From 7f51259fc614af8acf30d9b77379911f1007d92d Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Sat, 26 Sep 2026 03:46:09 +0300 Subject: [PATCH] feat: Wave C - pi-native steer/follow-up delivery Delivery now rides pi's OWN follow-up queue when a run is live, and a carry-everything idle wake otherwise. One registry, two vehicles; claimSteers and continuation coalescing are deleted (owner rulings, 2026-09-26: follow-ups only; one queue; the wake is the only appender). THE LIVE RUN (vehicle 1): the chat-handler registers the streaming turn's agent (registerLiveAgent, next to registerTurn); the pump hands every eligible queued marker to agent.followUp() - pi polls that queue when the run would otherwise stop, injects the markers (message_start/message_end on the live wire), and CONTINUES the same run. followUpMode "all" joins every handed marker in one continuation turn. The question stopper runs before the follow-up poll, so an open quiz is never overridden - a handed marker strands at settlement and waits (today's pacing). probe-handoffs never take this path (the handoff must START the teacher's run - a record-level branch). SETTLEMENT RECONCILIATION: settleSteers(chatId, persisted) at onFinish (after unregisterLiveAgent) resolves+flips entries the run delivered and returns stranded ones to pending - the registry stays the truth; the dead agent's queued copy is abandoned. This is what makes the handoff safe: pi owns no retry for a stranded follow-up. THE IDLE WAKE (vehicle 2): waitForIdle (unchanged gates) then ONE turn carrying EVERY pending marker as its input appends - dequeue-at-fire (the Wave B regression lesson: the wake run's own follow-up poll must never see its own cargo; a 409 pushes the cargo back and retries). CLIENT: applyWireEvent now upserts non-assistant messages by id, so in-run injected markers render live instead of dropping until settle. Tests: jobs 44/44 (live handoff, settlement stranding, handoff-waits, one-wake-carries-all); chat-flow 23/23 (the double-append pin drives the real pump through the new input); session-core 28/28 (custom-role wire appends). Full suite 997/997, tsc/lint/build green. Live smoke (PORT=3004): research job delivered mid-probe - parked behind the dangling questions exactly per semantics, landed once on the open gates, teacher folded the brief in and re-dispatched; the second (good) result rendered the full 10-source brief card and the lesson continued from it. No duplicate ids anywhere in the record. Known gap flagged to owner (NOT from this change): the server accepts malformed answer events via raw API (answers without the answers[] shape) and AnsweredCard then crashes the page render - surfaced by the smoke's scripted answers, repaired by hand in the scratch record. --- components/session-core.ts | 17 +- src/agent/teacher-agent.ts | 6 + src/core/input-events.ts | 8 +- src/server/chat-handler.ts | 56 +++---- src/server/jobs.ts | 313 +++++++++++++++++++++++++------------ tests/jobs.test.ts | 251 ++++++++++++++--------------- tests/session-core.test.ts | 40 +++++ 7 files changed, 426 insertions(+), 265 deletions(-) diff --git a/components/session-core.ts b/components/session-core.ts index bf6fe05..bb6c946 100644 --- a/components/session-core.ts +++ b/components/session-core.ts @@ -290,7 +290,22 @@ export class SessionModel { role?: string; id?: string; }; - if ((message as { role?: string }).role !== "assistant") return; + if ((message as { role?: string }).role !== "assistant") { + // Wave C: markers injected mid-run (pi's follow-up delivery) + // arrive as custom-role message_start/message_end events - + // UPSERT by id (markers always carry steer_ ids; id-less + // non-assistant events never occur - injected messages emit + // start+end only, no streaming partials). Without this the + // live stream dropped the marker until the settle sync. + const mid = message.id; + if (mid === undefined) return; + this.mutate(() => { + const idx = this.tail.findIndex((m) => m.id === mid); + this.tail = + idx === -1 ? [...this.tail, message as LhMessage] : this.tail.with(idx, message as LhMessage); + }); + return; + } // pi streams AssistantMessages with NO id — the id exists only // once the server's persistTurn stamps the record (the live // streaming bug, 2026-09-26: minting a fresh id per event made diff --git a/src/agent/teacher-agent.ts b/src/agent/teacher-agent.ts index 9dc78d3..f379818 100644 --- a/src/agent/teacher-agent.ts +++ b/src/agent/teacher-agent.ts @@ -165,6 +165,12 @@ function baseAgentOptions(kind: AgentKind, opts: AgentBuildOptions, tools: Agent }, streamFn: engine.streamFn, shouldStopAfterTurn: stopper, + // Wave C (2026-09-26): markers handed to a live run drain from pi's own + // follow-up queue — "all" joins every handed marker in ONE + // continuation turn, matching the idle wake's carry-everything + // semantics (the default "one-at-a-time" would spend one teacher + // turn per marker). + followUpMode: "all", // Wave B (owner-approved 2026-09-26): the model-facing conversion is // THE app seam — markers flatten to their canonical user text // (verdictText / HANDOFF_TEXT / steer text, all single-sourced), diff --git a/src/core/input-events.ts b/src/core/input-events.ts index 4ff0619..19f34b6 100644 --- a/src/core/input-events.ts +++ b/src/core/input-events.ts @@ -42,8 +42,12 @@ export type ChatInput = | { type: "verdict"; accepted: boolean; note?: string } | { type: "continue" }; -/** A wake-pipeline steer delivery — same turn composition, server-origin. */ -export type SteerInput = { type: "steer"; steer: LhMessage }; +/** A wake-pipeline steer delivery — same turn composition, server-origin. + * Wave C (2026-09-26): the idle wake carries EVERY queued marker as its + * input appends — with claimSteers deleted, the pump's wake is the only + * steer appender left (the in-run path hands markers to pi's own + * follow-up queue instead, so it never builds an input event). */ +export type SteerInput = { type: "steers"; steers: LhMessage[] }; /** Every input the turn composition accepts. */ export type TurnInput = ChatInput | SteerInput; diff --git a/src/server/chat-handler.ts b/src/server/chat-handler.ts index a2f0a79..2742c58 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -4,7 +4,7 @@ import { piEngineForSpec, type PiEngine } from "../agent/models.ts"; import { createJobRunner } from "../agent/subagents/runner.ts"; import { createExtractionPipeline } from "../agent/subagents/extraction.ts"; import type { ModelRole } from "../../config.ts"; -import { claimSteers, deliverJobResult, setJobRespawn, setTurnRunner } from "./jobs.ts"; +import { deliverJobResult, registerLiveAgent, setJobRespawn, setTurnRunner, settleSteers, unregisterLiveAgent } from "./jobs.ts"; import { piTurnResponse, runPiTurn } from "./pi/ui-stream.ts"; import { effectiveSettings, stores } from "./stores.ts"; import { searchAvailable, searchProvider } from "./search.ts"; @@ -184,14 +184,16 @@ export async function handleChatInput( // The input's native appends (pure; the honest 400 for an answer with no // pending question call). Computed BEFORE beginTurn so a refused input - // leaves no liveness effects; read against the CURRENT record. + // leaves no liveness effects; read against the CURRENT record. Wave C: + // the idle wake carries EVERY queued marker (steerDelivered filters a + // re-fire of an already-landed one) - with claimSteers gone, this wake + // input is the only steer appender; in-run delivery rides pi's own + // follow-up queue and never builds an input event at all. let appends; try { appends = - input.type === "steer" - ? steerDelivered(record.messages, input.steer.id) - ? [] - : [input.steer] + input.type === "steers" + ? input.steers.filter((s) => !steerDelivered(record.messages, s.id)) : inputAppends(record, input); } catch (err) { if (err instanceof ChatInputError) { @@ -208,30 +210,13 @@ export async function handleChatInput( 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 is the - // question call's tool result now). Claiming happens after beginTurn, so - // this turn owns the registry and the write-ahead persist below makes the - // claim durable. Steers already in the transcript (a wake raced in) are - // finalized by the claim itself — the append filter keeps exactly one copy. - // THE CARRIED FILTER (restored 2026-09-26, the Wave B regression of the - // 2026-09-18 duplicate-key bug): a pump-fired wake turn ALREADY CARRIES - // its steer in `appends` — and the pump only shifts its queue entry AFTER - // the turn registers, so the claim above pops that very steer too. The - // old record-side-only filter couldn't see the carried copy (it checks - // the PRE-write-ahead snapshot), and the marker persisted twice with the - // same `steer_` id — two view messages, one React key, crash. - const carried = input.type === "steer" ? input.steer.id : undefined; - const steers = claimSteers(chatId).filter( - (s) => s.id !== carried && !steerDelivered(record.messages, s.id), - ); - // Write-ahead persistence: the input's native appends go on disk BEFORE - // the turn starts — the learner's answer is durable the instant the + // the turn starts - the learner's answer is durable the instant the // request arrives, and any remount, second tab, or server crash reads - // truth instead of the pre-answer record. - const merged = [...record.messages, ...appends, ...steers]; + // truth instead of the pre-answer record. (Wave C: the wake's markers + // ride this same write-ahead - the pump de-queued them at fire, so this + // is the ONLY copy ever appended.) + const merged = [...record.messages, ...appends]; try { stores.chatStore.saveMessages(chatId, merged); } catch (err) { @@ -373,6 +358,13 @@ export async function handleChatInput( // nothing must find the persisted record current — the same // ordering rule as endTurn. unregisterTurn(chatId); + // Wave C steer settlement: the live agent is dead (no more + // followUp handoffs) - reconcile the steer registry against + // what this run actually persisted (in-run injections + // resolve; stranded markers return to pending), then let + // the pump take any stragglers down the idle wake. + unregisterLiveAgent(chatId); + settleSteers(chatId, stamped); }, }); // The turn is now a live background object: register it so any view @@ -381,6 +373,11 @@ export async function handleChatInput( // runPiTurn call is race-free: the agent loop only ticks in later // microtasks, so settle (and thus unregister) cannot beat us here. registerTurn(chatId, handle); + // Wave C: the pump hands queued steer markers to THIS run's agent + // (pi's own follow-up queue) - same race-freedom as registerTurn: + // the loop only ticks in later microtasks, so settlement cannot + // beat the registration. + registerLiveAgent(chatId, constructed); // The POST response: just this caller's view onto the running turn — // identical wire dialect to the reattach stream. return piTurnResponse(handle); @@ -388,6 +385,9 @@ export async function handleChatInput( // The stream never started, so no onFinish will fire — clear the registry // or a malformed request would lock the chat until the TTL eviction. endTurn(chatId); + // Wave C: the live-agent registration (if it happened) dies with the + // turn here too - nothing was handed (the loop never ticked). + unregisterLiveAgent(chatId); console.error(`[chat ${chatId}] turn failed to start:`, err); return Response.json({ error: "Failed to start the turn" }, { status: 500 }); } diff --git a/src/server/jobs.ts b/src/server/jobs.ts index f5fe66e..70c5336 100644 --- a/src/server/jobs.ts +++ b/src/server/jobs.ts @@ -1,24 +1,40 @@ /** * THE WAKE PIPELINE (contracts.ts "THE ASYNC PROTOCOL" §4) — server-side - * steer delivery for finished sub-agent jobs. + * steer delivery for finished sub-agent jobs. Wave C (2026-09-26): + * delivery rides pi's OWN follow-up queue when a run is live, and the + * idle wake otherwise. One registry, two vehicles. * - * deliverJobResult(steer) — the runner calls this on job completion. The - * steer joins a per-chat serialized queue and a pump drives it: + * deliverJobResult(steer) — the runner calls this on job completion. + * The steer joins a per-chat serialized registry and the pump drives + * it along ONE of two paths: * - * 1. WAIT FOR IDLE: while a turn is in flight (turn-liveness) OR a - * question tool part is pending (an unanswered client-side tool - * call breaks provider conversion — the steer fires right after - * the learner's continuation), poll. A short grace after the gates - * open avoids stealing the turn a client is about to send. - * 2. FIRE: a server-side teacher turn through the SAME composition as - * a client request (the injected turn runner wraps - * handleChatRequest): write-ahead persist, liveness, extraction — - * one code path. A 409 means a client turn raced in — retry with - * backoff. Idempotent per jobId (steerDelivered): a crashed wake - * or a duplicate deliver is skipped, not duplicated. - * 3. FLIP: only now does the job go done/error — a polling client - * that sees the flip re-syncs once and gets the finished record - * (steer + the teacher's response to it). + * 1. LIVE RUN → pi follow-up: a turn is streaming, so the marker is + * handed to the LIVE agent's followUp queue (agent.followUp). + * pi polls that queue when the run would otherwise stop and + * CONTINUES THE SAME RUN — the marker joins the context, rides + * the live event stream (message_start/message_end), and lands + * in the record at that run's persist. The question stopper + * runs BEFORE the follow-up poll and exits the loop, so an open + * quiz is never overridden: a handed marker strands (see + * settleSteers) and waits for the next run — today's pacing. + * probe-handoff markers NEVER take this path (owner ruling: + * the handoff must start the TEACHER's run — a record-level + * branch only a fresh wake can make). + * 2. IDLE WAKE: no live turn. WAIT FOR IDLE (turn-liveness OR a + * dangling question — an unanswered client-side tool call breaks + * provider conversion), then fire ONE server-side wake turn + * carrying EVERY queued marker as its input appends (write-ahead + * persist, liveness, extraction — one code path, the same + * composition as a client request). A 409 means a client turn + * raced in — re-queue and retry with backoff. Idempotent per + * jobId (steerDelivered): a crashed wake or a duplicate deliver + * is skipped, not duplicated. + * + * settleSteers(chatId, persisted) — the handler calls this at turn + * settlement (after unregisterLiveAgent): entries the run DELIVERED + * resolve + flip their jobs; entries that were handed to the dying + * run but never injected (stopper exit, abort, a missed final poll) + * return to pending and the pump restarts — no stranding, no loss. * * Crash recovery (recoverStaleJobs): a process restart can orphan jobs — * "running" with a result (the wake never fired) or without one (the @@ -44,10 +60,12 @@ import { danglingQuestionOf } from "../core/native-messages.ts"; import { fireJobAbort, isJobLive } from "../agent/subagents/aborts.ts"; import { stores } from "./stores.ts"; import { isTurnRunning } from "./turn-liveness.ts"; +import type { Agent } from "@earendil-works/pi-agent-core"; /** Fires one server-side teacher turn: (chatId, input) → HTTP-ish status. - * Wave B: the input is the STEER EVENT — the handler appends the marker - * natively under its own turn claim (one composition for every turn). */ + * Wave C: the input is the IDLE WAKE — every queued marker as the turn's + * input appends (the handler appends them natively under the wake's own + * turn claim; one composition for every turn). */ export type TurnRunner = (chatId: string, input: TurnInput) => Promise; /** Re-runs a job's sub-agent loop (crash recovery for orphaned jobs). */ @@ -57,13 +75,24 @@ interface QueuedSteer { steer: SteerMessage; resolve: () => void; reject: (reason: Error) => void; + /** The live agent this marker was handed to via followUp() — set by the + * pump's live path. Guards double-handoff (a re-pumped loop must not + * re-queue what a run already holds) and is CLEARED at settlement for + * markers the dying run never injected. */ + handedTo?: Agent; } interface WakeState { turnRunner?: TurnRunner; respawn?: JobRespawn; - /** chatId → steers awaiting delivery, in arrival order. */ + /** chatId → steers awaiting delivery, in arrival order. The single + * delivery registry (Wave C: the agent's own follow-up queue is a + * per-run VEHICLE, not the registry — settlement reconciles here). */ queues: Map; + /** chatId → the agent of the CURRENTLY streaming turn (cleared at + * settlement, before settleSteers). The pump's live path hands + * markers to this agent's follow-up queue. */ + liveAgents: Map; /** Chats whose pump loop is currently active (serialization per chat). */ pumping: Set; /** Throttle bookkeeping for the stale-job sweep. */ @@ -76,7 +105,7 @@ interface LhWakeGlobal { function state(): WakeState { const g = globalThis as typeof globalThis & LhWakeGlobal; - return (g.__lhWakePipeline ??= { queues: new Map(), pumping: new Set(), lastSweepAt: 0 }); + return (g.__lhWakePipeline ??= { queues: new Map(), liveAgents: new Map(), pumping: new Set(), lastSweepAt: 0 }); } /** chat-handler wires this at module init: its adapter around handleChatRequest. */ @@ -145,48 +174,55 @@ export function deliverJobResult(steer: SteerMessage): 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). +/** The chat-handler registers the CURRENTLY STREAMING turn's agent right + * after registerTurn — the pump's live path hands queued markers to this + * agent's pi-native follow-up queue (followUpMode "all": every handed + * marker joins the continuation turn together, matching the wake's + * carry-everything semantics). Registration is race-free the same way + * registerTurn is: the agent loop only ticks in later microtasks, so the + * settle seam cannot beat the registration. */ +export function registerLiveAgent(chatId: string, agent: Agent): void { + state().liveAgents.set(chatId, agent); +} + +/** Clear the live agent at settlement — MUST run before settleSteers so the + * reconciliation's re-pump can never hand a marker to the dying agent. */ +export function unregisterLiveAgent(chatId: string): void { + state().liveAgents.delete(chatId); +} + +/** TURN SETTLEMENT RECONCILIATION (Wave C): the handler calls this AFTER + * unregisterLiveAgent, with the run's persisted messages. Two jobs: * - * 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. + * - entries whose marker the run DELIVERED (in-run follow-up injection — + * their ids are in the persisted slice) resolve and flip their jobs; + * - entries handed to the dying run but never injected (the question + * stopper exited before the follow-up poll; an abort; a delivery that + * raced the final poll) return to PENDING (handedTo cleared) and the + * pump restarts — the idle wake delivers them once the gates open. * - * incoming (the turn's request body — the wake pump fires turns that - * ALREADY CARRY the steer it dequeues only after the turn returns 200, so - * the entry is still queued here): a steer already in the incoming list is - * being delivered by THIS VERY TURN — finalized and resolved, but never - * appended again. Without this the write-ahead persist would hold two - * copies of the same steer id (a live React duplicate-key warning the - * owner hit 2026-09-18; withMessageIds only heals it at onFinish). - */ -export function claimSteers(chatId: string): SteerMessage[] { - const s = state(); - const queue = s.queues.get(chatId); - if (!queue || queue.length === 0) return []; - s.queues.delete(chatId); - // The caller holds the turn registry (chat-handler claims after - // beginTurn), so the write-ahead persist is imminent and durable: - // every queued steer is claimed — its delivery is THIS turn's job. - // Steers already in the transcript (a wake raced in) still finalize - // here; the handler's append filter (steerDelivered) keeps exactly - // one copy in the record. - const claimed = queue.map((entry) => entry.steer); + * This is what makes the live hand-off safe: pi owns no retry for a marker + * stranded in a settled agent's queue, so the registry stays the truth and + * the dead agent's copy is simply abandoned (GC'd with the agent). */ +export function settleSteers(chatId: string, persisted: { id?: string }[]): void { + const queue = state().queues.get(chatId); + if (!queue || queue.length === 0) return; + const deliveredIds = new Set(persisted.map((m) => m.id).filter(Boolean)); + const remaining: QueuedSteer[] = []; for (const entry of queue) { - finalizeSteer(entry.steer); - entry.resolve(); + if (deliveredIds.has(entry.steer.id)) { + finalizeSteer(entry.steer); + entry.resolve(); + } else { + entry.handedTo = undefined; + remaining.push(entry); + } + } + if (remaining.length === 0) state().queues.delete(chatId); + else { + state().queues.set(chatId, remaining); + void pump(chatId); } - return claimed; } /** Result of an abort attempt — the endpoint maps these to HTTP. */ @@ -231,13 +267,18 @@ export function abortJob(jobId: string): AbortOutcome { fireJobAbort(jobId); // no-op when the loop already returned (delivering) // Drop any queued steers for this job (their deliver() resolves unfired). + // HANDED entries (already in a live run's follow-up queue) are past + // recall — same ruling as a fired wake: the result lands, and the abort + // note below dedupes against it via steerDelivered (same jobId). const queue = state().queues.get(job.chatId); if (queue) { - const remaining = queue.filter((e) => jobIdOf(e.steer) !== jobId); + const remaining = queue.filter( + (e) => jobIdOf(e.steer) !== jobId || e.handedTo !== undefined, + ); if (remaining.length === 0) state().queues.delete(job.chatId); else state().queues.set(job.chatId, remaining); for (const entry of queue) { - if (jobIdOf(entry.steer) === jobId) entry.resolve(); + if (jobIdOf(entry.steer) === jobId && entry.handedTo === undefined) entry.resolve(); } } @@ -262,8 +303,14 @@ function chatIdOf(steer: SteerMessage): string { } /** - * The per-chat pump: delivers every queued steer in order, one wake turn - * each. Only one loop per chat at a time; later arrivals ride the queue. + * The per-chat pump: delivers the registry along pi's two native paths. + * Only one loop per chat at a time; later arrivals ride the registry. + * + * Each iteration first clears dead weight (gone chat, already-delivered + * markers), then picks the vehicle: a LIVE turn takes every eligible + * marker via agent.followUp (the run continues at its own stop — nothing + * more to do here until settlement re-pumps); otherwise the IDLE WAKE + * fires ONE turn carrying every pending marker as its input appends. */ async function pump(chatId: string): Promise { if (!chatId) return; @@ -277,26 +324,69 @@ async function pump(chatId: string): Promise { s.queues.delete(chatId); return; } - const entry = queue[0]!; - const steerId = entry.steer.id; // Gone chat: nothing to deliver into. const record = stores.chatStore.getChat(chatId); if (!record) { - queue.shift(); - entry.reject(new Error(`chat ${chatId} not found — cannot deliver steer ${steerId}`)); + s.queues.delete(chatId); + for (const entry of queue) { + entry.reject(new Error(`chat ${chatId} not found — cannot deliver steer ${entry.steer.id}`)); + } continue; } - // Idempotent delivery: the transcript already carries this steer - // (a previous wake completed, or a duplicate deliver raced in). - if (steerDelivered(record.messages, steerId)) { - queue.shift(); - finalizeSteer(entry.steer); - entry.resolve(); - continue; + // Idempotent delivery: the transcript already carries these steers + // (an in-run injection persisted, a wake completed, or a duplicate + // deliver raced in). Resolve and flip without re-firing anything. + const stillPending = queue.filter((e) => !steerDelivered(record.messages, e.steer.id)); + for (const entry of queue) { + if (!stillPending.includes(entry)) { + finalizeSteer(entry.steer); + entry.resolve(); + } + } + if (stillPending.length === 0) { + s.queues.delete(chatId); + return; + } + s.queues.set(chatId, stillPending); + + // VEHICLE 1 — THE LIVE RUN: pi's own follow-up queue. The agent of + // the streaming turn takes every marker not already handed to IT; + // pi polls the queue when the run would otherwise stop, injects + // them (message_start/message_end on the live wire), and continues + // the run. Delivery resolves at settlement (settleSteers). Nothing + // else to do in this loop iteration — the turn owns the delivery. + const live = s.liveAgents.get(chatId); + if (live && isTurnRunning(chatId)) { + const handable = stillPending.filter( + (e) => e.handedTo !== live && e.steer.customType !== "probe-handoff", + ); + let handedAny = false; + for (const entry of handable) { + live.followUp(entry.steer); + entry.handedTo = live; + handedAny = true; + } + // probe-handoffs never ride a live run (owner ruling: the + // handoff must START the teacher's run) — if any remain + // pending, wait the run out and loop to the wake below. + const handoffWaits = stillPending.some( + (e) => e.steer.customType === "probe-handoff", + ); + if (handedAny && !handoffWaits) { + return; // the run delivers; settleSteers re-pumps the rest + } + if (handedAny) { + await delay(POLL_MS); + await waitForIdle(chatId); + continue; + } } + // VEHICLE 2 — THE IDLE WAKE: one server-side turn carrying EVERY + // pending marker as its input appends (the handler write-ahead + // persists them under the wake's own turn claim). await waitForIdle(chatId); // Re-fetch AFTER the gates open — waitForIdle can wait out a quiz @@ -304,59 +394,80 @@ async function pump(chatId: string): Promise { // the client's answer + continuation in the write-ahead persist. const fresh = stores.chatStore.getChat(chatId); if (!fresh) { - queue.shift(); - entry.reject(new Error(`chat ${chatId} not found — cannot deliver steer ${steerId}`)); - continue; + continue; // the gone-chat branch rejects next iteration } // The runner is required — chat-handler sets it at init. If some // module instance missed the wiring, fail loudly rather than hang. if (!s.turnRunner) { - queue.shift(); + s.queues.delete(chatId); const err = new Error("wake pipeline: no turn runner wired (setTurnRunner)"); - failSteer(entry.steer, err); - entry.reject(err); - continue; + for (const entry of stillPending) { + failSteer(entry.steer, err); + entry.reject(err); + } + return; } - // 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, steerId)) { - queue.shift(); - finalizeSteer(entry.steer); - entry.resolve(); - continue; + // A live run may have delivered some markers while the gates were + // closed — re-check: only still-missing markers ride the wake. + const fire = stillPending.filter((e) => !steerDelivered(fresh.messages, e.steer.id)); + for (const entry of stillPending) { + if (!fire.includes(entry)) { + finalizeSteer(entry.steer); + entry.resolve(); + } + } + if (fire.length === 0) { + s.queues.delete(chatId); + return; } - // Fire the server-side teacher turn (the steer EVENT: the handler - // appends the marker natively under its own turn claim). A 409 = a - // client turn raced in between idle and firing — back off and - // retry the same steer. - const status = await s.turnRunner(chatId, { type: "steer", steer: entry.steer }).catch((err: unknown) => { - console.error(`[steer ${steerId}] wake turn crashed:`, err); + // DEQUEUE-AT-FIRE (the Wave C lesson from the Wave B regression): + // the carried markers leave the registry BEFORE the turn runs — + // the wake run's OWN follow-up poll must never see its own cargo, + // and a 409 (client raced in) must be able to push them back. + s.queues.delete(chatId); + const steers = fire.map((e) => e.steer); + + // Fire the wake (every marker as the turn's input appends). A 409 = + // a client turn raced in between idle and firing — push the cargo + // back and retry after backoff (the racing turn's own settlement + // re-pumps if it beats us to the re-queue). + const status = await s.turnRunner(chatId, { type: "steers", steers }).catch((err: unknown) => { + console.error(`[steer ${steers.map((x) => x.id).join(",")}] wake turn crashed:`, err); return -1; }); if (status === 409) { + const back = s.queues.get(chatId) ?? []; + back.push(...fire); + s.queues.set(chatId, back); await delay(RETRY_BACKOFF_MS[Math.min(attempt++, RETRY_BACKOFF_MS.length - 1)]!); continue; } attempt = 0; - queue.shift(); if (status === 200) { - finalizeSteer(entry.steer); - entry.resolve(); + for (const entry of fire) { + finalizeSteer(entry.steer); + entry.resolve(); + } } else { const err = new Error(`wake turn failed with status ${status}`); - failSteer(entry.steer, err); - entry.reject(err); + for (const entry of fire) { + failSteer(entry.steer, err); + entry.reject(err); + } } } } finally { s.pumping.delete(chatId); // A steer may have arrived between the last check and the finally — - // restart the pump rather than strand it. - if ((state().queues.get(chatId) ?? []).length > 0) void pump(chatId); + // restart the pump rather than strand it. Entries already handed to + // a live run are NOT stragglers (settleSteers owns their resurrection) + // — restarting for them would just park in waitForIdle forever. + const q = state().queues.get(chatId) ?? []; + if (q.some((e) => e.handedTo === undefined)) void pump(chatId); } } diff --git a/tests/jobs.test.ts b/tests/jobs.test.ts index 025b3aa..ea73374 100644 --- a/tests/jobs.test.ts +++ b/tests/jobs.test.ts @@ -47,7 +47,16 @@ import { composeInstructions, createTeacherAgent } from "../src/agent/teacher-ag import { createTeacherTools } from "../src/agent/tools.ts"; import { webTools } from "../src/agent/tools/web.ts"; import { handleChatRequest } from "../src/server/chat-handler.ts"; -import { abortJob, claimSteers, deliverJobResult, recoverStaleJobs, setJobRespawn, setTurnRunner } from "../src/server/jobs.ts"; +import { + abortJob, + deliverJobResult, + registerLiveAgent, + recoverStaleJobs, + setJobRespawn, + setTurnRunner, + settleSteers, + unregisterLiveAgent, +} from "../src/server/jobs.ts"; import { registerJobAbort, unregisterJobAbort } from "../src/agent/subagents/aborts.ts"; import { beginTurn, endTurn } from "../src/server/turn-liveness.ts"; import { persistTurn } from "../src/server/turn-service.ts"; @@ -708,9 +717,9 @@ describe("wake pipeline (src/server/jobs.ts)", () => { // open question. chatStore.saveMessages(chat.id, nativeToolMessages("teach me calc", "probe_done", "call-pd", { ok: true })); const handoff = buildProbeHandoffSteer(chat.id); - let seen: { chatId: string; steer: SteerMessage } | undefined; + let seen: { chatId: string; steers: SteerMessage[] } | undefined; setTurnRunner(async (cid, input) => { - seen = { chatId: cid, steer: (input as SteerInput).steer as SteerMessage }; + seen = { chatId: cid, steers: (input as SteerInput).steers as SteerMessage[] }; return 200; }); @@ -718,7 +727,8 @@ describe("wake pipeline (src/server/jobs.ts)", () => { await waitFor(() => seen !== undefined, 8000, "handoff wake fired"); const firedHandoff = seen!; assert.equal(firedHandoff.chatId, chat.id); - assert.equal(firedHandoff.steer.id, `steer_probe_${chat.id}`, "the handoff IS the wake turn's input"); + assert.equal(firedHandoff.steers.length, 1); + assert.equal(firedHandoff.steers[0]!.id, `steer_probe_${chat.id}`, "the handoff IS the wake turn's input"); }); it("a redelivered handoff dedupes via the transcript (idempotent by steer id)", async () => { @@ -739,9 +749,9 @@ describe("wake pipeline (src/server/jobs.ts)", () => { 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; steer: SteerMessage } | undefined; + let seen: { chatId: string; steers: SteerMessage[] } | undefined; setTurnRunner(async (cid, input) => { - seen = { chatId: cid, steer: (input as SteerInput).steer as SteerMessage }; + seen = { chatId: cid, steers: (input as SteerInput).steers as SteerMessage[] }; return 200; }); @@ -752,7 +762,8 @@ describe("wake pipeline (src/server/jobs.ts)", () => { assert.equal(jobStore.getJob(jobId)!.activity, "delivered"); assert.ok(seen, "the turn runner fired"); assert.equal(seen.chatId, chatId); - assert.equal(seen.steer.id, steer.id, "the steer IS the wake turn's input"); + assert.equal(seen.steers.length, 1); + assert.equal(seen.steers[0]!.id, steer.id, "the steer IS the wake turn's input"); }); it("waits while a question part is pending, and fires once it clears", async () => { @@ -843,145 +854,119 @@ describe("wake pipeline (src/server/jobs.ts)", () => { // ─── 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 () => { +describe("Wave C: live follow-up handoff + settlement reconciliation", () => { + /** A minimal stand-in for pi's Agent — the pump touches only followUp(). */ + function fakeAgent(): { handed: SteerMessage[]; agent: unknown } { + const handed: SteerMessage[] = []; + return { handed, agent: { followUp: (m: SteerMessage) => handed.push(m) } }; + } + + it("hands queued markers to the LIVE run (pi followUp) — no wake fires, the registry waits", async () => { const { chatId, jobId, steer } = setup(); + const { handed, agent } = fakeAgent(); 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); + // A live turn is streaming (turn registry + live agent registered). + assert.ok(beginTurn(chatId)); + registerLiveAgent(chatId, agent as never); - 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; the handler's steerDelivered filter keeps one copy (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's top-of-loop idempotent skip run - const claimed = claimSteers(chatId); - // The pump itself finalizes already-delivered steers at the top of - // its loop — the queue is empty by the time a racing turn claims. - // (A steer that DID ride the queue gets claimed whole; the HANDLER's - // steerDelivered filter keeps exactly one copy at append.) - assert.equal(claimed.length, 0, "the pump finalized the delivered steer without firing a wake"); - assert.ok(steerDelivered(chatStore.getChat(chatId)!.messages, steer.id), "the record already holds it"); - 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 path: a steer coalesced into the record mid-flight is claimed, filtered at append, never double-fired (duplicate-key bug, 2026-09-18)", async () => { - const { chatId, jobId, steer } = setup(); - // Own runner: this chat's gates are OPEN, so the pump would otherwise - // fire whatever runner the NEXT test registers (leaked counter). - let fired = 0; - setTurnRunner(async () => { - fired++; - return 200; - }); - // The racing turn's write-ahead ALREADY landed the steer in the record - // while the entry is still queued — the exact window the handler - // claims in (the pump dequeues only after the turn returns 200). - const delivery = deliverJobResult(steer); - await delay(300); // the pump is inside waitForIdle's grace (750ms) - const incoming = [...chatStore.getChat(chatId)!.messages, steer]; - chatStore.saveMessages(chatId, incoming); // the racing turn's write-ahead persist - const claimed = claimSteers(chatId); - assert.equal(claimed.length, 1, "the queued steer is claimed — its delivery rides the turn"); - assert.equal(jobStore.getJob(jobId)!.status, "done", "job still finalizes — the claim resolves the delivery"); - await delivery; - assert.deepEqual(claimSteers(chatId), [], "queue is drained"); - // The handler appends only steers NOT already in the record - // (steerDelivered) — the write-ahead keeps exactly ONE copy of the - // steer id, and the pump's post-idle re-check reads the RECORD and - // skips the delivered id. - const appends = claimed.filter((c) => !steerDelivered(chatStore.getChat(chatId)!.messages, c.id)); - assert.equal(appends.length, 0, "the already-persisted steer never appends again"); - const outgoing = [...chatStore.getChat(chatId)!.messages, ...appends]; - const ids = outgoing.map((m) => m.id); - assert.equal(ids.filter((id) => id === steer.id).length, 1, "exactly one steer copy persists"); - await delay(800); // the pump's grace elapses — its re-check sees the delivered id - assert.equal(fired, 0, "the claimed-and-persisted steer never double-fires a wake"); - }); - - 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 dangling question call parks the pump in waitForIdle. - chatStore.saveMessages(chatId, [ - { id: "u1", role: "user", content: "hello", timestamp: 1 } as never, - { - id: "a1", - role: "assistant", - content: [{ type: "toolCall", id: "quiz-call", name: "quiz", arguments: { question: "q", choices: ["a", "b"] } as never }], - api: "openai-completions", - provider: "openrouter", - model: "test", - usage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, - stopReason: "toolUse", - timestamp: 1, - } as never, - ]); - const delivery = deliverJobResult(steer); + const delivery = deliverJobResult(steer); // floats — resolves at settlement + await waitFor(() => handed.length === 1, 8000, "marker handed to the live agent"); 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 answeredRecord = chatStore.getChat(chatId)!; - chatStore.saveMessages(chatId, [ - ...answeredRecord.messages, - { - id: "r1", - role: "toolResult", - toolCallId: "quiz-call", - toolName: "quiz", - content: [{ type: "text", text: "answered" }], - details: { answer: "a", correct: true, explanation: "" }, - isError: false, - timestamp: 1, - } as never, - ]); + assert.equal(fired, 0, "no wake while the run is live"); + assert.equal(jobStore.getJob(jobId)!.status, "running", "job flips only at delivery"); + // Settlement: the run persisted the injected marker. + unregisterLiveAgent(chatId); + settleSteers(chatId, [steer as never]); await delivery; - await delay(1500); - assert.equal(fired, 0, "no duplicate wake after coalescing"); + assert.equal(jobStore.getJob(jobId)!.status, "done", "delivery resolves at settlement"); + assert.equal(fired, 0, "never needed a wake"); + endTurn(chatId); + }); + + it("a handed marker the run never injected (stopper exit / missed poll) strands at settlement — the idle wake delivers it", async () => { + const { chatId, jobId, steer } = setup(); + const { handed, agent } = fakeAgent(); + let fired = 0; + setTurnRunner(async () => { + fired++; + return 200; + }); + + assert.ok(beginTurn(chatId)); + registerLiveAgent(chatId, agent as never); + + const delivery = deliverJobResult(steer); + await waitFor(() => handed.length === 1, 8000, "handed"); + // The run settles WITHOUT the marker (question stopper exited before + // the follow-up poll) — reconcile against the persisted truth. + unregisterLiveAgent(chatId); + settleSteers(chatId, []); + endTurn(chatId); + + // The marker returned to pending — the pump's idle wake delivers it. + await delivery; + assert.equal(fired, 1, "the stranded marker rode the idle wake"); assert.equal(jobStore.getJob(jobId)!.status, "done"); }); + + it("a probe-handoff NEVER rides a live run — it waits the turn out and wakes", async () => { + const chat = chatStore.createChat("Test: handoff never live"); + trackChat(chat.id); + chatStore.saveMessages(chat.id, nativeToolMessages("teach me calc", "probe_done", "call-pd", { ok: true })); + const handoff = buildProbeHandoffSteer(chat.id); + const { handed, agent } = fakeAgent(); + const fired: SteerMessage[][] = []; + setTurnRunner(async (_cid, input) => { + fired.push((input as SteerInput).steers as SteerMessage[]); + return 200; + }); + + assert.ok(beginTurn(chat.id)); + registerLiveAgent(chat.id, agent as never); + + const delivery = deliverJobResult(handoff); + await delay(1200); + assert.equal(handed.length, 0, "the handoff was NOT handed to the live agent"); + assert.equal(fired.length, 0, "no wake while the prober's turn streams"); + + // The prober's turn settles — now the handoff wakes the TEACHER. + unregisterLiveAgent(chat.id); + settleSteers(chat.id, []); + endTurn(chat.id); + await delivery; + assert.equal(fired.length, 1); + assert.equal(fired[0]![0]!.id, handoff.id, "the handoff rode the wake"); + }); + + it("the idle wake carries EVERY pending marker in ONE turn", async () => { + const { chatId, steer } = setup(); + const job2 = jobStore.createJob({ kind: "research", chatId, toolCallId: "call-2" }); + trackJob(job2.id); + const result: ResearchResult = { brief: "Second result.", sources: [] }; + jobStore.updateJob(job2.id, { result }); + const steer2 = buildSubagentSteer(job2, result); + + const fired: SteerMessage[][] = []; + setTurnRunner(async (_cid, input) => { + fired.push((input as SteerInput).steers as SteerMessage[]); + return 200; + }); + + await Promise.all([deliverJobResult(steer), deliverJobResult(steer2)]); + assert.equal(fired.length, 1, "one wake for the whole queue"); + assert.equal(fired[0]!.length, 2, "both markers ride the same turn"); + const ids = fired[0]!.map((s) => s.id).sort(); + assert.deepEqual(ids, [steer.id, steer2.id].sort()); + }); }); + it("fails the job (error status) when the wake turn errors unrecoverably", async () => { const { jobId, steer } = setup(); setTurnRunner(async () => 500); @@ -1238,7 +1223,7 @@ describe("abort (src/server/jobs.ts abortJob)", () => { const { jobId } = newChatWithResearchJob("abort running job", "call-abort1"); const fired: SteerMessage[] = []; setTurnRunner(async (_cid, input) => { - fired.push((input as SteerInput).steer as SteerMessage); + fired.push(...((input as SteerInput).steers as SteerMessage[])); return 200; }); researchModelOverride.current = hangingModel(); @@ -1316,7 +1301,7 @@ describe("abort (src/server/jobs.ts abortJob)", () => { assert.ok(beginTurn(chat.id)); const fired: SteerMessage[] = []; setTurnRunner(async (_cid, input) => { - fired.push((input as SteerInput).steer as SteerMessage); + fired.push(...((input as SteerInput).steers as SteerMessage[])); return 200; }); void deliverJobResult(buildSubagentSteer(job, result)).catch(() => { diff --git a/tests/session-core.test.ts b/tests/session-core.test.ts index 6bb68b9..acb8579 100644 --- a/tests/session-core.test.ts +++ b/tests/session-core.test.ts @@ -394,6 +394,46 @@ describe("send (one input event, verbatim consumption)", () => { await done; }); +it("Wave C: markers injected mid-run (custom role, follow-up delivery) append to the tail by id", () => { + // A never-settling fetch: the send only needs to flip the model into its + // streaming status synchronously — its POST floats (never rejects) while + // the injected wire events are asserted. + const m = model({ fetch: () => new Promise(() => {}) }); + m.adoptRecord([ + { id: "u1", role: "user", content: "hello", timestamp: 1 } as never, + ] as never); + void m.send({ type: "text", text: "hello" }); + + // pi emits message_start + message_end for an injected follow-up marker + // (no streaming partials) — both must UPSERT the same tail slot by id. + const marker = { + id: "steer_job_x_1", + role: "custom", + customType: "subagent-steer", + content: "The research result arrived.", + display: true, + details: { jobId: "job_x", kind: "research" }, + timestamp: 2, + }; + m.applyWireEvent({ type: "message_start", message: marker } as never); + m.applyWireEvent({ type: "message_end", message: marker } as never); + + const state = m.getState(); + const ids = state.messages.map((v) => v.parts.length); + assert.ok(state.messages.length >= 2, "the marker produced a view message"); + const tail = state.messages.at(-1)!; + assert.equal(tail.role, "user", "a steer marker projects as a user-role view message"); + assert.ok( + tail.parts.some((p) => p.type === "data-subagent-artifact"), + "the marker renders its artifact part (SubagentSteerCard)", + ); + + // A re-delivery of the same id (message_end replay) must NOT duplicate. + m.applyWireEvent({ type: "message_end", message: marker } as never); + const again = m.getState(); + assert.equal(again.messages.length, state.messages.length, "same id upserts, never appends a second copy"); +}); + it("a provider error rides VERBATIM (owner ruling) — no masking anywhere", async () => { globalThis.fetch = (async () => sseResponse(