fix(rag,ingestion): harden generation isolation, cache fail-closed stamps, summary verification, and edge/worker alignment

docs/branch-review-records/5349150e82f255e42866975fa7aa5e9fe76601433ebc427c75226677c5bef854.record.md

@@ -0,0 +1 @@
+| 2026-10-17 | claude/rag-ingestion-hardening | 2a9d44e0015358e4e7d197e4cf15f83df0a9ce35 | RAG retrieval/summary/cache generation isolation and ingestion worker/edge hardening | resolved | typecheck:internal, eslint, check:diff-integrity, check:migration-immutability, check:owner-scope, check:rag:fixtures, eval:rag:offline (26 files/979 tests), eval:rag:adversarial:offline (25 tests), vitest (8 suites/254 tests) |

docs/branch-review-records/94a7d7908ea7ff58edaecb6ebd9fd41f72d422ad6d345190046cfd5ed2d4372d.record.md

@@ -0,0 +1 @@
+| 2026-10-17 | claude/rag-ingestion-hardening | b8278363b9e55ca218133ac52f6191a2edf60502 | RAG and Ingestion review follow-ups | Resolved PR review findings, repaired bot-applied regressions, and restored rag.ts maintainability budget | npm run typecheck:internal; node scripts/check-maintainability-budgets.mjs; npx eslint; npx vitest run (11 suites) |

src/app/api/documents/route.ts

@@ -17,6 +17,7 @@ import {
   redactNonOwnedDocumentFields,
   withOwnerReadScope,
 } from "@/lib/public-api-access";
+import { committedIndexGeneration, isCommittedGenerationMetadata } from "@/lib/reindex-pipeline";
 import { parseRequestQuery, queryBoolean, queryInteger } from "@/lib/validation/query";
 
 export const runtime = "nodejs";
@@ -35,6 +36,11 @@ const PUBLIC_DOCUMENT_LIST_COLUMNS = [
   "updated_at",
 ].join(",");
 
