fix: steer-delivery starvation — continuation coalescing (claimSteers)
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.
This commit is contained in:
amosblomqvist
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -128,6 +128,45 @@ export function deliverJobResult(steer: SubagentSteerMessage): Promise<void> {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<void> {
|
||||
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);
|
||||
|
||||
+92
-1
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user