From 17ed37a4717468a06e07765c860788587100a799 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Fri, 11 Sep 2026 03:25:09 +0300 Subject: [PATCH 1/4] feat: abort sub-agent jobs (real kill) + runner usage capture into JobRecord.usage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wave 4 (agent-side observability + control), reconciled with the dock contract (master relay 2026-09): - abortJob in src/server/jobs.ts, exported through the server barrel for POST /api/activity/abort (master's thin route): fires the job's AbortController (real kill — the in-flight model call aborts, not just the next loop check), drops any queued result steer, flips the record to error "Aborted by learner" + activity "aborted" synchronously, and queues a one-line abort steer (buildAbortSteer) through the same wake pipeline so the teacher's promise doesn't hang in the lesson. Shares the result steer's deterministic id space, so abort + racing result can never both deliver (steerDelivered dedupe). - Abort registry (src/agent/subagents/aborts.ts) on globalThis, mirroring the wake pipeline's module-state pattern; runner registers/unregisters per run, so the route's module instance sees the loop's controller. - Runner usage capture: LoopContext (abortSignal + onUsage) threads into all three kind loops; per-step usage folds via accumulateJobUsage into JobRecord.usage (model from config MODELS, absent-never-zero), flushed in runJob's finally so an aborted job's money record survives the kill. - Contracts: JobUsage token fields optional + lastStep (most recent call's accounting, for the dock's per-blob context line vs CONTEXT_LIMITS); SubagentResultMeta.aborted; AbortSteerMessage; buildAbortSteer; SteerMessage union generalized. - demo-builder/visual-maker accept abortSignal + onUsage (visual folds generateText usage per iteration). - stillRunning/finishJob/jobContext guards: an aborted job is never overwritten by the runner (terminal no matter what the wake does); failJob and the pump's finalizeJob keep the flip. - UI: abort steer renders a quiet aborted chip (job-aborted-badge). - Prompt 20-jobs.md: one line about aborted jobs. Tests: accumulateJobUsage folds (undefined-preservation, lastStep), runner usage capture, real-kill abort (sync flip, silent unwind, abort steer wakes), usage-so-far survives the kill, aborting a delivering job drops the queued result steer, outcome reasons, abort-chip render. 330/330, typecheck/lint/build green. --- components/artifacts/steer-card.tsx | 23 ++- src/agent/prompts/20-jobs.md | 4 + src/agent/subagents/aborts.ts | 50 ++++++ src/agent/subagents/demo-builder.ts | 8 +- src/agent/subagents/runner.ts | 130 +++++++++++++-- src/agent/subagents/visual-maker.ts | 17 +- src/core/contracts/jobs.ts | 104 +++++++++++- src/server/jobs.ts | 80 ++++++++- tests/demos.test.tsx | 14 +- tests/jobs.test.ts | 247 +++++++++++++++++++++++++++- 10 files changed, 644 insertions(+), 33 deletions(-) create mode 100644 src/agent/subagents/aborts.ts diff --git a/components/artifacts/steer-card.tsx b/components/artifacts/steer-card.tsx index fab2503..003710b 100644 --- a/components/artifacts/steer-card.tsx +++ b/components/artifacts/steer-card.tsx @@ -1,6 +1,6 @@ "use client"; -import { CheckIcon } from "lucide-react"; +import { CheckIcon, CircleX } from "lucide-react"; import type { ResearchResult, SubagentArtifactData, TeacherUIMessage } from "@/src/core"; import { isSubagentFailure } from "@/src/core"; import { DemoOutputCard } from "@/components/artifacts/demo-card"; @@ -27,9 +27,15 @@ import { VisualOutputCard } from "@/components/artifacts/visual-card"; export function SubagentSteerCard({ message }: { message: TeacherUIMessage }) { const artifactPart = message.parts.find((p) => p.type === "data-subagent-artifact"); const data = artifactPart?.data as SubagentArtifactData | undefined; + // The abort steer (2026-09 dock semantics): no artifact part, only the + // metadata marker — a quiet chip, mirroring the complete-badge idiom. if (!data) { + const meta = ( + message.metadata as { subagentResult?: { aborted?: boolean; kind?: string } } | undefined + )?.subagentResult; + if (meta?.aborted) return ; return ( -
+
A sub-agent finished, but its result could not be displayed.
); @@ -48,6 +54,19 @@ export function SubagentSteerCard({ message }: { message: TeacherUIMessage }) { } } +/** The aborted-job chip: mirrors the complete-badge one-liner — the + * teacher's wake turn acknowledges the abort in the lesson itself, + * this just marks what happened. */ +function AbortedBadge({ kind }: { kind: string }) { + return ( +
+ + {kind} + aborted by the learner +
+ ); +} + /** Quiet one-liner, the dispatch-settled idiom — the lesson lives in the * teacher's wake turn; the badge just marks delivery. The persisted brief * rides along in a collapsed
(the wave-1 "view full research" diff --git a/src/agent/prompts/20-jobs.md b/src/agent/prompts/20-jobs.md index 26d47f7..e1d4026 100644 --- a/src/agent/prompts/20-jobs.md +++ b/src/agent/prompts/20-jobs.md @@ -27,3 +27,7 @@ Dispatch as many research jobs as you genuinely need — in one turn or across t parallel when the questions are independent — but don't fan out on one question; the researcher searches multiple angles on its own. If research fails, you'll get a failure message: say so plainly and continue with what you can establish, flagging the uncertainty. + +A job can also be **aborted by the learner** — you'll get a one-line "was aborted" +message instead of the result. Acknowledge it in one sentence and continue teaching +without it (offering to re-dispatch is fine if the learner wants the result). diff --git a/src/agent/subagents/aborts.ts b/src/agent/subagents/aborts.ts new file mode 100644 index 0000000..af357de --- /dev/null +++ b/src/agent/subagents/aborts.ts @@ -0,0 +1,50 @@ +/** + * Job abort registry — WHO can actually kill a running sub-agent. + * + * The runner creates one AbortController per running job and registers it + * here; the abort path (src/server/jobs.ts abortJob, driven by the dock's + * END button via POST /api/jobs/[jobId]/abort) fires it through fireJobAbort. + * Two module trees meet: the runner (src/agent) and the endpoint + * (src/server) — the registry lives on globalThis for the same reason as + * the turn registry and the wake pipeline: Next dev gives each route entry + * its own module instances, so a module-level Map would be a different + * object in each. Placed under src/agent (server→agent imports are an + * established direction; agent→server is not — see chat-handler). + * + * 2026-09 abort semantics (user-decided): REAL KILL, not mark-and-abandon — + * the controller aborts the in-flight model call (agent.generate / + * generateText abortSignal); headless tool runs are best-effort (the next + * model call is where a hung one dies). The route owns the bookkeeping + * (status error "Aborted by learner" + the abort steer); the runner only unwinds. + */ + +interface LhAbortGlobal { + __lhJobAborts?: Map; +} + +function registry(): Map { + const g = globalThis as typeof globalThis & LhAbortGlobal; + return (g.__lhJobAborts ??= new Map()); +} + +/** The runner registers its controller at job start (idempotent overwrite). */ +export function registerJobAbort(jobId: string, controller: AbortController): void { + registry().set(jobId, controller); +} + +/** The runner unregisters in its finally block (job settled or unwound). */ +export function unregisterJobAbort(jobId: string): void { + registry().delete(jobId); +} + +/** Abort a running job's loop. Returns false when no controller is live — + * the job's loop is not executing (crashed, or in the delivering-result + * phase where the runner has already returned); the caller then owns the + * whole abort on the record side. */ +export function fireJobAbort(jobId: string): boolean { + const controller = registry().get(jobId); + if (!controller) return false; + registry().delete(jobId); + controller.abort(); + return true; +} diff --git a/src/agent/subagents/demo-builder.ts b/src/agent/subagents/demo-builder.ts index 37ed194..5341eae 100644 --- a/src/agent/subagents/demo-builder.ts +++ b/src/agent/subagents/demo-builder.ts @@ -1,7 +1,7 @@ import { ToolLoopAgent, stepCountIs, tool, type LanguageModel, type ToolSet } from "ai"; import { z } from "zod"; import { modelFor } from "../models.ts"; -import type { DemoResult } from "../../core/contracts.ts"; +import type { DemoResult, UsageStepLike } from "../../core/contracts.ts"; import { getHeadless, type DemoRunResult, type HeadlessApi } from "./headless.ts"; /** @@ -32,6 +32,10 @@ export interface DemoBuilderOptions { maxSteps?: number; /** Live activity-line callback (updates the job record). */ onActivity?: (line: string) => void; + /** Kills the run mid-model-call (2026-09 real-kill abort semantics). */ + abortSignal?: AbortSignal; + /** Money channel: one call per agent step, folded into JobRecord.usage. */ + onUsage?: (usage: UsageStepLike) => void; } const INSTRUCTIONS = `You are a demo builder — a specialist sub-agent inside a learning application. You @@ -189,6 +193,8 @@ export async function runDemoBuilder(input: DemoBuilderInput, opts: DemoBuilderO try { const result = await agent.generate({ prompt, + abortSignal: opts.abortSignal, + onStepFinish: ({ usage }) => opts.onUsage?.(usage), onToolExecutionStart: ({ toolCall }) => { opts.onActivity?.(toolCall.toolName); }, diff --git a/src/agent/subagents/runner.ts b/src/agent/subagents/runner.ts index 924512c..95cc3c0 100644 --- a/src/agent/subagents/runner.ts +++ b/src/agent/subagents/runner.ts @@ -1,16 +1,21 @@ import { isToolUIPart, type LanguageModel, type UIMessage } from "ai"; import { + accumulateJobUsage, buildSubagentSteer, type DemoResult, type DemoToolInput, type JobKind, type JobRecord, + type JobUsage, type ResearchResult, type ResearchToolInput, type SubagentSteerMessage, + type UsageStepLike, type VisualResult, type VisualToolInput, } from "../../core/contracts.ts"; +import { registerJobAbort, unregisterJobAbort } from "./aborts.ts"; +import { MODELS } from "../../../config.ts"; import type { ChatStore, JobStore } from "../../core/ports.ts"; import { runDemoBuilder } from "./demo-builder.ts"; import { createResearcherAgent } from "./researcher.ts"; @@ -67,16 +72,70 @@ export interface JobRunner { runJob(jobId: string, input?: ResearchToolInput | DemoToolInput | VisualToolInput): Promise; } +/** One job loop's injected machinery: the kill switch (real abort, 2026-09 + * semantics) and the usage tap (money channel — every model call folds into + * JobRecord.usage as it completes). */ +interface LoopContext { + /** Fires when abortJob kills the run; abortSignal is passed into every + * model call so the spend stops mid-call, not at the next loop check. */ + abortSignal: AbortSignal; + /** Fold one model call's usage into the job record (progress write). */ + onUsage: (step: UsageStepLike) => void; +} + +/** The OpenRouter model id for a job kind — JobUsage.model. The kind→role + * table mirrors MODELS in config.ts (research→researcher, demo→demoBuilder, + * visual→visualMaker); the id is taken from config so a price-table sync + * re-prices jobs for free. Model overrides (tests) still bill under the + * configured model — usage is what is being captured, not which instance. */ +const KIND_MODEL: Record = { + research: MODELS.researcher.model, + demo: MODELS.demoBuilder.model, + visual: MODELS.visualMaker.model, +}; + export function createJobRunner(deps: JobRunnerDeps): JobRunner { const { chatStore, jobStore, deliver } = deps; + /** Running usage totals per live job (the loop's onUsage folds here, + * then writes through the job store so the poll sees progress). */ + const usageTotals = new Map(); + + function makeLoopContext(jobId: string, kind: JobKind): LoopContext { + const model = KIND_MODEL[kind]; + // Real-kill wiring (2026-09 abort semantics): the controller is what + // the abort endpoint fires (agent/subagents/aborts.ts registry); the + // signal rides into every model call of the loop. + const controller = new AbortController(); + registerJobAbort(jobId, controller); + return { + abortSignal: controller.signal, + onUsage: (step) => { + const next = accumulateJobUsage(usageTotals.get(jobId), model, step); + usageTotals.set(jobId, next); + jobStore.updateJob(jobId, { usage: next }); // merge patch — status untouched + }, + }; + } + + /** True while the job may still write terminal state (result, error, + * steer). False once abortJob has flipped it to error ("Aborted by learner") — + * the runner then unwinds silently; the abort path owns the record and + * the abort steer. */ + function stillRunning(jobId: string): boolean { + return jobStore.getJob(jobId)?.status === "running"; + } + /** The kind→runner map: one sub-agent loop per job kind. */ // Each kind's runner narrows the union input itself (dispatch-time inputs // are kind-correct by construction; transcript fallbacks are cast after). - const KIND_RUNNERS: Record Promise> = { - research: (jobId, input) => runResearchJob(jobId, researchModelOverride.current, input), - demo: (jobId, input) => runDemoJob(jobId, demoModelOverride.current, input), - visual: (jobId, input) => runVisualJob(jobId, visualModelOverride.current, input), + const KIND_RUNNERS: Record< + JobKind, + (jobId: string, input: JobToolInput | undefined, loop: LoopContext) => Promise + > = { + research: (jobId, input, loop) => runResearchJob(jobId, researchModelOverride.current, input, loop), + demo: (jobId, input, loop) => runDemoJob(jobId, demoModelOverride.current, input, loop), + visual: (jobId, input, loop) => runVisualJob(jobId, visualModelOverride.current, input, loop), }; async function runJob(jobId: string, input?: JobToolInput): Promise { @@ -92,7 +151,17 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner { jobStore.updateJob(jobId, { status: "error", error: `unknown kind ${job.kind}`, activity: "failed" }); return; } - await run(jobId, input); + try { + await run(jobId, input, makeLoopContext(jobId, job.kind)); + } finally { + unregisterJobAbort(jobId); + // The last usage write may have been mid-flight when the loop died; + // persist what was folded (the abort path patches only status, so + // this never resurrects an aborted job — merge patch). + const usage = usageTotals.get(jobId); + if (usage) jobStore.updateJob(jobId, { usage }); + usageTotals.delete(jobId); + } } /** Resolve the job's tool input: dispatch passes it directly; only a @@ -132,6 +201,10 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner { async function finishJob(jobId: string, result: ResearchResult | DemoResult | VisualResult): Promise { const job = jobStore.getJob(jobId); if (!job) return; + // Aborted mid-run or mid-delivery (abortJob flips the record before + // the loop unwinds): no result write, no steer — the abort path owns + // the record and queues the abort steer itself. + if (job.status !== "running") return; jobStore.updateJob(jobId, { result, activity: "delivering result" }); const steer = buildSubagentSteer(job, result); try { @@ -157,7 +230,12 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner { // ─── research ─────────────────────────────────────────────────────────── - async function runResearchJob(jobId: string, model?: LanguageModel, passedInput?: JobToolInput): Promise { + async function runResearchJob( + jobId: string, + model: LanguageModel | undefined, + passedInput: JobToolInput | undefined, + loop: LoopContext, + ): Promise { const ctx = jobContext(jobId); if (!ctx) return; const { job } = ctx; @@ -181,6 +259,10 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner { jobStore.updateJob(jobId, { activity: "researching" }); const result = await agent.generate({ prompt, + abortSignal: loop.abortSignal, + // Money channel: each agent step is one model call — fold its + // usage into the record as it completes (progress write). + onStepFinish: ({ usage }) => loop.onUsage(usage), onToolExecutionStart: ({ toolCall }) => { const input = toolCall.input as { query?: string; url?: string } | undefined; if (toolCall.toolName === "web_fetch" && typeof input?.url === "string") { @@ -195,15 +277,24 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner { const research: ResearchResult = { brief, sources: parseSources(brief, fetchedUrls) }; await finishJob(jobId, research); } catch (err) { + // Real-kill abort (2026-09): the abort path flipped the record and + // queued the abort steer — unwind silently, keep the usage + // folded so far (flushed by runJob's finally). + if (loop.abortSignal.aborted) return; const reason = err instanceof Error ? err.message : String(err); - jobStore.updateJob(jobId, { error: reason }); + if (stillRunning(jobId)) jobStore.updateJob(jobId, { error: reason }); await finishJob(jobId, { brief: `Research failed: ${reason}`, sources: [] }); } } // ─── demo + visual ────────────────────────────────────────────────────── - async function runDemoJob(jobId: string, model?: LanguageModel, passedInput?: JobToolInput): Promise { + async function runDemoJob( + jobId: string, + model: LanguageModel | undefined, + passedInput: JobToolInput | undefined, + loop: LoopContext, + ): Promise { const ctx = jobContext(jobId); if (!ctx) return; const { job } = ctx; @@ -218,19 +309,28 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner { { title, brief }, { model, + abortSignal: loop.abortSignal, + onUsage: loop.onUsage, onActivity: (line) => jobStore.updateJob(jobId, { activity: line }), }, ); - if (isDemoFailure(result)) jobStore.updateJob(jobId, { error: result.notes ?? "demo build failed" }); + if (isDemoFailure(result) && stillRunning(jobId)) + jobStore.updateJob(jobId, { error: result.notes ?? "demo build failed" }); await finishJob(jobId, result); } catch (err) { + if (loop.abortSignal.aborted) return; const reason = err instanceof Error ? err.message : String(err); - jobStore.updateJob(jobId, { error: reason }); + if (stillRunning(jobId)) jobStore.updateJob(jobId, { error: reason }); await finishJob(jobId, { title, html: "", tested: true, notes: `FAILED: ${reason}` }); } } - async function runVisualJob(jobId: string, model?: LanguageModel, passedInput?: JobToolInput): Promise { + async function runVisualJob( + jobId: string, + model: LanguageModel | undefined, + passedInput: JobToolInput | undefined, + loop: LoopContext, + ): Promise { const ctx = jobContext(jobId); if (!ctx) return; const { job } = ctx; @@ -244,14 +344,18 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner { { brief }, { model, + abortSignal: loop.abortSignal, + onUsage: loop.onUsage, onActivity: (line) => jobStore.updateJob(jobId, { activity: line }), }, ); - if (isVisualFailure(result)) jobStore.updateJob(jobId, { error: result.notes ?? "visual build failed" }); + if (isVisualFailure(result) && stillRunning(jobId)) + jobStore.updateJob(jobId, { error: result.notes ?? "visual build failed" }); await finishJob(jobId, result); } catch (err) { + if (loop.abortSignal.aborted) return; const reason = err instanceof Error ? err.message : String(err); - jobStore.updateJob(jobId, { error: reason }); + if (stillRunning(jobId)) jobStore.updateJob(jobId, { error: reason }); await finishJob(jobId, { svg: "", notes: `FAILED: ${reason}` }); } } diff --git a/src/agent/subagents/visual-maker.ts b/src/agent/subagents/visual-maker.ts index df0a798..7349093 100644 --- a/src/agent/subagents/visual-maker.ts +++ b/src/agent/subagents/visual-maker.ts @@ -1,7 +1,7 @@ import { generateText, tool, type LanguageModel, type ModelMessage, type ToolResultPart } from "ai"; import { z } from "zod"; import { modelFor } from "../models.ts"; -import type { VisualResult } from "../../core/contracts.ts"; +import type { UsageStepLike, VisualResult } from "../../core/contracts.ts"; import { getHeadless, type HeadlessApi } from "./headless.ts"; /** @@ -37,6 +37,11 @@ export interface VisualMakerOptions { maxIterations?: number; /** Live activity-line callback (updates the job record). */ onActivity?: (line: string) => void; + /** Kills the run mid-model-call (2026-09 real-kill abort semantics). */ + abortSignal?: AbortSignal; + /** Money channel: one call per loop iteration (generateText), folded + * into JobRecord.usage. */ + onUsage?: (usage: UsageStepLike) => void; } const INSTRUCTIONS = `You are a visual maker — a diagram author + renderer for precise, spatial figures. You @@ -135,7 +140,15 @@ export async function runVisualMaker(input: VisualMakerInput, opts: VisualMakerO try { for (let iteration = 0; iteration < maxIterations; iteration++) { - const result = await generateText({ model: opts.model ?? modelFor("visualMaker"), system: INSTRUCTIONS, messages, tools }); + const result = await generateText({ + model: opts.model ?? modelFor("visualMaker"), + system: INSTRUCTIONS, + messages, + tools, + abortSignal: opts.abortSignal, + }); + // Money channel: each iteration is one model call. + opts.onUsage?.(result.usage); messages = [...messages, ...result.response.messages]; const toolCalls = result.toolCalls; diff --git a/src/core/contracts/jobs.ts b/src/core/contracts/jobs.ts index d2f4448..8774652 100644 --- a/src/core/contracts/jobs.ts +++ b/src/core/contracts/jobs.ts @@ -7,6 +7,7 @@ * of "ai" (safe everywhere: server, client, tests). See docs/vision.md. */ import type { UIMessage } from "ai"; +import type { CallUsage } from "./chat.ts"; // ─── Async sub-agent jobs (vision §2.5; plan.md Wave 0 protocol) ──────────── @@ -36,18 +37,64 @@ export interface JobRecord { /** A sub-agent run's accumulated usage — the same accounting shape as a * chat turn (contracts/chat.ts CallUsage), flattened to one running total - * plus a step count. */ + * plus a step count. Token fields are optional exactly like CallUsage: a + * field NO model call reported stays undefined — absent is not zero (the + * money-channel doctrine; cost renders "—", never $0). */ export interface JobUsage { /** The OpenRouter model id that ran the sub-agent. */ model: string; /** Running totals summed over the run's model calls so far. */ - inputTokens: number; - outputTokens: number; - cacheReadTokens: number; - cacheWriteTokens: number; - reasoningTokens: number; + inputTokens: number | undefined; + outputTokens: number | undefined; + cacheReadTokens: number | undefined; + cacheWriteTokens: number | undefined; + reasoningTokens: number | undefined; /** How many model calls the run has made (each bills separately). */ steps: number; + /** The MOST RECENT model call's accounting (2026-09 dock contract): the + * per-blob context line reads last-step input+output against the model's + * CONTEXT_LIMITS — never the cumulative totals. Absent until the first + * call completes; absent provider fields stay absent (never zero). */ + lastStep?: CallUsage; +} + +/** The per-model-call usage shape we fold into JobUsage — the SDK's + * LanguageModelUsage narrowed structurally (no runtime "ai" import: any + * object with these fields folds, including the real thing and test fakes). */ +export interface UsageStepLike { + inputTokens?: number | undefined; + inputTokenDetails?: { cacheReadTokens?: number; cacheWriteTokens?: number } | undefined; + outputTokens?: number | undefined; + outputTokenDetails?: { reasoningTokens?: number } | undefined; +} + +/** Fold one model call's usage into the run's running total (pure). A + * field neither the total nor the step reported stays undefined; once + * reported, later unreported calls contribute 0. The `steps` counter is + * the billing count: one per model call, reported or not. */ +export function accumulateJobUsage( + total: JobUsage | undefined, + model: string, + step: UsageStepLike, +): JobUsage { + const add = (a: number | undefined, b: number | undefined): number | undefined => + a === undefined && b === undefined ? undefined : (a ?? 0) + (b ?? 0); + return { + model, + inputTokens: add(total?.inputTokens, step.inputTokens), + outputTokens: add(total?.outputTokens, step.outputTokens), + cacheReadTokens: add(total?.cacheReadTokens, step.inputTokenDetails?.cacheReadTokens), + cacheWriteTokens: add(total?.cacheWriteTokens, step.inputTokenDetails?.cacheWriteTokens), + reasoningTokens: add(total?.reasoningTokens, step.outputTokenDetails?.reasoningTokens), + steps: (total?.steps ?? 0) + 1, + lastStep: { + inputTokens: step.inputTokens, + outputTokens: step.outputTokens, + cacheReadTokens: step.inputTokenDetails?.cacheReadTokens, + cacheWriteTokens: step.inputTokenDetails?.cacheWriteTokens, + reasoningTokens: step.outputTokenDetails?.reasoningTokens, + }, + }; } /** @@ -155,6 +202,11 @@ export interface SubagentResultMeta { jobId: string; kind: JobKind; toolCallId: string; + /** Present only on an abort steer (2026-09 abort semantics): the + * learner aborted the job (it lands as error "Aborted by learner"); + * the teacher is told so its promise of a result doesn't hang + * silently in the lesson. */ + aborted?: true; } /** The data-subagent-artifact part payload: what the UI renders. */ @@ -178,6 +230,20 @@ export interface SubagentSteerMessage extends UIMessage { ]; } +/** The abort steer: user-role like any steer, but NO artifact part — + * there is no result to render, only the teacher-facing text. Shares the + * result steer's deterministic id space (`steer_${jobId}`): steerDelivered + * matches on metadata jobId, so an abort and a (racing) result steer + * can never BOTH deliver — exactly one wakes the teacher. */ +export interface AbortSteerMessage extends UIMessage { + role: "user"; + metadata: { subagentResult: SubagentResultMeta & { aborted: true } }; + parts: [{ type: "text"; text: string }]; +} + +/** Anything the wake pipeline may queue and deliver. */ +export type SteerMessage = SubagentSteerMessage | AbortSteerMessage; + /** Whether a result is the failure shape (html/svg empty, failure brief). */ export function isSubagentFailure(kind: JobKind, result: ResearchResult | DemoResult | VisualResult): boolean { if (kind === "research") { @@ -234,6 +300,32 @@ export function buildSubagentSteer( }; } +/** Build the abort steer for an aborted job (pure — shared by the + * abort path in src/server/jobs.ts and tests). Same id space as the result + * steer; see AbortSteerMessage. */ +export function buildAbortSteer( + job: Pick, +): AbortSteerMessage { + return { + id: `steer_${job.id}`, + role: "user", + metadata: { + subagentResult: { jobId: job.id, kind: job.kind, toolCallId: job.toolCallId, aborted: true }, + }, + parts: [{ type: "text", text: abortText(job.kind) }], + }; +} + +/** What the teacher reads: one line telling it the result is not coming, + * how to acknowledge, and that re-dispatching is allowed. */ +function abortText(kind: JobKind): string { + if (kind === "research") + return "Research aborted — the learner aborted the research job, so no brief is coming. Acknowledge it in one line and continue teaching without it (or offer to re-dispatch it)."; + if (kind === "demo") + return "The demo job was aborted by the learner — no demo is coming. Acknowledge it in one line and teach the topic in prose instead (a re-dispatch is fine if the learner wants it)."; + return "The figure job was aborted by the learner — no figure is coming. Acknowledge it in one line and teach without it (a re-dispatch is fine if the learner wants it)."; +} + /** True when the transcript already carries this job's steer (idempotent delivery). */ export function steerDelivered(messages: UIMessage[], jobId: string): boolean { return messages.some( diff --git a/src/server/jobs.ts b/src/server/jobs.ts index 780d6c5..f3a1778 100644 --- a/src/server/jobs.ts +++ b/src/server/jobs.ts @@ -33,13 +33,15 @@ import type { UIMessage } from "ai"; import { + buildAbortSteer, buildSubagentSteer, isSubagentFailure, steerDelivered, type JobRecord, - type SubagentSteerMessage, + type SteerMessage, } from "../core/contracts.ts"; import { findPendingToolCall } from "../core/pending.ts"; +import { fireJobAbort } from "../agent/subagents/aborts.ts"; import { stores } from "./stores.ts"; import { isTurnRunning } from "./turn-liveness.ts"; @@ -50,7 +52,7 @@ export type TurnRunner = (chatId: string, messages: UIMessage[]) => Promise void; interface QueuedSteer { - steer: SubagentSteerMessage; + steer: SteerMessage; resolve: () => void; reject: (reason: Error) => void; } @@ -108,7 +110,7 @@ function delay(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } -function jobIdOf(steer: SubagentSteerMessage): string { +function jobIdOf(steer: SteerMessage): string { return (steer.metadata as { subagentResult: { jobId: string } }).subagentResult.jobId; } @@ -117,7 +119,7 @@ function jobIdOf(steer: SubagentSteerMessage): string { * has completed (and the job status flipped); rejects when delivery fails * unrecoverably. Never rejects for queueing itself. */ -export function deliverJobResult(steer: SubagentSteerMessage): Promise { +export function deliverJobResult(steer: SteerMessage): Promise { return new Promise((resolve, reject) => { const s = state(); const chatId = chatIdOf(steer); @@ -145,13 +147,13 @@ export function deliverJobResult(steer: SubagentSteerMessage): Promise { * before firing, so a claim racing the pump converges to exactly one * delivery either way. */ -export function claimSteers(chatId: string): SubagentSteerMessage[] { +export function claimSteers(chatId: string): SteerMessage[] { const s = state(); const queue = s.queues.get(chatId); if (!queue || queue.length === 0) return []; const record = stores.chatStore.getChat(chatId); if (!record) return []; // the handler validated the chat; defensive only - const claimed: SubagentSteerMessage[] = []; + const claimed: SteerMessage[] = []; for (const entry of queue) { const jobId = jobIdOf(entry.steer); if (steerDelivered(record.messages, jobId)) { @@ -167,7 +169,66 @@ export function claimSteers(chatId: string): SubagentSteerMessage[] { return claimed; } -function chatIdOf(steer: SubagentSteerMessage): string { +/** Result of an abort attempt — the endpoint maps these to HTTP. */ +export interface AbortOutcome { + ok: boolean; + reason?: "not_found" | "not_running"; +} + +/** + * ABORT (2026-09 dock semantics, user-decided): real kill, error with + * reason, the teacher still learns. The dock's END button reaches this + * through POST /api/activity/abort (body {kind: "job", id: jobId}); the + * route is master's thin composition — this is the job-side mechanism. + * Four things happen, in order: + * + * 1. KILL the spend: fire the job's registered AbortController (the + * runner registered it at loop start) — the in-flight model call + * aborts, not just the next loop check. The runner unwinds silently + * (its catch checks abortSignal.aborted; finishJob checks status). + * 2. DROP any queued result steer for this job (the delivering-result + * phase): resolve its entry without firing the wake. If the wake + * turn is ALREADY in flight (the pump shifted the entry), it's too + * late — the result steer lands, and the cancellation steer dedupes + * against it via steerDelivered (same jobId metadata). + * 3. FLIP the record to error "Aborted by learner" immediately — the + * poll reflects the user's action at once (the user chose + * error-with-reason over a distinct cancelled status). finalizeJob + * never overwrites a non-running status, so the flip is terminal no + * matter what the wake does. + * 4. STEER the teacher: a one-line abort note (buildAbortSteer) through + * the same wake pipeline, so the teacher's promise of a result + * doesn't hang silently in the lesson. + * + * The runner writes the usage folded so far in its finally (merge patch), + * so an aborted job's money record survives the kill. + */ +export function abortJob(jobId: string): AbortOutcome { + const job = stores.jobStore.getJob(jobId); + if (!job) return { ok: false, reason: "not_found" }; + if (job.status !== "running") return { ok: false, reason: "not_running" }; + + fireJobAbort(jobId); // no-op when the loop already returned (delivering) + + // Drop any queued steers for this job (their deliver() resolves unfired). + const queue = state().queues.get(job.chatId); + if (queue) { + const remaining = queue.filter((e) => jobIdOf(e.steer) !== jobId); + if (remaining.length === 0) state().queues.delete(job.chatId); + else state().queues.set(job.chatId, remaining); + for (const entry of queue) { + if (jobIdOf(entry.steer) === jobId) entry.resolve(); + } + } + + stores.jobStore.updateJob(jobId, { status: "error", error: "Aborted by learner", activity: "aborted" }); + void deliverJobResult(buildAbortSteer(job)).catch(() => { + /* deliverJobResult marks the job error on unrecoverable failure */ + }); + return { ok: true }; +} + +function chatIdOf(steer: SteerMessage): string { // The chat is resolvable from the job record; the steer's metadata // carries jobId, which indexes the store. const jobId = jobIdOf(steer); @@ -285,6 +346,11 @@ function finalizeJob(jobId: string): void { } function failJob(jobId: string, err: Error): void { + // An aborted job stays aborted even when its (abort) steer + // delivery fails — the user's action is the terminal fact, not the + // delivery error. Non-running records are never overwritten here. + const job = stores.jobStore.getJob(jobId); + if (!job || job.status !== "running") return; stores.jobStore.updateJob(jobId, { status: "error", error: err.message, activity: "delivery failed" }); } diff --git a/tests/demos.test.tsx b/tests/demos.test.tsx index 84c151b..69c3ad8 100644 --- a/tests/demos.test.tsx +++ b/tests/demos.test.tsx @@ -49,7 +49,7 @@ import { runDemoBuilder } from "../src/agent/subagents/demo-builder.ts"; import { headlessOverride, type DemoRunResult, type HeadlessApi } from "../src/agent/subagents/headless.ts"; import { createJobRunner, demoModelOverride, visualModelOverride } from "../src/agent/subagents/runner.ts"; import { runVisualMaker } from "../src/agent/subagents/visual-maker.ts"; -import { buildSubagentSteer, isSubagentFailure, type SubagentSteerMessage } from "../src/core/contracts.ts"; +import { buildAbortSteer, buildSubagentSteer, isSubagentFailure, type SubagentSteerMessage } from "../src/core/contracts.ts"; import type { JobRecord } from "../src/core/contracts.ts"; import { createTeacherTools } from "../src/agent/tools.ts"; import { chatStore } from "../src/store/chats.ts"; @@ -342,6 +342,18 @@ describe("steer card rendering (the delivered artifact)", () => { assert.ok(!html.includes("Looking at the render"), "model-facing notes never render"); }); + it("renders the abort chip for an aborted job's steer (2026-09 dock semantics)", () => { + const steer = buildAbortSteer({ id: "jc", kind: "research", toolCallId: "cc" }); + assert.equal(steer.metadata.subagentResult.aborted, true); + assert.equal(steer.parts.length, 1, "no artifact part on an abort"); + const html = renderToStaticMarkup(); + assert.match(html, /job-aborted-badge/); + assert.match(html, /research/, "the job's kind, lower-cased like the complete-badge idiom"); + assert.match(html, /aborted by the learner/); + // the teacher-facing protocol text never renders for the learner + assert.ok(!html.includes("no brief is coming"), "no protocol glue in the chip"); + }); + it("renders a quiet Research-complete one-liner whose collapsed
carries the full brief", () => { const steer = buildSubagentSteer( { id: "jr", kind: "research", toolCallId: "cr" }, diff --git a/tests/jobs.test.ts b/tests/jobs.test.ts index 49590da..66ec4e8 100644 --- a/tests/jobs.test.ts +++ b/tests/jobs.test.ts @@ -31,18 +31,23 @@ import type { UIMessage } from "ai"; import { MockLanguageModelV4, simulateReadableStream } from "ai/test"; import type { LanguageModelV4StreamPart } from "@ai-sdk/provider"; import { + accumulateJobUsage, + buildAbortSteer, buildSubagentSteer, isSubagentFailure, steerDelivered, type DemoResult, + type JobUsage, type ResearchResult, + type SteerMessage, type SubagentSteerMessage, type VisualResult, } from "../src/core/contracts.ts"; import { composeInstructions, createTeacherAgent } from "../src/agent/teacher-agent.ts"; import { createTeacherTools } from "../src/agent/tools.ts"; import { handleChatRequest } from "../src/server/chat-handler.ts"; -import { claimSteers, deliverJobResult, setTurnRunner } from "../src/server/jobs.ts"; +import { abortJob, claimSteers, deliverJobResult, setTurnRunner } from "../src/server/jobs.ts"; +import { beginTurn, endTurn } from "../src/server/turn-liveness.ts"; import { persistTurn, withMessageIds } from "../src/server/turn-service.ts"; import { chatStore } from "../src/store/chats.ts"; import { jobStore } from "../src/store/jobs.ts"; @@ -170,6 +175,57 @@ describe("steer helpers (contracts.ts)", () => { assert.deepEqual(data.data.result, research); }); + it("accumulateJobUsage: sums per model call; fields NO call reported stay undefined (absent is not zero)", () => { + let total: JobUsage | undefined = undefined; + total = accumulateJobUsage(total, "z-ai/glm-5.3-flash", { + inputTokens: 10, + outputTokens: 5, + inputTokenDetails: { cacheReadTokens: 4 }, + }); + assert.deepEqual(total, { + model: "z-ai/glm-5.3-flash", + inputTokens: 10, + outputTokens: 5, + cacheReadTokens: 4, + cacheWriteTokens: undefined, + reasoningTokens: undefined, + steps: 1, + lastStep: { + inputTokens: 10, + outputTokens: 5, + cacheReadTokens: 4, + cacheWriteTokens: undefined, + reasoningTokens: undefined, + }, + }, "lastStep is the most recent call's accounting, not cumulative"); + // a later call that reports nothing new keeps reported fields summing + // (unreported contributes 0) and bumps the billing step count. + total = accumulateJobUsage(total, "z-ai/glm-5.3-flash", { inputTokens: 20, outputTokens: 0 }); + assert.equal(total.inputTokens, 30); + assert.equal(total.outputTokens, 5); + assert.equal(total.cacheReadTokens, 4, "unreported after reported contributes 0"); + assert.equal(total.cacheWriteTokens, undefined, "never reported stays absent — never zero"); + assert.equal(total.reasoningTokens, undefined); + assert.equal(total.steps, 2, "one step per model call, reported or not"); + assert.deepEqual(total.lastStep, { + inputTokens: 20, + outputTokens: 0, + cacheReadTokens: undefined, + cacheWriteTokens: undefined, + reasoningTokens: undefined, + }, "lastStep always mirrors the LATEST call, never the sums"); + }); + + it("buildAbortSteer: metadata-aborted, text-only (no artifact), same deterministic id space as the result steer", () => { + const steer = buildAbortSteer({ id: "job_c", kind: "demo", toolCallId: "call-c" }); + assert.equal(steer.id, "steer_job_c", "same id space — steerDelivered treats either as delivered"); + assert.equal(steer.role, "user"); + assert.equal(steer.metadata.subagentResult.aborted, true); + assert.equal(steer.metadata.subagentResult.jobId, "job_c"); + assert.equal(steer.parts.length, 1); + assert.ok(steer.parts[0]!.type === "text" && /aborted/.test(steer.parts[0].text)); + }); + it("isSubagentFailure: failure shapes only (empty html/svg, failure brief)", () => { assert.equal(isSubagentFailure("research", research), false); assert.equal(isSubagentFailure("research", { brief: "Research failed: x", sources: [] }), true); @@ -836,3 +892,192 @@ describe("prompt composition", () => { assert.ok(composeInstructions().length <= 60_000, "composed system prompt over the 60k guard"); }); }); + +// ─── abort (2026-09 semantics: real kill, cancelled status, teacher told) ─── + +describe("abort (src/server/jobs.ts abortJob)", () => { + /** A hanging model that only settles when its abort signal fires — the + * real-kill path: the in-flight call itself is killed. */ + function hangingModel(): MockLanguageModelV4 { + return new MockLanguageModelV4({ + doGenerate: ({ abortSignal }) => + new Promise((_resolve, reject) => { + // an ALREADY-aborted signal never fires the event — real + // providers reject an aborted fetch at once, the mock must too + if (abortSignal?.aborted) { + reject(new DOMException("The operation was aborted", "AbortError")); + return; + } + abortSignal?.addEventListener("abort", () => { + reject(new DOMException("The operation was aborted", "AbortError")); + }); + }), + }); + } + + /** One step (with usage) then a hang — proves usage-so-far survives the kill. */ + function oneStepThenHangModel(): MockLanguageModelV4 { + let turn = 0; + return new MockLanguageModelV4({ + doGenerate: ({ abortSignal }) => { + turn++; + if (turn === 1) { + return Promise.resolve({ + content: [ + { + type: "tool-call" as const, + toolCallId: "ws1", + toolName: "web_search", + input: JSON.stringify({ query: "anything" }), + }, + ], + finishReason: { unified: "tool-calls" as const, raw: "tool-calls" }, + usage: usage(), + warnings: [], + }); + } + return new Promise((_resolve, reject) => { + if (abortSignal?.aborted) { + reject(new DOMException("The operation was aborted", "AbortError")); + return; + } + abortSignal?.addEventListener("abort", () => { + reject(new DOMException("The operation was aborted", "AbortError")); + }); + }); + }, + }); + } + + function newChatWithResearchJob(label: string, toolCallId: string): { chatId: string; jobId: string } { + const chat = chatStore.createChat(`Test: ${label}`); + trackChat(chat.id); + chatStore.saveMessages(chat.id, [ + { id: "u1", role: "user", parts: [{ type: "text", text: "hi" }] }, + { + id: "a1", + role: "assistant", + parts: [ + { type: "text", text: "Checking." }, + { type: "tool-research", toolCallId, state: "output-available", input: RESEARCH_INPUT, output: "Dispatched." }, + ], + }, + ] as never); + const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId }); + trackJob(job.id); + return { chatId: chat.id, jobId: job.id }; + } + + it("kills the running loop: error \"Aborted by learner\" immediately, loop unwinds silently, teacher gets the abort steer", async () => { + const { jobId } = newChatWithResearchJob("abort running job", "call-abort1"); + const fired: SteerMessage[] = []; + setTurnRunner(async (_cid, messages) => { + fired.push(messages[messages.length - 1] as SteerMessage); + return 200; + }); + researchModelOverride.current = hangingModel(); + const runner = createJobRunner({ + chatStore, + jobStore, + deliver: async () => { + throw new Error("a cancelled job must never deliver a result steer"); + }, + }); + try { + const settled = runner.runJob(jobId, RESEARCH_INPUT); + await waitFor(() => jobStore.getJob(jobId)!.activity === "researching", 8000, "loop started"); + + const outcome = abortJob(jobId); + assert.ok(outcome.ok); + const killed = jobStore.getJob(jobId)!; + assert.equal(killed.status, "error", "flipped synchronously — the poll sees it at once"); + assert.equal(killed.error, "Aborted by learner", "abort reads as an error with a reason, not a failure"); + assert.equal(killed.activity, "aborted"); + + await settled; // unwinds — no hang + assert.equal(jobStore.getJob(jobId)!.status, "error", "the runner never overwrites the flip"); + assert.equal(jobStore.getJob(jobId)!.error, "Aborted by learner"); + assert.equal(jobStore.getJob(jobId)!.result, undefined, "no result on an aborted job"); + + await waitFor(() => fired.length > 0, 8000, "abort wake fired"); + const last = fired[0]!; + assert.equal(last.metadata.subagentResult.aborted, true); + assert.equal(last.id, `steer_${jobId}`); + assert.equal(last.parts.length, 1, "text only — no artifact part"); + // aborted is terminal: the wake completing must not flip it + assert.equal(jobStore.getJob(jobId)!.status, "error", "terminal after the wake"); + } finally { + researchModelOverride.current = undefined; + } + }); + + it("the usage folded so far survives the kill (money channel outlives the abort)", async () => { + const restoreFetch = stubFetch(); + try { + const { jobId } = newChatWithResearchJob("abort keeps usage", "call-abort2"); + setTurnRunner(async () => 200); + researchModelOverride.current = oneStepThenHangModel(); + const runner = createJobRunner({ chatStore, jobStore, deliver: async () => {} }); + try { + const settled = runner.runJob(jobId, RESEARCH_INPUT); + await waitFor(() => (jobStore.getJob(jobId)!.usage?.steps ?? 0) === 1, 8000, "first step folded"); + abortJob(jobId); + await settled; + const job = jobStore.getJob(jobId)!; + assert.equal(job.status, "error"); + assert.equal(job.error, "Aborted by learner"); + assert.equal(job.usage!.steps, 1, "usage-so-far persisted by the runner's finally"); + assert.equal(job.usage!.model, "z-ai/glm-5.3-flash", "the config role's OpenRouter id"); + assert.equal(job.usage!.inputTokens, 10); + assert.equal(job.usage!.outputTokens, 5); + } finally { + researchModelOverride.current = undefined; + } + } finally { + restoreFetch(); + } + }); + + it("aborting a delivering job drops the queued result steer and wakes with the abort note instead", async () => { + const chat = chatStore.createChat("Test: abort delivering job"); + trackChat(chat.id); + chatStore.saveMessages(chat.id, [{ id: "u1", role: "user", parts: [{ type: "text", text: "hello" }] }] as never); + const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-abort3" }); + trackJob(job.id); + const result: ResearchResult = { brief: "Done: it works.", sources: [{ title: "s", url: "https://x.test" }] }; + jobStore.updateJob(job.id, { result, activity: "delivering result" }); + + // Block the pump's idle gates so the result steer sits QUEUED, not in flight. + assert.ok(beginTurn(chat.id)); + const fired: SteerMessage[] = []; + setTurnRunner(async (_cid, messages) => { + fired.push(messages[messages.length - 1] as SteerMessage); + return 200; + }); + void deliverJobResult(buildSubagentSteer(job, result)).catch(() => { + /* dropped by the abort below */ + }); + await waitFor(() => jobStore.getJob(job.id)!.activity === "delivering result", 8000, "queued"); + + const outcome = abortJob(job.id); + assert.ok(outcome.ok); + assert.equal(jobStore.getJob(job.id)!.status, "error"); + assert.equal(jobStore.getJob(job.id)!.error, "Aborted by learner"); + + endTurn(chat.id); // gates open — whatever is left in the queue fires + await waitFor(() => fired.length > 0, 8000, "wake fired"); + assert.equal(fired.length, 1, "exactly one delivery — the result steer was dropped"); + assert.equal(fired[0]!.metadata.subagentResult.aborted, true, "the abort steer, not the result"); + assert.equal(jobStore.getJob(job.id)!.status, "error", "terminal after the wake"); + }); + + it("outcome reasons: not_found for unknown, not_running for settled jobs", () => { + assert.deepEqual(abortJob("job_does_not_exist"), { ok: false, reason: "not_found" }); + const chat = chatStore.createChat("Test: abort settled job"); + trackChat(chat.id); + const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-abort4" }); + trackJob(job.id); + jobStore.updateJob(job.id, { status: "done", activity: "delivered" }); + assert.deepEqual(abortJob(job.id), { ok: false, reason: "not_running" }); + }); +}); From 9174cde55e34befbabf45493af87d06cf0d7037f Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Fri, 11 Sep 2026 03:25:25 +0300 Subject: [PATCH 2/4] Extraction observability: usage capture, live status payload, true abort (Wave 4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- app/api/graph/status/route.ts | 4 + src/agent/subagents/extraction.ts | 126 ++++++++++++++++++++++++++---- src/core/contracts/chat.ts | 36 +++++++++ src/core/contracts/graph.ts | 11 +++ src/core/graph.ts | 13 +-- src/core/ports.ts | 3 + src/server/observability.ts | 2 +- src/store/chats.ts | 22 +++++- tests/extraction.test.ts | 111 +++++++++++++++++++++++++- tests/usage.test.ts | 33 +++++++- 10 files changed, 330 insertions(+), 31 deletions(-) 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`); +}); From fde46c28fd6db381201760ea7580d0b6144fc782 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Fri, 11 Sep 2026 03:29:21 +0300 Subject: [PATCH 3/4] =?UTF-8?q?Activity=20abort=20route:=20POST=20/api/act?= =?UTF-8?q?ivity/abort=20=E2=80=94=20uniform=20END=20control=20(jobs=20+?= =?UTF-8?q?=20extraction)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/activity/abort/route.ts | 42 +++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) create mode 100644 app/api/activity/abort/route.ts diff --git a/app/api/activity/abort/route.ts b/app/api/activity/abort/route.ts new file mode 100644 index 0000000..5b5de06 --- /dev/null +++ b/app/api/activity/abort/route.ts @@ -0,0 +1,42 @@ +/** + * POST /api/activity/abort — the dock's END/ABORT control, uniform across + * process kinds (user-decided contract): body {kind: "job"|"extraction", id} + * where id is the jobId for sub-agent jobs and the chatId for extraction + * runs (the pipeline registry is chat-keyed). Mechanisms live in their + * domains — abortJob (server/jobs.ts, agents stream) and abortExtraction + * (the pipeline module, graph stream); this route is the thin composition. + * + * Outcomes: a job's not_found -> 404, not_running -> 409, aborted -> 200. + * Extraction abort is idempotent by design (aborting with nothing in + * flight is success — the blob is gone either way) -> 200. + */ +import { abortExtraction, abortJob } from "@/src/server"; + +export async function POST(req: Request) { + let body: unknown; + try { + body = await req.json(); + } catch { + return Response.json({ error: "Invalid JSON body" }, { status: 400 }); + } + const { kind, id } = (body ?? {}) as { kind?: string; id?: string }; + if ((kind !== "job" && kind !== "extraction") || typeof id !== "string" || id === "") { + return Response.json({ error: 'Body must be {kind: "job"|"extraction", id: string}' }, { status: 400 }); + } + if (!/^[A-Za-z0-9_-]+$/.test(id)) { + return Response.json({ error: "Invalid id" }, { status: 400 }); + } + + if (kind === "job") { + const outcome = abortJob(id); + if (!outcome.ok) { + const status = outcome.reason === "not_found" ? 404 : 409; + return Response.json({ ok: false, reason: outcome.reason }, { status }); + } + return Response.json({ ok: true }); + } + // kind === "extraction": true cancellation; idempotent (nothing in + // flight is success — same dock outcome) + abortExtraction(id); + return Response.json({ ok: true }); +} From 090a39b77bae78192f8eb83e8811338e338f9cdc Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Fri, 11 Sep 2026 03:30:25 +0300 Subject: [PATCH 4/4] churn: next-env.d.ts --- next-env.d.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/next-env.d.ts b/next-env.d.ts index ce4e94a..a419cbe 100644 --- a/next-env.d.ts +++ b/next-env.d.ts @@ -1,7 +1,7 @@ /// /// -import "./.next/types/routes.d.ts"; -import "./.next/types/root-params.d.ts"; +import "./.next/dev/types/routes.d.ts"; +import "./.next/dev/types/root-params.d.ts"; // NOTE: This file should not be edited // see https://nextjs.org/docs/app/api-reference/config/typescript for more information.