feat: abort sub-agent jobs (real kill) + runner usage capture into JobRecord.usage
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.
This commit is contained in:
amosblomqvist
@@ -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;
|
||||
}
|
||||
@@ -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);
|
||||
},
|
||||
|
||||
+117
-13
@@ -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<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 {
|
||||
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. */
|
||||
// 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<JobKind, (jobId: string, input?: JobToolInput) => Promise<void>> = {
|
||||
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<void>
|
||||
> = {
|
||||
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> {
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
async function runResearchJob(
|
||||
jobId: string,
|
||||
model: LanguageModel | undefined,
|
||||
passedInput: JobToolInput | undefined,
|
||||
loop: LoopContext,
|
||||
): Promise<void> {
|
||||
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<void> {
|
||||
async function runDemoJob(
|
||||
jobId: string,
|
||||
model: LanguageModel | undefined,
|
||||
passedInput: JobToolInput | undefined,
|
||||
loop: LoopContext,
|
||||
): Promise<void> {
|
||||
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<void> {
|
||||
async function runVisualJob(
|
||||
jobId: string,
|
||||
model: LanguageModel | undefined,
|
||||
passedInput: JobToolInput | undefined,
|
||||
loop: LoopContext,
|
||||
): Promise<void> {
|
||||
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}` });
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in New Issue
Block a user