audit fixes (P14): stop one failure stranding a whole document or job batch in the worker

eval/docling/harness/legacy-tables.ts

@@ -0,0 +1,32 @@
+/**
+ * Table projection for the Docling lab's legacy-extractor runner.
+ *
+ * Extracted from run-legacy.ts so the key contract with
+ * worker/python/extract_pdf_assets.py can be tested without executing the
+ * runner's `main()`.
+ */
+import type { ExtractedImage } from "../../../src/lib/types";
+
+export type LegacyTable = {
+  markdown: string | null;
+  rows: number | null;
+  cols: number | null;
+};
+
+export function tableFromImage(image: ExtractedImage): LegacyTable | null {
+  if (image.sourceKind !== "table_crop") return null;
+  const metadata = image.metadata ?? {};
+  const markdown = metadata["accessible_table_markdown"];
+  // Audit L117: `table_rows` / `table_columns` are ARRAYS in
+  // worker/python/extract_pdf_assets.py (a list of row lists and a list of
+  // header cells). The counts live in `row_count` / `column_count`; reading the
+  // array keys through a `typeof === "number"` filter made both fields
+  // permanently null in out/raw.
+  const rows = metadata["row_count"];
+  const cols = metadata["column_count"];
+  return {
+    markdown: typeof markdown === "string" ? markdown : null,
+    rows: typeof rows === "number" ? rows : null,
+    cols: typeof cols === "number" ? cols : null,
+  };
+}

eval/docling/harness/run-legacy.ts

@@ -16,29 +16,11 @@
 import { readFile, writeFile, rm } from "node:fs/promises";
 import path from "node:path";
 import { extractDocument } from "../../../src/lib/extractors/document";
-import type { ExtractedDocument, ExtractedImage } from "../../../src/lib/types";
+import type { ExtractedDocument } from "../../../src/lib/types";
+import { tableFromImage, type LegacyTable } from "./legacy-tables";
 
 const TEXT_CAP_BYTES = Number(process.env.LAB_PER_DOC_TEXT_BYTES ?? 64 * 1024 * 1024);
 
