embedNewNodes: usage-on-error + pay-once (partial-failure accounting)
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.
This commit is contained in:
+13
-4
@@ -22,14 +22,23 @@ async function main() {
|
|||||||
}
|
}
|
||||||
const t0 = Date.now();
|
const t0 = Date.now();
|
||||||
console.log(`[embed] model: ${resolved.port.model} (provider embeddings, batch 32)`);
|
console.log(`[embed] model: ${resolved.port.model} (provider embeddings, batch 32)`);
|
||||||
const { embedded, total, usage } = await embedNewNodes(graphStore);
|
const { embedded, total, usage, failed } = await embedNewNodes(graphStore);
|
||||||
console.log(
|
const secs = ((Date.now() - t0) / 1000).toFixed(1);
|
||||||
`[embed] embedded ${embedded} nodes → ledger holds ${total} vectors (${((Date.now() - t0) / 1000).toFixed(1)}s)`,
|
|
||||||
);
|
|
||||||
if (usage) {
|
if (usage) {
|
||||||
const tokens = usage.steps.reduce((sum, s) => sum + (s.inputTokens ?? 0), 0);
|
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)`);
|
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) => {
|
main().catch((err) => {
|
||||||
|
|||||||
+45
-10
@@ -269,13 +269,21 @@ export function embeddingPort(): EmbeddingResolution {
|
|||||||
// ─── The post-merge batch (the extraction pipeline's embedding tail) ────────
|
// ─── The post-merge batch (the extraction pipeline's embedding tail) ────────
|
||||||
|
|
||||||
/** What the embedding tail reports back: counts for logs, the skip reason
|
/** What the embedding tail reports back: counts for logs, the skip reason
|
||||||
* when nothing ran (owner ruling: null model = info skip, not error), and
|
* when nothing ran (owner ruling: null model = info skip, not error), the
|
||||||
* the money-channel stage when it billed (one CallUsage step per API call,
|
* failure message when it died partway (owner ruling 2026-09-23:
|
||||||
* outputTokens: 0 — priced via EMBEDDING_PRICES). */
|
* 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 {
|
export interface EmbedBatchResult {
|
||||||
embedded: number;
|
embedded: number;
|
||||||
total: number;
|
total: number;
|
||||||
skipped?: string;
|
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;
|
usage?: ExtractionStageUsage;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -288,6 +296,11 @@ export interface EmbedBatchResult {
|
|||||||
* exist, then re-clusters ONCE per batch (consumers read the persisted
|
* exist, then re-clusters ONCE per batch (consumers read the persisted
|
||||||
* partition). Non-fatal by contract: the caller treats failure as
|
* partition). Non-fatal by contract: the caller treats failure as
|
||||||
* "clustering runs without the new nodes until the next merge retries".
|
* "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(
|
export async function embedNewNodes(
|
||||||
graphStore: {
|
graphStore: {
|
||||||
@@ -320,14 +333,26 @@ export async function embedNewNodes(
|
|||||||
if (!vectors[node.id]) todo.push(node);
|
if (!vectors[node.id]) todo.push(node);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let landed = 0;
|
||||||
|
let failed: string | undefined;
|
||||||
for (let i = 0; i < todo.length; i += EMBED_BATCH) {
|
for (let i = 0; i < todo.length; i += EMBED_BATCH) {
|
||||||
const batch = todo.slice(i, i + EMBED_BATCH);
|
const batch = todo.slice(i, i + EMBED_BATCH);
|
||||||
const result = await port.embed(
|
let result: EmbeddingResult;
|
||||||
batch.map((n) => n.statement),
|
try {
|
||||||
"document",
|
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++) {
|
for (let j = 0; j < batch.length; j++) {
|
||||||
vectors[batch[j]!.id] = result.vectors[j]!;
|
vectors[batch[j]!.id] = result.vectors[j]!;
|
||||||
|
landed++;
|
||||||
}
|
}
|
||||||
if (result.vectors[0]) dim = result.vectors[0].length;
|
if (result.vectors[0]) dim = result.vectors[0].length;
|
||||||
if (result.promptTokens !== undefined) {
|
if (result.promptTokens !== undefined) {
|
||||||
@@ -336,6 +361,11 @@ export async function embedNewNodes(
|
|||||||
steps.push(embeddingStep(result.promptTokens));
|
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.
|
// Prune ghosts: vectors for node ids that no longer exist.
|
||||||
let pruned = 0;
|
let pruned = 0;
|
||||||
@@ -351,7 +381,10 @@ export async function embedNewNodes(
|
|||||||
// consumers (snapshot, settle, namer) read the persisted partition.
|
// consumers (snapshot, settle, namer) read the persisted partition.
|
||||||
const clusters = semanticClusters(vectors, { similarityThreshold: SEMANTIC_SIM_THRESHOLD });
|
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) {
|
if (wrote) {
|
||||||
graphStore.writeEmbeddings({
|
graphStore.writeEmbeddings({
|
||||||
model: port.model,
|
model: port.model,
|
||||||
@@ -361,11 +394,13 @@ export async function embedNewNodes(
|
|||||||
updatedAt: new Date().toISOString(),
|
updatedAt: new Date().toISOString(),
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
return {
|
const result: EmbedBatchResult = {
|
||||||
embedded: todo.length,
|
embedded: landed,
|
||||||
total: Object.keys(vectors).length,
|
total: Object.keys(vectors).length,
|
||||||
usage: steps.length > 0 ? { model: port.model, steps } : undefined,
|
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) ──────
|
// ─── The federated embedding catalog (wave 6, owner ruling 2026-09-23) ──────
|
||||||
|
|
||||||
|
|||||||
@@ -109,6 +109,79 @@ function scriptedPort(model = "openai:test-embed", dim = 2, tokensPerCall = 120)
|
|||||||
|
|
||||||
const okPort = (port: EmbeddingPort): EmbeddingResolution => ({ ok: true, port });
|
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 ─────────────────────────────────
|
// ─── the money channel's embedding overlay ─────────────────────────────────
|
||||||
|
|
||||||
describe("the embedding price overlay (hand-maintained, outside the sync block)", () => {
|
describe("the embedding price overlay (hand-maintained, outside the sync block)", () => {
|
||||||
|
|||||||
Reference in New Issue
Block a user