diff --git a/app/api/graph/status/route.ts b/app/api/graph/status/route.ts index e056605..0ba43a6 100644 --- a/app/api/graph/status/route.ts +++ b/app/api/graph/status/route.ts @@ -15,6 +15,10 @@ export async function GET() { title: chatStore.getChat(run.chatId)?.title ?? run.chatId, phase: run.phase, startedAt: run.startedAt, + // Live per-stage usage (the money channel): model + per-step CallUsage, + // accumulated as steps bill. Cost and context-window folds are the + // consumer's (core/prices.ts helpers — nothing stores dollars). + usage: run.usage, })); return Response.json({ runs }); } diff --git a/src/agent/subagents/extraction.ts b/src/agent/subagents/extraction.ts index 629d221..2c7d308 100644 --- a/src/agent/subagents/extraction.ts +++ b/src/agent/subagents/extraction.ts @@ -1,7 +1,15 @@ import { ToolLoopAgent, stepCountIs, tool, type LanguageModel, type UIMessage } from "ai"; import { z } from "zod"; import { modelFor } from "../models.ts"; -import type { AuditResult, ChatRecord, GraphNode, SessionFindings } from "../../core/contracts.ts"; +import type { + AuditResult, + CallUsage, + ChatRecord, + ExtractionRunUsage, + ExtractionStageUsage, + GraphNode, + SessionFindings, +} from "../../core/contracts.ts"; import { getGraph, mergeFindings, type EmissionStats, type GraphDeps } from "../../core/graph.ts"; /** @@ -211,12 +219,6 @@ function createExtractor(model: LanguageModel) { }), }, stopWhen: stepCountIs(4), - // One line per step: a background worker's progress must be legible - // from the server log (§5 observability baseline). - onStepFinish: (step) => - console.info( - `[extraction] extractor step ${step.stepNumber}: ${step.finishReason}, tools: ${step.toolCalls.map((c) => c.toolName).join(", ") || "(none)"}`, - ), }); } @@ -300,10 +302,6 @@ function createAuditor(model: LanguageModel) { }), }, stopWhen: stepCountIs(4), - onStepFinish: (step) => - console.info( - `[extraction] auditor step ${step.stepNumber}: ${step.finishReason}, tools: ${step.toolCalls.map((c) => c.toolName).join(", ") || "(none)"}`, - ), }); } @@ -312,11 +310,18 @@ function createAuditor(model: LanguageModel) { /** The pipeline phases, in order — what the home status endpoint shows. */ export type ExtractionPhase = "extracting" | "auditing" | "merging"; -/** One in-flight pipeline run. Module-level: one process, one registry. */ -interface RunningExtraction { +/** One in-flight pipeline run. Module-level: one process, one registry. + * Exported for the status payload's consumer typing (the dock). */ +export interface RunningExtraction { chatId: string; phase: ExtractionPhase; startedAt: number; + /** Live per-stage usage, accumulated as steps bill (the dock's cost / + * context-window blobs read this through /api/graph/status). */ + usage: { extractor?: ExtractionStageUsage; auditor?: ExtractionStageUsage }; + /** True-cancellation handle (the dock's abort button): one controller per + * run, shared across phases; abort() cancels the in-flight model call. */ + controller: AbortController; } const runningExtractions = new Map(); @@ -326,6 +331,18 @@ export function extractionStatuses(): RunningExtraction[] { return [...runningExtractions.values()].sort((a, b) => a.startedAt - b.startedAt); } +/** True cancellation of the in-flight run for a chat (the dock's abort + * button): the pending model call receives the signal — the provider + * cancels the request, spend stops — and the run ends like a failure: + * nothing merged, nothing checkpointed, partial usage still persisted. + * Returns false when no run is in flight (idempotent no-op). */ +export function abortExtraction(chatId: string): boolean { + const run = runningExtractions.get(chatId); + if (!run) return false; + run.controller.abort(); + return true; +} + // ─── the pipeline ─────────────────────────────────────────────────────────── export interface ExtractionModels { @@ -350,21 +367,62 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti const inFlight = new Set(); function setPhase(chatId: string, phase: ExtractionPhase): void { + const prev = runningExtractions.get(chatId); runningExtractions.set(chatId, { chatId, phase, - startedAt: runningExtractions.get(chatId)?.startedAt ?? Date.now(), + startedAt: prev?.startedAt ?? Date.now(), + usage: prev?.usage ?? {}, // phase changes never lose usage + controller: prev?.controller ?? new AbortController(), }); } + /** Accumulate one billed step into the run's registry entry (live). */ + function recordStep(chatId: string, stage: "extractor" | "auditor", usage: CallUsage, model: string): void { + const run = runningExtractions.get(chatId); + if (!run) return; // not running (tests driving stages directly) + run.usage[stage] ??= { model, steps: [] }; + run.usage[stage]!.steps.push(usage); + } + + /** The SDK's per-step usage, flattened to the persisted CallUsage fields + * (the money channel's capture shape, contracts/chat.ts). */ + function callUsage(usage: { + inputTokens: number | undefined; + outputTokens: number | undefined; + inputTokenDetails?: { cacheReadTokens?: number; cacheWriteTokens?: number }; + outputTokenDetails?: { reasoningTokens?: number }; + }): CallUsage { + // Optional-chained: the SDK types the detail objects as required, but + // providers (and mocks) may leave them unset. + return { + inputTokens: usage.inputTokens, + outputTokens: usage.outputTokens, + cacheReadTokens: usage.inputTokenDetails?.cacheReadTokens, + cacheWriteTokens: usage.inputTokenDetails?.cacheWriteTokens, + reasoningTokens: usage.outputTokenDetails?.reasoningTokens, + }; + } + async function extract(chatId: string): Promise { const record = chatStore.getChat(chatId); if (!record) throw new Error(`Chat not found: ${chatId}`); + const signal = runningExtractions.get(chatId)?.controller.signal; + signal?.throwIfAborted(); // an abort during a sync gap ends the run here const view = transcriptView(record); console.info(`[extraction ${chatId}] extractor starting (${view.split("\n").length} transcript lines)`); const t0 = performance.now(); - const extraction = await extractorAgent.generate({ prompt: `Transcript:\n\n${view}` }); + const extraction = await extractorAgent.generate({ + prompt: `Transcript:\n\n${view}`, + abortSignal: signal, + onStepFinish: (step) => { + recordStep(chatId, "extractor", callUsage(step.usage), step.model.modelId); + console.info( + `[extraction] extractor step ${step.stepNumber}: ${step.finishReason}, tools: ${step.toolCalls.map((c) => c.toolName).join(", ") || "(none)"}`, + ); + }, + }); for (const part of extraction.content) { if (part.type === "tool-result" && part.toolName === "submit_findings") { console.info(`[extraction ${chatId}] extractor done in ${Math.round(performance.now() - t0)}ms`); @@ -377,6 +435,8 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti async function audit(chatId: string, findings: SessionFindings): Promise { const record = chatStore.getChat(chatId); if (!record) throw new Error(`Chat not found: ${chatId}`); + const signal = runningExtractions.get(chatId)?.controller.signal; + signal?.throwIfAborted(); // an abort during the extract→audit gap ends here const view = transcriptView(record); const graph = getGraph(graphStore); console.info( @@ -389,6 +449,13 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti prompt: `Transcript:\n\n${view}\n\nProposed findings:\n\n${JSON.stringify(findings, null, 2)}\n\n` + `Current knowledge graph (${graph.nodes.length} nodes):\n\n${graphView(graph.nodes)}`, + abortSignal: signal, + onStepFinish: (step) => { + recordStep(chatId, "auditor", callUsage(step.usage), step.model.modelId); + console.info( + `[extraction] auditor step ${step.stepNumber}: ${step.finishReason}, tools: ${step.toolCalls.map((c) => c.toolName).join(", ") || "(none)"}`, + ); + }, }); let auditResult: AuditResult | null = null; for (const part of review.content) { @@ -412,6 +479,25 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti return auditResult; } + /** Append the run's usage record to the chat's extractions log — success + * or failure: a failed run still spent money, and its partial usage is + * exactly what the record should show (stats absent = never merged). + * Idempotent per startedAt (the store dedupes). */ + function persistRunUsage(chatId: string, durationMs: number, stats: EmissionStats | undefined): void { + const run = runningExtractions.get(chatId); + if (!run) return; + chatStore.appendExtractionUsage(chatId, { + startedAt: run.startedAt, + at: new Date().toISOString(), + durationMs, + stats, + stages: { + extractor: run.usage.extractor ?? { model: "unknown", steps: [] }, + auditor: run.usage.auditor ?? { model: "unknown", steps: [] }, + }, + }); + } + async function runExtraction(chatId: string): Promise { if (inFlight.has(chatId)) return { nodes: 0, edges: 0, events: 0 }; inFlight.add(chatId); @@ -426,12 +512,22 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti // The merger re-reads the chat record itself — events are rebuilt // from the transcript, never from stale copies. const stats = mergeFindings(chatId, auditResult, { chatStore, graphStore }); + persistRunUsage(chatId, Math.round(performance.now() - t0), stats); console.info( `[extraction ${chatId}] merged: ${stats.nodes} nodes, ${stats.edges} edges, ${stats.events} events ` + `(${Math.round(performance.now() - t0)}ms total)`, ); return stats; } catch (err) { + persistRunUsage(chatId, Math.round(performance.now() - t0), undefined); + // Aborted (the dock's abort button): a deliberate stop, not a + // failure — same ledger discipline as a failure (nothing merged, + // nothing checkpointed, partial usage persisted), but the run + // reports zero emissions instead of throwing. + if (err instanceof Error && err.name === "AbortError") { + console.info(`[extraction ${chatId}] aborted (${Math.round(performance.now() - t0)}ms) — nothing merged, nothing checkpointed`); + return { nodes: 0, edges: 0, events: 0 }; + } // The throw travels to the caller; this line is what makes the // run's log legible end-to-end (no hanging "pipeline start"). console.error(`[extraction ${chatId}] pipeline failed (${Math.round(performance.now() - t0)}ms):`, err); diff --git a/src/core/contracts/chat.ts b/src/core/contracts/chat.ts index e80f903..3fd08c9 100644 --- a/src/core/contracts/chat.ts +++ b/src/core/contracts/chat.ts @@ -9,6 +9,7 @@ import type { UIMessage } from "ai"; import type { Phase3 } from "../phases.ts"; import type { PlanNodeStatus, PlanTree } from "./plan.ts"; +import type { EmissionStats } from "./graph.ts"; // ─── Chat records (the session store's on-disk format) ─────────────────────── @@ -32,6 +33,41 @@ export interface ChatRecord { * each entry onto its message; aggregates are folds over this log * (core/prices.ts). */ turns?: TurnUsage[]; + /** Append-only background-extraction usage log (2026-09 observability): + * one entry per finished pipeline run on THIS chat — extraction is real + * spend too. Same parallel-log discipline as `turns` (nothing rides + * inside messages). Keyed by startedAt; idempotent appends. */ + extractions?: ExtractionRunUsage[]; +} + +// ─── Background-extraction usage (the money channel, extraction side) ──────── + +/** One pipeline stage's model usage — the stage's model + every step it + * billed (CallUsage per agent step). Steps, not a sum: cost and context + * folds happen at read time (core/prices.ts). */ +export interface ExtractionStageUsage { + /** The OpenRouter model id that ran this stage. */ + model: string; + steps: CallUsage[]; +} + +/** One finished (or failed mid-flight) extraction run's usage record. + * Appended to the chat's `extractions` log when the run ends — success or + * failure, because money was spent either way; `stats` is present only + * when the merger actually ran. */ +export interface ExtractionRunUsage { + /** The run's registry startedAt (ms) — the idempotency key. */ + startedAt: number; + at: string; + durationMs: number; + /** What the run merged into the graph; absent when it failed first. */ + stats?: EmissionStats; + /** Model usage per stage, in pipeline order (merger is deterministic — + * no model call, no entry). */ + stages: { + extractor: ExtractionStageUsage; + auditor: ExtractionStageUsage; + }; } // ─── Usage accounting (the money channel) ───────────────────────────────────── diff --git a/src/core/contracts/graph.ts b/src/core/contracts/graph.ts index a0ba30d..24ea899 100644 --- a/src/core/contracts/graph.ts +++ b/src/core/contracts/graph.ts @@ -22,6 +22,17 @@ * strength — strength is derived at read time from topology. */ export type EdgeType = "follows" | "supports"; +/** What one extraction run merged into the graph ledgers — the merger's + * output, also persisted in extraction-run usage records (contracts/chat). */ +export interface EmissionStats { + /** New graph nodes appended. */ + nodes: number; + /** New edge assertions appended. */ + edges: number; + /** Ledger events rebuilt (quiz outcomes + observations). */ + events: number; +} + /** One provenance point — message-granular (DECIDED 2026-09-02): nodes and * edges record every (chat, message) where they came up, deep-linkable to * the exact transcript point. */ diff --git a/src/core/graph.ts b/src/core/graph.ts index f40d684..5164992 100644 --- a/src/core/graph.ts +++ b/src/core/graph.ts @@ -42,6 +42,7 @@ import { randomBytes } from "node:crypto"; import { isToolUIPart, type UIMessage } from "ai"; import type { AuditResult, + EmissionStats, GraphEdge, GraphNode, ProvenancePoint, @@ -108,15 +109,6 @@ const samePoint = (a: ProvenancePoint, b: ProvenancePoint): boolean => // ─── the merger (extraction pipeline stage 3 — deterministic) ──────────────── -export interface EmissionStats { - /** New graph nodes appended. */ - nodes: number; - /** New edge assertions appended. */ - edges: number; - /** New ledger events appended. */ - events: number; -} - /** * Merge one chat's audited findings into the graph ledgers — the pipeline's * apply stage. Deterministic + idempotent: re-merging the same verdicts (or @@ -470,3 +462,6 @@ export function graphSnapshot(deps: GraphDeps): GraphSnapshot { return { nodes, edges, sessions, nodeSessions, nodeObservations }; } + +// Re-export: the contract owns the shape; legacy importers still reach it. +export type { EmissionStats }; diff --git a/src/core/ports.ts b/src/core/ports.ts index 8baad24..b612877 100644 --- a/src/core/ports.ts +++ b/src/core/ports.ts @@ -18,6 +18,7 @@ import type { JobKind, JobRecord, LedgerEvent, + ExtractionRunUsage, NodeMapping, PlanTree, SessionCheckpoint, @@ -41,6 +42,8 @@ export interface ChatStore { * 2026-09): idempotent per messageId — a turn persisted twice (abort * replays, wake-turn retries) appends exactly once. */ appendTurnUsage(id: string, entry: TurnUsage): void; + /** Append one extraction-run usage entry — idempotent per startedAt. */ + appendExtractionUsage(id: string, entry: ExtractionRunUsage): void; } // ─── GraphStore — the append-only evidence ledgers (data/graph) ────────────── diff --git a/src/server/observability.ts b/src/server/observability.ts index e92ebdf..7b2ebc8 100644 --- a/src/server/observability.ts +++ b/src/server/observability.ts @@ -8,5 +8,5 @@ * this file is where the deep imports surface, so routes stay clean. */ -export { extractionStatuses, type ExtractionPhase } from "../agent/subagents/extraction.ts"; +export { abortExtraction, extractionStatuses, type ExtractionPhase, type RunningExtraction } from "../agent/subagents/extraction.ts"; export { chatStore } from "../store/chats.ts"; diff --git a/src/store/chats.ts b/src/store/chats.ts index 81a1f69..9f26f2b 100644 --- a/src/store/chats.ts +++ b/src/store/chats.ts @@ -11,7 +11,7 @@ import * as fs from "node:fs"; import * as path from "node:path"; import { PATHS } from "../../config.ts"; import type { ChatStore } from "../core/ports.ts"; -import type { ChatRecord, ChatSummary, PlanTree, TurnUsage } from "../core/contracts.ts"; +import type { ChatRecord, ChatSummary, ExtractionRunUsage, PlanTree, TurnUsage } from "../core/contracts.ts"; import type { UIMessage } from "ai"; import { findPendingToolCall } from "../core/pending.ts"; import { phaseOfTree } from "../core/phases.ts"; @@ -51,9 +51,10 @@ function listChats(): ChatSummary[] { .filter((f) => f.endsWith(".json")) .map((f) => getChat(f.slice(0, -5))) .filter((r): r is ChatRecord => r !== null) - .map(({ messages, tree, turns, ...summary }) => ({ - // turns stay out of the list payload (list = summaries; usage - // aggregates ride the full record, the picker never needs them). + .map(({ messages, tree, turns, extractions, ...summary }) => ({ + // turns and extractions stay out of the list payload (list = + // summaries; usage aggregates ride the full record, the picker + // never needs them). ...summary, messageCount: messages.length, // Record-derived status for the session picker: phase from the @@ -90,6 +91,18 @@ function appendTurnUsage(id: string, entry: TurnUsage): void { fs.writeFileSync(chatFile(id), JSON.stringify(record, null, "\t"), "utf-8"); } +/** Append one extraction-run usage entry to the record's extractions + * log — idempotent per startedAt: a retry that reaches the append twice + * (abort replays) does not double-count spend. */ +function appendExtractionUsage(id: string, entry: ExtractionRunUsage): void { + const record = getChat(id); + if (!record) throw new Error(`Chat not found: ${id}`); + record.extractions = record.extractions?.some((r) => r.startedAt === entry.startedAt) + ? record.extractions + : [...(record.extractions ?? []), entry]; + fs.writeFileSync(chatFile(id), JSON.stringify(record, null, "\t"), "utf-8"); +} + /** Persist the session plan tree (plan tools call this in execute). */ function saveTree(id: string, tree: PlanTree): void { const record = getChat(id); @@ -107,4 +120,5 @@ export const chatStore: ChatStore = { saveMessages, saveTree, appendTurnUsage, + appendExtractionUsage, }; diff --git a/tests/extraction.test.ts b/tests/extraction.test.ts index 5a5fbec..b4214e8 100644 --- a/tests/extraction.test.ts +++ b/tests/extraction.test.ts @@ -34,7 +34,7 @@ import type { SessionFindings, } from "../src/core/contracts.ts"; import type { GraphStore } from "../src/core/ports.ts"; -import { createExtractionPipeline, extractionStatuses, transcriptView } from "../src/agent/subagents/extraction.ts"; +import { abortExtraction, createExtractionPipeline, extractionStatuses, transcriptView } from "../src/agent/subagents/extraction.ts"; import { chatStore } from "../src/store/chats.ts"; import { PATHS } from "../config.ts"; @@ -197,6 +197,24 @@ function submitModel(name: string, input: unknown, onCall?: () => void): MockLan }); } +/** A model that hangs until its abort signal fires, then rejects — + * mimicking a cancelled provider request (the abort path). Like a real + * provider, an ALREADY-aborted signal rejects immediately (the abort + * can win the race before doGenerate attaches its listener). */ +function hangingModel(): MockLanguageModelV4 { + const abortReject = (reject: (reason: Error) => void) => + reject(Object.assign(new Error("The operation was aborted"), { name: "AbortError" })); + return new MockLanguageModelV4({ + doGenerate: async (options): Promise => { + await new Promise((_, reject) => { + if (options.abortSignal?.aborted) abortReject(reject); + else options.abortSignal?.addEventListener("abort", () => abortReject(reject)); + }); + throw new Error("unreachable"); + }, + }); +} + /** A model that never calls a tool (the failure path). */ function silentModel(): MockLanguageModelV4 { return new MockLanguageModelV4({ @@ -378,6 +396,97 @@ describe("the extraction pipeline (extractor → auditor → merger)", () => { assert.equal(extractionStatuses().length, 0, "the registry clears when the run ends"); }); + it("captures per-stage usage live, persists the run's usage into the chat record", async () => { + const { chatId } = fixtureChat(); + const ledgers = memoryLedgers(); + const existingId = "n_existing"; + ledgers.appendNode({ + id: existingId, + name: "Derivative", + statement: "Derivatives measure rates of change", + aliases: [], + createdAt: "2026-08-01T00:00:00.000Z", + provenance: [], + }); + const usageSeen: (number | undefined)[] = []; + const verdict: AuditResult = { + propositions: [{ ...FINDINGS.propositions[0]!, linkTo: existingId }], + edges: [], + attribution: { quiz: {} }, + dropped: [{ ref: "p2", reason: "evidence not found" }], + }; + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + // During the auditor's generate, the registry must already carry + // the extractor's fully accumulated live usage (2 steps × 10/5). + extractor: submitModel("submit_findings", FINDINGS), + auditor: submitModel("submit_review", verdict, () => + usageSeen.push(extractionStatuses().find((r) => r.chatId === chatId)?.usage.extractor?.steps.at(-1)?.inputTokens), + ), + }); + await pipeline.runExtraction(chatId); + + // Live: mid-audit the extractor's usage was fully accumulated + // (the mock bills 10 input / 5 output per step, flat per the SDK). + assert.deepEqual(usageSeen, [10]); + + // Persisted: the run appended ONE usage entry to the chat's + // extractions log — per-stage model + steps, with merge stats. + const record = chatStore.getChat(chatId)!; + assert.equal(record.extractions?.length, 1); + const run = record.extractions![0]!; + assert.equal(run.stats?.nodes, 0); // linked, not created + assert.equal(run.stats?.events, 3); + assert.equal(run.stages.extractor.model, "mock-model-id"); // the mock's model id + assert.equal(run.stages.extractor.steps.length, 2); // tool-call step + text step + assert.deepEqual(run.stages.extractor.steps[0], { + inputTokens: 10, + outputTokens: 5, + cacheReadTokens: 0, + cacheWriteTokens: 0, + reasoningTokens: 0, + }); + assert.equal(run.stages.auditor.steps.length, 2); + assert.ok(!Number.isNaN(Date.parse(run.at))); + }); + + it("aborts mid-run: cancels the model call, persists partial usage, checkpoints nothing", async () => { + const { chatId } = fixtureChat(); + const ledgers = memoryLedgers(); + + const pipeline = createExtractionPipeline({ + chatStore, + graphStore: ledgers, + extractor: hangingModel(), // hangs until aborted — no step finishes + auditor: submitModel("submit_review", { propositions: [], edges: [], attribution: { quiz: {} }, dropped: [] }), + }); + const runPromise = pipeline.runExtraction(chatId); + + // The registry entry exists synchronously after the call (setPhase + // runs before the first await); abort is TRUE cancellation. + assert.equal(abortExtraction(chatId), true); + const stats = await runPromise; + + // A deliberate stop, not a failure: zero emissions, no throw. + assert.deepEqual(stats, { nodes: 0, edges: 0, events: 0 }); + assert.equal(extractionStatuses().length, 0, "the registry clears"); + assert.equal(ledgers.readNodes().length, 0, "nothing merged"); + assert.equal(ledgers.readCheckpoints().length, 0, "nothing checkpointed — the next trigger re-runs"); + + // Partial usage still persisted (money was spent): stats absent, + // no completed steps in either stage. + const record = chatStore.getChat(chatId)!; + assert.equal(record.extractions?.length, 1); + const run = record.extractions![0]!; + assert.equal(run.stats, undefined); + assert.equal(run.stages.extractor.steps.length, 0); + assert.equal(run.stages.auditor.steps.length, 0); + + // Aborting a non-running chat is an idempotent no-op. + assert.equal(abortExtraction(chatId), false); + }); + it("is single-flight per chat — a concurrent second run is a no-op", async () => { const { chatId } = fixtureChat(); const ledgers = memoryLedgers(); diff --git a/tests/usage.test.ts b/tests/usage.test.ts index a2085bc..e71839c 100644 --- a/tests/usage.test.ts +++ b/tests/usage.test.ts @@ -12,7 +12,7 @@ import * as fs from "node:fs"; import { chatStore } from "../src/store/chats.ts"; import { PATHS } from "../config.ts"; import { callCost, contextWindow, jobCost, sessionCost, sumSteps, turnCost } from "../src/core/prices.ts"; -import type { CallUsage, JobRecord, TurnUsage } from "../src/core/contracts.ts"; +import type { CallUsage, ExtractionRunUsage, JobRecord, TurnUsage } from "../src/core/contracts.ts"; const call = (over: Partial = {}): CallUsage => ({ inputTokens: 0, @@ -134,3 +134,34 @@ test("appendTurnUsage: appends, is idempotent per messageId, survives saveMessag fs.unlinkSync(`${PATHS.chats}/${record.id}.json`); }); + +test("appendExtractionUsage: appends, idempotent per startedAt, not in list payload", () => { + const record = chatStore.createChat("extraction usage test chat"); + const runEntry = (startedAt: number): ExtractionRunUsage => ({ + startedAt, + at: "2026-09-09T00:00:00Z", + durationMs: 42_000, + stats: { nodes: 1, edges: 2, events: 3 }, + stages: { + extractor: { model: "z-ai/glm-5.3-flash", steps: [call({ inputTokens: 100, outputTokens: 10 })] }, + auditor: { model: "z-ai/glm-5.3-flash", steps: [call({ inputTokens: 50, outputTokens: 5 })] }, + }, + }); + chatStore.appendExtractionUsage(record.id, runEntry(1)); + chatStore.appendExtractionUsage(record.id, runEntry(1)); // same run appended twice → no-op + chatStore.appendExtractionUsage(record.id, runEntry(2)); // a retried run is a NEW spend + let reread = chatStore.getChat(record.id); + assert.equal(reread?.extractions?.length, 2); + // failed runs (stats absent) persist with usage — money was spent + chatStore.appendExtractionUsage(record.id, { ...runEntry(3), stats: undefined }); + reread = chatStore.getChat(record.id); + assert.equal(reread?.extractions?.length, 3); + assert.equal(reread?.extractions?.at(-1)?.stats, undefined); + + // the list payload must NOT leak the extractions log + const listed = chatStore.listChats().find((c) => c.id === record.id); + assert.ok(listed); + assert.equal("extractions" in listed, false); + + fs.unlinkSync(`${PATHS.chats}/${record.id}.json`); +});