fix(tui): preserve fleet and input stream truth

packages/tui/src/hydrate.ts

@@ -91,6 +91,7 @@ interface RigSpecJsonRead {
   pods?: Array<{ members?: Array<{ agentRef?: string }> }>;
 }
 interface NeedsYouItemRead {
+  hostId?: string;
   source: "agent" | "derived";
   identity: string;
   summary: string;
@@ -102,6 +103,7 @@ interface NeedsYouItemRead {
 interface ReviewFleetRead {
   needsYou?: { items?: NeedsYouItemRead[] };
   hosts?: Array<{ hostId: string; status: { status: string } }>;
+  registryError?: string | null;
 }
 interface AttentionAggregateRead {
   hosts?: Array<{ hostId: string; status: string; error?: string }>;
@@ -130,9 +132,15 @@ function toAgentRow(node: NodeInventoryRead): AgentRow {
     tokens: known ? fmtTokens(ctx.totalInputTokens, ctx.totalOutputTokens) : null,
     // Mirror the maintained web projection: lifecycle truth drives actions,
     // while session/terminal activity drives the visible status label.
-    status: node.sessionStatus === "running" || node.sessionStatus === "ready"
-      ? (node.terminalActive === true || (node.terminalActive == null && node.agentActivity?.state === "running") ? "active" : "idle")
-      : (node.sessionStatus ?? "unknown"),
+    status: node.startupStatus === "failed"
+      ? "failed"
+      : node.startupStatus === "attention_required"
+        ? "attention_required"
+        : node.agentActivity?.state === "needs_input"
+          ? "needs_input"
+          : node.sessionStatus === "running" || node.sessionStatus === "ready"
+            ? (node.terminalActive === true || (node.terminalActive == null && node.agentActivity?.state === "running") ? "active" : "idle")
+            : (node.sessionStatus ?? "unknown"),
     live: node.lifecycleState === "running",
     canRun: node.lifecycleState !== "running",
     session: node.canonicalSessionName,
@@ -157,7 +165,7 @@ function toNeedsItem(item: NeedsYouItemRead): NeedsItem {
   // session/where the daemon already names (identity prefix for derived rows)
   const target = item.source === "derived" ? (item.identity.split("|")[0] ?? item.where) : (item.destinationSession ?? item.where);
   const detail = item.derived ? `${item.summary} — ${item.derived.evidence}` : item.summary;
-  return { kind: item.derived?.kind ?? item.leg, target, detail };
+  return { kind: item.derived?.kind ?? item.leg, target, detail, ...(item.hostId ? { hostId: item.hostId } : {}) };
 }
 
 function resolveAgentRef(ref: string, agentSpecNames: Set<string>): string {
@@ -222,6 +230,7 @@ export async function hydrateSnapshot(client: DaemonClient, reviewCache?: SpecRe
   ]);
 
   const agentSpecNames = new Set((library ?? []).filter((entry) => entry.kind === "agent").map((entry) => entry.name));
+  if (review?.registryError) readErrors.push(`review-fleet registry: ${review.registryError}`);
 
   // Topology: the local host expands to the daemon's rigs; remote hosts come
   // from the aggregate with reachability only (per-rig start; the all-rigs
@@ -382,7 +391,7 @@ export async function hydrateSnapshot(client: DaemonClient, reviewCache?: SpecRe
     hosts: [localHost, ...remoteHosts],
     specs,
     needs,
-    humanQueueProbed: review != null && Array.isArray(review.hosts) && review.hosts.length > 0
+    humanQueueProbed: review != null && !review.registryError && Array.isArray(review.hosts) && review.hosts.length > 0
       && review.hosts.every((host) => host.status.status === "ok"),
     humanQueue,
     hostsDown,

packages/tui/src/index.ts

@@ -1,7 +1,7 @@
 export * from "./types.js";
 export { parseCommand } from "./grammar.js";
 export { createViewState, defaultSections, emptySnapshot, computeExplorerRows, findAgent, findSpec, findRig, agentsRunningSpec, agentsRunningSpecTargets } from "./state.js";
-export { decodeInput, resolveKeyAction, sgrClick, MOUSE_ENABLE, MOUSE_DISABLE, ALT_SCREEN_ON, ALT_SCREEN_OFF } from "./input.js";
+export { createInputDecoder, decodeInput, resolveKeyAction, sgrClick, MOUSE_ENABLE, MOUSE_DISABLE, ALT_SCREEN_ON, ALT_SCREEN_OFF } from "./input.js";
 export { renderScreen } from "./render.js";
 export { createControlSocket, defaultSocketPath, describeState, MAX_SOCKET_PATH_BYTES } from "./socket-server.js";
 export { DaemonClient } from "./daemon-client.js";

packages/tui/src/input.ts

@@ -3,47 +3,53 @@
 // Mouse events are resolved against the renderer's hit-map by the caller and
 // then dispatched through the SAME dispatch as commands and keys (PIN 1).
 import type { Action, InputEvent, Screen, ViewState } from "./types.js";
+import { StringDecoder } from "node:string_decoder";
 
-const SGR_MOUSE = /\x1b\[<(\d+);(\d+);(\d+)([Mm])/g;
-
-export function decodeInput(bytes: string | Buffer): InputEvent[] {
-  const text = typeof bytes === "string" ? bytes : bytes.toString("utf8");
+function parseText(text: string, final: boolean): { events: InputEvent[]; remainder: string } {
   const events: InputEvent[] = [];
-
-  const rest = text.replace(SGR_MOUSE, (_, b: string, x: string, y: string, kind: string) => {
-    const button = Number(b);
-    if (kind === "M" && (button & 3) !== 3 && button < 32) {
-      // press of button 0/1/2 (32+ are motion/wheel — out of the safe core)
-      events.push({ type: "mouse", button: button & 3, x: Number(x), y: Number(y) });
-    }
-    return "";
-  });
-
   let i = 0;
-  while (i < rest.length) {
-    const ch = rest[i] ?? "";
-    if (ch === "\x1b" && rest[i + 1] === "[") {
-      const code = rest[i + 2];
-      if ((code === "5" || code === "6") && rest[i + 3] === "~") {
-        const down = code === "6";
-        events.push({
-          type: "key",
-          key: down ? "pagedown" : "pageup",
-          action: { type: "content-scroll", delta: down ? 10 : -10 },
-        });
-        i += 4;
-        continue;
-      }
-      const key = code === "A" ? "up" : code === "B" ? "down" : code === "C" ? "right" : code === "D" ? "left" : null;
-      if (key) {
-        events.push({
-          type: "key",
-          key,
-          action: { type: "select", delta: key === "down" ? 1 : key === "up" ? -1 : 0 },
-        });
-        i += 3;
-        continue;
+  while (i < text.length) {
+    const ch = text[i] ?? "";
+    if (ch === "\x1b") {
+      if (i + 1 >= text.length && !final) break;
+      if (text[i + 1] === "[") {
+        if (i + 2 >= text.length && !final) break;
+        const code = text[i + 2];
+        if (code === "<") {
+          const tail = text.slice(i);
+          const match = tail.match(/^\x1b\[<(\d+);(\d+);(\d+)([Mm])/);
+          if (match) {
+            const button = Number(match[1]);
+            if (match[4] === "M" && (button & 3) !== 3 && button < 32)
+              events.push({ type: "mouse", button: button & 3, x: Number(match[2]), y: Number(match[3]) });
+            i += match[0].length;
+            continue;
+          }
+          if (!final && /^\x1b\[<[0-9;]*$/.test(tail)) break;
+        }
+        if ((code === "5" || code === "6") && i + 3 >= text.length && !final) break;
+        if ((code === "5" || code === "6") && text[i + 3] === "~") {
+          const down = code === "6";
+          events.push({
+            type: "key",
+            key: down ? "pagedown" : "pageup",
+            action: { type: "content-scroll", delta: down ? 10 : -10 },
+          });
+          i += 4;
+          continue;
+        }
+        const key = code === "A" ? "up" : code === "B" ? "down" : code === "C" ? "right" : code === "D" ? "left" : null;
+        if (key) {
+          events.push({
+            type: "key",
+            key,
+            action: { type: "select", delta: key === "down" ? 1 : key === "up" ? -1 : 0 },
+          });
+          i += 3;
+          continue;
+        }
       }
+      events.push({ type: "key", key: "escape" });
       i += 1;
       continue;
     }
@@ -57,15 +63,43 @@ export function decodeInput(bytes: string | Buffer): InputEvent[] {
       i += 1;
       continue;
     }
-    if (ch === "\x1b") {
-      events.push({ type: "key", key: "escape" });
-      i += 1;
-      continue;
-    }
     if (ch >= " ") events.push({ type: "char", ch });
     i += 1;
   }
-  return events;
+  return { events, remainder: text.slice(i) };
+}
+
+export interface InputDecoder {
+  write(bytes: string | Buffer): InputEvent[];
+  flush(): InputEvent[];
+  hasPending(): boolean;
+}
+
+/** Stateful terminal-stream decoder: retains split escape sequences and uses
+ * Node's StringDecoder so UTF-8 code points survive arbitrary Buffer chunks. */
+export function createInputDecoder(): InputDecoder {
+  const utf8 = new StringDecoder("utf8");
+  let pending = "";
+  return {
+    write(bytes) {
+      pending += typeof bytes === "string" ? bytes : utf8.write(bytes);
+      const parsed = parseText(pending, false);
+      pending = parsed.remainder;
+      return parsed.events;
+    },
+    flush() {
+      const parsed = parseText(pending, true);
+      pending = parsed.remainder;
+      return parsed.events;
+    },
+    hasPending: () => pending.length > 0,
+  };
+}
+
+/** Whole-buffer convenience used by tests and synthetic adapters. */
+export function decodeInput(bytes: string | Buffer): InputEvent[] {
+  const decoder = createInputDecoder();
+  return [...decoder.write(bytes), ...decoder.flush()];
 }
 
 /** Test/automation helper: the SGR bytes a terminal emits for a left click at (x, y). */

packages/tui/src/main.ts

@@ -7,7 +7,7 @@
 //   openrig-tui [--instance <id>] [--socket <path>] [--url <daemon>] [--demo]
 import { createViewState, computeExplorerRows, emptySnapshot } from "./state.js";
 import { parseCommand } from "./grammar.js";
-import { decodeInput, resolveKeyAction, MOUSE_ENABLE, MOUSE_DISABLE, ALT_SCREEN_ON, ALT_SCREEN_OFF } from "./input.js";
+import { createInputDecoder, resolveKeyAction, MOUSE_ENABLE, MOUSE_DISABLE, ALT_SCREEN_ON, ALT_SCREEN_OFF } from "./input.js";
 import { renderScreen } from "./render.js";
 import { createControlSocket, defaultSocketPath } from "./socket-server.js";
 import { demoSnapshot } from "./demo-data.js";
@@ -37,6 +37,8 @@ async function run(): Promise<void> {
 
   let inputLine = "";
   let lastScreen: Screen | null = null;
+  const inputDecoder = createInputDecoder();
+  let inputFlushTimer: NodeJS.Timeout | null = null;
 
   function draw(): void {
     const cols = process.stdout.columns ?? 120;
@@ -107,8 +109,8 @@ async function run(): Promise<void> {
   process.on("SIGTERM", () => void shutdown());
 
   if (process.stdin.isTTY) process.stdin.setRawMode(true);
-  process.stdin.on("data", (bytes: Buffer) => {
-    for (const ev of decodeInput(bytes)) {
+  function handleInput(events: ReturnType<typeof inputDecoder.write>): void {
+    for (const ev of events) {
       if (ev.type === "char") {
         if (ev.ch === "q" && inputLine === "") {
           void shutdown();
@@ -144,6 +146,17 @@ async function run(): Promise<void> {
       }
     }
     draw();
+  }
+
+  process.stdin.on("data", (bytes: Buffer) => {
+    if (inputFlushTimer) clearTimeout(inputFlushTimer);
+    handleInput(inputDecoder.write(bytes));
+    if (inputDecoder.hasPending()) {
+      inputFlushTimer = setTimeout(() => {
+        inputFlushTimer = null;
+        handleInput(inputDecoder.flush());
+      }, 25);
+    }
   });
 
   process.stdout.write(ALT_SCREEN_ON + MOUSE_ENABLE);

packages/tui/src/render.ts

@@ -68,9 +68,10 @@ function specTabsLine(state: ViewState, name: string): ContentLine {
 }
 
 function needsLine(prefix: string, item: NeedsItem, snap: FleetSnapshot): ContentLine {
-  const found = findAgentBySession(snap, item.target);
+  const found = findAgentBySession(snap, item.target, item.hostId);
+  const provenance = item.hostId ? ` [${item.hostId}]` : "";
   return {
-    text: `${prefix}${item.kind}  ${item.target}  — ${item.detail}${found ? "  (open ▸)" : ""}`,
+    text: `${prefix}${item.kind}${provenance}  ${item.target}  — ${item.detail}${found ? "  (open ▸)" : ""}`,
     ...(found ? { action: { type: "drill", resource: "agent", name: found.agent.name, target: { host: found.host.name, rig: found.rig.name, pod: found.pod.name } } as const } : {}),
   };
 }

packages/tui/src/state.ts

@@ -167,12 +167,14 @@ export function findSpec(snap: FleetSnapshot, name: string) {
 }
 
 /** Joins a Needs-You target (a session name) back to the topology agent. */
-export function findAgentBySession(snap: FleetSnapshot, session: string) {
+export function findAgentBySession(snap: FleetSnapshot, session: string, hostId?: string) {
+  const matches = [];
   for (const host of snap.hosts)
     for (const rig of host.rigs)
       for (const pod of rig.pods)
-        for (const agent of pod.agents) if (agent.session === session) return { host, rig, pod, agent };
-  return null;
+        for (const agent of pod.agents)
+          if (agent.session === session && (!hostId || host.name === hostId)) matches.push({ host, rig, pod, agent });
+  return matches.length === 1 ? matches[0]! : null;
 }
 
 function rigMatches(snap: FleetSnapshot, name: string, hostName?: string) {

packages/tui/src/types.ts

@@ -109,6 +109,8 @@ export interface NeedsItem {
   kind: string;
   target: string;
   detail: string;
+  /** fleet provenance; required to prevent a remote row opening a local twin */
+  hostId?: string;
 }
 
 /** Host/rig-down rows — composed BESIDE the Needs-You items at render, never

packages/tui/test/hydrate.test.ts

@@ -91,10 +91,10 @@ const FIXTURES: Record<string, unknown> = {
     scope: "rig",
     needsYou: {
       items: [
-        { source: "agent", identity: "qitem-2026080201", summary: "approve slice 11 proof", leg: "human-routed", where: "human@kernel", ageIso: "2026-08-02T08:40:00.000Z", priority: "high", tier: "human-gate", evidenceRef: "proof/qa.md", unblocks: null, qitemId: "qitem-2026080201", destinationSession: "human@kernel", derived: null },
+        { source: "agent", identity: "qitem-2026080201", summary: "approve slice 11 proof", leg: "human-routed", where: "human@kernel", ageIso: "2026-08-02T08:40:00.000Z", priority: "high", tier: "human-gate", evidenceRef: "proof/qa.md", unblocks: null, qitemId: "qitem-2026080201", destinationSession: "human@kernel", derived: null, hostId: "local" },
         { source: "agent", identity: "qitem-remote", summary: "remote founder gate", leg: "human-routed", where: "human@kernel", ageIso: "2026-08-02T08:41:00.000Z", priority: "high", tier: "human-gate", evidenceRef: null, unblocks: null, qitemId: "qitem-remote", destinationSession: "human@kernel", derived: null, hostId: "mm2-host" },
-        { source: "derived", identity: "dev-impl@myrig|stuck|2026-08-02T08:43:00.000Z", summary: "impl looks stuck", leg: "stuck", where: "rig", ageIso: null, priority: null, tier: null, evidenceRef: null, unblocks: null, qitemId: null, destinationSession: null, derived: { kind: "stuck", evidence: "idle 47m >= 30m default · holds 2", threshold: "idle-with-work >= 30m" } },
-        { source: "derived", identity: "dev-qa@myrig|too-long-in-state|2026-08-02T05:00:00.000Z", summary: "qa has not transitioned in 180m", leg: "stuck", where: "rig", ageIso: null, priority: null, tier: null, evidenceRef: null, unblocks: null, qitemId: null, destinationSession: null, derived: { kind: "stuck", evidence: "no transition for 180m >= 120m default · holds 2", threshold: "too-long-in-state >= 120m" } },
+        { source: "derived", identity: "dev-impl@myrig|stuck|2026-08-02T08:43:00.000Z", summary: "impl looks stuck", leg: "stuck", where: "rig", ageIso: null, priority: null, tier: null, evidenceRef: null, unblocks: null, qitemId: null, destinationSession: null, derived: { kind: "stuck", evidence: "idle 47m >= 30m default · holds 2", threshold: "idle-with-work >= 30m" }, hostId: "local" },
+        { source: "derived", identity: "dev-qa@myrig|too-long-in-state|2026-08-02T05:00:00.000Z", summary: "qa has not transitioned in 180m", leg: "stuck", where: "rig", ageIso: null, priority: null, tier: null, evidenceRef: null, unblocks: null, qitemId: null, destinationSession: null, derived: { kind: "stuck", evidence: "no transition for 180m >= 120m default · holds 2", threshold: "too-long-in-state >= 120m" }, hostId: "local" },
       ],
       provenance: "computed from queue+ps (rig scope) · window: today at 2026-08-02T09:30:00.000Z",
     },
@@ -146,7 +146,7 @@ describe("snapshot hydration over the §4.A reads (Phase 2)", () => {
     const snap = await hydrateSnapshot(fixtureClient());
     const stuck = snap.needs.filter((n) => n.kind === "stuck");
     expect(stuck).toHaveLength(2);
-    expect(stuck[0]?.target).toBe("dev-impl@myrig");
+    expect(stuck[0]).toMatchObject({ target: "dev-impl@myrig", hostId: "local" });
     expect(stuck[0]?.detail).toContain("idle 47m >= 30m default");
     expect(stuck[1]?.target).toBe("dev-qa@myrig");
     expect(stuck[1]?.detail).toContain("no transition for 180m >= 120m default");
@@ -213,8 +213,8 @@ describe("snapshot hydration over the §4.A reads (Phase 2)", () => {
     const snap = await hydrateSnapshot(fixtureClient());
     expect(snap.humanQueueProbed).toBe(true);
     expect(snap.humanQueue).toEqual([
-      { kind: "human-routed", target: "human@kernel", detail: "approve slice 11 proof" },
-      { kind: "human-routed", target: "human@kernel", detail: "remote founder gate" },
+      { kind: "human-routed", target: "human@kernel", detail: "approve slice 11 proof", hostId: "local" },
+      { kind: "human-routed", target: "human@kernel", detail: "remote founder gate", hostId: "mm2-host" },
     ]);
   });
 
@@ -232,6 +232,36 @@ describe("snapshot hydration over the §4.A reads (Phase 2)", () => {
     expect(snap.humanQueue).toEqual([]);
   });
 
+  it("treats a fleet registry error as named incomplete state, never proven-empty", async () => {
+    const snap = await hydrateSnapshot(fixtureClient({}, {
+      "/api/review/fleet": {
+        needsYou: { items: [] },
+        hosts: [{ hostId: "local", status: { hostId: "local", status: "ok" } }],
+        registryError: "failed to parse hosts.yaml",
+      },
+    }));
+    expect(snap.humanQueueProbed).toBe(false);
+    expect(snap.readErrors).toContain("review-fleet registry: failed to parse hosts.yaml");
+  });
+
+  it("gives failed, attention, and needs-input truth precedence over terminal active/idle", async () => {
+    const base = (FIXTURES["/api/rigs/01JRIG/nodes"] as Array<Record<string, unknown>>)[0]!;
+    const nodes = [
+      { ...base, logicalId: "dev.failed", startupStatus: "failed", lifecycleState: "attention_required", terminalActive: true },
+      { ...base, logicalId: "dev.attention", startupStatus: "attention_required", lifecycleState: "attention_required", terminalActive: false },
+      { ...base, logicalId: "dev.input", startupStatus: "ready", agentActivity: { state: "needs_input" }, terminalActive: true },
+      { ...base, logicalId: "dev.active", startupStatus: "ready", agentActivity: { state: "running" }, terminalActive: true },
+    ];
+    const snap = await hydrateSnapshot(fixtureClient({}, { "/api/rigs/01JRIG/nodes": nodes }));
+    const statuses = Object.fromEntries(snap.hosts[0]!.rigs[0]!.pods[0]!.agents.map((agent) => [agent.name, agent.status]));
+    expect(statuses).toEqual({
+      "dev.failed": "failed",
+      "dev.attention": "attention_required",
+      "dev.input": "needs_input",
+      "dev.active": "active",
+    });
+  });
+
   it("joins Specs↔Topology over existing reads: rig agentRefs + agent usedByRigs", async () => {
     const snap = await hydrateSnapshot(fixtureClient());
     expect(snap.specs.find((s) => s.name === "myrig")?.agentRefs).toEqual(["implementer", "qa-agent"]);

packages/tui/test/input-stream.test.ts

@@ -0,0 +1,29 @@
+import { describe, expect, it } from "vitest";
+import { createInputDecoder, decodeInput, sgrClick } from "../src/input.js";
+
+describe("stateful stdin decoding", () => {
+  it("preserves arrows, paging, and SGR mouse at every chunk split", () => {
+    const vectors = ["\x1b[A", "\x1b[B", "\x1b[C", "\x1b[D", "\x1b[5~", "\x1b[6~", sgrClick(47, 12)];
+    for (const vector of vectors) {
+      const expected = decodeInput(vector);
+      for (let split = 1; split < Buffer.byteLength(vector); split++) {
+        const decoder = createInputDecoder();
+        const bytes = Buffer.from(vector);
+        const actual = [
+          ...decoder.write(bytes.subarray(0, split)),
+          ...decoder.write(bytes.subarray(split)),
+        ];
+        expect(actual, `${JSON.stringify(vector)} split ${split}`).toEqual(expected);
+      }
+    }
+  });
+
+  it("preserves UTF-8 characters split between bytes", () => {
+    const bytes = Buffer.from("界");
+    for (let split = 1; split < bytes.length; split++) {
+      const decoder = createInputDecoder();
+      expect(decoder.write(bytes.subarray(0, split))).toEqual([]);
+      expect(decoder.write(bytes.subarray(split))).toEqual([{ type: "char", ch: "界" }]);
+    }
+  });
+});

packages/tui/test/render-regressions.test.ts

@@ -99,6 +99,23 @@ describe("live visual regressions", () => {
     expect(screen.hitMap.some((hit) => hit.y === lineIndex + 1 && hit.x1 > 30)).toBe(false);
   });
 
+  it("never opens a local seat for a remote Needs row with the same canonical session", () => {
+    const snap = demoSnapshot();
+    snap.needs = [{
+      kind: "stuck",
+      target: "dev50-guard@openrig-build",
+      hostId: "remote-a",
+      detail: "remote guard needs attention",
+    }];
+    const view = createViewState({ instanceId: "t", getSnapshot: () => snap });
+    view.dispatch({ type: "jump", section: "needs" });
+    const screen = renderScreen(view.get(), snap, { cols: 140, rows: 34 });
+    const row = screen.lines.find((line) => line.includes("remote guard needs attention"));
+    expect(row).toContain("[remote-a]");
+    expect(row).not.toContain("open ▸");
+    expect(screen.contentTargets).toHaveLength(0);
+  });
+
   it("renders the locked rig-spec structure with clickable agent refs", () => {
     const base = demoSnapshot();
     const snap: FleetSnapshot = {