fix(ingestion): phase-3 concurrency fixes — R11, R15/R16, R22, R24d

src/app/api/documents/[id]/reindex/route.ts

@@ -7,6 +7,7 @@ import { upsertDocumentDeepMemory } from "@/lib/deep-memory";
 import { jsonError } from "@/lib/http";
 import {
  checkIngestionMutationSafety,
+ hasActiveAgentEnrichmentJob,
  ingestionMutationSafetyPayload,
  ingestionRollbackFenceStamp,
 } from "@/lib/ingestion-mutation-safety";
@@ -135,6 +136,26 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
  if (!safety.ok) return NextResponse.json(ingestionMutationSafetyPayload(safety), { status: safety.status });
 
  if (mode === "enrichment") {
+ // Audit R24d: route-mode enrichment and the enrichment agent both
+ // delete-then-insert the same artifact families with no shared lock. If
+ // the agent is mid-pass, the interleaved deletes can strand a
+ // "completed/good" document with zero enrichment artifacts (no repair
+ // path exists). Refuse to run while a live agent pass holds the document.
+ const agentBusy = await hasActiveAgentEnrichmentJob({
+ supabase,
+ documentId: id,
+ staleAfterMinutes: env.WORKER_STALE_AFTER_MINUTES,
+ });
+ if (agentBusy) {
+ return NextResponse.json(
+ {
+ error:
+ "The enrichment agent is currently processing this document. Wait for it to finish before re-running enrichment.",
+ },
+ { status: 409 },
+ );
+ }
+
  const [chunks, images] = await Promise.all([
  selectReindexRowsInPages<ReindexChunk>({
  supabase,

src/app/api/documents/[id]/route.ts

@@ -5,6 +5,7 @@ import { rateLimitJsonResponse } from "@/lib/api-rate-limit";
 import { getDemoDocumentPayload } from "@/lib/demo-data";
 import { env, isDemoMode } from "@/lib/env";
 import { jsonError, PublicApiError } from "@/lib/http";
+import { buildStorageCleanupJobUpdate } from "@/lib/ingestion";
 import { invalidateRagCachesForDocumentMutation } from "@/lib/rag";
 import { committedIndexGeneration, isCommittedGenerationMetadata } from "@/lib/reindex-pipeline";
 import { createAdminClient } from "@/lib/supabase/admin";
@@ -206,20 +207,21 @@ async function updateStorageCleanupJob(args: {
  status: "completed" | "failed";
  storageRemoved: number;
  warnings: string[];
+ // Audit R11: set on every DELETE abort path so the ledger row's storage paths
+ // are cleared — the document survives the abort, so the janitor must never see
+ // its live paths queued for removal.
+ aborted?: boolean;
 }) {
  const { error } = await args.supabase
  .from("storage_cleanup_jobs")
- .update({
- status: args.status,
- attempts: 1,
- storage_removed: args.storageRemoved,
- last_error: args.warnings.length ? args.warnings.join("; ") : null,
- completed_at: args.status === "completed" ? new Date().toISOString() : null,
- metadata: {
- operation: "permanent_document_delete",
- storage_warnings: args.warnings,
- },
- })
+ .update(
+ buildStorageCleanupJobUpdate({
+ status: args.status,
+ storageRemoved: args.storageRemoved,
+ warnings: args.warnings,
+ aborted: args.aborted,
+ }),
+ )
  .eq("id", args.cleanupJobId);
 
  return error ? storageWarningsFrom(error, "Cleanup ledger") : null;
@@ -562,6 +564,7 @@ export async function DELETE(request: Request, { params }: { params: Promise<{ i
  status: "failed",
  storageRemoved: 0,
  warnings: [message],
+ aborted: true,
  });
  throw new PublicApiError(ledgerWarning ? `${message}; ${ledgerWarning}` : message, 409);
  }
@@ -576,6 +579,7 @@ export async function DELETE(request: Request, { params }: { params: Promise<{ i
  status: "failed",
  storageRemoved: 0,
  warnings: [message],
+ aborted: true,
  });
  throw new Error(ledgerWarning ? `${message}; ${ledgerWarning}` : message);
  }
@@ -588,6 +592,7 @@ export async function DELETE(request: Request, { params }: { params: Promise<{ i
  status: "failed",
  storageRemoved: 0,
  warnings: [`Database delete: ${deleteError.message}`],
+ aborted: true,
  });
  throw new Error(ledgerWarning ? `${deleteError.message}; ${ledgerWarning}` : deleteError.message);
  }

src/app/api/ingestion/jobs/[id]/retry/route.ts

@@ -2,6 +2,7 @@ import { NextResponse } from "next/server";
 import { z } from "zod";
 import { env, isDemoMode } from "@/lib/env";
 import { jsonError } from "@/lib/http";
+import { ingestionJobRetryRejectionReason, retryDocumentQueueUpdate } from "@/lib/ingestion";
 import { ingestionRollbackFenceStamp } from "@/lib/ingestion-mutation-safety";
 import { createAdminClient } from "@/lib/supabase/admin";
 import { AuthenticationError, requireAuthenticatedUser, unauthorizedResponse } from "@/lib/supabase/auth";
@@ -25,7 +26,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
  const { data: job, error: jobError } = await supabase
  .from("ingestion_jobs")
  .select(
- "id,document_id,batch_id,status,stage,progress,error_message,attempt_count,max_attempts,locked_at,locked_by,next_run_at,completed_at,documents!inner(owner_id)",
+ "id,document_id,batch_id,status,stage,progress,error_message,attempt_count,max_attempts,locked_at,locked_by,next_run_at,completed_at,documents!inner(owner_id,status)",
  )
  .eq("id", id)
  .eq("documents.owner_id", user.id)
@@ -34,6 +35,14 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
  if (jobError) throw new Error(jobError.message);
  if (!job) return NextResponse.json({ error: "Ingestion job not found." }, { status: 404 });
 
+ // IDX-R16: refuse to retry a job that already completed. Resetting a
+ // completed job re-pends terminal work; a worker then re-claims it and
+ // rebuilds against the live index (zombie re-ingest). Rebuild via reindex.
+ const retryRejection = ingestionJobRetryRejectionReason(job.status);
+ if (retryRejection) {
+ return NextResponse.json({ error: retryRejection }, { status: 409 });
+ }
+
  // IDX-C3 / B6: refuse to retry a job a live worker still holds, atomically.
  // A SELECT-then-UPDATE was a TOCTOU race: a worker could claim the job
  // between the read and the write, and the unguarded UPDATE would silently
@@ -88,9 +97,21 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
  // only re-queue; the prior index stays live until the worker commits a fresh one.
  // updated_at is stamped so the reindex routes' rollback fences (which
  // match on documents.updated_at) can see this competing queue-state write.
+ //
+ // IDX-R15: never demote an already-indexed document to `queued`. That write
+ // alone (independent of resetDocumentIndex above) forces the worker's
+ // destructive non-atomic path, which deletes the live index at job start.
+ // retryDocumentQueueUpdate keeps indexed documents on the atomic path.
+ const documentRow = (Array.isArray(job.documents) ? job.documents[0] : job.documents) as
+ { status?: string | null } | null | undefined;
  const { error: documentError } = await supabase
  .from("documents")
- .update({ status: "queued", error_message: null, updated_at: ingestionRollbackFenceStamp() })
+ .update(
+ retryDocumentQueueUpdate({
+ documentStatus: documentRow?.status ?? null,
+ fenceStamp: ingestionRollbackFenceStamp(),
+ }),
+ )
  .eq("id", job.document_id)
  .eq("owner_id", user.id);
  if (documentError) {

src/lib/ingestion-mutation-safety.ts

@@ -1,8 +1,16 @@
+import type { SupabaseClient } from "@supabase/supabase-js";
 import type { createAdminClient } from "@/lib/supabase/admin";
 import { probeSupabaseHealth } from "@/lib/supabase/health";
 
 type SupabaseAdminClient = ReturnType<typeof createAdminClient>;
 
+type AgentEnrichmentJobRow = {
+ document_id: string | null;
+ status: string | null;
+ locked_at: string | null;
+ updated_at: string | null;
+};
+
 type IngestionJobStatus = "pending" | "processing" | "failed" | string;
 
 type IngestionJobRow = {
@@ -77,6 +85,53 @@ export function ingestionRollbackFenceStamp(now = new Date()) {
  return now.toISOString().replace("Z", `${microseconds}Z`);
 }
 
+// Audit R24d: route-mode enrichment (reindex `mode:'enrichment'`) and the
+// enrichment agent (edge function) both delete-then-insert the same artifact
+// families. `checkIngestionMutationSafety` only reads `ingestion_jobs`, so it
+// cannot see a live agent pass — route enrichment runs freely against one and
+// the interleaved deletes can leave a "completed/good" document with ZERO
+// enrichment artifacts. This predicate flags a genuinely-live agent pass: a
+// `processing` agent job whose lease is still fresh. The agent lease has no
+// heartbeat, so a lock older than the stale threshold (or missing timestamps on
+// an abandoned row) is treated as dead and does NOT block. Steady-state
+// `pending`/`completed` agent jobs never block (they are not processing).
+export function isActiveAgentEnrichmentJob(
+ job: { status: string | null; locked_at: string | null; updated_at: string | null },
+ staleAfterMinutes: number,
+ nowMs: number,
+): boolean {
+ if (job.status !== "processing") return false;
+ const lockedAge = minutesAgo(job.locked_at, nowMs);
+ const updatedAge = minutesAgo(job.updated_at, nowMs);
+ const age = lockedAge ?? updatedAge;
+ if (age === null) return true; // just claimed / no timestamps → treat as live
+ return age < staleAfterMinutes; // fresh lease → live; stale lease → dead
+}
+
+export async function hasActiveAgentEnrichmentJob(args: {
+ supabase: SupabaseAdminClient;
+ documentId: string;
+ staleAfterMinutes: number;
+ now?: Date;
+}): Promise<boolean> {
+ // indexing_v3_agent_jobs is not in the generated Database types (it is a
+ // worker-state table added by migration), so query it through an untyped
+ // client the same way the reindex route paginates dynamic tables.
+ const client = args.supabase as unknown as SupabaseClient;
+ const { data, error } = await client
+ .from("indexing_v3_agent_jobs")
+ .select("document_id,status,locked_at,updated_at")
+ .eq("document_id", args.documentId)
+ .eq("status", "processing")
+ .limit(1);
+ if (error) throw new Error(error.message);
+
+ const nowMs = (args.now ?? new Date()).getTime();
+ return ((data ?? []) as AgentEnrichmentJobRow[]).some((job) =>
+ isActiveAgentEnrichmentJob(job, args.staleAfterMinutes, nowMs),
+ );
+}
+
 export async function checkIngestionMutationSafety(args: {
  supabase: SupabaseAdminClient;
  documentIds: string[];

src/lib/ingestion-recovery.ts

@@ -60,6 +60,15 @@ export function buildIngestionRecoveryPlan(args: {
  isRecoverableProcessingJob(job, now, args.staleAfterMinutes);
 
  if (isIndexedDocument && job.status !== "completed") {
+ // Audit R22: a `pending` job on an already-indexed document is a
+ // legitimately-queued reindex, not an abandoned leftover. Superseding it
+ // silently cancels the reindex ("completed / superseded by successful
+ // index"); routing it through the retry branch below would reset the live
+ // index (R19). Leave it untouched for the worker's atomic reindex path,
+ // which keeps the old generation live until the new commit swaps it.
+ if (job.status === "pending") {
+ continue;
+ }
  actions.push({ action: "supersede", jobId: job.id, documentId: job.document_id });
  continue;
  }

src/lib/ingestion.ts

@@ -29,3 +29,81 @@ export function terminalBatchStatus(args: { queued: number; processing: number;
  if (args.queued > 0 || args.processing > 0) return "processing";
  return args.failed > 0 ? "completed_with_errors" : "completed";
 }
+
+// Audit R16: retrying a `completed` ingestion job resurrects a terminal row
+// into a zombie re-ingest — a worker can re-claim it and interleave a fresh
+// build against the live committed index. Completed work is re-run via the
+// reindex route (which enqueues a NEW job); retry is only for jobs that have
+// not completed. Returns a rejection message, or null when the retry may run.
+export function ingestionJobRetryRejectionReason(status: string | null | undefined): string | null {
+ if (status === "completed") {
+ return "This ingestion job already completed. Reindex the document to rebuild it instead of retrying a completed job.";
+ }
+ return null;
+}
+
+// Audit R15/R16: the retry route must NOT demote an already-indexed document to
+// `queued`. A queued document takes the worker's non-atomic path, which runs
+// reset_document_index at job start and deletes the entire live committed index
+// hours before any replacement commit — so a transient failure of a healthy
+// indexed document would otherwise destroy its clinical index. Indexed
+// documents keep their status (atomic reindex: the old index stays live until
+// the new commit swaps the generation); only non-indexed documents are
+// re-queued. Either way we clear error_message and stamp the rollback fence.
+export function retryDocumentQueueUpdate(args: { documentStatus: string | null | undefined; fenceStamp: string }): {
+ status?: "queued";
+ error_message: null;
+ updated_at: string;
+} {
+ const base = { error_message: null as null, updated_at: args.fenceStamp };
+ if (args.documentStatus === "indexed") {
+ return base;
+ }
+ return { status: "queued", ...base };
+}
+
+export type StorageCleanupJobUpdate = {
+ status: "completed" | "failed";
+ attempts: number;
+ storage_removed: number;
+ last_error: string | null;
+ completed_at: string | null;
+ metadata: { operation: string; storage_warnings: string[] };
+ document_paths?: string[];
+ image_paths?: string[];
+};
+
+// Audit R11: the DELETE route creates the storage_cleanup_jobs ledger row (with
+// the LIVE document's source-PDF + image paths) before the point of no return.
+// If the delete then aborts — late re-check 409, trace-cleanup failure, or the
+// DB delete failing — the document is still alive, but the ledger row keeps its
+// populated paths and only its status flips to `failed`. The storage janitor
+// (scripts/cleanup-storage.ts) drains rows in status ('pending','failed') and
+// never checks the document still exists, so one routine janitor run then
+// permanently deletes a live document's PDF and images. Clearing the paths on
+// every abort path defuses the ledger row: the janitor may still pick it up but
+// has nothing to remove. Successful cleanup keeps its paths for auditability.
+export function buildStorageCleanupJobUpdate(args: {
+ status: "completed" | "failed";
+ storageRemoved: number;
+ warnings: string[];
+ aborted?: boolean;
+ now?: Date;
+}): StorageCleanupJobUpdate {
+ const update: StorageCleanupJobUpdate = {
+ status: args.status,
+ attempts: 1,
+ storage_removed: args.storageRemoved,
+ last_error: args.warnings.length ? args.warnings.join("; ") : null,
+ completed_at: args.status === "completed" ? (args.now ?? new Date()).toISOString() : null,
+ metadata: {
+ operation: "permanent_document_delete",
+ storage_warnings: args.warnings,
+ },
+ };
+ if (args.aborted) {
+ update.document_paths = [];
+ update.image_paths = [];
+ }
+ return update;
+}

tests/ingestion-mutation-safety.test.ts

@@ -0,0 +1,62 @@
+import { describe, expect, it } from "vitest";
+import { isActiveAgentEnrichmentJob } from "../src/lib/ingestion-mutation-safety";
+
+describe("active enrichment-agent detection (R24d)", () => {
+ const now = Date.parse("2026-07-07T00:00:00.000Z");
+ const staleAfterMinutes = 45;
+
+ it("treats a freshly-locked processing agent job as live (blocks route enrichment)", () => {
+ expect(
+ isActiveAgentEnrichmentJob(
+ { status: "processing", locked_at: "2026-07-06T23:50:00.000Z", updated_at: "2026-07-06T23:50:00.000Z" },
+ staleAfterMinutes,
+ now,
+ ),
+ ).toBe(true);
+ });
+
+ it("treats a processing job with a stale lease as dead (does not block)", () => {
+ expect(
+ isActiveAgentEnrichmentJob(
+ { status: "processing", locked_at: "2026-07-06T22:00:00.000Z", updated_at: "2026-07-06T22:00:00.000Z" },
+ staleAfterMinutes,
+ now,
+ ),
+ ).toBe(false);
+ });
+
+ it("never blocks on steady-state pending or completed agent jobs", () => {
+ for (const status of ["pending", "completed", "failed", "needs_enrichment_artifacts"]) {
+ expect(
+ isActiveAgentEnrichmentJob(
+ { status, locked_at: "2026-07-06T23:59:00.000Z", updated_at: "2026-07-06T23:59:00.000Z" },
+ staleAfterMinutes,
+ now,
+ ),
+ ).toBe(false);
+ }
+ });
+
+ it("treats a just-claimed processing job with no timestamps as live (conservative block)", () => {
+ expect(
+ isActiveAgentEnrichmentJob({ status: "processing", locked_at: null, updated_at: null }, staleAfterMinutes, now),
+ ).toBe(true);
+ });
+
+ it("falls back to updated_at when the lock timestamp is missing", () => {
+ expect(
+ isActiveAgentEnrichmentJob(
+ { status: "processing", locked_at: null, updated_at: "2026-07-06T23:58:00.000Z" },
+ staleAfterMinutes,
+ now,
+ ),
+ ).toBe(true);
+ expect(
+ isActiveAgentEnrichmentJob(
+ { status: "processing", locked_at: null, updated_at: "2026-07-06T21:00:00.000Z" },
+ staleAfterMinutes,
+ now,
+ ),
+ ).toBe(false);
+ });
+});