fix(dn): close promotion race and rerun gaps

src/lib/dn-pipeline/promote.test.ts

@@ -28,6 +28,7 @@ function makePb() {
         return {
           create: deliveryNotesCreate,
           update: deliveryNotesUpdate,
+          getFullList: deliveryNotesGetFullList,
         };
       case "delivery_note_items":
         return {
@@ -47,6 +48,7 @@ const dnBatchNotesUpdate = vi.fn();
 const dnReviewEventsCreate = vi.fn();
 const deliveryNotesCreate = vi.fn();
 const deliveryNotesUpdate = vi.fn();
+const deliveryNotesGetFullList = vi.fn();
 const deliveryNoteItemsGetFullList = vi.fn();
 const deliveryNoteItemsCreate = vi.fn();
 const deliveryNoteItemsDelete = vi.fn();
@@ -57,6 +59,7 @@ beforeEach(() => {
   dnReviewEventsCreate.mockReset();
   deliveryNotesCreate.mockReset();
   deliveryNotesUpdate.mockReset();
+  deliveryNotesGetFullList.mockReset();
   deliveryNoteItemsGetFullList.mockReset();
   deliveryNoteItemsCreate.mockReset();
   deliveryNoteItemsDelete.mockReset();
@@ -69,6 +72,7 @@ beforeEach(() => {
   deliveryNoteItemsCreate.mockResolvedValue({ id: "item1" });
   deliveryNoteItemsDelete.mockResolvedValue(undefined);
   deliveryNotesUpdate.mockResolvedValue({ id: "dnprior01" });
+  deliveryNotesGetFullList.mockResolvedValue([]);
   dnBatchNotesUpdate.mockResolvedValue({ id: "bn1" });
   dnReviewEventsCreate.mockResolvedValue({ id: "event1" });
 });
@@ -394,4 +398,70 @@ describe("promoteNote", () => {
     expect(deliveryNotesUpdate).not.toHaveBeenCalled();
     expect(deliveryNoteItemsCreate).not.toHaveBeenCalled();
   });
+
+  it("reconciles a unique-key create race to an already-posted winner without mutation", async () => {
+    dnBatchNotesGetOne
+      .mockResolvedValueOnce({ id: "bn1", delivery_note: "" })
+      .mockResolvedValueOnce({ id: "bn1", status: "promoted", stage: "posted", posted_at: "2026-09-08T00:00:00Z", delivery_note: "dnwinner" });
+    deliveryNotesCreate.mockRejectedValueOnce(new Error("batch_note unique constraint"));
+    deliveryNotesGetFullList.mockResolvedValueOnce([]).mockResolvedValueOnce([{ id: "dnwinner", batch_note: "bn1" }]);
+    deliveryNoteItemsGetFullList.mockResolvedValueOnce([{ id: "winner-item" }]);
+
+    const result = await promoteNote(input());
+
+    expect(result).toEqual({ deliveryNoteId: "dnwinner", itemsCreated: 1 });
+    expect(deliveryNotesCreate).toHaveBeenCalledTimes(1);
+    expect(deliveryNotesUpdate).not.toHaveBeenCalled();
+    expect(deliveryNoteItemsCreate).not.toHaveBeenCalled();
+    expect(dnBatchNotesUpdate).not.toHaveBeenCalled();
+    expect(dnReviewEventsCreate).not.toHaveBeenCalled();
+  });
+
+  it("surfaces an in-progress create race without touching the winner's lines", async () => {
+    dnBatchNotesGetOne
+      .mockResolvedValueOnce({ id: "bn1", delivery_note: "" })
+      .mockResolvedValueOnce({ id: "bn1", status: "blocked", stage: "needs_review", delivery_note: "dnpartial" });
+    deliveryNotesCreate.mockRejectedValueOnce(new Error("batch_note unique constraint"));
+    deliveryNotesGetFullList.mockResolvedValueOnce([]).mockResolvedValueOnce([{ id: "dnpartial", batch_note: "bn1" }]);
+
+    await expect(promoteNote(input())).rejects.toThrow("promotion_in_progress");
+    expect(deliveryNoteItemsGetFullList).not.toHaveBeenCalled();
+    expect(deliveryNoteItemsCreate).not.toHaveBeenCalled();
+    expect(deliveryNoteItemsDelete).not.toHaveBeenCalled();
+    expect(dnBatchNotesUpdate).not.toHaveBeenCalled();
+    expect(dnReviewEventsCreate).not.toHaveBeenCalled();
+  });
+
+  it("keeps a persisted review marker sufficient when the audit event write fails", async () => {
+    deliveryNoteItemsCreate.mockRejectedValueOnce(new Error("line write failed"));
+    dnReviewEventsCreate.mockRejectedValue(new Error("audit unavailable"));
+
+    await expect(promoteNote(input())).rejects.toThrow("line write failed");
+    expect(dnBatchNotesUpdate).toHaveBeenCalledWith("bn1", expect.objectContaining({ status: "blocked", stage: "needs_review" }));
+  });
+
+  it("keeps an audit event sufficient when candidate review-state writes fail", async () => {
+    dnBatchNotesUpdate.mockRejectedValue(new Error("candidate unavailable"));
+
+    await expect(promoteNote(input())).rejects.toThrow(/candidate unavailable|promotion failure persistence/);
+    expect(dnReviewEventsCreate).toHaveBeenCalled();
+  });
+
+  it("reports combined persistence failure when both review writes fail", async () => {
+    deliveryNoteItemsCreate.mockRejectedValueOnce(new Error("line write failed"));
+    dnReviewEventsCreate.mockRejectedValue(new Error("audit unavailable"));
+    dnBatchNotesUpdate.mockRejectedValue(new Error("candidate unavailable"));
+
+    await expect(promoteNote(input())).rejects.toThrow(/promotion failure persistence/);
+  });
+
+  it("does not relabel a candidate that becomes posted while failure handling runs", async () => {
+    deliveryNoteItemsCreate.mockRejectedValueOnce(new Error("line write failed"));
+    dnBatchNotesGetOne
+      .mockResolvedValueOnce({ id: "bn1", delivery_note: "dnpartial" })
+      .mockResolvedValueOnce({ id: "bn1", status: "promoted", stage: "posted", posted_at: "2026-09-08T00:00:00Z", delivery_note: "dnpartial" });
+
+    await expect(promoteNote(input())).rejects.toThrow("line write failed");
+    expect(dnBatchNotesUpdate).not.toHaveBeenCalledWith("bn1", expect.objectContaining({ status: "blocked" }));
+  });
 });

src/lib/dn-pipeline/promote.ts

@@ -49,6 +49,13 @@ type CandidateState = {
   posted_at?: string;
 };
 
+class PromotionInProgressError extends Error {
+  constructor() {
+    super("promotion_in_progress");
+    this.name = "PromotionInProgressError";
+  }
+}
+
 type PbCollection = ReturnType<PocketBase["collection"]> & {
   getFullList?: (options?: Record<string, unknown>) => Promise<Array<Record<string, unknown>>>;
 };
@@ -58,10 +65,7 @@ export async function promoteNote(input: PromoteInput): Promise<PromoteResult> {
 
   // The candidate row is the durable promotion key. A posted candidate is
   // immutable: retries may reconcile only an unposted partial write.
-  const candidate = (await pb
-    .collection("dn_batch_notes")
-    .getOne(batchNoteId, { fields: "id,delivery_note,status,stage,posted_at" })
-    .catch(() => null)) as CandidateState | null;
+  const candidate = await getCandidate(pb, batchNoteId);
   if (candidate && isPosted(candidate) && !input.explicitCorrection) {
     throw new Error("posted candidate is immutable without explicit correction");
   }
@@ -110,9 +114,21 @@ export async function promoteNote(input: PromoteInput): Promise<PromoteResult> {
 
   let dn: { id: string } | undefined;
   try {
-    dn = (existingDnId
-      ? await pb.collection("delivery_notes").update(existingDnId, form)
-      : await pb.collection("delivery_notes").create(form)) as { id: string };
+    if (existingDnId) {
+      dn = (await pb.collection("delivery_notes").update(existingDnId, form)) as { id: string };
+    } else {
+      try {
+        dn = (await pb.collection("delivery_notes").create(form)) as { id: string };
+      } catch (createError) {
+        // A unique batch_note error means another worker won the create race.
+        // Re-read the candidate and winner before touching any financial rows.
+        const winnerId = await findDeliveryNoteByCandidate(pb, batchNoteId);
+        if (!winnerId) throw createError;
+        const winner = await getCandidate(pb, batchNoteId);
+        if (winner && isPostedWinner(winner)) return loadExistingResult(pb, winnerId);
+        throw new PromotionInProgressError();
+      }
+    }
 
     // Replace the item set wholesale so a retry reconciles only this
     // candidate's unposted partial rows and cannot duplicate them.
@@ -161,27 +177,49 @@ export async function promoteNote(input: PromoteInput): Promise<PromoteResult> {
 
     return { deliveryNoteId: dn.id, itemsCreated: created };
   } catch (error) {
+    if (error instanceof PromotionInProgressError) throw error;
     const reason = error instanceof Error ? error.message : String(error);
     const recoveryDnId = dn?.id ?? existingDnId ?? null;
-    await recordPromotionFailure(pb, {
+    await persistPromotionFailure(pb, {
       batch: batchId,
       note: batchNoteId,
       deliveryNote: recoveryDnId,
       reason,
     });
-    // Keep the candidate review-visible while retaining its delivery_note
-    // relation (if one exists) so the next run can reconcile it.
-    await pb.collection("dn_batch_notes").update(batchNoteId, {
-      status: "blocked",
-      stage: "needs_review",
-      ...(recoveryDnId ? { delivery_note: recoveryDnId } : {}),
-    }).catch(() => undefined);
     throw error;
   }
 }
 
+async function getCandidate(pb: PocketBase, batchNoteId: string): Promise<CandidateState | null> {
+  return (await pb
+    .collection("dn_batch_notes")
+    .getOne(batchNoteId, { fields: "id,delivery_note,status,stage,posted_at" })
+    .catch(() => null)) as CandidateState | null;
+}
+
+async function getCandidateForFailure(
+  pb: PocketBase,
+  batchNoteId: string,
+): Promise<{ candidate: CandidateState | null; readFailed: boolean }> {
+  try {
+    return {
+      candidate: (await pb.collection("dn_batch_notes").getOne(batchNoteId, {
+        fields: "id,delivery_note,status,stage,posted_at",
+      })) as CandidateState,
+      readFailed: false,
+    };
+  } catch {
+    return { candidate: null, readFailed: true };
+  }
+}
+
 function isPosted(candidate: CandidateState): boolean {
-  return candidate.status === "promoted" || candidate.status === "posted" || candidate.stage === "posted" || Boolean(candidate.posted_at);
+  return candidate.status === "promoted" || candidate.status === "posted" || candidate.stage === "posted" ||
+    Boolean(candidate.posted_at);
+}
+
+function isPostedWinner(candidate: CandidateState): boolean {
+  return isPosted(candidate) || (Boolean(candidate.delivery_note) && !candidate.status && !candidate.stage && !candidate.posted_at);
 }
 
 async function findDeliveryNoteByCandidate(pb: PocketBase, batchNoteId: string): Promise<string | null> {
@@ -195,24 +233,69 @@ async function findDeliveryNoteByCandidate(pb: PocketBase, batchNoteId: string):
   return typeof id === "string" ? id : null;
 }
 
-async function recordPromotionFailure(
+async function loadExistingResult(pb: PocketBase, deliveryNoteId: string): Promise<PromoteResult> {
+  const rows = await pb.collection("delivery_note_items").getFullList({
+    filter: boundFilter(pb, "delivery_note = {:deliveryNote}", { deliveryNote: deliveryNoteId }),
+    fields: "id",
+  });
+  return { deliveryNoteId, itemsCreated: rows.length };
+}
+
+async function persistPromotionFailure(
   pb: PocketBase,
   fields: { batch: string; note: string; deliveryNote: string | null; reason: string },
 ): Promise<void> {
-  try {
-    await pb.collection("dn_review_events").create({
-      batch: fields.batch,
-      note: fields.note,
-      action: "promotion_failed",
-      before: { delivery_note: fields.deliveryNote, status: "candidate" },
-      after: { delivery_note: fields.deliveryNote, status: "blocked", stage: "needs_review" },
-      reason: fields.reason.slice(0, 500),
-      actor: "dn-pipeline",
-    });
-  } catch {
-    // The failed promotion itself must remain visible even on an older PB
-    // instance without the additive review-event collection.
+  let markerPersisted = false;
+  let eventPersisted = false;
+  let markerError: unknown = null;
+  let eventError: unknown = null;
+
+  // Never overwrite a candidate that another worker posted while this
+  // worker was handling its failure. The audit attempt remains independent.
+  const latest = await getCandidateForFailure(pb, fields.note);
+  if (!latest.readFailed && (!latest.candidate || !isPosted(latest.candidate))) {
+    const marker = {
+      status: "blocked",
+      stage: "needs_review",
+      ...(fields.deliveryNote ? { delivery_note: fields.deliveryNote } : {}),
+    };
+    for (let attempt = 0; attempt < 2 && !markerPersisted; attempt++) {
+      try {
+        await pb.collection("dn_batch_notes").update(fields.note, marker);
+        markerPersisted = true;
+      } catch (error) {
+        markerError = error;
+      }
+    }
   }
+
+  const event = {
+    batch: fields.batch,
+    note: fields.note,
+    action: "promotion_failed",
+    before: { delivery_note: fields.deliveryNote, status: "candidate" },
+    after: { delivery_note: fields.deliveryNote, status: "blocked", stage: "needs_review" },
+    reason: fields.reason.slice(0, 500),
+    actor: "dn-pipeline",
+  };
+  for (let attempt = 0; attempt < 2 && !eventPersisted; attempt++) {
+    try {
+      await pb.collection("dn_review_events").create(event);
+      eventPersisted = true;
+    } catch (error) {
+      eventError = error;
+    }
+  }
+
+  if (!markerPersisted && !eventPersisted) {
+    throw new Error(
+      `promotion failure persistence failed: marker=${formatFailure(markerError)}; event=${formatFailure(eventError)}; original=${fields.reason}`,
+    );
+  }
+}
+
+function formatFailure(error: unknown): string {
+  return error instanceof Error ? error.message : String(error ?? "not-attempted");
 }
 
 function boundFilter(pb: PocketBase, query: string, params: Record<string, unknown>): string {

src/lib/dn-pipeline/run.test.ts

@@ -99,6 +99,7 @@ function makePb() {
         };
       case "dn_batch_notes":
         return {
+          getOne: dnBatchNotesGetOne,
           getFullList: dnBatchNotesGetFullList,
           create: dnBatchNotesCreate,
           update: dnBatchNotesUpdate,
@@ -132,6 +133,7 @@ const purchaseOrdersGetFullList = vi.fn();
 const purchaseOrdersGetOne = vi.fn();
 const purchaseOrderItemsGetFullList = vi.fn();
 const dnBatchNotesGetFullList = vi.fn();
+const dnBatchNotesGetOne = vi.fn();
 const dnBatchNotesCreate = vi.fn();
 const dnBatchNotesUpdate = vi.fn();
 const dnItemAliasesGetFullList = vi.fn();
@@ -152,6 +154,7 @@ beforeEach(() => {
     purchaseOrdersGetOne,
     purchaseOrderItemsGetFullList,
     dnBatchNotesGetFullList,
+    dnBatchNotesGetOne,
     dnBatchNotesCreate,
     dnBatchNotesUpdate,
     dnItemAliasesGetFullList,
@@ -235,6 +238,7 @@ beforeEach(() => {
   supplierDocumentAliasesGetFullList.mockResolvedValue([]);
   deliveryNoteItemsGetFullList.mockResolvedValue([]);
   dnBatchNotesGetFullList.mockResolvedValue([]); // No existing → create.
+  dnBatchNotesGetOne.mockResolvedValue({ id: "bn1", status: "candidate" });
   dnBatchNotesCreate.mockResolvedValue({ id: "bn1" });
   dnBatchNotesUpdate.mockResolvedValue({ id: "bn1" });
 
@@ -586,6 +590,26 @@ describe("runBatch — per-note checks", () => {
     expect(form.page_to).toBe("3");
   });
 
+  it("passes authoritative raster MIME/name through promotion when the source is extensionless", async () => {
+    dnBatchesGetOne.mockResolvedValueOnce({ id: "batch1", site: "site1", file: "camera-capture" });
+    fetchMock.mockResolvedValueOnce(new Response(Buffer.from([0xff, 0xd8, 0xff, 0xd9]), {
+      status: 200,
+      headers: { "content-type": "application/octet-stream" },
+    }));
+    mExtractPageRange.mockResolvedValueOnce(Buffer.from([0xff, 0xd8, 0xff, 0xd9]));
+
+    await runBatch(makePb(), "batch1");
+
+    expect(mPromoteNote).toHaveBeenCalledWith(expect.objectContaining({
+      pageFileMime: "image/jpeg",
+      pageFileName: "camera-capture-note-1.jpg",
+    }));
+    const form = dnBatchNotesCreate.mock.calls[0][0] as FormData;
+    const file = form.get("page_file") as File;
+    expect(file.type).toBe("image/jpeg");
+    expect(file.name).toBe("camera-capture-note-1.jpg");
+  });
+
   it("names the per-note page file after the original batch filename", async () => {
     dnBatchesGetOne.mockResolvedValueOnce({
       id: "batch1",
@@ -885,6 +909,24 @@ describe("runBatch — batch checks and terminal status", () => {
     expect(lastBatchUpdate().notes_promoted).toBe(2);
   });
 
+  it("does not relabel a candidate that is posted while the runner handles a promotion error", async () => {
+    mRunNoteChecks.mockReturnValue([{ id: "x", passed: true, reason: "" }]);
+    mPromoteNote.mockRejectedValueOnce(new Error("promotion failed"));
+    dnBatchNotesGetOne.mockResolvedValueOnce({
+      id: "bn1",
+      status: "promoted",
+      stage: "posted",
+      posted_at: "2026-09-08T00:00:00Z",
+      delivery_note: "dn-posted",
+    });
+
+    const result = await runBatch(makePb(), "batch1");
+
+    expect(result.status).toBe("needs_attention");
+    expect(lastBatchUpdate().notes_promoted).toBe(0);
+    expect(dnBatchNotesUpdate).not.toHaveBeenCalledWith("bn1", expect.objectContaining({ status: "blocked" }));
+  });
+
   it("keeps needs_attention when a per-note check fails even though batch checks pass", async () => {
     mRunNoteChecks.mockReturnValueOnce([{ id: "po_resolved", passed: false, reason: "no" }]);
 
@@ -982,6 +1024,25 @@ describe("runBatch — a dismissal survives a re-run", () => {
 });
 
 describe("runBatch — a re-run must not block a note against its own ledger row", () => {
+  it("preserves a posted legacy candidate and counts it as reconciled", async () => {
+    dnBatchNotesGetFullList.mockResolvedValueOnce([{
+      id: "bn-posted",
+      status: "promoted",
+      stage: "posted",
+      posted_at: "2026-09-08T00:00:00Z",
+      delivery_note: "dn-posted",
+    }]);
+    deliveryNotesGetFullList.mockResolvedValue([{ delivery_note_number: "DQ0010107", batch: "batch1" }]);
+
+    const result = await runBatch(makePb(), "batch1");
+
+    expect(result.status).toBe("processed");
+    expect(result.notesPromoted).toBe(1);
+    expect(mPromoteNote).not.toHaveBeenCalled();
+    expect(dnBatchNotesUpdate).not.toHaveBeenCalledWith("bn-posted", expect.anything());
+    expect(dnBatchNotesCreate).not.toHaveBeenCalled();
+  });
+
   it("promotes a note whose only clashing ledger row is one it created itself", async () => {
     // Re-running a batch that had already promoted 8 notes took it to 0:
     // each note saw the delivery_notes row it created last time and

src/lib/dn-pipeline/run.ts

@@ -34,10 +34,11 @@ export interface RunBatchResult {
 export async function runBatch(pb: PocketBase, batchId: string): Promise<RunBatchResult> {
   const batch = await pb.collection("dn_batches").getOne(batchId);
   // getOne returns a generic RecordModel; we only need id/file here.
-  const { buffer, mime, filename } = await fetchBatchFile(
+  const { buffer, mime: declaredMime, filename } = await fetchBatchFile(
     pb,
     batch as unknown as { id: string; file: string },
   );
+  const mime = authoritativeMime(buffer, declaredMime, filename);
 
   // 1. Read every page independently.
   const images = await splitPages(buffer, mime);
@@ -143,14 +144,24 @@ export async function runBatch(pb: PocketBase, batchId: string): Promise<RunBatc
       pageFileMime: mime,
     });
 
-    allChecks.push(...checks.map((c) => ({ ...c, id: `note${note.note_index + 1}.${c.id}` })));
-
     // A dismissed note stands down. Promoting it would put a duplicate
     // on the ledger — the exact double count it was dismissed to avoid —
     // and re-reading the page tells us nothing about a decision a person
     // already made.
     if (candidate.dismissed) continue;
 
+    // Terminal legacy/staged candidates are immutable. Their existing
+    // financial row is already reconciled, even if this rerun extracts a
+    // different page file or now reports a different check result.
+    if (candidate.posted) {
+      promoted++;
+      existingNumbers.add(note.delivery_note_number);
+      existingNumbers.documentIdentities.add(documentIdentity(resolution.supplierId, note.delivery_note_number));
+      continue;
+    }
+
+    allChecks.push(...checks.map((c) => ({ ...c, id: `note${note.note_index + 1}.${c.id}` })));
+
     if (!blocked) {
       try {
         await promoteNote({
@@ -163,6 +174,7 @@ export async function runBatch(pb: PocketBase, batchId: string): Promise<RunBatc
           poItems: poItems.map((p) => ({ id: p.id, unit: p.unit, description: p.description })),
           pageFile,
           pageFileName,
+          pageFileMime: mime,
           lineMatches,
         });
         promoted++;
@@ -187,10 +199,9 @@ export async function runBatch(pb: PocketBase, batchId: string): Promise<RunBatc
           passed: false,
           reason: `could not write this note: ${message}`.slice(0, 500),
         });
-        await pb
-          .collection("dn_batch_notes")
-          .update(candidate.id, { status: "blocked" })
-          .catch(() => undefined);
+        if (message !== "promotion_in_progress") {
+          await markCandidateBlockedUnlessPosted(pb, candidate.id);
+        }
       }
     }
   }
@@ -393,13 +404,37 @@ function safeSiteFilter(pb: PocketBase, siteId: string): string {
 function candidateFileName(filename: string, mime: string, noteIndex: number): string {
   const clean = filename.split("/").pop() ?? filename;
   const base = clean.replace(/\.[^.]+$/, "");
-  const originalExt = clean.match(/\.([a-z0-9]+)$/i)?.[1];
-  const extension = mime === "application/pdf"
-    ? "pdf"
-    : (originalExt || mime.split("/")[1] || "bin");
+  const extension = extensionForMime(mime);
   return `${base}-note-${noteIndex + 1}.${extension}`;
 }
 
+function extensionForMime(mime: string): string {
+  const normal = mime.split(";", 1)[0].toLowerCase();
+  const known: Record<string, string> = {
+    "application/pdf": "pdf",
+    "image/jpeg": "jpg",
+    "image/png": "png",
+    "image/webp": "webp",
+    "image/gif": "gif",
+    "image/avif": "avif",
+    "image/bmp": "bmp",
+    "image/tiff": "tiff",
+  };
+  return known[normal] ?? normal.split("/")[1] ?? "bin";
+}
+
+function authoritativeMime(buffer: Buffer, declared: string, filename: string): string {
+  if (buffer.subarray(0, 5).equals(Buffer.from("%PDF-"))) return "application/pdf";
+  if (buffer.subarray(0, 3).equals(Buffer.from([0xff, 0xd8, 0xff]))) return "image/jpeg";
+  if (buffer.subarray(0, 8).equals(Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]))) return "image/png";
+  if (buffer.subarray(0, 4).toString("ascii") === "RIFF" && buffer.subarray(8, 12).toString("ascii") === "WEBP") return "image/webp";
+  if (buffer.subarray(0, 3).toString("ascii") === "GIF") return "image/gif";
+  if (buffer.subarray(0, 2).toString("ascii") === "BM") return "image/bmp";
+  if (declared && declared !== "application/octet-stream") return declared.split(";", 1)[0].toLowerCase();
+  const extension = filename.toLowerCase().match(/\.([a-z0-9]+)$/)?.[1];
+  return extension === "pdf" ? "application/pdf" : `image/${extension || "jpeg"}`;
+}
+
 async function upsertCandidate(
   pb: PocketBase,
   batchId: string,
@@ -422,7 +457,7 @@ async function upsertCandidate(
     .collection("dn_batch_notes")
     .getFullList({
       filter: boundFilter(pb, "batch = {:batch} && note_index = {:noteIndex}", { batch: batchId, noteIndex: note.note_index }),
-      fields: "id,status",
+      fields: "id,status,stage,posted_at,delivery_note",
     });
 
   // A dismissal is a human decision — "this is a duplicate of one
@@ -432,6 +467,17 @@ async function upsertCandidate(
   // verified dismissals and put them straight back in the queue.
   const wasDismissed =
     existing.length > 0 && (existing[0] as { status?: string }).status === "dismissed";
+  const existingCandidate = existing[0] as {
+    id?: string;
+    status?: string;
+    stage?: string;
+    posted_at?: string;
+    delivery_note?: string;
+  } | undefined;
+  const posted = Boolean(existingCandidate && isPostedCandidate(existingCandidate));
+  if (posted && existingCandidate?.id) {
+    return { id: existingCandidate.id, dismissed: false, posted: true };
+  }
 
   const form = new FormData();
   const fields: Record<string, unknown> = {
@@ -475,7 +521,28 @@ async function upsertCandidate(
     ? pb.collection("dn_batch_notes").update(existing[0].id, form)
     : pb.collection("dn_batch_notes").create(form));
 
-  return { id: record.id as string, dismissed: wasDismissed };
+  return { id: record.id as string, dismissed: wasDismissed, posted: false };
+}
+
+function isPostedCandidate(candidate: { status?: string; stage?: string; posted_at?: string; delivery_note?: string }): boolean {
+  return candidate.status === "promoted" || candidate.status === "posted" || candidate.stage === "posted" ||
+    Boolean(candidate.posted_at) || (Boolean(candidate.delivery_note) && !candidate.status && !candidate.stage);
+}
+
+async function markCandidateBlockedUnlessPosted(pb: PocketBase, candidateId: string): Promise<void> {
+  try {
+    const collection = pb.collection("dn_batch_notes") as unknown as {
+      getOne?: (id: string, options?: Record<string, unknown>) => Promise<Record<string, unknown>>;
+      update: (id: string, body: Record<string, unknown>) => Promise<unknown>;
+    };
+    if (typeof collection.getOne !== "function") return;
+    const current = await collection.getOne(candidateId, { fields: "status,stage,posted_at,delivery_note" });
+    if (isPostedCandidate(current)) return;
+    await collection.update(candidateId, { status: "blocked", stage: "needs_review" });
+  } catch {
+    // promoteNote owns durable failure visibility. The runner must not turn
+    // an unavailable read into a potentially unsafe overwrite.
+  }
 }
 
 function boundFilter(pb: PocketBase, query: string, params: Record<string, unknown>): string {