Ingestion hardening: Retry route and heartbeat (part a)

data/repo-awareness-snapshot.json

@@ -1,7 +1,7 @@
 {
   "version": "repo-awareness-snapshot-v3",
   "captured_revision": {
-    "committed_at": "2026-10-20"
+    "committed_at": "2026-10-21"
   },
   "routes": {
     "modes": [

docs/audit/tenancy-defense-in-depth-review.md

@@ -452,6 +452,7 @@ guard until it is reviewed. Entries live in `DERIVED_QUERY_INVENTORY` in
 | `src/app/api/images/[id]/signed-url/route.ts`     | `document_images`      | `GET`                                                     | parent-document-verified             | Image row fetched by id, then its parent document resolved through `withOwnerReadScope`; 404 and nothing signed unless the parent is visible.                                                                                                                                                                                                                                               |
 | `src/app/api/images/signed-urls/route.ts`         | `document_images`      | `POST`                                                    | parent-document-verified             | Batch form of the same pattern (`/api/documents/images/batch` is a bare re-export of this module); only images whose parent survived `withOwnerReadScope` are signed.                                                                                                                                                                                                                       |
 | `src/app/api/ingestion/jobs/route.ts`             | `ingestion_jobs`       | `GET` (×2)                                                | documents-inner-join                 | Page query and active-count query both join `documents!inner` and filter `.eq("documents.owner_id", user.id)` on the chain itself.                                                                                                                                                                                                                                                          |
+| `src/app/api/ingestion/jobs/[id]/retry/route.ts`  | `ingestion_jobs`       | `POST` (×2)                                               | documents-inner-join                 | Pre-recovery state read and post-recovery re-read both join `documents!inner` and filter `.eq("documents.owner_id", user.id)` on the chain.                                                                                                                                                                                                                                                 |
 | `src/app/api/ingestion/quality/route.ts`          | `ingestion_jobs`       | `GET`                                                     | owner-scoped-id-list (`documentIds`) | `.in("document_id", documentIds)` from the handler's `.eq("owner_id", user.id)` documents query.                                                                                                                                                                                                                                                                                            |
 | `src/app/api/ingestion/quality/route.ts`          | `ingestion_job_stages` | `GET`                                                     | owner-scoped-id-list (`documentIds`) | Same owner-scoped id list.                                                                                                                                                                                                                                                                                                                                                                  |
 | `src/app/api/ingestion/quality/route.ts`          | `document_pages`       | `GET`                                                     | owner-scoped-id-list (`documentIds`) | Same owner-scoped id list. This one reads page `text`, so it is the highest-value derived read in the handler.                                                                                                                                                                                                                                                                              |

docs/reindex-shadow-harness-design.md

@@ -6,6 +6,12 @@ The offline decision core (`src/lib/reindex-eval-gate.ts` — `decideReindexGate
 unit-tested; this doc specifies the remaining live pieces so they are turnkey when the billed
 OpenAI key + Supabase service secrets are available.
 
+**Until this harness is applied, the eval gate is advisory and by hand** (audit 2026-10-20 ING-A-04).
+Nothing calls `decideReindexGate`: the worker refuses only a zero-chunk, zero-image generation, so a
+worse re-extraction can replace a good live one unchecked. Before approving any library reload, run a
+baseline eval canary first, then after the reload run `npm run reindex:health` and a second canary,
+and compare the two. Revert on any recall drop or per-case regression.
+
 ## Goal
 
 Build the new (better-chunked, re-enriched) index for a document into a **staged** generation

docs/worker-deploy-runbook.md

@@ -725,6 +725,15 @@ The second Gate B caveat is already load-bearing above: docling's eager-mode lat
   node -v # must report v24.x
   ```
 
+- **Pausing the worker on purpose (audit EVAL-03):** the worker has no public
+  address, so monitoring cannot see that it is paused. Whenever you pause it
+  (`WORKER_PAUSED=true`, or a placeholder start command), also set the GitHub
+  repository variable `INGESTION_WORKER_PAUSED=true`; set it to `false` when the
+  worker runs again. The six-hourly ops digest then shows the worker as paused
+  rather than healthy, and the ingestion autopilot names the pause as the cause
+  when it alerts. A pause alone never alerts; a pending upload that nothing claims
+  for more than 30 minutes does, paused or not.
+
 ---
 
 ## Rollback

scripts/lib/tenancy-scan.mjs

@@ -1537,6 +1537,15 @@ export const DERIVED_QUERY_INVENTORY = [
     reason:
       'The page query, the active-count query and the failed-count query each join `documents!inner` and filter `.eq("documents.owner_id", user.id)`, so ownership rides on every one of the three query chains itself. 2 -> 3 on 2026-09-04 when #L15 added the pre-pagination failed-job head count, built as a copy of the active-count chain precisely so it would carry the same on-chain filter rather than need a weaker proof.',
   },
+  {
+    file: "src/app/api/ingestion/jobs/[id]/retry/route.ts",
+    table: "ingestion_jobs",
+    fn: "POST",
+    queries: 2,
+    proof: PROOF_KINDS.DOCUMENTS_INNER_JOIN,
+    reason:
+      'Admin Retry (audit 2026-10-20 DB-01): the pre-recovery state read and the post-recovery re-read both join `documents!inner` and filter `.eq("documents.owner_id", user.id)` on the chain itself.',
+  },
   {
     file: "src/app/api/ingestion/quality/route.ts",
     table: "ingestion_jobs",

scripts/reindex-health.ts

@@ -126,6 +126,35 @@ async function main() {
     .order("created_at", { ascending: true })
     .limit(25);
 
+  // Audit 2026-10-20 ING-A-02 / EVAL-09: the open-jobs preview above is the 25 oldest jobs, so a job
+  // stranded on its final attempt can sit behind old failed jobs unseen. Count it directly, along
+  // with indexed documents that have no searchable chunks. (Embeddings are NOT NULL, so no missing-embedding count.)
+  const nowIso = new Date().toISOString();
+  const integrityCounts = await Promise.all([
+    safeDerivedCount("jobs_stranded_final_attempt", async () => {
+      const { data, error } = await supabase
+        .from("ingestion_jobs")
+        .select("id,attempt_count,max_attempts")
+        .eq("status", "processing")
+        .lt("lease_expires_at", nowIso)
+        .limit(500);
+      if (error) throw new Error(error.message);
+      return (data ?? []).filter((job) => Number(job.attempt_count ?? 0) >= Number(job.max_attempts ?? 0));
+    }),
+    safeCount(
+      "documents_indexed_without_chunks",
+      supabase
+        .from("documents")
+        .select("id", { count: "exact", head: true })
+        .eq("status", "indexed")
+        .eq("chunk_count", 0),
+    ),
+    // document_chunks.embedding is NOT NULL, so a missing-embedding count would always be 0; not checked.
+  ]);
+  const integrityProblems = integrityCounts
+    .filter((item) => Number(item.count ?? 0) > 0)
+    .map((item) => `${item.label}: ${item.count}`);
+
   const counts = [
     indexed,
     queued,
@@ -137,7 +166,7 @@ async function main() {
     strandedQueued,
     chunksWithSynopsis,
   ];
-  const countErrors = counts.filter((item) => item.error);
+  const countErrors = [...counts, ...integrityCounts].filter((item) => item.error);
   const errors = [
     ...countErrors.map((item) => ({ label: item.label, error: unavailableMessage(item.error) ?? item.error })),
     ...(jobsError ? [{ label: "open_jobs", error: unavailableMessage(jobsError.message) ?? jobsError.message }] : []),
@@ -148,7 +177,8 @@ async function main() {
       {
         ok: countErrors.length === 0 && !jobsError,
         generatedAt: new Date().toISOString(),
-        counts: Object.fromEntries(counts.map((item) => [item.label, item.count])),
+        counts: Object.fromEntries([...counts, ...integrityCounts].map((item) => [item.label, item.count])),
+        integrityProblems,
         errors,
         openJobs: openJobs ?? [],
         recommendation:

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

@@ -240,7 +240,7 @@ export async function DELETE(request: Request, { params }: { params: Promise<{ i
     }
     if (result.outcome === "active_job") {
       throw new PublicApiError(
-        "Document has pending or processing indexing work. Stop or wait for the worker before deleting.",
+        "Document has pending or processing indexing work. Wait for it to finish, or recover a job stuck on its last attempt, before deleting.",
         409,
         { code: "document_indexing_active" },
       );

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

@@ -4,7 +4,11 @@ import { consumeApiRateLimit, rateLimitJsonResponse } from "@/lib/api-rate-limit
 import { env, isDemoMode } from "@/lib/env";
 import { jsonError, publicErrorResponse } from "@/lib/http";
 import { ingestionJobRetryRejectionReason } from "@/lib/ingestion";
-import { ingestionRollbackFenceStamp } from "@/lib/ingestion-mutation-safety";
+import {
+  applyIngestionRecoveryActions,
+  isFreshProcessingJob,
+  type IngestionRecoveryJob,
+} from "@/lib/ingestion-recovery";
 import { createAdminClient } from "@/lib/supabase/admin";
 import { AuthenticationError, requireAuthenticatedUser, unauthorizedResponse } from "@/lib/supabase/auth";
 import { parseRouteParams } from "@/lib/validation/params";
@@ -14,12 +18,28 @@ export const runtime = "nodejs";
 const ingestionRetryRouteParamsSchema = z.object({
   id: z.string().uuid(),
 });
-const ingestionRetryResultSchema = z.discriminatedUnion("outcome", [
-  z.object({ outcome: z.literal("not_found") }),
-  z.object({ outcome: z.literal("completed") }),
-  z.object({ outcome: z.literal("active_worker") }),
-  z.object({ outcome: z.literal("queued"), job: z.object({}).passthrough() }),
-]);
+
+const ACTIVE_JOB_MESSAGE =
+  "This job is still being processed by a worker. Wait for it to finish or go stale before retrying.";
+
+// Audit 2026-10-20 DB-01: Retry goes through recover_ingestion_job, the same hardened path the
+// operator recovery scripts use (document authority lock, managed-evidence hold, sibling-job and
+// live-lease checks), guarded on the row state read here. Each refusal is a 409 the panel shows.
+const RECOVERY_REFUSALS: Record<string, { message: string; code: string }> = {
+  active_lease: { message: ACTIVE_JOB_MESSAGE, code: "ingestion_job_active" },
+  state_changed: {
+    message: "This job or its document has changed since the list was loaded. Refresh and try again.",
+    code: "ingestion_job_state_changed",
+  },
+  managed_recovery_required: {
+    message: "This document has managed index evidence, so it needs operator recovery rather than Retry.",
+    code: "ingestion_job_managed_recovery_required",
+  },
+  authority_unavailable: {
+    message: "Another change to this document is in progress. Try again shortly.",
+    code: "ingestion_job_authority_unavailable",
+  },
+};
 
 export async function POST(request: Request, { params }: { params: Promise<{ id: string }> }) {
   try {
@@ -44,39 +64,75 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
       return rateLimitJsonResponse("Too many ingestion administration requests. Retry shortly.", rateLimit);
     }
 
-    const staleThreshold = new Date(Date.now() - env.WORKER_STALE_AFTER_MINUTES * 60_000).toISOString();
-    const resetNextRunAt = ingestionRollbackFenceStamp();
-    const { data, error } = await supabase.rpc("retry_ingestion_job_if_idle", {
-      p_job_id: id,
-      p_owner_id: user.id,
-      p_stale_before: staleThreshold,
-      p_max_attempts: env.WORKER_MAX_ATTEMPTS,
-      p_next_run_at: resetNextRunAt,
-      p_document_updated_at: ingestionRollbackFenceStamp(),
-    });
-    if (error) throw new Error(error.message);
-
-    const parsed = ingestionRetryResultSchema.safeParse(data);
-    if (!parsed.success) throw new Error("retry_ingestion_job_if_idle returned an invalid result.");
-    if (parsed.data.outcome === "not_found") {
+    const { data: row, error: readError } = await supabase
+      .from("ingestion_jobs")
+      .select(
+        "id,document_id,status,locked_at,lease_token,lease_expires_at,documents!inner(status,owner_id,chunk_count)",
+      )
+      .eq("id", id)
+      .eq("documents.owner_id", user.id)
+      .maybeSingle();
+    if (readError) throw new Error(readError.message);
+    const job = row as unknown as (IngestionRecoveryJob & { status: string | null }) | null;
+    if (!job || job.documents?.owner_id !== user.id) {
       return publicErrorResponse("Ingestion job not found.", 404, { code: "ingestion_job_not_found" });
     }
-    if (parsed.data.outcome === "completed") {
+    if (job.status === "completed") {
       return publicErrorResponse(
         ingestionJobRetryRejectionReason("completed") ?? "Completed ingestion jobs cannot be retried.",
         409,
         { code: "ingestion_job_completed" },
       );
     }
-    if (parsed.data.outcome === "active_worker") {
+    if (job.status === "pending") {
+      return publicErrorResponse("This job is already queued.", 409, { code: "ingestion_job_queued" });
+    }
+    if (isFreshProcessingJob(job, new Date(), env.WORKER_STALE_AFTER_MINUTES)) {
+      return publicErrorResponse(ACTIVE_JOB_MESSAGE, 409, { code: "ingestion_job_active" });
+    }
+    if (job.documents?.status === "indexed" && Number(job.documents.chunk_count ?? 0) > 0) {
       return publicErrorResponse(
-        "This job is still being processed by a worker. Wait for it to finish or go stale before retrying.",
+        "This document is already indexed, so its old job cannot be retried. Use Re-index to rebuild it.",
         409,
-        { code: "ingestion_job_active" },
+        { code: "ingestion_document_indexed" },
       );
     }
 
-    return NextResponse.json({ job: parsed.data.job });
+    const [result] = await applyIngestionRecoveryActions(supabase, [
+      {
+        action: { action: "retry", jobId: job.id, documentId: job.document_id, resetDocument: false },
+        expected: {
+          jobStatus: job.status ?? null,
+          jobLockedAt: job.locked_at ?? null,
+          jobLeaseToken: job.lease_token ?? null,
+          jobLeaseExpiresAt: job.lease_expires_at ?? null,
+          documentStatus: job.documents?.status ?? null,
+          documentOwnerId: job.documents?.owner_id ?? null,
+        },
+      },
+    ]);
+    if (!result?.applied) {
+      if (result?.reason === "job_closed") {
+        return publicErrorResponse("Ingestion job not found.", 404, { code: "ingestion_job_not_found" });
+      }
+      const refusal = RECOVERY_REFUSALS[result?.reason ?? ""];
+      if (refusal) return publicErrorResponse(refusal.message, 409, { code: refusal.code });
+      throw new Error(`recover_ingestion_job refused the retry: ${result?.reason ?? "no result"}.`);
+    }
+
+    const { data: queued, error: queuedError } = await supabase
+      .from("ingestion_jobs")
+      .select("*, documents!inner(owner_id)")
+      .eq("id", job.id)
+      .eq("documents.owner_id", user.id)
+      .maybeSingle();
+    if (queuedError) throw new Error(queuedError.message);
+    // The owner join is only the tenancy filter; keep the response shape the panel already reads.
+    const queuedJob = queued ? { ...(queued as Record<string, unknown>) } : null;
+    if (queuedJob) delete queuedJob.documents;
+    return NextResponse.json({
+      job: queuedJob ?? { id: job.id, document_id: job.document_id, status: "pending" },
+    });
   } catch (error) {
     if (error instanceof AuthenticationError) return unauthorizedResponse();
     return jsonError(error);