-type LegacyTable = {
-  markdown: string | null;
-  rows: number | null;
-  cols: number | null;
-};
-
-function tableFromImage(image: ExtractedImage): LegacyTable | null {
-  if (image.sourceKind !== "table_crop") return null;
-  const metadata = image.metadata ?? {};
-  const markdown = metadata["accessible_table_markdown"];
-  const rows = metadata["table_rows"];
-  const cols = metadata["table_columns"];
-  return {
-    markdown: typeof markdown === "string" ? markdown : null,
-    rows: typeof rows === "number" ? rows : null,
-    cols: typeof cols === "number" ? cols : null,
-  };
-}
-
 function argValue(flag: string): string {
   const index = process.argv.indexOf(flag);
   const value = index === -1 ? undefined : process.argv[index + 1];

src/lib/rag/rag-cache.ts

@@ -32,6 +32,24 @@ const ragCacheDependencyVersion = "rag-cache-v21";
 const cacheIndexingVersionTtlMs = 5000;
 const cacheIndexingVersionMaxEntries = 512;
 const cacheIndexingVersionCache = new Map<string, { expiresAt: number; value: string }>();
+/**
+ * The two corpus-staleness stamps that carry no information about the corpus.
+ *
+ * Audit L21: a PostgREST `error` on the `documents` stamp query used to be
+ * folded into INDEXING_STAMP_EMPTY_CORPUS, making a failed read indistinguishable
+ * from "nothing is indexed". Both constants are stable across requests, so two
+ * stamp failures bracketing a worker-side reindex (which cannot call this
+ * process's in-process invalidation) let a pre-reindex answer be served with
+ * `answer_cache_hit` — the exact staleness this stamp exists to prevent. An
+ * error now produces the distinct UNAVAILABLE stamp, and neither read nor write
+ * trusts it.
+ */
+const INDEXING_STAMP_UNAVAILABLE = "index-stamp-unavailable";
+const INDEXING_STAMP_EMPTY_CORPUS = "no-indexed-documents";
+
+function isUnavailableIndexingStamp(indexingVersion: string) {
+  return indexingVersion.endsWith(`:${INDEXING_STAMP_UNAVAILABLE}`);
+}
 /**
  * Invalidation generations for deferred `setCachedAnswer` promotions. Review /
  * table-fact mutations do not change the documents indexing stamp, so the
@@ -215,6 +233,12 @@ export async function getCachedAnswer(
     return null;
   }
 
+  // LRU recency bump (audit L134): a Map preserves insertion order, so without
+  // the re-insert a repeatedly-read hot answer still ages out in insertion order
+  // and is evicted before answers nobody has asked for since.
+  answerCache.delete(key);
+  answerCache.set(key, cached);
+
   const answer = cloneAnswer(cached.answer);
   answer.routingReason = answer.routingReason ? `${answer.routingReason}; answer_cache_hit` : "answer_cache_hit";
   answer.latencyTimings = {
@@ -264,7 +288,13 @@ export async function setCachedAnswer(
   const indexingVersion = await cacheIndexingVersion(args, { forceRefresh: true });
   if (invalidationEpochChanged(args.ownerId, invalidationEpochAtStart)) return;
   if (options?.indexingVersionAtRetrievalStart && indexingVersion !== options.indexingVersionAtRetrievalStart) return;
+  // An unavailable stamp cannot vouch for the corpus, so a write under it would
+  // be matched later by another failed read and served as fresh (audit L21).
+  if (isUnavailableIndexingStamp(indexingVersion)) return;
   const key = scopedAnswerCacheKey(args);
+  // Delete before set so a refreshed answer moves to the most-recent position
+  // rather than keeping its original insertion slot (audit L134).
+  answerCache.delete(key);
   answerCache.set(key, {
     expiresAt: Date.now() + env.RAG_ANSWER_CACHE_TTL_MS,
     answer: cloneAnswer(answer),
@@ -420,7 +450,9 @@ export async function setCachedSearch(
   const indexingVersion = await cacheIndexingVersion(args, { forceRefresh: true });
   throwIfAborted(args.signal);
   if (options?.indexingVersionAtRetrievalStart && indexingVersion !== options.indexingVersionAtRetrievalStart) return;
+  if (isUnavailableIndexingStamp(indexingVersion)) return;
   const key = scopedSearchCacheKey(args, telemetry.query_class, queryVariants);
+  searchCache.delete(key);
   searchCache.set(key, {
     expiresAt: Date.now() + env.RAG_SEARCH_CACHE_TTL_MS,
     results: clonedResults,
@@ -485,7 +517,7 @@ export async function cacheIndexingVersion(
   const cached = readExpiringCacheEntry(cacheIndexingVersionCache, cacheKey);
   if (cached) return cached.value;
 
-  let value = `${ragDeepMemoryVersion}:index-stamp-unavailable`;
+  let value = `${ragDeepMemoryVersion}:${INDEXING_STAMP_UNAVAILABLE}`;
   try {
     const supabase = createAdminClient();
     const documentFilters = args.documentIds?.length ? args.documentIds : args.documentId ? [args.documentId] : null;
@@ -507,8 +539,11 @@ export async function cacheIndexingVersion(
     if (args.signal) query = query.abortSignal(args.signal);
     const { data, error } = await query;
     throwIfAborted(args.signal);
-    if (error || !data?.length) {
-      value = `${ragDeepMemoryVersion}:no-indexed-documents`;
+    if (error) {
+      // A failed read is NOT an empty corpus (audit L21).
+      value = `${ragDeepMemoryVersion}:${INDEXING_STAMP_UNAVAILABLE}`;
+    } else if (!data?.length) {
+      value = `${ragDeepMemoryVersion}:${INDEXING_STAMP_EMPTY_CORPUS}`;
     } else {
       const latest = data[0] as { id?: string; updated_at?: string | null; metadata?: unknown };
       const metadata = normalizeSourceMetadata(latest.metadata);
@@ -521,7 +556,7 @@ export async function cacheIndexingVersion(
     }
   } catch {
     if (args.signal?.aborted) throw abortReason(args.signal);
-    value = `${ragDeepMemoryVersion}:index-stamp-unavailable`;
+    value = `${ragDeepMemoryVersion}:${INDEXING_STAMP_UNAVAILABLE}`;
   }
   throwIfAborted(args.signal);
   writeBoundedExpiringCacheEntry(

src/lib/reindex-pipeline.ts

@@ -47,10 +47,25 @@ export function isAtomicReindexCandidate(document: { status?: string | null; met
   return document.status === "indexed";
 }
 
+/**
+ * Mirror of the SQL predicate `public.is_committed_document_generation`:
+ *
+ *   select row_generation is null or row_generation = document_generation;
+ *
+ * Audit L11: the second arm used to fail OPEN in TypeScript — a document whose
+ * `metadata.index_generation_id` was absent accepted rows from ANY generation,
+ * while the SQL comparison against a NULL document generation yields NULL and
+ * excludes them. During the first atomic reindex of a legacy (never-stamped)
+ * indexed document, that let the document viewer interleave staged,
+ * uncommitted chunks/pages/images/table facts with the live ones — duplicate or
+ * half-built evidence on screen that `search_document_chunks` would never
+ * return. A row that carries a generation is committed only when the document
+ * names that same generation.
+ */
 export function isCommittedGenerationMetadata(args: { rowMetadata?: unknown; committedGeneration?: string | null }) {
   const rowGeneration = committedIndexGeneration(args.rowMetadata);
   if (!rowGeneration) return true;
-  if (!args.committedGeneration) return true;
+  if (!args.committedGeneration) return false;
   return rowGeneration === args.committedGeneration;
 }
 

supabase/functions/indexing-v3-agent/behavior.ts

@@ -181,3 +181,95 @@ export function deferralDecision(args: {
     },
   };
 }
