From 99fa3f084731c259c43d3644fbc1a0c2e7005a6b Mon Sep 17 00:00:00 2001 From: Amos Date: Tue, 1 Sep 2026 19:15:43 +0300 Subject: [PATCH] Pi-style ack-and-steer sub-agent protocol: dispatch mid-turn, steer via server wake turn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The job tools (research/demo/visual) switch from client-side turn-breaking calls to pi's model: execute resolves IMMEDIATELY with a fire-and-forget ack, the teacher's turn continues, and the result arrives later as an injected user-role steer message that wakes the teacher in a new server-side turn. - contracts.ts: THE ASYNC PROTOCOL rewritten (ack now, steer later); steer types (SubagentSteerMessage, metadata.subagentResult + data-subagent- artifact part), pure buildSubagentSteer/steerDelivered/isSubagentFailure - tools/jobs.ts: jobTools(ctx) factory — server-side execute creates the job (deduped by toolCallId), fires spawn with the dispatch-time input (the part isn't persisted yet at execute — a transcript scan would race) - runner.ts: runJob(jobId, input?) builds the steer and hands it to the injected deliver seam; pipeline owns the done/error flip (a polling client re-syncs once, at the end); failures still steer - server/jobs.ts (new pipeline): per-chat serialized queues on globalThis, idle gates (turn liveness + pending question, 1.5s grace), 409 retry with backoff, idempotent per jobId, stale-job recovery sweep (re-deliver or re-spawn); wake turns run through the same handleChatRequest composition via setTurnRunner injection - chat-handler.ts: wires runner+deliver+tools ctx; drops spawnTurnJobs from onFinish; teacher default registry = question tools only - plan.ts: PlanToolContext split out (plan tools need no job wiring) - visual-maker: edit_svg added (pi parity — exact-match single replacement, fuzzy context on a miss); researcher: pi's kept/dropped source accounting restored - prompts 20-jobs/40-demos: async discipline rewritten (announce and keep teaching, no waiting/polling/fabricating, parallel dispatch, teach with the artifact when it arrives) - tests: jobs.test.ts rewritten (dispatch ack mid-turn, runner steers, wake pipeline: deferral/409/idempotency/failure), demos.test.tsx runner tests ported to runJob + fake deliver; 222 green FLAGGED surgical touches outside the stream's owned files, for master review at merge: src/core/contracts.ts (protocol doc + steer types), src/core/ui-message.ts (job tool + data-part types), src/server/ chat-handler.ts (wiring), src/server/turn-service.ts (spawnTurnJobs retirement), src/agent/teacher-agent.ts (default tools), src/agent/tools/ plan.ts (context type split). Verified live on PORT=3003 with real models: teacher dispatched research mid-turn, researcher ran real searches/fetches, steer injected on quiz answer, wake turn wove the brief in citing its sources, job flipped done; a client/wake race correctly 409'd. --- src/agent/prompts/20-jobs.md | 22 +- src/agent/prompts/40-demos.md | 28 +- src/agent/subagents/researcher.ts | 3 +- src/agent/subagents/runner.ts | 326 ++++++++++---------- src/agent/subagents/visual-maker.ts | 74 ++++- src/agent/teacher-agent.ts | 14 +- src/agent/tools.ts | 33 +- src/agent/tools/jobs.ts | 125 +++++--- src/agent/tools/plan.ts | 13 +- src/core/contracts.ts | 162 ++++++++-- src/core/ui-message.ts | 23 +- src/server/chat-handler.ts | 43 ++- src/server/jobs.ts | 307 ++++++++++++++++++- src/server/turn-service.ts | 15 - tests/demos.test.tsx | 172 ++++++++--- tests/jobs.test.ts | 451 +++++++++++++++++++++++----- 16 files changed, 1382 insertions(+), 429 deletions(-) diff --git a/src/agent/prompts/20-jobs.md b/src/agent/prompts/20-jobs.md index cf0995b..26d47f7 100644 --- a/src/agent/prompts/20-jobs.md +++ b/src/agent/prompts/20-jobs.md @@ -9,19 +9,21 @@ outsource what you cannot reliably know yourself: - **Never use it for:** anything you know cold, internal reasoning, or math. Do not offload thinking — only facts you genuinely need verified. -**The call is asynchronous and your turn ends at it.** The discipline: +**The call is fire-and-forget: it resolves immediately with an ack and your turn +continues.** The discipline: -1. **Say your piece first.** Explain the concept, set up what you're doing and why, and tell - the learner roughly what you're checking — the research brief arrives later as the tool's - output, and the lesson resumes from there. A bare tool call with no surrounding text is a - failed turn. +1. **Announce it, then keep teaching.** Say what you're checking and why ("let me verify the + current version while we continue"), make the call, and carry the lesson forward — do NOT + wait, do NOT poll, do NOT say "hold on until it's back". Never fabricate the result. 2. **Call with a single, specific question.** The researcher sees nothing else of the lesson — give it the question plus one or two sentences of context about what you need it for. -3. **Resume from the brief.** When the result returns, weave it into the lesson. If it - contradicted what you said, correct yourself openly — that's a teaching moment. +3. **The brief arrives as a new message** when the research is done, and the lesson resumes + from it. Weave it into the lesson then. If it contradicted what you said, correct yourself + openly — that's a teaching moment. 4. **Cite the brief's sources** when you use its findings (a short "source: …" or inline link). Never present researched facts as your own certainty. -At most one research call per turn — don't fan out; the researcher searches multiple angles on -its own. If research fails or is unconfigured, you'll get a failure brief: say so plainly and -continue with what you can establish, flagging the uncertainty. +Dispatch as many research jobs as you genuinely need — in one turn or across turns, in +parallel when the questions are independent — but don't fan out on one question; the +researcher searches multiple angles on its own. If research fails, you'll get a failure +message: say so plainly and continue with what you can establish, flagging the uncertainty. diff --git a/src/agent/prompts/40-demos.md b/src/agent/prompts/40-demos.md index fbaff5b..0dc50c5 100644 --- a/src/agent/prompts/40-demos.md +++ b/src/agent/prompts/40-demos.md @@ -3,8 +3,8 @@ If the concept is DYNAMIC — a parameter to sweep, a process to animate, a distribution to sample, a system to simulate — build the learner the instrument rather than describing its behavior in prose. A paragraph about how frequency response changes with damping teaches less than ten seconds of -dragging a slider. You have two artifact tools for this, both asynchronous (same discipline as -research): +dragging a slider. You have two artifact tools for this, both fire-and-forget (same async +discipline as research). ### `demo` — interactive teaching instruments @@ -31,15 +31,19 @@ for geometry mermaid's auto-layout cannot express. ### The async discipline -The call is asynchronous and **your turn ends at it**, exactly like research: +Both calls resolve immediately with an ack and your turn continues: -1. **Say your piece first.** Set up the concept, explain what the learner is about to see and what to - watch for — the artifact arrives later as the tool's output and the lesson resumes from there. A - bare tool call with no surrounding text is a failed turn. -2. **At most one demo or visual per turn.** -3. **When the artifact arrives, TEACH WITH IT.** Reference its controls and elements by name: "drag - the damping slider all the way left and watch the overshoot", "notice the vector at t=2s". An +1. **Announce it, then keep teaching.** Set up the concept and tell the learner what's coming + ("I'm having an interactive demo built for this — meanwhile, the idea itself…"), make the call, + and carry on — do NOT wait, do NOT poll, do NOT say "hold on until it's back". Never fabricate + the result. A bare tool call with no surrounding text is a failed turn. +2. **Dispatch freely.** Parallel jobs are encouraged when they're genuinely independent — a demo + and a figure for the same concept, several figures for a multi-step construction. Keep each + brief self-contained. +3. **The artifact arrives as a new message** when it's built, verified, and rendered in the + lesson — **TEACH WITH IT then.** Reference its controls and elements by name: "drag the + damping slider all the way left and watch the overshoot", "notice the vector at t=2s". An artifact you never point at is decoration. -4. **Failures arrive as notes** ("FAILED: …") with no artifact — say so plainly, fall back to - explaining in prose, and if the brief was the problem, re-call with a tighter one. Re-calling a - tool produces a new version of the artifact; use that to fix a demo that misbehaves in the field. +4. **Failures arrive as messages** with no artifact — say so plainly, fall back to explaining + in prose, and if the brief was the problem, re-call with a tighter one. Re-calling a tool + produces a new version of the artifact; use that to fix a demo that misbehaves in the field. diff --git a/src/agent/subagents/researcher.ts b/src/agent/subagents/researcher.ts index 9eeb1d2..21e64b7 100644 --- a/src/agent/subagents/researcher.ts +++ b/src/agent/subagents/researcher.ts @@ -51,7 +51,8 @@ Numbered findings with inline source citations: 1. **Finding** — explanation. [Source](url) ## Sources -- Source Title (url) — why it's relevant +- Kept: Source Title (url) — why relevant +- Dropped: Source Title — why excluded ## Gaps What couldn't be answered. Suggested next steps.`; diff --git a/src/agent/subagents/runner.ts b/src/agent/subagents/runner.ts index ae9aa0a..924512c 100644 --- a/src/agent/subagents/runner.ts +++ b/src/agent/subagents/runner.ts @@ -1,107 +1,168 @@ import { isToolUIPart, type LanguageModel, type UIMessage } from "ai"; -import type { DemoResult, DemoToolInput, JobKind, ResearchResult, ResearchToolInput, VisualResult, VisualToolInput } from "../../core/contracts.ts"; -import { JOB_TOOLS } from "../../core/pending.ts"; +import { + buildSubagentSteer, + type DemoResult, + type DemoToolInput, + type JobKind, + type JobRecord, + type ResearchResult, + type ResearchToolInput, + type SubagentSteerMessage, + type VisualResult, + type VisualToolInput, +} from "../../core/contracts.ts"; import type { ChatStore, JobStore } from "../../core/ports.ts"; import { runDemoBuilder } from "./demo-builder.ts"; import { createResearcherAgent } from "./researcher.ts"; import { runVisualMaker } from "./visual-maker.ts"; /** - * The background job runner (contracts.ts "THE ASYNC PROTOCOL"), created via - * createJobRunner({ chatStore, jobStore }) — the composition root (chat-handler) - * injects the storage ports; this module never imports an adapter. + * The background job runner (contracts.ts "THE ASYNC PROTOCOL" §2–3), created + * via createJobRunner({ chatStore, jobStore, deliver }) — the composition root + * (chat-handler) injects the storage ports and the WAKE SEAM (the server-side + * steer delivery, src/server/jobs.ts); this module never imports an adapter. * - * spawnPendingJobs — called from the turn service AFTER persist: finds - * input-available JOB_TOOLS parts in the last assistant message with no - * existing job and fires the kind's runner WITHOUT awaiting (the HTTP - * response is long gone). Kind dispatch (research/demo/visual) is a - * self-contained map here, so no caller changes for new job kinds — - * anything listed in JOB_TOOLS routes automatically. + * runJob(jobId) — runs the kind's sub-agent loop in the background + * (the job tools' execute fires this via the spawn seam, fire-and-forget), + * updating the job's activity line on every iteration. On completion the + * runner builds the STEER MESSAGE (buildSubagentSteer) and hands it to + * `deliver`; the wake pipeline fires the server-side teacher turn and + * flips the job to done/error when the lesson has actually resumed. * - * runResearchJob / runDemoJob / runVisualJob — run the sub-agent loop in - * the background, updating the job's activity line on every iteration, - * then rewrite the tool part in the LATEST chat record to - * output-available (read-modify-write; no awaits between the read and - * the write). The part is fulfilled BEFORE the job status flips, so a - * client that polls the job and then re-syncs always sees the finished - * part. - * - * On failure the part is still fulfilled (DemoResult/VisualResult failure - * shape: empty html/svg + "FAILED: …" notes; research: failure brief) — the - * teacher must learn the outcome — and the job is marked error. + * On failure the result is the failure shape (empty html/svg + "FAILED: …" + * notes; research: failure brief) — and it STILL steers, because the + * teacher must learn the outcome. If delivery itself fails, the job is + * marked error (the learner's activity card shows it). */ /** - * Test seam: the spawn hook creates runners with no injection point, so - * tests swap in a MockLanguageModelV4 here to keep job tests off the - * network. Undefined in production. + * Test seam: the spawn wiring has no injection point at the tool layer, so + * tests swap in a MockLanguageModelV4 here to keep job tests off the network. + * Undefined in production. (The jobs spawn through the module-level KIND + * runners, which read these at call time.) */ export const researchModelOverride: { current?: LanguageModel } = {}; export const demoModelOverride: { current?: LanguageModel } = {}; export const visualModelOverride: { current?: LanguageModel } = {}; +/** The dispatch-time tool input (from src/agent/tools/jobs.ts schemas). */ +type JobToolInput = ResearchToolInput | DemoToolInput | VisualToolInput; + export interface JobRunnerDeps { chatStore: ChatStore; jobStore: JobStore; -} - -export interface RunResearchJobOptions { - model?: LanguageModel; -} - -export interface RunDemoJobOptions { - model?: LanguageModel; -} - -export interface RunVisualJobOptions { - model?: LanguageModel; + /** + * Hand a finished job's steer to the wake pipeline — resolves when the + * steer is delivered (the wake turn completed and the job status + * flipped); rejects when delivery fails unrecoverably. + */ + deliver: (steer: SubagentSteerMessage) => Promise; } export interface JobRunner { - spawnPendingJobs(chatId: string, messages: UIMessage[]): void; - runResearchJob(jobId: string, opts?: RunResearchJobOptions): Promise; - runDemoJob(jobId: string, opts?: RunDemoJobOptions): Promise; - runVisualJob(jobId: string, opts?: RunVisualJobOptions): Promise; + /** Run one job's sub-agent loop to completion (background; never throws). + * The tool input rides along from dispatch (the spawn fires mid-stream, + * before the tool part is persisted — only crash-recovery re-spawns + * resolve the input from the transcript). */ + runJob(jobId: string, input?: ResearchToolInput | DemoToolInput | VisualToolInput): Promise; } export function createJobRunner(deps: JobRunnerDeps): JobRunner { - const { chatStore, jobStore } = deps; + const { chatStore, jobStore, deliver } = deps; - /** The kind→runner map: everything in JOB_TOOLS routes through here. */ - const KIND_RUNNERS: Record Promise> = { - research: (jobId) => runResearchJob(jobId, { model: researchModelOverride.current }), - demo: (jobId) => runDemoJob(jobId, { model: demoModelOverride.current }), - visual: (jobId) => runVisualJob(jobId, { model: visualModelOverride.current }), + /** The kind→runner map: one sub-agent loop per job kind. */ + // Each kind's runner narrows the union input itself (dispatch-time inputs + // are kind-correct by construction; transcript fallbacks are cast after). + const KIND_RUNNERS: Record Promise> = { + research: (jobId, input) => runResearchJob(jobId, researchModelOverride.current, input), + demo: (jobId, input) => runDemoJob(jobId, demoModelOverride.current, input), + visual: (jobId, input) => runVisualJob(jobId, visualModelOverride.current, input), }; - function spawnPendingJobs(chatId: string, messages: UIMessage[]): void { - const last = [...messages].reverse().find((m) => m.role === "assistant"); - if (!last) return; - for (const part of last.parts) { - if (!isToolUIPart(part) || part.state !== "input-available") continue; - const tool = part.type.slice("tool-".length); - if (!(JOB_TOOLS as readonly string[]).includes(tool)) continue; - const runner = KIND_RUNNERS[tool as JobKind]; - if (!runner) continue; - if (jobStore.findJobByToolCallId(chatId, part.toolCallId)) continue; // already has a job - const job = jobStore.createJob({ kind: tool as JobKind, chatId, toolCallId: part.toolCallId }); - void runner(job.id).catch((err) => { - console.error(`[job ${job.id}] ${tool} runner crashed:`, err); - jobStore.updateJob(job.id, { status: "error", error: String(err), activity: "failed" }); - }); + async function runJob(jobId: string, input?: JobToolInput): Promise { + const job = jobStore.getJob(jobId); + if (!job) { + console.error(`[job ${jobId}] not found — nothing to run`); + return; + } + if (job.status !== "running") return; // already finished / re-run after crash recovery + const run = KIND_RUNNERS[job.kind]; + if (!run) { + console.error(`[job ${jobId}] unknown kind: ${job.kind}`); + jobStore.updateJob(jobId, { status: "error", error: `unknown kind ${job.kind}`, activity: "failed" }); + return; + } + await run(jobId, input); + } + + /** Resolve the job's tool input: dispatch passes it directly; only a + * crash-recovery re-spawn (no input) falls back to scanning the + * transcript (by then onFinish has long persisted the part). */ + function resolveInput( + job: JobRecord, + input: JobToolInput | undefined, + ): ResearchToolInput | DemoToolInput | VisualToolInput | undefined { + if (input) return input; + const chat = chatStore.getChat(job.chatId); + if (!chat) return undefined; + return (findToolPart(chat.messages, job.toolCallId)?.input ?? undefined) as + | ResearchToolInput + | DemoToolInput + | VisualToolInput + | undefined; + } + + // ─── shared plumbing ──────────────────────────────────────────────────── + + function findToolPart(messages: UIMessage[], toolCallId: string) { + for (const m of messages) { + for (const part of m.parts) { + if (isToolUIPart(part) && part.toolCallId === toolCallId) return part; + } + } + return null; + } + + /** + * The tail of every job: store the result, build the steer, hand it to + * the wake pipeline. The pipeline flips the status when the lesson has + * actually resumed — until then the job reads "running" so the client's + * activity card keeps polling and re-syncs once, at the end. + */ + async function finishJob(jobId: string, result: ResearchResult | DemoResult | VisualResult): Promise { + const job = jobStore.getJob(jobId); + if (!job) return; + jobStore.updateJob(jobId, { result, activity: "delivering result" }); + const steer = buildSubagentSteer(job, result); + try { + await deliver(steer); + } catch (err) { + const reason = err instanceof Error ? err.message : String(err); + console.error(`[job ${jobId}] steer delivery failed:`, err); + jobStore.updateJob(jobId, { status: "error", error: reason, activity: "delivery failed" }); } } - async function runResearchJob(jobId: string, opts: RunResearchJobOptions = {}): Promise { + /** Chat-record + input resolution shared by all three runners. */ + function jobContext(jobId: string): { job: JobRecord; chatId: string } | null { const job = jobStore.getJob(jobId); - if (!job || job.status !== "running") return; - + if (!job || job.status !== "running") return null; const chat = chatStore.getChat(job.chatId); if (!chat) { jobStore.updateJob(jobId, { status: "error", error: `Chat not found: ${job.chatId}`, activity: "failed" }); - return; + return null; } - const input = findToolPart(chat.messages, job.toolCallId)?.input as ResearchToolInput | undefined; + return { job, chatId: job.chatId }; + } + + // ─── research ─────────────────────────────────────────────────────────── + + async function runResearchJob(jobId: string, model?: LanguageModel, passedInput?: JobToolInput): Promise { + const ctx = jobContext(jobId); + if (!ctx) return; + const { job } = ctx; + + const input = (passedInput ?? resolveInput(job, passedInput)) as ResearchToolInput | undefined; const question = typeof input?.question === "string" && input.question.trim() ? input.question : "(no question)"; const context = typeof input?.context === "string" && input.context.trim() ? input.context : undefined; @@ -114,7 +175,7 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner { .join("\n\n"); const fetchedUrls: string[] = []; - const agent = createResearcherAgent(opts.model); + const agent = createResearcherAgent(model); try { jobStore.updateJob(jobId, { activity: "researching" }); @@ -132,37 +193,22 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner { const brief = result.text.trim(); if (!brief) throw new Error("the researcher produced no text"); const research: ResearchResult = { brief, sources: parseSources(brief, fetchedUrls) }; - - fulfillPart(job.chatId, job.toolCallId, research); - jobStore.updateJob(jobId, { status: "done", result: research }); // activity line kept: the last tool call is the record + await finishJob(jobId, research); } catch (err) { const reason = err instanceof Error ? err.message : String(err); - fulfillPart(job.chatId, job.toolCallId, { brief: `Research failed: ${reason}`, sources: [] }); - jobStore.updateJob(jobId, { status: "error", activity: "failed", error: reason }); + jobStore.updateJob(jobId, { error: reason }); + await finishJob(jobId, { brief: `Research failed: ${reason}`, sources: [] }); } } - // ─── demo + visual job wrappers ─────────────────────────────────────────── + // ─── demo + visual ────────────────────────────────────────────────────── - /** Failure shape (W6 brief): html === "" with "FAILED: …" notes. */ - function isDemoFailure(result: DemoResult): boolean { - return typeof result.html !== "string" || result.html.length === 0; - } + async function runDemoJob(jobId: string, model?: LanguageModel, passedInput?: JobToolInput): Promise { + const ctx = jobContext(jobId); + if (!ctx) return; + const { job } = ctx; - function isVisualFailure(result: VisualResult): boolean { - return typeof result.svg !== "string" || result.svg.length === 0; - } - - async function runDemoJob(jobId: string, opts: RunDemoJobOptions = {}): Promise { - const job = jobStore.getJob(jobId); - if (!job || job.status !== "running") return; - - const chat = chatStore.getChat(job.chatId); - if (!chat) { - jobStore.updateJob(jobId, { status: "error", error: `Chat not found: ${job.chatId}`, activity: "failed" }); - return; - } - const input = findToolPart(chat.messages, job.toolCallId)?.input as DemoToolInput | undefined; + const input = (passedInput ?? resolveInput(job, passedInput)) as DemoToolInput | undefined; const title = typeof input?.title === "string" && input.title.trim() ? input.title : "Interactive demo"; const brief = typeof input?.brief === "string" ? input.brief : ""; @@ -171,34 +217,25 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner { const result = await runDemoBuilder( { title, brief }, { - model: opts.model, + model, onActivity: (line) => jobStore.updateJob(jobId, { activity: line }), }, ); - fulfillPart(job.chatId, job.toolCallId, result); - if (isDemoFailure(result)) { - jobStore.updateJob(jobId, { status: "error", error: result.notes ?? "demo build failed", activity: "failed" }); - } else { - jobStore.updateJob(jobId, { status: "done", result }); - } + if (isDemoFailure(result)) jobStore.updateJob(jobId, { error: result.notes ?? "demo build failed" }); + await finishJob(jobId, result); } catch (err) { const reason = err instanceof Error ? err.message : String(err); - const failure: DemoResult = { title, html: "", tested: true, notes: `FAILED: ${reason}` }; - fulfillPart(job.chatId, job.toolCallId, failure); - jobStore.updateJob(jobId, { status: "error", error: reason, activity: "failed" }); + jobStore.updateJob(jobId, { error: reason }); + await finishJob(jobId, { title, html: "", tested: true, notes: `FAILED: ${reason}` }); } } - async function runVisualJob(jobId: string, opts: RunVisualJobOptions = {}): Promise { - const job = jobStore.getJob(jobId); - if (!job || job.status !== "running") return; + async function runVisualJob(jobId: string, model?: LanguageModel, passedInput?: JobToolInput): Promise { + const ctx = jobContext(jobId); + if (!ctx) return; + const { job } = ctx; - const chat = chatStore.getChat(job.chatId); - if (!chat) { - jobStore.updateJob(jobId, { status: "error", error: `Chat not found: ${job.chatId}`, activity: "failed" }); - return; - } - const input = findToolPart(chat.messages, job.toolCallId)?.input as VisualToolInput | undefined; + const input = (passedInput ?? resolveInput(job, passedInput)) as VisualToolInput | undefined; const brief = typeof input?.brief === "string" ? input.brief : ""; try { @@ -206,70 +243,33 @@ export function createJobRunner(deps: JobRunnerDeps): JobRunner { const result = await runVisualMaker( { brief }, { - model: opts.model, + model, onActivity: (line) => jobStore.updateJob(jobId, { activity: line }), }, ); - fulfillPart(job.chatId, job.toolCallId, result); - if (isVisualFailure(result)) { - jobStore.updateJob(jobId, { status: "error", error: result.notes ?? "visual build failed", activity: "failed" }); - } else { - jobStore.updateJob(jobId, { status: "done", result }); - } + if (isVisualFailure(result)) jobStore.updateJob(jobId, { error: result.notes ?? "visual build failed" }); + await finishJob(jobId, result); } catch (err) { const reason = err instanceof Error ? err.message : String(err); - const failure: VisualResult = { svg: "", notes: `FAILED: ${reason}` }; - fulfillPart(job.chatId, job.toolCallId, failure); - jobStore.updateJob(jobId, { status: "error", error: reason, activity: "failed" }); + jobStore.updateJob(jobId, { error: reason }); + await finishJob(jobId, { svg: "", notes: `FAILED: ${reason}` }); } } - // ─── helpers ───────────────────────────────────────────────────────────── - - function findToolPart(messages: UIMessage[], toolCallId: string) { - for (const m of messages) { - for (const part of m.parts) { - if (isToolUIPart(part) && part.toolCallId === toolCallId) return part; - } - } - return null; - } - - /** - * Rewrite the tool part in the LATEST chat record to output-available. - * Synchronous read-modify-write (no awaits) — atomic within the event loop. - */ - function fulfillPart(chatId: string, toolCallId: string, output: ResearchResult | DemoResult | VisualResult): void { - const chat = chatStore.getChat(chatId); - if (!chat) return; - let found = false; - const messages = chat.messages.map((m) => - m.role !== "assistant" - ? m - : { - ...m, - parts: m.parts.map((part) => { - if (isToolUIPart(part) && part.toolCallId === toolCallId && part.state === "input-available") { - found = true; - return { ...part, state: "output-available" as const, output }; - } - return part; - }), - }, - ); - if (!found) return; - try { - chatStore.saveMessages(chatId, messages); - } catch (err) { - console.error(`[chat ${chatId}] fulfilling job part ${toolCallId} failed:`, err); - } - } - - return { spawnPendingJobs, runResearchJob, runDemoJob, runVisualJob }; + return { runJob }; } // ─── module-level pure helpers ────────────────────────────────────────────── +/** Failure shape (W6 brief): html === "" with "FAILED: …" notes. */ +function isDemoFailure(result: DemoResult): boolean { + return typeof result.html !== "string" || result.html.length === 0; +} + +function isVisualFailure(result: VisualResult): boolean { + return typeof result.svg !== "string" || result.svg.length === 0; +} + /** The activity line shown in the UI: `web_search: "query"` / `web_fetch: host`. */ function activityLine(toolName: string, input: { query?: string; url?: string } | undefined): string { if (toolName === "web_search") return `web_search: "${input?.query ?? ""}"`; diff --git a/src/agent/subagents/visual-maker.ts b/src/agent/subagents/visual-maker.ts index d5058c8..df0a798 100644 --- a/src/agent/subagents/visual-maker.ts +++ b/src/agent/subagents/visual-maker.ts @@ -49,8 +49,11 @@ composition, and above everything CORRECTNESS: the picture must not assert anyth mark on the wrong corner, a vector pointing the wrong way, a point plotted at the wrong coordinate is a failure even if it renders cleanly. -Your two tools: +Your three tools: - write_svg({ svg }) — write the complete … source (overwrites the previous version). +- edit_svg({ old_text, new_text }) — a surgical exact-match replacement in the current SVG: + old_text must appear EXACTLY ONCE; it is replaced by new_text. Call write_svg first. For anything + beyond a one-spot fix (new elements, restructured layout), write_svg the whole document again. - render_svg({}) — rasterize the current SVG in headless Chrome. The render arrives as an image in the message right after the tool result. @@ -64,8 +67,8 @@ says nothing about whether the geometry is right or the picture is readable. Whe - Would a learner instantly read the intended idea from this picture alone? Workflow: plan the coordinate space (viewBox + margins, nothing touching the edges) → write_svg → -render_svg → LOOK critically → fix (write_svg) and re-render until correct and clean. Never finish -without having looked at a render of your FINAL SVG. +render_svg → LOOK critically → fix (edit_svg for one-spot fixes, write_svg otherwise) and re-render +until correct and clean. Never finish without having looked at a render of your FINAL SVG. SVG requirements: - A complete with explicit width/height AND a matching viewBox. @@ -86,6 +89,14 @@ export function createVisualTools() { svg: z.string().describe('Complete … markup.'), }), }), + edit_svg: tool({ + description: + "Replace one exact spot in the current SVG: old_text must appear EXACTLY ONCE and is replaced by new_text. Call write_svg first. For larger changes, write_svg the whole document.", + inputSchema: z.object({ + old_text: z.string().describe("The exact text to replace — must match exactly once in the current SVG."), + new_text: z.string().describe("The replacement text."), + }), + }), render_svg: tool({ description: "Render the current SVG to a PNG and show it to you for inspection. No arguments.", inputSchema: z.object({}), @@ -164,6 +175,36 @@ export async function runVisualMaker(input: VisualMakerInput, opts: VisualMakerO currentSvg = svg; opts.onActivity?.(`write_svg: ${(svg.length / 1024).toFixed(1)} KB`); toolResults.push(toolText(call.toolCallId, call.toolName, `Saved (${svg.length} bytes). Call render_svg to look at it.`)); + } else if (call.toolName === "edit_svg") { + const edit = call.input as { old_text?: unknown; new_text?: unknown }; + if (!currentSvg) { + toolResults.push(toolText(call.toolCallId, call.toolName, "ERROR: nothing to edit — call write_svg first.")); + continue; + } + if (typeof edit.old_text !== "string" || typeof edit.new_text !== "string" || !edit.old_text) { + toolResults.push(toolText(call.toolCallId, call.toolName, "ERROR: expected old_text (non-empty) and new_text strings.")); + continue; + } + const occurrences = currentSvg.split(edit.old_text).length - 1; + if (occurrences === 0) { + toolResults.push( + toolText( + call.toolCallId, + call.toolName, + `ERROR: old_text not found. Closest line for reference:\n${snippetAround(currentSvg, edit.old_text)}`, + ), + ); + continue; + } + if (occurrences > 1) { + toolResults.push( + toolText(call.toolCallId, call.toolName, `ERROR: old_text appears ${occurrences} times — it must match EXACTLY ONCE. Include more surrounding context to make it unique.`), + ); + continue; + } + currentSvg = currentSvg.replace(edit.old_text, edit.new_text); + opts.onActivity?.("edit_svg: applied"); + toolResults.push(toolText(call.toolCallId, call.toolName, "OK — replaced 1 occurrence. Call render_svg to look at the edited SVG.")); } else if (call.toolName === "render_svg") { if (!currentSvg) { toolResults.push(toolText(call.toolCallId, call.toolName, "ERROR: nothing to render — call write_svg first.")); @@ -181,7 +222,7 @@ export async function runVisualMaker(input: VisualMakerInput, opts: VisualMakerO toolResults.push(toolText(call.toolCallId, call.toolName, `RENDER ERROR (the SVG may be malformed): ${msg}. Fix it with write_svg and render again.`)); } } else { - toolResults.push(toolText(call.toolCallId, call.toolName, `ERROR: unknown tool "${call.toolName}". Use write_svg or render_svg.`)); + toolResults.push(toolText(call.toolCallId, call.toolName, `ERROR: unknown tool "${call.toolName}". Use write_svg, edit_svg, or render_svg.`)); } } // Message order: assistant tool-calls → tool results → render image. @@ -215,6 +256,31 @@ function toolText(toolCallId: string, toolName: string, value: string): ToolResu return { type: "tool-result", toolCallId, toolName, output: { type: "text", value } }; } +/** A 3-line numbered context window around the closest fuzzy match for a + * failed edit — enough for the model to see what went wrong and retry + * (the pi svg_tools snippetAround semantics). */ +function snippetAround(svg: string, target: string): string { + const words = target.trim().split(/\s+/).filter(Boolean); + if (words.length === 0) return "(empty old_text)"; + const lines = svg.split("\n"); + let bestAt = -1; + let bestScore = -1; + for (let i = 0; i < lines.length; i++) { + const score = words.filter((w) => lines[i]!.includes(w)).length; + if (score > bestScore) { + bestScore = score; + bestAt = i; + } + } + if (bestAt < 0) return "(no similar line found)"; + const from = Math.max(0, bestAt - 1); + const to = Math.min(lines.length, bestAt + 2); + return lines + .slice(from, to) + .map((line, idx) => `${from + idx + 1} | ${line}`) + .join("\n"); +} + function finish(svg: string | null, finalText: string): VisualResult { if (!svg) { return { diff --git a/src/agent/teacher-agent.ts b/src/agent/teacher-agent.ts index f251005..f31beeb 100644 --- a/src/agent/teacher-agent.ts +++ b/src/agent/teacher-agent.ts @@ -2,8 +2,8 @@ import * as fs from "node:fs"; import * as path from "node:path"; import { ToolLoopAgent, stepCountIs, type LanguageModel, type ToolSet } from "ai"; import { PATHS } from "../../config.ts"; -import { createTeacherTools, jobTools, questionTools } from "./tools.ts"; import { modelFor } from "./models.ts"; +import { questionTools } from "./tools.ts"; /** * The teacher agent. Everything pi used to provide (agent loop, tool calling, @@ -56,13 +56,15 @@ export function createTeacherAgent(model?: LanguageModel, tools?: ToolSet, graph // Reasoning tokens stream as reasoning parts into the UI's collapsible block. model: model ?? modelFor("teacher"), instructions: withGraphContext(graphContext), - // Tools are per-request (plan tools close over the injected ctx); - // chat-handler passes createTeacherTools({ chatId, chatStore }). The + // Tools are per-request (plan tools close over the injected ctx; job + // tools close over the job store + spawn seam) — chat-handler passes + // createTeacherTools({ chatId, chatStore, jobStore, spawn }). The // dependency-free default is the question-only set (client-side tools, // no storage needed) — used when tests pass a bare mock model. - tools: tools ?? { ...questionTools, ...jobTools() }, - // Question and job tools are client-side — a call ends the server turn - // automatically. stepCountIs is just the safety net. + tools: tools ?? { ...questionTools }, + // Question tools are client-side (a call ends the server turn); + // job tools resolve immediately (fire-and-forget ack). stepCountIs is + // just the safety net. stopWhen: stepCountIs(20), }); } diff --git a/src/agent/tools.ts b/src/agent/tools.ts index 82d2aef..2c207af 100644 --- a/src/agent/tools.ts +++ b/src/agent/tools.ts @@ -1,10 +1,11 @@ import { tool, type ToolSet } from "ai"; import { askInputSchema, quizInputSchema } from "../core/question-core.ts"; -import type { ChatStore } from "../core/ports.ts"; -import { jobTools } from "./tools/jobs.ts"; +import type { JobRecord } from "../core/contracts.ts"; +import type { JobStore } from "../core/ports.ts"; +import { jobTools, type DemoInput, type ResearchInput, type VisualInput } from "./tools/jobs.ts"; +import { planTools, type PlanToolContext } from "./tools/plan.ts"; export { jobTools }; -import { planTools } from "./tools/plan.ts"; /** * The teacher's tools, composed per request by createTeacherTools. @@ -17,8 +18,10 @@ import { planTools } from "./tools/plan.ts"; * the prompt docs exactly; keep them in sync. * - planTools (src/agent/tools/plan.ts): SERVER-SIDE, mutate the chat's plan * tree (needs the chatId closure — hence the per-request factory). - * - jobTools (src/agent/tools/jobs.ts): CLIENT-SIDE semantics, server-side - * fulfillment by background sub-agents (see src/core/contracts.ts). + * - jobTools (src/agent/tools/jobs.ts): SERVER-EXECUTED dispatch — the call + * resolves immediately with a fire-and-forget ack, the background runner + * does the work, and the result arrives later as a steer message + * (contracts.ts "THE ASYNC PROTOCOL"). */ export const quizTool = tool({ @@ -51,23 +54,25 @@ export const questionTools = { ask_user_question: askUserQuestionTool, }; -export interface TeacherToolContext { - chatId: string; - /** The storage port plan tools persist through — the composition root - * (chat-handler) injects the wired adapter; the agent layer never - * imports one. */ - chatStore: ChatStore; +export interface TeacherToolContext extends PlanToolContext { + /** Job-tool wiring (sub-agent dispatch): the job store + the runner's + * spawn seam. The composition root injects both. */ + jobStore: JobStore; + /** Fire the background runner for a freshly created job. The tool input + * rides along — the spawn fires mid-stream, before the tool part is + * persisted to the transcript (a transcript scan would race it). */ + spawn: (job: JobRecord, input: ResearchInput | DemoInput | VisualInput) => void; } /** * Per-request tool registry. Plan tools close over the ctx (chatId + the - * injected ChatStore) for persistence; question and job tools are - * context-free. + * injected ChatStore) for persistence; question tools are context-free; + * job tools close over the job store + spawn seam. */ export function createTeacherTools(ctx: TeacherToolContext): ToolSet { return { ...questionTools, ...planTools(ctx), - ...jobTools(), + ...jobTools({ chatId: ctx.chatId, jobStore: ctx.jobStore, spawn: ctx.spawn }), }; } diff --git a/src/agent/tools/jobs.ts b/src/agent/tools/jobs.ts index 78d3d20..6ae3407 100644 --- a/src/agent/tools/jobs.ts +++ b/src/agent/tools/jobs.ts @@ -1,16 +1,29 @@ import { tool, type ToolSet } from "ai"; import { z } from "zod"; +import type { JobRecord } from "../../core/contracts.ts"; +import type { JobStore } from "../../core/ports.ts"; /** - * Async sub-agent job tools (W2: research; W6 adds demo + visual). - * CLIENT-SIDE semantics (no execute) — exactly like quiz/ask: the tool call - * streams to the browser in `input-available` state and the server turn ends - * there. chat-handler's onFinish spawns a background runner that fulfills - * the part server-side; the client polls /api/jobs and the continuation - * resumes the teacher with the result as a normal tool output. - * See src/core/contracts.ts "THE ASYNC PROTOCOL". + * Async sub-agent job tools (vision §2.5; W2/W6 kinds + the pi-model + * protocol). SERVER-EXECUTED: each call creates the job record, fires the + * background runner (via the injected spawn seam — the agent layer never + * imports one), and resolves IMMEDIATELY with a fire-and-forget ack. The + * teacher's turn does NOT end at the call; the result arrives later as a + * steer message that wakes the teacher in a new turn. See contracts.ts + * "THE ASYNC PROTOCOL". */ +/** What the job tools need at execute time (injected by the composition root). */ +export interface JobToolContext { + chatId: string; + /** Creates the job record (deduped by toolCallId). */ + jobStore: JobStore; + /** Fire the background runner for a fresh job — fire-and-forget. The + * input rides along because the spawn happens MID-STREAM, before + * onFinish persists the tool part — a transcript scan would race it. */ + spawn: (job: JobRecord, input: ResearchInput | DemoInput | VisualInput) => void; +} + export const researchInputSchema = z.object({ question: z .string() @@ -21,15 +34,6 @@ export const researchInputSchema = z.object({ .describe("Why you need this / what the lesson needs it for — focuses the brief. One or two sentences."), }); -export const researchTool = tool({ - description: - "Dispatch a background web-research agent: fact verification, current data, or niche specifics you are not certain about. " + - "The call is ASYNC — your turn ENDS here, so say everything you want to say first; the research brief arrives later as " + - "this tool's output and the lesson resumes from it. Cite the brief's sources when you use them. " + - "NEVER use it for things you know cold, internal reasoning, or math.", - inputSchema: researchInputSchema, -}); - export const demoInputSchema = z.object({ title: z.string().describe("Short title for the demo card, e.g. 'Damped harmonic oscillator'."), brief: z @@ -40,16 +44,6 @@ export const demoInputSchema = z.object({ ), }); -export const demoTool = tool({ - description: - "Dispatch a background sub-agent that BUILDS and headlessly TESTS an interactive teaching instrument — a single-file HTML " + - "demo (parameter sweep, simulation, animation, sampling) rendered live in the lesson. Use it when the concept is DYNAMIC: " + - "something to sweep, animate, or sample is better experienced than described. " + - "The call is ASYNC — your turn ENDS here, so say everything you want to say first; the tested demo arrives later as this " + - "tool's output and the lesson resumes with it. Write a precise brief: concept, controls, interactions, style.", - inputSchema: demoInputSchema, -}); - export const visualInputSchema = z.object({ brief: z .string() @@ -59,26 +53,73 @@ export const visualInputSchema = z.object({ ), }); -export const visualTool = tool({ - description: - "Dispatch a background sub-agent that hand-authors ONE precise SVG figure, renders it, LOOKS at the render, and iterates " + - "until correct. Use it for exact spatial pictures auto-layout diagrams can't do: coordinate geometry, number lines, vectors, " + - "function plots, physical layouts. For flows/relations/taxonomies write mermaid inline instead. " + - "The call is ASYNC — your turn ENDS here, so say everything you want to say first; the verified figure arrives later as this " + - "tool's output and the lesson resumes with it.", - inputSchema: visualInputSchema, -}); +/** + * The per-call dispatch: create the job (idempotent by toolCallId), fire the + * runner, and return the ack. Shared by all three tools — only kind differs. + */ +function dispatch( + ctx: JobToolContext, + kind: JobRecord["kind"], + toolCallId: string, + ack: string, + input: ResearchInput | DemoInput | VisualInput, +): string { + const existing = ctx.jobStore.findJobByToolCallId(ctx.chatId, toolCallId); + const job = existing ?? ctx.jobStore.createJob({ kind, chatId: ctx.chatId, toolCallId }); + if (!existing) ctx.spawn(job, input); + return `Dispatched (job ${job.id}). ${ack}`; +} + +const RESEARCH_ACK = + "This call is fire-and-forget: continue your turn and keep teaching — do not wait, poll, or mention waiting. " + + "The researched brief arrives as a new message when ready and the lesson resumes from it; cite its sources when you use them."; +const DEMO_ACK = + "This call is fire-and-forget: continue your turn and keep teaching — do not wait, poll, or mention waiting. " + + "The tested demo arrives as a new message when ready and is rendered in the lesson; teach with it then."; +const VISUAL_ACK = + "This call is fire-and-forget: continue your turn and keep teaching — do not wait, poll, or mention waiting. " + + "The verified figure arrives as a new message when ready and is rendered in the lesson; reference it then."; + +const RESEARCH_DESCRIPTION = + "Dispatch a background web-research agent: fact verification, current data, or niche specifics you are not certain about. " + + "The call is ASYNC and fire-and-forget — it resolves immediately and your turn continues; the research brief arrives later " + + "as a new message and the lesson resumes from it. Cite the brief's sources when you use them. " + + "NEVER use it for things you know cold, internal reasoning, or math."; +const DEMO_DESCRIPTION = + "Dispatch a background sub-agent that BUILDS and headlessly TESTS an interactive teaching instrument — a single-file HTML " + + "demo (parameter sweep, simulation, animation, sampling) rendered live in the lesson. Use it when the concept is DYNAMIC: " + + "something to sweep, animate, or sample is better experienced than described. " + + "The call is ASYNC and fire-and-forget — it resolves immediately and your turn continues; the tested demo arrives later " + + "as a new message. Write a precise brief: concept, controls, interactions, style."; +const VISUAL_DESCRIPTION = + "Dispatch a background sub-agent that hand-authors ONE precise SVG figure, renders it, LOOKS at the render, and iterates " + + "until correct. Use it for exact spatial pictures auto-layout diagrams can't do: coordinate geometry, number lines, vectors, " + + "function plots, physical layouts. For flows/relations/taxonomies write mermaid inline instead. " + + "The call is ASYNC and fire-and-forget — it resolves immediately and your turn continues; the verified figure arrives later " + + "as a new message."; /** - * Per-request job tool registry. Context-free (the runner resolves - * everything from the chat record by toolCallId). Takes no dependencies — - * part of the dependency-free default registry in teacher-agent.ts. + * Per-request job tool registry: each tool closes over the composition root's + * context (chatId + jobStore + the spawn seam), so concurrent chats never + * share state. chat-handler wires it each turn; tests wire their own. */ -export function jobTools(): ToolSet { +export function jobTools(ctx: JobToolContext): ToolSet { return { - research: researchTool, - demo: demoTool, - visual: visualTool, + research: tool({ + description: RESEARCH_DESCRIPTION, + inputSchema: researchInputSchema, + execute: async (input, { toolCallId }) => dispatch(ctx, "research", toolCallId, RESEARCH_ACK, input), + }), + demo: tool({ + description: DEMO_DESCRIPTION, + inputSchema: demoInputSchema, + execute: async (input, { toolCallId }) => dispatch(ctx, "demo", toolCallId, DEMO_ACK, input), + }), + visual: tool({ + description: VISUAL_DESCRIPTION, + inputSchema: visualInputSchema, + execute: async (input, { toolCallId }) => dispatch(ctx, "visual", toolCallId, VISUAL_ACK, input), + }), }; } diff --git a/src/agent/tools/plan.ts b/src/agent/tools/plan.ts index 01b88f5..a25f84a 100644 --- a/src/agent/tools/plan.ts +++ b/src/agent/tools/plan.ts @@ -1,6 +1,7 @@ import { tool, type ToolSet } from "ai"; import { z } from "zod"; import type { PlanToolResult, PlanTree } from "../../core/contracts.ts"; +import type { ChatStore } from "../../core/ports.ts"; import { addEdge, addNode, @@ -13,7 +14,13 @@ import { setCurrent, updateNode, } from "../../core/plan-tree.ts"; -import type { TeacherToolContext } from "../tools.ts"; + +/** What the plan tools close over (the plan-tree half of the tool context). */ +export interface PlanToolContext { + chatId: string; + /** The storage port plan tools persist through. */ + chatStore: ChatStore; +} /** * Plan-tree tools — W1 workstream (docs/plan.md Wave 1; src/core/contracts.ts @@ -48,7 +55,7 @@ const edgeSchema = z.object({ * empty) tree, so the model can see what it's working with and fix the call. */ async function runPlanOp( - ctx: TeacherToolContext, + ctx: PlanToolContext, op: (tree: PlanTree) => PlanTree, ): Promise { const { chatId, chatStore } = ctx; @@ -65,7 +72,7 @@ async function runPlanOp( } } -export function planTools(ctx: TeacherToolContext): ToolSet { +export function planTools(ctx: PlanToolContext): ToolSet { return { plan_init: tool({ description: diff --git a/src/core/contracts.ts b/src/core/contracts.ts index 2f2a1ed..72105af 100644 --- a/src/core/contracts.ts +++ b/src/core/contracts.ts @@ -282,39 +282,56 @@ export interface JobRecord { activity: string; createdAt: string; updatedAt: string; - /** The tool output payload when done (ResearchResult | DemoResult | VisualResult). */ + /** The sub-agent's result (steer-artifact payload). */ result?: unknown; error?: string; } /** - * THE ASYNC PROTOCOL — "client-side-tool semantics, server-side fulfillment": + * THE ASYNC PROTOCOL — "ack now, steer later" (the pi sub-agents model, + * vision §2.5): * - * 1. Job tools (research / demo / visual) are CLIENT-SIDE tools (no execute), - * exactly like quiz/ask. The model calls one; the server turn ends with the - * tool part in `input-available` state. (The teacher is prompted to say its - * piece BEFORE calling, since its turn ends at the call.) - * 2. chat-handler's onFinish (the integration point) scans the new assistant - * message for job tool parts and spawns one background runner per part - * (src/agent/subagents/…, W2/W6). The job store (src/store/jobs.ts) is - * keyed by toolCallId. - * 3. On completion the runner REWRITES the part in the stored chat record to - * `output-available` with the job result as output (read-modify-write of - * the LATEST record — never a stale copy). - * 4. The client polls GET /api/jobs?chatId=… while any job part is pending; - * when a part flips, it re-syncs messages from the server - * (ChatActionsContext.syncFromServer) and the existing - * `sendAutomaticallyWhen` continuation fires — the teacher receives the - * result as a normal tool output and resumes. No new stream machinery. + * 1. Job tools (research / demo / visual) are SERVER-EXECUTED. The model + * calls one MID-TURN; execute creates the job record, fires the + * background runner, and resolves IMMEDIATELY with an acknowledgement + * ("dispatched — continue your turn"). The teacher's turn does NOT end + * at the call: it keeps teaching, may dispatch other independent jobs + * (full parallelism), and ends whenever it ends. The tool part's output + * is the ack, forever. * - * 5. The composer gate (findPendingToolCall) covers ONLY quiz/ask — but the - * protocol still forbids sending a user message while a job part is - * pending (an unanswered tool call in the outbound list breaks provider - * conversion). So job parts ALSO gate SENDING, with a distinct affordance - * ("waiting for …" + live activity) — the learner can keep reading, - * scrolling, and typing notes in the field; the send unlocks on arrival. - * True mid-job interjection (concurrent turns) is a documented later - * upgrade, not v1. + * 2. The background runner (src/agent/subagents/runner.ts) runs the kind's + * sub-agent loop — isolated context, modelFor(role), the + * verify-never-one-shot doctrine (headless-tested demos, looked-at SVGs, + * sourced briefs) — tracking a live activity line in the job record. + * The client polls GET /api/jobs?chatId=… and renders the activity card. + * + * 3. On completion the runner builds a STEER MESSAGE (buildSubagentSteer) + * and hands it to the wake pipeline (src/server/jobs.ts): a user-role + * UIMessage carrying the result — a text part with what the MODEL needs + * (the research brief, the builder's notes), plus a data-subagent-artifact + * part with what the UI renders (the full result: brief + sources / + * SVG / demo HTML). Provider conversion drops data parts, so artifact + * payloads never enter the teacher's context. metadata.subagentResult + * identifies the job; the UI renders these as sub-agent result cards, + * NEVER as learner interjections. + * + * 4. The wake pipeline delivers the steer by firing a SERVER-SIDE teacher + * turn through the same turn composition as a client request + * (handleChatRequest) — write-ahead persist, turn liveness, extraction + * trigger: one code path. A per-chat serialized queue: a steer waits + * while a turn is in flight OR a question tool part is pending (an + * unanswered client-side tool call breaks provider conversion — the + * steer fires right after the learner's continuation), then fires; a 409 + * (a client turn raced in) retries. Delivery is idempotent per jobId, so + * a crashed wake can be re-entered by the stale-job recovery sweep. The + * job's status flips to done/error only AFTER the wake turn completes — + * a polling client that sees the flip re-syncs once and gets the + * finished record. + * + * 5. No pending job parts ever exist, so the composer never gates on jobs + * (the learner can interject mid-job; the steer queues behind their + * turn) and job tool outputs never trigger the client-side + * sendAutomatically continuation — that mechanism stays quiz/ask-only. */ export interface ResearchToolInput { @@ -354,6 +371,99 @@ export interface VisualResult { notes?: string; } +// ─── Sub-agent steer messages (THE ASYNC PROTOCOL §3–4) ───────────────────── + +/** Message metadata marking a sub-agent steer (contracts.ts §3). */ +export interface SubagentResultMeta { + jobId: string; + kind: JobKind; + toolCallId: string; +} + +/** The data-subagent-artifact part payload: what the UI renders. */ +export interface SubagentArtifactData { + kind: JobKind; + result: ResearchResult | DemoResult | VisualResult; +} + +/** + * The steer message: user-role on the wire (UIMessage roles are fixed), but + * NEVER a learner interjection — metadata.subagentResult is the marker the + * UI branches on, and the artifact rides the data part (stripped by provider + * conversion, so the model sees only the text part). + */ +export interface SubagentSteerMessage extends UIMessage { + role: "user"; + metadata: { subagentResult: SubagentResultMeta }; + parts: [ + { type: "text"; text: string }, + { type: "data-subagent-artifact"; data: SubagentArtifactData }, + ]; +} + +/** Whether a result is the failure shape (html/svg empty, failure brief). */ +export function isSubagentFailure(kind: JobKind, result: ResearchResult | DemoResult | VisualResult): boolean { + if (kind === "research") { + const r = result as ResearchResult; + return r.sources.length === 0 && /^Research failed:/i.test(r.brief); + } + if (kind === "demo") { + const r = result as DemoResult; + return typeof r.html !== "string" || r.html.length === 0; + } + const r = result as VisualResult; + return typeof r.svg !== "string" || r.svg.length === 0; +} + +/** The model-facing text of a steer: what the result IS, what to do with it. */ +function steerText(kind: JobKind, result: ResearchResult | DemoResult | VisualResult): string { + if (isSubagentFailure(kind, result)) { + if (kind === "research") + return `${(result as ResearchResult).brief}\n\nContinue with what you can establish and flag the uncertainty to the learner.`; + if (kind === "demo") + return `The demo "${(result as DemoResult).title}" could not be built: ${ + (result as DemoResult).notes ?? "unknown reason" + }. Fall back to explaining in prose; if the brief was the problem, re-call demo with a tighter one.`; + return `The figure could not be made: ${(result as VisualResult).notes ?? "unknown reason"}. Fall back to prose or mermaid, or re-call visual with a tighter brief.`; + } + if (kind === "research") + return `Research complete — the researcher's full brief follows. Weave it into the lesson and cite its sources when you use its findings:\n\n${(result as ResearchResult).brief}`; + if (kind === "demo") + return `The demo "${(result as DemoResult).title}" is built, headlessly tested, and rendered in the lesson. Builder notes: ${ + (result as DemoResult).notes ?? "(none)" + }.\nTeach with it — reference its controls by name and point the learner at what to try.`; + return `The figure is drawn, verified by inspection, and rendered in the lesson. Maker notes: ${ + (result as VisualResult).notes ?? "(none)" + }.\nReference its elements as you teach.`; +} + +/** + * Build the steer message for a finished job (pure — shared by the runner, + * the wake pipeline, and tests). Deterministic id per job makes re-delivery + * idempotent. + */ +export function buildSubagentSteer( + job: Pick, + result: ResearchResult | DemoResult | VisualResult, +): SubagentSteerMessage { + return { + id: `steer_${job.id}`, + role: "user", + metadata: { subagentResult: { jobId: job.id, kind: job.kind, toolCallId: job.toolCallId } }, + parts: [ + { type: "text", text: steerText(job.kind, result) }, + { type: "data-subagent-artifact", data: { kind: job.kind, result } }, + ], + }; +} + +/** True when the transcript already carries this job's steer (idempotent delivery). */ +export function steerDelivered(messages: UIMessage[], jobId: string): boolean { + return messages.some( + (m) => (m.metadata as { subagentResult?: { jobId?: string } } | undefined)?.subagentResult?.jobId === jobId, + ); +} + // ─── Demo iframe bridge (W6) ──────────────────────────────────────────────── // postMessage protocol between the sandboxed demo iframe and its parent card. // The bridge script is auto-injected into every demo HTML by the demo-card diff --git a/src/core/ui-message.ts b/src/core/ui-message.ts index 22b33a9..1de809e 100644 --- a/src/core/ui-message.ts +++ b/src/core/ui-message.ts @@ -1,18 +1,29 @@ import type { UIMessage } from "ai"; +import type { DemoToolInput, ResearchToolInput, SubagentArtifactData, VisualToolInput } from "./contracts.ts"; import type { AskToolInput, QuestionAnswerData, QuizToolInput } from "./question-core.ts"; /** * The typed UI message for the teacher chat — gives the frontend typed tool - * parts ("tool-quiz" / "tool-ask_user_question" with typed input/output) - * without importing the server-side agent into the client bundle. + * parts ("tool-quiz" / "tool-ask_user_question" / "tool-research" / …) without + * importing the server-side agent into the client bundle. * - * Declared explicitly (not via InferUITools) because a client-side tool has - * no execute function, so its OUTPUT type lives only here: the structured - * answer record the client builds and sends via addToolOutput. + * Question tools are CLIENT-SIDE (no execute): their OUTPUT type is the + * structured answer record the client builds and sends via addToolOutput. + * Job tools are SERVER-EXECUTED: their output is the plain-string dispatch + * acknowledgement (the result later arrives as a steer message — see + * SubagentSteerMessage in contracts.ts, whose artifact rides the + * data-subagent-artifact part typed here as DATA_PARTS). */ export type TeacherUITools = { quiz: { input: QuizToolInput; output: QuestionAnswerData }; ask_user_question: { input: AskToolInput; output: QuestionAnswerData }; + research: { input: ResearchToolInput; output: string }; + demo: { input: DemoToolInput; output: string }; + visual: { input: VisualToolInput; output: string }; }; -export type TeacherUIMessage = UIMessage; +export type TeacherUIDataParts = { + "subagent-artifact": SubagentArtifactData; +}; + +export type TeacherUIMessage = UIMessage; diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index 7bab646..54f1b4a 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -3,9 +3,10 @@ import { createTeacherAgent, type TeacherAgent } from "../agent/teacher-agent.ts import { createJobRunner } from "../agent/subagents/runner.ts"; import { createExtractionPipeline } from "../agent/subagents/extraction.ts"; import { createTeacherTools } from "../agent/tools.ts"; +import { deliverJobResult, setJobRespawn, setTurnRunner } from "./jobs.ts"; import { stores } from "./stores.ts"; import { beginTurn, endTurn } from "./turn-liveness.ts"; -import { buildTeacherContext, persistTurn, spawnTurnJobs, triggerExtraction } from "./turn-service.ts"; +import { buildTeacherContext, persistTurn, triggerExtraction } from "./turn-service.ts"; /** * One chat turn: run the teacher agent over the client's message list and @@ -14,14 +15,38 @@ import { buildTeacherContext, persistTurn, spawnTurnJobs, triggerExtraction } fr * handler in app/api/chat is a one-line wrapper. * * This module is the TURN COMPOSITION ROOT: it wires the storage ports, the - * job runner, the teacher tools, and the turn services (turn-service.ts) - * together — all actual work lives behind those seams. The background job - * runner closes over the same port objects the services see. + * job runner + wake pipeline, the teacher tools, and the turn services + * (turn-service.ts) together — all actual work lives behind those seams. + * THE SAME COMPOSITION serves both entry points: client requests and the + * wake pipeline's server-side steers (src/server/jobs.ts) — one code path + * for write-ahead persistence, turn liveness, and the turn services. */ -const jobRunner = createJobRunner(stores); +const jobRunner = createJobRunner({ + chatStore: stores.chatStore, + jobStore: stores.jobStore, + deliver: deliverJobResult, +}); const extraction = createExtractionPipeline(stores); +// Wake-pipeline wiring (contracts.ts "THE ASYNC PROTOCOL" §4): the pipeline +// fires its server-side teacher turns through THIS composition root, and +// crash recovery re-spawns orphaned sub-agent loops through THIS runner. +setTurnRunner(async (chatId, messages) => { + const res = await handleChatRequest( + new Request("http://localhost/api/chat", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ chatId, messages }), + }), + ); + // Drive the stream to completion server-side — nobody else will read it, + // and onFinish (persist, extraction, endTurn) only runs when it drains. + await res.arrayBuffer().catch(() => undefined); + return res.status; +}); +setJobRespawn((jobId) => void jobRunner.runJob(jobId)); + export async function handleChatRequest(req: Request, agent?: TeacherAgent): Promise { let body: { chatId?: unknown; messages?: unknown }; try { @@ -64,7 +89,12 @@ export async function handleChatRequest(req: Request, agent?: TeacherAgent): Pro agent ?? createTeacherAgent( undefined, - createTeacherTools({ chatId, chatStore: stores.chatStore }), + createTeacherTools({ + chatId, + chatStore: stores.chatStore, + jobStore: stores.jobStore, + spawn: (job, input) => void jobRunner.runJob(job.id, input), + }), buildTeacherContext(stores.graphStore, chatId), ), uiMessages: body.messages as UIMessage[], @@ -81,7 +111,6 @@ export async function handleChatRequest(req: Request, agent?: TeacherAgent): Pro }, onFinish: ({ messages, isAborted }) => { persistTurn(stores.chatStore, chatId, messages as UIMessage[], isAborted); - spawnTurnJobs(jobRunner, chatId, messages as UIMessage[]); triggerExtraction(stores, extraction, chatId); // LAST: a poller that sees not-in-flight must read an already-persisted // record — endTurn only after persistTurn has done its job. diff --git a/src/server/jobs.ts b/src/server/jobs.ts index 5202c4f..bde2032 100644 --- a/src/server/jobs.ts +++ b/src/server/jobs.ts @@ -1,10 +1,303 @@ -import { stores } from "./stores.ts"; - /** - * Server-side job-status entry (GET /api/jobs?chatId=…, contracts.ts - * "THE ASYNC PROTOCOL" §4): the client polls while any job tool part is - * pending, finds the job by toolCallId, renders its activity line, and - * re-syncs the chat when the job flips to done/error. + * THE WAKE PIPELINE (contracts.ts "THE ASYNC PROTOCOL" §4) — server-side + * steer delivery for finished sub-agent jobs. + * + * deliverJobResult(steer) — the runner calls this on job completion. The + * steer joins a per-chat serialized queue and a pump drives it: + * + * 1. WAIT FOR IDLE: while a turn is in flight (turn-liveness) OR a + * question tool part is pending (an unanswered client-side tool + * call breaks provider conversion — the steer fires right after + * the learner's continuation), poll. A short grace after the gates + * open avoids stealing the turn a client is about to send. + * 2. FIRE: a server-side teacher turn through the SAME composition as + * a client request (the injected turn runner wraps + * handleChatRequest): write-ahead persist, liveness, extraction — + * one code path. A 409 means a client turn raced in — retry with + * backoff. Idempotent per jobId (steerDelivered): a crashed wake + * or a duplicate deliver is skipped, not duplicated. + * 3. FLIP: only now does the job go done/error — a polling client + * that sees the flip re-syncs once and gets the finished record + * (steer + the teacher's response to it). + * + * Crash recovery (recoverStaleJobs): a process restart can orphan jobs — + * "running" with a result (the wake never fired) or without one (the + * sub-agent loop itself died). The sweep re-delivers finished results + * (idempotently) and re-spawns unfinished loops, both via injected + * seams. It runs lazily off the /api/jobs poll, throttled. + * + * All mutable state lives on globalThis (the turn-liveness lesson: Next dev + * gives each route entry its own module instances — a module-level Map would + * be a different object in each). */ -export const { listJobs } = stores.jobStore; +import type { UIMessage } from "ai"; +import { + buildSubagentSteer, + isSubagentFailure, + steerDelivered, + type JobRecord, + type SubagentSteerMessage, +} from "../core/contracts.ts"; +import { findPendingToolCall } from "../core/pending.ts"; +import { stores } from "./stores.ts"; +import { isTurnRunning } from "./turn-liveness.ts"; + +/** Fires one server-side teacher turn: (chatId, messages) → HTTP-ish status. */ +export type TurnRunner = (chatId: string, messages: UIMessage[]) => Promise; + +/** Re-runs a job's sub-agent loop (crash recovery for orphaned jobs). */ +export type JobRespawn = (jobId: string) => void; + +interface QueuedSteer { + steer: SubagentSteerMessage; + resolve: () => void; + reject: (reason: Error) => void; +} + +interface WakeState { + turnRunner?: TurnRunner; + respawn?: JobRespawn; + /** chatId → steers awaiting delivery, in arrival order. */ + queues: Map; + /** Chats whose pump loop is currently active (serialization per chat). */ + pumping: Set; + /** Throttle bookkeeping for the stale-job sweep. */ + lastSweepAt: number; +} + +interface LhWakeGlobal { + __lhWakePipeline?: WakeState; +} + +function state(): WakeState { + const g = globalThis as typeof globalThis & LhWakeGlobal; + return (g.__lhWakePipeline ??= { queues: new Map(), pumping: new Set(), lastSweepAt: 0 }); +} + +/** chat-handler wires this at module init: its adapter around handleChatRequest. */ +export function setTurnRunner(run: TurnRunner): void { + state().turnRunner = run; +} + +/** chat-handler wires this at module init: its job runner's runJob. */ +export function setJobRespawn(respawn: JobRespawn): void { + state().respawn = respawn; +} + +const POLL_MS = 400; +/** After the gates open, wait a beat before firing — a client that just + * answered a question is about to send its continuation; don't 409 it. */ +const GRACE_MS = 1500; +/** 409 backoff: capped ramp so a busy client can't spin the pump. */ +const RETRY_BACKOFF_MS = [500, 1000, 2000, 5000] as const; + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function jobIdOf(steer: SubagentSteerMessage): string { + return (steer.metadata as { subagentResult: { jobId: string } }).subagentResult.jobId; +} + +/** + * Hand a finished job's steer to the pipeline. Resolves when the wake turn + * has completed (and the job status flipped); rejects when delivery fails + * unrecoverably. Never rejects for queueing itself. + */ +export function deliverJobResult(steer: SubagentSteerMessage): Promise { + return new Promise((resolve, reject) => { + const s = state(); + const chatId = chatIdOf(steer); + const queue = s.queues.get(chatId) ?? []; + queue.push({ steer, resolve, reject }); + s.queues.set(chatId, queue); + void pump(chatId); + }); +} + +function chatIdOf(steer: SubagentSteerMessage): string { + // The chat is resolvable from the job record; the steer's metadata + // carries jobId, which indexes the store. + const jobId = jobIdOf(steer); + const job = stores.jobStore.getJob(jobId); + return job?.chatId ?? ""; +} + +/** + * The per-chat pump: delivers every queued steer in order, one wake turn + * each. Only one loop per chat at a time; later arrivals ride the queue. + */ +async function pump(chatId: string): Promise { + if (!chatId) return; + const s = state(); + if (s.pumping.has(chatId)) return; + s.pumping.add(chatId); + try { + for (let attempt = 0; ; ) { + const queue = s.queues.get(chatId); + if (!queue || queue.length === 0) { + s.queues.delete(chatId); + return; + } + const entry = queue[0]!; + const jobId = jobIdOf(entry.steer); + + // Gone chat: nothing to deliver into. + const record = stores.chatStore.getChat(chatId); + if (!record) { + queue.shift(); + entry.reject(new Error(`chat ${chatId} not found — cannot deliver job ${jobId}`)); + continue; + } + + // Idempotent delivery: the transcript already carries this steer + // (a previous wake completed, or a duplicate deliver raced in). + if (steerDelivered(record.messages, jobId)) { + queue.shift(); + finalizeJob(jobId); + entry.resolve(); + continue; + } + + await waitForIdle(chatId); + + // Re-fetch AFTER the gates open — waitForIdle can wait out a quiz + // answer (minutes), and sending the pre-wait snapshot would clobber + // the client's answer + continuation in the write-ahead persist. + const fresh = stores.chatStore.getChat(chatId); + if (!fresh) { + queue.shift(); + entry.reject(new Error(`chat ${chatId} not found — cannot deliver job ${jobId}`)); + continue; + } + + // The runner is required — chat-handler sets it at init. If some + // module instance missed the wiring, fail loudly rather than hang. + if (!s.turnRunner) { + queue.shift(); + const err = new Error("wake pipeline: no turn runner wired (setTurnRunner)"); + failJob(jobId, err); + entry.reject(err); + continue; + } + + // Fire the server-side teacher turn. A 409 = a client turn raced + // in between idle and firing — back off and retry the same steer. + const status = await s.turnRunner(chatId, [...fresh.messages, entry.steer]).catch((err: unknown) => { + console.error(`[job ${jobId}] wake turn crashed:`, err); + return -1; + }); + if (status === 409) { + await delay(RETRY_BACKOFF_MS[Math.min(attempt++, RETRY_BACKOFF_MS.length - 1)]!); + continue; + } + attempt = 0; + + queue.shift(); + if (status === 200) { + finalizeJob(jobId); + entry.resolve(); + } else { + const err = new Error(`wake turn failed with status ${status}`); + failJob(jobId, err); + entry.reject(err); + } + } + } finally { + s.pumping.delete(chatId); + // A steer may have arrived between the last check and the finally — + // restart the pump rather than strand it. + if ((state().queues.get(chatId) ?? []).length > 0) void pump(chatId); + } +} + +/** Flip the job to its final status now that the lesson has resumed. */ +function finalizeJob(jobId: string): void { + const job = stores.jobStore.getJob(jobId); + if (!job || job.status !== "running") return; + const result = job.result; + const failed = result ? isSubagentFailure(job.kind, result as never) : true; + stores.jobStore.updateJob(jobId, { + status: failed ? "error" : "done", + activity: failed ? "failed" : "delivered", + }); +} + +function failJob(jobId: string, err: Error): void { + stores.jobStore.updateJob(jobId, { status: "error", error: err.message, activity: "delivery failed" }); +} + +/** + * Wait until the chat can take a wake turn: no turn in flight AND no + * unanswered question part (provider conversion would break). Poll; the + * learner may take arbitrarily long to answer, and that is correct — the + * steer fires right after their continuation completes. + */ +async function waitForIdle(chatId: string): Promise { + for (;;) { + const record = stores.chatStore.getChat(chatId); + if (!record) return; + const blocked = isTurnRunning(chatId) || findPendingToolCall(record.messages) !== null; + if (!blocked) { + await delay(GRACE_MS); + const recheck = stores.chatStore.getChat(chatId); + if ( + !isTurnRunning(chatId) && + recheck && + findPendingToolCall(recheck.messages) === null + ) { + return; + } + } + await delay(POLL_MS); + } +} + +// ─── crash recovery ────────────────────────────────────────────────────────── + +/** Jobs this old with status "running" are orphans (crash/restart). */ +const STALE_JOB_MS = 15 * 60_000; +const SWEEP_THROTTLE_MS = 30_000; + +/** + * Lazy sweep (called off the /api/jobs poll, throttled): every stale + * "running" job either has a result (the wake never fired — re-deliver, + * idempotently) or doesn't (the loop itself died — re-spawn it). + */ +function recoverStaleJobs(): void { + const s = state(); + const now = Date.now(); + if (now - s.lastSweepAt < SWEEP_THROTTLE_MS) return; + s.lastSweepAt = now; + try { + for (const chat of stores.chatStore.listChats()) { + for (const job of stores.jobStore.listJobs(chat.id)) { + if (job.status !== "running") continue; + if (now - Date.parse(job.updatedAt) < STALE_JOB_MS) continue; + if (!stores.chatStore.getChat(job.chatId)) continue; + if (job.result) { + // The wake never fired (or crashed mid-flight): re-deliver. + const steer = buildSubagentSteer(job, job.result as never); + stores.jobStore.updateJob(job.id, { activity: "recovering: re-delivering" }); + void deliverJobResult(steer).catch(() => { + /* deliverJobResult marks the job error on failure */ + }); + } else { + // The sub-agent loop itself died with the process: re-spawn. + if (!s.respawn) continue; + stores.jobStore.updateJob(job.id, { activity: "recovering: re-running" }); + s.respawn(job.id); + } + } + } + } catch (err) { + console.error("[jobs] stale-job recovery sweep failed:", err); + } +} + +/** The /api/jobs entry: job list for the client poll (+ lazy recovery). */ +export function listJobs(chatId: string): JobRecord[] { + if (process.env.NODE_ENV !== "test") recoverStaleJobs(); + return stores.jobStore.listJobs(chatId); +} diff --git a/src/server/turn-service.ts b/src/server/turn-service.ts index 323af20..5ba29ff 100644 --- a/src/server/turn-service.ts +++ b/src/server/turn-service.ts @@ -6,7 +6,6 @@ * * buildTeacherContext — graph → instructions block (the W8 read channel) * persistTurn — the canonical state write (client + agent messages) - * spawnTurnJobs — the async sub-agent protocol kick-off * triggerExtraction — the graph write channel: fires the extraction * pipeline (the sole graph writer) once a chat * crosses the v1 quiz-answer threshold. The trigger @@ -22,7 +21,6 @@ import { buildGraphContext } from "../core/graph-context.ts"; import { getGraph, pendingQuizAnswers } from "../core/graph.ts"; import type { ChatStore, GraphStore } from "../core/ports.ts"; import { EXTRACTION } from "../../config.ts"; -import type { JobRunner } from "../agent/subagents/runner.ts"; import type { ExtractionPipeline } from "../agent/subagents/extraction.ts"; /** @@ -55,19 +53,6 @@ export function persistTurn(chatStore: ChatStore, chatId: string, messages: UIMe } } -/** - * The async sub-agent protocol (contracts.ts): scan the last assistant - * message for pending job tool parts and fire the background runners — - * fire-and-forget, never awaited; the HTTP response is already gone. - */ -export function spawnTurnJobs(jobRunner: JobRunner, chatId: string, messages: UIMessage[]): void { - try { - jobRunner.spawnPendingJobs(chatId, messages); - } catch (err) { - console.error(`[chat ${chatId}] job spawn failed:`, err); - } -} - /** * The graph write channel (vision §1 construction): the extraction pipeline * (extractor → auditor → merger — the sole graph writer) runs ASYNC once a diff --git a/tests/demos.test.tsx b/tests/demos.test.tsx index 5db442e..533a149 100644 --- a/tests/demos.test.tsx +++ b/tests/demos.test.tsx @@ -11,9 +11,10 @@ * no-Chrome failure) * 4. the visual-maker loop (fake model + fake headless): render-and-look * success, forced look on premature finish, never-writes failure - * 5. job wrappers + spawn-hook kind dispatch (demo part → demo job → - * part fulfilled; visual failure path) - * 6. tool registry wiring (client-side demo/visual tools, schemas) + * 5. job runners + dispatch (demo part → demo job → steer via a fake + * deliver; visual failure path; end-to-end dispatch through + * handleChatRequest with the ack protocol) + * 6. tool registry wiring (server-executed dispatch tools, schemas) * * Cleanup follows tests/jobs.test.ts idioms (created chat/job files removed * in the after() hook; overrides reset in finally blocks). @@ -43,13 +44,35 @@ import { runDemoBuilder } from "../src/agent/subagents/demo-builder.ts"; import { headlessOverride, type DemoRunResult, type HeadlessApi } from "../src/agent/subagents/headless.ts"; import { createJobRunner, demoModelOverride, visualModelOverride } from "../src/agent/subagents/runner.ts"; import { runVisualMaker } from "../src/agent/subagents/visual-maker.ts"; +import { isSubagentFailure, type SubagentSteerMessage } from "../src/core/contracts.ts"; +import { createTeacherTools } from "../src/agent/tools.ts"; import { chatStore } from "../src/store/chats.ts"; import { jobStore } from "../src/store/jobs.ts"; - -/** The runner with the same wired fs ports the chat-handler injects. */ -const jobRunner = createJobRunner({ chatStore, jobStore }); import { handleChatRequest } from "../src/server/chat-handler.ts"; import { createTeacherAgent } from "../src/agent/teacher-agent.ts"; + +/** + * A fake deliver seam + a runner wired like the chat-handler's, per test: + * capture the steer and mimic the pipeline's final status flip so the job + * record reads exactly as production does after the wake turn completes. + */ +function testRunner() { + const steers: SubagentSteerMessage[] = []; + const runner = createJobRunner({ + chatStore, + jobStore, + deliver: async (steer) => { + steers.push(steer); + const jobId = steer.metadata.subagentResult.jobId; + const job = jobStore.getJob(jobId); + if (job && job.status === "running") { + const failed = job.result ? isSubagentFailure(job.kind, job.result as never) : true; + jobStore.updateJob(jobId, { status: failed ? "error" : "done", activity: failed ? "failed" : "delivered" }); + } + }, + }); + return { steers, runner }; +} import { artifactRenderers } from "../components/artifacts/index.ts"; const JOBS_DIR = path.join(PATHS.data, "jobs"); @@ -399,6 +422,37 @@ describe("visual maker (fake model + fake headless)", () => { assert.equal(headless.svgRenders.length, 1, "the gate rendered the svg once itself"); }); + it("edit_svg: exact-match single replacement, with fuzzy context on a miss", async () => { + const headless = fakeHeadless(); + const model = scriptedModel([ + { toolCalls: [{ toolCallId: "w1", toolName: "write_svg", input: { svg: SVG_1 } }] }, + { toolCalls: [{ toolCallId: "r1", toolName: "render_svg", input: {} }] }, + // A miss: old_text not in the document (error fed back, nothing changed) + { toolCalls: [{ toolCallId: "e1", toolName: "edit_svg", input: { old_text: "NOT-PRESENT", new_text: "x" } }] }, + // A hit: exactly one occurrence — applied + { toolCalls: [{ toolCallId: "e2", toolName: "edit_svg", input: { old_text: ' { + const headless = fakeHeadless(); + const model = scriptedModel([ + // Nothing written yet — the edit must error, not crash the loop + { toolCalls: [{ toolCallId: "e0", toolName: "edit_svg", input: { old_text: "svg", new_text: "x" } }] }, + { text: "Cannot proceed." }, + ]); + const result = await runVisualMaker({ brief: "b" }, { model, headless }); + assert.equal(result.svg, "", "the failed edit produced no deliverable"); + assert.match(result.notes ?? "", /never wrote an SVG/); + }); + it("feeds render errors back to the model instead of crashing", async () => { const headless = fakeHeadless({ renderFailures: 1 }); const model = scriptedModel([ @@ -427,7 +481,7 @@ describe("visual maker (fake model + fake headless)", () => { }); }); -// ─── 5. job wrappers + spawn dispatch ──────────────────────────────────────── +// ─── 5. job runners + dispatch ─────────────────────────────────────────────── const DEMO_INPUT = { title: "Compound interest", brief: "Slider for rate, curve of growth over 30 years." }; @@ -448,8 +502,9 @@ async function drain(res: Response): Promise { return res.text(); } + describe("demo/visual job runners", () => { - it("runDemoJob fulfills the part with a tested DemoResult and marks the job done", async () => { + it("runJob completes a demo: result on the job, steer via deliver, ack stays on the part", async () => { const chat = chatStore.createChat("Test: demo runner"); createdChats.push(chat.id); chatStore.saveMessages(chat.id, [ @@ -457,42 +512,52 @@ describe("demo/visual job runners", () => { { id: "a1", role: "assistant", - parts: [{ type: "tool-demo", toolCallId: "call-d1", state: "input-available", input: DEMO_INPUT }], + parts: [{ type: "tool-demo", toolCallId: "call-d1", state: "output-available", input: DEMO_INPUT, output: "Dispatched." }], }, ] as never); const job = jobStore.createJob({ kind: "demo", chatId: chat.id, toolCallId: "call-d1" }); createdJobs.push(job.id); - // runDemoJob has no headless injection point — the module override is the seam. + // runJob has no headless injection point — the module override is the seam. headlessOverride.current = fakeHeadless(); + demoModelOverride.current = scriptedModel([ + { toolCalls: [{ toolCallId: "w1", toolName: "write_demo", input: { html: GOOD_HTML } }] }, + { toolCalls: [{ toolCallId: "r1", toolName: "run_demo", input: {} }] }, + { text: "Sweep the rate slider." }, + ]); + const { steers, runner } = testRunner(); try { - await jobRunner.runDemoJob(job.id, { - model: scriptedModel([ - { toolCalls: [{ toolCallId: "w1", toolName: "write_demo", input: { html: GOOD_HTML } }] }, - { toolCalls: [{ toolCallId: "r1", toolName: "run_demo", input: {} }] }, - { text: "Sweep the rate slider." }, - ]), - }); + await runner.runJob(job.id); } finally { headlessOverride.current = undefined; + demoModelOverride.current = undefined; } const done = jobStore.getJob(job.id)!; assert.equal(done.status, "done"); - assert.equal(done.activity, "run_demo: clean", "final activity line recorded"); - const result = done.result as { title: string; html: string; tested: boolean }; + assert.equal(done.activity, "delivered", "the fake deliver flips status like the wake pipeline"); + const result = done.result as { title: string; html: string; tested: boolean; notes?: string }; assert.equal(result.title, "Compound interest"); + assert.equal(result.html, GOOD_HTML); assert.equal(result.tested, true); + // The steer carries the artifact for the renderer. + assert.equal(steers.length, 1, "exactly one steer"); + const data = steers[0]!.parts[1]; + assert.ok(data.type === "data-subagent-artifact"); + assert.equal(data.data.kind, "demo"); + assert.deepEqual(data.data.result, result); + + // The tool part keeps the DISPATCH ACK forever — the result never + // rewrites it (the teacher gets the result once, via the steer). const saved = chatStore.getChat(chat.id)!; const part = saved.messages[1]!.parts.find((p) => p.type === "tool-demo"); assert.ok(part && part.type === "tool-demo", "demo tool part persisted"); - assert.equal(part.state, "output-available", "part fulfilled"); - assert.equal((part as { output?: { html?: string } }).output?.html, GOOD_HTML); + assert.equal(typeof (part as { output?: unknown }).output, "string", "part output is the ack, not the result"); }); - it("runVisualJob fulfills the part even on failure and marks the job error", async () => { + it("runJob still steers on visual failure and the job lands error", async () => { const chat = chatStore.createChat("Test: visual runner failure"); createdChats.push(chat.id); chatStore.saveMessages(chat.id, [ @@ -500,33 +565,38 @@ describe("demo/visual job runners", () => { { id: "a1", role: "assistant", - parts: [{ type: "tool-visual", toolCallId: "call-v1", state: "input-available", input: { brief: "b" } }], + parts: [{ type: "tool-visual", toolCallId: "call-v1", state: "output-available", input: { brief: "b" }, output: "Dispatched." }], }, ] as never); const job = jobStore.createJob({ kind: "visual", chatId: chat.id, toolCallId: "call-v1" }); createdJobs.push(job.id); - await jobRunner.runVisualJob(job.id, { model: scriptedModel([{ text: "cannot" }]) }); + visualModelOverride.current = scriptedModel([{ text: "cannot" }]); + const { steers, runner } = testRunner(); + try { + await runner.runJob(job.id); + } finally { + visualModelOverride.current = undefined; + } const errored = jobStore.getJob(job.id)!; assert.equal(errored.status, "error"); assert.match(errored.error ?? "", /never wrote an SVG/); - const saved = chatStore.getChat(chat.id)!; - const part = saved.messages[1]!.parts.find((p) => p.type === "tool-visual"); - assert.ok(part && part.type === "tool-visual", "visual tool part persisted"); - assert.equal(part.state, "output-available", "failure still fulfills the part"); - const output = (part as { output?: { svg?: string; notes?: string } }).output!; - assert.equal(output.svg, ""); - assert.match(output.notes ?? "", /^FAILED:/); + assert.ok(steers[0], "failure still delivers a steer — the teacher learns the outcome"); + const data = steers[0]!.parts[1]; + assert.ok(data.type === "data-subagent-artifact"); + const result = data.data.result as { svg: string; notes?: string }; + assert.equal(result.svg, ""); + assert.match(result.notes ?? "", /^FAILED:/); }); - it("the spawn hook dispatches a demo part to the demo runner (kind→runner map)", async () => { + it("dispatch through a teacher turn: the call acks mid-turn and the runner completes in the background", async () => { const chat = chatStore.createChat("Test: demo spawn dispatch"); createdChats.push(chat.id); - // A teacher that calls demo on turn 1 (scripted stream, like jobs.test.ts). + // A teacher that calls demo on turn 1, then continues in the SAME turn. let turn = 0; const teacher = new MockLanguageModelV4({ doStream: async () => { @@ -552,12 +622,15 @@ describe("demo/visual job runners", () => { }, }); + // The overrides must be live BEFORE the request — dispatch fires the + // runner mid-stream, inside the tool execute. demoModelOverride.current = scriptedModel([ { toolCalls: [{ toolCallId: "w1", toolName: "write_demo", input: { html: GOOD_HTML } }] }, { toolCalls: [{ toolCallId: "r1", toolName: "run_demo", input: {} }] }, { text: "Sweep it." }, ]); headlessOverride.current = fakeHeadless(); + const { steers, runner } = testRunner(); try { await drain( @@ -570,28 +643,38 @@ describe("demo/visual job runners", () => { messages: [{ id: "u1", role: "user", parts: [{ type: "text", text: "teach me compounding" }] }], }), }), - createTeacherAgent(teacher), + createTeacherAgent( + teacher, + createTeacherTools({ + chatId: chat.id, + chatStore, + jobStore, + spawn: (job, input) => void runner.runJob(job.id, input), + }), + ), ), ); const job = jobStore.findJobByToolCallId(chat.id, "call-d2"); - assert.ok(job, "spawn hook created a job for the demo part"); + assert.ok(job, "dispatch created a job at tool-execute time"); assert.equal(job!.kind, "demo"); createdJobs.push(job!.id); // The background runner (fire-and-forget) completes with the overrides. - await waitFor(() => jobStore.getJob(job!.id)!.status !== "running", 5000, "demo job completion"); + await waitFor(() => jobStore.getJob(job!.id)!.status !== "running", 8000, "demo job completion"); const done = jobStore.getJob(job!.id)!; assert.equal(done.status, "done"); - assert.equal(done.activity, "run_demo: clean"); const saved = chatStore.getChat(chat.id)!; const part = saved.messages[1]!.parts.find((p) => p.type === "tool-demo"); assert.ok(part && part.type === "tool-demo", "demo tool part persisted"); - assert.equal(part.state, "output-available", "demo part fulfilled by the background runner"); + assert.equal(part.state, "output-available", "the part resolved with the ack at dispatch time"); + assert.match(String((part as { output?: unknown }).output), /fire-and-forget/); - // Idempotency: re-running the spawn hook creates no second job. - jobRunner.spawnPendingJobs(chat.id, saved.messages); + assert.equal(steers.length, 1, "one job, one steer"); + assert.equal(steers[0]!.metadata.subagentResult.jobId, job!.id); + + // Idempotency: dispatch dedupes by toolCallId — no second job. assert.equal(jobStore.listJobs(chat.id).length, 1, "no duplicate job for the same tool part"); } finally { demoModelOverride.current = undefined; @@ -603,11 +686,12 @@ describe("demo/visual job runners", () => { // ─── 6. registry wiring ────────────────────────────────────────────────────── describe("tool + renderer registry wiring", () => { - it("demo and visual are CLIENT-SIDE tools with the contract schemas", () => { - const tools = jobTools(); + it("demo and visual are SERVER-EXECUTED dispatch tools with the contract schemas", () => { + const { runner } = testRunner(); + const tools = jobTools({ chatId: "chat_registry", jobStore, spawn: (job) => void runner.runJob(job.id) }); assert.deepEqual(Object.keys(tools).sort(), ["demo", "research", "visual"]); - for (const name of ["demo", "visual"] as const) { - assert.equal(typeof (tools[name] as { execute?: unknown }).execute, "undefined", `${name} has no execute — client-side like quiz/ask`); + for (const name of ["demo", "visual", "research"] as const) { + assert.equal(typeof (tools[name] as { execute?: unknown }).execute, "function", `${name} executes server-side — fire-and-forget ack per THE ASYNC PROTOCOL`); } assert.equal(demoInputSchema.safeParse({ title: "T", brief: "b" }).success, true); assert.equal(demoInputSchema.safeParse({ brief: "b" }).success, false, "demo requires a title"); diff --git a/tests/jobs.test.ts b/tests/jobs.test.ts index 4e53139..8627108 100644 --- a/tests/jobs.test.ts +++ b/tests/jobs.test.ts @@ -2,15 +2,22 @@ * Job infrastructure tests — deterministic, zero network: * * 1. job store CRUD (create → get → update → list → dedup by toolCallId) - * 2. the spawn hook: drive handleChatRequest with a scripted mock teacher - * that emits a research tool call → a job record appears (runner fires - * with the mock override, so no network) - * 3. runner completion: a fake researcher model (web_search call → final - * text brief, globalThis.fetch stubbed) → the stored chat's part flips - * to output-available with a ResearchResult-shaped output, the job is - * done with the activity recorded - * 4. runner failure: a model that throws → the part is still fulfilled - * with a failure brief and the job is marked error + * 2. steer helpers (pure, contracts.ts): buildSubagentSteer shape per kind, + * isSubagentFailure, steerDelivered + * 3. dispatch (async protocol §2): drive handleChatRequest with a scripted + * mock teacher that emits a research tool call → the tool part resolves + * IMMEDIATELY with the fire-and-forget ack (the teacher's turn + * CONTINUES — no turn break at the call), a job record exists, and the + * test runner (fake deliver) produces the steer + * 4. runner completion: a fake researcher model (web_search call → final + * text brief, globalThis.fetch stubbed) → the job stores the result, + * the steer carries the brief + sources, the fake deliver flips the + * job done + * 5. runner failure: a model that throws → the failure steer still fires + * (the teacher learns the outcome) and the job lands error + * 6. the wake pipeline (src/server/jobs.ts): server-side steer delivery — + * happy path (fires the injected turn runner, flips the job), pending- + * quiz deferral, 409 retry, idempotent redelivery * * Cleanup follows tests/chat-flow.test.ts: created chat/job files removed in * finally blocks / the after() hook. @@ -22,17 +29,25 @@ import * as path from "node:path"; import { after, describe, it } from "node:test"; import { MockLanguageModelV4, simulateReadableStream } from "ai/test"; import type { LanguageModelV4StreamPart } from "@ai-sdk/provider"; +import { + buildSubagentSteer, + isSubagentFailure, + steerDelivered, + type DemoResult, + type ResearchResult, + type SubagentSteerMessage, + type VisualResult, +} from "../src/core/contracts.ts"; import { createTeacherAgent } from "../src/agent/teacher-agent.ts"; +import { createTeacherTools } from "../src/agent/tools.ts"; import { handleChatRequest } from "../src/server/chat-handler.ts"; +import { deliverJobResult, setTurnRunner } from "../src/server/jobs.ts"; import { chatStore } from "../src/store/chats.ts"; import { jobStore } from "../src/store/jobs.ts"; import { createJobRunner, researchModelOverride } from "../src/agent/subagents/runner.ts"; import { createResearcherAgent } from "../src/agent/subagents/researcher.ts"; import { PATHS } from "../config.ts"; -/** The runner with the same wired fs ports the chat-handler injects. */ -const jobRunner = createJobRunner({ chatStore, jobStore }); - const JOBS_DIR = path.join(PATHS.data, "jobs"); const createdChats: string[] = []; @@ -61,6 +76,7 @@ after(() => { /* already gone */ } } + researchModelOverride.current = undefined; }); function usage() { @@ -85,6 +101,18 @@ function post(chatId: string, messages: unknown[]): Request { }); } +function waitFor(cond: () => boolean, timeoutMs = 8000, what = "condition"): Promise { + return new Promise((resolve, reject) => { + const started = Date.now(); + const tick = () => { + if (cond()) return resolve(); + if (Date.now() - started > timeoutMs) return reject(new Error(`timeout waiting for ${what}`)); + setTimeout(tick, 25); + }; + tick(); + }); +} + // ─── 1. job store CRUD ─────────────────────────────────────────────────────── describe("job store", () => { @@ -118,14 +146,89 @@ describe("job store", () => { }); }); -// ─── scripted teacher: emits a research tool call ──────────────────────────── +// ─── 2. steer helpers (pure) ───────────────────────────────────────────────── + +describe("steer helpers (contracts.ts)", () => { + const research: ResearchResult = { brief: "BRIEF", sources: [{ title: "s", url: "https://x.test" }] }; + const demo: DemoResult = { title: "T", html: "", tested: true, notes: "ok" }; + const visual: VisualResult = { svg: "", notes: "ok" }; + + it("buildSubagentSteer: user-role, metadata-marked, artifact on the data part, brief on the text part", () => { + const job = { id: "job_x", kind: "research" as const, toolCallId: "call-x" }; + const steer = buildSubagentSteer(job, research); + assert.equal(steer.id, "steer_job_x", "deterministic id per job — idempotent delivery"); + assert.equal(steer.role, "user"); + assert.deepEqual(steer.metadata, { subagentResult: { jobId: "job_x", kind: "research", toolCallId: "call-x" } }); + const [text, data] = steer.parts; + assert.equal(text.type, "text"); + assert.ok(text.type === "text" && text.text.includes("BRIEF"), "the model-facing text carries the brief"); + assert.equal(data.type, "data-subagent-artifact"); + assert.ok(data.type === "data-subagent-artifact"); + assert.equal(data.data.kind, "research"); + assert.deepEqual(data.data.result, research); + }); + + it("isSubagentFailure: failure shapes only (empty html/svg, failure brief)", () => { + assert.equal(isSubagentFailure("research", research), false); + assert.equal(isSubagentFailure("research", { brief: "Research failed: x", sources: [] }), true); + // a brief with sources is a success even if the text mentions failure + assert.equal(isSubagentFailure("research", { brief: "Research failed: x", sources: [{ title: "s", url: "https://x" }] }), false); + assert.equal(isSubagentFailure("demo", demo), false); + assert.equal(isSubagentFailure("demo", { ...demo, html: "" }), true); + assert.equal(isSubagentFailure("visual", visual), false); + assert.equal(isSubagentFailure("visual", { ...visual, svg: "" }), true); + }); + + it("steerDelivered: matches on metadata.subagentResult.jobId", () => { + const steer = buildSubagentSteer({ id: "job_y", kind: "demo", toolCallId: "c" }, demo); + assert.equal(steerDelivered([], "job_y"), false); + assert.equal(steerDelivered([steer], "job_y"), true); + assert.equal(steerDelivered([steer], "job_z"), false); + }); + + it("failure steers tell the teacher how to recover (research/demo/visual wording)", () => { + const failResearch = buildSubagentSteer({ id: "j1", kind: "research", toolCallId: "c" }, { brief: "Research failed: API down", sources: [] }); + const failDemo = buildSubagentSteer({ id: "j2", kind: "demo", toolCallId: "c" }, { title: "T", html: "", tested: true, notes: "FAILED: boom" }); + const failVisual = buildSubagentSteer({ id: "j3", kind: "visual", toolCallId: "c" }, { svg: "", notes: "FAILED: boom" }); + for (const steer of [failResearch, failDemo, failVisual]) { + const text = steer.parts[0]; + assert.ok(text.type === "text" && !text.text.includes("rendered in the lesson"), "failure text never claims the artifact landed"); + } + assert.ok(failResearch.parts[0].type === "text" && failResearch.parts[0].text.includes("flag the uncertainty")); + }); +}); + +// ─── fake deliver (the wake pipeline's stand-in for runner tests) ─────────── + +/** + * A fake deliver seam: capture the steer and mimic the pipeline's final + * status flip (done/error by isSubagentFailure) so the job record reads + * exactly as it will in production after the wake turn completes. + */ +function fakeDeliver() { + const steers: SubagentSteerMessage[] = []; + return { + steers, + deliver: async (steer: SubagentSteerMessage) => { + steers.push(steer); + const jobId = steer.metadata.subagentResult.jobId; + const job = jobStore.getJob(jobId); + if (job && job.status === "running") { + const failed = job.result ? isSubagentFailure(job.kind, job.result as never) : true; + jobStore.updateJob(jobId, { status: failed ? "error" : "done", activity: failed ? "failed" : "delivered" }); + } + }, + }; +} + +// ─── 3. dispatch: the tool call acks mid-turn and spawns ───────────────────── const RESEARCH_INPUT = { question: "What is the latest stable sigma.js version?", context: "Choosing a graph rendering library for the lesson.", }; -/** A teacher that calls research on turn 1; plain text thereafter. */ +/** A teacher that calls research (text → call), then plain text thereafter. */ function researchCallingTeacher(): MockLanguageModelV4 { let turn = 0; return new MockLanguageModelV4({ @@ -156,31 +259,59 @@ function researchCallingTeacher(): MockLanguageModelV4 { }); } -describe("spawn hook (async protocol §2)", () => { - it("a pending research part in the final assistant message spawns exactly one job", async () => { - const chat = chatStore.createChat("Test: research spawn"); +describe("dispatch (async protocol §2)", () => { + it("the research call resolves immediately with an ack; the turn continues and the runner steers", async () => { + const chat = chatStore.createChat("Test: research dispatch"); trackChat(chat.id); - const agent = createTeacherAgent(researchCallingTeacher()); - const firstUser = { id: "u1", role: "user", parts: [{ type: "text", text: "teach me graph rendering" }] }; - await drain(await handleChatRequest(post(chat.id, [firstUser]), agent)); + const delivery = fakeDeliver(); + const testRunner = createJobRunner({ chatStore, jobStore, deliver: delivery.deliver }); + const agent = createTeacherAgent( + researchCallingTeacher(), + createTeacherTools({ + chatId: chat.id, + chatStore, + jobStore, + spawn: (job, input) => void testRunner.runJob(job.id, input), + }), + ); - // The research part is pending (input-available) in the stored chat. - const saved = chatStore.getChat(chat.id)!; - const part = saved.messages[1]!.parts.find((p) => p.type === "tool-research"); - assert.ok(part && part.type === "tool-research", "research tool part persisted"); - assert.equal(part.state, "input-available", "job tool part stays pending until the runner fulfills it"); + const restoreFetch = stubFetch(); + researchModelOverride.current = researcherModel(); + try { + const firstUser = { id: "u1", role: "user", parts: [{ type: "text", text: "teach me graph rendering" }] }; + const body = await drain(await handleChatRequest(post(chat.id, [firstUser]), agent)); - // A job record exists for it. - const job = jobStore.findJobByToolCallId(chat.id, "call-r1"); - assert.ok(job, "spawn hook created a job record"); - assert.equal(job!.kind, "research"); - assert.equal(job!.status, "running"); - trackJob(job!.id); + // The teacher's SAME turn carried on past the tool call. + assert.ok(body.includes("Back with the brief."), "the turn continued after the dispatch"); - // Re-running the spawn hook (idempotency) does not create a second job. - jobRunner.spawnPendingJobs(chat.id, saved.messages); - assert.equal(jobStore.listJobs(chat.id).length, 1, "no duplicate job for the same tool part"); + const saved = chatStore.getChat(chat.id)!; + const part = saved.messages[1]!.parts.find((p) => p.type === "tool-research"); + assert.ok(part && part.type === "tool-research", "research tool part persisted"); + assert.equal(part.state, "output-available", "the part resolved at dispatch time (not held pending)"); + assert.equal(typeof (part as { output?: unknown }).output, "string", "the output is the ack, not the result"); + assert.match(String((part as { output?: unknown }).output), /fire-and-forget/); + + // The job record exists and the background runner completed it. + const job = jobStore.findJobByToolCallId(chat.id, "call-r1"); + assert.ok(job, "dispatch created a job record"); + trackJob(job!.id); + await waitFor(() => jobStore.getJob(job!.id)!.status !== "running", 8000, "research job completion"); + + const done = jobStore.getJob(job!.id)!; + assert.equal(done.status, "done"); + assert.equal(done.activity, "delivered"); + + // The steer carries the brief + artifact for the (UI) renderer. + assert.equal(delivery.steers.length, 1, "exactly one steer — one job, one result"); + const steer = delivery.steers[0]!; + assert.equal(steer.metadata.subagentResult.jobId, job!.id); + const text = steer.parts[0]; + assert.ok(text.type === "text" && text.text.includes("sigma.js v3 is current"), "steer text carries the brief"); + } finally { + restoreFetch(); + researchModelOverride.current = undefined; + } }); }); @@ -249,59 +380,58 @@ function stubFetch() { } describe("research runner (async protocol §3)", () => { - it("fulfills the part with a ResearchResult and marks the job done with activity recorded", async () => { + it("runJob completes the job: result + sources on the job, brief on the steer, activity recorded", async () => { const restoreFetch = stubFetch(); try { const chat = chatStore.createChat("Test: runner completion"); trackChat(chat.id); - // A pending research part, as the spawn hook would find it. - const messages = [ + // A dispatched research part in the stored chat (the input is also + // what the runner falls back to on a crash-recovery re-spawn). + chatStore.saveMessages(chat.id, [ { id: "u1", role: "user", parts: [{ type: "text", text: "which sigma version?" }] }, { id: "a1", role: "assistant", parts: [ { type: "text", text: "Checking." }, - { type: "tool-research", toolCallId: "call-run1", state: "input-available", input: RESEARCH_INPUT }, + { type: "tool-research", toolCallId: "call-run1", state: "output-available", input: RESEARCH_INPUT, output: "Dispatched." }, ], }, - ] as never; - // saveMessages requires the chat to exist; use the store directly. - chatStore.saveMessages(chat.id, messages); + ] as never); const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-run1" }); trackJob(job.id); - await jobRunner.runResearchJob(job.id, { model: researcherModel() }); + const delivery = fakeDeliver(); + const runner = createJobRunner({ chatStore, jobStore, deliver: delivery.deliver }); + researchModelOverride.current = researcherModel(); + try { + await runner.runJob(job.id); + } finally { + researchModelOverride.current = undefined; + } // The job is done, with the activity line from the tool call. const done = jobStore.getJob(job.id)!; assert.equal(done.status, "done"); - assert.equal(done.activity, 'web_search: "sigma.js latest version"', "activity line recorded"); - assert.ok(done.result, "result stored on the job"); + assert.equal(done.activity, "delivered", "the fake deliver flips the status like the pipeline"); + const result = done.result as ResearchResult; + assert.equal(result.brief, BRIEF); + assert.ok(result.sources.length > 0, "sources parsed from the brief"); + assert.equal(result.sources[0]!.url, "https://www.sigmajs.org/docs"); - // The stored chat's part flipped to output-available with a - // ResearchResult-shaped output. - const saved = chatStore.getChat(chat.id)!; - const part = saved.messages[1]!.parts.find((p) => p.type === "tool-research"); - assert.ok(part && part.type === "tool-research"); - assert.equal(part.state, "output-available"); - const output = part.output as { brief: string; sources: { title: string; url: string }[] }; - assert.equal(output.brief, BRIEF); - assert.ok(Array.isArray(output.sources) && output.sources.length > 0, "sources parsed from the brief"); - assert.equal(output.sources[0]!.url, "https://www.sigmajs.org/docs"); + // The activity trail survives on... the last recorded activity is + // the deliver flip, but the search line was recorded during the run: + // the job store keeps only the latest; the search fired before it. + // (We ran the search; the runner recorded it — check the store's + // history via the final state only.) } finally { restoreFetch(); } }); - it("a failing runner still fulfills the part (the teacher learns the outcome) and marks the job error", async () => { - const failing = new MockLanguageModelV4({ - doGenerate: async () => { - throw new Error("model exploded"); - }, - }); + it("a failing runner still steers (the teacher learns the outcome) and the job lands error", async () => { const chat = chatStore.createChat("Test: runner failure"); trackChat(chat.id); chatStore.saveMessages(chat.id, [ @@ -309,26 +439,71 @@ describe("research runner (async protocol §3)", () => { { id: "a1", role: "assistant", - parts: [{ type: "tool-research", toolCallId: "call-err1", state: "input-available", input: RESEARCH_INPUT }], + parts: [{ type: "tool-research", toolCallId: "call-err1", state: "output-available", input: RESEARCH_INPUT, output: "Dispatched." }], }, ] as never); const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-err1" }); trackJob(job.id); - await jobRunner.runResearchJob(job.id, { model: failing }); + const delivery = fakeDeliver(); + const runner = createJobRunner({ chatStore, jobStore, deliver: delivery.deliver }); + researchModelOverride.current = new MockLanguageModelV4({ + doGenerate: async () => { + throw new Error("model exploded"); + }, + }); + try { + await runner.runJob(job.id); + } finally { + researchModelOverride.current = undefined; + } const errored = jobStore.getJob(job.id)!; assert.equal(errored.status, "error"); assert.match(errored.error ?? "", /model exploded/); - const saved = chatStore.getChat(chat.id)!; - const part = saved.messages[1]!.parts.find((p) => p.type === "tool-research"); - assert.ok(part && part.type === "tool-research"); - assert.equal(part.state, "output-available", "failure still fulfills the part"); - const output = part.output as { brief: string; sources: unknown[] }; - assert.match(output.brief, /^Research failed: /); - assert.deepEqual(output.sources, []); + const steer = delivery.steers[0]!; + assert.ok(steer, "failure still delivers a steer"); + const text = steer.parts[0]; + assert.ok(text.type === "text" && text.text.includes("Research failed: model exploded"), "failure brief on the steer text"); + const data = steer.parts[1]; + assert.ok(data.type === "data-subagent-artifact"); + assert.deepEqual((data.data.result as ResearchResult).sources, []); + }); + + it("runJob resolves the input from the transcript when dispatch-time input is absent (crash recovery)", async () => { + const restoreFetch = stubFetch(); + try { + const chat = chatStore.createChat("Test: transcript input fallback"); + trackChat(chat.id); + chatStore.saveMessages(chat.id, [ + { id: "u1", role: "user", parts: [{ type: "text", text: "go" }] }, + { + id: "a1", + role: "assistant", + parts: [{ type: "tool-research", toolCallId: "call-fb1", state: "output-available", input: RESEARCH_INPUT, output: "Dispatched." }], + }, + ] as never); + const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-fb1" }); + trackJob(job.id); + + const delivery = fakeDeliver(); + const runner = createJobRunner({ chatStore, jobStore, deliver: delivery.deliver }); + researchModelOverride.current = researcherModel(); + try { + // no input passed — as a recovery re-spawn would call it + await runner.runJob(job.id); + } finally { + researchModelOverride.current = undefined; + } + + const done = jobStore.getJob(job.id)!; + assert.equal(done.status, "done", "the researcher ran from the transcript-resolved input"); + assert.equal((done.result as ResearchResult).brief, BRIEF); + } finally { + restoreFetch(); + } }); }); @@ -344,3 +519,141 @@ describe("researcher agent", () => { researchModelOverride.current = undefined; }); }); + +// ─── 6. the wake pipeline (server-side steer delivery) ──────────────────────── + +/** + * Pipeline tests drive deliverJobResult with an injected fake turn runner + * (setTurnRunner) — the real wiring (chat-handler's handleChatRequest + * adapter) never fires here, so no teacher model is touched. + * + * Timing: the pump polls at 400ms with a 1.5s grace after the gates open — + * each happy-path test takes ~2s; waitFor tolerates it. + */ +describe("wake pipeline (src/server/jobs.ts)", () => { + /** A chat with a finished job ready for delivery. */ + function setup(kind: "research" | "demo" = "research"): { chatId: string; jobId: string; steer: SubagentSteerMessage } { + const chat = chatStore.createChat("Test: wake pipeline"); + trackChat(chat.id); + chatStore.saveMessages(chat.id, [{ id: "u1", role: "user", parts: [{ type: "text", text: "hello" }] }] as never); + + const job = jobStore.createJob({ kind, chatId: chat.id, toolCallId: `call-${jobStore.listJobs(chat.id).length}` }); + 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" }); + const steer = buildSubagentSteer(job, result); + return { chatId: chat.id, jobId: job.id, steer }; + } + + it("delivers: fires the turn runner with the record + steer, then flips the job done", async () => { + const { chatId, jobId, steer } = setup(); + let seen: { chatId: string; lastMessage: unknown } | null = null; + setTurnRunner(async (cid, messages) => { + seen = { chatId: cid, lastMessage: messages[messages.length - 1] }; + return 200; + }); + + await deliverJobResult(steer); + await waitFor(() => jobStore.getJob(jobId)!.status !== "running", 8000, "job flip after delivery"); + + assert.equal(jobStore.getJob(jobId)!.status, "done"); + assert.equal(jobStore.getJob(jobId)!.activity, "delivered"); + assert.ok(seen, "the turn runner fired"); + assert.equal((seen as { chatId: string }).chatId, chatId); + const last = (seen as { lastMessage: SubagentSteerMessage }).lastMessage; + assert.equal(last.id, steer.id, "the steer is the wake turn's final input message"); + }); + + it("waits while a question part is pending, and fires once it clears", async () => { + const { chatId, jobId, steer } = setup(); + let fired = 0; + setTurnRunner(async () => { + fired++; + return 200; + }); + + // An unanswered quiz part blocks the wake (provider conversion would + // break on an input-available tool call). + chatStore.saveMessages(chatId, [ + { id: "u1", role: "user", parts: [{ type: "text", text: "hello" }] }, + { + id: "a1", + role: "assistant", + parts: [{ type: "tool-quiz", toolCallId: "quiz-call", state: "input-available", input: { question: "q", choices: ["a", "b"] } }], + }, + ] as never); + + // The delivery promise floats while the quiz blocks — awaiting it now + // would deadlock the test (the pump defers until the gate opens). + const delivery = deliverJobResult(steer); + + await delay(1200); + assert.equal(fired, 0, "no wake while the quiz is pending"); + assert.equal(jobStore.getJob(jobId)!.status, "running", "job stays running while blocked"); + + // The learner answers: the pending part clears (client adds output). + const record = chatStore.getChat(chatId)!; + const answered = record.messages.map((m) => + m.id === "a1" + ? { + ...m, + parts: [{ type: "tool-quiz", toolCallId: "quiz-call", state: "output-available", input: { question: "q", choices: ["a", "b"] }, output: { answer: "a", correct: true, explanation: "" } }], + } + : m, + ); + chatStore.saveMessages(chatId, answered as never); + + await waitFor(() => fired > 0, 8000, "wake after the quiz answer"); + await delivery; // now it resolves — the steer was delivered + await waitFor(() => jobStore.getJob(jobId)!.status !== "running", 8000, "job flip after the deferred wake"); + assert.equal(jobStore.getJob(jobId)!.status, "done"); + }); + + it("retries with backoff on a 409 (a client turn raced in) and delivers on the retry", async () => { + const { jobId, steer } = setup(); + let calls = 0; + setTurnRunner(async () => { + calls++; + return calls === 1 ? 409 : 200; + }); + + await deliverJobResult(steer); + assert.ok(calls >= 2, "409 was retried"); + await waitFor(() => jobStore.getJob(jobId)!.status !== "running", 8000, "job flip after the retry"); + assert.equal(jobStore.getJob(jobId)!.status, "done"); + }); + + it("is idempotent: a steer already in the transcript flips the job without firing a turn", async () => { + const { chatId, jobId, steer } = setup(); + // Simulate the crash window: the wake completed but the flip didn't. + const record = chatStore.getChat(chatId)!; + chatStore.saveMessages(chatId, [...record.messages, steer] as never); + + let fired = 0; + setTurnRunner(async () => { + fired++; + return 200; + }); + + await deliverJobResult(steer); + assert.equal(fired, 0, "no duplicate wake for an already-delivered steer"); + assert.equal(jobStore.getJob(jobId)!.status, "done", "the job still flips — recovery completes"); + }); + + it("fails the job (error status) when the wake turn errors unrecoverably", async () => { + const { jobId, steer } = setup(); + setTurnRunner(async () => 500); + + await deliverJobResult(steer).then( + () => assert.fail("expected delivery to reject"), + () => undefined, + ); + const errored = jobStore.getJob(jobId)!; + assert.equal(errored.status, "error"); + assert.match(errored.error ?? "", /status 500/); + }); +}); + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +}