fix(daemon): pi-runner stream heartbeat keeps long-running inference reading as working (#2336)

A Pi seat mid-turn now keeps reading `working` through a long, quiet model request (local inference, a slow provider) instead of flipping to idle or unknown after 15 seconds. The Pi runner already had an idle heartbeat; it now also posts an `active` beat while its `streaming` flag says a turn is in flight. The heartbeat timer ticks every 5 seconds and a beat goes out once 5 seconds have passed since the runner's last post, real or beat, so a working hook stays inside the seat-activity oracle's 15-second `HOOK_AUTHORITY_WINDOW_MS` on ordinary on-time ticks. The beat stops when `streaming` clears at the turn's end, an abort or error, and with the runner's exit; frequent real events postpone it, and the idle heartbeat's three-minute gate is unchanged. OMP keeps its no-op timer. `main()` starts the same exported timer the new fake-timer tests drive.

Runner +55/-13, tests +167/-4.

packages/daemon/src/adapters/pi-runner.ts

@@ -218,8 +218,10 @@ export function mapPiEvent(event: Record<string, unknown>, runtime: RunnerRuntim
       // Pi keeps main's mapping: turn_end carries no notice.
       return runtime === "omp" ? { mirrorLines: [], ...modelErrorNotice(event, runtime) } : { mirrorLines: [] };
     case "turn_start":
-      // Some turn paths omit agent_start; turn_start still opens a turn, so it reports the seat working.
-      return { mirrorLines: [], activity: { hookEvent: "active", subtype: "turn_start" } };
+      // Some turn paths omit agent_start; turn_start still opens a turn, so it reports the seat
+      // working and opens the streaming flag — the beat must cover a turn whose only opening
+      // event was turn_start, or a long quiet turn on that path still loses its working state.
+      return { mirrorLines: [], activity: { hookEvent: "active", subtype: "turn_start" }, streaming: runtime === "omp" ? undefined : true };
     case "message_start":
       return { mirrorLines: [] };
     case "message_update": {
@@ -387,6 +389,9 @@ export class RunnerCore {
   private lastEntryId: string | undefined;
   private ready = false;
   private lastHeartbeatMs: number | null = null;
+  /** When this runner last posted any activity payload, so the streaming beat spaces itself
+   *  behind real events (a tool call is evidence enough; the beat covers the silent gaps). */
+  private lastActivityPostMs: number | null = null;
   /** Hook event of the last activity this runner posted; null until it posts one. */
   private lastPostedHook: string | null = null;
   private assistantErrorShown = false;
@@ -712,17 +717,37 @@ export class RunnerCore {
     this.io.postActivity(this.activityPayload("Notification", next === "transient_turn_error" ? "runtime_error" : next));
   }
 
-  /** Idle heartbeat. A Pi runner only reports when something happens, and the daemon reads any hook older than
-   *  its freshness window as `unknown` (`stale_runtime_hook`), so an idle seat read `unknown` five minutes after
-   *  its last turn, and a restored seat read `unknown` (`generation_mismatch`) until its first turn. Re-asserting
-   *  idle while ready, not mid-turn and not waiting on the operator keeps the verdict honest: a runner that dies
-   *  stops beating and the seat goes back to `unknown`. Pi only; the OMP identity handshake has its own retry. */
+  /** Idle and streaming heartbeat. A Pi runner only reports at turn and tool boundaries, and the
+   *  daemon reads any hook older than its freshness window as `unknown` (`stale_runtime_hook`): an
+   *  idle seat read `unknown` five minutes after its last turn, and a seat mid-turn read `unknown`
+   *  through a long silent LLM request (the seat-activity oracle stops a bare working hook from
+   *  deciding after 15s, HOOK_AUTHORITY_WINDOW_MS). Beat while ready and not waiting on the
+   *  operator: idle (Stop) when no turn is in flight, streaming (active) while `streaming` says
+   *  one is, each at most once per its own interval behind the newest real event. Same honesty
+   *  rule: a runner that dies stops beating and the seat goes back to `unknown` (handlePiExit
+   *  clears `ready`, so no beat outlives the process it vouches for). Pi only; the OMP identity
+   *  handshake has its own retry.
+   *
+   *  Liveness of `streaming` (CodeRabbit 2336): the flag opens on agent_start/turn_start and
+   *  closes only on agent_end — can a turn end without one? Pi's run loop (dist/bundle/chunks,
+   *  runLoop, pi 0.84.4) emits agent_end on every exit: `stopReason === "error" ||
+   *  "aborted"` → finishTurn, turn_end, agent_end, return; decision end → turn_end, agent_end,
+   *  return; loop break → agent_end. That covers /abort (stopReason "aborted"), assistant errors
+   *  and exhausted auto-retries (stopReason "error"), and turns opened by turn_start alone
+   *  (turn_start fires only inside runLoop). A child crash is handlePiExit's honest exit, not a
+   *  silent working state. So a lost agent_end is not a Pi outcome, and no time limit is needed
+   *  on the beat — long silent turns stay covered. */
   heartbeat(): void {
-    if (this.runtime !== "pi" || !this.ready || this.streaming || this.attention) return;
+    if (this.runtime !== "pi" || !this.ready || this.attention) return;
+    const now = Date.parse(this.io.now());
+    if (this.streaming) {
+      if (this.lastActivityPostMs !== null && now - this.lastActivityPostMs < STREAM_HEARTBEAT_MS) return;
+      this.io.postActivity(this.activityPayload("active", "streaming"));
+      return;
+    }
     // `streaming` follows only agent_start/agent_end; turn_start, compaction and auto-retry also report the agent
     // working. Beat only when the last thing posted was a Stop (or nothing yet), so a beat never lands mid-turn.
     if (this.lastPostedHook !== null && this.lastPostedHook !== "Stop") return;
-    const now = Date.parse(this.io.now());
     if (this.lastHeartbeatMs !== null && now - this.lastHeartbeatMs < IDLE_HEARTBEAT_MS) return;
     this.io.postActivity(this.activityPayload("Stop", "idle_heartbeat"));
   }
@@ -879,6 +904,7 @@ export class RunnerCore {
 
   private activityPayload(hookEvent: string, subtype: string | null): Record<string, unknown> {
     this.lastPostedHook = hookEvent;
+    this.lastActivityPostMs = Date.parse(this.io.now());
     // Any posted Stop is an idle report, so the next beat is due a full interval after it, not just after a beat.
     if (hookEvent === "Stop") this.lastHeartbeatMs = Date.parse(this.io.now());
     return {
@@ -1040,7 +1066,24 @@ export function inOrder<A, R>(send: (arg: A) => Promise<R>): (arg: A) => Promise
 const IDENTITY_RETRY_MS = 30_000;
 /** Below the daemon's 5 minute hook freshness window (AGENT_ACTIVITY_FRESHNESS_MS). */
 const IDLE_HEARTBEAT_MS = 3 * 60_000;
-const IDLE_HEARTBEAT_TICK_MS = 15_000;
+/** One 5s scheduler tick drives both beats. CodeRabbit 2336: the first cut ticked every 15s, so a
+ *  streaming beat could land ~15s (+ delivery) behind a real event — the edge of
+ *  HOOK_AUTHORITY_WINDOW_MS, with a worst case near 25s. A 5s tick with a 5s minimum age posts a
+ *  streaming beat 5–10s after the newest event or beat: inside the window with headroom. The idle
+ *  beat's own 3-minute age gate makes the extra ticks free. */
+const PI_HEARTBEAT_TICK_MS = 5_000;
+/** Minimum age of the newest activity post before the streaming beat may re-assert it. */
+const STREAM_HEARTBEAT_MS = 5_000;
+
+/** The Pi heartbeat scheduler — one 5s tick drives both beats (see PI_HEARTBEAT_TICK_MS).
+ *  Exported so tests drive the real interval with fake timers instead of calling heartbeat()
+ *  by hand, which is exactly the gap that hid the 15s-tick latency in the first cut. */
+export function startHeartbeatTimer(core: { heartbeat(): void }, runtime: RunnerRuntime): { stop(): void } {
+  if (runtime !== "pi") return { stop() {} };
+  const timer = setInterval(() => core.heartbeat(), PI_HEARTBEAT_TICK_MS);
+  timer.unref();
+  return { stop: () => clearInterval(timer) };
+}
 
 export interface ExecutableResolverOps {
   isExecutable(path: string): boolean;
@@ -1346,8 +1389,7 @@ export async function main(argv = process.argv.slice(2)): Promise<void> {
   // Re-deliver OMP identity until the daemon confirms the resume token.
   const identityRetry = runtime === "omp" ? setInterval(() => core.retrySessionIdentity(), IDENTITY_RETRY_MS) : undefined;
   identityRetry?.unref();
-  const idleHeartbeat = runtime === "pi" ? setInterval(() => core.heartbeat(), IDLE_HEARTBEAT_TICK_MS) : undefined;
-  idleHeartbeat?.unref();
+  const heartbeatTimer = startHeartbeatTimer(core, runtime);
   // The last few stderr lines, so an OMP that exits during startup is
   // reported with its own words in the error, not only above it.
   const stderrTail: string[] = [];
@@ -1398,7 +1440,7 @@ export async function main(argv = process.argv.slice(2)): Promise<void> {
     if (exited) return;
     exited = true;
     clearInterval(identityRetry);
-    clearInterval(idleHeartbeat);
+    heartbeatTimer.stop();
     if (!core.isReady()) {
       const how = signal ? `signal ${signal}` : `exit code ${code ?? "unknown"}`;
       const output = stderrTail.length > 0 ? ` Its last output: ${stderrTail.join(" | ")}` : " It printed nothing on stderr.";

packages/daemon/test/pi-runner.test.ts

@@ -8,7 +8,7 @@ import { describe, it, expect, vi } from "vitest";
 import { PassThrough } from "node:stream";
 import {
   createRunnerInput, inOrder, MAX_PI_INPUT_BYTES, RunnerCore, mapPiEvent, parseRunnerArgs,
-  prepareRunnerSidecar,
+  prepareRunnerSidecar, startHeartbeatTimer,
   type RunnerIo,
 } from "../src/adapters/pi-runner.js";
 import { PI_RUNNER_READY_MARKER, PI_RUNNER_EXIT_MARKER, buildPiRunnerCommand, buildPiChildArgs, type PiRunnerState } from "../src/adapters/pi-runner-protocol.js";
@@ -144,10 +144,10 @@ describe("idle heartbeat", () => {
     expect(c.beats()).toHaveLength(1);
   });
 
-  // agent_start is not the only way a turn runs: turn_start, compaction and auto-retry report the agent working
-  // without setting `streaming`, and a beat in those stretches would read idle mid-turn.
+  // agent_start is not the only way a turn runs, but turn_start DOES open the streaming flag:
+  // a turn opened by turn_start alone gets beats. Compaction and auto-retry report the agent
+  // working without opening a turn, and a beat in those stretches would read idle mid-turn.
   it.each([
-    ["a turn opened by turn_start alone", { type: "turn_start" }],
     ["a compaction", { type: "compaction_start" }],
     ["an automatic retry", { type: "auto_retry_start" }],
   ])("stays quiet during %s, and resumes after agent_end", (_name, event) => {
@@ -206,6 +206,161 @@ describe("idle heartbeat", () => {
   });
 });
 
+// ── Streaming heartbeat ─────────────────────────────────────────────────────
+
+describe("streaming heartbeat", () => {
+  function clocked(runtime?: "pi" | "omp") {
+    const f = fakeIo();
+    let t = Date.parse("2026-07-06T10:00:00Z");
+    f.io.now = () => new Date(t).toISOString();
+    const core = new RunnerCore(f.io, { sessionName: SESSION, nodeId: "node-1", launchId: "launch-77" }, runtime ? { runtime } : undefined);
+    core.start();
+    const beats = () => f.activity.filter(a => a.subtype === "streaming");
+    return { core, f, beats, advance: (ms: number) => { t += ms; } };
+  }
+  const ready = (core: RunnerCore) => core.handlePiLine(JSON.stringify({
+    type: "response", id: "pi-runner-get-state", data: { sessionFile: SESSION_FILE, sessionId: "0197a2f0" } }));
+
+  // A silent inference leaves no events for the daemon; without a beat, a `running` hook older than
+  // the freshness window decays to `unknown` minutes into the turn it saw start.
+  it("re-asserts working behind a turn that runs silently, at most once per interval", () => {
+    const c = clocked();
+    ready(c.core);
+    c.core.handlePiLine(JSON.stringify({ type: "agent_start" }));
+    c.advance(10_000); c.core.heartbeat();
+    expect(c.beats()).toEqual([expect.objectContaining({ sessionName: SESSION, hookEvent: "active", subtype: "streaming", runtime: "pi" })]);
+    c.advance(10_000); c.core.heartbeat();
+    expect(c.beats()).toHaveLength(2);
+  });
+
+  // CodeRabbit 2336 — some turn paths omit agent_start; turn_start alone opens the turn.
+  it("keeps beating on a turn opened by turn_start alone", () => {
+    const c = clocked();
+    ready(c.core);
+    c.core.handlePiLine(JSON.stringify({ type: "turn_start" }));
+    c.advance(10_000); c.core.heartbeat();
+    expect(c.beats()).toEqual([expect.objectContaining({ subtype: "streaming" })]);
+    c.core.handlePiLine(JSON.stringify({ type: "agent_end" }));
+    c.advance(10_000); c.core.heartbeat();
+    expect(c.beats()).toHaveLength(1); // closed: idle branch resumes
+  });
+
+  it("stays quiet while real events keep arriving, then spaces the beat behind the newest one", () => {
+    const c = clocked();
+    ready(c.core);
+    c.core.handlePiLine(JSON.stringify({ type: "agent_start" }));
+    for (let i = 0; i < 10; i++) { c.advance(1_000); c.core.handlePiLine(JSON.stringify({ type: "tool_execution_start", toolName: "bash" })); }
+    c.core.heartbeat();
+    expect(c.beats()).toEqual([]); // the newest post (a tool call) is 1s old — well inside the interval
+    c.advance(10_000); c.core.heartbeat();
+    expect(c.beats()).toHaveLength(1);
+  });
+
+  it("stops when agent_end closes the turn, and after the pi child exits", () => {
+    const c = clocked();
+    ready(c.core);
+    c.core.handlePiLine(JSON.stringify({ type: "agent_start" }));
+    c.advance(20_000);
+    c.core.handlePiLine(JSON.stringify({ type: "agent_end" }));
+    c.advance(20_000); c.core.heartbeat();
+    expect(c.beats()).toEqual([]);
+    c.core.handlePiLine(JSON.stringify({ type: "agent_start" }));
+    c.core.handlePiExit(1);
+    c.advance(20_000); c.core.heartbeat();
+    expect(c.beats()).toEqual([]);
+  });
+
+  it("says nothing before the runner is ready", () => {
+    const c = clocked();
+    c.core.handlePiLine(JSON.stringify({ type: "agent_start" }));
+    c.advance(20_000); c.core.heartbeat();
+    expect(c.beats()).toEqual([]);
+  });
+
+  it("is a Pi-only behaviour", () => {
+    const c = clocked("omp");
+    ready(c.core);
+    c.core.handlePiLine(JSON.stringify({ type: "agent_start" }));
+    c.advance(20_000); c.core.heartbeat();
+    expect(c.beats()).toEqual([]);
+  });
+});
+
+// ── The real heartbeat scheduler (CodeRabbit 2336: the first cut's tests called
+// heartbeat() by hand and never drove the interval, hiding that a 15s tick landed the
+// streaming beat at the edge of HOOK_AUTHORITY_WINDOW_MS). Fake timers run the actual
+// startHeartbeatTimer wiring, and io.now follows the faked clock so age checks are real.
+
+describe("heartbeat scheduler cadence", () => {
+  function scheduledCore() {
+    vi.useFakeTimers({ now: Date.parse("2026-07-06T10:00:00Z") });
+    const f = fakeIo();
+    f.io.now = () => new Date().toISOString();
+    const core = new RunnerCore(f.io, { sessionName: SESSION, nodeId: "node-1", launchId: "launch-77" });
+    core.start();
+    core.handlePiLine(JSON.stringify({
+      type: "response", id: "pi-runner-get-state",
+      data: { sessionFile: SESSION_FILE, sessionId: "0197a2f0" },
+    }));
+    const timer = startHeartbeatTimer(core, "pi");
+    const beats = () => f.activity.filter(a => a.subtype === "streaming");
+    return { f, core, timer, beats, advance: (ms: number) => vi.advanceTimersByTime(ms) };
+  }
+  const after = (c: ReturnType<typeof scheduledCore>) => { c.timer.stop(); vi.useRealTimers(); };
+
+  it("lands the first streaming beat 5–10s after the turn opens, then every ~5s — inside the 15s authority window", () => {
+    const c = scheduledCore();
+    try {
+      c.core.handlePiLine(JSON.stringify({ type: "agent_start" }));
+      c.advance(4_000);
+      expect(c.beats()).toEqual([]); // too young
+      c.advance(1_000); // t+5s: the tick fires, the age gate passes
+      expect(c.beats()).toHaveLength(1);
+      c.advance(5_000); // t+10s
+      expect(c.beats()).toHaveLength(2);
+      c.advance(5_000); // t+15s: still inside the authority window per beat
+      expect(c.beats()).toHaveLength(3);
+    } finally { after(c); }
+  });
+
+  it("spaces the beat behind the newest real event, not just behind the turn", () => {
+    const c = scheduledCore();
+    try {
+      c.core.handlePiLine(JSON.stringify({ type: "agent_start" }));
+      c.advance(5_000);
+      expect(c.beats()).toHaveLength(1); // 5s after agent_start
+      c.core.handlePiLine(JSON.stringify({ type: "tool_execution_start", toolName: "bash" }));
+      c.advance(4_000);
+      expect(c.beats()).toHaveLength(1); // tool event only 4s old — age gate holds the beat
+      c.advance(1_000); // 5s after the tool event
+      expect(c.beats()).toHaveLength(2);
+    } finally { after(c); }
+  });
+
+  it("the idle beat still keeps its 3-minute interval under the 5s tick", () => {
+    const c = scheduledCore();
+    try {
+      const idles = () => c.f.activity.filter(a => a.subtype === "idle_heartbeat");
+      c.advance(3 * 60_000);
+      const first = idles().length; // 1: the very first tick posts (nothing older than the gate yet)
+      expect(first).toBe(1);
+      c.advance(4_000); // total 184s since open: 179s since the first beat — the age gate holds
+      expect(idles()).toHaveLength(first);
+      c.advance(1_000); // total 185s = 180s after the first beat
+      expect(idles()).toHaveLength(first + 1);
+    } finally { after(c); }
+  });
+
+  it("a non-pi runtime starts no timer", () => {
+    const f = fakeIo();
+    const core = new RunnerCore(f.io, { sessionName: SESSION, nodeId: "node-1", launchId: "launch-77" }, { runtime: "omp" });
+    core.start();
+    const timer = startHeartbeatTimer(core, "omp");
+    timer.stop(); // noop, and nothing threw
+    expect(true).toBe(true);
+  });
+});
+
 // ── Actual Node line editor + framed input ──────────────────────────────────
 
 describe("runner input", () => {
@@ -507,6 +662,14 @@ describe("mapPiEvent", () => {
     });
   });
 
+  // Some turn paths omit agent_start; turn_start alone opens the turn, so it must open the
+  // streaming flag too — the beat's only in-flight signal — or that path still goes blind.
+  it("turn_start opens streaming for pi", () => {
+    expect(mapPiEvent({ type: "turn_start" })).toMatchObject({
+      streaming: true, activity: { hookEvent: "active", subtype: "turn_start" },
+    });
+  });
+
   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({