Extraction observability: usage capture, live status payload, true abort (Wave 4)

Extraction is real spend and background work; the dock needs to see
and stop it. User decisions: persist all cost data (chat record),
per-stage usage granularity, and TRUE cancellation for abort.

USAGE CAPTURE (live + persisted):
- extraction.ts: per-call onStepFinish captures each billed step's
  usage (SDK LanguageModelUsage -> CallUsage: input/output/cache
  read+write/reasoning) into the registry's live per-stage usage
  (extractor/auditor; the merger is deterministic — no model call).
- /api/graph/status runs now carry usage (model + steps per stage);
  cost/context folds are the consumer's (core/prices.ts — nothing
  stores dollars).
- ChatRecord.extractions: append-only per-run usage log (mirrors the
  turns log): startedAt key, at, durationMs, merge stats when the
  merger ran, stages. Persisted on success AND failure — a failed run
  still spent money. Idempotent per startedAt; the list payload
  (ChatSummary) never leaks it.
- EmissionStats moved to contracts/graph.ts (its contract home);
  core/graph.ts re-exports for existing importers.
- Verified end-to-end against the real pipeline (backfill one chat):
  per-step token counts, reasoning tokens, and the auditor's growing
  prompt-cache reads captured and persisted.

TRUE ABORT (the dock's abort button, user decision):
- RunningExtraction owns one AbortController per run, shared across
  phases; abortExtraction(chatId) fires it and returns whether a run
  was aborted (idempotent no-op otherwise). The pending model call
  receives the signal — the provider cancels the request, spend
  stops. throwIfAborted() at phase boundaries ends runs aborted in
  sync gaps.
- Abort is a deliberate stop, not a failure: same ledger discipline
  (nothing merged, nothing checkpointed, partial usage persisted) but
  the run reports zero emissions instead of throwing. Exported via
  src/server/observability.ts for master's /api/activity/abort route.
- Tests: mid-run abort (hanging mock that rejects on the signal),
  live-accumulation + persisted-record assertions, store-seam
  idempotency, and the summary-payload leak guard.
This commit is contained in:
amosblomqvist
2026-09-11 03:25:25 +03:00
parent 83a00800e9
commit 9174cde55e
10 changed files with 330 additions and 31 deletions
+36
View File
@@ -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) ─────────────────────────────────────
+11
View File
@@ -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. */
+4 -9
View File
@@ -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 };
+3
View File
@@ -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) ──────────────