From 59e1fee8502ac8a7d304fd97f9036fa488ea1a4d Mon Sep 17 00:00:00 2001 From: Amos Date: Mon, 31 Aug 2026 22:16:03 +0300 Subject: [PATCH] Turn services + agent DI: chat-handler becomes the composition root (stage 4) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The chat turn is now composed, not implemented, by chat-handler.ts: - src/server/turn-service.ts: one named, error-contained seam per turn responsibility — buildTeacherContext (W8 read channel), persistTurn (canonical state write), spawnTurnJobs (async protocol kick-off), foldTurnIntoGraph (session→graph consolidation; the extraction-pipeline stage swaps this body behind the same seam). Each takes its deps explicitly; failures log one [chat ] line and never touch the response. - chat-handler.ts wires: storage ports (stores), the job runner, the teacher tools (ctx = chatId + injected ChatStore), and the services. It no longer contains any logic beyond request validation. Agent layer now receives dependencies instead of importing them — a new lint zone enforces it (src/agent never imports src/server or src/store; verified to fire both ways): - runner.ts: createJobRunner({ chatStore, jobStore }) factory returning the spawn hook + the three job runners; the model-override test seam stays module-level. Bodies otherwise verbatim. - plan tools: persist through the ctx-injected ChatStore (TeacherToolContext = { chatId, chatStore }). - teacher-agent's dependency-free default registry is now the true client-side set: questionTools + jobTools (jobTools was ctx-parameter- shaped but never used its ctx — now honestly context-free). Production always passes createTeacherTools({ chatId, chatStore }); tests that drive a bare mock model get the client-side set unchanged. Zero behavior change: tsc clean, 190/190 tests, eslint clean, production build clean. Tests inject the same fs ports the chat-handler does. --- README.md | 6 +- eslint.config.mjs | 9 + src/agent/subagents/runner.ts | 414 ++++++++++++++++++---------------- src/agent/teacher-agent.ts | 11 +- src/agent/tools.ts | 14 +- src/agent/tools/jobs.ts | 7 +- src/agent/tools/plan.ts | 22 +- src/server/chat-handler.ts | 63 ++---- src/server/turn-service.ts | 81 +++++++ tests/demos.test.tsx | 13 +- tests/jobs.test.ts | 11 +- tests/plan-tree.test.ts | 14 +- 12 files changed, 383 insertions(+), 282 deletions(-) create mode 100644 src/server/turn-service.ts diff --git a/README.md b/README.md index 306bf90..0fec74c 100644 --- a/README.md +++ b/README.md @@ -58,7 +58,7 @@ graph TB end subgraph TEACHER["Teacher agent (src/agent)"] - CH["chat-handler.ts (framework-free)
graph context · persist · spawn jobs · emit"] + CH["chat-handler.ts — turn composition root
(services: context · persist · spawn · fold)"] TA["ToolLoopAgent
system-prompt.md + prompts/*.md fragments
+ graph-context block"] QT["quiz · ask_user_question
(client-side tools, no execute)"] PT["plan_* tools (server-side,
mutate the chat's tree)"] @@ -125,7 +125,9 @@ src/ tools/jobs.ts research / demo / visual tool surfaces subagents/ researcher · demo-builder · visual-maker, runner (async protocol), headless Chrome helper - server/chat-handler.ts one chat turn: stream, persist, spawn, emit + server/chat-handler.ts one chat turn: compose agent + services, stream + server/turn-service.ts turn services: teacher context · persist · + spawn jobs · fold-into-graph (error-contained) server/stores.ts THE composition root — the one place adapters wire in core/contracts.ts THE contracts file — types + protocol docs core/ports.ts storage ports — ChatStore · GraphStore · JobStore diff --git a/eslint.config.mjs b/eslint.config.mjs index 1e5d7d0..1a6aad8 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -12,6 +12,9 @@ import importPlugin from "eslint-plugin-import"; * app/** → "@/src/core" + "@/src/server" (the server barrel) — ONLY * src/core/** → never imports src/agent, src/server, src/store (pure * core: adapters implement core/ports.ts, never the reverse) + * src/agent/** → never imports src/server, src/store (dependencies — + * the storage ports, the composition — arrive via + * factories/closures from the chat-handler) * * tests/ and scripts/ are internal — unrestricted. */ @@ -55,6 +58,12 @@ export default defineConfig([ message: "Pure core never imports IO layers (agent/server/store): adapters implement core/ports.ts, and the composition root (src/server/stores.ts) passes them in (architecture: docs/architecture.md).", }, + { + target: "./src/agent", + from: ["./src/server", "./src/store"], + message: + "The agent layer receives its dependencies (storage ports via factories/closures) — it never imports the server layer or store adapters (architecture: docs/architecture.md).", + }, ], }, ], diff --git a/src/agent/subagents/runner.ts b/src/agent/subagents/runner.ts index c640641..ae9aa0a 100644 --- a/src/agent/subagents/runner.ts +++ b/src/agent/subagents/runner.ts @@ -1,22 +1,22 @@ 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 { chatStore } from "../../store/chats.ts"; -import { jobStore } from "../../store/jobs.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 + the chat-handler spawn hook (contracts.ts - * "THE ASYNC PROTOCOL"). + * 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. * - * spawnPendingJobs — called from onFinish AFTER saveMessages: finds + * 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 chat-handler needs no changes for new - * job kinds — anything listed in JOB_TOOLS routes automatically. + * self-contained map here, so no caller changes for new job kinds — + * anything listed in JOB_TOOLS routes automatically. * * runResearchJob / runDemoJob / runVisualJob — run the sub-agent loop in * the background, updating the job's activity line on every iteration, @@ -40,186 +40,236 @@ export const researchModelOverride: { current?: LanguageModel } = {}; export const demoModelOverride: { current?: LanguageModel } = {}; export const visualModelOverride: { current?: LanguageModel } = {}; +export interface JobRunnerDeps { + chatStore: ChatStore; + jobStore: JobStore; +} + export interface RunResearchJobOptions { model?: LanguageModel; } -/** 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 }), -}; - -export 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" }); - }); - } -} - -export async function runResearchJob(jobId: string, opts: RunResearchJobOptions = {}): 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 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; - - const prompt = [ - `Research question: ${question}`, - context ? `Context (what this is needed for): ${context}` : null, - "Produce your final brief in the required format.", - ] - .filter(Boolean) - .join("\n\n"); - - const fetchedUrls: string[] = []; - const agent = createResearcherAgent(opts.model); - - try { - jobStore.updateJob(jobId, { activity: "researching" }); - const result = await agent.generate({ - prompt, - onToolExecutionStart: ({ toolCall }) => { - const input = toolCall.input as { query?: string; url?: string } | undefined; - if (toolCall.toolName === "web_fetch" && typeof input?.url === "string") { - fetchedUrls.push(input.url); - } - jobStore.updateJob(jobId, { activity: activityLine(toolCall.toolName, input) }); - }, - }); - - 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 - } 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 }); - } -} - -// ─── demo + visual job wrappers ─────────────────────────────────────────── - export interface RunDemoJobOptions { model?: LanguageModel; } -/** 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; -} - -export 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 title = typeof input?.title === "string" && input.title.trim() ? input.title : "Interactive demo"; - const brief = typeof input?.brief === "string" ? input.brief : ""; - - try { - jobStore.updateJob(jobId, { activity: "building demo" }); - const result = await runDemoBuilder( - { title, brief }, - { - model: opts.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 }); - } - } 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" }); - } -} - export interface RunVisualJobOptions { model?: LanguageModel; } -export async function runVisualJob(jobId: string, opts: RunVisualJobOptions = {}): Promise { - const job = jobStore.getJob(jobId); - if (!job || job.status !== "running") return; +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; +} - const chat = chatStore.getChat(job.chatId); - if (!chat) { - jobStore.updateJob(jobId, { status: "error", error: `Chat not found: ${job.chatId}`, activity: "failed" }); - return; +export function createJobRunner(deps: JobRunnerDeps): JobRunner { + const { chatStore, jobStore } = 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 }), + }; + + 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" }); + }); + } } - const input = findToolPart(chat.messages, job.toolCallId)?.input as VisualToolInput | undefined; - const brief = typeof input?.brief === "string" ? input.brief : ""; - try { - jobStore.updateJob(jobId, { activity: "drawing figure" }); - const result = await runVisualMaker( - { brief }, - { - model: opts.model, - onActivity: (line) => jobStore.updateJob(jobId, { activity: line }), - }, + async function runResearchJob(jobId: string, opts: RunResearchJobOptions = {}): 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 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; + + const prompt = [ + `Research question: ${question}`, + context ? `Context (what this is needed for): ${context}` : null, + "Produce your final brief in the required format.", + ] + .filter(Boolean) + .join("\n\n"); + + const fetchedUrls: string[] = []; + const agent = createResearcherAgent(opts.model); + + try { + jobStore.updateJob(jobId, { activity: "researching" }); + const result = await agent.generate({ + prompt, + onToolExecutionStart: ({ toolCall }) => { + const input = toolCall.input as { query?: string; url?: string } | undefined; + if (toolCall.toolName === "web_fetch" && typeof input?.url === "string") { + fetchedUrls.push(input.url); + } + jobStore.updateJob(jobId, { activity: activityLine(toolCall.toolName, input) }); + }, + }); + + 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 + } 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 }); + } + } + + // ─── demo + visual job wrappers ─────────────────────────────────────────── + + /** 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; + } + + 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 title = typeof input?.title === "string" && input.title.trim() ? input.title : "Interactive demo"; + const brief = typeof input?.brief === "string" ? input.brief : ""; + + try { + jobStore.updateJob(jobId, { activity: "building demo" }); + const result = await runDemoBuilder( + { title, brief }, + { + model: opts.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 }); + } + } 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" }); + } + } + + async function runVisualJob(jobId: string, opts: RunVisualJobOptions = {}): 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 VisualToolInput | undefined; + const brief = typeof input?.brief === "string" ? input.brief : ""; + + try { + jobStore.updateJob(jobId, { activity: "drawing figure" }); + const result = await runVisualMaker( + { brief }, + { + model: opts.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 }); + } + } 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" }); + } + } + + // ─── 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; + }), + }, ); - 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 }); - } - } 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" }); - } -} - -// ─── 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; + if (!found) return; + try { + chatStore.saveMessages(chatId, messages); + } catch (err) { + console.error(`[chat ${chatId}] fulfilling job part ${toolCallId} failed:`, err); } } - return null; + + return { spawnPendingJobs, runResearchJob, runDemoJob, runVisualJob }; } +// ─── module-level pure helpers ────────────────────────────────────────────── + /** 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 ?? ""}"`; @@ -236,36 +286,6 @@ function hostOf(url: string | undefined): string { } } -/** - * 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); - } -} - /** Sources for the result: markdown links cited in the brief, else fetched URLs. */ function parseSources(brief: string, fetchedUrls: string[]): ResearchResult["sources"] { const sources: ResearchResult["sources"] = []; diff --git a/src/agent/teacher-agent.ts b/src/agent/teacher-agent.ts index 6687759..3d3f4eb 100644 --- a/src/agent/teacher-agent.ts +++ b/src/agent/teacher-agent.ts @@ -3,7 +3,7 @@ import * as path from "node:path"; import { ToolLoopAgent, stepCountIs, type LanguageModel, type ToolSet } from "ai"; import { openrouter } from "@openrouter/ai-sdk-provider"; import { PATHS } from "../../config.ts"; -import { createTeacherTools } from "./tools.ts"; +import { createTeacherTools, jobTools, questionTools } from "./tools.ts"; /** * The teacher agent. Everything pi used to provide (agent loop, tool calling, @@ -62,10 +62,11 @@ export function createTeacherAgent(model?: LanguageModel, tools?: ToolSet, graph reasoning: { enabled: true, effort: "medium" }, }), instructions: withGraphContext(graphContext), - // Tools are per-request (plan tools close over chatId); chat-handler - // passes createTeacherTools({ chatId }). Tests pass a mock model and get - // the default question-only set. - tools: tools ?? createTeacherTools({ chatId: "" }), + // Tools are per-request (plan tools close over the injected ctx); + // chat-handler passes createTeacherTools({ chatId, chatStore }). 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. stopWhen: stepCountIs(20), diff --git a/src/agent/tools.ts b/src/agent/tools.ts index 1cdbcc2..e8a0028 100644 --- a/src/agent/tools.ts +++ b/src/agent/tools.ts @@ -1,6 +1,9 @@ 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"; + +export { jobTools }; import { planTools } from "./tools/plan.ts"; /** @@ -43,16 +46,21 @@ export const questionTools = { 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; } /** - * Per-request tool registry. Plan tools close over chatId for persistence; - * question and job tools are context-free. + * Per-request tool registry. Plan tools close over the ctx (chatId + the + * injected ChatStore) for persistence; question and job tools are + * context-free. */ export function createTeacherTools(ctx: TeacherToolContext): ToolSet { return { ...questionTools, ...planTools(ctx), - ...jobTools(ctx), + ...jobTools(), }; } diff --git a/src/agent/tools/jobs.ts b/src/agent/tools/jobs.ts index bf3b8e2..78d3d20 100644 --- a/src/agent/tools/jobs.ts +++ b/src/agent/tools/jobs.ts @@ -1,6 +1,5 @@ import { tool, type ToolSet } from "ai"; import { z } from "zod"; -import type { TeacherToolContext } from "../tools.ts"; /** * Async sub-agent job tools (W2: research; W6 adds demo + visual). @@ -72,10 +71,10 @@ export const visualTool = tool({ /** * Per-request job tool registry. Context-free (the runner resolves - * everything from the chat record by toolCallId) — the ctx parameter keeps - * the createTeacherTools factory shape stable. + * everything from the chat record by toolCallId). Takes no dependencies — + * part of the dependency-free default registry in teacher-agent.ts. */ -export function jobTools(ctx: TeacherToolContext): ToolSet { +export function jobTools(): ToolSet { return { research: researchTool, demo: demoTool, diff --git a/src/agent/tools/plan.ts b/src/agent/tools/plan.ts index 2efd281..01b88f5 100644 --- a/src/agent/tools/plan.ts +++ b/src/agent/tools/plan.ts @@ -13,7 +13,6 @@ import { setCurrent, updateNode, } from "../../core/plan-tree.ts"; -import { chatStore } from "../../store/chats.ts"; import type { TeacherToolContext } from "../tools.ts"; /** @@ -49,9 +48,10 @@ const edgeSchema = z.object({ * empty) tree, so the model can see what it's working with and fix the call. */ async function runPlanOp( - chatId: string, + ctx: TeacherToolContext, op: (tree: PlanTree) => PlanTree, ): Promise { + const { chatId, chatStore } = ctx; const record = chatStore.getChat(chatId); if (!record) return { ok: false, error: `Chat not found: ${chatId}`, tree: emptyTree() }; const current = record.tree ?? emptyTree(); @@ -81,7 +81,7 @@ export function planTools(ctx: TeacherToolContext): ToolSet { .describe("Initial `requires` edges between the nodes. Edge direction: from = prerequisite, to = dependent."), }), execute: async ({ nodes, edges }): Promise => { - const record = chatStore.getChat(ctx.chatId); + const record = ctx.chatStore.getChat(ctx.chatId); if (!record) return { ok: false, error: `Chat not found: ${ctx.chatId}`, tree: emptyTree() }; const existing = record.tree; if (existing && existing.nodes.length > 0) { @@ -91,7 +91,7 @@ export function planTools(ctx: TeacherToolContext): ToolSet { tree: existing, }; } - return runPlanOp(ctx.chatId, () => createTree(nodes, edges ?? [])); + return runPlanOp(ctx, () => createTree(nodes, edges ?? [])); }, }), @@ -109,7 +109,7 @@ export function planTools(ctx: TeacherToolContext): ToolSet { .describe("Ids of prerequisite nodes — each gets a `requires` edge pointing to this node. All must already exist."), }), execute: async ({ id, title, summary, parents }): Promise => - runPlanOp(ctx.chatId, (tree) => addNode(tree, { id, title, ...(summary !== undefined ? { summary } : {}) }, parents)), + runPlanOp(ctx, (tree) => addNode(tree, { id, title, ...(summary !== undefined ? { summary } : {}) }, parents)), }), plan_update_node: tool({ @@ -125,7 +125,7 @@ export function planTools(ctx: TeacherToolContext): ToolSet { message: "Pass at least one of `title` or `summary` — an update with neither changes nothing.", }), execute: async ({ id, title, summary }): Promise => - runPlanOp(ctx.chatId, (tree) => updateNode(tree, id, { ...(title !== undefined ? { title } : {}), ...(summary !== undefined ? { summary } : {}) })), + runPlanOp(ctx, (tree) => updateNode(tree, id, { ...(title !== undefined ? { title } : {}), ...(summary !== undefined ? { summary } : {}) })), }), plan_remove_node: tool({ @@ -133,7 +133,7 @@ export function planTools(ctx: TeacherToolContext): ToolSet { "Remove a node from the plan tree — prune what proves unnecessary. Cascades: every edge touching the node (both directions), its current " + "status, and its done status are removed with it.", inputSchema: z.object({ id: nodeId }), - execute: async ({ id }): Promise => runPlanOp(ctx.chatId, (tree) => removeNode(tree, id)), + execute: async ({ id }): Promise => runPlanOp(ctx, (tree) => removeNode(tree, id)), }), plan_add_edge: tool({ @@ -144,7 +144,7 @@ export function planTools(ctx: TeacherToolContext): ToolSet { from: nodeId.describe("Prerequisite node id (must be understood before `to`)."), to: nodeId.describe("Dependent node id (requires `from`)."), }), - execute: async ({ from, to }): Promise => runPlanOp(ctx.chatId, (tree) => addEdge(tree, from, to)), + execute: async ({ from, to }): Promise => runPlanOp(ctx, (tree) => addEdge(tree, from, to)), }), plan_remove_edge: tool({ @@ -154,7 +154,7 @@ export function planTools(ctx: TeacherToolContext): ToolSet { from: nodeId.describe("Prerequisite end of the edge to remove."), to: nodeId.describe("Dependent end of the edge to remove."), }), - execute: async ({ from, to }): Promise => runPlanOp(ctx.chatId, (tree) => removeEdge(tree, from, to)), + execute: async ({ from, to }): Promise => runPlanOp(ctx, (tree) => removeEdge(tree, from, to)), }), plan_set_current: tool({ @@ -162,7 +162,7 @@ export function planTools(ctx: TeacherToolContext): ToolSet { "Set the CURRENT node — where the lesson is right now. Call it the moment you START teaching a node, so the learner sees their position " + "on the map. The node must exist. This marks position only; it does not mark anything done.", inputSchema: z.object({ id: nodeId.describe("The node you are starting to teach now.") }), - execute: async ({ id }): Promise => runPlanOp(ctx.chatId, (tree) => setCurrent(tree, id)), + execute: async ({ id }): Promise => runPlanOp(ctx, (tree) => setCurrent(tree, id)), }), plan_mark_done: tool({ @@ -170,7 +170,7 @@ export function planTools(ctx: TeacherToolContext): ToolSet { "Mark a node DONE — and only when the learner has DEMONSTRATED it (a correct check), not when you finished explaining it. " + "The node must exist. Idempotent.", inputSchema: z.object({ id: nodeId.describe("The node the learner just demonstrated.") }), - execute: async ({ id }): Promise => runPlanOp(ctx.chatId, (tree) => markDone(tree, id)), + execute: async ({ id }): Promise => runPlanOp(ctx, (tree) => markDone(tree, id)), }), }; } diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index 9937e8b..ea0bdb7 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -1,10 +1,9 @@ import { createAgentUIStreamResponse, type UIMessage } from "ai"; import { createTeacherAgent, type TeacherAgent } from "../agent/teacher-agent.ts"; +import { createJobRunner } from "../agent/subagents/runner.ts"; import { createTeacherTools } from "../agent/tools.ts"; -import { spawnPendingJobs } from "../agent/subagents/runner.ts"; -import { emitChat, getGraph } from "../core/graph.ts"; -import { buildGraphContext } from "../core/graph-context.ts"; import { stores } from "./stores.ts"; +import { buildTeacherContext, foldTurnIntoGraph, persistTurn, spawnTurnJobs } from "./turn-service.ts"; /** * One chat turn: run the teacher agent over the client's message list and @@ -12,9 +11,14 @@ import { stores } from "./stores.ts"; * so tests can drive it directly with a scripted mock model — the Next route * handler in app/api/chat is a one-line wrapper. * - * The client owns the message list between turns (useChat); the server - * persists the canonical state (client messages + agent response) onFinish. + * 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. */ + +const jobRunner = createJobRunner(stores); + export async function handleChatRequest(req: Request, agent?: TeacherAgent): Promise { let body: { chatId?: unknown; messages?: unknown }; try { @@ -30,48 +34,19 @@ export async function handleChatRequest(req: Request, agent?: TeacherAgent): Pro return Response.json({ error: "messages must be an array" }, { status: 400 }); } - // W8 (vision §1 "the graph as session context"): the learner's knowledge - // graph rides along as part of the teacher's instructions. Built per - // request — stateless, and self-updating as the graph grows mid-session - // (the emission hook below folds each turn). A read failure must never - // take the lesson down: the block is context, not a dependency. - let graphContext: string | null = null; - try { - graphContext = buildGraphContext(getGraph(stores.graphStore)); - } catch (err) { - console.error(`[chat ${chatId}] graph context build failed:`, err); - } - return createAgentUIStreamResponse({ - agent: agent ?? createTeacherAgent(undefined, createTeacherTools({ chatId }), graphContext), + agent: + agent ?? + createTeacherAgent( + undefined, + createTeacherTools({ chatId, chatStore: stores.chatStore }), + buildTeacherContext(stores.graphStore, chatId), + ), uiMessages: body.messages as UIMessage[], onFinish: ({ messages, isAborted }) => { - // Persist even when aborted: the partial transcript (including any - // pending question part) is what makes resume work. - try { - stores.chatStore.saveMessages(chatId, messages as UIMessage[]); - } catch (err) { - console.error(`[chat ${chatId}] persist failed:`, err, { isAborted }); - } - // 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. - try { - spawnPendingJobs(chatId, messages as UIMessage[]); - } catch (err) { - console.error(`[chat ${chatId}] job spawn failed:`, err); - } - // W4: fold this chat into the knowledge-graph ledgers (tree nodes / - // edges, quiz outcomes, taught events) — synchronous but tiny - // (append-only JSONL), and errors never touch the response. - // Skipped under tests: scripted chats are fixtures, not knowledge. - if (process.env.NODE_ENV !== "test") { - try { - emitChat(chatId, stores); - } catch (err) { - console.error(`[chat ${chatId}] graph emission failed:`, err); - } - } + persistTurn(stores.chatStore, chatId, messages as UIMessage[], isAborted); + spawnTurnJobs(jobRunner, chatId, messages as UIMessage[]); + foldTurnIntoGraph(stores, chatId); }, }); } diff --git a/src/server/turn-service.ts b/src/server/turn-service.ts new file mode 100644 index 0000000..009ea86 --- /dev/null +++ b/src/server/turn-service.ts @@ -0,0 +1,81 @@ +/** + * The chat-turn services — one named, error-contained seam per turn + * responsibility, extracted from the chat-handler so it composes instead of + * implementing. Each takes its dependencies explicitly (storage ports, the + * job runner) — no module here imports an adapter. + * + * buildTeacherContext — graph → instructions block (the W8 read channel) + * persistTurn — the canonical state write (client + agent messages) + * spawnTurnJobs — the async sub-agent protocol kick-off + * foldTurnIntoGraph — session → graph consolidation (the write channel; + * the extraction-pipeline stage swaps this body for + * the extractor/auditor/merger trigger behind the + * same seam) + * + * Logs are the minimal-observability baseline: one line per failure, tagged + * [chat ] — structured logging lands behind these seams later. + */ + +import type { UIMessage } from "ai"; +import { buildGraphContext } from "../core/graph-context.ts"; +import { emitChat, getGraph, type GraphDeps } from "../core/graph.ts"; +import type { ChatStore, GraphStore } from "../core/ports.ts"; +import type { JobRunner } from "../agent/subagents/runner.ts"; + +/** + * W8 (vision §1 "the graph as session context"): the learner's knowledge + * graph rides along as part of the teacher's instructions. Built per request + * — stateless, and self-updating as the graph grows mid-session. A read + * failure must never take the lesson down: the block is context, not a + * dependency. + */ +export function buildTeacherContext(graphStore: GraphStore, chatId: string): string | null { + try { + return buildGraphContext(getGraph(graphStore)); + } catch (err) { + console.error(`[chat ${chatId}] graph context build failed:`, err); + return null; + } +} + +/** + * The canonical state write: the client owns the message list between turns + * (useChat); this persists client messages + the agent response. Persists + * even when aborted — the partial transcript (including any pending + * question part) is what makes resume work. + */ +export function persistTurn(chatStore: ChatStore, chatId: string, messages: UIMessage[], isAborted: boolean): void { + try { + chatStore.saveMessages(chatId, messages); + } catch (err) { + console.error(`[chat ${chatId}] persist failed:`, err, { isAborted }); + } +} + +/** + * 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); + } +} + +/** + * W4: fold this chat into the knowledge-graph ledgers (tree nodes / edges, + * quiz outcomes, taught events) — synchronous but tiny (append-only JSONL), + * and errors never touch the response. Skipped under tests: scripted chats + * are fixtures, not knowledge. + */ +export function foldTurnIntoGraph(deps: GraphDeps, chatId: string): void { + if (process.env.NODE_ENV === "test") return; + try { + emitChat(chatId, deps); + } catch (err) { + console.error(`[chat ${chatId}] graph emission failed:`, err); + } +} diff --git a/tests/demos.test.tsx b/tests/demos.test.tsx index 984e6bc..5db442e 100644 --- a/tests/demos.test.tsx +++ b/tests/demos.test.tsx @@ -41,10 +41,13 @@ import { sanitizeSvg, VisualCard } from "../components/artifacts/visual-card.tsx import { demoInputSchema, jobTools, visualInputSchema } from "../src/agent/tools/jobs.ts"; import { runDemoBuilder } from "../src/agent/subagents/demo-builder.ts"; import { headlessOverride, type DemoRunResult, type HeadlessApi } from "../src/agent/subagents/headless.ts"; -import { demoModelOverride, runDemoJob, runVisualJob, spawnPendingJobs, visualModelOverride } from "../src/agent/subagents/runner.ts"; +import { createJobRunner, demoModelOverride, visualModelOverride } from "../src/agent/subagents/runner.ts"; import { runVisualMaker } from "../src/agent/subagents/visual-maker.ts"; import { 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"; import { artifactRenderers } from "../components/artifacts/index.ts"; @@ -464,7 +467,7 @@ describe("demo/visual job runners", () => { // runDemoJob has no headless injection point — the module override is the seam. headlessOverride.current = fakeHeadless(); try { - await runDemoJob(job.id, { + await jobRunner.runDemoJob(job.id, { model: scriptedModel([ { toolCalls: [{ toolCallId: "w1", toolName: "write_demo", input: { html: GOOD_HTML } }] }, { toolCalls: [{ toolCallId: "r1", toolName: "run_demo", input: {} }] }, @@ -504,7 +507,7 @@ describe("demo/visual job runners", () => { const job = jobStore.createJob({ kind: "visual", chatId: chat.id, toolCallId: "call-v1" }); createdJobs.push(job.id); - await runVisualJob(job.id, { model: scriptedModel([{ text: "cannot" }]) }); + await jobRunner.runVisualJob(job.id, { model: scriptedModel([{ text: "cannot" }]) }); const errored = jobStore.getJob(job.id)!; assert.equal(errored.status, "error"); @@ -588,7 +591,7 @@ describe("demo/visual job runners", () => { assert.equal(part.state, "output-available", "demo part fulfilled by the background runner"); // Idempotency: re-running the spawn hook creates no second job. - spawnPendingJobs(chat.id, saved.messages); + jobRunner.spawnPendingJobs(chat.id, saved.messages); assert.equal(jobStore.listJobs(chat.id).length, 1, "no duplicate job for the same tool part"); } finally { demoModelOverride.current = undefined; @@ -601,7 +604,7 @@ describe("demo/visual job runners", () => { describe("tool + renderer registry wiring", () => { it("demo and visual are CLIENT-SIDE tools with the contract schemas", () => { - const tools = jobTools({ chatId: "c" }); + const tools = jobTools(); 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`); diff --git a/tests/jobs.test.ts b/tests/jobs.test.ts index 7a6ba3e..4e53139 100644 --- a/tests/jobs.test.ts +++ b/tests/jobs.test.ts @@ -26,10 +26,13 @@ import { createTeacherAgent } from "../src/agent/teacher-agent.ts"; import { handleChatRequest } from "../src/server/chat-handler.ts"; import { chatStore } from "../src/store/chats.ts"; import { jobStore } from "../src/store/jobs.ts"; -import { researchModelOverride, runResearchJob, spawnPendingJobs } from "../src/agent/subagents/runner.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[] = []; @@ -176,7 +179,7 @@ describe("spawn hook (async protocol §2)", () => { trackJob(job!.id); // Re-running the spawn hook (idempotency) does not create a second job. - spawnPendingJobs(chat.id, saved.messages); + jobRunner.spawnPendingJobs(chat.id, saved.messages); assert.equal(jobStore.listJobs(chat.id).length, 1, "no duplicate job for the same tool part"); }); }); @@ -270,7 +273,7 @@ describe("research runner (async protocol §3)", () => { const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-run1" }); trackJob(job.id); - await runResearchJob(job.id, { model: researcherModel() }); + await jobRunner.runResearchJob(job.id, { model: researcherModel() }); // The job is done, with the activity line from the tool call. const done = jobStore.getJob(job.id)!; @@ -313,7 +316,7 @@ describe("research runner (async protocol §3)", () => { const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-err1" }); trackJob(job.id); - await runResearchJob(job.id, { model: failing }); + await jobRunner.runResearchJob(job.id, { model: failing }); const errored = jobStore.getJob(job.id)!; assert.equal(errored.status, "error"); diff --git a/tests/plan-tree.test.ts b/tests/plan-tree.test.ts index bbdad7b..a6ba0c5 100644 --- a/tests/plan-tree.test.ts +++ b/tests/plan-tree.test.ts @@ -236,7 +236,7 @@ function newChat(title: string): string { describe("plan tools (execute against the real store)", () => { it("plan_init creates and persists the tree; result is ok with the full tree", async () => { const chatId = newChat("Plan tools: init"); - const tools = planTools({ chatId }); + const tools = planTools({ chatId, chatStore }); const res = await exec(tools, "plan_init", { nodes: [ @@ -258,7 +258,7 @@ describe("plan tools (execute against the real store)", () => { it("plan_init refuses to re-init and never silently wipes the existing tree", async () => { const chatId = newChat("Plan tools: re-init"); - const tools = planTools({ chatId }); + const tools = planTools({ chatId, chatStore }); await exec(tools, "plan_init", { nodes: [{ id: "a", title: "A" }] }); const res = await exec(tools, "plan_init", { @@ -274,7 +274,7 @@ describe("plan tools (execute against the real store)", () => { it("plan_init returns an error result on invalid input (cycle) without persisting", async () => { const chatId = newChat("Plan tools: bad init"); - const tools = planTools({ chatId }); + const tools = planTools({ chatId, chatStore }); const res = await exec(tools, "plan_init", { nodes: [ { id: "a", title: "A" }, @@ -293,7 +293,7 @@ describe("plan tools (execute against the real store)", () => { it("mutation round-trip: add node with parents, set current, mark done — each persists", async () => { const chatId = newChat("Plan tools: round-trip"); - const tools = planTools({ chatId }); + const tools = planTools({ chatId, chatStore }); await exec(tools, "plan_init", { nodes: [ @@ -331,7 +331,7 @@ describe("plan tools (execute against the real store)", () => { it("invalid mutations return ok:false with the current tree and persist nothing", async () => { const chatId = newChat("Plan tools: invalid ops"); - const tools = planTools({ chatId }); + const tools = planTools({ chatId, chatStore }); await exec(tools, "plan_init", { nodes: [{ id: "a", title: "A" }] }); const before = chatStore.getChat(chatId)!.tree!; @@ -353,7 +353,7 @@ describe("plan tools (execute against the real store)", () => { it("mutation tools work on a fresh (uninitialized) chat: the tree seeds from empty", async () => { const chatId = newChat("Plan tools: no init"); - const tools = planTools({ chatId }); + const tools = planTools({ chatId, chatStore }); const res = await exec(tools, "plan_add_node", { id: "a", title: "A" }); assert.equal(res.ok, true); assert.equal(res.tree.nodes.length, 1); @@ -361,7 +361,7 @@ describe("plan tools (execute against the real store)", () => { }); it("unknown chat returns ok:false with the empty tree", async () => { - const tools = planTools({ chatId: "chat_nonexistent" }); + const tools = planTools({ chatId: "chat_nonexistent", chatStore }); const res = await exec(tools, "plan_mark_done", { id: "a" }); assert.equal(res.ok, false); assert.match(res.error!, /Chat not found/);