fix: close bug-hunter stale-state paths (re-open)

AGENTS.md

@@ -89,6 +89,28 @@ After setup:
 
 <!-- END:dependency-shortcut -->
 
+<!-- BEGIN:bug-hunter-shortcut -->
+
+## Bug-hunter shortcut
+
+When the user types exactly `bug-hunter` as the entire task message, after trimming surrounding whitespace, treat it as a shortcut for targeted defect discovery.
+
+Execution rules:
+
+- Invoke the `bug-hunter` skill first.
+- Prioritize reproducible defects over code style, naming, or formatting feedback.
+- Trace realistic failure paths: invalid input, empty states, retries, race/concurrency issues, stale state/cache, network/auth failures, permissions, and boundary values.
+- For each finding, include trigger, expected behavior, actual risk, and the smallest proof (or targeted test/check) that would catch it.
+- If no high-confidence defect is found, explicitly state that and list the most likely residual risk area.
+
+Scope and safety:
+
+- Keep the hunt scoped to code touched by the user request unless the defect clearly crosses module boundaries.
+- Do not make broad refactors while hunting; propose minimal fixes for confirmed issues.
+- Run the smallest focused verification for each confirmed defect, then expand only if needed.
+
+<!-- END:bug-hunter-shortcut -->
+
 <!-- BEGIN:local-server-safety -->
 
 # Local server safety

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

@@ -5,7 +5,11 @@ import { env, isDemoMode } from "@/lib/env";
 import { upsertDocumentEnrichment } from "@/lib/document-enrichment";
 import { upsertDocumentDeepMemory } from "@/lib/deep-memory";
 import { jsonError } from "@/lib/http";
-import { checkIngestionMutationSafety, ingestionMutationSafetyPayload } from "@/lib/ingestion-mutation-safety";
+import {
+ checkIngestionMutationSafety,
+ ingestionMutationSafetyPayload,
+ ingestionRollbackFenceStamp,
+} from "@/lib/ingestion-mutation-safety";
 import { consumeApiRateLimit, rateLimitJsonResponse } from "@/lib/api-rate-limit";
 import {
  committedIndexGeneration,
@@ -111,7 +115,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
 
  const { data: document, error: documentError } = await supabase
  .from("documents")
- .select("id,owner_id,title,file_name,source_path,import_batch_id,status,metadata")
+ .select("id,owner_id,title,file_name,source_path,import_batch_id,status,error_message,page_count,chunk_count,image_count,metadata")
  .eq("id", id)
  .eq("owner_id", user.id)
  .maybeSingle();
@@ -180,12 +184,36 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
  }
 
  const atomicReindex = isAtomicReindexCandidate(document);
+ // Rollback fence: the queue-state write stamps updated_at with a
+ // per-request value and the rollback below matches on that stamp, making
+ // it a single conditional UPDATE that is atomic server-side. An
+ // overlapping reindex/retry re-stamps the row before enqueueing its own
+ // job, so a stale rollback from this request matches zero rows instead of
+ // reverting the newer queue state. The competing-job SELECT below is only
+ // a cheap fast path; the fence is what closes the check-then-write race.
+ const rollbackFence = ingestionRollbackFenceStamp();
+ const rollbackDocumentPayload = atomicReindex
+ ? { error_message: document.error_message ?? null }
+ : {
+ status: document.status ?? null,
+ error_message: document.error_message ?? null,
+ page_count: document.page_count ?? 0,
+ chunk_count: document.chunk_count ?? 0,
+ image_count: document.image_count ?? 0,
+ };
  const { error: updateError } = await supabase
  .from("documents")
  .update(
  atomicReindex
- ? { error_message: null }
- : { status: "queued", error_message: null, page_count: 0, chunk_count: 0, image_count: 0 },
+ ? { error_message: null, updated_at: rollbackFence }
+ : {
+ status: "queued",
+ error_message: null,
+ page_count: 0,
+ chunk_count: 0,
+ image_count: 0,
+ updated_at: rollbackFence,
+ },
  )
  .eq("id", id)
  .eq("owner_id", user.id);
@@ -204,7 +232,29 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
  .select()
  .single();
 
