From b1d4eadb93fcf2693c98ee7b9cf8980d82a082dd Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Wed, 23 Sep 2026 12:21:36 +0300 Subject: [PATCH] embedNewNodes: usage-on-error + pay-once (partial-failure accounting) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Teacher's Wave-6 review flagged a money-ledger hole: a mid-batch embed throw discarded the spend of batches that had already landed (extraction only sees success results). Owner ruling 2026-09-23 (option A, pay-once): the landed vectors persist too — a retry embeds only what remains, so each text is billed exactly once. - embedNewNodes never throws on embed failures: per-batch try/catch, result gains `failed` (error message), `embedded` counts what actually landed, and usage carries the completed batches' steps either way (usage-on-error). - Zero-landed failures write nothing — the old ledger (its own model/dim) survives untouched. Stale-wipe semantics unchanged. - scripts/embed-graph.ts: honest failure output (lands count, not the in-memory total), exit 1 on failure. - tests: 4 new cases — mid-run failure (32 land, 8 remain; retry resumes and re-bills only the rest), first-batch failure (no write, no spend, old vectors survive), success carries no failed key. Live-verified on port 3004: bogus-model 400 from OpenRouter → exit 1, ledger untouched; healthy re-run embeds 0 (no re-spend); reset scratch setting to null. Gates: typecheck, lint, 899/899 tests, build. --- scripts/embed-graph.ts | 17 +++++++--- src/server/embeddings.ts | 55 ++++++++++++++++++++++++------ tests/embeddings.test.ts | 73 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 131 insertions(+), 14 deletions(-) diff --git a/scripts/embed-graph.ts b/scripts/embed-graph.ts index 60f7a9e..b739722 100644 --- a/scripts/embed-graph.ts +++ b/scripts/embed-graph.ts @@ -22,14 +22,23 @@ async function main() { } const t0 = Date.now(); console.log(`[embed] model: ${resolved.port.model} (provider embeddings, batch 32)`); - const { embedded, total, usage } = await embedNewNodes(graphStore); - console.log( - `[embed] embedded ${embedded} nodes → ledger holds ${total} vectors (${((Date.now() - t0) / 1000).toFixed(1)}s)`, - ); + const { embedded, total, usage, failed } = await embedNewNodes(graphStore); + const secs = ((Date.now() - t0) / 1000).toFixed(1); if (usage) { const tokens = usage.steps.reduce((sum, s) => sum + (s.inputTokens ?? 0), 0); console.log(`[embed] embedding spend: ${tokens} prompt tokens over ${usage.steps.length} call(s)`); } + // Usage-on-error + pay-once (owner ruling 2026-09-23): a mid-run provider + // failure still reports the landed batches' spend above; the landed + // vectors are persisted, so the honest count is what LANDED — the + // in-memory `total` would read as a ledger claim it can't make. + if (failed) { + console.error( + `[embed] failed after ${embedded} node(s) landed (${secs}s) — the ledger keeps exactly those (a retry embeds only the rest): ${failed}`, + ); + process.exit(1); + } + console.log(`[embed] embedded ${embedded} nodes → ledger holds ${total} vectors (${secs}s)`); } main().catch((err) => { diff --git a/src/server/embeddings.ts b/src/server/embeddings.ts index d09fd71..30fb4eb 100644 --- a/src/server/embeddings.ts +++ b/src/server/embeddings.ts @@ -269,13 +269,21 @@ export function embeddingPort(): EmbeddingResolution { // ─── The post-merge batch (the extraction pipeline's embedding tail) ──────── /** What the embedding tail reports back: counts for logs, the skip reason - * when nothing ran (owner ruling: null model = info skip, not error), and - * the money-channel stage when it billed (one CallUsage step per API call, - * outputTokens: 0 — priced via EMBEDDING_PRICES). */ + * when nothing ran (owner ruling: null model = info skip, not error), the + * failure message when it died partway (owner ruling 2026-09-23: + * usage-on-error — a multi-batch embed that fails mid-run still reports + * the spend of the batches that landed, so the money channel never + * loses a billed call), and the money-channel stage when it billed (one + * CallUsage step per API call, outputTokens: 0 — priced via + * EMBEDDING_PRICES). */ export interface EmbedBatchResult { embedded: number; total: number; skipped?: string; + /** Set when an embed batch failed after others may have landed. The + * landed vectors are already persisted (pay-once, owner ruling A + * 2026-09-23) — the next run embeds only what remains. */ + failed?: string; usage?: ExtractionStageUsage; } @@ -288,6 +296,11 @@ export interface EmbedBatchResult { * exist, then re-clusters ONCE per batch (consumers read the persisted * partition). Non-fatal by contract: the caller treats failure as * "clustering runs without the new nodes until the next merge retries". + * Usage-on-error + pay-once (owner ruling 2026-09-23): an embed batch + * that throws never propagates — the batches that already landed are + * persisted (each text is paid for exactly once; a retry continues with + * the remaining nodes), their spend still reaches the money channel via + * usage, and `failed` carries the error for logs and callers. */ export async function embedNewNodes( graphStore: { @@ -320,14 +333,26 @@ export async function embedNewNodes( if (!vectors[node.id]) todo.push(node); } + let landed = 0; + let failed: string | undefined; for (let i = 0; i < todo.length; i += EMBED_BATCH) { const batch = todo.slice(i, i + EMBED_BATCH); - const result = await port.embed( - batch.map((n) => n.statement), - "document", - ); + let result: EmbeddingResult; + try { + result = await port.embed( + batch.map((n) => n.statement), + "document", + ); + } catch (err) { + // Usage-on-error + pay-once (owner ruling 2026-09-23): stop + // here, keep the batches that landed, report the failure — the + // throw never escapes this function. + failed = err instanceof Error ? err.message : String(err); + break; + } for (let j = 0; j < batch.length; j++) { vectors[batch[j]!.id] = result.vectors[j]!; + landed++; } if (result.vectors[0]) dim = result.vectors[0].length; if (result.promptTokens !== undefined) { @@ -336,6 +361,11 @@ export async function embedNewNodes( steps.push(embeddingStep(result.promptTokens)); } } + if (failed !== undefined) { + console.error( + `[embeddings] embedding failed after ${landed}/${todo.length} nodes (non-fatal; landed vectors persist, the next run embeds only what remains): ${failed}`, + ); + } // Prune ghosts: vectors for node ids that no longer exist. let pruned = 0; @@ -351,7 +381,10 @@ export async function embedNewNodes( // consumers (snapshot, settle, namer) read the persisted partition. const clusters = semanticClusters(vectors, { similarityThreshold: SEMANTIC_SIM_THRESHOLD }); - const wrote = todo.length > 0 || pruned > 0 || stale || ledger.clusters.length !== clusters.length; + const wrote = + failed !== undefined + ? landed > 0 + : todo.length > 0 || pruned > 0 || stale || ledger.clusters.length !== clusters.length; if (wrote) { graphStore.writeEmbeddings({ model: port.model, @@ -361,11 +394,13 @@ export async function embedNewNodes( updatedAt: new Date().toISOString(), }); } - return { - embedded: todo.length, + const result: EmbedBatchResult = { + embedded: landed, total: Object.keys(vectors).length, usage: steps.length > 0 ? { model: port.model, steps } : undefined, }; + if (failed !== undefined) result.failed = failed; + return result; } // ─── The federated embedding catalog (wave 6, owner ruling 2026-09-23) ────── diff --git a/tests/embeddings.test.ts b/tests/embeddings.test.ts index 1990d7e..d807037 100644 --- a/tests/embeddings.test.ts +++ b/tests/embeddings.test.ts @@ -109,6 +109,79 @@ function scriptedPort(model = "openai:test-embed", dim = 2, tokensPerCall = 120) const okPort = (port: EmbeddingPort): EmbeddingResolution => ({ ok: true, port }); +// ─── usage-on-error + pay-once (owner ruling 2026-09-23, interlude) ────────── + +describe("embedNewNodes partial-failure accounting (Teacher review follow-up)", () => { + it("a mid-run batch failure: landed vectors persist, usage survives, failed carries the error, next run resumes", async () => { + // 40 nodes → batch 1 (32) lands, batch 2 (8) throws. + const many: GraphNode[] = Array.from({ length: 40 }, (_, i) => node(`n${i}`, `statement ${i}`)); + const ledger: EmbeddingsLedger = { model: "m", dim: 2, vectors: {}, clusters: [], updatedAt: "" }; + let writes = 0; + let call = 0; + const failing: EmbeddingPort = { + model: "m", + async embed(texts) { + call++; + if (call === 2) throw new Error("provider exploded mid-run"); + return { + vectors: texts.map((_, i) => (i === 0 ? [1, 0] : [0, 1])), + model: "m", + dim: 2, + promptTokens: 100, + }; + }, + }; + const store = { + readNodes: () => many, + readEmbeddings: () => ledger, + writeEmbeddings: (next: EmbeddingsLedger) => { + writes++; + Object.assign(ledger, next); + }, + }; + const result = await embedNewNodes(store, () => okPort(failing)); + // Usage-on-error: batch 1 billed — the money channel keeps it. + assert.equal(result.failed, "provider exploded mid-run"); + assert.equal(result.embedded, 32, "only the landed batch counts as embedded"); + assert.equal(result.usage?.steps.length, 1); + assert.equal(result.usage?.steps[0]?.inputTokens, 100); + // Pay-once: the landed vectors are persisted — a retry embeds only the rest. + assert.equal(writes, 1); + assert.equal(Object.keys(ledger.vectors).length, 32); + assert.equal(ledger.model, "m"); + const resume = await embedNewNodes(store, () => okPort(scriptedPort("m", 2).port)); + assert.equal(resume.embedded, 8, "the retry resumes — it re-embeds only the 8 remaining"); + assert.equal(resume.failed, undefined); + assert.equal(Object.keys(ledger.vectors).length, 40); + assert.equal(resume.usage?.steps[0]?.inputTokens, 120, "scripted port tokens — the retry billed once"); + }); + + it("a FIRST-batch failure: nothing written, nothing billed, the old ledger stays intact", async () => { + const g = fakeGraphStore({ vectors: { n1: [1, 0] } }); + const alwaysFails: EmbeddingPort = { + model: "m", + embed: async () => { + throw new Error("total outage"); + }, + }; + const result = await embedNewNodes(g.store, () => okPort(alwaysFails)); + assert.equal(result.failed, "total outage"); + assert.equal(result.embedded, 0); + assert.equal(result.usage, undefined, "no call landed — nothing to bill"); + assert.equal(g.writes.length, 0, "the ledger is untouched (its own model/dim still true)"); + assert.ok(g.ledger.vectors.n1, "the old vector survives a zero-landed failure"); + }); + + it("success still carries no `failed` key and counts every todo node", async () => { + const g = fakeGraphStore(); + const scripted = scriptedPort(g.ledger.model, 2); + const result = await embedNewNodes(g.store, () => okPort(scripted.port)); + assert.equal(result.failed, undefined); + assert.equal(result.embedded, 2); + assert.equal(result.skipped, undefined); + }); +}); + // ─── the money channel's embedding overlay ───────────────────────────────── describe("the embedding price overlay (hand-maintained, outside the sync block)", () => {