+
+/**
+ * Claim limit for the agent endpoint.
+ *
+ * `Number(url.searchParams.get("limit"))` alone lets `?limit=abc` reach the
+ * claim RPC as `NaN::integer`, which Postgres rejects with a cast error — a 500
+ * before a single job is claimed. Mirror the ingestion-worker function: finite
+ * check first, then truncate and clamp.
+ */
+export const AGENT_CLAIM_LIMIT_DEFAULT = 8;
+export const AGENT_CLAIM_LIMIT_MAX = 50;
+
+export function parseAgentClaimLimit(
+  raw: string | null | undefined,
+  fallback = AGENT_CLAIM_LIMIT_DEFAULT,
+  max = AGENT_CLAIM_LIMIT_MAX,
+): number {
+  if (raw === null || raw === undefined || raw.trim() === "") return fallback;
+  const parsed = Number(raw);
+  if (!Number.isFinite(parsed)) return fallback;
+  return Math.max(1, Math.min(max, Math.trunc(parsed)));
+}
+
+/** The endpoint claims and mutates jobs; only POST may reach it. */
+export function isAllowedAgentMethod(method: string): boolean {
+  return method === "POST";
+}
+
+export type ClaimedBatchResult<TJob> = {
+  processed: number;
+  deferred: number;
+  failed: number;
+  deferrals: Array<{ job: TJob; missing: string[] }>;
+  failures: Array<{ job: TJob; error: string; failure_record_error: string | null }>;
+};
+
+function errorMessage(error: unknown): string {
+  return error instanceof Error ? error.message : JSON.stringify(error);
+}
+
+/**
+ * Run one claimed batch, isolating BOTH the per-job work and the per-job
+ * failure recording.
+ *
+ * The bug this exists to prevent (audit L13): `markJobFailure` throws whenever
+ * the status RPC returns `ok:false` or the database errors. Called unguarded in
+ * the loop's catch, that throw escaped to the request-level catch and returned
+ * 500 with every not-yet-processed job in the batch still `processing` under
+ * its lock — invisible until the 45-minute stale reclaim, each one attempt
+ * closer to the terminal `failed` state that is never re-queued. One job's
+ * failure must never abandon its siblings, so the failure-recording call gets
+ * its own guard and the loop continues.
+ */
+export async function runClaimedJobBatch<TJob>(
+  jobs: readonly TJob[],
+  handlers: {
+    processJob: (job: TJob) => Promise<{ status: "completed" | "deferred"; missing: string[] }>;
+    markJobFailure: (job: TJob, message: string) => Promise<unknown>;
+  },
+): Promise<ClaimedBatchResult<TJob>> {
+  const result: ClaimedBatchResult<TJob> = {
+    processed: 0,
+    deferred: 0,
+    failed: 0,
+    deferrals: [],
+    failures: [],
+  };
+
+  for (const job of jobs) {
+    try {
+      const outcome = await handlers.processJob(job);
+      if (outcome.status === "completed") {
+        result.processed += 1;
+      } else {
+        result.deferred += 1;
+        result.deferrals.push({ job, missing: outcome.missing });
+      }
+    } catch (error) {
+      const message = errorMessage(error);
+      let failureRecordError: string | null = null;
+      try {
+        await handlers.markJobFailure(job, message);
+      } catch (recordError) {
+        failureRecordError = errorMessage(recordError);
+      }
+      result.failed += 1;
+      result.failures.push({ job, error: message, failure_record_error: failureRecordError });
+    }
+  }
+
+  return result;
+}

supabase/functions/indexing-v3-agent/index.ts

