From b60f7fd3ce442d59bf8b84fbfd5a3e0e2fd6b4a1 Mon Sep 17 00:00:00 2001 From: Dave Date: Sat, 17 Oct 2026 09:24:04 +0200 Subject: [PATCH] fix(collection-ai): keep reads on errors, record pairing failures, budget reads Co-Authored-By: Claude Opus 5.5 --- src/lib/server/collection-ai-service.test.ts | 107 ++++++++++- src/lib/server/collection-ai-service.ts | 183 ++++++++++++------- 2 files changed, 214 insertions(+), 76 deletions(-) diff --git a/src/lib/server/collection-ai-service.test.ts b/src/lib/server/collection-ai-service.test.ts index 6596cab..52c776d 100644 --- a/src/lib/server/collection-ai-service.test.ts +++ b/src/lib/server/collection-ai-service.test.ts @@ -14,6 +14,10 @@ let currentActor: Actor | null = null; let orderItemFields = ["description", "ai_check"]; let schemaChecks = 0; let idSeq = 0; +let failSkuFill = false; +/** Throw from the content_hash lookup after this many lookups (null = never). */ +let hashLookupFailsAfter: number | null = null; +let hashLookups = 0; function table(name: string): Map { let t = db.get(name); @@ -159,9 +163,15 @@ const pb = { collection(name: string) { return { getFullList: async (opts: { filter?: string; sort?: string } = {}) => list(name, opts), - getList: async (_page: number, perPage: number, opts: { filter?: string; sort?: string } = {}) => ({ - items: list(name, opts).slice(0, perPage), - }), + getList: async (_page: number, perPage: number, opts: { filter?: string; sort?: string } = {}) => { + if (name === "collection_doc_reads" && opts.filter?.startsWith("content_hash")) { + hashLookups += 1; + if (hashLookupFailsAfter !== null && hashLookups > hashLookupFailsAfter) { + throw new Error("hash lookup down"); + } + } + return { items: list(name, opts).slice(0, perPage) }; + }, getOne: async (id: string) => { const r = table(name).get(id); if (!r) throw new Error(`${name}/${id} not found`); @@ -202,6 +212,7 @@ vi.mock("@/lib/server/audit-context", () => ({ const skuCalls: { id: string; sku: string; actor: { id: string; name: string }; audit: Actor | null }[] = []; vi.mock("@/lib/server/collection-confirm-service", () => ({ updateOrderItemSku: async (id: string, args: { sku: string; actor: { id: string; name: string } }) => { + if (failSkuFill) throw new Error("sku boom"); skuCalls.push({ id, sku: args.sku, actor: args.actor, audit: currentActor }); const r = table("order_items").get(id)!; r.sku = args.sku; @@ -443,6 +454,9 @@ beforeEach(() => { orderItemFields = ["description", "ai_check"]; schemaChecks = 0; idSeq = 0; + failSkuFill = false; + hashLookupFailsAfter = null; + hashLookups = 0; pairPlan = { 1: { id: "l1", same: "sure" }, 2: { id: "l2", same: "sure" }, @@ -668,6 +682,47 @@ describe("checkCollectionCard", () => { expect(aiCheck("order_items", "l1")?.sku_filled_value).toBe("SM293306030"); }); + it("keeps a SKU a rep sets while the check is running (no AI fill, no marker)", async () => { + seedNss(); + const { deps } = makeDeps(); + const pairImpl = deps.pairLines.getMockImplementation()!; + deps.pairLines.mockImplementation(async (docLines, poLines) => { + table("order_items").get("l1")!.sku = "REP"; + return pairImpl(docLines, poLines); + }); + + await checkCollectionCard(PO_CEM, { deps }); + + expect(skuCalls).toEqual([]); + expect(table("order_items").get("l1")?.sku).toBe("REP"); + expect(aiCheck("order_items", "l1")?.sku_filled_value).toBe(""); + }); + + it("still writes the line's check when the SKU fill fails", async () => { + seedNss(); + const { deps } = makeDeps(); + failSkuFill = true; + + const r = await checkCollectionCard(PO_CEM, { deps }); + + expect(r.error).toMatch(/SKU fill failed for line l1: sku boom/); + expect(aiCheck("order_items", "l1")).toMatchObject({ sku_filled_value: "", same_item: "sure" }); + expect(r.skus_filled).toBe(0); + }); + + it("keeps reads already made when a later file's hash lookup throws", async () => { + seedNss({ attachments: ["first.jpg", "second.jpg"] }); + const { deps } = makeDeps(); + hashLookupFailsAfter = 1; + + const r = await checkCollectionCard(PO_CEM, { deps }); + + expect(deps.readDocument).toHaveBeenCalledTimes(1); + expect(r.error).toMatch(/second\.jpg: hash lookup down/); + expect(reads().map((x) => x.filename)).toEqual(["first.jpg"]); + expect(reads()[0].status).toBe("read"); + }); + it("never fills on a 'different' or 'likely' pairing", async () => { seedNss(); table("order_items").get("l3")!.sku = ""; @@ -916,8 +971,50 @@ describe("runCollectionAiTick", () => { expect(r.summary?.cards_checked).toBe(2); expect(r.summary?.errors).toEqual(["PO202610002XX: pairing failed: pairing exploded"]); - expect(aiCheck("order_items", "c1")).not.toBeNull(); - expect(aiCheck("order_items", "c2")).toBeNull(); + expect(aiCheck("order_items", "c1")?.doc_line).not.toBeNull(); + // The failure is recorded on the line (all red), not silently dropped. + expect(aiCheck("order_items", "c2")).toMatchObject({ + doc_line: null, + findings: ["AI couldn't pair this document: pairing exploded"], + }); + }); + + it("does not retry a card whose pairing failed on the next tick; a manual check does", async () => { + seedCards(1, { withRead: true }); + const pairLines = vi.fn(async () => { + throw new Error("pairing exploded"); + }); + const { deps } = makeDeps({ pairLines }); + + await runCollectionAiTick(deps); + expect(pairLines).toHaveBeenCalledTimes(1); + + settings.collection_ai_last_run_at = ""; + const second = await runCollectionAiTick(deps); + expect(pairLines).toHaveBeenCalledTimes(1); + expect(second.summary?.cards_checked).toBe(0); + + await checkCollectionCard("PO202610001XX", { deps }); + expect(pairLines).toHaveBeenCalledTimes(2); + }); + + it("stops starting new reads inside a card once the time budget has passed", async () => { + addPo("pob", "PO202610055XX"); + addOrderLine("b1", "pob", { order_item_attachements: ["one.jpg", "two.jpg"] }); + let clock = NOW.getTime(); + const { deps } = makeDeps({ now: () => new Date(clock) }); + const readImpl = deps.readDocument.getMockImplementation()!; + deps.readDocument.mockImplementation(async (img) => { + clock += TICK_TIME_BUDGET_MS; + return readImpl(img); + }); + + const r = await runCollectionAiTick(deps); + + expect(deps.readDocument).toHaveBeenCalledTimes(1); + expect(r.summary?.files_read).toBe(1); + // The finished read is kept. + expect(reads().map((x) => x.filename)).toEqual(["one.jpg"]); }); it("keeps the summary bounded: at most 20 errors of at most 300 characters", async () => { diff --git a/src/lib/server/collection-ai-service.ts b/src/lib/server/collection-ai-service.ts index 4536f46..520e26b 100644 --- a/src/lib/server/collection-ai-service.ts +++ b/src/lib/server/collection-ai-service.ts @@ -485,7 +485,14 @@ export async function checkCollectionCard( async function checkCard( poNumber: string, - opts: { actorName?: string; maxFiles?: number; deps?: CollectionAiDeps; retryFailed: boolean }, + opts: { + actorName?: string; + maxFiles?: number; + deps?: CollectionAiDeps; + retryFailed: boolean; + /** Epoch ms after which no new file read is started (tick budget). */ + deadline?: number; + }, ): Promise { const deps = opts.deps ?? {}; const readDocument = deps.readDocument ?? readDocumentImage; @@ -528,75 +535,84 @@ async function checkCard( const cardLines = await loadLinesForPos(pb, cardPos, true); if (cardLines.length === 0) return result; - // 2. Read each file once. Writes are queued so "AI off" writes nothing. + // 2. Read each file once. Writes are queued so "AI off" writes nothing; + // on any other exit the queue is flushed, so a read is never lost. const files = filesForLines(cardLines, stockPurchases); const readsByFile = await loadReadsForFiles(pb, files); const byHash = new Map(); const queued: Array<{ file: FileRef; fields: ReadFields; replaceId: string | null }> = []; - try { - for (const file of files) { - const existing = readsByFile.get(fileKey(file)); - if (existing && !(opts.retryFailed && existing.status === "failed")) continue; - const replaceId = existing?.id ?? null; - const extension = ext(file.filename); - if (!IMAGE_EXT.has(extension)) { - const fields = emptyRead(file, now); - fields.notes = [extension === "pdf" ? PDF_NOTE : `Not an image (.${extension || "?"})`]; - queued.push({ file, fields, replaceId }); - continue; - } - if (result.files_read >= maxFiles) continue; - let fetched: Awaited>>; - try { - fetched = await fetchFile({ collection: file.collection, id: file.id }, file.filename); - } catch (e) { - // Transient: no row, so the next run tries again. - fail(`${file.filename}: ${errMessage(e)}`); - continue; - } - const hash = createHash("sha256").update(fetched.bytes).digest("hex"); - let known = byHash.get(hash) ?? null; - if (!known) { - const found = await findReadByHash(pb, hash); - if (found) { - const rest: Partial = { ...found }; - delete rest.id; - known = rest as ReadFields; - } - } - if (known) { - queued.push({ file, fields: copyOf(file, known), replaceId }); - continue; - } - - let fields: ReadFields; - try { - const read = await readDocument({ base64: fetched.base64, contentType: fetched.contentType }); - fields = fromResult(file, hash, read, now); - result.files_read += 1; - } catch (e) { - if (e instanceof AiDisabledError) throw new DisabledSignal(); - const msg = errMessage(e); - fields = { ...emptyRead(file, now), content_hash: hash }; - if (msg.startsWith("Unsupported image type")) { - fields.notes = [msg]; - } else { - fields.status = "failed"; - fields.error = msg.slice(0, 2000); - fail(`read failed for ${file.filename}: ${msg}`); - result.files_read += 1; - } - } - if (fields.status !== "failed") byHash.set(hash, fields); + const readOneFile = async (file: FileRef): Promise => { + const existing = readsByFile.get(fileKey(file)); + if (existing && !(opts.retryFailed && existing.status === "failed")) return; + const replaceId = existing?.id ?? null; + const extension = ext(file.filename); + if (!IMAGE_EXT.has(extension)) { + const fields = emptyRead(file, now); + fields.notes = [extension === "pdf" ? PDF_NOTE : `Not an image (.${extension || "?"})`]; queued.push({ file, fields, replaceId }); + return; } - } catch (e) { - if (e instanceof DisabledSignal) { - result.error = COLLECTION_AI_DISABLED_ERROR; - return result; + if (result.files_read >= maxFiles) return; + // Time budget: never start a new read late in a tick. + if (opts.deadline !== undefined && nowFn().getTime() >= opts.deadline) return; + + let fetched: Awaited>>; + try { + fetched = await fetchFile({ collection: file.collection, id: file.id }, file.filename); + } catch (e) { + // Transient: no row, so the next run tries again. + fail(`${file.filename}: ${errMessage(e)}`); + return; + } + const hash = createHash("sha256").update(fetched.bytes).digest("hex"); + let known = byHash.get(hash) ?? null; + if (!known) { + const found = await findReadByHash(pb, hash); + if (found) { + const rest: Partial = { ...found }; + delete rest.id; + known = rest as ReadFields; + } + } + if (known) { + queued.push({ file, fields: copyOf(file, known), replaceId }); + return; + } + + let fields: ReadFields; + try { + const read = await readDocument({ base64: fetched.base64, contentType: fetched.contentType }); + fields = fromResult(file, hash, read, now); + result.files_read += 1; + } catch (e) { + if (e instanceof AiDisabledError) throw new DisabledSignal(); + const msg = errMessage(e); + fields = { ...emptyRead(file, now), content_hash: hash }; + if (msg.startsWith("Unsupported image type")) { + fields.notes = [msg]; + } else { + fields.status = "failed"; + fields.error = msg.slice(0, 2000); + fail(`read failed for ${file.filename}: ${msg}`); + result.files_read += 1; + } + } + if (fields.status !== "failed") byHash.set(hash, fields); + queued.push({ file, fields, replaceId }); + }; + + for (const file of files) { + try { + await readOneFile(file); + } catch (e) { + if (e instanceof DisabledSignal) { + result.error = COLLECTION_AI_DISABLED_ERROR; + return result; + } + // One file's lookup failing must not drop the reads already made. + fail(`${file.filename}: ${errMessage(e)}`); } - throw e; } for (const { file, fields, replaceId } of queued) { try { @@ -629,6 +645,7 @@ async function checkCard( const lineDocs = new Map(); const linesById = new Map(cardLines.map((l) => [l.id, l])); const checkedDocs: CollectionDocRead[] = []; + const pairFailures: Array<{ doc: CollectionDocRead; finding: string }> = []; for (const doc of docs) { try { @@ -666,6 +683,12 @@ async function checkCard( return result; } fail(`pairing failed: ${errMessage(e)}`); + // Recorded on the lines below, so the tick doesn't retry this + // card every run; a manual "Check with AI" always re-pairs. + pairFailures.push({ + doc, + finding: `AI couldn't pair this document: ${errMessage(e)}`.slice(0, 500), + }); continue; } @@ -699,8 +722,11 @@ async function checkCard( fail(`document ${doc.document_no || doc.filename}: ${errMessage(e)}`); } } - if (checkedDocs.length === 0) return result; - const newestDoc = checkedDocs.reduce((a, b) => (newer(b, a) ? b : a)); + if (checkedDocs.length === 0 && pairFailures.length === 0) return result; + const newestDoc = (checkedDocs.length > 0 ? checkedDocs : pairFailures.map((f) => f.doc)).reduce((a, b) => + newer(b, a) ? b : a, + ); + const failureFindings = pairFailures.map((f) => f.finding); // 5. Write ai_check on every collected line (card + other POs the documents named). for (const line of linesById.values()) { @@ -715,20 +741,34 @@ async function checkCard( const fresh = await pb.collection(line.collection).getOne(line.id); const ownSku = str(fresh.sku).trim(); const previous = (fresh.ai_check as CollectionAiCheck | null) ?? null; + // Pairing failed outright: only mark lines that have no check yet, + // so a transient failure never wipes an earlier good check. + if (checkedDocs.length === 0 && previous) continue; // The robot marker survives a re-run only while the SKU is still the AI's. let skuFilled = previous?.sku_filled_value && str(fresh.sku) === previous.sku_filled_value ? previous.sku_filled_value : ""; const code = chosen.docLine?.code.trim() ?? ""; if (!ownSku && code && code.length <= MAX_SKU_LENGTH && chosen.decision?.same_item === "sure") { - if (line.kind === "customer") { - await updateOrderItemSku(line.id, { sku: code, actor: { id: SKU_ACTOR_ID, name: actorName } }); - } else { - await updateStockPurchaseItem({ id: line.id, patch: { sku: code } }); + // A failed fill must not cost the line its check. + try { + if (line.kind === "customer") { + await updateOrderItemSku(line.id, { sku: code, actor: { id: SKU_ACTOR_ID, name: actorName } }); + } else { + await updateStockPurchaseItem({ id: line.id, patch: { sku: code } }); + } + skuFilled = code; + result.skus_filled += 1; + } catch (e) { + skuFilled = ""; + fail(`SKU fill failed for line ${line.id}: ${errMessage(e)}`); } - skuFilled = code; - result.skus_filled += 1; } const docNo = chosen.doc.document_no || "the document"; + const lineFindings = chosen.docLine + ? [] + : checkedDocs.length > 0 + ? [`Not found on ${docNo}.`, ...failureFindings] + : failureFindings; const check: CollectionAiCheck = { read_id: chosen.doc.id, document_no: chosen.doc.document_no, @@ -738,7 +778,7 @@ async function checkCard( same_item: chosen.decision?.same_item ?? null, reason: chosen.decision?.reason ?? "", sku_filled_value: skuFilled, - findings: chosen.docLine ? [] : [`Not found on ${docNo}.`], + findings: lineFindings, checked_at: now.toISOString(), model: chosen.model, }; @@ -804,6 +844,7 @@ export async function runCollectionAiTick( try { const r = await checkCard(poNumber, { maxFiles: TICK_MAX_FILES - summary.files_read, + deadline: startMs + TICK_TIME_BUDGET_MS, deps, retryFailed: false, });