fix: serialize reindex with active enrichment

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
 

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);