feat(liveness): honest SSE states — thinking must be backed by a live stream (L1, #638)

packages/extension/src/sse_client.ts

@@ -1,6 +1,7 @@
 import * as http from "node:http";
 import * as vscode from "vscode";
 import type { StatusBarManager } from "./status_bar";
+import { SseLivenessTracker, type SseState } from "./sse_liveness";
 
 // ============================================================================
 // OpencodeEventClient — Channel 1 of the bidirectional plumbing.
@@ -25,6 +26,9 @@ export interface SseClientOptions {
    *  without it the fork 401s /event and the reconnect loop spins forever.
    *  Never logged; the value exists only on the wire. */
   authorization?: string;
+  /** Liveness states (L1, #638): fired on every transition. The status bar
+   *  is driven automatically from this; this hook is for any other sink. */
+  onSseState?: (from: SseState, to: SseState) => void;
 }
 
 export class OpencodeEventClient implements vscode.Disposable {
@@ -34,17 +38,67 @@ export class OpencodeEventClient implements vscode.Disposable {
   private url?: URL;
   private reconnectTimer?: NodeJS.Timeout;
   private disposed = false;
+  private liveness?: SseLivenessTracker;
 
   constructor(private readonly opts: SseClientOptions) {}
 
   connect(serverUrl: URL): void {
     this.url = new URL("/event", serverUrl);
     this.opts.channel.appendLine(`[sse] connecting to ${this.url}`);
+    // Liveness (L1, #638): frames are the truth; the probe splits a quiet
+    // half-open stream from a dead server; every transition reaches the
+    // status bar and the caller's hook — "thinking" is never unbacked.
+    this.liveness?.dispose();
+    this.liveness = new SseLivenessTracker({
+      probe: () => this.probeServer(),
+      log: (line) => this.opts.channel.appendLine(line),
+      onStateChange: (from, to) => {
+        this.opts.onSseState?.(from, to);
+        this.opts.statusBar?.setSseState(to);
+      },
+      onReconnectNeeded: () => {
+        // half-open: the server is alive but this stream will never speak
+        // again — tear it down; the reconnect loop reopens fresh
+        try {
+          this.req?.destroy();
+        } catch {
+          /* noop */
+        }
+        try {
+          this.res?.destroy();
+        } catch {
+          /* noop */
+        }
+        this.scheduleReconnect();
+      },
+    });
+    this.liveness.start();
     this.openOnce();
   }
 
+  /** The stream's honest state — for callers and tests. */
+  get sseState(): SseState {
+    return this.liveness?.getState() ?? "connecting";
+  }
+
+  /** A short-timeout GET on the server root with the same credential the
+   *  stream uses — the STALE branch's dead-or-alive question. */
+  private async probeServer(): Promise<boolean> {
+    if (!this.url) return false;
+    try {
+      const r = await fetch(this.url.origin, {
+        signal: AbortSignal.timeout(2_000),
+        headers: this.opts.authorization ? { Authorization: this.opts.authorization } : undefined,
+      });
+      return r.status > 0;
+    } catch {
+      return false;
+    }
+  }
+
   dispose(): void {
     this.disposed = true;
+    this.liveness?.dispose();
     if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
     try {
       this.req?.destroy();
@@ -74,25 +128,30 @@ export class OpencodeEventClient implements vscode.Disposable {
       (res) => {
         this.res = res;
         if (res.statusCode !== 200) {
+          this.liveness?.noteDisconnected();
           this.opts.channel.appendLine(`[sse] bad status ${res.statusCode}; will retry`);
           this.scheduleReconnect();
           return;
         }
         this.opts.channel.appendLine(`[sse] connected`);
+        this.liveness?.noteConnected();
         res.setEncoding("utf8");
         res.on("data", (chunk: string) => this.onChunk(chunk));
         res.on("end", () => {
           this.opts.channel.appendLine(`[sse] stream ended; will retry`);
+          this.liveness?.noteDisconnected();
           this.scheduleReconnect();
         });
         res.on("error", (err) => {
           this.opts.channel.appendLine(`[sse] response error: ${err.message}`);
+          this.liveness?.noteDisconnected();
           this.scheduleReconnect();
         });
       },
     );
     req.on("error", (err) => {
       this.opts.channel.appendLine(`[sse] req error: ${err.message}`);
+      this.liveness?.noteDisconnected();
       this.scheduleReconnect();
     });
     req.end();
@@ -109,6 +168,9 @@ export class OpencodeEventClient implements vscode.Disposable {
   }
 
   private onChunk(chunk: string): void {
+    // ANY bytes = the stream is speaking (events or comment-only pings —
+    // the liveness signal the old client discarded, #638)
+    this.liveness?.noteFrame();
     this.buf += chunk;
     // SSE events terminate on a blank line (\n\n).
     let sep: number;

packages/extension/src/sse_liveness.ts

@@ -0,0 +1,139 @@
+// Liveness state machine (L1, #638): evidence-backed states for the event
+// stream, so "thinking" is never rendered without a live connection behind it.
+//
+// The truth hierarchy: frames are the primary signal (ANY SSE block — event
+// or comment-only ping; opencode sends pings the old client discarded). No
+// frames for stalenessMs → STALE. The STALE branch probes the server to split
+// "alive but stream quiet" (a half-open TCP connection — tear it down and
+// reconnect; never wait for TCP to notice) from "server gone" (DEAD). Every
+// transition fires onStateChange with the evidence logged — the UI gets the
+// truth, the output channel gets the why.
+//
+// Pure by injection: the probe, the clock, and the state-change sink are
+// dependencies. The production caller is sse_client; T4a's serve-daemon tests
+// reuse the same semantics. Tests use fake timers — zero real sleeps.
+
+export type SseState = "connecting" | "live" | "stale" | "dead";
+
+export const DEFAULT_STALENESS_MS = 30_000;
+export const DEFAULT_TICK_MS = 5_000;
+
+export interface SseLivenessOptions {
+  /** ms without a frame before the stream is STALE (default 30s). */
+  stalenessMs?: number;
+  /** tick cadence for staleness checks (default 5s). */
+  tickMs?: number;
+  /** Probe the server: resolve true = alive. Injected so the STALE branch
+   *  can split stale-from-dead without this module owning fetch. */
+  probe: () => Promise<boolean>;
+  /** Fired on every state TRANSITION (never twice for the same state). */
+  onStateChange?: (from: SseState, to: SseState) => void;
+  /** Fired once when the server proves alive but the stream stayed quiet —
+   *  the half-open signal: the owner tears the connection down and
+   *  reconnects rather than waiting for TCP to notice. */
+  onReconnectNeeded?: () => void;
+  /** Diagnostics sink (the output channel); liveness never crashes the host. */
+  log?: (line: string) => void;
+  /** Clock — injected; tests pass a controllable one, production wall clock. */
+  now?: () => number;
+}
+
+export class SseLivenessTracker {
+  private state: SseState = "connecting";
+  private lastFrameAt: number | undefined;
+  private connectedAt: number | undefined;
+  private readonly stalenessMs: number;
+  private readonly tickMs: number;
+  private readonly probeFn: () => Promise<boolean>;
+  private readonly onStateChange?: (from: SseState, to: SseState) => void;
+  private readonly onReconnectNeeded?: () => void;
+  private readonly log?: (line: string) => void;
+  private readonly now: () => number;
+  private timer: ReturnType<typeof setInterval> | undefined;
+  private probing = false;
+
+  constructor(opts: SseLivenessOptions) {
+    this.stalenessMs = opts.stalenessMs ?? DEFAULT_STALENESS_MS;
+    this.tickMs = opts.tickMs ?? DEFAULT_TICK_MS;
+    this.probeFn = opts.probe;
+    this.onStateChange = opts.onStateChange;
+    this.onReconnectNeeded = opts.onReconnectNeeded;
+    this.log = opts.log;
+    this.now = opts.now ?? Date.now;
+  }
+
+  /** Start ticking. Idempotent. */
+  start(): void {
+    if (this.timer !== undefined) return;
+    this.timer = setInterval(() => void this.tick(), this.tickMs);
+  }
+
+  dispose(): void {
+    if (this.timer !== undefined) {
+      clearInterval(this.timer);
+      this.timer = undefined;
+    }
+  }
+
+  getState(): SseState {
+    return this.state;
+  }
+
+  /** The stream connected (2xx on the event request). */
+  noteConnected(): void {
+    this.connectedAt = this.now();
+    this.transition("live");
+  }
+
+  /** The stream ended/errored — a reconnect is in flight. */
+  noteDisconnected(): void {
+    this.transition("connecting");
+  }
+
+  /** ANY block arrived — event or comment-only ping. The discarded-ping fix:
+   *  this is the liveness signal. Frames recover STALE back to LIVE. */
+  noteFrame(): void {
+    this.lastFrameAt = this.now();
+    if (this.state === "stale") {
+      this.log?.("[liveness] frames returned — stream live again");
+    }
+    // a speaking stream is live in every state — believe the frames
+    this.transition("live");
+  }
+
+  private transition(to: SseState): void {
+    if (this.state === to) return;
+    const from = this.state;
+    this.state = to;
+    this.log?.(`[liveness] ${from} → ${to}`);
+    this.onStateChange?.(from, to);
+  }
+
+  private async tick(): Promise<void> {
+    if (this.state !== "live" && this.state !== "stale") return;
+    const now = this.now();
+    const last = this.lastFrameAt ?? this.connectedAt ?? now;
+    if (now - last < this.stalenessMs) return;
+    if (this.state === "live") {
+      this.transition("stale");
+    }
+    if (this.probing) return;
+    this.probing = true;
+    let alive = false;
+    try {
+      alive = await this.probeFn();
+    } catch {
+      alive = false; /* a throwing probe is a dead probe, never a crash */
+    } finally {
+      this.probing = false;
+    }
+    // state may have moved on while the probe was in flight (frames returned)
+    if (this.state !== "stale") return;
+    if (!alive) {
+      this.transition("dead");
+      return;
+    }
+    this.log?.("[liveness] server alive but stream quiet — half-open; reconnect");
+    this.onReconnectNeeded?.();
+  }
+}

packages/extension/src/status_bar.ts

@@ -1,19 +1,44 @@
 import * as vscode from "vscode";
+import type { SseState } from "./sse_liveness";
 
 // ============================================================================
-// StatusBarManager — minimal server-ready indicator. The live-solve indicator
-// (iter/fidelity) is removed in #351 — the Work Column Run Inspector tab is
-// now the sole surface for solve status.
+// StatusBarManager — the honest server/stream indicator. The live-solve
+// indicator (iter/fidelity) is removed in #351 — the Work Column Run
+// Inspector tab is the sole surface for solve status. The stream states (L1,
+// #638) are evidence-backed: "thinking" is only ever true while the event
+// stream is LIVE; a stalled stream says so instead of pretending.
 // ============================================================================
 
-export function statusBarLabel(serverReady: boolean): { text: string; tooltip: string } {
+export function statusBarLabel(
+  serverReady: boolean,
+  // default "live" preserves the historical ready-state label for callers
+  // that only know the server is ready; the manager always passes the real
+  // stream state (L1, #638)
+  sseState: SseState = "live",
+): { text: string; tooltip: string } {
   if (!serverReady) return { text: "$(loading~spin) Amicode (booting)", tooltip: "Spawning opencode server…" };
-  return { text: "$(comment-discussion) Amicode", tooltip: "Amicode — chat + Work Column inspectors" };
+  switch (sseState) {
+    case "live":
+      return { text: "$(comment-discussion) Amicode", tooltip: "Amicode — chat + Work Column inspectors" };
+    case "stale":
+      return {
+        text: "$(debug-disconnect) Amicode — stream stalled",
+        tooltip: "Event stream stalled — probing the harness, reconnecting if needed. A 'thinking' indicator is not truth right now.",
+      };
+    case "dead":
+      return {
+        text: "$(error) Amicode — server unreachable",
+        tooltip: "Harness unreachable — reconnecting. Session state is preserved; work is not lost.",
+      };
+    case "connecting":
+      return { text: "$(loading~spin) Amicode", tooltip: "Connecting to the harness event stream…" };
+  }
 }
 
 export class StatusBarManager {
   private readonly item: vscode.StatusBarItem;
   private serverReady = false;
+  private sseState: SseState = "connecting";
 
   constructor() {
     this.item = vscode.window.createStatusBarItem(vscode.StatusBarAlignment.Right, 100);
@@ -27,12 +52,19 @@ export class StatusBarManager {
     this.render();
   }
 
+  /** The stream's honest state (L1, #638) — driven by the SSE client's
+   *  liveness transitions; boot states stay untouched. */
+  setSseState(state: SseState): void {
+    this.sseState = state;
+    this.render();
+  }
+
   dispose(): void {
     this.item.dispose();
   }
 
   private render(): void {
-    const { text, tooltip } = statusBarLabel(this.serverReady);
+    const { text, tooltip } = statusBarLabel(this.serverReady, this.sseState);
     this.item.text = text;
     this.item.tooltip = tooltip;
   }