From 7490587e507c0e204ee3a9ce92019e7ffd7ae11f Mon Sep 17 00:00:00 2001 From: danielkuykendall23-boop Date: Thu, 24 Sep 2026 16:26:37 -0400 Subject: [PATCH] fix(daemon): mirror Pi replies from assistantMessageEvent deltas Pi 0.84.0 changed JSON and RPC message_update events to carry only the assistantMessageEvent delta and dropped the cumulative message, so the pi-runner mirrored no assistant text and Pi seats (and rig transcript) showed only the user's lines and tool summaries. Mirror text_delta deltas instead. Older Pi sent the same deltas next to the cumulative message, so this also stops the reply from being appended again on every update there. Thinking and tool-call argument deltas stay out of the pane, matching the previous text-only mirror. --- packages/daemon/src/adapters/pi-runner.ts | 32 +++--------- packages/daemon/test/pi-runner.test.ts | 62 ++++++++++++++++++----- 2 files changed, 58 insertions(+), 36 deletions(-) diff --git a/packages/daemon/src/adapters/pi-runner.ts b/packages/daemon/src/adapters/pi-runner.ts index 04975032f..c97cf0df5 100644 --- a/packages/daemon/src/adapters/pi-runner.ts +++ b/packages/daemon/src/adapters/pi-runner.ts @@ -82,29 +82,6 @@ export interface MirrorAndActivity { streaming?: boolean; } -/** Tolerant text extraction from a Pi message shape (string, {text}, or - * content-block arrays). Exact field calibration is a VM-proof concern; this - * covers the documented shapes without throwing on unknowns. */ -export function extractMessageText(message: unknown): string { - if (typeof message === "string") return message; - if (message === null || typeof message !== "object") return ""; - const m = message as Record; - if (typeof m.text === "string") return m.text; - if (Array.isArray(m.content)) { - return m.content - .map((block) => { - if (typeof block === "string") return block; - if (block !== null && typeof block === "object" && typeof (block as Record).text === "string") { - return (block as Record).text as string; - } - return ""; - }) - .join(""); - } - if (typeof m.content === "string") return m.content; - return ""; -} - export function mapPiEvent(event: Record): MirrorAndActivity { const type = typeof event.type === "string" ? event.type : ""; switch (type) { @@ -117,7 +94,14 @@ export function mapPiEvent(event: Record): MirrorAndActivity { case "message_start": return { mirrorLines: [] }; case "message_update": { - const delta = extractMessageText(event.message ?? event.delta); + // Pi streams assistant text as `assistantMessageEvent` text_delta + // records. Only the delta is mirrored: Pi 0.84.0 dropped the cumulative + // `message` from RPC message_update, and older Pi sent it alongside the + // same deltas, so appending it repeated the growing text. Thinking and + // tool-call argument deltas stay out of the pane; tool calls get their + // own one-line summaries from tool_execution_*. + const update = event.assistantMessageEvent as Record | undefined; + const delta = update?.type === "text_delta" && typeof update.delta === "string" ? update.delta : ""; return delta ? { mirrorLines: [], mirrorAppend: delta } : { mirrorLines: [] }; } case "message_end": { diff --git a/packages/daemon/test/pi-runner.test.ts b/packages/daemon/test/pi-runner.test.ts index a558ed58f..c4289772f 100644 --- a/packages/daemon/test/pi-runner.test.ts +++ b/packages/daemon/test/pi-runner.test.ts @@ -6,7 +6,7 @@ import { describe, it, expect, vi } from "vitest"; import { - PasteAggregator, RunnerCore, mapPiEvent, extractMessageText, parseRunnerArgs, + PasteAggregator, RunnerCore, mapPiEvent, parseRunnerArgs, prepareRunnerSidecar, type RunnerIo, } from "../src/adapters/pi-runner.js"; @@ -171,6 +171,37 @@ describe("RunnerCore identity + sidecar", () => { // ── event → mirror / activity mapping ──────────────────────────────────────── +// Shapes from a real `pi --mode rpc` 0.87.1 run (OpenRig's child argv) that +// streamed a reply from a local OpenAI-compatible mock. Since Pi 0.84.0, +// message_update carries only `usage` and the `assistantMessageEvent` delta. +const PI_USAGE = { + input: 12, output: 5, cacheRead: 0, cacheWrite: 0, reasoning: 0, totalTokens: 17, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, +}; +const PI_REPLY = { + role: "assistant", + content: [ + { type: "thinking", thinking: "User wants a greeting.", thinkingSignature: "reasoning_content" }, + { type: "text", text: "Hello from the mock server." }, + ], + api: "openai-completions", provider: "mock", model: "mock-model", usage: PI_USAGE, + stopReason: "stop", timestamp: 1790281248272, responseId: "chatcmpl-mock", +}; +const piUpdate = (assistantMessageEvent: Record) => + ({ type: "message_update", usage: PI_USAGE, assistantMessageEvent }); +const PI_REPLY_EVENTS = [ + { type: "message_start", message: { ...PI_REPLY, content: [], stopReason: "pending" } }, + piUpdate({ type: "thinking_start", contentIndex: 0 }), + piUpdate({ type: "thinking_delta", contentIndex: 0, delta: "User wants a greeting." }), + piUpdate({ type: "text_start", contentIndex: 1 }), + piUpdate({ type: "text_delta", contentIndex: 1, delta: "Hello" }), + piUpdate({ type: "text_delta", contentIndex: 1, delta: " from the" }), + piUpdate({ type: "text_delta", contentIndex: 1, delta: " mock server." }), + piUpdate({ type: "thinking_end", contentIndex: 0, content: "User wants a greeting." }), + piUpdate({ type: "text_end", contentIndex: 1, content: "Hello from the mock server." }), + { type: "message_end", message: PI_REPLY }, +]; + describe("mapPiEvent", () => { it("agent_start/agent_end drive streaming + running/idle activity", () => { expect(mapPiEvent({ type: "agent_start" })).toMatchObject({ @@ -181,10 +212,14 @@ describe("mapPiEvent", () => { }); }); - it("message_update deltas append to the mirror; message_end terminates the line", () => { - const update = mapPiEvent({ type: "message_update", message: { content: [{ type: "text", text: "hel" }] } }); - expect(update.mirrorAppend).toBe("hel"); - expect(mapPiEvent({ type: "message_end" }).mirrorLines).toEqual([""]); + it("pre-0.84 message_update appends only the delta, never the cumulative message", () => { + const partial = { ...PI_REPLY, content: [{ type: "text", text: "Hello from the" }], stopReason: "pending" }; + const legacy = mapPiEvent({ + type: "message_update", + message: partial, + assistantMessageEvent: { type: "text_delta", contentIndex: 1, delta: " from the", partial }, + }); + expect(legacy.mirrorAppend).toBe(" from the"); }); it("tool executions render compact one-line summaries + PreToolUse activity", () => { @@ -199,14 +234,17 @@ describe("mapPiEvent", () => { expect(mapPiEvent({ type: "compaction_start" }).activity).toEqual({ hookEvent: "active", subtype: "compaction" }); expect(mapPiEvent({ type: "auto_retry_start" }).activity).toEqual({ hookEvent: "active", subtype: "auto_retry" }); }); +}); - it("extractMessageText tolerates string / {text} / content-block shapes", () => { - expect(extractMessageText("plain")).toBe("plain"); - expect(extractMessageText({ text: "t" })).toBe("t"); - expect(extractMessageText({ content: [{ type: "text", text: "a" }, { type: "text", text: "b" }] })).toBe("ab"); - expect(extractMessageText({ content: "inline" })).toBe("inline"); - expect(extractMessageText(null)).toBe(""); - expect(extractMessageText({ weird: true })).toBe(""); +describe("RunnerCore assistant reply mirror", () => { + it("streams a Pi 0.84+ reply from text deltas and ends the line on message_end", () => { + const f = fakeIo(); + const { core } = readyCore(f); + let pane = ""; + f.io.mirrorLine = (line) => { pane += `${line}\n`; }; + f.io.mirrorAppend = (text) => { pane += text; }; + for (const event of PI_REPLY_EVENTS) core.handlePiLine(JSON.stringify(event)); + expect(pane).toBe("Hello from the mock server.\n"); }); });