Merge main: wave-4 backend (activity abort endpoint, lastStep capture)

This commit is contained in:
amosblomqvist
2026-09-11 03:33:15 +03:00
21 changed files with 1021 additions and 65 deletions
+42
View File
@@ -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 });
}
+4
View File
@@ -15,6 +15,10 @@ export async function GET() {
title: chatStore.getChat(run.chatId)?.title ?? run.chatId, title: chatStore.getChat(run.chatId)?.title ?? run.chatId,
phase: run.phase, phase: run.phase,
startedAt: run.startedAt, 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 }); return Response.json({ runs });
} }
+21 -2
View File
@@ -1,6 +1,6 @@
"use client"; "use client";
import { CheckIcon } from "lucide-react"; import { CheckIcon, CircleX } from "lucide-react";
import type { ResearchResult, SubagentArtifactData, TeacherUIMessage } from "@/src/core"; import type { ResearchResult, SubagentArtifactData, TeacherUIMessage } from "@/src/core";
import { isSubagentFailure } from "@/src/core"; import { isSubagentFailure } from "@/src/core";
import { DemoOutputCard } from "@/components/artifacts/demo-card"; import { DemoOutputCard } from "@/components/artifacts/demo-card";
@@ -27,9 +27,15 @@ import { VisualOutputCard } from "@/components/artifacts/visual-card";
export function SubagentSteerCard({ message }: { message: TeacherUIMessage }) { export function SubagentSteerCard({ message }: { message: TeacherUIMessage }) {
const artifactPart = message.parts.find((p) => p.type === "data-subagent-artifact"); const artifactPart = message.parts.find((p) => p.type === "data-subagent-artifact");
const data = artifactPart?.data as SubagentArtifactData | undefined; 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) { if (!data) {
const meta = (
message.metadata as { subagentResult?: { aborted?: boolean; kind?: string } } | undefined
)?.subagentResult;
if (meta?.aborted) return <AbortedBadge kind={meta.kind ?? "job"} />;
return ( return (
<div className="rounded-xl border bg-card p-4 text-sm text-muted-foreground"> <div className="rounded-xl border bg-card p-4 text-sm text-muted-foreground">
A sub-agent finished, but its result could not be displayed. A sub-agent finished, but its result could not be displayed.
</div> </div>
); );
@@ -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 (
<div data-testid="job-aborted-badge" className="flex items-center gap-2 text-xs text-muted-foreground">
<CircleX className="size-3.5" />
<span className="font-medium uppercase tracking-wide">{kind}</span>
<span>aborted by the learner</span>
</div>
);
}
/** Quiet one-liner, the dispatch-settled idiom — the lesson lives in the /** Quiet one-liner, the dispatch-settled idiom — the lesson lives in the
* teacher's wake turn; the badge just marks delivery. The persisted brief * teacher's wake turn; the badge just marks delivery. The persisted brief
* rides along in a collapsed <details> (the wave-1 "view full research" * rides along in a collapsed <details> (the wave-1 "view full research"
+4
View File
@@ -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 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 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. 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).
+50
View File
@@ -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<string, AbortController>;
}
function registry(): Map<string, AbortController> {
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;
}
+7 -1
View File
@@ -1,7 +1,7 @@
import { ToolLoopAgent, stepCountIs, tool, type LanguageModel, type ToolSet } from "ai"; import { ToolLoopAgent, stepCountIs, tool, type LanguageModel, type ToolSet } from "ai";
import { z } from "zod"; import { z } from "zod";
import { modelFor } from "../models.ts"; 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"; import { getHeadless, type DemoRunResult, type HeadlessApi } from "./headless.ts";
/** /**
@@ -32,6 +32,10 @@ export interface DemoBuilderOptions {
maxSteps?: number; maxSteps?: number;
/** Live activity-line callback (updates the job record). */ /** Live activity-line callback (updates the job record). */
onActivity?: (line: string) => void; 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 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 { try {
const result = await agent.generate({ const result = await agent.generate({
prompt, prompt,
abortSignal: opts.abortSignal,
onStepFinish: ({ usage }) => opts.onUsage?.(usage),
onToolExecutionStart: ({ toolCall }) => { onToolExecutionStart: ({ toolCall }) => {
opts.onActivity?.(toolCall.toolName); opts.onActivity?.(toolCall.toolName);
}, },
+111 -15
View File
@@ -1,7 +1,15 @@
import { ToolLoopAgent, stepCountIs, tool, type LanguageModel, type UIMessage } from "ai"; import { ToolLoopAgent, stepCountIs, tool, type LanguageModel, type UIMessage } from "ai";
import { z } from "zod"; import { z } from "zod";
import { modelFor } from "../models.ts"; 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"; import { getGraph, mergeFindings, type EmissionStats, type GraphDeps } from "../../core/graph.ts";
/** /**
@@ -211,12 +219,6 @@ function createExtractor(model: LanguageModel) {
}), }),
}, },
stopWhen: stepCountIs(4), 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), 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. */ /** The pipeline phases, in order — what the home status endpoint shows. */
export type ExtractionPhase = "extracting" | "auditing" | "merging"; export type ExtractionPhase = "extracting" | "auditing" | "merging";
/** One in-flight pipeline run. Module-level: one process, one registry. */ /** One in-flight pipeline run. Module-level: one process, one registry.
interface RunningExtraction { * Exported for the status payload's consumer typing (the dock). */
export interface RunningExtraction {
chatId: string; chatId: string;
phase: ExtractionPhase; phase: ExtractionPhase;
startedAt: number; 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<string, RunningExtraction>(); const runningExtractions = new Map<string, RunningExtraction>();
@@ -326,6 +331,18 @@ export function extractionStatuses(): RunningExtraction[] {
return [...runningExtractions.values()].sort((a, b) => a.startedAt - b.startedAt); 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 ─────────────────────────────────────────────────────────── // ─── the pipeline ───────────────────────────────────────────────────────────
export interface ExtractionModels { export interface ExtractionModels {
@@ -350,21 +367,62 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
const inFlight = new Set<string>(); const inFlight = new Set<string>();
function setPhase(chatId: string, phase: ExtractionPhase): void { function setPhase(chatId: string, phase: ExtractionPhase): void {
const prev = runningExtractions.get(chatId);
runningExtractions.set(chatId, { runningExtractions.set(chatId, {
chatId, chatId,
phase, 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<SessionFindings> { async function extract(chatId: string): Promise<SessionFindings> {
const record = chatStore.getChat(chatId); const record = chatStore.getChat(chatId);
if (!record) throw new Error(`Chat not found: ${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); const view = transcriptView(record);
console.info(`[extraction ${chatId}] extractor starting (${view.split("\n").length} transcript lines)`); console.info(`[extraction ${chatId}] extractor starting (${view.split("\n").length} transcript lines)`);
const t0 = performance.now(); 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) { for (const part of extraction.content) {
if (part.type === "tool-result" && part.toolName === "submit_findings") { if (part.type === "tool-result" && part.toolName === "submit_findings") {
console.info(`[extraction ${chatId}] extractor done in ${Math.round(performance.now() - t0)}ms`); 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<AuditResult> { async function audit(chatId: string, findings: SessionFindings): Promise<AuditResult> {
const record = chatStore.getChat(chatId); const record = chatStore.getChat(chatId);
if (!record) throw new Error(`Chat not found: ${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 view = transcriptView(record);
const graph = getGraph(graphStore); const graph = getGraph(graphStore);
console.info( console.info(
@@ -389,6 +449,13 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
prompt: prompt:
`Transcript:\n\n${view}\n\nProposed findings:\n\n${JSON.stringify(findings, null, 2)}\n\n` + `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)}`, `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; let auditResult: AuditResult | null = null;
for (const part of review.content) { for (const part of review.content) {
@@ -412,6 +479,25 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
return auditResult; 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<EmissionStats> { async function runExtraction(chatId: string): Promise<EmissionStats> {
if (inFlight.has(chatId)) return { nodes: 0, edges: 0, events: 0 }; if (inFlight.has(chatId)) return { nodes: 0, edges: 0, events: 0 };
inFlight.add(chatId); inFlight.add(chatId);
@@ -426,12 +512,22 @@ export function createExtractionPipeline(deps: ExtractionPipelineDeps): Extracti
// The merger re-reads the chat record itself — events are rebuilt // The merger re-reads the chat record itself — events are rebuilt
// from the transcript, never from stale copies. // from the transcript, never from stale copies.
const stats = mergeFindings(chatId, auditResult, { chatStore, graphStore }); const stats = mergeFindings(chatId, auditResult, { chatStore, graphStore });
persistRunUsage(chatId, Math.round(performance.now() - t0), stats);
console.info( console.info(
`[extraction ${chatId}] merged: ${stats.nodes} nodes, ${stats.edges} edges, ${stats.events} events ` + `[extraction ${chatId}] merged: ${stats.nodes} nodes, ${stats.edges} edges, ${stats.events} events ` +
`(${Math.round(performance.now() - t0)}ms total)`, `(${Math.round(performance.now() - t0)}ms total)`,
); );
return stats; return stats;
} catch (err) { } 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 // The throw travels to the caller; this line is what makes the
// run's log legible end-to-end (no hanging "pipeline start"). // run's log legible end-to-end (no hanging "pipeline start").
console.error(`[extraction ${chatId}] pipeline failed (${Math.round(performance.now() - t0)}ms):`, err); console.error(`[extraction ${chatId}] pipeline failed (${Math.round(performance.now() - t0)}ms):`, err);
+117 -13
View File
@@ -1,16 +1,21 @@
import { isToolUIPart, type LanguageModel, type UIMessage } from "ai"; import { isToolUIPart, type LanguageModel, type UIMessage } from "ai";
import { import {
accumulateJobUsage,
buildSubagentSteer, buildSubagentSteer,
type DemoResult, type DemoResult,
type DemoToolInput, type DemoToolInput,
type JobKind, type JobKind,
type JobRecord, type JobRecord,
type JobUsage,
type ResearchResult, type ResearchResult,
type ResearchToolInput, type ResearchToolInput,
type SubagentSteerMessage, type SubagentSteerMessage,
type UsageStepLike,
type VisualResult, type VisualResult,
type VisualToolInput, type VisualToolInput,
} from "../../core/contracts.ts"; } from "../../core/contracts.ts";
import { registerJobAbort, unregisterJobAbort } from "./aborts.ts";
import { MODELS } from "../../../config.ts";
import type { ChatStore, JobStore } from "../../core/ports.ts"; import type { ChatStore, JobStore } from "../../core/ports.ts";
import { runDemoBuilder } from "./demo-builder.ts"; import { runDemoBuilder } from "./demo-builder.ts";
import { createResearcherAgent } from "./researcher.ts"; import { createResearcherAgent } from "./researcher.ts";
@@ -67,16 +72,70 @@ export interface JobRunner {
runJob(jobId: string, input?: ResearchToolInput | DemoToolInput | VisualToolInput): Promise<void>; runJob(jobId: string, input?: ResearchToolInput | DemoToolInput | VisualToolInput): Promise<void>;
} }
/** 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<JobKind, string> = {
research: MODELS.researcher.model,
demo: MODELS.demoBuilder.model,
visual: MODELS.visualMaker.model,
};
export function createJobRunner(deps: JobRunnerDeps): JobRunner { export function createJobRunner(deps: JobRunnerDeps): JobRunner {
const { chatStore, jobStore, deliver } = deps; 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<string, JobUsage>();
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. */ /** The kind→runner map: one sub-agent loop per job kind. */
// Each kind's runner narrows the union input itself (dispatch-time inputs // Each kind's runner narrows the union input itself (dispatch-time inputs
// are kind-correct by construction; transcript fallbacks are cast after). // are kind-correct by construction; transcript fallbacks are cast after).
const KIND_RUNNERS: Record<JobKind, (jobId: string, input?: JobToolInput) => Promise<void>> = { const KIND_RUNNERS: Record<
research: (jobId, input) => runResearchJob(jobId, researchModelOverride.current, input), JobKind,
demo: (jobId, input) => runDemoJob(jobId, demoModelOverride.current, input), (jobId: string, input: JobToolInput | undefined, loop: LoopContext) => Promise<void>
visual: (jobId, input) => runVisualJob(jobId, visualModelOverride.current, input), > = {
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<void> { async function runJob(jobId: string, input?: JobToolInput): Promise<void> {
@@ -92,7 +151,17 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
jobStore.updateJob(jobId, { status: "error", error: `unknown kind ${job.kind}`, activity: "failed" }); jobStore.updateJob(jobId, { status: "error", error: `unknown kind ${job.kind}`, activity: "failed" });
return; 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 /** 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<void> { async function finishJob(jobId: string, result: ResearchResult | DemoResult | VisualResult): Promise<void> {
const job = jobStore.getJob(jobId); const job = jobStore.getJob(jobId);
if (!job) return; 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" }); jobStore.updateJob(jobId, { result, activity: "delivering result" });
const steer = buildSubagentSteer(job, result); const steer = buildSubagentSteer(job, result);
try { try {
@@ -157,7 +230,12 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
// ─── research ─────────────────────────────────────────────────────────── // ─── research ───────────────────────────────────────────────────────────
async function runResearchJob(jobId: string, model?: LanguageModel, passedInput?: JobToolInput): Promise<void> { async function runResearchJob(
jobId: string,
model: LanguageModel | undefined,
passedInput: JobToolInput | undefined,
loop: LoopContext,
): Promise<void> {
const ctx = jobContext(jobId); const ctx = jobContext(jobId);
if (!ctx) return; if (!ctx) return;
const { job } = ctx; const { job } = ctx;
@@ -181,6 +259,10 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
jobStore.updateJob(jobId, { activity: "researching" }); jobStore.updateJob(jobId, { activity: "researching" });
const result = await agent.generate({ const result = await agent.generate({
prompt, 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 }) => { onToolExecutionStart: ({ toolCall }) => {
const input = toolCall.input as { query?: string; url?: string } | undefined; const input = toolCall.input as { query?: string; url?: string } | undefined;
if (toolCall.toolName === "web_fetch" && typeof input?.url === "string") { 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) }; const research: ResearchResult = { brief, sources: parseSources(brief, fetchedUrls) };
await finishJob(jobId, research); await finishJob(jobId, research);
} catch (err) { } 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); 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: [] }); await finishJob(jobId, { brief: `Research failed: ${reason}`, sources: [] });
} }
} }
// ─── demo + visual ────────────────────────────────────────────────────── // ─── demo + visual ──────────────────────────────────────────────────────
async function runDemoJob(jobId: string, model?: LanguageModel, passedInput?: JobToolInput): Promise<void> { async function runDemoJob(
jobId: string,
model: LanguageModel | undefined,
passedInput: JobToolInput | undefined,
loop: LoopContext,
): Promise<void> {
const ctx = jobContext(jobId); const ctx = jobContext(jobId);
if (!ctx) return; if (!ctx) return;
const { job } = ctx; const { job } = ctx;
@@ -218,19 +309,28 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
{ title, brief }, { title, brief },
{ {
model, model,
abortSignal: loop.abortSignal,
onUsage: loop.onUsage,
onActivity: (line) => jobStore.updateJob(jobId, { activity: line }), 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); await finishJob(jobId, result);
} catch (err) { } catch (err) {
if (loop.abortSignal.aborted) return;
const reason = err instanceof Error ? err.message : String(err); 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}` }); await finishJob(jobId, { title, html: "", tested: true, notes: `FAILED: ${reason}` });
} }
} }
async function runVisualJob(jobId: string, model?: LanguageModel, passedInput?: JobToolInput): Promise<void> { async function runVisualJob(
jobId: string,
model: LanguageModel | undefined,
passedInput: JobToolInput | undefined,
loop: LoopContext,
): Promise<void> {
const ctx = jobContext(jobId); const ctx = jobContext(jobId);
if (!ctx) return; if (!ctx) return;
const { job } = ctx; const { job } = ctx;
@@ -244,14 +344,18 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner {
{ brief }, { brief },
{ {
model, model,
abortSignal: loop.abortSignal,
onUsage: loop.onUsage,
onActivity: (line) => jobStore.updateJob(jobId, { activity: line }), 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); await finishJob(jobId, result);
} catch (err) { } catch (err) {
if (loop.abortSignal.aborted) return;
const reason = err instanceof Error ? err.message : String(err); 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}` }); await finishJob(jobId, { svg: "", notes: `FAILED: ${reason}` });
} }
} }
+15 -2
View File
@@ -1,7 +1,7 @@
import { generateText, tool, type LanguageModel, type ModelMessage, type ToolResultPart } from "ai"; import { generateText, tool, type LanguageModel, type ModelMessage, type ToolResultPart } from "ai";
import { z } from "zod"; import { z } from "zod";
import { modelFor } from "../models.ts"; 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"; import { getHeadless, type HeadlessApi } from "./headless.ts";
/** /**
@@ -37,6 +37,11 @@ export interface VisualMakerOptions {
maxIterations?: number; maxIterations?: number;
/** Live activity-line callback (updates the job record). */ /** Live activity-line callback (updates the job record). */
onActivity?: (line: string) => void; 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 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 { try {
for (let iteration = 0; iteration < maxIterations; iteration++) { 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]; messages = [...messages, ...result.response.messages];
const toolCalls = result.toolCalls; const toolCalls = result.toolCalls;
+36
View File
@@ -9,6 +9,7 @@
import type { UIMessage } from "ai"; import type { UIMessage } from "ai";
import type { Phase3 } from "../phases.ts"; import type { Phase3 } from "../phases.ts";
import type { PlanNodeStatus, PlanTree } from "./plan.ts"; import type { PlanNodeStatus, PlanTree } from "./plan.ts";
import type { EmissionStats } from "./graph.ts";
// ─── Chat records (the session store's on-disk format) ─────────────────────── // ─── 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 * each entry onto its message; aggregates are folds over this log
* (core/prices.ts). */ * (core/prices.ts). */
turns?: TurnUsage[]; 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) ───────────────────────────────────── // ─── Usage accounting (the money channel) ─────────────────────────────────────
+11
View File
@@ -22,6 +22,17 @@
* strength — strength is derived at read time from topology. */ * strength — strength is derived at read time from topology. */
export type EdgeType = "follows" | "supports"; 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 /** One provenance point — message-granular (DECIDED 2026-09-02): nodes and
* edges record every (chat, message) where they came up, deep-linkable to * edges record every (chat, message) where they came up, deep-linkable to
* the exact transcript point. */ * the exact transcript point. */
+98 -6
View File
@@ -7,6 +7,7 @@
* of "ai" (safe everywhere: server, client, tests). See docs/vision.md. * of "ai" (safe everywhere: server, client, tests). See docs/vision.md.
*/ */
import type { UIMessage } from "ai"; import type { UIMessage } from "ai";
import type { CallUsage } from "./chat.ts";
// ─── Async sub-agent jobs (vision §2.5; plan.md Wave 0 protocol) ──────────── // ─── 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 /** A sub-agent run's accumulated usage — the same accounting shape as a
* chat turn (contracts/chat.ts CallUsage), flattened to one running total * 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 { export interface JobUsage {
/** The OpenRouter model id that ran the sub-agent. */ /** The OpenRouter model id that ran the sub-agent. */
model: string; model: string;
/** Running totals summed over the run's model calls so far. */ /** Running totals summed over the run's model calls so far. */
inputTokens: number; inputTokens: number | undefined;
outputTokens: number; outputTokens: number | undefined;
cacheReadTokens: number; cacheReadTokens: number | undefined;
cacheWriteTokens: number; cacheWriteTokens: number | undefined;
reasoningTokens: number; reasoningTokens: number | undefined;
/** How many model calls the run has made (each bills separately). */ /** How many model calls the run has made (each bills separately). */
steps: number; 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; jobId: string;
kind: JobKind; kind: JobKind;
toolCallId: string; 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. */ /** 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). */ /** Whether a result is the failure shape (html/svg empty, failure brief). */
export function isSubagentFailure(kind: JobKind, result: ResearchResult | DemoResult | VisualResult): boolean { export function isSubagentFailure(kind: JobKind, result: ResearchResult | DemoResult | VisualResult): boolean {
if (kind === "research") { 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<JobRecord, "id" | "kind" | "toolCallId">,
): 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). */ /** True when the transcript already carries this job's steer (idempotent delivery). */
export function steerDelivered(messages: UIMessage[], jobId: string): boolean { export function steerDelivered(messages: UIMessage[], jobId: string): boolean {
return messages.some( return messages.some(
+4 -9
View File
@@ -42,6 +42,7 @@ import { randomBytes } from "node:crypto";
import { isToolUIPart, type UIMessage } from "ai"; import { isToolUIPart, type UIMessage } from "ai";
import type { import type {
AuditResult, AuditResult,
EmissionStats,
GraphEdge, GraphEdge,
GraphNode, GraphNode,
ProvenancePoint, ProvenancePoint,
@@ -108,15 +109,6 @@ const samePoint = (a: ProvenancePoint, b: ProvenancePoint): boolean =>
// ─── the merger (extraction pipeline stage 3 — deterministic) ──────────────── // ─── 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 * Merge one chat's audited findings into the graph ledgers — the pipeline's
* apply stage. Deterministic + idempotent: re-merging the same verdicts (or * 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 }; 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, JobKind,
JobRecord, JobRecord,
LedgerEvent, LedgerEvent,
ExtractionRunUsage,
NodeMapping, NodeMapping,
PlanTree, PlanTree,
SessionCheckpoint, SessionCheckpoint,
@@ -41,6 +42,8 @@ export interface ChatStore {
* 2026-09): idempotent per messageId — a turn persisted twice (abort * 2026-09): idempotent per messageId — a turn persisted twice (abort
* replays, wake-turn retries) appends exactly once. */ * replays, wake-turn retries) appends exactly once. */
appendTurnUsage(id: string, entry: TurnUsage): void; 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) ────────────── // ─── GraphStore — the append-only evidence ledgers (data/graph) ──────────────
+73 -7
View File
@@ -33,13 +33,15 @@
import type { UIMessage } from "ai"; import type { UIMessage } from "ai";
import { import {
buildAbortSteer,
buildSubagentSteer, buildSubagentSteer,
isSubagentFailure, isSubagentFailure,
steerDelivered, steerDelivered,
type JobRecord, type JobRecord,
type SubagentSteerMessage, type SteerMessage,
} from "../core/contracts.ts"; } from "../core/contracts.ts";
import { findPendingToolCall } from "../core/pending.ts"; import { findPendingToolCall } from "../core/pending.ts";
import { fireJobAbort } from "../agent/subagents/aborts.ts";
import { stores } from "./stores.ts"; import { stores } from "./stores.ts";
import { isTurnRunning } from "./turn-liveness.ts"; import { isTurnRunning } from "./turn-liveness.ts";
@@ -50,7 +52,7 @@ export type TurnRunner = (chatId: string, messages: UIMessage[]) => Promise<numb
export type JobRespawn = (jobId: string) => void; export type JobRespawn = (jobId: string) => void;
interface QueuedSteer { interface QueuedSteer {
steer: SubagentSteerMessage; steer: SteerMessage;
resolve: () => void; resolve: () => void;
reject: (reason: Error) => void; reject: (reason: Error) => void;
} }
@@ -108,7 +110,7 @@ function delay(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms)); 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; 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 * has completed (and the job status flipped); rejects when delivery fails
* unrecoverably. Never rejects for queueing itself. * unrecoverably. Never rejects for queueing itself.
*/ */
export function deliverJobResult(steer: SubagentSteerMessage): Promise<void> { export function deliverJobResult(steer: SteerMessage): Promise<void> {
return new Promise<void>((resolve, reject) => { return new Promise<void>((resolve, reject) => {
const s = state(); const s = state();
const chatId = chatIdOf(steer); const chatId = chatIdOf(steer);
@@ -145,13 +147,13 @@ export function deliverJobResult(steer: SubagentSteerMessage): Promise<void> {
* before firing, so a claim racing the pump converges to exactly one * before firing, so a claim racing the pump converges to exactly one
* delivery either way. * delivery either way.
*/ */
export function claimSteers(chatId: string): SubagentSteerMessage[] { export function claimSteers(chatId: string): SteerMessage[] {
const s = state(); const s = state();
const queue = s.queues.get(chatId); const queue = s.queues.get(chatId);
if (!queue || queue.length === 0) return []; if (!queue || queue.length === 0) return [];
const record = stores.chatStore.getChat(chatId); const record = stores.chatStore.getChat(chatId);
if (!record) return []; // the handler validated the chat; defensive only if (!record) return []; // the handler validated the chat; defensive only
const claimed: SubagentSteerMessage[] = []; const claimed: SteerMessage[] = [];
for (const entry of queue) { for (const entry of queue) {
const jobId = jobIdOf(entry.steer); const jobId = jobIdOf(entry.steer);
if (steerDelivered(record.messages, jobId)) { if (steerDelivered(record.messages, jobId)) {
@@ -167,7 +169,66 @@ export function claimSteers(chatId: string): SubagentSteerMessage[] {
return claimed; 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 // The chat is resolvable from the job record; the steer's metadata
// carries jobId, which indexes the store. // carries jobId, which indexes the store.
const jobId = jobIdOf(steer); const jobId = jobIdOf(steer);
@@ -285,6 +346,11 @@ function finalizeJob(jobId: string): void {
} }
function failJob(jobId: string, err: Error): 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" }); stores.jobStore.updateJob(jobId, { status: "error", error: err.message, activity: "delivery failed" });
} }
+6 -2
View File
@@ -13,6 +13,10 @@
* (chat-transport); the dock is a different consumer with a different * (chat-transport); the dock is a different consumer with a different
* shape, served here. Single sources of truth are unchanged: the job * shape, served here. Single sources of truth are unchanged: the job
* store and the pipeline registry — this file only composes them. * store and the pipeline registry — this file only composes them.
*
* The extraction side of the dock's controls also surfaces here:
* abortExtraction (true mid-run cancellation, user-decided) is exported
* from the pipeline's module for the abort route to call.
*/ */
import { extractionStatuses, type ExtractionPhase } from "../agent/subagents/extraction.ts"; import { extractionStatuses, type ExtractionPhase } from "../agent/subagents/extraction.ts";
@@ -20,8 +24,8 @@ import { chatStore } from "../store/chats.ts";
import { listJobs } from "./jobs.ts"; import { listJobs } from "./jobs.ts";
import type { JobRecord } from "../core/contracts.ts"; import type { JobRecord } from "../core/contracts.ts";
export { extractionStatuses, type ExtractionPhase }; export { abortExtraction, extractionStatuses, type ExtractionPhase, type RunningExtraction } from "../agent/subagents/extraction.ts";
export { chatStore }; export { chatStore } from "../store/chats.ts";
/** A running sub-agent job, joined with its chat's title for the dock. */ /** A running sub-agent job, joined with its chat's title for the dock. */
export interface ActivityJob extends JobRecord { export interface ActivityJob extends JobRecord {
+18 -4
View File
@@ -11,7 +11,7 @@ import * as fs from "node:fs";
import * as path from "node:path"; import * as path from "node:path";
import { PATHS } from "../../config.ts"; import { PATHS } from "../../config.ts";
import type { ChatStore } from "../core/ports.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 type { UIMessage } from "ai";
import { findPendingToolCall } from "../core/pending.ts"; import { findPendingToolCall } from "../core/pending.ts";
import { phaseOfTree } from "../core/phases.ts"; import { phaseOfTree } from "../core/phases.ts";
@@ -51,9 +51,10 @@ function listChats(): ChatSummary[] {
.filter((f) => f.endsWith(".json")) .filter((f) => f.endsWith(".json"))
.map((f) => getChat(f.slice(0, -5))) .map((f) => getChat(f.slice(0, -5)))
.filter((r): r is ChatRecord => r !== null) .filter((r): r is ChatRecord => r !== null)
.map(({ messages, tree, turns, ...summary }) => ({ .map(({ messages, tree, turns, extractions, ...summary }) => ({
// turns stay out of the list payload (list = summaries; usage // turns and extractions stay out of the list payload (list =
// aggregates ride the full record, the picker never needs them). // summaries; usage aggregates ride the full record, the picker
// never needs them).
...summary, ...summary,
messageCount: messages.length, messageCount: messages.length,
// Record-derived status for the session picker: phase from the // 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"); 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). */ /** Persist the session plan tree (plan tools call this in execute). */
function saveTree(id: string, tree: PlanTree): void { function saveTree(id: string, tree: PlanTree): void {
const record = getChat(id); const record = getChat(id);
@@ -107,4 +120,5 @@ export const chatStore: ChatStore = {
saveMessages, saveMessages,
saveTree, saveTree,
appendTurnUsage, appendTurnUsage,
appendExtractionUsage,
}; };
+13 -1
View File
@@ -48,7 +48,7 @@ import { runDemoBuilder } from "../src/agent/subagents/demo-builder.ts";
import { headlessOverride, type DemoRunResult, type HeadlessApi } from "../src/agent/subagents/headless.ts"; import { headlessOverride, type DemoRunResult, type HeadlessApi } from "../src/agent/subagents/headless.ts";
import { createJobRunner, demoModelOverride, visualModelOverride } from "../src/agent/subagents/runner.ts"; import { createJobRunner, demoModelOverride, visualModelOverride } from "../src/agent/subagents/runner.ts";
import { runVisualMaker } from "../src/agent/subagents/visual-maker.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 type { JobRecord } from "../src/core/contracts.ts";
import { createTeacherTools } from "../src/agent/tools.ts"; import { createTeacherTools } from "../src/agent/tools.ts";
import { chatStore } from "../src/store/chats.ts"; import { chatStore } from "../src/store/chats.ts";
@@ -341,6 +341,18 @@ describe("steer card rendering (the delivered artifact)", () => {
assert.ok(!html.includes("Looking at the render"), "model-facing notes never render"); 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(<SubagentSteerCard message={steer as never} />);
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 <details> carries the full brief", () => { it("renders a quiet Research-complete one-liner whose collapsed <details> carries the full brief", () => {
const steer = buildSubagentSteer( const steer = buildSubagentSteer(
{ id: "jr", kind: "research", toolCallId: "cr" }, { id: "jr", kind: "research", toolCallId: "cr" },
+110 -1
View File
@@ -34,7 +34,7 @@ import type {
SessionFindings, SessionFindings,
} from "../src/core/contracts.ts"; } from "../src/core/contracts.ts";
import type { GraphStore } from "../src/core/ports.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 { chatStore } from "../src/store/chats.ts";
import { PATHS } from "../config.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<never> => {
await new Promise<never>((_, 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). */ /** A model that never calls a tool (the failure path). */
function silentModel(): MockLanguageModelV4 { function silentModel(): MockLanguageModelV4 {
return new 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"); 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 () => { it("is single-flight per chat — a concurrent second run is a no-op", async () => {
const { chatId } = fixtureChat(); const { chatId } = fixtureChat();
const ledgers = memoryLedgers(); const ledgers = memoryLedgers();
+246 -1
View File
@@ -31,18 +31,23 @@ import type { UIMessage } from "ai";
import { MockLanguageModelV4, simulateReadableStream } from "ai/test"; import { MockLanguageModelV4, simulateReadableStream } from "ai/test";
import type { LanguageModelV4StreamPart } from "@ai-sdk/provider"; import type { LanguageModelV4StreamPart } from "@ai-sdk/provider";
import { import {
accumulateJobUsage,
buildAbortSteer,
buildSubagentSteer, buildSubagentSteer,
isSubagentFailure, isSubagentFailure,
steerDelivered, steerDelivered,
type DemoResult, type DemoResult,
type JobUsage,
type ResearchResult, type ResearchResult,
type SteerMessage,
type SubagentSteerMessage, type SubagentSteerMessage,
type VisualResult, type VisualResult,
} from "../src/core/contracts.ts"; } from "../src/core/contracts.ts";
import { composeInstructions, createTeacherAgent } from "../src/agent/teacher-agent.ts"; import { composeInstructions, createTeacherAgent } from "../src/agent/teacher-agent.ts";
import { createTeacherTools } from "../src/agent/tools.ts"; import { createTeacherTools } from "../src/agent/tools.ts";
import { handleChatRequest } from "../src/server/chat-handler.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 { persistTurn, withMessageIds } from "../src/server/turn-service.ts";
import { chatStore } from "../src/store/chats.ts"; import { chatStore } from "../src/store/chats.ts";
import { jobStore } from "../src/store/jobs.ts"; import { jobStore } from "../src/store/jobs.ts";
@@ -170,6 +175,57 @@ describe("steer helpers (contracts.ts)", () => {
assert.deepEqual(data.data.result, research); 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)", () => { it("isSubagentFailure: failure shapes only (empty html/svg, failure brief)", () => {
assert.equal(isSubagentFailure("research", research), false); assert.equal(isSubagentFailure("research", research), false);
assert.equal(isSubagentFailure("research", { brief: "Research failed: x", sources: [] }), true); 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"); 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" });
});
});
+32 -1
View File
@@ -12,7 +12,7 @@ import * as fs from "node:fs";
import { chatStore } from "../src/store/chats.ts"; import { chatStore } from "../src/store/chats.ts";
import { PATHS } from "../config.ts"; import { PATHS } from "../config.ts";
import { callCost, contextWindow, jobCost, sessionCost, sumSteps, turnCost } from "../src/core/prices.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> = {}): CallUsage => ({ const call = (over: Partial<CallUsage> = {}): CallUsage => ({
inputTokens: 0, inputTokens: 0,
@@ -134,3 +134,34 @@ test("appendTurnUsage: appends, is idempotent per messageId, survives saveMessag
fs.unlinkSync(`${PATHS.chats}/${record.id}.json`); 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`);
});