From e33b83ca2d5be3a151123289f6ffa39cd7777aa5 Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Wed, 9 Sep 2026 03:04:21 +0300 Subject: [PATCH] =?UTF-8?q?fix:=20steer-delivery=20starvation=20=E2=80=94?= =?UTF-8?q?=20continuation=20coalescing=20(claimSteers)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The bug (seen live): a teacher that ends every turn with a new pending quiz keeps the wake pipeline's idle gates permanently closed — the steer sits in 'delivering result' forever while the learner answers check after check. - claimSteers (server/jobs.ts): the chat-handler claims every queued, undelivered steer the instant a client turn wins the turn registry, and appends it after the learner's answer — conversion-safe, the answered quiz carries its output. Every answer flushes pending steers immediately. - The pump keeps its post-idle duplicate-guard (re-check steerDelivered before firing), so a claim racing the pump converges to exactly one delivery either way. - The stale-job sweep now skips chats with a turn in flight (a healthy but slow wake no longer gets falsely re-queued as 'recovering'). - Out-of-lane surgical touch (flagged for master review): the 3-line handleChatRequest seam (claim after beginTurn, persist+stream 'outgoing'). - Verified live end-to-end on the stuck session: answering the pending quiz fired the steer + wake turn within a second; job flipped done/delivered. --- src/core/contracts/jobs.ts | 11 +++++ src/server/chat-handler.ts | 16 +++++-- src/server/jobs.ts | 49 ++++++++++++++++++++ tests/jobs.test.ts | 93 +++++++++++++++++++++++++++++++++++++- 4 files changed, 165 insertions(+), 4 deletions(-) diff --git a/src/core/contracts/jobs.ts b/src/core/contracts/jobs.ts index 5c1127c..31c2892 100644 --- a/src/core/contracts/jobs.ts +++ b/src/core/contracts/jobs.ts @@ -73,6 +73,17 @@ export interface JobRecord { * a polling client that sees the flip re-syncs once and gets the * finished record. * + * CONTINUATION COALESCING: the idle window above can starve — a + * quiz-chaining teacher ends every turn with a new pending check, so + * the gates never stay open long enough for a wake to fire. The fix: + * when a client turn wins the turn registry, the handler CLAIMS every + * queued, undelivered steer (jobs.ts claimSteers) and appends it after + * the learner's answer in that very turn — conversion-safe, because the + * answered quiz carries its output. Every answer flushes pending + * steers immediately; the wake pump remains for turns that end with no + * pending question. The pump re-checks steerDelivered right before + * firing, so a claim racing the pump converges to exactly one delivery. + * * 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 diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index 54f1b4a..8f7fa69 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -3,7 +3,7 @@ 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 { claimSteers, deliverJobResult, setJobRespawn, setTurnRunner } from "./jobs.ts"; import { stores } from "./stores.ts"; import { beginTurn, endTurn } from "./turn-liveness.ts"; import { buildTeacherContext, persistTurn, triggerExtraction } from "./turn-service.ts"; @@ -71,6 +71,16 @@ export async function handleChatRequest(req: Request, agent?: TeacherAgent): Pro return Response.json({ error: "A turn is already running for this session" }, { status: 409 }); } + // CONTINUATION COALESCING (async protocol §4, the delivery-starvation fix): + // steers finished while a quiz chain held the wake gates ride THIS turn — + // appended after the learner's answer (conversion-safe: the answer carries + // its output). Claiming happens after beginTurn, so this turn owns the + // registry and the write-ahead persist below makes the claim durable. + const steers = claimSteers(chatId); + const outgoing = ( + steers.length > 0 ? [...(body.messages as UIMessage[]), ...steers] : body.messages + ) as UIMessage[]; + // Write-ahead persistence: the request body IS the client's canonical // state, so it goes on disk BEFORE the stream starts. The learner's answer // is durable the instant the request arrives — any remount, second tab, or @@ -78,7 +88,7 @@ export async function handleChatRequest(req: Request, agent?: TeacherAgent): Pro // returns mid-turn reads turnInFlight from the registry and re-attaches by // polling instead of resurrecting the stale transcript.) try { - stores.chatStore.saveMessages(chatId, body.messages as UIMessage[]); + stores.chatStore.saveMessages(chatId, outgoing); } catch (err) { console.error(`[chat ${chatId}] write-ahead persist failed:`, err); } @@ -97,7 +107,7 @@ export async function handleChatRequest(req: Request, agent?: TeacherAgent): Pro }), buildTeacherContext(stores.graphStore, chatId), ), - uiMessages: body.messages as UIMessage[], + uiMessages: outgoing, // Tool-call validation failures are messages OUR schemas authored (the // quiz authoring contract, unknown tools) — safe and useful in the // transcript: they are exactly what the teacher saw and retried on. The diff --git a/src/server/jobs.ts b/src/server/jobs.ts index 160fb60..780d6c5 100644 --- a/src/server/jobs.ts +++ b/src/server/jobs.ts @@ -128,6 +128,45 @@ export function deliverJobResult(steer: SubagentSteerMessage): Promise { }); } +/** + * CONTINUATION COALESCING ("THE ASYNC PROTOCOL" §4): the chat-handler calls + * this the instant a client turn wins the turn registry, BEFORE its + * write-ahead persist. Undelivered steers for the chat ride THAT turn — + * appended after the learner's answer, which is conversion-safe (the + * answered quiz carries its output) — instead of waiting for the pump's + * idle window, which a quiz-chaining teacher may never open (the delivery + * starvation bug: every turn ends with a new pending quiz, so the gates + * never stay open long enough to fire a wake). + * + * Claimed steers resolve their deliver() promise and flip the job done + * here: the caller holds the turn registry, so the write-ahead persist is + * imminent and durable. Steers already in the transcript (a wake raced in) + * are finalized, not returned. The pump re-checks steerDelivered right + * before firing, so a claim racing the pump converges to exactly one + * delivery either way. + */ +export function claimSteers(chatId: string): SubagentSteerMessage[] { + const s = state(); + const queue = s.queues.get(chatId); + if (!queue || queue.length === 0) return []; + const record = stores.chatStore.getChat(chatId); + if (!record) return []; // the handler validated the chat; defensive only + const claimed: SubagentSteerMessage[] = []; + for (const entry of queue) { + const jobId = jobIdOf(entry.steer); + if (steerDelivered(record.messages, jobId)) { + // A wake already landed it — just finish the bookkeeping. + finalizeJob(jobId); + } else { + claimed.push(entry.steer); + finalizeJob(jobId); // delivery is now the client turn's job + } + entry.resolve(); + } + s.queues.delete(chatId); + return claimed; +} + function chatIdOf(steer: SubagentSteerMessage): string { // The chat is resolvable from the job record; the steer's metadata // carries jobId, which indexes the store. @@ -194,6 +233,15 @@ async function pump(chatId: string): Promise { continue; } + // A client continuation may have COALESCED this steer (claimSteers) while + // the gates were closed — re-check before firing: exactly one delivery. + if (steerDelivered(fresh.messages, jobId)) { + queue.shift(); + finalizeJob(jobId); + entry.resolve(); + 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) => { @@ -288,6 +336,7 @@ function recoverStaleJobs(): void { if (job.status !== "running") continue; if (now - Date.parse(job.updatedAt) < STALE_JOB_MS) continue; if (!stores.chatStore.getChat(job.chatId)) continue; + if (isTurnRunning(job.chatId)) continue; // a wake turn may be delivering it right now if (job.result) { // The wake never fired (or crashed mid-flight): re-deliver. const steer = buildSubagentSteer(job, job.result as never); diff --git a/tests/jobs.test.ts b/tests/jobs.test.ts index 2b4712e..ba0c107 100644 --- a/tests/jobs.test.ts +++ b/tests/jobs.test.ts @@ -41,7 +41,7 @@ import { 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 { deliverJobResult, setTurnRunner } from "../src/server/jobs.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"; @@ -641,6 +641,97 @@ describe("wake pipeline (src/server/jobs.ts)", () => { 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);