+// Public rows are read with `metadata` only so the hidden-label and committed-generation filters can
+// run server-side; `projectPublicFields` strips it (and anything else not in the public list) before
+// the response is built.
+const PUBLIC_DOCUMENT_QUERY_COLUMNS = `${PUBLIC_DOCUMENT_LIST_COLUMNS},metadata`;
+
 const DOCUMENT_LIST_COLUMNS = [
   "id",
   "owner_id",
@@ -68,6 +74,8 @@ const PUBLIC_LABEL_LIST_COLUMNS = [
   "updated_at",
 ].join(",");
 
+const PUBLIC_LABEL_QUERY_COLUMNS = `${PUBLIC_LABEL_LIST_COLUMNS},metadata`;
+
 const LABEL_LIST_COLUMNS = [
   "id",
   "document_id",
@@ -83,6 +91,8 @@ const LABEL_LIST_COLUMNS = [
 
 const PUBLIC_SUMMARY_LIST_COLUMNS = ["id", "document_id", "summary", "clinical_specifics", "generated_at"].join(",");
 
+const PUBLIC_SUMMARY_QUERY_COLUMNS = `${PUBLIC_SUMMARY_LIST_COLUMNS},metadata`;
+
 const SUMMARY_LIST_COLUMNS = [
   "id",
   "document_id",
@@ -180,7 +190,7 @@ export async function GET(request: Request) {
     }
 
     const effectiveIncludeMeta = access.authenticated ? includeMeta : false;
-    const listColumns = access.authenticated ? DOCUMENT_LIST_COLUMNS : PUBLIC_DOCUMENT_LIST_COLUMNS;
+    const listColumns = access.authenticated ? DOCUMENT_LIST_COLUMNS : PUBLIC_DOCUMENT_QUERY_COLUMNS;
     let query = withOwnerReadScope(supabase.from("documents").select(listColumns, { count: "exact" }), access.ownerId)
       .order("created_at", { ascending: false })
       .range(offset, offset + limit - 1);
@@ -210,13 +220,16 @@ export async function GET(request: Request) {
       .filter((document) => !ownedDocumentIds.has(document.id))
       .map((document) => document.id);
     const documents = rawDocuments.map((document) => redactNonOwnedDocumentFields(document, access.ownerId));
+    const responseDocuments = documents.map((document) =>
+      ownedDocumentIds.has(document.id) ? document : projectPublicFields(document, PUBLIC_DOCUMENT_LIST_COLUMNS),
+    );
     const documentIds = documents.map((document) => document.id);
     const indexing = indexingState(documents);
 
     const pagination = offsetPagination({ limit, offset, pageLength: documents.length, count });
 
     if (documentIds.length === 0 || !effectiveIncludeMeta) {
-      return documentsResponse({ documents, pagination }, indexing);
+      return documentsResponse({ documents: responseDocuments, pagination }, indexing);
     }
 
     const ownedIds = [...ownedDocumentIds];
@@ -234,7 +247,7 @@ export async function GET(request: Request) {
       publicLabelsPromises.push(
         supabase
           .from("document_labels")
-          .select(PUBLIC_LABEL_LIST_COLUMNS)
+          .select(PUBLIC_LABEL_QUERY_COLUMNS)
           .in("document_id", publicDocumentIds.slice(i, i + 100)),
       );
     }
@@ -244,16 +257,18 @@ export async function GET(request: Request) {
         supabase
           .from("document_summaries")
           .select(SUMMARY_LIST_COLUMNS)
-          .in("document_id", ownedIds.slice(i, i + 100)),
+          .in("document_id", ownedIds.slice(i, i + 100))
+          .order("generated_at", { ascending: false }),
       );
     }
     const publicSummariesPromises = [];
     for (let i = 0; i < publicDocumentIds.length; i += 100) {
       publicSummariesPromises.push(
         supabase
           .from("document_summaries")
-          .select(PUBLIC_SUMMARY_LIST_COLUMNS)
-          .in("document_id", publicDocumentIds.slice(i, i + 100)),
+          .select(PUBLIC_SUMMARY_QUERY_COLUMNS)
+          .in("document_id", publicDocumentIds.slice(i, i + 100))
+          .order("generated_at", { ascending: false }),
       );
     }
 
@@ -273,12 +288,35 @@ export async function GET(request: Request) {
       if (res.error) throw new Error(res.error.message);
     }
 
+    const committedGenerationByDocument = new Map<string, string | null>(
+      rawDocuments.map((doc) => [doc.id, committedIndexGeneration((doc as { metadata?: unknown }).metadata)]),
+    );
+    const isHiddenLabelRow = (row: Record<string, unknown>) => {
+      if (!row.metadata || typeof row.metadata !== "object" || Array.isArray(row.metadata)) return false;
+      const meta = row.metadata as Record<string, unknown>;
+      return (
+        meta.hidden === true ||
+        meta.hidden === "true" ||
+        meta.review_status === "hidden" ||
+        meta.label_review_status === "hidden"
+      );
+    };
+
     const labelsByDocument = new Map<string, unknown[]>();
     const labelRows = parseListRows(
       [...ownedLabelsResults.flatMap((res) => res.data ?? []), ...publicLabelsResults.flatMap((res) => res.data ?? [])],
       labelListRowSchema,
     );
     for (const label of labelRows) {
+      if (isHiddenLabelRow(label)) continue;
+      if (
+        !isCommittedGenerationMetadata({
+          rowMetadata: label.metadata,
+          committedGeneration: committedGenerationByDocument.get(label.document_id) ?? null,
+        })
+      ) {
+        continue;
+      }
       const existing = labelsByDocument.get(label.document_id) ?? [];
       existing.push(
         ownedDocumentIds.has(label.document_id) ? label : projectPublicFields(label, PUBLIC_LABEL_LIST_COLUMNS),
@@ -292,19 +330,36 @@ export async function GET(request: Request) {
       ],
       summaryListRowSchema,
     );
-    const summariesByDocument = new Map(
-      summaryRows.map((summary) => [
-        summary.document_id,
-        ownedDocumentIds.has(summary.document_id) ? summary : projectPublicFields(summary, PUBLIC_SUMMARY_LIST_COLUMNS),
-      ]),
-    );
+    const summariesByDocument = new Map<string, { row: unknown; generatedAt: string }>();
+    for (const summary of summaryRows) {
+      if (
+        !isCommittedGenerationMetadata({
+          rowMetadata: summary.metadata,
+          committedGeneration: committedGenerationByDocument.get(summary.document_id) ?? null,
+        })
+      ) {
+        continue;
+      }
+      const generatedAt = typeof summary.generated_at === "string" ? summary.generated_at : "";
+      const existing = summariesByDocument.get(summary.document_id);
+      if (!existing || (generatedAt && (!existing.generatedAt || generatedAt > existing.generatedAt))) {
+        summariesByDocument.set(summary.document_id, {
+          row: ownedDocumentIds.has(summary.document_id)
+            ? summary
+            : projectPublicFields(summary, PUBLIC_SUMMARY_LIST_COLUMNS),
+          generatedAt,
+        });
+      }
+    }
 
     return documentsResponse(
       {
         documents: documents.map((document) => ({
-          ...document,
+          ...(ownedDocumentIds.has(document.id)
+            ? document
+            : projectPublicFields(document, PUBLIC_DOCUMENT_LIST_COLUMNS)),
           labels: labelsByDocument.get(document.id) ?? [],
-          summary: summariesByDocument.get(document.id) ?? null,
+          summary: summariesByDocument.get(document.id)?.row ?? null,
         })),
         pagination,
       },

src/app/api/ingestion/quality/route.ts

@@ -214,13 +214,29 @@ function buildReviewItems(args: {
     const documentStages = stagesByDocument.get(document.id) ?? [];
     const documentPages = pagesByDocument.get(document.id) ?? [];
     const documentImages = imagesByDocument.get(document.id) ?? [];
-    const failedJob = documentJobs.find((job) => job.status === "failed");
+    const latestJob = documentJobs[0];
+    const failedJob =
+      latestJob?.status === "failed"
+        ? latestJob
+        : document.status === "failed"
+          ? documentJobs.find((job) => job.status === "failed")
+          : undefined;
     const ingestionJobIds = new Set(documentJobs.map((job) => job.id));
-    const failedOcrStage = documentStages.find(
-      (stage) =>
-        stage.stage_status === "failed" &&
-        includesAny(`${stage.stage_name ?? ""} ${stage.error_message ?? ""}`, ["ocr", "extract"]),
-    );
+    const latestStagesByName = new Map<string, (typeof documentStages)[number]>();
+    for (const stage of documentStages) {
+      const key = stage.stage_name ?? stage.id;
+      if (!latestStagesByName.has(key)) latestStagesByName.set(key, stage);
+    }
+    const failedOcrStage = Array.from(latestStagesByName.values()).find((stage) => {
+      if (stage.stage_status !== "failed") return false;
+      if (!includesAny(`${stage.stage_name ?? ""} ${stage.error_message ?? ""}`, ["ocr", "extract"])) return false;
+      if (latestJob?.status !== "completed") return true;
+      if (stage.job_id && ingestionJobIds.has(stage.job_id)) {
+        return stage.job_id === latestJob.id;
+      }
+      const stageTimestamp = stage.finished_at ?? stage.started_at;
+      return !latestJob.updated_at || !stageTimestamp || stageTimestamp >= latestJob.updated_at;
+    });
     const warnings = Array.isArray(metadata.extraction_warnings) ? metadata.extraction_warnings.map(String) : [];
     const ocrWarning = unique([...warnings, ...issues]).find((warning) => includesAny(warning, ["ocr", "scanned"]));
     const lowTextPages = documentPages.filter((page) => String(page.text ?? "").trim().length < 80);

src/lib/document-enrichment.ts

@@ -31,6 +31,7 @@ import type { SupabaseClient } from "@supabase/supabase-js";
 import { z } from "zod";
 import { classifyDocumentOrganization } from "@/lib/document-organization";
 import { env } from "@/lib/env";
+import { committedIndexGeneration, isCommittedGenerationMetadata } from "@/lib/reindex-pipeline";
 import {
   PUBLIC_OWNER_FILTER_SENTINEL,
   retrievalAccessScopeForArgs,
@@ -1074,10 +1075,21 @@ export async function fetchRelatedDocumentMetadata(args: {
     rpcData = [...merged.values()];
   }
 
+  const isHiddenLabel = (label: { metadata?: unknown }) => {
+    if (!label.metadata || typeof label.metadata !== "object" || Array.isArray(label.metadata)) return false;
+    const meta = label.metadata as Record<string, unknown>;
+    return (
+      meta.hidden === true ||
+      meta.hidden === "true" ||
+      meta.review_status === "hidden" ||
+      meta.label_review_status === "hidden"
+    );
+  };
+
   if (!rpcError) {
     return ((rpcData ?? []) as RelatedDocumentMetadataRow[]).map((row) => ({
       document_id: row.document_id,
-      labels: row.labels ?? [],
+      labels: (row.labels ?? []).filter((label) => !isHiddenLabel(label)),
       summary: row.summary ?? null,
     }));
   }
@@ -1088,8 +1100,11 @@ export async function fetchRelatedDocumentMetadata(args: {
     .in("document_id", args.documentIds);
   let summariesQuery = args.supabase
     .from("document_summaries")
-    .select("document_id,owner_id,summary")
+    .select("document_id,owner_id,summary,generated_at,metadata")
     .in("document_id", args.documentIds);
+  if (typeof (summariesQuery as { order?: unknown }).order === "function") {
+    summariesQuery = summariesQuery.order("generated_at", { ascending: false });
+  }
 
   if (accessScope.ownerId && accessScope.includePublic) {
     labelsQuery = labelsQuery.or(`owner_id.eq.${accessScope.ownerId},owner_id.is.null`);
@@ -1111,22 +1126,67 @@ export async function fetchRelatedDocumentMetadata(args: {
   args.signal?.throwIfAborted();
   const metadataError = labelsResult.error ?? summariesResult.error;
   if (metadataError) throw metadataError;
-  const labels = (labelsResult.data ?? []) as DocumentLabel[];
-  const summaries = (summariesResult.data ?? []) as Pick<DocumentSummary, "document_id" | "summary">[];
+  const rawLabels = (labelsResult.data ?? []) as DocumentLabel[];
+  const rawSummaries = (summariesResult.data ?? []) as Array<
+    Pick<DocumentSummary, "document_id" | "summary"> & { generated_at?: string | null; metadata?: unknown }
+  >;
+  const hasGenerationStampedMetadata =
+    rawLabels.some((label) => Boolean(committedIndexGeneration(label.metadata))) ||
+    rawSummaries.some((summary) => Boolean(committedIndexGeneration(summary.metadata)));
+  const committedGenerationByDocument = new Map<string, string | null>();
+  if (hasGenerationStampedMetadata) {
+    let documentsQuery = args.supabase.from("documents").select("id,metadata").in("id", args.documentIds);
+    if (accessScope.ownerId && accessScope.includePublic) {
+      documentsQuery = documentsQuery.or(`owner_id.eq.${accessScope.ownerId},owner_id.is.null`);
+    } else if (accessScope.ownerId) {
+      documentsQuery = documentsQuery.eq("owner_id", accessScope.ownerId);
+    } else {
+      documentsQuery = documentsQuery.is("owner_id", null);
+    }
+    if (args.signal) {
+      documentsQuery = documentsQuery.abortSignal(args.signal);
+    }
+    const documentsResult = await documentsQuery;
+    args.signal?.throwIfAborted();
+    if (documentsResult?.error) throw documentsResult.error;
+    for (const doc of (documentsResult?.data ?? []) as Array<{ id: string; metadata?: unknown }>) {
+      committedGenerationByDocument.set(doc.id, committedIndexGeneration(doc.metadata));
+    }
+  }
+  const labels = rawLabels.filter(
+    (label) =>
+      !isHiddenLabel(label) &&
+      isCommittedGenerationMetadata({
+        rowMetadata: label.metadata,
+        committedGeneration: committedGenerationByDocument.get(label.document_id) ?? null,
+      }),
+  );
+  const summaries = rawSummaries.filter((summary) =>
+    isCommittedGenerationMetadata({
+      rowMetadata: summary.metadata,
+      committedGeneration: committedGenerationByDocument.get(summary.document_id) ?? null,
+    }),
+  );
   const labelsByDocument = new Map<string, DocumentLabel[]>();
-  const summariesByDocument = new Map<string, string | null>();
+  const summariesByDocument = new Map<string, { summary: string | null; generatedAt: string }>();
 
   for (const label of labels) {
     const existing = labelsByDocument.get(label.document_id) ?? [];
     existing.push(label);
     labelsByDocument.set(label.document_id, existing);
   }
-  for (const summary of summaries) summariesByDocument.set(summary.document_id, summary.summary ?? null);
+  for (const summary of summaries) {
+    const generatedAt = typeof summary.generated_at === "string" ? summary.generated_at : "";
+    const existing = summariesByDocument.get(summary.document_id);
+    if (!existing || (generatedAt && (!existing.generatedAt || generatedAt > existing.generatedAt))) {
+      summariesByDocument.set(summary.document_id, { summary: summary.summary ?? null, generatedAt });
+    }
+  }
 
   return args.documentIds.map((documentId) => ({
     document_id: documentId,
     labels: labelsByDocument.get(documentId) ?? [],
-    summary: summariesByDocument.get(documentId) ?? null,
+    summary: summariesByDocument.get(documentId)?.summary ?? null,
   }));
 }
 

src/lib/env.ts

@@ -80,11 +80,11 @@ const envSchema = z
     // Railway deploy webhook -> chat forwarder. Railway only lets you configure a
     // target URL (no custom headers), so this secret travels as `?token=` in the
     // configured URL and is compared constant-time. Min 16 chars.
-    RAILWAY_WEBHOOK_SECRET: z.string().min(16).optional(),
+    RAILWAY_WEBHOOK_SECRET: z.preprocess(coerceBlankEnv, z.string().min(16).optional()),
     // Supabase Database Webhook -> ingestion enqueue. Supabase webhooks DO allow
     // custom headers, so this secret is sent as `Authorization: Bearer` (or the
     // `x-webhook-secret` header) and compared constant-time. Min 16 chars.
-    SUPABASE_INGESTION_WEBHOOK_SECRET: z.string().min(16).optional(),
+    SUPABASE_INGESTION_WEBHOOK_SECRET: z.preprocess(coerceBlankEnv, z.string().min(16).optional()),
     // Optional outbound chat destinations shared by every /api/webhooks/* forwarder
     // and the CI-failure GitHub workflow. Set either, both, or neither; a receiver
     // with no destination configured accepts the event and reports it undelivered.

src/lib/ingestion-enqueue.ts

@@ -73,7 +73,10 @@ export async function enqueueDocumentReindexJob(args: {
     .eq("id", document.id);
   // Owner scope is the app's single tenancy layer (RLS is bypassed by the
   // service-role client); scope the write to the owning row when we have it.
-  const { error: updateError } = await (ownerId ? queueUpdate.eq("owner_id", ownerId) : queueUpdate);
+  const applyOwnerFilter = <T extends { eq: (col: string, val: string) => T; is?: (col: string, val: null) => T }>(
+    query: T,
+  ) => (ownerId ? query.eq("owner_id", ownerId) : typeof query.is === "function" ? query.is("owner_id", null) : query);
+  const { error: updateError } = await applyOwnerFilter(queueUpdate);
   if (updateError) throw new Error(updateError.message);
 
   const { data: job, error: jobError } = await supabase
@@ -102,17 +105,24 @@ export async function enqueueDocumentReindexJob(args: {
   if (jobError.code === "23505") {
     const { data: competingJobs, error: competingJobsError } = await supabase
       .from("ingestion_jobs")
-      .select("id")
+      .select("id,status")
       .eq("document_id", document.id)
       .in("status", ["pending", "processing"])
       .limit(1);
-    if (!competingJobsError && (competingJobs?.length ?? 0) === 0) {
+    const hasCompetingProcessingJob =
+      !atomicReindex &&
+      Boolean(document.status) &&
+      (competingJobs ?? []).some((row) => (row as { status?: unknown }).status === "processing");
+    if (!competingJobsError && ((competingJobs?.length ?? 0) === 0 || hasCompetingProcessingJob)) {
+      const payloadToRestore = hasCompetingProcessingJob
+        ? { ...rollbackDocumentPayload, status: "processing", error_message: null }
+        : rollbackDocumentPayload;
       const rollback = supabase
         .from("documents")
-        .update(rollbackDocumentPayload)
+        .update(payloadToRestore)
         .eq("id", document.id)
         .eq("updated_at", rollbackFence);
-      await (ownerId ? rollback.eq("owner_id", ownerId) : rollback);
+      await applyOwnerFilter(rollback);
     }
     return { outcome: "already_active" };
   }