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" });