diff --git a/.env.example b/.env.example index b0d54b3..940a0ce 100644 --- a/.env.example +++ b/.env.example @@ -24,8 +24,17 @@ COLLIE_HOST=127.0.0.1 # --- Herdr connection --- # Usually injected by the systemd unit; override only if your socket is elsewhere. # HERDR_SOCKET_PATH=/home/you/.config/herdr/herdr.sock +# Transport to reach Herdr: auto (default) = the herdr CLI on Windows (its control socket is a named +# pipe Bun can't open), the Unix socket elsewhere. Force cli/socket to exercise one regardless of +# platform (e.g. the CLI path on macOS/Linux). +# COLLIE_HERDR_TRANSPORT=auto +# Path to the herdr binary for the CLI transport (Windows). Defaults to `herdr` on PATH. +# COLLIE_HERDR_BIN=C:\Users\you\AppData\Local\Programs\Herdr\bin\herdr.exe # Poll cadence (ms) and scrollback lines pulled for the detail view. COLLIE_POLL_MS=1500 +# Poll cadence when the transport has no event stream at all (the Windows CLI). Its own knob because +# COLLIE_POLL_MS would spawn herdr.exe too often and COLLIE_POLL_IDLE_MS is too stale. +# COLLIE_POLL_NO_EVENTS_MS=4000 COLLIE_READ_LINES=200 # Keys sent to submit a reply after the text (comma-separated). Agent-dependent. COLLIE_SUBMIT_KEYS=Enter diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 2be6b04..2484f94 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -124,6 +124,17 @@ app. Closing this needs the server-side blocking-message capture described above `agent.send`, `events.subscribe`, …). It translates to/from an internal domain model (`AgentStatus`, `AgentView`, `SnapshotResponse` — `bridge/types.ts`), so a Herdr API rename is a one-file fix, not a shatter. +- **One client, a pluggable transport chosen by platform.** `herdr-client.ts` exports a single + `HerdrClient` class — every method and its verified doc comment, written once — over a `Transport` + seam (`request()` + `subscribeEvents()`), selected by `createHerdrClient()`. The divergence between + platforms is transport, not semantics: `SocketTransport` (mac/Linux) opens Herdr's Unix socket + directly, the verified default; `CliTransport` (Windows) spawns the `herdr` CLI once per RPC via a + method→argv table, because Herdr maps the socket path onto a Windows named pipe that + `Bun.connect({unix})` can't reach. Both yield identical JSON envelopes, so nothing above the client + changes, and adding a Herdr method is one method plus one argv entry — never a change duplicated + across platforms. `COLLIE_HERDR_TRANSPORT=auto|cli|socket` forces a transport (so the CLI path can + be exercised off-Windows). The CLI has no `events.subscribe` equivalent, so `subscribeEvents()` + returns `null` and the poll below carries the whole load — which it's designed to do anyway. - **Output model: poll, not stream — event-poked.** Herdr exposes `pane.read` (snapshot) and `pane.output_matched` (regex event) but **no raw output-stream event**, so there is nothing to stream even if we wanted to; the live pane view is poll-on-status-change + caching. The bridge's @@ -133,8 +144,10 @@ app. Closing this needs the server-side blocking-message capture described above alongside purely to **poke** that poll: lifecycle events plus a per-agent-pane `pane.agent_status_changed` subscription trigger an immediate debounced re-poll, while the interval relaxes to `COLLIE_POLL_IDLE_MS` (12 s default) whenever the stream is healthy and drops back to - the fast `COLLIE_POLL_MS` when it isn't. **The snapshot poll stays the source of truth throughout — - a missed event costs one interval, never correctness.** + the fast `COLLIE_POLL_MS` when it transiently drops. A transport with **no** stream at all (the + Windows CLI) is a distinct steady state — not a flapping stream — so it polls on its own + `COLLIE_POLL_NO_EVENTS_MS` cadence (4 s default) with no reconnect loop. **The snapshot poll stays + the source of truth throughout — a missed event costs one interval, never correctness.** - **The browser polls too.** `useRevalidator` → `/api/snapshot` on an adaptive interval. There is no WebSocket fan-out to the browser and no push of state; pulling is what makes the two recovery loops below trivial. diff --git a/README.md b/README.md index cd4a16a..9c16ee7 100644 --- a/README.md +++ b/README.md @@ -131,6 +131,22 @@ deps by hand — the build runs `bun install` for you; the backend imports only [`web-push`](https://www.npmjs.com/package/web-push) is optional and lazy (see [Web Push](#web-push-optional)). +### Windows + +Collie runs on Windows with two caveats: + +- **`herdr` must be on `PATH`** (or point `HERDR_BIN_PATH` / `COLLIE_HERDR_BIN` at it). On Windows, + Herdr's control socket is a named pipe the bridge can't open directly, so it talks to Herdr by + spawning the `herdr` CLI per request instead — this needs the same-version binary reachable. +- **Change detection is poll-only.** There's no live `events.subscribe` stream over the CLI, so the + UI refreshes on a dedicated poll cadence (`COLLIE_POLL_NO_EVENTS_MS`, default 4s) rather than being + poked instantly on a herd change. Events were only ever a poke — polling is the source of truth — so + nothing but refresh latency changes. (`COLLIE_HERDR_TRANSPORT=auto|cli|socket` forces the transport; + `auto` picks the CLI on Windows and the socket elsewhere.) + +The control script (`collie-ctl.sh`) shells out via `bash`, so run it under Git Bash; without +`systemd --user` it supervises the bridge as a `nohup` process with a pidfile. + ## Install On the host, not your phone. Two ways in. diff --git a/bridge/config.ts b/bridge/config.ts index 221dbe1..b81f59c 100644 --- a/bridge/config.ts +++ b/bridge/config.ts @@ -1,6 +1,8 @@ import { homedir } from "node:os"; import { join } from "node:path"; +import type { TransportMode } from "./herdr-client.ts"; + // All bridge configuration, resolved once at startup. Env-driven so the systemd unit and the // plugin launcher can configure it without code changes. Defaults are safe for a single-user, // tailnet-only deployment. @@ -32,6 +34,20 @@ function envInt( return n; } +/** + * Read an env var constrained to a fixed set of string values, falling back (with a warning) on + * anything not in `allowed`. Empty/unset → `fallback`. Case-insensitive. + */ +function envEnum(name: string, allowed: readonly T[], fallback: T): T { + const raw = process.env[name]; + if (raw === undefined || raw.trim() === "") return fallback; + const v = raw.trim().toLowerCase(); + const match = allowed.find((a) => a.toLowerCase() === v); + if (match !== undefined) return match; + console.warn(`[config] ${name}="${raw}" is not one of ${allowed.join("|")} — using default ${fallback}`); + return fallback; +} + function envList(name: string): string[] { return (process.env[name] ?? "") .split(",") @@ -57,6 +73,20 @@ function envBool(name: string, fallback: boolean): boolean { export interface Config { /** Path to Herdr's control socket. A non-Herdr-launched daemon must discover this itself. */ socketPath: string; + /** + * Path to the `herdr` binary. Only used by the CLI transport (Windows, or a forced + * {@link transportMode}), where the bridge talks to Herdr by spawning this CLI — Herdr's socket is + * a Windows named pipe Bun can't open directly (see herdr-client.ts). Herdr injects `HERDR_BIN_PATH` + * into plugin commands; we fall back to that, then `COLLIE_HERDR_BIN`, then a bare `herdr` on PATH. + * Optional: the socket transport ignores it, so mac/Linux configs (and tests) can omit it. + */ + herdrBin?: string; + /** + * Which Herdr transport to use. `auto` (default) = the CLI on Windows, the Unix socket elsewhere. + * `cli`/`socket` force one regardless of platform — the only way someone on macOS/Linux (no Windows + * box) can exercise the CLI path. Set via `COLLIE_HERDR_TRANSPORT`. + */ + transportMode: TransportMode; /** TCP port the bridge listens on (loopback only). `tailscale serve` proxies to it. */ port: number; /** @@ -72,6 +102,14 @@ export interface Config { * one of these, never correctness. Falls back to {@link pollMs} the moment the stream drops. */ pollIdleMs: number; + /** + * Poll cadence, ms, when the transport has NO event stream at all (the Windows CLI). This is a + * steady state, not a transient drop, so it gets its own knob between the two socket cadences: + * {@link pollMs} (fast, transient-drop) would spawn herdr.exe far too often, and {@link pollIdleMs} + * (relaxed, stream-healthy) is too stale when polling is the ONLY signal. Set via + * `COLLIE_POLL_NO_EVENTS_MS`. + */ + pollNoEventsMs: number; /** * Debounce window before a blocked/done transition becomes a push, ms. An agent that resolves * within this window (you handled it at your desk) never notifies; one that fires is retracted @@ -149,10 +187,13 @@ export function loadConfig(): Config { return { socketPath: process.env.HERDR_SOCKET_PATH ?? join(homedir(), ".config", "herdr", "herdr.sock"), + herdrBin: process.env.HERDR_BIN_PATH ?? process.env.COLLIE_HERDR_BIN ?? "herdr", + transportMode: envEnum("COLLIE_HERDR_TRANSPORT", ["auto", "cli", "socket"] as const, "auto"), port: envInt("COLLIE_PORT", 8787, { min: 1, max: 65535 }), host: process.env.COLLIE_HOST ?? "127.0.0.1", pollMs: envInt("COLLIE_POLL_MS", 1500, { min: 250 }), pollIdleMs: envInt("COLLIE_POLL_IDLE_MS", 12_000, { min: 1000 }), + pollNoEventsMs: envInt("COLLIE_POLL_NO_EVENTS_MS", 4000, { min: 1000 }), notifyDelayMs: envInt("COLLIE_NOTIFY_DELAY_MS", 30_000, { min: 0 }), readLines: envInt("COLLIE_READ_LINES", 200, { min: 1 }), submitKeys: submitKeys.length ? submitKeys : ["Enter"], diff --git a/bridge/event-poker.test.ts b/bridge/event-poker.test.ts index 3e9508b..50eeacf 100644 --- a/bridge/event-poker.test.ts +++ b/bridge/event-poker.test.ts @@ -1,6 +1,6 @@ import { describe, expect, test } from "bun:test"; -import { buildSubscriptions, EventPoker, sameIdSet, type Subscription } from "./event-poker.ts"; +import { buildSubscriptions, EventPoker, type PollMode, sameIdSet, type Subscription } from "./event-poker.ts"; import type { HerdrClient } from "./herdr-client.ts"; // EventPoker owns the stream lifecycle (ack → healthy, events → debounced poke, down → backoff @@ -19,12 +19,15 @@ interface FakeStream { class FakeClient { readonly streams: FakeStream[] = []; + /** When true, subscribeEvents returns null — modelling a transport (Windows CLI) that can't stream. */ + constructor(private readonly noStream = false) {} subscribeEvents(opts: { subscriptions: Subscription[]; onUp: () => void; onEvent: (event: string, data: unknown) => void; onDown: (reason: string) => void; - }): { close(): void } { + }): { close(): void } | null { + if (this.noStream) return null; const stream: FakeStream = { ...opts, closed: false }; this.streams.push(stream); return { @@ -42,17 +45,17 @@ class FakeClient { } } -function makePoker(opts?: { debounceMs?: number; backoffMs?: number[] }) { - const client = new FakeClient(); +function makePoker(opts?: { debounceMs?: number; backoffMs?: number[]; noStream?: boolean }) { + const client = new FakeClient(opts?.noStream ?? false); const poker = new EventPoker(client as unknown as HerdrClient, { debounceMs: opts?.debounceMs ?? 10, backoffMs: opts?.backoffMs ?? [10, 20], }); const pokes: number[] = []; - const health: boolean[] = []; + const modes: PollMode[] = []; poker.onPoke(() => pokes.push(1)); - poker.onHealth((h) => health.push(h)); - return { client, poker, pokes, health }; + poker.onPollMode((m) => modes.push(m)); + return { client, poker, pokes, modes }; } describe("buildSubscriptions / sameIdSet", () => { @@ -82,16 +85,30 @@ describe("buildSubscriptions / sameIdSet", () => { }); }); -describe("EventPoker — health", () => { - test("goes healthy on ack and unhealthy on down, notifying each transition once", () => { - const { client, poker, health } = makePoker(); +describe("EventPoker — poll mode", () => { + test("goes streaming on ack and fast on down, notifying each transition once", () => { + const { client, poker, modes } = makePoker(); poker.start(); expect(client.streams.length).toBe(1); client.last.onUp(); client.last.onUp(); // duplicate ack — no second notify - expect(health).toEqual([true]); + expect(modes).toEqual(["streaming"]); client.last.onDown("socket error"); - expect(health).toEqual([true, false]); + expect(modes).toEqual(["streaming", "fast"]); + poker.stop(); + }); + + test("a transport with no stream reports no-events once and never reconnects", async () => { + const { client, poker, modes } = makePoker({ noStream: true, backoffMs: [10] }); + poker.start(); + // subscribeEvents returned null: no stream tracked, mode latched to no-events. + expect(client.streams.length).toBe(0); + expect(modes).toEqual(["no-events"]); + // A changing pane set must NOT trigger a resubscribe/reconnect attempt on a poll-only transport. + poker.setAgentPanes(["w1:p1"]); + await sleep(30); + expect(client.streams.length).toBe(0); + expect(modes).toEqual(["no-events"]); // still just the one transition poker.stop(); }); }); @@ -113,15 +130,15 @@ describe("EventPoker — debounced poke", () => { describe("EventPoker — reconnect backoff", () => { test("reconnects after a down per the backoff schedule and resets on the next ack", async () => { - const { client, poker, health } = makePoker({ backoffMs: [15, 40] }); + const { client, poker, modes } = makePoker({ backoffMs: [15, 40] }); poker.start(); client.last.onUp(); client.last.onDown("boom"); expect(client.streams.length).toBe(1); // not yet — waiting out the backoff await sleep(30); expect(client.streams.length).toBe(2); // reconnected (first backoff step) - client.last.onUp(); // healthy again → backoff reset - expect(health).toEqual([true, false, true]); + client.last.onUp(); // streaming again → backoff reset + expect(modes).toEqual(["streaming", "fast", "streaming"]); poker.stop(); }); }); @@ -157,13 +174,13 @@ describe("EventPoker — stop()", () => { expect(client.streams.length).toBe(1); // the scheduled reconnect was cancelled }); - test("closing an up stream on stop() does not flip health or schedule work", async () => { - const { client, poker, health } = makePoker(); + test("closing an up stream on stop() does not flip the mode or schedule work", async () => { + const { client, poker, modes } = makePoker(); poker.start(); client.last.onUp(); poker.stop(); expect(client.last.closed).toBe(true); // stop() closed it - expect(health).toEqual([true]); // the deliberate close is stale — no spurious unhealthy + expect(modes).toEqual(["streaming"]); // the deliberate close is stale — no spurious fast flip await sleep(20); expect(client.streams.length).toBe(1); }); diff --git a/bridge/event-poker.ts b/bridge/event-poker.ts index 2df51d1..3966278 100644 --- a/bridge/event-poker.ts +++ b/bridge/event-poker.ts @@ -6,11 +6,24 @@ import type { HerdrClient } from "./herdr-client.ts"; // relaxes to a safety-net cadence; when it's down the engine falls back to fast // polling. Events are never state here — a missed event costs one interval, never // correctness — so the snapshot poll stays the single source of truth. See index.ts. +// +// Some transports (the Windows CLI) have NO event stream: subscribeEvents returns +// null. That's a steady state, not a transient drop — so it gets its own poll mode +// (`no-events`) with a dedicated cadence and NO reconnect loop, rather than being +// treated as a stream that keeps failing. // ───────────────────────────────────────────────────────────────────────────── /** A subscription request entry: global (just `type`) or pane-scoped (needs `pane_id`). */ export type Subscription = { type: string; pane_id?: string }; +/** + * How the engine should be polling, derived from the event stream's state: + * • `streaming` — stream healthy; events poke re-polls, so relax to the safety-net cadence. + * • `fast` — stream is (transiently) down and reconnecting; poll fast until it recovers. + * • `no-events` — transport can't stream at all; poll on the dedicated no-events cadence forever. + */ +export type PollMode = "streaming" | "fast" | "no-events"; + // Global events that change what Collie's snapshot renders. We deliberately DROP layout.*, // worktree.*, pane.scroll_changed and pane.output_matched — none of them alter the herd view we // poll for, so subscribing would only add pokes that re-fetch identical state. Also NO @@ -67,15 +80,17 @@ export class EventPoker { private readonly backoff: number[]; private agentPanes: string[] = []; private started = false; - private healthy = false; + private mode: PollMode = "fast"; private backoffIdx = 0; // The active stream handle; identity-compared in callbacks so a superseded stream's late `onDown` - // (from a deliberate close during reconnect/stop) is ignored instead of flapping health. + // (from a deliberate close during reconnect/stop) is ignored instead of flapping the mode. private stream: { close(): void } | null = null; + // Set once a connect() attempt returns null: this transport has no event stream, so stop trying. + private streamUnsupported = false; private debounceTimer: ReturnType | null = null; private reconnectTimer: ReturnType | null = null; private readonly pokeListeners = new Set<() => void>(); - private readonly healthListeners = new Set<(healthy: boolean) => void>(); + private readonly modeListeners = new Set<(mode: PollMode) => void>(); constructor( private readonly client: HerdrClient, @@ -90,9 +105,10 @@ export class EventPoker { return () => this.pokeListeners.delete(cb); } - onHealth(cb: (healthy: boolean) => void): () => void { - this.healthListeners.add(cb); - return () => this.healthListeners.delete(cb); + /** Notified whenever the poll mode changes (drives StateEngine's cadence). Fires once per change. */ + onPollMode(cb: (mode: PollMode) => void): () => void { + this.modeListeners.add(cb); + return () => this.modeListeners.delete(cb); } start(): void { @@ -121,7 +137,8 @@ export class EventPoker { setAgentPanes(ids: string[]): void { if (sameIdSet(ids, this.agentPanes)) return; this.agentPanes = [...ids]; - if (this.started) this.reconnect(); + // No stream to re-scope on a poll-only transport — the snapshot poll already covers every pane. + if (this.started && !this.streamUnsupported) this.reconnect(); } private connect(): void { @@ -131,10 +148,10 @@ export class EventPoker { onUp: () => { if (this.stream !== handle) return; this.backoffIdx = 0; - // A resubscribe acks while already healthy, so setHealthy dedupes it silently — but it's - // the only journal evidence that the per-pane subscriptions followed the herd. Log it. - if (this.healthy) console.log(`[events] resubscribed (${subs.length} subscriptions)`); - this.setHealthy(true, subs.length); + // A resubscribe acks while already streaming, so setMode dedupes it silently — but it's the + // only journal evidence that the per-pane subscriptions followed the herd. Log it. + if (this.mode === "streaming") console.log(`[events] resubscribed (${subs.length} subscriptions)`); + this.setMode("streaming", subs.length); }, onEvent: () => { if (this.stream !== handle) return; @@ -143,10 +160,19 @@ export class EventPoker { onDown: (reason) => { if (this.stream !== handle) return; this.stream = null; - this.setHealthy(false, subs.length, reason); + this.setMode("fast", subs.length, reason); if (this.started) this.scheduleReconnect(); }, }); + // A null handle means this transport can't stream at all (Windows CLI). Latch it: no stream to + // track, no reconnect loop, and poll on the dedicated no-events cadence. Correctness is + // unaffected — events were only ever a poke; the snapshot poll is the source of truth. + if (handle == null) { + this.streamUnsupported = true; + this.stream = null; + this.setMode("no-events", subs.length); + return; + } this.stream = handle; } @@ -179,11 +205,12 @@ export class EventPoker { }, this.debounceMs); } - private setHealthy(healthy: boolean, subCount: number, reason?: string): void { - if (this.healthy === healthy) return; - this.healthy = healthy; - if (healthy) console.log(`[events] stream up (${subCount} subscriptions)`); - else console.log(`[events] stream down: ${reason ?? "unknown"} — fast polling until it recovers`); - for (const cb of this.healthListeners) cb(healthy); + private setMode(mode: PollMode, subCount: number, reason?: string): void { + if (this.mode === mode) return; + this.mode = mode; + if (mode === "streaming") console.log(`[events] stream up (${subCount} subscriptions)`); + else if (mode === "fast") console.log(`[events] stream down: ${reason ?? "unknown"} — fast polling until it recovers`); + else console.log("[events] no event stream on this transport — polling on the no-events cadence"); + for (const cb of this.modeListeners) cb(mode); } } diff --git a/bridge/herdr-client.test.ts b/bridge/herdr-client.test.ts new file mode 100644 index 0000000..805fe8b --- /dev/null +++ b/bridge/herdr-client.test.ts @@ -0,0 +1,177 @@ +import { describe, expect, test } from "bun:test"; + +import { type CliResult, type CliRunner, CliTransport, HerdrClient } from "./herdr-client.ts"; + +// The CLI transport's argv table is the one part of the Windows path neither reader can verify by +// eye and no non-Windows CI can exercise for real. CliTransport takes its runner as a constructor +// arg, so here we inject a recording fake: assert the exact argv each method spawns, and that the +// JSON-envelope / raw-text / void result handling is right. No process is ever spawned. + +interface Call { + bin: string; + args: string[]; + env: Record; +} + +/** A fake runner that records every invocation and returns a scripted result (default: empty exit 0). */ +function recordingRunner(result: Partial = {}) { + const calls: Call[] = []; + const runner: CliRunner = async (bin, args, env) => { + calls.push({ bin, args, env }); + return { code: result.code ?? 0, stdout: result.stdout ?? "", stderr: result.stderr ?? "" }; + }; + return { calls, runner }; +} + +/** Build a HerdrClient whose CLI transport uses the given runner. */ +function cliClient(runner: CliRunner, bin = "C:\\herdr.exe", socket = "C:\\herdr.sock") { + return new HerdrClient(new CliTransport(socket, bin, 5000, runner)); +} + +/** Wrap a result payload in the full `{"id","result":{...}}` envelope the CLI prints on stdout. */ +function envelope(result: unknown): string { + return JSON.stringify({ id: "x", result }); +} + +describe("CliTransport — argv table", () => { + test("list/snapshot calls map to their subcommands", async () => { + const cases: Array<{ run: (c: HerdrClient) => Promise; args: string[]; result: unknown }> = [ + { run: (c) => c.listWorkspaces(), args: ["workspace", "list"], result: { workspaces: [] } }, + { run: (c) => c.listPanes(), args: ["pane", "list"], result: { panes: [] } }, + { run: (c) => c.listTabs(), args: ["tab", "list"], result: { tabs: [] } }, + { + run: (c) => c.sessionSnapshot(), + args: ["api", "snapshot"], + result: { type: "snapshot", snapshot: { version: "0.7.5", protocol: 16, workspaces: [], tabs: [], panes: [] } }, + }, + ]; + for (const { run, args, result } of cases) { + const { calls, runner } = recordingRunner({ stdout: envelope(result) }); + await run(cliClient(runner)); + expect(calls).toHaveLength(1); + expect(calls[0]!.args).toEqual(args); + } + }); + + test("createTab includes --no-focus and optional --label/--cwd", async () => { + const rootPane = { pane_id: "w1:p9", workspace_id: "w1", tab_id: "t9", cwd: "/x" }; + const { calls, runner } = recordingRunner({ stdout: envelope({ root_pane: rootPane }) }); + const client = cliClient(runner); + + await client.createTab("w1"); + expect(calls[0]!.args).toEqual(["tab", "create", "--workspace", "w1", "--no-focus"]); + + await client.createTab("w1", { label: "build", cwd: "/repo" }); + expect(calls[1]!.args).toEqual([ + "tab", "create", "--workspace", "w1", "--no-focus", "--label", "build", "--cwd", "/repo", + ]); + }); + + test("createWorkspace passes --cwd, --no-focus and optional --label", async () => { + const result = { + workspace: { workspace_id: "w2", label: "space", number: 2 }, + root_pane: { pane_id: "w2:p1", workspace_id: "w2", tab_id: "t1", cwd: "/repo" }, + }; + const { calls, runner } = recordingRunner({ stdout: envelope(result) }); + const client = cliClient(runner); + + await client.createWorkspace({ cwd: "/repo" }); + expect(calls[0]!.args).toEqual(["workspace", "create", "--cwd", "/repo", "--no-focus"]); + + await client.createWorkspace({ cwd: "/repo", label: "work" }); + expect(calls[1]!.args).toEqual(["workspace", "create", "--cwd", "/repo", "--no-focus", "--label", "work"]); + }); + + test("send/close/rename map to their subcommands", async () => { + const { calls, runner } = recordingRunner(); + const client = cliClient(runner); + + // Reply text with quotes/&/%/spaces goes through as ONE argv element (real .exe — no cmd.exe shim). + await client.sendPaneText("w1:p1", 'echo "a & b" 100%'); + expect(calls[0]!.args).toEqual(["pane", "send-text", "w1:p1", 'echo "a & b" 100%']); + + await client.sendPaneKeys("w1:p1", ["ctrl+a", "Enter"]); + expect(calls[1]!.args).toEqual(["pane", "send-keys", "w1:p1", "ctrl+a", "Enter"]); + + await client.closePane("w1:p1"); + expect(calls[2]!.args).toEqual(["pane", "close", "w1:p1"]); + + await client.renamePane("w1:p1", "hot"); + expect(calls[3]!.args).toEqual(["pane", "rename", "w1:p1", "hot"]); + + // A null label clears via --clear, not a literal "null". + await client.renamePane("w1:p1", null); + expect(calls[4]!.args).toEqual(["pane", "rename", "w1:p1", "--clear"]); + + await client.renameTab("t1", "logs"); + expect(calls[5]!.args).toEqual(["tab", "rename", "t1", "logs"]); + + await client.closeTab("t1"); + expect(calls[6]!.args).toEqual(["tab", "close", "t1"]); + }); + + test("HERDR_SOCKET_PATH is injected into every invocation's env", async () => { + const { calls, runner } = recordingRunner({ stdout: envelope({ panes: [] }) }); + await cliClient(runner, "C:\\herdr.exe", "C:\\my.sock").listPanes(); + expect(calls[0]!.env.HERDR_SOCKET_PATH).toBe("C:\\my.sock"); + expect(calls[0]!.bin).toBe("C:\\herdr.exe"); + }); +}); + +describe("CliTransport — pane.read (raw text → envelope)", () => { + test("wraps raw stdout into a PaneRead with source args and revision 0", async () => { + const { calls, runner } = recordingRunner({ stdout: "line-1\nline-2\n" }); + const read = await cliClient(runner).readPane("w1:p1", "recent", 200, "ansi"); + expect(calls[0]!.args).toEqual([ + "pane", "read", "w1:p1", "--source", "recent", "--lines", "200", "--format", "ansi", + ]); + expect(read.pane_id).toBe("w1:p1"); + expect(read.text).toBe("line-1\nline-2\n"); + expect(read.revision).toBe(0); + }); + + test("truncated is approximated true when the read fills the requested window", async () => { + const full = Array.from({ length: 200 }, (_, i) => `l${i}`).join("\n"); + const { runner } = recordingRunner({ stdout: full }); + const read = await cliClient(runner).readPane("w1:p1", "recent", 200, "text"); + // 200 lines returned for a 200-line request → older scrollback almost certainly exists. + expect(read.truncated).toBe(true); + }); + + test("truncated is false when fewer lines than requested come back", async () => { + const { runner } = recordingRunner({ stdout: "only\ntwo\n" }); + const read = await cliClient(runner).readPane("w1:p1", "recent", 200, "text"); + expect(read.truncated).toBe(false); + }); +}); + +describe("CliTransport — result handling", () => { + test("a non-zero exit throws with the decoded error message", async () => { + const { runner } = recordingRunner({ + code: 1, + stderr: JSON.stringify({ error: { code: "pane_not_found", message: "no such pane" } }), + }); + await expect(cliClient(runner).closePane("bad")).rejects.toThrow(/pane_not_found: no such pane/); + }); + + test("an 'unknown variant' snapshot error surfaces verbatim (drives StateEngine's fallback)", async () => { + const { runner } = recordingRunner({ code: 1, stderr: "Error: unknown variant `session.snapshot`" }); + await expect(cliClient(runner).sessionSnapshot()).rejects.toThrow(/unknown variant/); + }); + + test("subscribeEvents returns null — the CLI has no event stream", () => { + const { runner } = recordingRunner(); + const stream = cliClient(runner).subscribeEvents({ + subscriptions: [], + onUp: () => {}, + onEvent: () => {}, + onDown: () => {}, + }); + expect(stream).toBeNull(); + }); + + test("ping resolves false when the underlying call fails (does not throw)", async () => { + const { runner } = recordingRunner({ code: 1, stderr: "boom" }); + expect(await cliClient(runner).ping()).toBe(false); + }); +}); diff --git a/bridge/herdr-client.ts b/bridge/herdr-client.ts index dfaeda7..d232b6c 100644 --- a/bridge/herdr-client.ts +++ b/bridge/herdr-client.ts @@ -2,12 +2,23 @@ import type { AgentStatus } from "./types.ts"; import { decodeReplyLine, decodeStreamLine } from "./wire.ts"; // ───────────────────────────────────────────────────────────────────────────── -// The Herdr socket adapter. THIS IS THE ONLY FILE that knows Herdr's method names -// and wire shapes. Everything else talks to the typed methods below, so a Herdr -// API change is a one-file fix. Protocol facts are documented in HERDR_API.md. +// The Herdr adapter. THIS IS THE ONLY FILE that knows Herdr's method names and +// wire shapes. Everything else talks to the typed HerdrClient class below, so a +// Herdr API change is a one-file fix. Protocol facts are documented in HERDR_API.md. // -// Key fact: RPC is ONE-SHOT — the server closes the connection after a single -// response. So every request opens a fresh connection, reads one line, closes. +// ONE CLIENT, PLUGGABLE TRANSPORT: +// • HerdrClient — every method and its verified doc comment, written once. It +// never touches a socket or a process; it only calls Transport.request(). +// • SocketTransport — mac/Linux. Opens Herdr's Unix socket directly (the verified +// upstream path). RPC is ONE-SHOT: the server closes after one response, so +// every request opens a fresh connection. Streams events.subscribe. +// • CliTransport — Windows. Herdr does NOT expose a filesystem AF_UNIX socket +// there; it maps the socket path onto a Windows named pipe (see that class for +// the full why). So we shell out to the `herdr` binary per RPC. No event stream. +// +// The divergence between platforms is TRANSPORT, not semantics — which is why the +// seam is here and not one-class-per-platform. createHerdrClient() picks the transport; +// callers only ever see HerdrClient. // ───────────────────────────────────────────────────────────────────────────── /** Raw wire shape of a workspace from `workspace.list`. */ @@ -88,16 +99,223 @@ export interface PaneRead { type ReadSource = "visible" | "recent" | "recent-unwrapped"; type ReadFormat = "text" | "ansi"; -let idCounter = 0; +export interface SubscribeOptions { + subscriptions: Array<{ type: string; pane_id?: string }>; + onUp: () => void; + onEvent: (event: string, data: unknown) => void; + onDown: (reason: string) => void; +} + +/** Handle to an open event stream; `close()` is idempotent. */ +export interface StreamHandle { + close(): void; +} + +/** + * The transport seam. A transport moves ONE request to Herdr and (optionally) streams events. + * Everything platform-specific lives behind this — the socket vs. the CLI — so HerdrClient's methods + * are written once on top. `request()` returns the `result` payload of Herdr's reply envelope. + */ +export interface Transport { + request(method: string, params?: Record): Promise; + /** + * Open a long-lived event stream, or return `null` when this transport cannot stream (the CLI has + * no `events.subscribe` equivalent). A `null` return is a first-class signal to EventPoker that + * this herd is poll-only — NOT an error to retry. + */ + subscribeEvents(opts: SubscribeOptions): StreamHandle | null; +} +/** + * The typed Herdr contract every consumer talks to. One implementation, transport-agnostic: each + * method builds params and delegates to {@link Transport.request}, so a Herdr API change is one + * method plus (on Windows) one argv-table entry — never a change duplicated across platforms. + */ export class HerdrClient { + constructor(private readonly transport: Transport) {} + + async listWorkspaces(): Promise { + const r = await this.transport.request<{ workspaces: WireWorkspace[] }>("workspace.list"); + return r.workspaces; + } + + async listPanes(): Promise { + const r = await this.transport.request<{ panes: WirePane[] }>("pane.list"); + return r.panes; + } + + /** All tabs across every workspace (`tab.list` with no filter returns the full set). */ + async listTabs(): Promise { + const r = await this.transport.request<{ tabs: WireTab[] }>("tab.list"); + return r.tabs; + } + + /** + * The whole herd in one round-trip (herdr ≥ 0.7.2). Replaces workspace.list + pane.list + + * tab.list for the poll loop. An older server rejects the method with an "unknown variant" error + * reply — StateEngine treats only that as a permanent signal to fall back to the three list calls. + */ + async sessionSnapshot(): Promise { + const r = await this.transport.request<{ type: string; snapshot: WireSnapshot }>("session.snapshot"); + return r.snapshot; + } + + /** + * Open a LONG-LIVED `events.subscribe` stream, or return `null` on a transport that can't stream + * (Windows/CLI). When non-null: after the ack, each line is an event; it exists ONLY to poke + * re-polls — callers must not treat events as state. `onDown` fires exactly once when the stream + * ends for any reason; `close()` is idempotent. Reconnect/backoff live in the caller (EventPoker). + */ + subscribeEvents(opts: SubscribeOptions): StreamHandle | null { + return this.transport.subscribeEvents(opts); + } + + /** + * Create a new tab in a workspace, opening a fresh shell pane. `cwd` is optional — omitted, the + * tab inherits the workspace's directory (verified). `focus:false` so we never yank the desktop + * TUI's focus. Returns the new shell pane to navigate into. + */ + async createTab(workspaceId: string, opts: { label?: string; cwd?: string } = {}): Promise { + const params: Record = { workspace_id: workspaceId, focus: false }; + if (opts.label) params.label = opts.label; + if (opts.cwd) params.cwd = opts.cwd; + const r = await this.transport.request<{ root_pane: WirePane }>("tab.create", params); + const p = r.root_pane; + return { paneId: p.pane_id, workspaceId: p.workspace_id, tabId: p.tab_id, cwd: p.cwd }; + } + + /** + * Create a new workspace ("space") with a fresh shell pane rooted at `cwd`. `focus:false` to + * leave the desktop TUI undisturbed. Returns the new shell pane (with its workspace label). + */ + async createWorkspace(opts: { cwd: string; label?: string }): Promise { + const params: Record = { cwd: opts.cwd, focus: false }; + if (opts.label) params.label = opts.label; + const r = await this.transport.request<{ workspace: WireWorkspace; root_pane: WirePane }>( + "workspace.create", + params, + ); + const p = r.root_pane; + return { + paneId: p.pane_id, + workspaceId: p.workspace_id, + workspaceLabel: r.workspace.label, + tabId: p.tab_id, + cwd: p.cwd, + }; + } + + async readPane( + paneId: string, + source: ReadSource, + lines: number, + format: ReadFormat = "text", + ): Promise { + const r = await this.transport.request<{ read: PaneRead }>("pane.read", { + pane_id: paneId, + source, + lines, + // "text" = plain (no escapes); "ansi" = SGR color codes (verified: no cursor sequences), + // parsed + escaped safely on the client to render a faithful, colored terminal mirror. + format, + }); + return r.read; + } + + /** Type literal text into a pane's terminal (does not submit). */ + sendPaneText(paneId: string, text: string): Promise { + return this.transport.request("pane.send_text", { pane_id: paneId, text }); + } + + /** Send key names (e.g. ["Enter"]) to a pane — used to submit a reply. */ + sendPaneKeys(paneId: string, keys: string[]): Promise { + return this.transport.request("pane.send_keys", { pane_id: paneId, keys }); + } + + /** Close a pane, terminating its agent ("kill"). Resolves on Herdr's `{type:"ok"}` reply. */ + closePane(paneId: string): Promise { + return this.transport.request("pane.close", { pane_id: paneId }); + } + + /** + * Set or clear a pane's label. `label: null` clears it (the key then disappears from pane + * records). Resolves on Herdr's `pane_info` reply — the returned pane isn't consumed here, the + * next snapshot poll carries the new label (pane.rename emits no event). Bad id → `pane_not_found`. + */ + renamePane(paneId: string, label: string | null): Promise { + return this.transport.request("pane.rename", { pane_id: paneId, label }); + } + + /** + * Set a tab's label. Unlike {@link renamePane}, `label` is a NON-null string: herdr's `tab.rename` + * rejects `null` (`invalid type: null, expected a string`) and stores an empty string literally + * rather than clearing to the default number — both live-verified 2026-07-19 — so a tab has no + * "clear". Resolves on herdr's `tab_info` reply; the new label surfaces on the next snapshot poll + * (tab.rename also emits a `tab_renamed` event, which Collie doesn't consume). Bad id → `tab_not_found`. + */ + renameTab(tabId: string, label: string): Promise { + return this.transport.request("tab.rename", { tab_id: tabId, label }); + } + + /** + * Close a tab, terminating EVERY pane inside it (live-verified 2026-07-19: the tab's shell/agent + * panes all disappear with it — closing a tab is a bulk pane-close). Resolves on herdr's + * `{type:"ok"}` reply; the closure surfaces on the next `session.snapshot` poll (tab.close also + * emits a `tab_closed` event, which Collie doesn't consume). Bad id → `tab_not_found`. + */ + closeTab(tabId: string): Promise { + return this.transport.request("tab.close", { tab_id: tabId }); + } + + /** Reachability check for the connected/disconnected banner. */ + async ping(): Promise { + try { + await this.listWorkspaces(); + return true; + } catch { + return false; + } + } +} + +/** Which transport to use. `auto` = CLI on Windows, socket elsewhere; the others force it. */ +export type TransportMode = "auto" | "cli" | "socket"; + +/** + * Build a HerdrClient over the right transport. `auto` (the default) picks the `herdr` CLI on + * Windows — where Herdr's socket is a named pipe Bun can't open (see {@link CliTransport}) — and the + * Unix socket everywhere else. `cli`/`socket` force one regardless of platform, which is how someone + * on macOS/Linux can exercise the Windows path (COLLIE_HERDR_TRANSPORT). `herdrBin` is only consulted + * by the CLI transport. + */ +export function createHerdrClient(opts: { + socketPath: string; + herdrBin: string; + transport?: TransportMode; + timeoutMs?: number; +}): HerdrClient { + const mode = opts.transport ?? "auto"; + const useCli = mode === "cli" || (mode === "auto" && process.platform === "win32"); + const transport: Transport = useCli + ? new CliTransport(opts.socketPath, opts.herdrBin, opts.timeoutMs) + : new SocketTransport(opts.socketPath, opts.timeoutMs); + return new HerdrClient(transport); +} + +let idCounter = 0; + +// ───────────────────────────────────────────────────────────────────────────── +// mac/Linux transport: one-shot JSON-RPC over Herdr's Unix socket, plus the +// long-lived events.subscribe stream. +// ───────────────────────────────────────────────────────────────────────────── +export class SocketTransport implements Transport { constructor( private readonly socketPath: string, private readonly timeoutMs = 5000, ) {} /** One request, one reply, one connection. Rejects on error reply, timeout, or early close. */ - private request(method: string, params: Record = {}): Promise { + request(method: string, params: Record = {}): Promise { const id = `b${++idCounter}`; return new Promise((resolve, reject) => { let buf = ""; @@ -177,45 +395,12 @@ export class HerdrClient { }); } - async listWorkspaces(): Promise { - const r = await this.request<{ workspaces: WireWorkspace[] }>("workspace.list"); - return r.workspaces; - } - - async listPanes(): Promise { - const r = await this.request<{ panes: WirePane[] }>("pane.list"); - return r.panes; - } - - /** All tabs across every workspace (`tab.list` with no filter returns the full set). */ - async listTabs(): Promise { - const r = await this.request<{ tabs: WireTab[] }>("tab.list"); - return r.tabs; - } - /** - * The whole herd in one round-trip (herdr ≥ 0.7.2). Replaces workspace.list + pane.list + - * tab.list for the poll loop. An older server rejects the method with an "unknown variant" error - * reply — StateEngine treats only that as a permanent signal to fall back to the three list calls. + * The socket streams, so this always returns a handle. After the ack each line is an event; + * `onDown` fires exactly once when the stream ends for any reason (error line, socket error, + * close, or a 5s ack timeout); `close()` is idempotent and ends it with reason "closed". */ - async sessionSnapshot(): Promise { - const r = await this.request<{ type: string; snapshot: WireSnapshot }>("session.snapshot"); - return r.snapshot; - } - - /** - * Open a LONG-LIVED `events.subscribe` stream. Unlike every other method here (one-shot), this - * connection stays open: after the ack, each line is an event. It exists ONLY to poke re-polls — - * callers must not treat events as state. `onDown` fires exactly once when the stream ends for any - * reason (error line, socket error, close, or a 5s ack timeout); `close()` is idempotent and also - * ends it with reason "closed". Reconnect/backoff live in the caller (see EventPoker). - */ - subscribeEvents(opts: { - subscriptions: Array<{ type: string; pane_id?: string }>; - onUp: () => void; - onEvent: (event: string, data: unknown) => void; - onDown: (reason: string) => void; - }): { close(): void } { + subscribeEvents(opts: SubscribeOptions): StreamHandle { const id = `es${++idCounter}`; const decoder = new TextDecoder("utf-8"); let buf = ""; @@ -309,111 +494,231 @@ export class HerdrClient { return { close: () => fireDown("closed") }; } +} - /** - * Create a new tab in a workspace, opening a fresh shell pane. `cwd` is optional — omitted, the - * tab inherits the workspace's directory (verified). `focus:false` so we never yank the desktop - * TUI's focus. Returns the new shell pane to navigate into. - */ - async createTab(workspaceId: string, opts: { label?: string; cwd?: string } = {}): Promise { - const params: Record = { workspace_id: workspaceId, focus: false }; - if (opts.label) params.label = opts.label; - if (opts.cwd) params.cwd = opts.cwd; - const r = await this.request<{ root_pane: WirePane }>("tab.create", params); - const p = r.root_pane; - return { paneId: p.pane_id, workspaceId: p.workspace_id, tabId: p.tab_id, cwd: p.cwd }; - } +// ───────────────────────────────────────────────────────────────────────────── +// Windows transport: spawn the `herdr` CLI per RPC. +// +// WHY THIS SPAWNS THE CLI INSTEAD OF OPENING THE SOCKET: +// On Windows, Herdr does NOT expose a filesystem AF_UNIX socket. It uses the Rust +// `interprocess` crate, which maps the socket path onto a Windows *named pipe* +// (`\\.\pipe\`) guarded by an in-crate handshake. Bun's `Bun.connect({unix})` +// targets native AF_UNIX and can't reach a named pipe at all; even a pipe-aware raw +// client (Node net / .NET NamedPipeClientStream) gets the connection accepted and +// then immediately EOF'd, because it doesn't speak the crate's handshake. Verified +// empirically 2026-07-13. So the ONLY reliable local client for that pipe is the +// same-version `herdr` binary itself — which exposes every method Collie needs as a +// CLI subcommand and emits the identical JSON envelopes. We shell out to it. +// +// One process spawn per RPC. `events.subscribe` has no CLI equivalent, so subscribeEvents +// returns null (see below) — StateEngine already treats events as a mere poke, never a +// source of truth, so correctness is unaffected; only poke latency changes. +// ───────────────────────────────────────────────────────────────────────────── - /** - * Create a new workspace ("space") with a fresh shell pane rooted at `cwd`. `focus:false` to - * leave the desktop TUI undisturbed. Returns the new shell pane (with its workspace label). - */ - async createWorkspace(opts: { cwd: string; label?: string }): Promise { - const params: Record = { cwd: opts.cwd, focus: false }; - if (opts.label) params.label = opts.label; - const r = await this.request<{ - workspace: WireWorkspace; - root_pane: WirePane; - }>("workspace.create", params); - const p = r.root_pane; - return { - paneId: p.pane_id, - workspaceId: p.workspace_id, - workspaceLabel: r.workspace.label, - tabId: p.tab_id, - cwd: p.cwd, - }; - } +/** Outcome of one `herdr` CLI invocation: captured stdout/stderr and the process exit code. */ +export interface CliResult { + code: number; + stdout: string; + stderr: string; +} - async readPane( - paneId: string, - source: ReadSource, - lines: number, - format: ReadFormat = "text", - ): Promise { - const r = await this.request<{ read: PaneRead }>("pane.read", { - pane_id: paneId, - source, - lines, - // "text" = plain (no escapes); "ansi" = SGR color codes (verified: no cursor sequences), - // parsed + escaped safely on the client to render a faithful, colored terminal mirror. - format, - }); - return r.read; - } +/** + * Spawns one `herdr` invocation and captures its result. Injected into {@link CliTransport} so the + * argv table — the one part neither reader can verify by eye — is exercised by `bun test` with a + * fake runner, instead of shelling out for real. + */ +export type CliRunner = ( + bin: string, + args: string[], + env: Record, + timeoutMs: number, +) => Promise; + +/** How to interpret a CLI invocation's output for a given method. */ +type ResultKind = "json" | "text" | "void"; + +/** Per-method CLI mapping: how to build argv from the JSON-RPC params, and how to read the result. */ +interface CliSpec { + argv: (params: Record) => string[]; + kind: ResultKind; +} - /** Type literal text into a pane's terminal (does not submit). */ - sendPaneText(paneId: string, text: string): Promise { - return this.request("pane.send_text", { pane_id: paneId, text }); - } +const str = (v: unknown): string => String(v ?? ""); + +// method → (argv builder, result kind). The single source of truth for the Windows CLI surface; +// adding a Herdr method means one entry here plus the typed method on HerdrClient. +const CLI_SPECS: Record = { + "workspace.list": { kind: "json", argv: () => ["workspace", "list"] }, + "pane.list": { kind: "json", argv: () => ["pane", "list"] }, + "tab.list": { kind: "json", argv: () => ["tab", "list"] }, + "session.snapshot": { kind: "json", argv: () => ["api", "snapshot"] }, + "tab.create": { + kind: "json", + argv: (p) => { + const args = ["tab", "create", "--workspace", str(p.workspace_id), "--no-focus"]; + if (p.label) args.push("--label", str(p.label)); + if (p.cwd) args.push("--cwd", str(p.cwd)); + return args; + }, + }, + "workspace.create": { + kind: "json", + argv: (p) => { + const args = ["workspace", "create", "--cwd", str(p.cwd), "--no-focus"]; + if (p.label) args.push("--label", str(p.label)); + return args; + }, + }, + "pane.read": { + kind: "text", + argv: (p) => [ + "pane", + "read", + str(p.pane_id), + "--source", + str(p.source), + "--lines", + str(p.lines), + "--format", + str(p.format), + ], + }, + "pane.send_text": { kind: "void", argv: (p) => ["pane", "send-text", str(p.pane_id), str(p.text)] }, + "pane.send_keys": { + kind: "void", + argv: (p) => ["pane", "send-keys", str(p.pane_id), ...(p.keys as string[])], + }, + "pane.close": { kind: "void", argv: (p) => ["pane", "close", str(p.pane_id)] }, + "pane.rename": { + kind: "void", + // herdr's socket clears a label with `label:null`; the CLI clears it with `--clear`. + argv: (p) => ["pane", "rename", str(p.pane_id), p.label === null ? "--clear" : str(p.label)], + }, + "tab.rename": { kind: "void", argv: (p) => ["tab", "rename", str(p.tab_id), str(p.label)] }, + "tab.close": { kind: "void", argv: (p) => ["tab", "close", str(p.tab_id)] }, +}; - /** Send key names (e.g. ["Enter"]) to a pane — used to submit a reply. */ - sendPaneKeys(paneId: string, keys: string[]): Promise { - return this.request("pane.send_keys", { pane_id: paneId, keys }); +/** + * Default runner: spawn `herdr ` with the session's socket in the env, capturing + * stdout/stderr/exit, killing a hung CLI at `timeoutMs`. A missing binary is translated to an + * actionable message rather than a raw ENOENT, since that's the one setup error a Windows user hits. + */ +const defaultCliRunner: CliRunner = async (bin, args, env, timeoutMs) => { + let proc: Bun.Subprocess<"ignore", "pipe", "pipe">; + try { + proc = Bun.spawn([bin, ...args], { env, stdout: "pipe", stderr: "pipe", stdin: "ignore" }); + } catch (e) { + const err = e as { code?: string; message?: string }; + if (err.code === "ENOENT") { + throw new Error( + `herdr not found at "${bin}" — set HERDR_BIN_PATH or COLLIE_HERDR_BIN to the herdr executable`, + ); + } + throw new Error(`herdr spawn failed ("${bin}"): ${err.message ?? String(e)}`); } - /** Close a pane, terminating its agent ("kill"). Resolves on Herdr's `{type:"ok"}` reply. */ - closePane(paneId: string): Promise { - return this.request("pane.close", { pane_id: paneId }); + // Kill a hung CLI so a wedged pipe can't stall a poll tick forever (mirrors the socket timeout). + const killer = setTimeout(() => { + try { + proc.kill(); + } catch { + /* already exited */ + } + }, timeoutMs); + + try { + const [stdout, stderr, code] = await Promise.all([ + new Response(proc.stdout).text(), + new Response(proc.stderr).text(), + proc.exited, + ]); + return { code, stdout, stderr }; + } finally { + clearTimeout(killer); } +}; +export class CliTransport implements Transport { /** - * Set or clear a pane's label. `label: null` clears it (the key then disappears from pane - * records). Resolves on Herdr's `pane_info` reply — the returned pane isn't consumed here, the - * next snapshot poll carries the new label (pane.rename emits no event). Bad id → `pane_not_found`. + * @param socketPath Herdr's control socket/pipe path. Passed to every CLI call via + * `HERDR_SOCKET_PATH` so a multi-session bridge targets the right herd. + * @param herdrBin Absolute path to `herdr` (or `herdr.exe`). Resolved once in config. + * @param timeoutMs Per-invocation wall-clock budget; a hung CLI is killed and the call rejects. + * @param runner Spawns one invocation; injected so the argv table is unit-testable. */ - renamePane(paneId: string, label: string | null): Promise { - return this.request("pane.rename", { pane_id: paneId, label }); - } + constructor( + private readonly socketPath: string, + private readonly herdrBin: string, + private readonly timeoutMs = 5000, + private readonly runner: CliRunner = defaultCliRunner, + ) {} - /** - * Set a tab's label. Unlike {@link renamePane}, `label` is a NON-null string: herdr's `tab.rename` - * rejects `null` (`invalid type: null, expected a string`) and stores an empty string literally - * rather than clearing to the default number — both live-verified 2026-07-19 — so a tab has no - * "clear". Resolves on herdr's `tab_info` reply; the new label surfaces on the next snapshot poll - * (tab.rename also emits a `tab_renamed` event, which Collie doesn't consume). Bad id → `tab_not_found`. - */ - renameTab(tabId: string, label: string): Promise { - return this.request("tab.rename", { tab_id: tabId, label }); + async request(method: string, params: Record = {}): Promise { + const spec = CLI_SPECS[method]; + if (!spec) throw new Error(`herdr ${method}: no CLI mapping (unsupported on the Windows transport)`); + + const r = await this.runner(this.herdrBin, spec.argv(params), { ...process.env, HERDR_SOCKET_PATH: this.socketPath }, this.timeoutMs); + + if (spec.kind === "void") { + if (r.code !== 0) throw new Error(`herdr ${method}: ${this.errText(r)}`); + return undefined as T; + } + + if (spec.kind === "text") { + // `herdr pane read` prints the pane's RAW TEXT to stdout, not a JSON envelope — so synthesize + // the {read} envelope the rest of the bridge expects. `revision` is a stub on herdr 0.7.x + // (always 0), matching what the socket path returned anyway. + if (r.code !== 0) throw new Error(`herdr ${method}: ${this.errText(r)}`); + return { + read: { + pane_id: str(params.pane_id), + text: r.stdout, + truncated: approximateTruncated(r.stdout, Number(params.lines)), + revision: 0, + }, + } as T; + } + + // json: the CLI prints a full `{"id","result":{...}}` envelope on stdout (exit 0), or an error + // envelope / plain transport line on stderr (exit ≠ 0). decodeReplyLine unwraps `result` and + // throws the error message — which carries "unknown variant" for the session.snapshot fallback. + if (r.code === 0 && r.stdout.trim()) return decodeReplyLine(r.stdout.trim(), method); + throw new Error(`herdr ${method}: ${this.errText(r)}`); } - /** - * Close a tab, terminating EVERY pane inside it (live-verified 2026-07-19: the tab's shell/agent - * panes all disappear with it — closing a tab is a bulk pane-close). Resolves on herdr's - * `{type:"ok"}` reply; the closure surfaces on the next `session.snapshot` poll (tab.close also - * emits a `tab_closed` event, which Collie doesn't consume). Bad id → `tab_not_found`. - */ - closeTab(tabId: string): Promise { - return this.request("tab.close", { tab_id: tabId }); + /** The CLI has no streaming transport; poll-only. Null tells EventPoker so — not an error to retry. */ + subscribeEvents(_opts: SubscribeOptions): StreamHandle | null { + return null; } - /** Reachability check for the connected/disconnected banner. */ - async ping(): Promise { + /** Human-readable failure text from a CLI result (error-envelope message, else raw stderr). */ + private errText(r: CliResult): string { + const raw = (r.stderr || r.stdout).trim(); try { - await this.listWorkspaces(); - return true; + const parsed = JSON.parse(raw) as { + error?: { code?: string; message?: string }; + code?: string; + message?: string; + }; + const err = parsed.error ?? parsed; + if (err && (err.code || err.message)) return `${err.code ?? "error"}: ${err.message ?? ""}`.trim(); } catch { - return false; + /* not JSON — fall through to raw */ } + return raw || `exited ${r.code}`; } } + +/** + * Approximate herdr's `truncated` flag, which the CLI doesn't expose. If a `--lines N` read comes + * back with at least N lines, older scrollback almost certainly exists beyond the window, so report + * truncated — that's what gates agent-chat's "load older lines" button. An exact count isn't + * possible without the flag; this errs toward showing the button rather than hiding history. + */ +function approximateTruncated(text: string, requestedLines: number): boolean { + if (!Number.isFinite(requestedLines) || requestedLines <= 0) return false; + const lines = text.split(/\r?\n/); + // A trailing newline yields a final empty element that isn't a real line. + if (lines.length > 0 && lines[lines.length - 1] === "") lines.pop(); + return lines.length >= requestedLines; +} diff --git a/bridge/index.ts b/bridge/index.ts index 25ae2b8..4b3a44e 100644 --- a/bridge/index.ts +++ b/bridge/index.ts @@ -4,8 +4,8 @@ import { join } from "node:path"; import { AuditLog, fileAuditAppender } from "./audit.ts"; import { loadConfig } from "./config.ts"; -import { EventPoker } from "./event-poker.ts"; -import { HerdrClient } from "./herdr-client.ts"; +import { EventPoker, type PollMode } from "./event-poker.ts"; +import { createHerdrClient } from "./herdr-client.ts"; import { NotificationCoordinator, makeNotifySink, type NotifyClock } from "./notifications.ts"; import { NotifyPrefsStore } from "./notify-prefs.ts"; import { Push } from "./push.ts"; @@ -104,16 +104,22 @@ updateTimer.unref(); // registry calls this for the primary at construction and for each session discovered later. Push, // snooze, notify-prefs, the audit log and the uploads dir stay process-global (shared here). const makeSession: SessionFactory = (name, socketPath, isPrimary) => { - const herdr = new HerdrClient(socketPath); + const herdr = createHerdrClient({ socketPath, herdrBin: cfg.herdrBin ?? "herdr", transport: cfg.transportMode }); const engine = new StateEngine(herdr, cfg.pollMs); // Event-poked polling: a long-lived events.subscribe stream pokes an immediate re-poll on any herd // change, and while it's healthy the interval relaxes to the safety-net cadence. Events are ONLY a // poke — the snapshot poll stays the source of truth — so a missed event costs one interval, not // correctness. The fresh snapshot after any pane lifecycle change re-scopes the subscriptions. + // A transport with no stream (Windows CLI) reports `no-events` and polls on its own cadence. const poker = new EventPoker(herdr); + const cadenceFor: Record = { + streaming: cfg.pollIdleMs, + fast: cfg.pollMs, + "no-events": cfg.pollNoEventsMs, + }; poker.onPoke(() => engine.pokeNow()); - poker.onHealth((h) => engine.setCadence(h ? cfg.pollIdleMs : cfg.pollMs)); + poker.onPollMode((mode) => engine.setCadence(cadenceFor[mode])); engine.onUpdate((s) => poker.setAgentPanes(s.agents.map((a) => a.paneId))); // Background notifications on lifecycle transitions (foreground toasts are computed client-side by diff --git a/bridge/server.test.ts b/bridge/server.test.ts index 329255b..320120d 100644 --- a/bridge/server.test.ts +++ b/bridge/server.test.ts @@ -31,10 +31,12 @@ function req(headers: Record): Request { function cfg(overrides: Partial = {}): Config { return { socketPath: "/tmp/herdr.sock", + transportMode: "auto", port: 8787, host: "127.0.0.1", pollMs: 1500, pollIdleMs: 12_000, + pollNoEventsMs: 4000, notifyDelayMs: 30_000, readLines: 200, submitKeys: ["Enter"], diff --git a/herdr-plugin.toml b/herdr-plugin.toml index 77babf4..35fc486 100644 --- a/herdr-plugin.toml +++ b/herdr-plugin.toml @@ -3,7 +3,7 @@ name = "Collie" version = "0.14.2" min_herdr_version = "0.7.0" description = "Mobile web UI to monitor and reply to your agent herd, served over Tailscale" -platforms = ["linux", "macos"] +platforms = ["linux", "macos", "windows"] # This plugin is a thin launcher. The actual bridge runs as a systemd --user service so it # outlives Herdr restarts (see ARCHITECTURE.md §3). The actions below shell out to @@ -16,7 +16,7 @@ platforms = ["linux", "macos"] # Still needs Bun on PATH — building the frontend is the one unavoidable Bun step. [[build]] command = ["bash", "scripts/collie-ctl.sh", "build"] -platforms = ["linux", "macos"] +platforms = ["linux", "macos", "windows"] [[actions]] id = "start"