From cbb76a5a8bcfe5e64c615f2f2c8148f023c4785a Mon Sep 17 00:00:00 2001 From: v-openrig-build Date: Sun, 2 Aug 2026 13:01:25 +0000 Subject: [PATCH] fix(tui): preserve fleet and input stream truth --- packages/tui/src/hydrate.ts | 19 ++- packages/tui/src/index.ts | 2 +- packages/tui/src/input.ts | 120 ++++++++++++------- packages/tui/src/main.ts | 19 ++- packages/tui/src/render.ts | 5 +- packages/tui/src/state.ts | 8 +- packages/tui/src/types.ts | 2 + packages/tui/test/hydrate.test.ts | 42 ++++++- packages/tui/test/input-stream.test.ts | 29 +++++ packages/tui/test/render-regressions.test.ts | 17 +++ 10 files changed, 200 insertions(+), 63 deletions(-) create mode 100644 packages/tui/test/input-stream.test.ts diff --git a/packages/tui/src/hydrate.ts b/packages/tui/src/hydrate.ts index bbc97da95..33ee8c247 100644 --- a/packages/tui/src/hydrate.ts +++ b/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 { @@ -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, diff --git a/packages/tui/src/index.ts b/packages/tui/src/index.ts index aadbfd376..bbcd21c48 100644 --- a/packages/tui/src/index.ts +++ b/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"; diff --git a/packages/tui/src/input.ts b/packages/tui/src/input.ts index 3a3b886a5..dc85ee718 100644 --- a/packages/tui/src/input.ts +++ b/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). */ diff --git a/packages/tui/src/main.ts b/packages/tui/src/main.ts index f64148f6c..1bb82413a 100644 --- a/packages/tui/src/main.ts +++ b/packages/tui/src/main.ts @@ -7,7 +7,7 @@ // openrig-tui [--instance ] [--socket ] [--url ] [--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 { 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 { 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): 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 { } } 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); diff --git a/packages/tui/src/render.ts b/packages/tui/src/render.ts index f008c3f36..887b7490b 100644 --- a/packages/tui/src/render.ts +++ b/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 } : {}), }; } diff --git a/packages/tui/src/state.ts b/packages/tui/src/state.ts index 7097b0212..a498be37a 100644 --- a/packages/tui/src/state.ts +++ b/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) { diff --git a/packages/tui/src/types.ts b/packages/tui/src/types.ts index 7cc0ce3ad..9f7115e01 100644 --- a/packages/tui/src/types.ts +++ b/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 diff --git a/packages/tui/test/hydrate.test.ts b/packages/tui/test/hydrate.test.ts index 3c2bd0b3d..37c09df61 100644 --- a/packages/tui/test/hydrate.test.ts +++ b/packages/tui/test/hydrate.test.ts @@ -91,10 +91,10 @@ const FIXTURES: Record = { 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>)[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"]); diff --git a/packages/tui/test/input-stream.test.ts b/packages/tui/test/input-stream.test.ts new file mode 100644 index 000000000..5bd0e2942 --- /dev/null +++ b/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: "界" }]); + } + }); +}); diff --git a/packages/tui/test/render-regressions.test.ts b/packages/tui/test/render-regressions.test.ts index b5855c73e..38b5e5dc9 100644 --- a/packages/tui/test/render-regressions.test.ts +++ b/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 = {