- if (jobError) throw new Error(jobError.message);
+ if (jobError) {
+ const { data: competingJobs, error: competingJobsError } = await supabase
+ .from("ingestion_jobs")
+ .select("id")
+ .eq("document_id", id)
+ .in("status", ["pending", "processing"])
+ .limit(1);
+ if (competingJobsError) {
+ throw new Error(`Failed to enqueue reindex job: ${jobError.message}; competing-job check failed: ${competingJobsError.message}`);
+ }
+ if ((competingJobs?.length ?? 0) === 0) {
+ const { error: rollbackError } = await supabase
+ .from("documents")
+ .update(rollbackDocumentPayload)
+ .eq("id", id)
+ .eq("owner_id", user.id)
+ .eq("updated_at", rollbackFence);
+ if (rollbackError) {
+ throw new Error(`Failed to enqueue reindex job: ${jobError.message}; rollback failed: ${rollbackError.message}`);
+ }
+ }
+ throw new Error(jobError.message);
+ }
  return NextResponse.json({ job }, { status: 201 });
  } catch (error) {
  if (error instanceof AuthenticationError) return unauthorizedResponse();

src/app/api/documents/bulk/reindex/route.ts

@@ -5,7 +5,11 @@ import { upsertDocumentDeepMemory } from "@/lib/deep-memory";
 import { upsertDocumentEnrichment } from "@/lib/document-enrichment";
 import { env, isDemoMode } from "@/lib/env";
 import { jsonError, PublicApiError } from "@/lib/http";
-import { checkIngestionMutationSafety, ingestionMutationSafetyPayload } from "@/lib/ingestion-mutation-safety";
+import {
+ checkIngestionMutationSafety,
+ ingestionMutationSafetyPayload,
+ ingestionRollbackFenceStamp,
+} from "@/lib/ingestion-mutation-safety";
 import { consumeApiRateLimit, rateLimitJsonResponse } from "@/lib/api-rate-limit";
 import { invalidateRagCachesForOwner } from "@/lib/rag";
 import {
@@ -101,7 +105,7 @@ export async function POST(request: Request) {
  const documentIds = Array.from(new Set(parsed.documentIds));
  const { data: documents, error: documentError } = await supabase
  .from("documents")
- .select("id,owner_id,title,file_name,source_path,import_batch_id,status,metadata")
+ .select("id,owner_id,title,file_name,source_path,import_batch_id,status,error_message,page_count,chunk_count,image_count,metadata")
  .eq("owner_id", user.id)
  .in("id", documentIds);
  if (documentError) throw new Error(documentError.message);
@@ -176,12 +180,34 @@ export async function POST(request: Request) {
  }
 
  const atomicReindex = isAtomicReindexCandidate(document);
+ // Rollback fence: same pattern as the single-document reindex route —
+ // the queue-state write stamps updated_at and the rollback matches on
+ // the stamp, so a stale rollback cannot revert a newer queue state
+ // written by an overlapping reindex/retry. The competing-job SELECT is
+ // only a fast path; the fence closes the check-then-write race.
+ const rollbackFence = ingestionRollbackFenceStamp();
+ const rollbackDocumentPayload = atomicReindex
+ ? { error_message: document.error_message ?? null }
+ : {
+ status: document.status ?? null,
+ error_message: document.error_message ?? null,
+ page_count: document.page_count ?? 0,
+ chunk_count: document.chunk_count ?? 0,
+ image_count: document.image_count ?? 0,
+ };
  const { error: updateError } = await supabase
  .from("documents")
  .update(
  atomicReindex
- ? { error_message: null }
- : { status: "queued", error_message: null, page_count: 0, chunk_count: 0, image_count: 0 },
+ ? { error_message: null, updated_at: rollbackFence }
+ : {
+ status: "queued",
+ error_message: null,
+ page_count: 0,
+ chunk_count: 0,
+ image_count: 0,
+ updated_at: rollbackFence,
+ },
  )
  .eq("id", document.id)
  .eq("owner_id", user.id);
@@ -198,7 +224,31 @@ export async function POST(request: Request) {
  })
  .select("id")
  .single();
- if (jobError) throw new Error(jobError.message);
+ if (jobError) {
+ const { data: competingJobs, error: competingJobsError } = await supabase
+ .from("ingestion_jobs")
+ .select("id")
+ .eq("document_id", document.id)
+ .in("status", ["pending", "processing"])
+ .limit(1);
+ if (competingJobsError) {
+ throw new Error(
+ `Failed to enqueue bulk reindex job: ${jobError.message}; competing-job check failed: ${competingJobsError.message}`,
+ );
+ }
+ if ((competingJobs?.length ?? 0) === 0) {
+ const { error: rollbackError } = await supabase
+ .from("documents")
+ .update(rollbackDocumentPayload)
+ .eq("id", document.id)
+ .eq("owner_id", user.id)
+ .eq("updated_at", rollbackFence);
+ if (rollbackError) {
+ throw new Error(`Failed to enqueue bulk reindex job: ${jobError.message}; rollback failed: ${rollbackError.message}`);
+ }
+ }
+ throw new Error(jobError.message);
+ }
  results.push({ documentId: document.id, mode: parsed.mode, ok: true, jobId: job.id });
  } catch (error) {
  results.push({

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

@@ -2,6 +2,7 @@ import { NextResponse } from "next/server";
 import { z } from "zod";
 import { env, isDemoMode } from "@/lib/env";
 import { jsonError } from "@/lib/http";
+import { ingestionRollbackFenceStamp } from "@/lib/ingestion-mutation-safety";
 import { createAdminClient } from "@/lib/supabase/admin";
 import { AuthenticationError, requireAuthenticatedUser, unauthorizedResponse } from "@/lib/supabase/auth";
 import { parseRouteParams } from "@/lib/validation/params";
@@ -23,7 +24,9 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
 
  const { data: job, error: jobError } = await supabase
  .from("ingestion_jobs")
- .select("id,document_id,batch_id,status,locked_at,documents!inner(owner_id)")
+ .select(
+ "id,document_id,batch_id,status,stage,progress,error_message,attempt_count,max_attempts,locked_at,locked_by,next_run_at,completed_at,documents!inner(owner_id)",
+ )
  .eq("id", id)
  .eq("documents.owner_id", user.id)
  .maybeSingle();
@@ -41,6 +44,13 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
  // job is NOT processing, OR its lock is already stale, OR it has no lock.
  const staleThreshold = new Date(Date.now() - env.WORKER_STALE_AFTER_MINUTES * 60_000).toISOString();
 
+ // Rollback fence: next_run_at doubles as a per-request stamp. Two
+ // overlapping retries write generically identical resets (pending/queued/
+ // attempt 0), so the rollback below additionally matches on this exact
+ // value — a stale rollback from the losing request affects zero rows
+ // instead of reverting the winning request's reset.
+ const resetNextRunAt = ingestionRollbackFenceStamp();
+
  const { data, error } = await supabase
  .from("ingestion_jobs")
  .update({
@@ -52,7 +62,7 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
  max_attempts: env.WORKER_MAX_ATTEMPTS,
  locked_at: null,
  locked_by: null,
- next_run_at: new Date().toISOString(),
+ next_run_at: resetNextRunAt,
  completed_at: null,
  })
  .eq("id", id)
@@ -76,12 +86,41 @@ export async function POST(request: Request, { params }: { params: Promise<{ id:
  // job start (worker/main.ts), so resetting before enqueue would leave a previously-good
  // clinical document with zero index if the worker never runs or fails permanently. We
  // only re-queue; the prior index stays live until the worker commits a fresh one.
+ // updated_at is stamped so the reindex routes' rollback fences (which
+ // match on documents.updated_at) can see this competing queue-state write.
  const { error: documentError } = await supabase
  .from("documents")
- .update({ status: "queued", error_message: null })
+ .update({ status: "queued", error_message: null, updated_at: ingestionRollbackFenceStamp() })
  .eq("id", job.document_id)
  .eq("owner_id", user.id);
- if (documentError) throw new Error(documentError.message);
+ if (documentError) {
+ const { error: rollbackError } = await supabase
+ .from("ingestion_jobs")
+ .update({
+ status: job.status,
+ stage: job.stage,
+ progress: job.progress,
+ error_message: job.error_message,
+ attempt_count: job.attempt_count,
+ max_attempts: job.max_attempts,
+ locked_at: job.locked_at,
+ locked_by: job.locked_by,
+ next_run_at: job.next_run_at,
+ completed_at: job.completed_at,
+ })
+ .eq("id", id)
+ .eq("status", "pending")
+ .eq("stage", "queued")
+ .eq("progress", 0)
+ .eq("attempt_count", 0)
+ .is("locked_at", null)
+ .is("locked_by", null)
+ .eq("next_run_at", resetNextRunAt);
+ if (rollbackError) {
+ throw new Error(`${documentError.message}; failed to roll back retried job state: ${rollbackError.message}`);
+ }
+ throw new Error(documentError.message);
+ }
 
  return NextResponse.json({ job: data });
  } catch (error) {

src/app/api/upload/route.ts

@@ -24,6 +24,8 @@ const uploadMetadataSchema = z
 export async function POST(request: Request) {
  let supabase: ReturnType<typeof createAdminClient> | null = null;
  let uploadedPath: string | null = null;
+ let insertedDocumentId: string | null = null;
+ let insertedDocumentOwnerId: string | null = null;
 
  try {
  supabase = createAdminClient();
@@ -141,6 +143,8 @@ export async function POST(request: Request) {
  .single();
 
  if (documentError) throw new Error(documentError.message);
+ insertedDocumentId = documentId;
+ insertedDocumentOwnerId = user.id;
 
  const { data: job, error: jobError } = await supabase
  .from("ingestion_jobs")
@@ -155,7 +159,19 @@ export async function POST(request: Request) {
  .select()
  .single();
 
- if (jobError) throw new Error(jobError.message);
+ if (jobError) {
+ const { error: rollbackDocumentError } = await supabase
+ .from("documents")
+ .delete()
+ .eq("id", documentId)
+ .eq("owner_id", user.id);
+ if (rollbackDocumentError) {
+ throw new Error(`Failed to enqueue ingestion job: ${jobError.message}; rollback failed: ${rollbackDocumentError.message}`);
+ }
+ insertedDocumentId = null;
+ insertedDocumentOwnerId = null;
+ throw new Error(jobError.message);
+ }
 
  await writeAuditLog(supabase, {
  ownerId: user.id,
@@ -167,9 +183,38 @@ export async function POST(request: Request) {
 
  return NextResponse.json({ document, job }, { status: 201 });
  } catch (error) {
+ if (insertedDocumentId && insertedDocumentOwnerId && supabase) {
+ try {
+ const { error: cleanupDeleteError } = await supabase
+ .from("documents")
+ .delete()
+ .eq("id", insertedDocumentId)
+ .eq("owner_id", insertedDocumentOwnerId);
+ if (cleanupDeleteError) {
+ logger.error("Upload cleanup failed; document row may be orphaned", {
+ documentId: insertedDocumentId,
+ ownerId: insertedDocumentOwnerId,
+ message: cleanupDeleteError.message,
+ });
+ }
+ } catch (cleanupError) {
+ logger.error("Upload cleanup failed; document row may be orphaned", {
+ documentId: insertedDocumentId,
+ ownerId: insertedDocumentOwnerId,
+ message: cleanupError instanceof Error ? cleanupError.message : String(cleanupError),
+ });
+ }
+ }
+
  if (uploadedPath && supabase) {
  try {
- await supabase.storage.from(env.SUPABASE_DOCUMENT_BUCKET).remove([uploadedPath]);
+ const { error: cleanupStorageError } = await supabase.storage.from(env.SUPABASE_DOCUMENT_BUCKET).remove([uploadedPath]);
+ if (cleanupStorageError) {
+ logger.error("Upload cleanup failed; storage object may be orphaned", {
+ storagePath: uploadedPath,
+ message: cleanupStorageError.message,
+ });
+ }
  } catch (cleanupError) {
  // Cleanup is best-effort, but a silent failure leaves an orphaned storage
  // object. Record the path so it can be reconciled instead of dropping it.