diff --git a/docs/branch-review-ledger.md b/docs/branch-review-ledger.md
index d354facd1..d3f05d9f8 100644
--- a/docs/branch-review-ledger.md
+++ b/docs/branch-review-ledger.md
@@ -711,6 +711,7 @@ Use this ledger to prevent repeated branch and PR reviews when the reviewed HEAD
 | 2026-07-21 | claude/clinical-kb-pwa-review-asi3wb (Option A wave verdict — no code change; #1040 merged as cde6c5c) | canary run 29827012719 (#61, main cde6c5c) vs banked #60 (29800029819) | OPTION A WAVE ADOPTED — FIRST FULLY-GREEN 44-CASE CANARY IN PROGRAM HISTORY (Blocking failures: None). (1) Option A payoff EXCEEDED: citation_failure_rate 0.0227→0; the neuroleptic-side-effect-escalation case flipped from wrong-doc→failed-generation→1-citation-fallback to **strong route, successful gpt-5.6-sol generation, passed in 15.4s with no fallback marker** — the rescued S3 retrieval fixed generation itself, not just the citation count; expected_source_hit 0.6364→0.6591. (2) Golden held exactly as the blast-radius analysis promised: 36/36 PASS, content_recall 1.0, mrr@10 0.8921 BYTE-IDENTICAL to the pre-wave baseline (zero ordering movement — no golden case fires the predicate), irrelevant@10 0.1083→0.0917 (slightly better). (3) Parity payoff PARTIAL: monitoring targeting 1/5→2/5 (olanzapine-lai flipped — previously called a retrieval-depth residual; quetiapine-dose also flipped on the dose side); lithium-range (232ch) + metabolic (73ch, byte-identical answer to #60) did NOT flip despite offline-proven fixes — their live chunk sets evidently contain no admissible schedule sentence even under the widened gate → reclassified as retrieval-depth/live-content residuals joining adhd; below the ≥3/5 target but strictly improved, no regression anywhere. Dose 2/5 vs 2/4: same passing count, applicable set grew (new quality-metformin-renal-dosing miss = eval-set churn, not regression). (4) No-worse EXCEEDED: relevance 0.5333→0.6 (the two-step watch-item slide FULLY REVERSED to the #58 level), targeting_rate 0.6667→0.6957, fail_closed 0.9 held, readability/artifact_leaks 1.0, route ceilings 0, grounded 1.0, unsupported_correct 1.0, numeric 0, p95 22.8s, red_result 3/3. Adoption per the measured-gain rule: primary goal achieved, three case flips, relevance recovered, zero regressions. Residual queue: monitoring retrieval-depth trio (lithium-range/metabolic/adhd), E-3d H2 discards, weekly ANSWER_CASE_LIMIT 8→44 raise now unblocked (gate would be green), comparison-class coverage. Wave spend +~$2-4 → Phase E + Option A total ~$12-20 of ≤$20. | Evidence: run #61 job log read (Threshold Status: None; Answer Metrics; neuroleptic diagnostics row; targeting metric_rates + 6-miss list; golden 36 PASS lines + summary). Revert drill NOT triggered. |
 | 2026-07-21 | claude/clinical-kb-pwa-review-asi3wb (PR: I9 weekly coverage raise) | see PR head | ADDENDUM 5 post-green item I9 (plan-authorized "after reds fixed"): weekly scheduled canary ANSWER_CASE_LIMIT default 8→44 — the Sunday 18:00 UTC cron now guards the FULL answer-quality case set instead of the first 8 (both #57 blocking reds historically lived OUTSIDE the first 8, leaving the weekly gate blind to them). Unblocked by run #61 proving the citation gate green on the full 44. Cost: est +$1-2/week (user-authorized in the plan). Contract test pin updated in lockstep (eval-canary-workflow.test.ts). Dispatch shapes unchanged (input override still wins); operational-risk diff, plain-revert rollback. | check:github-actions PASS; check:ci-scope PASS; check:gate-manifest PASS (20/20); eval-canary-workflow contract 4/4; prettier clean; no provider calls |
 | 2026-07-21 | claude/database-governance-audit-10b6ed (PR #1051: source-governance audit — safe subset) | cee396730 | Governance-metadata observability + UI display + provenance flow test; no ranking/retrieval/generation surface touched. | IMPLEMENTED + handed off (not a review of prior work). Resolved audit #1 (logger.warn on unrecognized enum values; return value unchanged), #2 (review_due_source added to frontendVisibleWarningCodes → answer-level badge; warning-severity, no refusal impact), #9 (source_metadata retained on safety-finding citations + governance pill in SafetyFindingsListContent), #13 (new tests/provenance-flow.test.ts: DB-normalize→governance→client payload sources+safety citations→render policy). Deferred #4/5/6/8/10 (RAG-protected ranking/selection/LLM-context/cache — need live eval-canary+approval), #11/#5 flag debt (D5/D4), #3 (is_public schema/RLS), #7 (conflict-detection scope), #12 (canary automation). Rebased onto origin/main (was 18 behind; conflict-free — none of the 18 commits touched the 8 files). PR-policy CI green (confirmed no ragRankingPatterns match). | verify:pr-local exit 0 (351 files/3129 tests, production build, client-bundle secret scan, offline RAG fixtures 36/36); typecheck + lint + prettier green. verify:ui NOT run locally: pre-existing globals.css Tailwind/Turbopack dev-compile error (git-clean, unrelated; prod build passed) — CI Production UI job covers it. check:production-readiness deferred (offline env/config validator; PR changes no env/secret/config inputs; secretless worktree). No provider calls. |
+| 2026-07-24 | codex/reindex-agent-enrichment-guard (PR #1143) | f82cf9cfa | Run PR sweep: CI fix + threads + drift | Before: PR required green, 1 unresolved reindex/agent-claim serialization thread, branch behind main. After: remote branch already had atomic RPC fix 2bb0470d plus merge-main d2d57d7c; validated pushed head, resolved thread via GraphQL; reply mutation 403 noted in commit f82cf9cf. | node scripts/run-vitest.mjs run --reporter=dot tests/reindex-enrichment-lease.test.ts tests/private-access-routes.test.ts tests/supabase-schema.test.ts tests/drift-detection.test.ts PASS (218/218); npm run check:migration-role PASS; npm run typecheck PASS; npm run check:production-readiness PASS with non-blocking warnings; no live eval gates run. |
 
 ## 2026-07-24 — work search chrome behaviour review
 
diff --git a/src/app/api/documents/[id]/reindex/route.ts b/src/app/api/documents/[id]/reindex/route.ts
index e9ee68cd2..0cbce5a1f 100644
--- a/src/app/api/documents/[id]/reindex/route.ts
+++ b/src/app/api/documents/[id]/reindex/route.ts
@@ -8,11 +8,9 @@ import {
   checkIngestionMutationSafety,
   hasActiveAgentEnrichmentJob,
   ingestionMutationSafetyPayload,
-  ingestionRollbackFenceStamp,
   type IngestionJobRow,
 } from "@/lib/ingestion-mutation-safety";
 import { consumeApiRateLimit, rateLimitJsonResponse } from "@/lib/api-rate-limit";
-import { isAtomicReindexCandidate } from "@/lib/reindex-pipeline";
 import { createAdminClient } from "@/lib/supabase/admin";
 import { AuthenticationError, requireAuthenticatedUser, unauthorizedResponse } from "@/lib/supabase/auth";
 import { parseJsonBodyOrDefault } from "@/lib/validation/body";
@@ -28,6 +26,12 @@ const reindexModeSchema = z
 const reindexRouteParamsSchema = z.object({
   id: z.string().uuid(),
 });
+const reindexRequestResultSchema = z.discriminatedUnion("outcome", [
+  z.object({ outcome: z.literal("not_found") }),
+  z.object({ outcome: z.literal("agent_enrichment_active") }),
+  z.object({ outcome: z.literal("ingestion_active") }),
+  z.object({ outcome: z.literal("queued"), job: z.object({ id: z.string() }).passthrough() }),
+]);
 
 async function readMode(request: Request) {
   const parsed = await parseJsonBodyOrDefault(request, reindexModeSchema, { mode: "full" });
@@ -96,113 +100,49 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
       return NextResponse.json({ mode, queued }, { status: 202 });
     }
 
-    const atomicReindex = isAtomicReindexCandidate(document);
-    // Rollback fence: the queue-state write stamps updated_at with a
-    // per-request value and the rollback below matches on that stamp, making
-    // it a single conditional UPDATE that is atomic server-side. An
-    // overlapping reindex/retry re-stamps the row before enqueueing its own
-    // job, so a stale rollback from this request matches zero rows instead of
-    // reverting the newer queue state. The competing-job SELECT below is only
-    // a cheap fast path; the fence is what closes the check-then-write race.
-    const rollbackFence = ingestionRollbackFenceStamp();
-    const rollbackDocumentPayload = atomicReindex
-      ? { error_message: document.error_message ?? null }
-      : {
-          status: document.status ?? null,
-          error_message: document.error_message ?? null,
-          page_count: document.page_count ?? 0,
-          chunk_count: document.chunk_count ?? 0,
-          image_count: document.image_count ?? 0,
-        };
-    const { error: updateError } = await supabase
-      .from("documents")
-      .update(
-        atomicReindex
-          ? { error_message: null, updated_at: rollbackFence }
-          : {
-              status: "queued",
-              error_message: null,
-              page_count: 0,
-              chunk_count: 0,
-              image_count: 0,
-              updated_at: rollbackFence,
-            },
-      )
-      .eq("id", id)
-      .eq("owner_id", user.id);
-    if (updateError) throw new Error(updateError.message);
-
-    const { data: job, error: jobError } = await supabase
-      .from("ingestion_jobs")
-      .insert({
-        document_id: id,
-        batch_id: document.import_batch_id ?? null,
-        status: "pending",
-        stage: "queued",
-        progress: 0,
-        max_attempts: env.WORKER_MAX_ATTEMPTS,
-      })
-      .select()
-      .single();
+    const staleBefore = new Date(Date.now() - env.WORKER_STALE_AFTER_MINUTES * 60_000).toISOString();
+    const { data: reindexResult, error: reindexError } = await supabase.rpc("request_ingestion_reindex_if_agent_idle", {
+      p_document_id: id,
+      p_owner_id: user.id,
+      p_stale_before: staleBefore,
+      p_max_attempts: env.WORKER_MAX_ATTEMPTS,
+    });
+    if (reindexError) throw new Error(reindexError.message);
 
-    if (jobError) {
-      // A concurrent transactional delete holds the parent row lock while it
-      // removes the document. Once deletion commits, the FK check for this job
-      // fails with 23503. That is a normal lifecycle conflict, not a server
-      // fault, and there is no surviving document state to roll back.
-      if (jobError.code === "23503") {
-        return NextResponse.json(
-          { error: "Document was deleted while reindexing. Refresh the document list and retry." },
-          { status: 409 },
-        );
-      }
-      // R17: a unique index on ingestion_jobs(document_id) where status in
-      // (pending,processing) can reject this insert with 23505 when a
-      // concurrent request won the race between the pre-check above and this
-      // insert. That is the same "already queued" condition the pre-check
-      // reports, so surface it the same way (409, not a raw constraint 500).
-      if (jobError.code === "23505") {
-        const { data: raceJobs, error: raceJobsError } = await supabase
-          .from("ingestion_jobs")
-          .select(activeIngestionJobColumns)
-          .eq("document_id", id)
-          .in("status", ["pending", "processing"]);
-        if (!raceJobsError && (raceJobs?.length ?? 0) > 0) {
-          const safety = buildActiveJobsSafetyResult(
-            raceJobs as IngestionJobRow[],
-            env.WORKER_STALE_AFTER_MINUTES,
-            new Date().toISOString(),
-          );
-          return NextResponse.json(ingestionMutationSafetyPayload(safety), { status: 409 });
-        }
-      }
+    const parsedResult = reindexRequestResultSchema.safeParse(reindexResult);
+    if (!parsedResult.success) throw new Error("request_ingestion_reindex_if_agent_idle returned an invalid result.");
+    if (parsedResult.data.outcome === "not_found") {
+      return NextResponse.json(
+        { error: "Document was deleted while reindexing. Refresh the document list and retry." },
+        { status: 409 },
+      );
+    }
+    if (parsedResult.data.outcome === "agent_enrichment_active") {
+      return NextResponse.json(
+        { error: "Document has active agent enrichment work. Wait for it to finish before reindexing." },
+        { status: 409 },
+      );
+    }
+    if (parsedResult.data.outcome === "ingestion_active") {
       const { data: competingJobs, error: competingJobsError } = await supabase
         .from("ingestion_jobs")
-        .select("id")
+        .select(activeIngestionJobColumns)
         .eq("document_id", id)
-        .in("status", ["pending", "processing"])
-        .limit(1);
+        .in("status", ["pending", "processing"]);
       if (competingJobsError) {
-        throw new Error(
-          `Failed to enqueue reindex job: ${jobError.message}; competing-job check failed: ${competingJobsError.message}`,
-        );
+        throw new Error(`Atomic reindex found competing work but could not load it: ${competingJobsError.message}`);
       }
-      if ((competingJobs?.length ?? 0) === 0) {
-        const { error: rollbackError } = await supabase
-          .from("documents")
-          .update(rollbackDocumentPayload)
-          .eq("id", id)
-          .eq("owner_id", user.id)
-          .eq("updated_at", rollbackFence);
-        if (rollbackError) {
-          throw new Error(
-            `Failed to enqueue reindex job: ${jobError.message}; rollback failed: ${rollbackError.message}`,
-          );
-        }
+      if ((competingJobs?.length ?? 0) > 0) {
+        const safety = buildActiveJobsSafetyResult(
+          competingJobs as IngestionJobRow[],
+          env.WORKER_STALE_AFTER_MINUTES,
+          new Date().toISOString(),
+        );
+        return NextResponse.json(ingestionMutationSafetyPayload(safety), { status: 409 });
       }
-      throw new Error(jobError.message);
+      return NextResponse.json({ error: "Document already has active indexing work." }, { status: 409 });
     }
-    return NextResponse.json({ job }, { status: 201 });
+    return NextResponse.json({ job: parsedResult.data.job }, { status: 201 });
   } catch (error) {
     if (error instanceof AuthenticationError) return unauthorizedResponse();
     return jsonError(error);
diff --git a/src/app/api/documents/bulk/reindex/route.ts b/src/app/api/documents/bulk/reindex/route.ts
index 8ccc26002..558d61f21 100644
--- a/src/app/api/documents/bulk/reindex/route.ts
+++ b/src/app/api/documents/bulk/reindex/route.ts
@@ -3,17 +3,12 @@ import { z } from "zod";
 import { env, isDemoMode } from "@/lib/env";
 import { jsonError, PublicApiError } from "@/lib/http";
 import {
-  activeIngestionJobColumns,
-  buildActiveJobsSafetyResult,
   checkIngestionMutationSafety,
   ingestionMutationSafetyPayload,
-  ingestionRollbackFenceStamp,
   listDocumentsWithActiveAgentEnrichment,
-  type IngestionJobRow,
 } from "@/lib/ingestion-mutation-safety";
 import { consumeApiRateLimit, rateLimitJsonResponse } from "@/lib/api-rate-limit";
 import { invalidateRagCachesForOwner } from "@/lib/rag/rag";
-import { isAtomicReindexCandidate } from "@/lib/reindex-pipeline";
 import { createAdminClient } from "@/lib/supabase/admin";
 import { AuthenticationError, requireAuthenticatedUser, unauthorizedResponse } from "@/lib/supabase/auth";
 import { parseJsonBody } from "@/lib/validation/body";
@@ -24,6 +19,12 @@ const bulkReindexSchema = z.object({
   documentIds: z.array(z.string().uuid()).min(1).max(10),
   mode: z.enum(["enrichment", "full", "retry_failed"]).default("enrichment"),
 });
+const reindexRequestResultSchema = z.discriminatedUnion("outcome", [
+  z.object({ outcome: z.literal("not_found") }),
+  z.object({ outcome: z.literal("agent_enrichment_active") }),
+  z.object({ outcome: z.literal("ingestion_active") }),
+  z.object({ outcome: z.literal("queued"), job: z.object({ id: z.string() }).passthrough() }),
+]);
 
 export async function POST(request: Request) {
   try {
@@ -113,103 +114,32 @@ export async function POST(request: Request) {
           continue;
         }
 
-        const atomicReindex = isAtomicReindexCandidate(document);
-        // Rollback fence: same pattern as the single-document reindex route —
-        // the queue-state write stamps updated_at and the rollback matches on
-        // the stamp, so a stale rollback cannot revert a newer queue state
-        // written by an overlapping reindex/retry. The competing-job SELECT is
-        // only a fast path; the fence closes the check-then-write race.
-        const rollbackFence = ingestionRollbackFenceStamp();
-        const rollbackDocumentPayload = atomicReindex
-          ? { error_message: document.error_message ?? null }
-          : {
-              status: document.status ?? null,
-              error_message: document.error_message ?? null,
-              page_count: document.page_count ?? 0,
-              chunk_count: document.chunk_count ?? 0,
-              image_count: document.image_count ?? 0,
-            };
-        const { error: updateError } = await supabase
-          .from("documents")
-          .update(
-            atomicReindex
-              ? { error_message: null, updated_at: rollbackFence }
-              : {
-                  status: "queued",
-                  error_message: null,
-                  page_count: 0,
-                  chunk_count: 0,
-                  image_count: 0,
-                  updated_at: rollbackFence,
-                },
-          )
-          .eq("id", document.id)
-          .eq("owner_id", user.id);
-        if (updateError) throw new Error(updateError.message);
-        const { data: job, error: jobError } = await supabase
-          .from("ingestion_jobs")
-          .insert({
-            document_id: document.id,
-            batch_id: document.import_batch_id ?? null,
-            status: "pending",
-            stage: "queued",
-            progress: 0,
-            max_attempts: env.WORKER_MAX_ATTEMPTS,
-          })
-          .select("id")
-          .single();
-        if (jobError) {
-          if (jobError.code === "23503") {
-            throw new Error("Document was deleted while reindexing. Refresh the document list and retry.");
-          }
-          // R17: same race as the single-document reindex route — a unique
-          // index on ingestion_jobs(document_id) where status in
-          // (pending,processing) can reject this insert with 23505 when a
-          // concurrent request won the race after the pre-check ran. Surface
-          // the friendly "already queued" message instead of the raw
-          // constraint-violation text.
-          if (jobError.code === "23505") {
-            const { data: raceJobs, error: raceJobsError } = await supabase
-              .from("ingestion_jobs")
-              .select(activeIngestionJobColumns)
-              .eq("document_id", document.id)
-              .in("status", ["pending", "processing"]);
-            if (!raceJobsError && (raceJobs?.length ?? 0) > 0) {
-              const safety = buildActiveJobsSafetyResult(
-                raceJobs as IngestionJobRow[],
-                env.WORKER_STALE_AFTER_MINUTES,
-                new Date().toISOString(),
-              );
-              throw new Error(safety.message);
-            }
-          }
-          const { data: competingJobs, error: competingJobsError } = await supabase
-            .from("ingestion_jobs")
-            .select("id")
-            .eq("document_id", document.id)
-            .in("status", ["pending", "processing"])
-            .limit(1);
-          if (competingJobsError) {
-            throw new Error(
-              `Failed to enqueue bulk reindex job: ${jobError.message}; competing-job check failed: ${competingJobsError.message}`,
-            );
-          }
-          if ((competingJobs?.length ?? 0) === 0) {
-            const { error: rollbackError } = await supabase
-              .from("documents")
-              .update(rollbackDocumentPayload)
-              .eq("id", document.id)
-              .eq("owner_id", user.id)
-              .eq("updated_at", rollbackFence);
-            if (rollbackError) {
-              throw new Error(
-                `Failed to enqueue bulk reindex job: ${jobError.message}; rollback failed: ${rollbackError.message}`,
-              );
-            }
-          }
-          throw new Error(jobError.message);
+        const staleBefore = new Date(Date.now() - env.WORKER_STALE_AFTER_MINUTES * 60_000).toISOString();
+        const { data: reindexResult, error: reindexError } = await supabase.rpc(
+          "request_ingestion_reindex_if_agent_idle",
+          {
+            p_document_id: document.id,
+            p_owner_id: user.id,
+            p_stale_before: staleBefore,
+            p_max_attempts: env.WORKER_MAX_ATTEMPTS,
+          },
+        );
+        if (reindexError) throw new Error(reindexError.message);
+
+        const parsedResult = reindexRequestResultSchema.safeParse(reindexResult);
+        if (!parsedResult.success) {
+          throw new Error("request_ingestion_reindex_if_agent_idle returned an invalid result.");
+        }
+        if (parsedResult.data.outcome === "not_found") {
+          throw new Error("Document was deleted while reindexing. Refresh the document list and retry.");
+        }
+        if (parsedResult.data.outcome === "agent_enrichment_active") {
+          throw new Error("Document has active agent enrichment work. Wait for it to finish before reindexing.");
+        }
+        if (parsedResult.data.outcome === "ingestion_active") {
+          throw new Error("Document already has pending or processing indexing work.");
         }
-        results.push({ documentId: document.id, mode: parsed.mode, ok: true, jobId: job.id });
+        results.push({ documentId: document.id, mode: parsed.mode, ok: true, jobId: parsedResult.data.job.id });
       } catch (error) {
         results.push({
           documentId: document.id,
diff --git a/src/lib/supabase/database.types.ts b/src/lib/supabase/database.types.ts
index 58e20c4de..0b7e1f553 100644
--- a/src/lib/supabase/database.types.ts
+++ b/src/lib/supabase/database.types.ts
@@ -2402,6 +2402,15 @@ export type Database = {
         };
         Returns: Json;
       };
+      request_ingestion_reindex_if_agent_idle: {
+        Args: {
+          p_document_id: string;
+          p_max_attempts: number;
+          p_owner_id: string;
+          p_stale_before: string;
+        };
+        Returns: Json;
+      };
       detect_legacy_ivfflat_indexes: { Args: never; Returns: string[] };
       document_label_metadata: {
         Args: { p_document_id: string };
diff --git a/supabase/drift-manifest.json b/supabase/drift-manifest.json
index 4f47aad29..544324da2 100644
--- a/supabase/drift-manifest.json
+++ b/supabase/drift-manifest.json
@@ -1,9 +1,9 @@
 {
-  "generated_at": "2026-07-24T05:47:11.755Z",
+  "generated_at": "2026-07-24T06:18:38.047Z",
   "generator": "scripts/generate-drift-manifest.ts",
   "postgres_image": "supabase/postgres:17.6.1.127",
-  "schema_sha256": "81fd99d0849bde3c68a7a476e923034e0bd0258fda08e8e67f35b1c426f110e4",
-  "replay_seconds": 17,
+  "schema_sha256": "7c22a34be4b56894e19136e97897589a000b0502f7b666fffc9060c4e6c907a0",
+  "replay_seconds": 18,
   "snapshot": {
     "views": [
       {
@@ -6661,7 +6661,7 @@
           "postgres=X/postgres",
           "service_role=X/postgres"
         ],
-        "def_hash": "55f95e3b799f7716a7fd264d2b3cdd5a",
+        "def_hash": "d061668633026acc816e1d63403765d1",
         "signature": "public.claim_indexing_v3_agent_jobs(text,integer,integer)"
       },
       {
@@ -7213,6 +7213,14 @@
         "def_hash": "baa8c0bda37cc561026baecdf1cca1c2",
         "signature": "public.request_indexing_v3_enrichment(uuid,uuid)"
       },
+      {
+        "acl": [
+          "postgres=X/postgres",
+          "service_role=X/postgres"
+        ],
+        "def_hash": "608a60e6ec287dda67f9ae93c5889ceb",
+        "signature": "public.request_ingestion_reindex_if_agent_idle(uuid,uuid,timestamp with time zone,integer)"
+      },
       {
         "acl": [
           "postgres=X/postgres",
diff --git a/supabase/migrations/20260724060000_atomic_reindex_agent_guard.sql b/supabase/migrations/20260724060000_atomic_reindex_agent_guard.sql
new file mode 100644
index 000000000..49e397a97
--- /dev/null
+++ b/supabase/migrations/20260724060000_atomic_reindex_agent_guard.sql
@@ -0,0 +1,190 @@
+-- Serialize agent enrichment claims with full/retry ingestion reindex requests.
+-- The claim path now locks the document row as well as the agent row. The
+-- reindex RPC locks that same document before checking the agent lease and
+-- creating the ingestion job, so the decision and queue mutation share one
+-- transaction and cannot interleave.
+
+create or replace function public.claim_indexing_v3_agent_jobs(
+  p_worker_id text,
+  p_claim_limit integer default 1,
+  p_stale_after_minutes integer default 45
+)
+returns table (
+  id uuid, document_id uuid, batch_id uuid, status text, stage text, progress integer,
+  error_message text, attempt_count integer, max_attempts integer, locked_at timestamptz,
+  locked_by text, documents jsonb
+)
+language plpgsql
+set search_path = public, extensions, pg_temp
+as $$
+begin
+  insert into public.indexing_v3_agent_jobs (
+    document_id, status, enrichment_status, next_run_at, version, metadata, created_at, updated_at
+  )
+  select d.id, 'pending', coalesce(d.metadata->>'enrichment_status', 'pending'),
+    case when coalesce(d.metadata->>'indexing_v3_agent_next_run_at', '') ~ '^[0-9]{4}-[0-9]{2}-[0-9]{2}'
+      then (d.metadata->>'indexing_v3_agent_next_run_at')::timestamptz else null end,
+    coalesce(nullif(d.metadata->>'indexing_v3_agent_version', ''), 'visual-core-v3'),
+    '{}'::jsonb, coalesce(d.created_at, now()), now()
+  from public.documents d
+  where d.status = 'indexed'
+    and not exists (
+      select 1 from public.ingestion_jobs i
+      where i.document_id = d.id and i.status in ('pending', 'processing')
+    )
+    and d.metadata ? 'indexing_v3_agent_status'
+    and coalesce(d.metadata->>'indexing_v3_agent_status', 'pending')
+          not in ('completed', 'needs_enrichment_artifacts')
+  on conflict do nothing;
+
+  return query
+  with eligible_jobs as (
+    select j.id, j.document_id, j.attempt_count, j.max_attempts
+    from public.indexing_v3_agent_jobs j
+    join public.documents d on d.id = j.document_id and d.status = 'indexed'
+    where j.status not in ('completed', 'needs_enrichment_artifacts')
+      and not exists (
+        select 1 from public.ingestion_jobs i
+        where i.document_id = j.document_id and i.status in ('pending', 'processing')
+      )
+      and j.enrichment_status in ('pending', 'failed', 'processing')
+      and j.attempt_count < j.max_attempts
+      and coalesce(j.next_run_at, now()) <= now()
+      and (j.status <> 'processing' or j.locked_at is null
+        or j.locked_at < now() - make_interval(mins => p_stale_after_minutes))
+    order by coalesce(j.next_run_at, j.updated_at), j.id
+    limit greatest(p_claim_limit, 1)
+    for update of j, d skip locked
+  ),
+  claimed_jobs as (
+    update public.indexing_v3_agent_jobs j
+    set status = 'processing', enrichment_status = 'processing', locked_by = p_worker_id,
+      locked_at = now(), attempt_count = e.attempt_count + 1, last_error = null,
+      next_run_at = null, updated_at = now()
+    from eligible_jobs e where j.id = e.id returning j.*
+  ),
+  patched_documents as (
+    update public.documents d
+    set metadata = jsonb_strip_nulls(
+        (coalesce(d.metadata, '{}'::jsonb) - 'indexing_v3_agent_next_run_at' - 'indexing_v3_agent_last_error')
+        || jsonb_build_object(
+          'indexing_v3_agent_status', 'processing',
+          'indexing_v3_agent_version', cj.version,
+          'indexing_v3_agent_locked_by', p_worker_id,
+          'indexing_v3_agent_locked_at', cj.locked_at,
+          'indexing_v3_agent_attempt_count', cj.attempt_count,
+          'indexing_v3_agent_max_attempts', cj.max_attempts,
+          'indexing_v3_agent_updated_at', now(),
+          'enrichment_status', 'processing'
+        )
+      ), updated_at = now()
+    from claimed_jobs cj
+    where d.id = cj.document_id and d.status = 'indexed'
+    returning d.*, cj.id as job_id, cj.attempt_count as job_attempt_count,
+              cj.max_attempts as job_max_attempts, cj.locked_at as job_locked_at
+  )
+  select pd.job_id, pd.id, pd.import_batch_id, 'processing'::text, 'v3 enrichment claimed'::text,
+    95::integer, null::text, pd.job_attempt_count, pd.job_max_attempts, pd.job_locked_at,
+    p_worker_id,
+    to_jsonb(pd.*) - 'job_id' - 'job_attempt_count' - 'job_max_attempts' - 'job_locked_at'
+  from patched_documents pd;
+end;
+$$;
+
+revoke execute on function public.claim_indexing_v3_agent_jobs(text, integer, integer)
+  from public, anon, authenticated;
+grant execute on function public.claim_indexing_v3_agent_jobs(text, integer, integer)
+  to service_role;
+
+create or replace function public.request_ingestion_reindex_if_agent_idle(
+  p_document_id uuid,
+  p_owner_id uuid,
+  p_stale_before timestamptz,
+  p_max_attempts integer
+)
+returns jsonb
+language plpgsql
+security definer
+set search_path = ''
+as $$
+declare
+  v_document public.documents%rowtype;
+  v_job public.ingestion_jobs%rowtype;
+begin
+  if p_document_id is null or p_owner_id is null or p_stale_before is null
+     or p_max_attempts is null or p_max_attempts < 1 then
+    raise exception 'Reindex identifiers, stale cutoff, and max attempts are required.' using errcode = '22023';
+  end if;
+
+  select d.*
+    into v_document
+  from public.documents d
+  where d.id = p_document_id
+    and d.owner_id = p_owner_id
+  for update;
+
+  if not found then
+    return jsonb_build_object('outcome', 'not_found');
+  end if;
+
+  if exists (
+    select 1
+    from public.indexing_v3_agent_jobs a
+    where a.document_id = p_document_id
+      and a.status = 'processing'
+      and (
+        coalesce(a.locked_at, a.updated_at) is null
+        or coalesce(a.locked_at, a.updated_at) > p_stale_before
+      )
+  ) then
+    return jsonb_build_object('outcome', 'agent_enrichment_active');
+  end if;
+
+  if exists (
+    select 1
+    from public.ingestion_jobs i
+    where i.document_id = p_document_id
+      and i.status in ('pending', 'processing')
+  ) then
+    return jsonb_build_object('outcome', 'ingestion_active');
+  end if;
+
+  begin
+    update public.documents
+    set status = case when v_document.status = 'indexed' then status else 'queued' end,
+        error_message = null,
+        page_count = case when v_document.status = 'indexed' then page_count else 0 end,
+        chunk_count = case when v_document.status = 'indexed' then chunk_count else 0 end,
+        image_count = case when v_document.status = 'indexed' then image_count else 0 end,
+        updated_at = now()
+    where id = p_document_id
+      and owner_id = p_owner_id;
+
+    insert into public.ingestion_jobs (
+      document_id,
+      batch_id,
+      status,
+      stage,
+      progress,
+      max_attempts
+    ) values (
+      p_document_id,
+      v_document.import_batch_id,
+      'pending',
+      'queued',
+      0,
+      p_max_attempts
+    )
+    returning * into v_job;
+  exception when unique_violation then
+    return jsonb_build_object('outcome', 'ingestion_active');
+  end;
+
+  return jsonb_build_object('outcome', 'queued', 'job', to_jsonb(v_job));
+end;
+$$;
+
+revoke all on function public.request_ingestion_reindex_if_agent_idle(uuid, uuid, timestamptz, integer)
+  from public, anon, authenticated;
+grant execute on function public.request_ingestion_reindex_if_agent_idle(uuid, uuid, timestamptz, integer)
+  to service_role;
diff --git a/supabase/schema.sql b/supabase/schema.sql
index 6c3ee542d..36c8009a9 100644
--- a/supabase/schema.sql
+++ b/supabase/schema.sql
@@ -1253,7 +1253,7 @@ begin
     and d.metadata ? 'indexing_v3_agent_status'
     and coalesce(d.metadata->>'indexing_v3_agent_status', 'pending')
           not in ('completed', 'needs_enrichment_artifacts')
-  on conflict (document_id) do nothing;
+  on conflict do nothing;
 
   return query
   with eligible_jobs as (
@@ -1272,7 +1272,7 @@ begin
         or j.locked_at < now() - make_interval(mins => p_stale_after_minutes))
     order by coalesce(j.next_run_at, j.updated_at), j.id
     limit greatest(p_claim_limit, 1)
-    for update of j skip locked
+    for update of j, d skip locked
   ),
   claimed_jobs as (
     update public.indexing_v3_agent_jobs j
@@ -8421,6 +8421,99 @@ revoke all on function public.retry_ingestion_job_if_idle(uuid, uuid, timestampt
   from public, anon, authenticated;
 grant execute on function public.retry_ingestion_job_if_idle(uuid, uuid, timestamptz, integer, timestamptz, timestamptz)
   to service_role;
+
+create or replace function public.request_ingestion_reindex_if_agent_idle(
+  p_document_id uuid,
+  p_owner_id uuid,
+  p_stale_before timestamptz,
+  p_max_attempts integer
+)
+returns jsonb
+language plpgsql
+security definer
+set search_path = ''
+as $$
+declare
+  v_document public.documents%rowtype;
+  v_job public.ingestion_jobs%rowtype;
+begin
+  if p_document_id is null or p_owner_id is null or p_stale_before is null
+     or p_max_attempts is null or p_max_attempts < 1 then
+    raise exception 'Reindex identifiers, stale cutoff, and max attempts are required.' using errcode = '22023';
+  end if;
+
+  select d.*
+    into v_document
+  from public.documents d
+  where d.id = p_document_id
+    and d.owner_id = p_owner_id
+  for update;
+
+  if not found then
+    return jsonb_build_object('outcome', 'not_found');
+  end if;
+
+  if exists (
+    select 1
+    from public.indexing_v3_agent_jobs a
+    where a.document_id = p_document_id
+      and a.status = 'processing'
+      and (
+        coalesce(a.locked_at, a.updated_at) is null
+        or coalesce(a.locked_at, a.updated_at) > p_stale_before
+      )
+  ) then
+    return jsonb_build_object('outcome', 'agent_enrichment_active');
+  end if;
+
+  if exists (
+    select 1
+    from public.ingestion_jobs i
+    where i.document_id = p_document_id
+      and i.status in ('pending', 'processing')
+  ) then
+    return jsonb_build_object('outcome', 'ingestion_active');
+  end if;
+
+  begin
+    update public.documents
+    set status = case when v_document.status = 'indexed' then status else 'queued' end,
+        error_message = null,
+        page_count = case when v_document.status = 'indexed' then page_count else 0 end,
+        chunk_count = case when v_document.status = 'indexed' then chunk_count else 0 end,
+        image_count = case when v_document.status = 'indexed' then image_count else 0 end,
+        updated_at = now()
+    where id = p_document_id
+      and owner_id = p_owner_id;
+
+    insert into public.ingestion_jobs (
+      document_id,
+      batch_id,
+      status,
+      stage,
+      progress,
+      max_attempts
+    ) values (
+      p_document_id,
+      v_document.import_batch_id,
+      'pending',
+      'queued',
+      0,
+      p_max_attempts
+    )
+    returning * into v_job;
+  exception when unique_violation then
+    return jsonb_build_object('outcome', 'ingestion_active');
+  end;
+
+  return jsonb_build_object('outcome', 'queued', 'job', to_jsonb(v_job));
+end;
+$$;
+
+revoke all on function public.request_ingestion_reindex_if_agent_idle(uuid, uuid, timestamptz, integer)
+  from public, anon, authenticated;
+grant execute on function public.request_ingestion_reindex_if_agent_idle(uuid, uuid, timestamptz, integer)
+  to service_role;
 -- Catalog-level, fail-closed verification for future objects created by
 -- postgres. A missing pg_default_acl row must be interpreted through
 -- acldefault(), including PostgreSQL's built-in PUBLIC EXECUTE on functions.
diff --git a/tests/private-access-routes.test.ts b/tests/private-access-routes.test.ts
index ec8d99996..3646b608b 100644
--- a/tests/private-access-routes.test.ts
+++ b/tests/private-access-routes.test.ts
@@ -269,6 +269,9 @@ function createSupabaseMock(resolve: QueryResolver = defaultQueryResolver) {
         error: null,
       };
     }
+    if (name === "request_ingestion_reindex_if_agent_idle") {
+      return ok({ outcome: "queued", job: { id: "reindex-job" } });
+    }
     return ok([]);
   });
   const client = {
@@ -296,6 +299,17 @@ function createSupabaseMock(resolve: QueryResolver = defaultQueryResolver) {
   return client;
 }
 
+function mockAtomicReindexRpc(
+  client: ReturnType<typeof createSupabaseMock>,
+  resolve: (args: Record<string, unknown>) => QueryResult | Promise<QueryResult>,
+) {
+  client.rpc.mockImplementation(async (name: string, args?: Record<string, unknown>) => {
+    if (name === "consume_api_rate_limit") return ok([rateLimitRow()]);
+    if (name === "request_ingestion_reindex_if_agent_idle") return resolve(args ?? {});
+    return ok([]);
+  });
+}
+
 function mockRuntime(
   client: ReturnType<typeof createSupabaseMock>,
   ragMock?: Record<string, unknown>,
@@ -1978,6 +1992,16 @@ describe("private document API access", () => {
     ];
     const client = createSupabaseMock((call) => {
       if (call.table === "documents" && call.operation === "select") return ok(document);
+      if (call.table === "indexing_v3_agent_jobs") {
+        return ok([
+          {
+            document_id: documentId,
+            status: "processing",
+            locked_at: new Date().toISOString(),
+            updated_at: new Date().toISOString(),
+          },
+        ]);
+      }
       if (call.table === "document_chunks") return ok(chunks);
       if (call.table === "document_images") return ok(images);
       return ok([]);
@@ -2011,6 +2035,7 @@ describe("private document API access", () => {
     expect(upsertDocumentEnrichment).not.toHaveBeenCalled();
     expect(client.calls[0].selected).toContain("metadata");
     expect(client.calls[0].filters).toContainEqual({ column: "owner_id", value: userId });
+    expect(client.calls.some((call) => call.table === "indexing_v3_agent_jobs")).toBe(false);
     expect(upsertDocumentDeepMemory).not.toHaveBeenCalled();
   });
 
@@ -2191,6 +2216,100 @@ describe("private document API access", () => {
     expect(upsertDocumentDeepMemory).not.toHaveBeenCalled();
   });
 
+  it("blocks full reindex while a fresh agent-enrichment lease is processing", async () => {
+    const document = {
+      id: documentId,
+      owner_id: userId,
+      title: "Agent-Enriched Protocol",
+      file_name: "agent-enriched.pdf",
+      source_path: null,
+      import_batch_id: null,
+      status: "indexed",
+      error_message: null,
+      page_count: 3,
+      chunk_count: 8,
+      image_count: 1,
+      metadata: {},
+    };
+    const now = new Date().toISOString();
+    const client = createSupabaseMock((call) => {
+      if (call.table === "documents" && call.operation === "select") return ok(document);
+      if (call.table === "import_batches") return ok([]);
+      if (call.table === "ingestion_jobs" && call.operation === "select") return ok([]);
+      if (call.table === "indexing_v3_agent_jobs") {
+        return ok([{ document_id: documentId, status: "processing", locked_at: now, updated_at: now }]);
+      }
+      return ok([]);
+    });
+    mockRuntime(client);
+    const { POST } = await import("../src/app/api/documents/[id]/reindex/route");
+
+    const response = await POST(
+      authenticatedRequest(`/api/documents/${documentId}/reindex`, {
+        method: "POST",
+        body: JSON.stringify({ mode: "full" }),
+      }),
+      { params: Promise.resolve({ id: documentId }) },
+    );
+
+    expect(response.status).toBe(409);
+    expect(await payload(response)).toMatchObject({
+      error: "Reindex is paused while enrichment is active.",
+    });
+    expect(client.calls.some((call) => call.table === "documents" && call.operation === "update")).toBe(false);
+    expect(client.calls.some((call) => call.table === "ingestion_jobs" && call.operation === "insert")).toBe(false);
+  });
+
+  it("allows full reindex after an agent-enrichment lease becomes stale", async () => {
+    const document = {
+      id: documentId,
+      owner_id: userId,
+      title: "Stale Agent Protocol",
+      file_name: "stale-agent.pdf",
+      source_path: null,
+      import_batch_id: null,
+      status: "indexed",
+      error_message: null,
+      page_count: 3,
+      chunk_count: 8,
+      image_count: 1,
+      metadata: {},
+    };
+    const stale = new Date(Date.now() - 11 * 60_000).toISOString();
+    const client = createSupabaseMock((call) => {
+      if (call.table === "documents" && call.operation === "select") return ok(document);
+      if (call.table === "import_batches") return ok([]);
+      if (call.table === "ingestion_jobs" && call.operation === "select") return ok([]);
+      if (call.table === "indexing_v3_agent_jobs") {
+        return ok([{ document_id: documentId, status: "processing", locked_at: stale, updated_at: stale }]);
+      }
+      return ok([]);
+    });
+    mockRuntime(client);
+    const { POST } = await import("../src/app/api/documents/[id]/reindex/route");
+
+    const response = await POST(
+      authenticatedRequest(`/api/documents/${documentId}/reindex`, {
+        method: "POST",
+        body: JSON.stringify({ mode: "full" }),
+      }),
+      { params: Promise.resolve({ id: documentId }) },
+    );
+
+    expect(response.status).toBe(201);
+    expect(client.rpc).toHaveBeenCalledWith(
+      "request_ingestion_reindex_if_agent_idle",
+      expect.objectContaining({
+        p_document_id: documentId,
+        p_owner_id: userId,
+        p_stale_before: expect.any(String),
+        p_max_attempts: 3,
+      }),
+    );
+    expect(client.calls.some((call) => call.table === "documents" && call.operation === "update")).toBe(false);
+    expect(client.calls.some((call) => call.table === "ingestion_jobs" && call.operation === "insert")).toBe(false);
+  });
+
   it("blocks full reindex when the selected document already has active indexing work", async () => {
     const document = {
       id: documentId,
@@ -2343,7 +2462,7 @@ describe("private document API access", () => {
     expect(client.calls.some((call) => call.table === "documents" && call.operation === "update")).toBe(false);
   });
 
-  it("rolls back single-document queue mutation when full reindex job enqueue fails", async () => {
+  it("leaves queue mutation to the atomic RPC when full reindex enqueue fails", async () => {
     const document = {
       id: documentId,
       owner_id: userId,
@@ -2362,10 +2481,9 @@ describe("private document API access", () => {
       if (call.table === "documents" && call.operation === "select") return ok(document);
       if (call.table === "import_batches") return ok([]);
       if (call.table === "ingestion_jobs" && call.operation === "select") return ok([]);
-      if (call.table === "ingestion_jobs" && call.operation === "insert") return fail("job insert failed");
-      if (call.table === "documents" && call.operation === "update") return ok([]);
       return ok([]);
     });
+    mockAtomicReindexRpc(client, () => fail("transaction rolled back"));
     mockRuntime(client);
     const { POST } = await import("../src/app/api/documents/[id]/reindex/route");
 
@@ -2377,31 +2495,10 @@ describe("private document API access", () => {
       { params: Promise.resolve({ id: documentId }) },
     );
     const body = await payload(response);
-    const documentUpdates = client.calls.filter((call) => call.table === "documents" && call.operation === "update");
 
     expect(response.status).toBe(500);
     expect(body).toMatchObject({ error: "Request failed." });
-    expect(documentUpdates).toHaveLength(2);
-    expect(documentUpdates[0]?.updatePayload).toEqual({
-      status: "queued",
-      error_message: null,
-      page_count: 0,
-      chunk_count: 0,
-      image_count: 0,
-      updated_at: expect.any(String),
-    });
-    expect(documentUpdates[1]?.updatePayload).toEqual({
-      status: "failed",
-      error_message: "older failure",
-      page_count: 12,
-      chunk_count: 34,
-      image_count: 2,
-    });
-    // Rollback fence: the rollback must be conditional on the updated_at
-    // stamp the queue-state write set, so it is a single atomic UPDATE that
-    // cannot revert a newer queue state written by an overlapping request.
-    const fence = (documentUpdates[0]?.updatePayload as { updated_at?: string }).updated_at;
-    expect(documentUpdates[1]?.filters).toContainEqual({ column: "updated_at", value: fence });
+    expect(client.calls.some((call) => call.operation === "update" || call.operation === "insert")).toBe(false);
   });
 
   it("returns 409 without rollback when deletion wins the single-document reindex race", async () => {
@@ -2423,12 +2520,9 @@ describe("private document API access", () => {
       if (call.table === "documents" && call.operation === "select") return ok(document);
       if (call.table === "import_batches") return ok([]);
       if (call.table === "ingestion_jobs" && call.operation === "select") return ok([]);
-      if (call.table === "documents" && call.operation === "update") return ok([]);
-      if (call.table === "ingestion_jobs" && call.operation === "insert") {
-        return fail("ingestion_jobs_document_id_fkey", "23503");
-      }
       return ok([]);
     });
+    mockAtomicReindexRpc(client, () => ok({ outcome: "not_found" }));
     mockRuntime(client);
     const { POST } = await import("../src/app/api/documents/[id]/reindex/route");
 
@@ -2444,10 +2538,10 @@ describe("private document API access", () => {
     expect(await payload(response)).toMatchObject({
       error: "Document was deleted while reindexing. Refresh the document list and retry.",
     });
-    expect(client.calls.filter((call) => call.table === "documents" && call.operation === "update")).toHaveLength(1);
+    expect(client.calls.some((call) => call.operation === "update" || call.operation === "insert")).toBe(false);
   });
 
-  it("skips single-document rollback when a competing active job appears after the safety check", async () => {
+  it("returns the atomic RPC conflict when active ingestion appears after the safety check", async () => {
     const document = {
       id: documentId,
       owner_id: userId,
@@ -2462,17 +2556,31 @@ describe("private document API access", () => {
       image_count: 2,
       metadata: {},
     };
+    let ingestionReads = 0;
     const client = createSupabaseMock((call) => {
       if (call.table === "documents" && call.operation === "select") return ok(document);
       if (call.table === "import_batches") return ok([]);
-      if (call.table === "ingestion_jobs" && call.operation === "select" && call.limitCount === 1) {
-        return ok([{ id: "competing-job" }]);
+      if (call.table === "ingestion_jobs" && call.operation === "select" && call.selected?.includes("locked_at")) {
+        ingestionReads += 1;
+        if (ingestionReads === 1) return ok([]);
+        return ok([
+          {
+            id: "competing-job",
+            document_id: documentId,
+            status: "pending",
+            stage: "queued",
+            locked_at: null,
+            updated_at: new Date().toISOString(),
+            error_message: null,
+            attempt_count: 0,
+            max_attempts: 3,
+          },
+        ]);
       }
       if (call.table === "ingestion_jobs" && call.operation === "select") return ok([]);
-      if (call.table === "ingestion_jobs" && call.operation === "insert") return fail("job insert failed");
-      if (call.table === "documents" && call.operation === "update") return ok([]);
       return ok([]);
     });
+    mockAtomicReindexRpc(client, () => ok({ outcome: "ingestion_active" }));
     mockRuntime(client);
     const { POST } = await import("../src/app/api/documents/[id]/reindex/route");
 
@@ -2484,22 +2592,16 @@ describe("private document API access", () => {
       { params: Promise.resolve({ id: documentId }) },
     );
     const body = await payload(response);
-    const documentUpdates = client.calls.filter((call) => call.table === "documents" && call.operation === "update");
 
-    expect(response.status).toBe(500);
-    expect(body).toMatchObject({ error: "Request failed." });
-    expect(documentUpdates).toHaveLength(1);
-    expect(documentUpdates[0]?.updatePayload).toEqual({
-      status: "queued",
-      error_message: null,
-      page_count: 0,
-      chunk_count: 0,
-      image_count: 0,
-      updated_at: expect.any(String),
+    expect(response.status).toBe(409);
+    expect(body).toMatchObject({
+      error: "Document already has pending or processing indexing work.",
+      safety: { reason: "active_jobs", activeJobCount: 1 },
     });
+    expect(client.calls.some((call) => call.operation === "update" || call.operation === "insert")).toBe(false);
   });
 
-  it("rolls back per-document queue mutation when bulk full reindex enqueue fails", async () => {
+  it("reports an atomic RPC failure without direct bulk queue mutation", async () => {
     const document = {
       id: documentId,
       owner_id: userId,
@@ -2518,10 +2620,9 @@ describe("private document API access", () => {
       if (call.table === "documents" && call.operation === "select") return ok([document]);
       if (call.table === "import_batches") return ok([]);
       if (call.table === "ingestion_jobs" && call.operation === "select") return ok([]);
-      if (call.table === "documents" && call.operation === "update") return ok([]);
-      if (call.table === "ingestion_jobs" && call.operation === "insert") return fail("bulk job insert failed");
       return ok([]);
     });
+    mockAtomicReindexRpc(client, () => fail("transaction rolled back"));
     mockRuntime(client, { invalidateRagCachesForOwner: vi.fn() });
     const { POST } = await import("../src/app/api/documents/bulk/reindex/route");
 
@@ -2532,7 +2633,6 @@ describe("private document API access", () => {
       }),
     );
     const body = await payload(response);
-    const documentUpdates = client.calls.filter((call) => call.table === "documents" && call.operation === "update");
 
     expect(response.status).toBe(200);
     expect(body).toMatchObject({
@@ -2542,30 +2642,54 @@ describe("private document API access", () => {
           documentId,
           mode: "full",
           ok: false,
-          error: "bulk job insert failed",
+          error: "transaction rolled back",
         },
       ],
     });
-    expect(documentUpdates).toHaveLength(2);
-    expect(documentUpdates[0]?.updatePayload).toEqual({
-      status: "queued",
-      error_message: null,
-      page_count: 0,
-      chunk_count: 0,
-      image_count: 0,
-      updated_at: expect.any(String),
-    });
-    expect(documentUpdates[1]?.updatePayload).toEqual({
+    expect(client.calls.some((call) => call.operation === "update" || call.operation === "insert")).toBe(false);
+  });
+
+  it("blocks bulk retry before mutation while a selected document has fresh agent enrichment", async () => {
+    const document = {
+      id: documentId,
+      owner_id: userId,
+      title: "Bulk Agent-Enriched Protocol",
+      file_name: "bulk-agent-enriched.pdf",
+      source_path: null,
+      import_batch_id: null,
       status: "failed",
-      error_message: "older failure",
+      error_message: "earlier failure",
       page_count: 3,
       chunk_count: 8,
       image_count: 1,
+      metadata: {},
+    };
+    const now = new Date().toISOString();
+    const client = createSupabaseMock((call) => {
+      if (call.table === "documents" && call.operation === "select") return ok([document]);
+      if (call.table === "import_batches") return ok([]);
+      if (call.table === "ingestion_jobs" && call.operation === "select") return ok([]);
+      if (call.table === "indexing_v3_agent_jobs") {
+        return ok([{ document_id: documentId, status: "processing", locked_at: now, updated_at: now }]);
+      }
+      return ok([]);
     });
-    // Rollback fence: same atomic-conditional guard as the single-document
-    // reindex route — the rollback matches on the stamp this request wrote.
-    const fence = (documentUpdates[0]?.updatePayload as { updated_at?: string }).updated_at;
-    expect(documentUpdates[1]?.filters).toContainEqual({ column: "updated_at", value: fence });
+    mockRuntime(client, { invalidateRagCachesForOwner: vi.fn() });
+    const { POST } = await import("../src/app/api/documents/bulk/reindex/route");
+
+    const response = await POST(
+      authenticatedRequest("/api/documents/bulk/reindex", {
+        method: "POST",
+        body: JSON.stringify({ documentIds: [documentId], mode: "retry_failed" }),
+      }),
+    );
+
+    expect(response.status).toBe(409);
+    expect(await payload(response)).toMatchObject({
+      error: "Bulk reindex is paused while enrichment is active for one or more selected documents.",
+    });
+    expect(client.calls.some((call) => call.table === "documents" && call.operation === "update")).toBe(false);
+    expect(client.calls.some((call) => call.table === "ingestion_jobs" && call.operation === "insert")).toBe(false);
   });
 
   it("returns a successful partial-result response when deletion wins one bulk reindex race", async () => {
@@ -2593,15 +2717,15 @@ describe("private document API access", () => {
       if (call.table === "documents" && call.operation === "select") return ok([deletedDocument, queuedDocument]);
       if (call.table === "import_batches") return ok([]);
       if (call.table === "ingestion_jobs" && call.operation === "select") return ok([]);
-      if (call.table === "documents" && call.operation === "update") return ok([]);
-      if (call.table === "ingestion_jobs" && call.operation === "insert") {
-        const inserted = call.insertPayload as { document_id?: string };
-        return inserted.document_id === documentId
-          ? fail("ingestion_jobs_document_id_fkey", "23503")
-          : ok({ id: "surviving-job" });
-      }
       return ok([]);
     });
+    mockAtomicReindexRpc(client, (args) =>
+      ok(
+        args.p_document_id === documentId
+          ? { outcome: "not_found" }
+          : { outcome: "queued", job: { id: "surviving-job" } },
+      ),
+    );
     mockRuntime(client, { invalidateRagCachesForOwner: vi.fn() });
     const { POST } = await import("../src/app/api/documents/bulk/reindex/route");
 
@@ -2629,7 +2753,7 @@ describe("private document API access", () => {
       ],
       missingDocumentIds: [],
     });
-    expect(client.calls.filter((call) => call.table === "documents" && call.operation === "update")).toHaveLength(2);
+    expect(client.calls.some((call) => call.operation === "update" || call.operation === "insert")).toBe(false);
   });
 
   it("returns missing document ids as a completed partial batch", async () => {
@@ -2651,10 +2775,9 @@ describe("private document API access", () => {
       if (call.table === "documents" && call.operation === "select") return ok([document]);
       if (call.table === "import_batches") return ok([]);
       if (call.table === "ingestion_jobs" && call.operation === "select") return ok([]);
-      if (call.table === "documents" && call.operation === "update") return ok([]);
-      if (call.table === "ingestion_jobs" && call.operation === "insert") return ok({ id: "available-job" });
       return ok([]);
     });
+    mockAtomicReindexRpc(client, () => ok({ outcome: "queued", job: { id: "available-job" } }));
     mockRuntime(client, { invalidateRagCachesForOwner: vi.fn() });
     const { POST } = await import("../src/app/api/documents/bulk/reindex/route");
 
@@ -2673,7 +2796,7 @@ describe("private document API access", () => {
     });
   });
 
-  it("skips bulk rollback when a competing active job appears after the safety check", async () => {
+  it("reports an atomic bulk conflict when active ingestion appears after the safety check", async () => {
     const document = {
       id: documentId,
       owner_id: userId,
@@ -2691,14 +2814,10 @@ describe("private document API access", () => {
     const client = createSupabaseMock((call) => {
       if (call.table === "documents" && call.operation === "select") return ok([document]);
       if (call.table === "import_batches") return ok([]);
-      if (call.table === "ingestion_jobs" && call.operation === "select" && call.limitCount === 1) {
-        return ok([{ id: "competing-job" }]);
-      }
       if (call.table === "ingestion_jobs" && call.operation === "select") return ok([]);
-      if (call.table === "documents" && call.operation === "update") return ok([]);
-      if (call.table === "ingestion_jobs" && call.operation === "insert") return fail("bulk job insert failed");
       return ok([]);
     });
+    mockAtomicReindexRpc(client, () => ok({ outcome: "ingestion_active" }));
     mockRuntime(client, { invalidateRagCachesForOwner: vi.fn() });
     const { POST } = await import("../src/app/api/documents/bulk/reindex/route");
 
@@ -2709,7 +2828,6 @@ describe("private document API access", () => {
       }),
     );
     const body = await payload(response);
-    const documentUpdates = client.calls.filter((call) => call.table === "documents" && call.operation === "update");
 
     expect(response.status).toBe(200);
     expect(body).toMatchObject({
@@ -2719,19 +2837,11 @@ describe("private document API access", () => {
           documentId,
           mode: "full",
           ok: false,
-          error: "bulk job insert failed",
+          error: "Document already has pending or processing indexing work.",
         },
       ],
     });
-    expect(documentUpdates).toHaveLength(1);
-    expect(documentUpdates[0]?.updatePayload).toEqual({
-      status: "queued",
-      error_message: null,
-      page_count: 0,
-      chunk_count: 0,
-      image_count: 0,
-      updated_at: expect.any(String),
-    });
+    expect(client.calls.some((call) => call.operation === "update" || call.operation === "insert")).toBe(false);
   });
 
   it("cleans up uploaded storage when document insert fails", async () => {
diff --git a/tests/supabase-schema.test.ts b/tests/supabase-schema.test.ts
index cfd6d03d1..0a8a57c3d 100644
--- a/tests/supabase-schema.test.ts
+++ b/tests/supabase-schema.test.ts
@@ -30,6 +30,10 @@ const documentChangeWebhookMigration = readFileSync(
   new URL("../supabase/migrations/20260723150000_document_change_ingestion_webhook.sql", import.meta.url),
   "utf8",
 ).replace(/\s+/g, " ");
+const atomicReindexAgentGuardMigration = readFileSync(
+  new URL("../supabase/migrations/20260724060000_atomic_reindex_agent_guard.sql", import.meta.url),
+  "utf8",
+).replace(/\s+/g, " ");
 const dropStageJobIdFkMigration = readFileSync(
   new URL("../supabase/migrations/20260708140000_drop_ingestion_job_stages_job_id_fk.sql", import.meta.url),
   "utf8",
@@ -562,6 +566,37 @@ describe("Supabase schema Data API grants", () => {
     }
   });
 
+  it("serializes agent claims with transactional reindex enqueue", () => {
+    for (const sql of [schema, atomicReindexAgentGuardMigration]) {
+      const claimStart = sql.indexOf("create or replace function public.claim_indexing_v3_agent_jobs(");
+      const claimBody = sql.slice(claimStart, sql.indexOf("$$;", claimStart));
+      const reindexStart = sql.indexOf("create or replace function public.request_ingestion_reindex_if_agent_idle(");
+      const reindexBody = sql.slice(reindexStart, sql.indexOf("$$;", reindexStart));
+
+      expect(claimStart).toBeGreaterThanOrEqual(0);
+      expect(claimBody).toContain("on conflict do nothing");
+      expect(claimBody).toContain("for update of j, d skip locked");
+      expect(reindexStart).toBeGreaterThanOrEqual(0);
+      expect(reindexBody).toContain("and d.owner_id = p_owner_id for update;");
+      expect(reindexBody.indexOf("for update;")).toBeLessThan(
+        reindexBody.indexOf("from public.indexing_v3_agent_jobs a"),
+      );
+      expect(reindexBody.indexOf("from public.indexing_v3_agent_jobs a")).toBeLessThan(
+        reindexBody.indexOf("update public.documents"),
+      );
+      expect(reindexBody.indexOf("update public.documents")).toBeLessThan(
+        reindexBody.indexOf("insert into public.ingestion_jobs"),
+      );
+      expect(reindexBody).toContain("exception when unique_violation then");
+      expect(sql).toContain(
+        "revoke all on function public.request_ingestion_reindex_if_agent_idle(uuid, uuid, timestamptz, integer) from public, anon, authenticated;",
+      );
+      expect(sql).toContain(
+        "grant execute on function public.request_ingestion_reindex_if_agent_idle(uuid, uuid, timestamptz, integer) to service_role;",
+      );
+    }
+  });
+
   it("drops the stale duplicate ingestion_job_stages document index", () => {
     for (const sql of [schema, dropDuplicateStageIndexMigration]) {
       expect(sql).toContain("drop index if exists public.ingestion_job_stages_doc_idx");