From 6a2dd937c4f09de37198338e5712932b454cca00 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 23 Jun 2026 04:39:37 +0000 Subject: [PATCH 1/7] Initial plan From d88c2694862ba22cc372f442bea4c459228bb8b1 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 23 Jun 2026 04:57:05 +0000 Subject: [PATCH 2/7] feat: simplify operational tooling and runbooks (#56) - recover-ingestion-queue.ts: replace JSON output with human-readable summary; add interactive confirmation prompt (--yes / --apply to skip); extract confirm() to shared cli-utils.ts - supabase-recovery-status.ts: replace JSON output with readable dashboard - scripts/reindex.ts: new consolidated pipeline command that encapsulates the full reindex runbook (health check, recovery, worker, repeat) - package.json: add 'reindex' npm script - docs/reindex-runbook.md: document the new npm run reindex command --- docs/reindex-runbook.md | 29 ++- package.json | 1 + scripts/cli-utils.ts | 15 ++ scripts/recover-ingestion-queue.ts | 70 ++++-- scripts/reindex.ts | 340 ++++++++++++++++++++++++++++ scripts/supabase-recovery-status.ts | 53 +++-- 6 files changed, 460 insertions(+), 48 deletions(-) create mode 100644 scripts/cli-utils.ts create mode 100644 scripts/reindex.ts diff --git a/docs/reindex-runbook.md b/docs/reindex-runbook.md index a3e8f5bb0..de09cbb16 100644 --- a/docs/reindex-runbook.md +++ b/docs/reindex-runbook.md @@ -2,13 +2,38 @@ Use this sequence when applying RAG indexing changes to the live `Clinical KB Database` Supabase project. -## Safe sequence +## Consolidated pipeline command + +The quickest way to run a full reindex cycle is the consolidated pipeline command: + +```sh +npm run reindex +``` + +This single command encapsulates the full safe sequence below: it confirms the target project, checks Supabase health, snapshots reindex health, applies ingestion queue recovery (with an interactive confirmation prompt), runs the worker, and repeats until the queue is clear or the round limit is reached. + +Options: + +| Flag | Default | Description | +|---|---|---| +| `--yes` | off | Skip confirmation prompts (non-interactive / CI use) | +| `--max-rounds` | 10 | Maximum worker iterations before stopping | +| `--limit` | 20 | Recovery action limit per round | + +```sh +# Non-interactive, up to 5 worker rounds: +npm run reindex -- --yes --max-rounds 5 +``` + +## Manual safe sequence + +If you prefer to run each step by hand (or need to apply a migration in between): 1. Confirm the target project is `sjrfecxgysukkwxsowpy`. 2. Run `npm run supabase:recovery-status`. 3. Apply any pending local migration only after the probe succeeds. 4. Run `npm run reindex:health`. -5. If stale or failed jobs exist, run `npm run recover:ingestion -- --apply --limit 20`. +5. If stale or failed jobs exist, run `npm run recover:ingestion` and confirm the prompt. 6. Run `npm run worker:once` with the conservative defaults from `.env.example`. 7. Repeat `npm run reindex:health` and `npm run worker:once` until the queue is clear. 8. Run indexing and RAG evals only after documents have adaptive chunks and retrieval synopses. diff --git a/package.json b/package.json index 04ca223b0..d3e2c34e0 100644 --- a/package.json +++ b/package.json @@ -37,6 +37,7 @@ "check:supabase-project": "tsx scripts/check-supabase-project.ts", "check:indexing": "tsx scripts/check-indexing.ts", "recover:ingestion": "tsx scripts/recover-ingestion-queue.ts", + "reindex": "tsx scripts/reindex.ts", "reindex:health": "tsx scripts/reindex-health.ts", "supabase:recovery-status": "tsx scripts/supabase-recovery-status.ts", "promote:query-misses": "tsx scripts/promote-query-misses.ts", diff --git a/scripts/cli-utils.ts b/scripts/cli-utils.ts new file mode 100644 index 000000000..d1862f0c9 --- /dev/null +++ b/scripts/cli-utils.ts @@ -0,0 +1,15 @@ +import { createInterface } from "node:readline"; + +/** + * Prompts the user with a yes/no question and returns their answer. + * Returns `false` if stdin is not a TTY (e.g. when piped). + */ +export function confirm(question: string): Promise { + return new Promise((resolve) => { + const rl = createInterface({ input: process.stdin, output: process.stdout }); + rl.question(`${question} (y/N) `, (answer: string) => { + rl.close(); + resolve(answer.trim().toLowerCase() === "y"); + }); + }); +} diff --git a/scripts/recover-ingestion-queue.ts b/scripts/recover-ingestion-queue.ts index 17d6c13ea..e6560c18d 100644 --- a/scripts/recover-ingestion-queue.ts +++ b/scripts/recover-ingestion-queue.ts @@ -1,5 +1,6 @@ import { loadEnvConfig } from "@next/env"; import type { IngestionRecoveryJob } from "@/lib/ingestion-recovery"; +import { confirm } from "./cli-utils"; loadEnvConfig(process.cwd()); @@ -9,6 +10,14 @@ type RecoveryDocument = { chunk_count?: number | null; }; +type RawJobRow = { + id: string; + document_id: string; + status: string | null; + locked_at: string | null; + documents: RecoveryDocument | RecoveryDocument[] | null; +}; + function supabaseStageError(stage: string, error: unknown) { const record = error && typeof error === "object" ? (error as Record) : {}; const message = @@ -35,6 +44,7 @@ function parseArgs(argv: string[]) { }; return { apply: argv.includes("--apply"), + yes: argv.includes("--yes"), staleAfterMinutes: Number.parseInt(valueFor("stale-after-minutes") ?? "", 10), limit: Number.parseInt(valueFor("limit") ?? "", 10), }; @@ -59,7 +69,12 @@ async function main() { : env.WORKER_STALE_AFTER_MINUTES; const limit = Number.isFinite(args.limit) ? args.limit : 20; const supabase = createAdminClient(); + + console.log("=== Ingestion Queue Recovery ==="); + console.log(`Checking Supabase health...`); assertSupabaseHealthy(await probeSupabaseHealth(supabase), "Ingestion queue recovery"); + console.log(" Supabase is healthy.\n"); + const { data, error } = await supabase .from("ingestion_jobs") .select("id,document_id,status,locked_at,documents(status,page_count,chunk_count)") @@ -68,7 +83,7 @@ async function main() { if (error) throw supabaseStageError("load open ingestion jobs", error); - const jobs = (data ?? []).map((job) => ({ + const jobs = (data ?? []).map((job: RawJobRow) => ({ ...job, documents: Array.isArray(job.documents) ? (job.documents[0] as RecoveryDocument | undefined) : job.documents, })) as IngestionRecoveryJob[]; @@ -77,29 +92,43 @@ async function main() { const resetDocumentIds = Array.from( new Set(actions.filter((action) => action.action === "retry").map((action) => action.documentId)), ); + const supersedeCount = actions.filter((action) => action.action === "supersede").length; + const retryCount = actions.filter((action) => action.action === "retry").length; + const remainingCount = Math.max(0, plan.actions.length - actions.length); - console.log( - JSON.stringify( - { - mode: args.apply ? "apply" : "dry-run", - staleAfterMinutes, - limit, - scannedJobs: jobs.length, - resetDocuments: resetDocumentIds.length, - supersedeJobs: actions.filter((action) => action.action === "supersede").length, - retryJobs: actions.filter((action) => action.action === "retry").length, - remainingActions: Math.max(0, plan.actions.length - actions.length), - }, - null, - 2, - ), - ); + console.log(`Stale-after threshold : ${staleAfterMinutes} min`); + console.log(`Action limit : ${limit}`); + console.log(`Scanned jobs : ${jobs.length}`); + console.log(`Documents to reset : ${resetDocumentIds.length}`); + console.log(`Jobs to supersede : ${supersedeCount}`); + console.log(`Jobs to retry : ${retryCount}`); + if (remainingCount > 0) { + console.log(`Remaining (over limit): ${remainingCount}`); + } - if (!args.apply) { - console.log("Dry run only. Re-run with --apply to mutate the ingestion queue."); + if (actions.length === 0) { + console.log("\nNothing to recover. Queue looks healthy."); return; } + let shouldApply = args.apply; + + if (!shouldApply) { + if (args.yes) { + shouldApply = true; + } else { + console.log(""); + shouldApply = await confirm("Apply these changes?"); + } + } + + if (!shouldApply) { + console.log("\nNo changes applied. Re-run with --apply or confirm interactively to mutate the ingestion queue."); + return; + } + + console.log("\nApplying recovery..."); + for (const documentId of resetDocumentIds) { const { error: resetError } = await supabase.rpc("reset_document_index", { p_document_id: documentId }); if (resetError) throw supabaseStageError("reset document index", resetError); @@ -146,6 +175,9 @@ async function main() { } console.log("Ingestion queue recovery applied."); + if (remainingCount > 0) { + console.log(`\n${remainingCount} action(s) remain over the limit. Re-run to process the next batch.`); + } } main().catch((error) => { diff --git a/scripts/reindex.ts b/scripts/reindex.ts new file mode 100644 index 000000000..987ea74a8 --- /dev/null +++ b/scripts/reindex.ts @@ -0,0 +1,340 @@ +/** + * Consolidated reindex pipeline. + * + * Runs the full safe reindex sequence from the runbook in a single command: + * 1. Confirm target Supabase project + * 2. Probe Supabase health + * 3. Snapshot reindex health + * 4. Apply ingestion queue recovery if stale/failed jobs are present + * 5. Run worker:once + * 6. Repeat steps 3–5 until the queue is clear or --max-rounds is reached + * + * Usage: + * npm run reindex # interactive – prompts before each recovery + * npm run reindex -- --yes # non-interactive – auto-confirm every prompt + * npm run reindex -- --max-rounds 5 # limit worker iterations (default: 10) + * npm run reindex -- --limit 50 # recovery action limit per round (default: 20) + */ + +import { spawn } from "node:child_process"; +import { loadEnvConfig } from "@next/env"; +import { confirm } from "./cli-utils"; + +loadEnvConfig(process.cwd()); + +function parseArgs(argv: string[]) { + const valueFor = (name: string) => { + const inline = argv.find((arg) => arg.startsWith(`--${name}=`))?.split("=")[1]; + if (inline) return inline; + const index = argv.indexOf(`--${name}`); + return index >= 0 ? argv[index + 1] : undefined; + }; + return { + yes: argv.includes("--yes"), + maxRounds: Number.parseInt(valueFor("max-rounds") ?? "10", 10), + limit: Number.parseInt(valueFor("limit") ?? "20", 10), + }; +} + +function runWorkerOnce(): Promise { + return new Promise((resolve, reject) => { + const child = spawn("npx", ["tsx", "worker/index.ts", "--once"], { + stdio: "inherit", + shell: false, + }); + child.on("close", (code: number | null) => { + if (code === 0) { + resolve(); + } else { + reject(new Error(`worker:once exited with code ${code}`)); + } + }); + child.on("error", reject); + }); +} + +type CountResult = { + count: number | null; + error: { message?: string } | null; +}; + +async function safeCount(label: string, countPromise: PromiseLike) { + try { + const result = await countPromise; + if (result.error) { + return { label, count: null, error: result.error.message ?? "Supabase query failed." }; + } + return { label, count: result.count ?? 0, error: null }; + } catch (error) { + return { label, count: null, error: error instanceof Error ? error.message : String(error) }; + } +} + +async function main() { + const [ + { env, requireServerEnv }, + { buildIngestionRecoveryPlan }, + { createAdminClient }, + { assertSupabaseHealthy, probeSupabaseHealth }, + ] = await Promise.all([ + import("@/lib/env"), + import("@/lib/ingestion-recovery"), + import("@/lib/supabase/admin"), + import("@/lib/supabase/health"), + ]); + requireServerEnv(); + + const args = parseArgs(process.argv.slice(2)); + const staleAfterMinutes = env.WORKER_STALE_AFTER_MINUTES; + const limit = args.limit; + + console.log("=== Reindex Pipeline ===\n"); + + // Step 1 – Confirm target project + const { checkSupabaseProjectConfig, expectedSupabaseProject, formatSupabaseProjectCheck } = await import( + "@/lib/supabase/project" + ); + const projectCheck = checkSupabaseProjectConfig({ + NEXT_PUBLIC_SUPABASE_URL: process.env.NEXT_PUBLIC_SUPABASE_URL, + SUPABASE_PROJECT_REF: process.env.SUPABASE_PROJECT_REF, + SUPABASE_PROJECT_NAME: process.env.SUPABASE_PROJECT_NAME, + }); + if (projectCheck.status === "missing" || projectCheck.status === "mismatch") { + console.error(`Project check failed: ${formatSupabaseProjectCheck(projectCheck)}`); + process.exitCode = 1; + return; + } + console.log(`Project : ${expectedSupabaseProject.name} (${expectedSupabaseProject.ref})`); + + // Step 2 – Supabase health + console.log("\n[Step 1] Supabase health check..."); + const supabase = createAdminClient(); + assertSupabaseHealthy(await probeSupabaseHealth(supabase), "Reindex pipeline"); + console.log(" Supabase is healthy."); + + // Steps 3–5 – Iterative health snapshot → recovery → worker + for (let round = 1; round <= args.maxRounds; round++) { + console.log(`\n[Step 2] Reindex health snapshot (round ${round}/${args.maxRounds})...`); + + const staleCutoff = new Date(Date.now() - staleAfterMinutes * 60_000).toISOString(); + const [pendingJobs, processingJobs, failedJobs, staleJobs, queuedDocuments, failedDocuments, indexedDocuments] = + await Promise.all([ + safeCount( + "jobs_pending", + supabase.from("ingestion_jobs").select("id", { count: "exact", head: true }).eq("status", "pending"), + ), + safeCount( + "jobs_processing", + supabase.from("ingestion_jobs").select("id", { count: "exact", head: true }).eq("status", "processing"), + ), + safeCount( + "jobs_failed", + supabase.from("ingestion_jobs").select("id", { count: "exact", head: true }).eq("status", "failed"), + ), + safeCount( + "jobs_stale", + supabase + .from("ingestion_jobs") + .select("id", { count: "exact", head: true }) + .eq("status", "processing") + .lt("locked_at", staleCutoff), + ), + safeCount( + "documents_queued", + supabase.from("documents").select("id", { count: "exact", head: true }).eq("status", "queued"), + ), + safeCount( + "documents_failed", + supabase.from("documents").select("id", { count: "exact", head: true }).eq("status", "failed"), + ), + safeCount( + "documents_indexed", + supabase.from("documents").select("id", { count: "exact", head: true }).eq("status", "indexed"), + ), + ]); + + const countItems = [ + pendingJobs, + processingJobs, + failedJobs, + staleJobs, + queuedDocuments, + failedDocuments, + indexedDocuments, + ]; + const countErrors = countItems.filter((item) => item.error); + + console.log(` Documents indexed : ${indexedDocuments.count ?? "-"}`); + console.log(` Documents queued : ${queuedDocuments.count ?? "-"}`); + console.log(` Documents failed : ${failedDocuments.count ?? "-"}`); + console.log(` Jobs pending : ${pendingJobs.count ?? "-"}`); + console.log(` Jobs processing : ${processingJobs.count ?? "-"}`); + console.log(` Jobs failed : ${failedJobs.count ?? "-"}`); + console.log(` Jobs stale : ${staleJobs.count ?? "-"}`); + + if (countErrors.length > 0) { + console.error("\n Read errors:"); + for (const item of countErrors) { + console.error(` ${item.label}: ${item.error}`); + } + console.error("\n Fix the reported read errors before running workers or evals."); + process.exitCode = 1; + return; + } + + const openJobs = (pendingJobs.count ?? 0) + (processingJobs.count ?? 0) + (failedJobs.count ?? 0); + const needsRecovery = (staleJobs.count ?? 0) > 0 || (failedJobs.count ?? 0) > 0; + + if (openJobs === 0 && (queuedDocuments.count ?? 0) === 0) { + console.log("\nQueue is clear. Reindex pipeline complete."); + return; + } + + // Step 3 – Recovery (only if stale/failed jobs present) + if (needsRecovery) { + console.log(`\n[Step 3] Queue recovery (stale: ${staleJobs.count}, failed: ${failedJobs.count})...`); + + const { data, error } = await supabase + .from("ingestion_jobs") + .select("id,document_id,status,locked_at,documents(status,page_count,chunk_count)") + .in("status", ["processing", "failed"]) + .order("created_at", { ascending: true }); + + if (error) { + console.error(` Failed to load jobs for recovery: ${error.message}`); + process.exitCode = 1; + return; + } + + type RecoveryDocument = { status?: string | null; page_count?: number | null; chunk_count?: number | null }; + type RawJobRow = { + id: string; + document_id: string; + status: string | null; + locked_at: string | null; + documents: RecoveryDocument | RecoveryDocument[] | null; + }; + const jobs = (data ?? []).map((job: RawJobRow) => ({ + ...job, + documents: Array.isArray(job.documents) + ? (job.documents[0] as RecoveryDocument | undefined) + : (job.documents as RecoveryDocument | undefined), + })); + const plan = buildIngestionRecoveryPlan({ jobs, staleAfterMinutes }); + const actions = plan.actions.slice(0, limit); + const resetDocumentIds = Array.from( + new Set(actions.filter((a) => a.action === "retry").map((a) => a.documentId)), + ); + const supersedeCount = actions.filter((a) => a.action === "supersede").length; + const retryCount = actions.filter((a) => a.action === "retry").length; + + console.log(` Documents to reset: ${resetDocumentIds.length}`); + console.log(` Jobs to supersede : ${supersedeCount}`); + console.log(` Jobs to retry : ${retryCount}`); + + if (actions.length === 0) { + console.log(" Nothing to recover in this round."); + } else { + let shouldApply = args.yes; + if (!shouldApply) { + console.log(""); + shouldApply = await confirm(" Apply recovery changes?"); + } + + if (!shouldApply) { + console.log(" Recovery skipped. Continuing to worker run."); + } else { + for (const documentId of resetDocumentIds) { + const { error: resetError } = await supabase.rpc("reset_document_index", { + p_document_id: documentId, + }); + if (resetError) { + console.error(` Failed to reset document index ${documentId}: ${resetError.message}`); + process.exitCode = 1; + return; + } + const { error: docError } = await supabase + .from("documents") + .update({ status: "queued", error_message: null, page_count: 0, chunk_count: 0, image_count: 0 }) + .eq("id", documentId); + if (docError) { + console.error(` Failed to reset document status ${documentId}: ${docError.message}`); + process.exitCode = 1; + return; + } + } + + for (const action of actions) { + if (action.action === "supersede") { + const { error: supersedeError } = await supabase + .from("ingestion_jobs") + .update({ + status: "completed", + stage: "superseded by successful index", + progress: 100, + error_message: null, + locked_at: null, + locked_by: null, + completed_at: new Date().toISOString(), + }) + .eq("id", action.jobId); + if (supersedeError) { + console.error(` Failed to supersede job ${action.jobId}: ${supersedeError.message}`); + process.exitCode = 1; + return; + } + continue; + } + + const { error: retryError } = await supabase + .from("ingestion_jobs") + .update({ + status: "pending", + stage: "queued after recovery", + progress: 0, + attempt_count: 0, + error_message: null, + locked_at: null, + locked_by: null, + next_run_at: new Date().toISOString(), + completed_at: null, + }) + .eq("id", action.jobId); + if (retryError) { + console.error(` Failed to requeue job ${action.jobId}: ${retryError.message}`); + process.exitCode = 1; + return; + } + } + + console.log(" Recovery applied."); + } + } + } + + // Step 4 – Worker run + console.log("\n[Step 4] Running worker:once..."); + try { + await runWorkerOnce(); + console.log(" Worker run complete."); + } catch (err) { + console.error(` Worker run failed: ${err instanceof Error ? err.message : String(err)}`); + process.exitCode = 1; + return; + } + } + + console.log(`\nReached maximum rounds (${args.maxRounds}). Re-run to continue processing.`); +} + +main().catch((error) => { + import("@/lib/privacy") + .then(({ safeErrorLogDetails }) => { + console.error("Reindex pipeline failed", safeErrorLogDetails(error)); + process.exitCode = 1; + }) + .catch(() => { + console.error(error instanceof Error ? error.message : error); + process.exitCode = 1; + }); +}); diff --git a/scripts/supabase-recovery-status.ts b/scripts/supabase-recovery-status.ts index 34bf03a03..efbc2bd40 100644 --- a/scripts/supabase-recovery-status.ts +++ b/scripts/supabase-recovery-status.ts @@ -24,22 +24,17 @@ async function main() { ]); requireServerEnv(); const supabase = createAdminClient(); + + console.log("=== Supabase Recovery Status ==="); const health = await probeSupabaseHealth(supabase); if (!health.ok) { + console.log(`Status : UNAVAILABLE`); + console.log(`Checked at: ${health.checkedAt}`); + console.log(`Error : ${health.message}`); console.log( - JSON.stringify( - { - ok: false, - status: "supabase_unavailable", - checkedAt: health.checkedAt, - error: health.message, - recommendation: - "Do not run migrations, imports, workers, recovery mutations, or evals. Retry later or contact Supabase support if this persists for more than 30 minutes with local workers stopped.", - }, - null, - 2, - ), + "\nDo not run migrations, imports, workers, recovery mutations, or evals. " + + "Retry later or contact Supabase support if this persists for more than 30 minutes with local workers stopped.", ); process.exitCode = 1; return; @@ -85,23 +80,27 @@ async function main() { : openJobs === 0 ? "Queue is clear. Run indexing checks or resume imports in small waves." : (staleJobs.count ?? 0) > 0 || (failedJobs.count ?? 0) > 0 - ? "Run npm run recover:ingestion -- --apply --limit 20, then npm run worker:once." + ? "Run npm run recover:ingestion, then npm run worker:once." : "Run npm run worker:once with conservative defaults."; - console.log( - JSON.stringify( - { - ok: errors.length === 0, - status: errors.length ? "read_errors" : "ready", - checkedAt: health.checkedAt, - counts: Object.fromEntries(counts.map((item) => [item.label, item.count])), - errors, - recommendation, - }, - null, - 2, - ), - ); + const status = errors.length ? "READ ERRORS" : "READY"; + console.log(`Status : ${status}`); + console.log(`Checked at: ${health.checkedAt}`); + console.log(""); + console.log("Queue counts:"); + for (const item of counts) { + const value = item.error ? `ERROR: ${item.error}` : String(item.count ?? "-"); + console.log(` ${item.label.padEnd(24)}: ${value}`); + } + + if (errors.length > 0) { + console.log("\nRead errors:"); + for (const item of errors) { + console.log(` ${item.label}: ${item.error}`); + } + } + + console.log(`\nRecommendation: ${recommendation}`); if (errors.length) process.exitCode = 1; } From a7393deefb8f11653f06424f0126586188c4acdb Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Fri, 26 Jun 2026 13:50:42 +0800 Subject: [PATCH 3/7] Fix reindex pipeline recovery edge cases --- scripts/cli-utils.ts | 4 ++++ scripts/reindex.ts | 17 +++++++++++++++-- 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/scripts/cli-utils.ts b/scripts/cli-utils.ts index d1862f0c9..24e704814 100644 --- a/scripts/cli-utils.ts +++ b/scripts/cli-utils.ts @@ -5,6 +5,10 @@ import { createInterface } from "node:readline"; * Returns `false` if stdin is not a TTY (e.g. when piped). */ export function confirm(question: string): Promise { + if (!process.stdin.isTTY) { + return Promise.resolve(false); + } + return new Promise((resolve) => { const rl = createInterface({ input: process.stdin, output: process.stdout }); rl.question(`${question} (y/N) `, (answer: string) => { diff --git a/scripts/reindex.ts b/scripts/reindex.ts index 987ea74a8..1d6e4ac4e 100644 --- a/scripts/reindex.ts +++ b/scripts/reindex.ts @@ -137,7 +137,7 @@ async function main() { .from("ingestion_jobs") .select("id", { count: "exact", head: true }) .eq("status", "processing") - .lt("locked_at", staleCutoff), + .or(`locked_at.is.null,locked_at.lt.${staleCutoff}`), ), safeCount( "documents_queued", @@ -184,6 +184,7 @@ async function main() { const openJobs = (pendingJobs.count ?? 0) + (processingJobs.count ?? 0) + (failedJobs.count ?? 0); const needsRecovery = (staleJobs.count ?? 0) > 0 || (failedJobs.count ?? 0) > 0; + let skipWorkerRun = false; if (openJobs === 0 && (queuedDocuments.count ?? 0) === 0) { console.log("\nQueue is clear. Reindex pipeline complete."); @@ -242,7 +243,8 @@ async function main() { } if (!shouldApply) { - console.log(" Recovery skipped. Continuing to worker run."); + console.log(" Recovery skipped. Re-checking queue before worker run."); + skipWorkerRun = true; } else { for (const documentId of resetDocumentIds) { const { error: resetError } = await supabase.rpc("reset_document_index", { @@ -313,6 +315,17 @@ async function main() { } // Step 4 – Worker run + if (skipWorkerRun) { + continue; + } + + if ((processingJobs.count ?? 0) > 0 && !needsRecovery) { + console.log( + "\n Active processing jobs detected with no recovery candidates; skipping worker run to avoid concurrency spikes.", + ); + continue; + } + console.log("\n[Step 4] Running worker:once..."); try { await runWorkerOnce(); From 6735eea7fcb22c82be54b1b2326c64d90fc15bb2 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Fri, 26 Jun 2026 14:09:34 +0800 Subject: [PATCH 4/7] chore(reindex): harden worker recovery and completion paths --- scripts/cli-utils.ts | 1 + scripts/reindex.ts | 29 +++++++++++++++++++++++------ 2 files changed, 24 insertions(+), 6 deletions(-) diff --git a/scripts/cli-utils.ts b/scripts/cli-utils.ts index 24e704814..2d3ff19ab 100644 --- a/scripts/cli-utils.ts +++ b/scripts/cli-utils.ts @@ -6,6 +6,7 @@ import { createInterface } from "node:readline"; */ export function confirm(question: string): Promise { if (!process.stdin.isTTY) { + console.log(" Non-interactive input detected; defaulting to No."); return Promise.resolve(false); } diff --git a/scripts/reindex.ts b/scripts/reindex.ts index 1d6e4ac4e..ac8708623 100644 --- a/scripts/reindex.ts +++ b/scripts/reindex.ts @@ -183,10 +183,11 @@ async function main() { } const openJobs = (pendingJobs.count ?? 0) + (processingJobs.count ?? 0) + (failedJobs.count ?? 0); + const hasFailedDocuments = (failedDocuments.count ?? 0) > 0; const needsRecovery = (staleJobs.count ?? 0) > 0 || (failedJobs.count ?? 0) > 0; let skipWorkerRun = false; - if (openJobs === 0 && (queuedDocuments.count ?? 0) === 0) { + if (openJobs === 0 && (queuedDocuments.count ?? 0) === 0 && !hasFailedDocuments) { console.log("\nQueue is clear. Reindex pipeline complete."); return; } @@ -215,12 +216,26 @@ async function main() { locked_at: string | null; documents: RecoveryDocument | RecoveryDocument[] | null; }; - const jobs = (data ?? []).map((job: RawJobRow) => ({ + const staleAfterCutoff = new Date(staleCutoff); + const jobs = (data ?? []) + .map((job: RawJobRow) => ({ ...job, documents: Array.isArray(job.documents) ? (job.documents[0] as RecoveryDocument | undefined) : (job.documents as RecoveryDocument | undefined), - })); + })) + .filter((job) => { + if (job.status === "failed") { + return true; + } + if (job.status !== "processing") { + return false; + } + if (job.locked_at === null) { + return true; + } + return new Date(job.locked_at) <= staleAfterCutoff; + }); const plan = buildIngestionRecoveryPlan({ jobs, staleAfterMinutes }); const actions = plan.actions.slice(0, limit); const resetDocumentIds = Array.from( @@ -235,6 +250,7 @@ async function main() { if (actions.length === 0) { console.log(" Nothing to recover in this round."); + skipWorkerRun = true; } else { let shouldApply = args.yes; if (!shouldApply) { @@ -319,9 +335,9 @@ async function main() { continue; } - if ((processingJobs.count ?? 0) > 0 && !needsRecovery) { + if ((processingJobs.count ?? 0) > 0) { console.log( - "\n Active processing jobs detected with no recovery candidates; skipping worker run to avoid concurrency spikes.", + "\n Active processing jobs detected; skipping worker run to avoid concurrency spikes.", ); continue; } @@ -337,7 +353,8 @@ async function main() { } } - console.log(`\nReached maximum rounds (${args.maxRounds}). Re-run to continue processing.`); + console.log(`\nReached maximum rounds (${args.maxRounds}). Queue did not clear; re-run with a higher --max-rounds or inspect stale jobs first.`); + process.exitCode = 1; } main().catch((error) => { From 34de6f15f14ca4bef92d9b3099365f19cd5657a6 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Fri, 26 Jun 2026 14:51:25 +0800 Subject: [PATCH 5/7] Ignore Claude and visual temp files in ESLint --- eslint.config.mjs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/eslint.config.mjs b/eslint.config.mjs index e5dee5498..a8172a51a 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -16,6 +16,10 @@ const eslintConfig = defineConfig([ "test-results/**", "sample-documents/**", "scratch/**", + ".claude/**", + "**/.claude/**", + ".tmp-visual/**", + "**/.tmp-visual/**", "next-env.d.ts", ]), ]); From 64411cf657e53e745933b87dd279f48741696dd4 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Fri, 26 Jun 2026 15:23:27 +0800 Subject: [PATCH 6/7] Update work --- tests/deep-memory.test.ts | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/tests/deep-memory.test.ts b/tests/deep-memory.test.ts index b86b98eeb..9f5d2fca2 100644 --- a/tests/deep-memory.test.ts +++ b/tests/deep-memory.test.ts @@ -259,8 +259,11 @@ describe("deep RAG memory indexing", () => { insertedMemoryRows.push(...payload); return Promise.resolve({ data: null, error: null }); }, - select: () => ({ - eq: () => { + select: (columns: string) => ({ + eq: (column: string, value: unknown) => { + void columns; + void column; + void value; if (table === "documents") { return { single: async () => ({ data: { metadata: {} }, error: null }), @@ -278,7 +281,9 @@ describe("deep RAG memory indexing", () => { }, }), update: (payload: Record) => ({ - eq: () => { + eq: (column: string, value: unknown) => { + void column; + void value; updatedRows.set(table, [...(updatedRows.get(table) ?? []), payload]); return Promise.resolve({ data: null, error: null }); }, @@ -319,3 +324,4 @@ describe("deep RAG memory indexing", () => { ); }); }); + From d7d3b464dca75bee390760beccc202b1cc4e9753 Mon Sep 17 00:00:00 2001 From: BigSimmo <87357024+BigSimmo@users.noreply.github.com> Date: Fri, 26 Jun 2026 15:50:31 +0800 Subject: [PATCH 7/7] fix(reindex): tighten recovery completion checks --- scripts/reindex.ts | 62 ++++++++++++++++++++++++++------ src/lib/ingestion-recovery.ts | 30 +++++++++++++--- src/lib/reindex-pipeline.ts | 23 ++++++++++++ tests/ingestion-recovery.test.ts | 38 +++++++++++++++++++- tests/reindex-pipeline.test.ts | 45 +++++++++++++++++++++++ 5 files changed, 182 insertions(+), 16 deletions(-) create mode 100644 src/lib/reindex-pipeline.ts create mode 100644 tests/reindex-pipeline.test.ts diff --git a/scripts/reindex.ts b/scripts/reindex.ts index ac8708623..662cf36b1 100644 --- a/scripts/reindex.ts +++ b/scripts/reindex.ts @@ -73,12 +73,14 @@ async function safeCount(label: string, countPromise: PromiseLike) async function main() { const [ { env, requireServerEnv }, - { buildIngestionRecoveryPlan }, + { buildIngestionRecoveryPlan, isFreshProcessingJob }, + { hasIncompleteDocumentsWithoutOpenJobs, isReindexQueueClear }, { createAdminClient }, { assertSupabaseHealthy, probeSupabaseHealth }, ] = await Promise.all([ import("@/lib/env"), import("@/lib/ingestion-recovery"), + import("@/lib/reindex-pipeline"), import("@/lib/supabase/admin"), import("@/lib/supabase/health"), ]); @@ -116,8 +118,18 @@ async function main() { for (let round = 1; round <= args.maxRounds; round++) { console.log(`\n[Step 2] Reindex health snapshot (round ${round}/${args.maxRounds})...`); - const staleCutoff = new Date(Date.now() - staleAfterMinutes * 60_000).toISOString(); - const [pendingJobs, processingJobs, failedJobs, staleJobs, queuedDocuments, failedDocuments, indexedDocuments] = + const snapshotNow = new Date(); + const staleCutoff = new Date(snapshotNow.getTime() - staleAfterMinutes * 60_000).toISOString(); + const [ + pendingJobs, + processingJobs, + failedJobs, + staleJobs, + queuedDocuments, + processingDocuments, + failedDocuments, + indexedDocuments, + ] = await Promise.all([ safeCount( "jobs_pending", @@ -143,6 +155,10 @@ async function main() { "documents_queued", supabase.from("documents").select("id", { count: "exact", head: true }).eq("status", "queued"), ), + safeCount( + "documents_processing", + supabase.from("documents").select("id", { count: "exact", head: true }).eq("status", "processing"), + ), safeCount( "documents_failed", supabase.from("documents").select("id", { count: "exact", head: true }).eq("status", "failed"), @@ -159,6 +175,7 @@ async function main() { failedJobs, staleJobs, queuedDocuments, + processingDocuments, failedDocuments, indexedDocuments, ]; @@ -166,6 +183,7 @@ async function main() { console.log(` Documents indexed : ${indexedDocuments.count ?? "-"}`); console.log(` Documents queued : ${queuedDocuments.count ?? "-"}`); + console.log(` Documents processing: ${processingDocuments.count ?? "-"}`); console.log(` Documents failed : ${failedDocuments.count ?? "-"}`); console.log(` Jobs pending : ${pendingJobs.count ?? "-"}`); console.log(` Jobs processing : ${processingJobs.count ?? "-"}`); @@ -183,15 +201,29 @@ async function main() { } const openJobs = (pendingJobs.count ?? 0) + (processingJobs.count ?? 0) + (failedJobs.count ?? 0); - const hasFailedDocuments = (failedDocuments.count ?? 0) > 0; const needsRecovery = (staleJobs.count ?? 0) > 0 || (failedJobs.count ?? 0) > 0; let skipWorkerRun = false; - - if (openJobs === 0 && (queuedDocuments.count ?? 0) === 0 && !hasFailedDocuments) { + let hasActiveProcessingJobs = !needsRecovery && (processingJobs.count ?? 0) > 0; + const queueSnapshot = { + openJobs, + queuedDocuments: queuedDocuments.count ?? 0, + processingDocuments: processingDocuments.count ?? 0, + failedDocuments: failedDocuments.count ?? 0, + }; + + if (isReindexQueueClear(queueSnapshot)) { console.log("\nQueue is clear. Reindex pipeline complete."); return; } + if (hasIncompleteDocumentsWithoutOpenJobs(queueSnapshot)) { + console.error( + "\nNo open ingestion jobs remain, but some documents are still processing or failed. Run targeted recovery or manual inspection before declaring the corpus reindexed.", + ); + process.exitCode = 1; + return; + } + // Step 3 – Recovery (only if stale/failed jobs present) if (needsRecovery) { console.log(`\n[Step 3] Queue recovery (stale: ${staleJobs.count}, failed: ${failedJobs.count})...`); @@ -219,10 +251,10 @@ async function main() { const staleAfterCutoff = new Date(staleCutoff); const jobs = (data ?? []) .map((job: RawJobRow) => ({ - ...job, - documents: Array.isArray(job.documents) - ? (job.documents[0] as RecoveryDocument | undefined) - : (job.documents as RecoveryDocument | undefined), + ...job, + documents: Array.isArray(job.documents) + ? (job.documents[0] as RecoveryDocument | undefined) + : (job.documents as RecoveryDocument | undefined), })) .filter((job) => { if (job.status === "failed") { @@ -236,6 +268,14 @@ async function main() { } return new Date(job.locked_at) <= staleAfterCutoff; }); + hasActiveProcessingJobs = (data ?? []) + .map((job: RawJobRow) => ({ + ...job, + documents: Array.isArray(job.documents) + ? (job.documents[0] as RecoveryDocument | undefined) + : (job.documents as RecoveryDocument | undefined), + })) + .some((job) => isFreshProcessingJob(job, snapshotNow, staleAfterMinutes)); const plan = buildIngestionRecoveryPlan({ jobs, staleAfterMinutes }); const actions = plan.actions.slice(0, limit); const resetDocumentIds = Array.from( @@ -335,7 +375,7 @@ async function main() { continue; } - if ((processingJobs.count ?? 0) > 0) { + if (hasActiveProcessingJobs) { console.log( "\n Active processing jobs detected; skipping worker run to avoid concurrency spikes.", ); diff --git a/src/lib/ingestion-recovery.ts b/src/lib/ingestion-recovery.ts index 989bc9d40..e2498ffd4 100644 --- a/src/lib/ingestion-recovery.ts +++ b/src/lib/ingestion-recovery.ts @@ -14,11 +14,31 @@ export type IngestionRecoveryAction = | { action: "supersede"; jobId: string; documentId: string } | { action: "retry"; jobId: string; documentId: string; resetDocument: boolean }; +function parseLockedAt(value: string | null | undefined) { + if (!value) return null; + const time = Date.parse(value); + return Number.isFinite(time) ? time : null; +} + export function isStaleProcessingJob(job: IngestionRecoveryJob, now: Date, staleAfterMinutes: number) { if (job.status !== "processing" || !job.locked_at) return false; - const lockedAt = new Date(job.locked_at); - if (Number.isNaN(lockedAt.getTime())) return false; - return lockedAt.getTime() < now.getTime() - staleAfterMinutes * 60_000; + const lockedAt = parseLockedAt(job.locked_at); + if (lockedAt === null) return false; + return lockedAt < now.getTime() - staleAfterMinutes * 60_000; +} + +export function isRecoverableProcessingJob(job: IngestionRecoveryJob, now: Date, staleAfterMinutes: number) { + if (job.status !== "processing") return false; + const lockedAt = parseLockedAt(job.locked_at); + if (lockedAt === null) return true; + return lockedAt < now.getTime() - staleAfterMinutes * 60_000; +} + +export function isFreshProcessingJob(job: IngestionRecoveryJob, now: Date, staleAfterMinutes: number) { + if (job.status !== "processing" || !job.locked_at) return false; + const lockedAt = parseLockedAt(job.locked_at); + if (lockedAt === null) return false; + return lockedAt >= now.getTime() - staleAfterMinutes * 60_000; } export function buildIngestionRecoveryPlan(args: { @@ -35,7 +55,9 @@ export function buildIngestionRecoveryPlan(args: { const chunkCount = Number(job.documents?.chunk_count ?? 0); const isIndexedDocument = documentStatus === "indexed" && chunkCount > 0; const isRecoverableStatus = - job.status === "failed" || job.status === "pending" || isStaleProcessingJob(job, now, args.staleAfterMinutes); + job.status === "failed" || + job.status === "pending" || + isRecoverableProcessingJob(job, now, args.staleAfterMinutes); if (isIndexedDocument && job.status !== "completed") { actions.push({ action: "supersede", jobId: job.id, documentId: job.document_id }); diff --git a/src/lib/reindex-pipeline.ts b/src/lib/reindex-pipeline.ts new file mode 100644 index 000000000..1a10710d6 --- /dev/null +++ b/src/lib/reindex-pipeline.ts @@ -0,0 +1,23 @@ +export type ReindexQueueSnapshot = { + openJobs: number; + queuedDocuments: number; + processingDocuments: number; + failedDocuments: number; +}; + +export function isReindexQueueClear(snapshot: ReindexQueueSnapshot) { + return ( + snapshot.openJobs === 0 && + snapshot.queuedDocuments === 0 && + snapshot.processingDocuments === 0 && + snapshot.failedDocuments === 0 + ); +} + +export function hasIncompleteDocumentsWithoutOpenJobs(snapshot: ReindexQueueSnapshot) { + return ( + snapshot.openJobs === 0 && + snapshot.queuedDocuments === 0 && + (snapshot.processingDocuments > 0 || snapshot.failedDocuments > 0) + ); +} diff --git a/tests/ingestion-recovery.test.ts b/tests/ingestion-recovery.test.ts index 95d7e63f1..349713925 100644 --- a/tests/ingestion-recovery.test.ts +++ b/tests/ingestion-recovery.test.ts @@ -1,5 +1,10 @@ import { describe, expect, it } from "vitest"; -import { buildIngestionRecoveryPlan, isStaleProcessingJob } from "../src/lib/ingestion-recovery"; +import { + buildIngestionRecoveryPlan, + isFreshProcessingJob, + isRecoverableProcessingJob, + isStaleProcessingJob, +} from "../src/lib/ingestion-recovery"; describe("ingestion queue recovery planning", () => { const now = new Date("2026-06-15T00:00:00.000Z"); @@ -62,4 +67,35 @@ describe("ingestion queue recovery planning", () => { ), ).toBe(false); }); + + it("retries processing jobs with no lock timestamp", () => { + const plan = buildIngestionRecoveryPlan({ + now, + staleAfterMinutes: 45, + jobs: [ + { + id: "null-lock", + document_id: "doc-null", + status: "processing", + locked_at: null, + documents: { status: "processing", chunk_count: 0 }, + }, + ], + }); + + expect(plan.retryCount).toBe(1); + expect(plan.actions[0]).toMatchObject({ action: "retry", jobId: "null-lock", documentId: "doc-null" }); + }); + + it("treats fresh processing jobs as active but not recoverable", () => { + const job = { + id: "fresh", + document_id: "doc", + status: "processing" as const, + locked_at: "2026-06-14T23:30:00.000Z", + }; + + expect(isRecoverableProcessingJob(job, now, 45)).toBe(false); + expect(isFreshProcessingJob(job, now, 45)).toBe(true); + }); }); diff --git a/tests/reindex-pipeline.test.ts b/tests/reindex-pipeline.test.ts new file mode 100644 index 000000000..b76d2791c --- /dev/null +++ b/tests/reindex-pipeline.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it } from "vitest"; +import { hasIncompleteDocumentsWithoutOpenJobs, isReindexQueueClear } from "../src/lib/reindex-pipeline"; + +describe("reindex pipeline queue state", () => { + it("does not declare the queue clear while documents are still processing", () => { + expect( + isReindexQueueClear({ + openJobs: 0, + queuedDocuments: 0, + processingDocuments: 1, + failedDocuments: 0, + }), + ).toBe(false); + }); + + it("flags orphaned incomplete documents when no jobs remain", () => { + expect( + hasIncompleteDocumentsWithoutOpenJobs({ + openJobs: 0, + queuedDocuments: 0, + processingDocuments: 1, + failedDocuments: 0, + }), + ).toBe(true); + expect( + hasIncompleteDocumentsWithoutOpenJobs({ + openJobs: 0, + queuedDocuments: 1, + processingDocuments: 0, + failedDocuments: 1, + }), + ).toBe(false); + }); + + it("declares the queue clear only when no open jobs or incomplete documents remain", () => { + expect( + isReindexQueueClear({ + openJobs: 0, + queuedDocuments: 0, + processingDocuments: 0, + failedDocuments: 0, + }), + ).toBe(true); + }); +});