From ad0a6e5201c502d4297190cf8ad2edbb16d9dbba Mon Sep 17 00:00:00 2001 From: amosblomqvist Date: Sat, 26 Sep 2026 04:31:07 +0300 Subject: [PATCH] feat: Wave C pivot - steer markers, not follow-ups Owner correction (2026-09-26, later): a real lesson is ONE continuous teach->quiz->teach run - there are no run-would-stop moments mid-lesson, and the run ends that DO exist are quiz stopper exits where delivery must not fire. Subagent results must land BETWEEN messages INSIDE the run, never splitting a toolCall from its toolResult. Vehicle 1 now hands markers to agent.steer() - pi's STEERING queue: the loop drains it at TURN BOUNDARIES (after a completed assistant message AND its tool results - the bonded pair survives injection) and again at RUN START (a continuation drains before its first LLM call), then CONTINUES the run with the marker in context. steeringMode "all" replaces followUpMode. registerLiveAgent moved BEFORE runPiTurn and now hands pending markers itself: the start poll drains the queue SYNCHRONOUSLY inside the runPiTurn call, so a later registration would miss it. A marker stranded by a quiz-terminated run therefore rides the ANSWER run's first LLM call - the native replacement for the old continuation coalescing. Settlement reconciliation, the idle wake, probe-handoff exclusion, and the registry-as-truth design are UNCHANGED; stranding is now exactly the stopper-exit case. Tests: jobs 45/45 (fakeAgent.steer; new pin - registration steers pending markers into the new run, parked wake never fires); chat-flow 24/24 (new e2e through the real composition - a gated mid-run turn, a marker delivered while the turn streams, record shape [call, result, MARKER, response] with the marker adjacent to the completed unit, the model SEES the marker in its next prompt, zero wakes); full suite 999/999, tsc/lint/build green. Live smoke (PORT=3004): research dispatched in the probe turn parked behind the dangling quiz; on the quiz answer the marker injected at the answer run's START poll - record [answer result, marker, teacher response], job flipped done/delivered at settlement, brief card rendered with its sources toggle. Plan doc: docs/PI-NATIVE-MIGRATION-PLAN.md gains the pivot section. Deferred (owner list): thinking blocks rendering after messages; vanishing tool-call output tokens. --- docs/PI-NATIVE-MIGRATION-PLAN.md | 33 ++++++++++ src/agent/teacher-agent.ts | 14 +++-- src/server/chat-handler.ts | 14 +++-- src/server/jobs.ts | 100 ++++++++++++++++++------------- tests/chat-flow.test.ts | 78 ++++++++++++++++++++++++ tests/jobs.test.ts | 55 +++++++++++++++-- 6 files changed, 239 insertions(+), 55 deletions(-) diff --git a/docs/PI-NATIVE-MIGRATION-PLAN.md b/docs/PI-NATIVE-MIGRATION-PLAN.md index aafb84e..350c08e 100644 --- a/docs/PI-NATIVE-MIGRATION-PLAN.md +++ b/docs/PI-NATIVE-MIGRATION-PLAN.md @@ -589,3 +589,36 @@ again — Wave C is the structural migration onto pi's own delivery seams. 2. Client `applyWireEvent` non-assistant appends. 3. Pump simplification + `claimSteers` deletion + handler cleanup. 4. Test migration (jobs, chat-flow, session-core) + live smoke. + +## Wave C pivot (owner correction, later 2026-09-26): follow-ups → STEERING + +The owner re-examined the follow-ups-only ruling and overturned it: a real +learning session is ONE continuous run — teach a thing, verify it with a +quiz tool call, begin teaching the next thing. There are no "run would +otherwise stop" moments mid-lesson, and the few run ends that do exist are +exactly the ones where delivery must NOT happen (the question stopper). So +subagent results must be delivered MID-RUN, BETWEEN messages. + +The pivot (all in the same two-vehicle skeleton): + +- Vehicle 1 hands markers to `agent.steer()` — pi's STEERING queue. The + loop drains it at TURN BOUNDARIES: after a completed assistant message + AND its tool results (never between a toolCall and its toolResult — the + bonded pair survives injection), and again at RUN START (a continuation + drains the queue BEFORE its first LLM call). The run CONTINUES with the + marker in context. +- `steeringMode: "all"` (was `followUpMode`) — every handed marker joins + one boundary injection. +- `registerLiveAgent` moved BEFORE `runPiTurn` in the handler and now + hands pending markers itself: the loop's start poll drains the queue + SYNCHRONOUSLY inside the `runPiTurn` call, so a later registration would + miss it. A marker stranded by a quiz-terminated run therefore rides the + ANSWER run's first LLM call — the native replacement for the old + continuation coalescing. +- Settlement reconciliation, the idle wake, probe-handoff exclusion, and + the registry-as-truth design are UNCHANGED (stranding is now exactly the + stopper-exit case; the wake still covers a cold-and-idle queue). + +Deferred after the pivot (owner's list): thinking blocks sometimes +rendering after messages; output tokens of certain tool calls vanishing +with no tool call actually happening. diff --git a/src/agent/teacher-agent.ts b/src/agent/teacher-agent.ts index f379818..9ab012c 100644 --- a/src/agent/teacher-agent.ts +++ b/src/agent/teacher-agent.ts @@ -165,12 +165,14 @@ 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 C (2026-09-26, owner correction same day): markers handed to a + // live run drain from pi's STEERING queue — the lesson loop is one + // continuous teach→quiz→teach run, so delivery must inject at turn + // boundaries INSIDE the run, not wait for it to stop (follow-ups). + // "all" joins every handed marker at one boundary, matching the idle + // wake's carry-everything semantics (the default "one-at-a-time" + // would spend one teacher turn per marker). + steeringMode: "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/server/chat-handler.ts b/src/server/chat-handler.ts index 2742c58..d4b4498 100644 --- a/src/server/chat-handler.ts +++ b/src/server/chat-handler.ts @@ -315,6 +315,15 @@ export async function handleChatInput( // is just the FIRST subscriber (a view). A client disconnect can no // longer abort the turn; the remote kill is the abort channel, and // late views reattach via GET /api/chats/[id]/stream. + // Wave C (owner correction, 2026-09-26 — steering): register the live + // agent BEFORE runPiTurn — the loop's start steering-poll drains the + // queue SYNCHRONOUSLY inside that call, so a later registration would + // miss it. Registration also steers any pending registry markers into + // this run: a marker parked behind a just-answered quiz rides THIS + // run's first LLM call (the native replacement for continuation + // coalescing). Settlement still cannot beat the flow below: it needs + // a streamed response, which never happens inside the constructor. + registerLiveAgent(chatId, constructed); const handle = runPiTurn({ agent: constructed, // Stream-level failures are provider/transport internals — masked, @@ -373,11 +382,6 @@ 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); diff --git a/src/server/jobs.ts b/src/server/jobs.ts index 70c5336..35435a2 100644 --- a/src/server/jobs.ts +++ b/src/server/jobs.ts @@ -8,18 +8,26 @@ * The steer joins a per-chat serialized registry and the pump drives * it along ONE of two paths: * - * 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). + * 1. LIVE RUN → pi steering: the marker is handed to the LIVE + * agent's steering queue (agent.steer). pi polls that queue at + * TURN BOUNDARIES — after a completed assistant message AND its + * tool results (never between a toolCall and its result), and + * again at the run's start (a continuation drains the queue + * BEFORE its first LLM call) — and CONTINUES THE SAME RUN: the + * marker joins the context mid-lesson, rides the live event + * stream (message_start/message_end), and lands in the record + * at that run's persist. This is the owner's 2026-09-26 + * correction: a real lesson is one continuous teach→quiz→teach + * loop, so delivery must happen BETWEEN messages inside the + * run — waiting for the run to stop (follow-ups) only ever + * fires at a stopper exit, which is exactly when it must not. + * The question stopper runs BEFORE the post-turn poll and + * exits the loop, so a quiz-terminated run never injects: the + * handed marker strands (see settleSteers) and waits — the + * ANSWER run then picks it up at its start poll. 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 @@ -75,10 +83,10 @@ 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. */ + /** The live agent this marker was steered into (agent.steer) — set by + * the hand-off helper. Guards double-handoff (a re-pumped loop or a + * re-registered run must not re-queue what a run already holds) and is + * CLEARED at settlement for markers the dying run never injected. */ handedTo?: Agent; } @@ -174,15 +182,34 @@ export function deliverJobResult(steer: SteerMessage): Promise { }); } -/** 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. */ +/** Hand every eligible pending marker to a live agent's steering queue + * (idempotent per agent via handedTo). Called from BOTH the pump's live + * path and registerLiveAgent — a parked marker (stranded behind a quiz) + * must not wait for the pump loop to notice the new run: the handler + * registers the answer run's agent BEFORE runPiTurn, so this hand-off + * beats the loop's synchronous start-poll drain and the marker rides the + * run's FIRST LLM call (the native equivalent of the old continuation + * coalescing). */ +function handOffEligible(chatId: string, agent: Agent): void { + const queue = state().queues.get(chatId); + if (!queue || queue.length === 0) return; + for (const entry of queue) { + if (entry.handedTo !== agent && entry.steer.customType !== "probe-handoff") { + agent.steer(entry.steer); + entry.handedTo = agent; + } + } +} + +/** The chat-handler registers the CURRENTLY STREAMING turn's agent BEFORE + * runPiTurn (the loop's start steering-poll drains synchronously inside + * that call — a later registration would miss it). Pending registry + * markers are steered into the run right here. The settle seam cannot + * beat the registration: settlement needs a streamed response, which + * never happens inside the constructor call. */ export function registerLiveAgent(chatId: string, agent: Agent): void { state().liveAgents.set(chatId, agent); + handOffEligible(chatId, agent); } /** Clear the live agent at settlement — MUST run before settleSteers so the @@ -351,33 +378,26 @@ async function pump(chatId: string): Promise { } 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. + // VEHICLE 1 — THE LIVE RUN: pi's steering queue. The agent of the + // streaming turn takes every eligible marker; pi injects at the + // next turn boundary (message_start/message_end on the live wire, + // between completed assistant+toolResult units) 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; - } + handOffEligible(chatId, live); + const handable = stillPending.filter((e) => e.handedTo === live); // 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) { + if (handable.length > 0 && !handoffWaits) { return; // the run delivers; settleSteers re-pumps the rest } - if (handedAny) { + if (handable.length > 0) { await delay(POLL_MS); await waitForIdle(chatId); continue; diff --git a/tests/chat-flow.test.ts b/tests/chat-flow.test.ts index 470fc22..050da3e 100644 --- a/tests/chat-flow.test.ts +++ b/tests/chat-flow.test.ts @@ -389,6 +389,84 @@ describe("teacher chat flow (scripted model)", () => { assert.equal((record.messages[at + 1] as { role?: string } | undefined)?.role, "assistant", "the wake response follows the marker"); }); + it("Wave C steering: a mid-run result injects BETWEEN turn units — never splitting a toolCall from its toolResult", { timeout: 15000 }, async () => { + // THE OWNER'S CORRECTION (2026-09-26): a real lesson is ONE continuous + // teach→quiz→teach run, so a subagent result must land BETWEEN messages + // inside the run — pi's steering queue, drained at turn boundaries + // (after a completed assistant message AND its tool results). Waiting + // for run end (follow-ups) can never fire mid-lesson: the question + // stopper pre-empts that poll at exactly the moments a run DOES end. + const chat = createConfiguredChat("Test: mid-run steering"); + createdChats.push(chat.id); + + // Step 1 (gated): a non-question tool — an unknown name yields an + // error toolResult WITHOUT terminating, so the run continues. Step 2: + // the reaction AFTER the boundary injection (records its prompt). + let release!: () => void; + const gate = new Promise((r) => (release = r)); + const seen: string[] = []; + const faux = createFauxCore({}); + faux.setResponses([ + async () => { + await gate; + return fauxAssistantMessage( + [fauxToolCall("no_such_tool", { note: "x" }, { id: "call-mid" })], + { stopReason: "toolUse" }, + ); + }, + (context) => { + seen.push(context.messages.map((m) => JSON.stringify((m as { content: unknown }).content)).join("\n")); + return fauxAssistantMessage([fauxText("Folded the marker in.")]); + }, + ]); + const engine: PiEngine = { model: faux.getModel(), streamFn: faux.streamSimple }; + + // A wake here would be a BUG (delivery must ride the live run) — the + // counter keeps an erroneous wake hermetic and assertable. + let wakes = 0; + setTurnRunner(async () => { + wakes++; + return 200; + }); + + // The turn floats mid-run: step 1 is gated, the run is live. + const res = handleChatRequest(post(chat.id, { type: "text", text: "teach" }), agentFactory(engine)); + await waitFor(() => isTurnRunning(chat.id), 8000, "the turn is live"); + + // The result lands MID-RUN: the pump hands it to the live agent's + // steering queue (vehicle 1) — the queued marker waits for the + // in-flight turn to complete so the injection lands at its boundary. + const job = jobStore.createJob({ kind: "research", chatId: chat.id, toolCallId: "call-mid" }); + const steer = buildSubagentSteer(job, { brief: "Mid-run brief.", sources: [] }); + const delivery = deliverJobResult(steer); + await new Promise((r) => setTimeout(r, 300)); // vehicle 1 hands off + assert.equal(wakes, 0, "no wake — the live run owns the delivery"); + + release(); // step 1 completes → tool result → TURN BOUNDARY → inject → step 2 + await drain(await res); + await delivery; // resolves at settleSteers (the run persisted the marker) + + // Record shape: the marker sits BETWEEN completed turn units — after + // the toolResult it must never split, adjacent to it. + const record = chatStore.getChat(chat.id)!; + const at = (role: string, pred: (m: LhMessage) => boolean) => + record.messages.findIndex((m) => (m as { role?: string }).role === role && pred(m as never)); + const callAt = record.messages.findIndex( + (m) => (m as { role?: string }).role === "assistant" && + JSON.stringify((m as { content?: unknown }).content).includes("call-mid"), + ); + const resultAt = at("toolResult", (m) => (m as { toolCallId?: string }).toolCallId === "call-mid"); + const steerAt = record.messages.findIndex((m) => (m as { id?: string }).id === steer.id); + const finalAt = at("assistant", (m) => JSON.stringify((m as { content?: unknown }).content).includes("Folded the marker in.")); + assert.ok(callAt !== -1 && resultAt !== -1 && steerAt !== -1 && finalAt !== -1, `all units present: ${callAt}/${resultAt}/${steerAt}/${finalAt}`); + assert.ok(callAt < resultAt, "the toolCall precedes its toolResult (bonded pair)"); + assert.equal(steerAt, resultAt + 1, "the marker injected at the turn boundary — immediately after the completed unit"); + assert.ok(steerAt < finalAt, "the run CONTINUED after the injection (same run, no wake)"); + assert.equal(record.messages.filter((m) => (m as { id?: string }).id === steer.id).length, 1, "exactly one marker copy"); + assert.ok(seen[0]!.includes("Mid-run brief."), "the model SAW the marker in its next prompt"); + assert.equal(wakes, 0, "delivery rode the live run end to end"); + }); + it("a cancelled view NEVER aborts the turn — it runs to completion (owner ruling 2026-09-25)", async () => { let release!: () => void; const gate = new Promise((r) => (release = r)); diff --git a/tests/jobs.test.ts b/tests/jobs.test.ts index ea73374..b68ad42 100644 --- a/tests/jobs.test.ts +++ b/tests/jobs.test.ts @@ -854,14 +854,14 @@ describe("wake pipeline (src/server/jobs.ts)", () => { // ─── continuation coalescing (the delivery-starvation fix) ─────────────────── -describe("Wave C: live follow-up handoff + settlement reconciliation", () => { - /** A minimal stand-in for pi's Agent — the pump touches only followUp(). */ +describe("Wave C: live steering handoff + settlement reconciliation", () => { + /** A minimal stand-in for pi's Agent — the pump touches only steer(). */ function fakeAgent(): { handed: SteerMessage[]; agent: unknown } { const handed: SteerMessage[] = []; - return { handed, agent: { followUp: (m: SteerMessage) => handed.push(m) } }; + return { handed, agent: { steer: (m: SteerMessage) => handed.push(m) } }; } - it("hands queued markers to the LIVE run (pi followUp) — no wake fires, the registry waits", async () => { + it("hands queued markers to the LIVE run (pi steering queue) — no wake fires, the registry waits", async () => { const { chatId, jobId, steer } = setup(); const { handed, agent } = fakeAgent(); let fired = 0; @@ -944,6 +944,53 @@ describe("Wave C: live follow-up handoff + settlement reconciliation", () => { assert.equal(fired[0]![0]!.id, handoff.id, "the handoff rode the wake"); }); + it("registerLiveAgent immediately steers pending markers into the NEW run (the parked-behind-quiz path)", async () => { + const { chatId, jobId, steer } = setup(); + const { handed, agent } = fakeAgent(); + let fired = 0; + setTurnRunner(async () => { + fired++; + return 200; + }); + + // A dangling question has the pump parked in waitForIdle: the + // marker is pending but no live run exists to take it. + chatStore.saveMessages(chatId, [ + ...nativeToolMessages("teach me", "probe_done", "call-q", { ok: true }), + { + id: "a-q", + 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); + await delay(300); // let the pump park behind the dangling question + assert.equal(handed.length, 0, "nothing to hand to — no live agent"); + + // The learner answers: the handler registers the answer run's agent + // BEFORE runPiTurn, so the marker is steered into it synchronously — + // it rides the run's START steering poll (first LLM call). + assert.ok(beginTurn(chatId)); + registerLiveAgent(chatId, agent as never); + assert.equal(handed.length, 1, "registration itself steered the marker in"); + assert.equal(handed[0]!.id, steer.id); + + // The answer run persisted the marker — the parked pump's wake + // re-check resolves it without ever firing a turn. + unregisterLiveAgent(chatId); + settleSteers(chatId, [steer as never]); + endTurn(chatId); + await delivery; + assert.equal(fired, 0, "the parked wake never fired — the run delivered"); + assert.equal(jobStore.getJob(jobId)!.status, "done"); + }); + 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" });