amosblomqvist
91ae2e137d
With a steer coalesced into a quiz-answer continuation, the steer sits AFTER the assistant message being continued, so the SDK's onFinish list carries the input copy AND the extended response copy as two entries with the SAME id — persistTurn saved both, the jobs poller's syncFromServer imported the dup, and React keyed a single message twice (the user's live test hit it). withMessageIds now dedupes by id keeping the LAST occurrence — the SDK's final merge state (a superset of the earlier copy) — which also yields the right semantic shape: [user, steer, assistant(extended)]. Out-of-lane surgical touch (flagged for master review): turn-service.ts. Squashed the live session's already-corrupted record by hand (same dedupe).
838 lines
34 KiB
TypeScript
838 lines
34 KiB
TypeScript
/**
|
|
* Job infrastructure tests — deterministic, zero network:
|
|
*
|
|
* 1. job store CRUD (create → get → update → list → dedup by toolCallId)
|
|
* 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.
|
|
*/
|
|
|
|
import assert from "node:assert/strict";
|
|
import * as fs from "node:fs";
|
|
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 { composeInstructions, createTeacherAgent } from "../src/agent/teacher-agent.ts";
|
|
import { createTeacherTools } from "../src/agent/tools.ts";
|
|
import { handleChatRequest } from "../src/server/chat-handler.ts";
|
|
import { claimSteers, deliverJobResult, setTurnRunner } from "../src/server/jobs.ts";
|
|
import { persistTurn, withMessageIds } from "../src/server/turn-service.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";
|
|
|
|
const JOBS_DIR = path.join(PATHS.data, "jobs");
|
|
|
|
const createdChats: string[] = [];
|
|
const createdJobs: string[] = [];
|
|
|
|
function trackChat(id: string) {
|
|
createdChats.push(id);
|
|
}
|
|
|
|
function trackJob(id: string) {
|
|
createdJobs.push(id);
|
|
}
|
|
|
|
after(() => {
|
|
for (const id of createdChats) {
|
|
try {
|
|
fs.unlinkSync(path.join(PATHS.chats, `${id}.json`));
|
|
} catch {
|
|
/* already gone */
|
|
}
|
|
}
|
|
for (const id of createdJobs) {
|
|
try {
|
|
fs.unlinkSync(path.join(JOBS_DIR, `${id}.json`));
|
|
} catch {
|
|
/* already gone */
|
|
}
|
|
}
|
|
researchModelOverride.current = undefined;
|
|
});
|
|
|
|
function usage() {
|
|
return {
|
|
inputTokens: { total: 10, noCache: 10, cacheRead: 0, cacheWrite: 0 },
|
|
outputTokens: { total: 5, text: 5, reasoning: 0 },
|
|
};
|
|
}
|
|
|
|
async function drain(res: Response): Promise<string> {
|
|
if (res.status !== 200) {
|
|
assert.fail(`expected 200, got ${res.status}: ${await res.text()}`);
|
|
}
|
|
return res.text();
|
|
}
|
|
|
|
function post(chatId: string, messages: unknown[]): Request {
|
|
return new Request("http://test/api/chat", {
|
|
method: "POST",
|
|
headers: { "Content-Type": "application/json" },
|
|
body: JSON.stringify({ chatId, messages }),
|
|
});
|
|
}
|
|
|
|
function waitFor(cond: () => boolean, timeoutMs = 8000, what = "condition"): Promise<void> {
|
|
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", () => {
|
|
it("create → get → update → list round-trips", () => {
|
|
const job = jobStore.createJob({ kind: "research", chatId: "chat_storetest", toolCallId: "call-store" });
|
|
trackJob(job.id);
|
|
|
|
assert.match(job.id, /^job_/);
|
|
assert.equal(job.status, "running");
|
|
assert.equal(job.activity, "starting");
|
|
|
|
const loaded = jobStore.getJob(job.id)!;
|
|
assert.equal(loaded.toolCallId, "call-store");
|
|
assert.equal(loaded.kind, "research");
|
|
|
|
const updated = jobStore.updateJob(job.id, { activity: 'web_search: "test"', status: "done", result: { brief: "b" } })!;
|
|
assert.equal(updated.activity, 'web_search: "test"');
|
|
assert.equal(updated.status, "done");
|
|
assert.ok(updated.updatedAt >= job.updatedAt);
|
|
// immutable identity fields survive a patch
|
|
assert.equal(updated.toolCallId, "call-store");
|
|
assert.equal(jobStore.getJob(updated.id)!.toolCallId, "call-store");
|
|
|
|
const listed = jobStore.listJobs("chat_storetest");
|
|
assert.ok(listed.some((j) => j.id === job.id));
|
|
assert.equal(jobStore.listJobs("chat_other").some((j) => j.id === job.id), false, "jobs are scoped to their chat");
|
|
|
|
const byTool = jobStore.findJobByToolCallId("chat_storetest", "call-store");
|
|
assert.equal(byTool?.id, job.id);
|
|
assert.equal(jobStore.findJobByToolCallId("chat_storetest", "call-missing"), null);
|
|
});
|
|
});
|
|
|
|
// ─── 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: "<html></html>", tested: true, notes: "ok" };
|
|
const visual: VisualResult = { svg: "<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 (text → call), then plain text thereafter. */
|
|
function researchCallingTeacher(): MockLanguageModelV4 {
|
|
let turn = 0;
|
|
return new MockLanguageModelV4({
|
|
doStream: async () => {
|
|
turn++;
|
|
const chunks: LanguageModelV4StreamPart[] =
|
|
turn === 1
|
|
? [
|
|
{ type: "stream-start", warnings: [] },
|
|
{ type: "text-start", id: "t1" },
|
|
{ type: "text-delta", id: "t1", delta: "Let me check the current state of the library." },
|
|
{ type: "text-end", id: "t1" },
|
|
{ type: "tool-input-start", id: "call-r1", toolName: "research" },
|
|
{ type: "tool-input-delta", id: "call-r1", delta: JSON.stringify(RESEARCH_INPUT) },
|
|
{ type: "tool-input-end", id: "call-r1" },
|
|
{ type: "tool-call", toolCallId: "call-r1", toolName: "research", input: JSON.stringify(RESEARCH_INPUT) },
|
|
{ type: "finish", finishReason: { unified: "tool-calls", raw: "tool_calls" }, usage: usage() },
|
|
]
|
|
: [
|
|
{ type: "stream-start", warnings: [] },
|
|
{ type: "text-start", id: "t2" },
|
|
{ type: "text-delta", id: "t2", delta: "Back with the brief." },
|
|
{ type: "text-end", id: "t2" },
|
|
{ type: "finish", finishReason: { unified: "stop", raw: "stop" }, usage: usage() },
|
|
];
|
|
return { stream: simulateReadableStream({ chunks, initialDelayInMs: 0, chunkDelayInMs: 0 }) };
|
|
},
|
|
});
|
|
}
|
|
|
|
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 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),
|
|
}),
|
|
);
|
|
|
|
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));
|
|
|
|
// 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");
|
|
|
|
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;
|
|
}
|
|
});
|
|
});
|
|
|
|
// ─── the researcher loop with a fake model + stubbed fetch ──────────────────
|
|
|
|
const BRIEF = `## Summary
|
|
sigma.js v3 is current.
|
|
|
|
## Findings
|
|
1. **v3 is the latest major.** [sigma.js docs](https://www.sigmajs.org/docs)
|
|
|
|
## Sources
|
|
- sigma.js docs (https://www.sigmajs.org/docs)`;
|
|
|
|
/**
|
|
* The researcher loop runs via agent.generate (non-streaming) — the mock
|
|
* therefore scripts doGenerate. Turn 1: a web_search tool call; turn 2: the
|
|
* final text brief.
|
|
*/
|
|
function researcherModel(): MockLanguageModelV4 {
|
|
let turn = 0;
|
|
return new MockLanguageModelV4({
|
|
doGenerate: async () => {
|
|
turn++;
|
|
return turn === 1
|
|
? {
|
|
content: [
|
|
{
|
|
type: "tool-call" as const,
|
|
toolCallId: "ws1",
|
|
toolName: "web_search",
|
|
input: JSON.stringify({ query: "sigma.js latest version" }),
|
|
},
|
|
],
|
|
finishReason: { unified: "tool-calls" as const, raw: "tool-calls" },
|
|
usage: usage(),
|
|
warnings: [],
|
|
}
|
|
: {
|
|
content: [{ type: "text" as const, text: BRIEF }],
|
|
finishReason: { unified: "stop" as const, raw: "stop" },
|
|
usage: usage(),
|
|
warnings: [],
|
|
};
|
|
},
|
|
});
|
|
}
|
|
|
|
function stubFetch() {
|
|
const original = globalThis.fetch;
|
|
globalThis.fetch = (async (input: RequestInfo | URL) => {
|
|
const url = String(input);
|
|
if (url.startsWith("https://www.googleapis.com/")) {
|
|
return new Response(
|
|
JSON.stringify({
|
|
items: [{ title: "sigma.js", link: "https://www.sigmajs.org/", snippet: "v3 released" }],
|
|
}),
|
|
{ status: 200, headers: { "Content-Type": "application/json" } },
|
|
);
|
|
}
|
|
return new Response("unexpected fetch: " + url, { status: 500 });
|
|
}) as typeof fetch;
|
|
return () => {
|
|
globalThis.fetch = original;
|
|
};
|
|
}
|
|
|
|
describe("research runner (async protocol §3)", () => {
|
|
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 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: "output-available", input: RESEARCH_INPUT, output: "Dispatched." },
|
|
],
|
|
},
|
|
] as never);
|
|
|
|
const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-run1" });
|
|
trackJob(job.id);
|
|
|
|
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, "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 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 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, [
|
|
{ id: "u1", role: "user", parts: [{ type: "text", text: "hi" }] },
|
|
{
|
|
id: "a1",
|
|
role: "assistant",
|
|
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);
|
|
|
|
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 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();
|
|
}
|
|
});
|
|
});
|
|
|
|
// ─── the researcher agent construction (model resolution seam) ──────────────
|
|
|
|
describe("researcher agent", () => {
|
|
it("builds with the web tool set (no network touched)", () => {
|
|
const agent = createResearcherAgent(researcherModel());
|
|
assert.ok(agent.tools.web_search, "web_search available");
|
|
assert.ok(agent.tools.web_fetch, "web_fetch available");
|
|
assert.equal(Object.keys(agent.tools).length, 2, "ONLY the web tools — no question/plan tools");
|
|
// reset any accidental global override
|
|
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");
|
|
});
|
|
|
|
// ─── continuation coalescing (the delivery-starvation fix) ───────────────────
|
|
|
|
describe("claimSteers + pump duplicate-guard (continuation coalescing)", () => {
|
|
it("claims queued undelivered steers: resolves deliver(), flips the job done, empties the queue", async () => {
|
|
const { chatId, jobId, steer } = setup();
|
|
let fired = 0;
|
|
setTurnRunner(async () => {
|
|
fired++;
|
|
return 200;
|
|
});
|
|
|
|
const delivery = deliverJobResult(steer);
|
|
await delay(300); // the pump is inside waitForIdle's grace (750ms)
|
|
const claimed = claimSteers(chatId);
|
|
|
|
assert.equal(claimed.length, 1, "the queued steer is claimed");
|
|
assert.equal(claimed[0]!.id, steer.id);
|
|
// The handler's write-ahead persist lands the claim immediately after
|
|
// claiming (the very next statement in handleChatRequest).
|
|
const record = chatStore.getChat(chatId)!;
|
|
chatStore.saveMessages(chatId, [...record.messages, ...claimed] as never);
|
|
await delivery; // the runner's deliver() promise resolves on claim
|
|
assert.equal(jobStore.getJob(jobId)!.status, "done", "job flips done at claim");
|
|
assert.deepEqual(claimSteers(chatId), [], "queue is empty after the claim");
|
|
await delay(800);
|
|
assert.equal(fired, 0, "the pump never fires a wake for a claimed steer");
|
|
});
|
|
|
|
it("finalizes already-delivered steers without returning them (a wake raced in)", async () => {
|
|
const { chatId, jobId, steer } = setup();
|
|
const record = chatStore.getChat(chatId)!;
|
|
chatStore.saveMessages(chatId, [...record.messages, steer] as never);
|
|
|
|
const delivery = deliverJobResult(steer);
|
|
await delay(100); // let the pump see it first (its top-of-loop idempotent skip)
|
|
const claimed = claimSteers(chatId);
|
|
assert.equal(claimed.length, 0, "delivered steers are not re-claimed");
|
|
assert.equal(jobStore.getJob(jobId)!.status, "done");
|
|
await delivery;
|
|
});
|
|
|
|
it("claimSteers on a chat with no queue returns [] and touches nothing", () => {
|
|
assert.deepEqual(claimSteers("chat_never_existed"), []);
|
|
});
|
|
|
|
it("pump duplicate-guard: a steer coalesced into the record while the gates were closed is never double-fired", async () => {
|
|
const { chatId, jobId, steer } = setup();
|
|
let fired = 0;
|
|
setTurnRunner(async () => {
|
|
fired++;
|
|
return 200;
|
|
});
|
|
|
|
// Gates closed: a pending quiz parks the pump in waitForIdle.
|
|
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);
|
|
const delivery = deliverJobResult(steer);
|
|
await delay(1200);
|
|
assert.equal(fired, 0, "parked behind the pending quiz");
|
|
|
|
// The client turn coalesces: it claims the steer and lands it in the record.
|
|
const claimed = claimSteers(chatId);
|
|
assert.equal(claimed.length, 1, "coalescing claims the parked steer");
|
|
const record = chatStore.getChat(chatId)!;
|
|
chatStore.saveMessages(chatId, [...record.messages, steer] as never); // write-ahead persist
|
|
|
|
// The learner answers: the gates open — the parked pump must see the
|
|
// delivered steer via its post-idle re-check, NOT fire a second turn.
|
|
const answered = chatStore.getChat(chatId)!.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 delivery;
|
|
await delay(1500);
|
|
assert.equal(fired, 0, "no duplicate wake after coalescing");
|
|
assert.equal(jobStore.getJob(jobId)!.status, "done");
|
|
});
|
|
});
|
|
|
|
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<void> {
|
|
return new Promise((resolve) => setTimeout(resolve, ms));
|
|
}
|
|
|
|
// ─── persist id hygiene (the duplicate-empty-key bug) ────────────────────────
|
|
|
|
describe("persistTurn id hygiene", () => {
|
|
it("withMessageIds fills EMPTY ids (server-only synthesized messages) and preserves existing ones", () => {
|
|
const filled = withMessageIds([
|
|
{ id: "", role: "assistant", parts: [{ type: "text", text: "wake turn" }] },
|
|
{ id: "client-id", role: "user", parts: [{ type: "text", text: "hi" }] },
|
|
]);
|
|
assert.notEqual(filled[0]!.id, "", "empty id replaced");
|
|
assert.match(filled[0]!.id, /^msg_/);
|
|
assert.equal(filled[1]!.id, "client-id", "client-authored id untouched");
|
|
assert.notEqual(filled[0]!.id, filled[1]!.id);
|
|
});
|
|
|
|
it("persistTurn dedupes same-id continuation copies, keeping the LAST (the SDK's merged superset)", () => {
|
|
// The coalescing shape: a steer (user-role) rides a quiz-answer
|
|
// continuation AFTER the assistant message being continued, so the
|
|
// SDK's onFinish list carries the input copy AND the extended
|
|
// response copy under one id — the record must keep only the merge.
|
|
const chat = chatStore.createChat("Test: dedupe continuation copies");
|
|
trackChat(chat.id);
|
|
const short = { id: "msg_dup_1", role: "assistant", parts: [{ type: "text", text: "step 1" }] };
|
|
const long = { id: "msg_dup_1", role: "assistant", parts: [{ type: "text", text: "step 1" }, { type: "text", text: "step 2 (continued)" }] };
|
|
persistTurn(chatStore, chat.id, [
|
|
{ id: "u1", role: "user", parts: [{ type: "text", text: "hi" }] },
|
|
short,
|
|
{ id: "steer_x", role: "user", parts: [{ type: "text", text: "research complete" }] },
|
|
long,
|
|
], false);
|
|
const record = chatStore.getChat(chat.id)!;
|
|
assert.equal(record.messages.length, 3, "the duplicate entry is dropped");
|
|
assert.deepEqual(
|
|
record.messages.map((m) => m.id),
|
|
["u1", "steer_x", "msg_dup_1"],
|
|
"order preserved, steer before the merged assistant copy",
|
|
);
|
|
const merged = record.messages[2]!;
|
|
assert.equal(merged.parts.length, 2, "the KEPT copy is the extended one");
|
|
assert.equal((merged.parts[1] as { text: string }).text, "step 2 (continued)");
|
|
});
|
|
|
|
it("persistTurn saves every message with a unique non-empty id (wake turns never re-POSTed)", () => {
|
|
const chat = chatStore.createChat("Test: persist ids");
|
|
trackChat(chat.id);
|
|
persistTurn(chatStore, chat.id, [
|
|
{ id: "", role: "user", parts: [{ type: "text", text: "q" }] },
|
|
{ id: "", role: "assistant", parts: [{ type: "text", text: "a1" }] },
|
|
{ id: "", role: "assistant", parts: [{ type: "text", text: "a2" }] },
|
|
], false);
|
|
const record = chatStore.getChat(chat.id);
|
|
const ids = record!.messages.map((m) => m.id);
|
|
assert.ok(ids.every((id) => id && id.length > 0), "no empty ids survive persist");
|
|
assert.equal(new Set(ids).size, ids.length, "ids unique — React keys safe");
|
|
});
|
|
});
|
|
|
|
// ─── prompt composition hygiene (the README leak) ─────────────────────────────
|
|
|
|
describe("prompt composition", () => {
|
|
it("loads ONLY numeric-prefix fragments — README.md never enters the system prompt", () => {
|
|
const instructions = composeInstructions();
|
|
assert.ok(!instructions.includes("Prompt fragments"), "README content must not leak");
|
|
assert.ok(!instructions.includes("Rules for adding a fragment"), "README content must not leak (rules)");
|
|
// every numeric fragment DOES compose, verbatim, after the base prompt
|
|
const base = fs.readFileSync("src/agent/system-prompt.md", "utf-8").trim();
|
|
assert.ok(instructions.startsWith(base), "base prompt first");
|
|
for (const f of fs.readdirSync("src/agent/prompts").filter((f) => /^\d+-[\w.-]+\.md$/.test(f)).sort()) {
|
|
const body = fs.readFileSync(`src/agent/prompts/${f}`, "utf-8").trim();
|
|
assert.ok(instructions.includes(body), `fragment ${f} must compose verbatim`);
|
|
}
|
|
});
|
|
|
|
it("guards prompt size — no fragment can silently balloon (the README-leak class)", () => {
|
|
// The wave-1 leak: README.md (815 bytes / ~200 tokens) rode into every
|
|
// teacher turn until the numeric-prefix filter stopped it. This guard
|
|
// catches the sibling failure mode — an accidental dump pasted into a
|
|
// fragment. Thresholds: largest real fragment is 40-demos.md at ~3.2k
|
|
// chars; 8k is generous headroom. Composed total ≤ 60k chars.
|
|
for (const f of fs.readdirSync("src/agent/prompts").filter((f) => /^\d+-[\w.-]+\.md$/.test(f))) {
|
|
const body = fs.readFileSync(`src/agent/prompts/${f}`, "utf-8").trim();
|
|
assert.ok(body.length <= 8_000, `fragment ${f} is ${body.length} chars — over the 8k guard`);
|
|
}
|
|
assert.ok(composeInstructions().length <= 60_000, "composed system prompt over the 60k guard");
|
|
});
|
|
});
|