@@ -22,8 +22,11 @@ import {
   completionGateFromRow,
   agentFailureDecision,
   deferralDecision,
+  isAllowedAgentMethod,
   missingArtifactPlan,
+  parseAgentClaimLimit,
   parseJobStatusRpcResult,
+  runClaimedJobBatch,
   shouldRunVisualArtifacts,
   type CompletionGate,
   type CompletionGateRow,
@@ -1911,14 +1914,15 @@ async function processJob(job: ClaimedJob): Promise<{ status: "completed" | "def
 
 Deno.serve({ port: Number(Deno.env.get("PORT") ?? "8000") }, async (req: Request) => {
   try {
-    if (req.method !== "POST" && req.method !== "GET") {
-      return new Response("Method not allowed", { status: 405 });
+    // Claiming jobs mutates the queue, so a GET must not reach it (audit L13).
+    if (!isAllowedAgentMethod(req.method)) {
+      return new Response("Method not allowed", { status: 405, headers: { Allow: "POST" } });
     }
     const unauthorized = await authorizeRequest(req);
     if (unauthorized) return unauthorized;
 
     const url = new URL(req.url);
-    const limit = Math.max(1, Math.min(50, Number(url.searchParams.get("limit") ?? "8")));
+    const limit = parseAgentClaimLimit(url.searchParams.get("limit"));
     const workerId = `indexing-v3-agent-${crypto.randomUUID()}`;
 
     const claimSource = "documents";
@@ -1930,39 +1934,35 @@ Deno.serve({ port: Number(Deno.env.get("PORT") ?? "8000") }, async (req: Request
       return Response.json({ ok: true, claimed: 0, processed: 0, failed: 0 });
     }
 
-    let processed = 0;
-    let deferred = 0;
-    let failed = 0;
-    const failures: Array<{ job_id: string; document_id: string; error: string }> = [];
-    const deferrals: Array<{ job_id: string; document_id: string; missing: string[] }> = [];
-
-    for (const job of claimed) {
-      try {
-        const result = await processJob(job);
-        if (result.status === "completed") {
-          processed += 1;
-        } else {
-          deferred += 1;
-          deferrals.push({ job_id: job.id, document_id: job.document_id, missing: result.missing });
-        }
-      } catch (e) {
-        failed += 1;
-        const msg = e instanceof Error ? e.message : JSON.stringify(e);
-        failures.push({ job_id: job.id, document_id: job.document_id, error: msg });
-        await markJobFailure(job, msg);
-      }
-    }
+    // Both the per-job work and the per-job failure recording are isolated in
+    // here: markJobFailure throws on an ok:false status RPC or a DB error, and
+    // when that throw escaped the loop it returned 500 and left every
+    // not-yet-processed sibling `processing` under its lock until the
+    // 45-minute stale reclaim (audit L13).
+    const batch = await runClaimedJobBatch(claimed, {
+      processJob,
+      markJobFailure: (job, message) => markJobFailure(job, message),
+    });
 
     return Response.json({
       ok: true,
       worker: workerId,
       claim_source: claimSource,
       claimed: claimed.length,
-      processed,
-      deferred,
-      failed,
-      deferrals,
-      failures,
+      processed: batch.processed,
+      deferred: batch.deferred,
+      failed: batch.failed,
+      deferrals: batch.deferrals.map((entry) => ({
+        job_id: entry.job.id,
+        document_id: entry.job.document_id,
+        missing: entry.missing,
+      })),
+      failures: batch.failures.map((entry) => ({
+        job_id: entry.job.id,
+        document_id: entry.job.document_id,
+        error: entry.error,
+        failure_record_error: entry.failure_record_error,
+      })),
     });
   } catch (e) {
     const message = e instanceof Error ? e.message : JSON.stringify(e);

supabase/functions/ingestion-worker/index.ts

@@ -2,6 +2,7 @@ import "jsr:@supabase/functions-js/edge-runtime.d.ts";
 import postgres from "npm:postgres@3.4.7";
 
 import { hasServiceRoleAuthorization } from "./auth.ts";
+import { INGESTION_WORKER_RETIRED, retiredIngestionWorkerResponse } from "./retirement.ts";
 
 declare const Supabase: {
   ai: {
@@ -265,6 +266,11 @@ async function processJob(job: ClaimedJob, workerId: string): Promise<"completed
 
 Deno.serve(async (req: Request) => {
   try {
+    // Audit L24: this function performs no extraction, yet claims real jobs from
+    // the shared queue and stamps documents `indexed` on its way through the
+    // catch. Refuse before touching the queue; see ./retirement.ts.
+    if (INGESTION_WORKER_RETIRED) return retiredIngestionWorkerResponse();
+
     if (req.method !== "POST") {
       return new Response("Method not allowed", { status: 405, headers: { Allow: "POST" } });
     }