From 5cad1582a8e300ec06ffc776335cbc98519d8419 Mon Sep 17 00:00:00 2001 From: Zoltan Onody Date: Sat, 29 Aug 2026 17:11:19 +0200 Subject: [PATCH] feat(herdr): add herdr reporter extension for lifecycle state and session identity Add a kimchi-side herdr integration that reports lifecycle state (idle/working/blocked), blocked prompts, and session identity to herdr when running inside a herdr-managed pane. - src/extensions/herdr-reporter.ts: JSON-RPC socket client, single-writer async queue, lifecycle state machine, session reference reporting, and HERDR_ENV gating - src/cli.ts: register herdrReporterExtension in extensionFactories - src/extensions/herdr-reporter.test.ts: tests for env gating, state transitions, blocked refcounting, queue ordering, session reporting, socket error resilience, and beforeExit listener cleanup - docs/herdr.md: document running kimchi inside herdr and resuming with --session --- docs/herdr.md | 99 +++ src/cli.ts | 4 + src/extensions/herdr-reporter.test.ts | 970 ++++++++++++++++++++++++++ src/extensions/herdr-reporter.ts | 541 ++++++++++++++ 4 files changed, 1614 insertions(+) create mode 100644 docs/herdr.md create mode 100644 src/extensions/herdr-reporter.test.ts create mode 100644 src/extensions/herdr-reporter.ts diff --git a/docs/herdr.md b/docs/herdr.md new file mode 100644 index 000000000..294af2951 --- /dev/null +++ b/docs/herdr.md @@ -0,0 +1,99 @@ +# Herdr + +Kimchi runs inside [herdr](https://herdr.dev/) panes and reports its lifecycle state and session identity automatically — no kimchi-side configuration required. + +When herdr is present, kimchi publishes: + +- Its current **lifecycle state** (`idle` / `working` / `blocked`) +- Any **blocked prompt** awaiting input +- Its **session ID** and session file path so herdr can resume the session later + +herdr listens on its injected control socket and surfaces this information in the pane UI and status bar. + +--- + +## Running kimchi in herdr + +Start herdr in your project directory and open a kimchi pane: + +```bash +herdr # launches herdr in the current directory +``` + +Inside herdr, open `kimchi` in a new pane. herdr automatically injects the following environment variables into the pane: + +| Variable | Meaning | +|----------|---------| +| `HERDR_ENV=1` | Marks the pane as running under herdr | +| `HERDR_PANE_ID` | The pane identifier herdr uses to address the agent | +| `HERDR_SOCKET_PATH` | Path to the herdr control socket kimchi writes state to | +| `HERDR_BIN_PATH` | Path to the herdr binary, used by wrapper hints | + +Kimchi detects `HERDR_ENV=1` at session start and enables reporting. **No manual configuration is needed on the kimchi side** — there are no flags, environment variables, or settings entries to add. + +--- + +## Lifecycle state reporting + +Kimchi publishes one of three states to herdr's control socket as it runs: + +| State | When | +|-------|------| +| `idle` | Kimchi is waiting for the next user prompt (or between turns). | +| `working` | Kimchi is actively processing a turn — running tools, calling the model, etc. | +| `blocked` | A permission prompt or question is open and requires user input. | + +The state transitions track the session prompt: `idle` while the editor is waiting, `working` from the first tool call or model token through the end of the turn, and `blocked` whenever a confirmation/question overlay is shown. If a prompt times out or is dismissed, kimchi falls back to `idle`. + +### Blocked prompts + +When kimchi enters the `blocked` state, the payload includes the prompt text (truncated) so herdr can display a notification or surface it in the pane footer. + +### Session identity + +Kimchi also publishes: + +- **`session_id`** — the resumable session identifier (the same value accepted by `--session`). +- **`session_path`** — absolute path to the on-disk session file, for inspection and backup. + +herdr stores these so it can reattach a pane to the same session after a reconnect or restart. + +--- + +## Detection note + +herdr auto-detects the agent running in a pane from process and version metadata. Because kimchi is built on [pi-mono](https://github.com/badlogic/pi-mono), herdr may initially classify a kimchi pane as a **Pi agent** rather than as kimchi. + +This does not affect any reporting — state, blocked prompts, and session identity are all published regardless of how herdr labels the pane. + +To make herdr display the pane as kimchi, you have two options: + +1. **Wrapper hint** — set `HERDR_AGENT=kimchi` in the pane environment. herdr uses this value as the agent label: + + ```bash + HERDR_AGENT=kimchi kimchi + ``` + +2. **Wait for upstream detection** — a future herdr release will add native kimchi detection based on kimchi's own process metadata. + +--- + +## Session restore + +Kimchi persists sessions automatically on every turn. To resume a session from outside herdr: + +```bash +kimchi --session +``` + +The `` is the same value kimchi publishes to herdr as `session_id`. + +In herdr, the reported session reference lets herdr **store the resume id** for each pane, but the actual resume happens via the kimchi CLI — herdr does not replay session state itself. To restore a herdr-attached session, close the pane and re-launch kimchi with `--session `, or wire your own restore command to the pane's stored session id. + +--- + +## Troubleshooting + +- **No state shown in herdr** — confirm `HERDR_ENV=1` is present in the pane (`echo $HERDR_ENV`). If it's missing, the pane isn't running under herdr's agent manager. +- **`HERDR_SOCKET_PATH` not writable** — kimchi will log a warning and continue without reporting. State reporting is best-effort and never blocks the session. +- **Pane labeled as Pi** — see [Detection note](#detection-note) above; reporting still works, only the display label is affected. diff --git a/src/cli.ts b/src/cli.ts index 3bf492c1a..0e0589e36 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -60,6 +60,7 @@ import { setExperimentalFeaturesEnabled } from "./extensions/experimental.js" import explorationGuardExtension from "./extensions/exploration-guard.js" import fermentExtension from "./extensions/ferment/index.js" import helpExtension from "./extensions/help.js" +import herdrReporterExtension from "./extensions/herdr-reporter.js" import hiddenToolGuidanceExtension from "./extensions/hidden-tool-guidance.js" import hideThinkingExtension from "./extensions/hide-thinking.js" import ideAdapterExtension from "./extensions/ide-adapter/index.js" @@ -559,6 +560,9 @@ try { // First so its session_start handler syncs project trust onto the // settings watcher before any other handler reads settings. settingsTrustSyncExtension, + // Always-on lifecycle reporter — surfaces agent state to herdr (an + // out-of-process UI). No-op when HERDR_ENV is unset. + herdrReporterExtension, autoUpdateSettingsExtension, startupUpdateExtension, packageInstallGuardExtension, diff --git a/src/extensions/herdr-reporter.test.ts b/src/extensions/herdr-reporter.test.ts new file mode 100644 index 000000000..5205d8253 --- /dev/null +++ b/src/extensions/herdr-reporter.test.ts @@ -0,0 +1,970 @@ +/** + * Co-located tests for the herdr reporter extension. + * + * The reporter talks to herdr over a Unix domain socket using newline- + * delimited JSON-RPC. To keep these tests fast and hermetic we mock + * `node:net` with an in-memory fake socket server. The fake socket is + * itself an EventEmitter so it mirrors the surface the reporter uses + * (`connect`, `close`, `error`, `write`, `end`, `destroy`). + */ + +import { EventEmitter } from "node:events" +import type { EventBus, ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent" +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest" + +// --------------------------------------------------------------------------- +// node:net mock +// --------------------------------------------------------------------------- + +interface RecordedRequest { + jsonrpc: string + id: string + method: string + params: Record +} + +interface FakeSocket extends EventEmitter { + write: ReturnType + end: ReturnType + destroy: ReturnType +} + +const recordedRequests: RecordedRequest[] = [] +let destroyOnConnect = false + +function makeFakeSocket(): FakeSocket { + const socket = new EventEmitter() as FakeSocket + socket.write = vi.fn((data: string | Buffer, cb?: (err?: Error | null) => void) => { + try { + const text = typeof data === "string" ? data : data.toString("utf8") + const trimmed = text.replace(/\n$/, "") + if (trimmed.length > 0) { + const parsed = JSON.parse(trimmed) as RecordedRequest + recordedRequests.push(parsed) + } + } catch (err) { + cb?.(err as Error) + return false + } + cb?.(null) + return true + }) + // The reporter now writes via `socket.end(data, callback)` and resolves + // on the callback firing (Node's "finish" semantics: data flushed + + // FIN queued). Mirror that by parsing the data argument and invoking + // the callback on the next microtask, instead of relying on a `close` + // event — the peer's half of a real socket can stay open well past + // our short timeouts. + socket.end = vi.fn((data?: string | Buffer, cb?: () => void) => { + if (data !== undefined && data !== null) { + try { + const text = typeof data === "string" ? data : data.toString("utf8") + const trimmed = text.replace(/\n$/, "") + if (trimmed.length > 0) { + const parsed = JSON.parse(trimmed) as RecordedRequest + recordedRequests.push(parsed) + } + } catch { + queueMicrotask(() => cb?.()) + return + } + } + queueMicrotask(() => cb?.()) + }) + socket.destroy = vi.fn(() => { + queueMicrotask(() => socket.emit("close")) + }) + return socket +} + +vi.mock("node:net", () => { + const createConnection = vi.fn((_address: string) => { + const sock = makeFakeSocket() + // Defer the connect event so listeners attach first, just like a + // real socket would. + queueMicrotask(() => { + if (destroyOnConnect) { + sock.destroy() + sock.emit("error", new Error("herdr socket destroyed on connect")) + } else { + sock.emit("connect") + } + }) + return sock + }) + return { + default: { createConnection }, + createConnection, + } +}) + +// Import after the mock is registered so the reporter picks it up. +const herdrReporterModule = await import("./herdr-reporter.js") +const { createHerdrReporter, readHerdrEnv, beforeExitReleasers } = herdrReporterModule +const herdrReporterExtension = herdrReporterModule.default + +// --------------------------------------------------------------------------- +// Test helpers +// --------------------------------------------------------------------------- + +function makeCtx(overrides: Partial = {}): ExtensionContext { + return { + mode: "tui", + hasUI: true, + cwd: "/tmp", + ui: {} as ExtensionContext["ui"], + sessionManager: { + getSessionFile: () => "/tmp/session.jsonl", + getSessionId: () => "sess-abc", + } as unknown as ExtensionContext["sessionManager"], + modelRegistry: {} as ExtensionContext["modelRegistry"], + model: undefined, + scopedModels: [], + isIdle: () => true, + isProjectTrusted: () => true, + signal: undefined, + abort: () => {}, + hasPendingMessages: () => false, + shutdown: () => {}, + getContextUsage: () => undefined, + compact: () => {}, + getSystemPrompt: () => "", + ...overrides, + } +} + +interface FakePi { + handlers: Map unknown>> + events: EventBus & { emit: ReturnType; listeners: Map void>> } + api: ExtensionAPI & { herdrReporter?: unknown } +} + +function makeFakePi(): FakePi { + const handlers = new Map unknown>>() + const listeners = new Map void>>() + const events = { + emit: vi.fn((channel: string, data: unknown) => { + const list = listeners.get(channel) ?? [] + for (const fn of list) fn(data) + }), + on: vi.fn((channel: string, handler: (data: unknown) => void) => { + if (!listeners.has(channel)) listeners.set(channel, []) + listeners.get(channel)?.push(handler) + return () => { + const list = listeners.get(channel) ?? [] + const idx = list.indexOf(handler) + if (idx >= 0) list.splice(idx, 1) + } + }), + listeners, + } as unknown as EventBus & { emit: ReturnType; listeners: Map void>> } + + const api = { + on: vi.fn((event: string, handler: (...args: unknown[]) => unknown) => { + if (!handlers.has(event)) handlers.set(event, []) + handlers.get(event)?.push(handler) + }), + events, + } as unknown as ExtensionAPI & { herdrReporter?: unknown } + + return { handlers, events, api } +} + +function getHandler(pi: FakePi, event: string): (...args: unknown[]) => Promise | void { + const list = pi.handlers.get(event) + if (!list || list.length === 0) throw new Error(`No handler for ${event}`) + return list[0] as (...args: unknown[]) => Promise | void +} + +// --------------------------------------------------------------------------- +// Environment + mock setup +// --------------------------------------------------------------------------- + +let originalEnv: Record + +beforeEach(() => { + originalEnv = { + HERDR_ENV: process.env.HERDR_ENV, + HERDR_SOCKET_PATH: process.env.HERDR_SOCKET_PATH, + HERDR_PANE_ID: process.env.HERDR_PANE_ID, + HERDR_BIN_PATH: process.env.HERDR_BIN_PATH, + } + recordedRequests.length = 0 + destroyOnConnect = false +}) + +afterEach(() => { + for (const [key, value] of Object.entries(originalEnv)) { + if (value === undefined) { + delete process.env[key] + } else { + process.env[key] = value + } + } +}) + +// --------------------------------------------------------------------------- +// readHerdrEnv +// --------------------------------------------------------------------------- + +describe("readHerdrEnv", () => { + it("returns enabled=false when HERDR_ENV is unset", () => { + delete process.env.HERDR_ENV + delete process.env.HERDR_SOCKET_PATH + delete process.env.HERDR_PANE_ID + + const view = readHerdrEnv() + expect(view.enabled).toBe(false) + }) + + it("returns enabled=false when HERDR_ENV is not '1'", () => { + process.env.HERDR_ENV = "yes" + process.env.HERDR_SOCKET_PATH = "/tmp/herdr.sock" + process.env.HERDR_PANE_ID = "pane-1" + + const view = readHerdrEnv() + expect(view.enabled).toBe(false) + }) + + it("returns enabled=false when HERDR_ENV=1 but socket path is missing", () => { + process.env.HERDR_ENV = "1" + delete process.env.HERDR_SOCKET_PATH + process.env.HERDR_PANE_ID = "pane-1" + + expect(readHerdrEnv().enabled).toBe(false) + }) + + it("returns enabled=false when HERDR_ENV=1 but pane id is missing", () => { + process.env.HERDR_ENV = "1" + process.env.HERDR_SOCKET_PATH = "/tmp/herdr.sock" + delete process.env.HERDR_PANE_ID + + expect(readHerdrEnv().enabled).toBe(false) + }) + + it("returns enabled=true with all fields populated when fully configured", () => { + process.env.HERDR_ENV = "1" + process.env.HERDR_SOCKET_PATH = "/tmp/herdr.sock" + process.env.HERDR_PANE_ID = "pane-42" + process.env.HERDR_BIN_PATH = "/usr/local/bin/herdr" + + const view = readHerdrEnv() + expect(view.enabled).toBe(true) + expect(view.socketPath).toBe("/tmp/herdr.sock") + expect(view.paneId).toBe("pane-42") + expect(view.binPath).toBe("/usr/local/bin/herdr") + }) +}) + +// --------------------------------------------------------------------------- +// createHerdrReporter +// --------------------------------------------------------------------------- + +function flushMicrotasks(): Promise { + return new Promise((resolve) => setImmediate(resolve)) +} + +describe("createHerdrReporter", () => { + it("sends pane.report_agent with state, monotonic seq, and a session id", async () => { + const reporter = createHerdrReporter({ + paneId: "pane-1", + socketPath: "/tmp/herdr.sock", + source: "herdr:kimchi", + agent: "kimchi", + }) + reporter.updateSessionRef({ id: "sess-42" }) + + reporter.reportState("working") + await reporter.drain() + + expect(recordedRequests).toHaveLength(1) + const [req] = recordedRequests + expect(req.method).toBe("pane.report_agent") + expect(req.params.state).toBe("working") + expect(req.params.pane_id).toBe("pane-1") + expect(req.params.source).toBe("herdr:kimchi") + expect(req.params.agent).toBe("kimchi") + expect(req.params.agent_session_id).toBe("sess-42") + expect(req.params.seq).toBe(Number(req.id)) + expect(typeof req.id).toBe("string") + }) + + it("prefers agent_session_path over agent_session_id when both are set", async () => { + const reporter = createHerdrReporter({ + paneId: "pane-1", + socketPath: "/tmp/herdr.sock", + source: "herdr:kimchi", + agent: "kimchi", + }) + reporter.updateSessionRef({ id: "sess-id", path: "/tmp/sess.jsonl" }) + + reporter.reportState("idle") + await reporter.drain() + + expect(recordedRequests).toHaveLength(1) + expect(recordedRequests[0].params.agent_session_path).toBe("/tmp/sess.jsonl") + expect(recordedRequests[0].params.agent_session_id).toBeUndefined() + }) + + it("sends pane.report_agent_session with session_start_source", async () => { + const reporter = createHerdrReporter({ + paneId: "pane-1", + socketPath: "/tmp/herdr.sock", + source: "herdr:kimchi", + agent: "kimchi", + }) + + reporter.reportSession({ id: "sess-99" }, "startup") + await reporter.drain() + + expect(recordedRequests).toHaveLength(1) + const [req] = recordedRequests + expect(req.method).toBe("pane.report_agent_session") + expect(req.params.agent_session_id).toBe("sess-99") + expect(req.params.session_start_source).toBe("startup") + expect(req.params.pane_id).toBe("pane-1") + expect(req.params.source).toBe("herdr:kimchi") + expect(req.params.agent).toBe("kimchi") + }) + + it("omits session_start_source when not provided", async () => { + const reporter = createHerdrReporter({ + paneId: "pane-1", + socketPath: "/tmp/herdr.sock", + source: "herdr:kimchi", + agent: "kimchi", + }) + + reporter.reportSession({ id: "sess-99" }) + await reporter.drain() + + expect(recordedRequests[0].params.session_start_source).toBeUndefined() + }) + + it("reports queued reports in submission order with strictly increasing seq", async () => { + const reporter = createHerdrReporter({ + paneId: "pane-1", + socketPath: "/tmp/herdr.sock", + source: "herdr:kimchi", + agent: "kimchi", + }) + reporter.updateSessionRef({ id: "sess-1" }) + + reporter.reportState("working") + reporter.reportState("idle") + reporter.reportState("blocked", "Permission: bash") + reporter.reportSession({ id: "sess-2" }, "resume") + await reporter.drain() + + const seqs = recordedRequests.map((r) => Number(r.id)) + for (let i = 1; i < seqs.length; i++) { + expect(seqs[i]).toBeGreaterThan(seqs[i - 1]) + } + + expect(recordedRequests.map((r) => r.method)).toEqual([ + "pane.report_agent", + "pane.report_agent", + "pane.report_agent", + "pane.report_agent_session", + ]) + expect(recordedRequests[0].params.state).toBe("working") + expect(recordedRequests[1].params.state).toBe("idle") + expect(recordedRequests[2].params.state).toBe("blocked") + expect(recordedRequests[2].params.message).toBe("Permission: bash") + expect(recordedRequests[3].params.agent_session_id).toBe("sess-2") + expect(recordedRequests[3].params.session_start_source).toBe("resume") + }) + + it("propagates the per-method seq into params.seq", async () => { + const reporter = createHerdrReporter({ + paneId: "pane-1", + socketPath: "/tmp/herdr.sock", + source: "herdr:kimchi", + agent: "kimchi", + }) + + reporter.reportState("working") + reporter.reportState("idle") + await reporter.drain() + + for (const req of recordedRequests) { + expect(req.params.seq).toBe(Number(req.id)) + } + }) + + it("swallows socket errors without throwing", async () => { + destroyOnConnect = true + const reporter = createHerdrReporter({ + paneId: "pane-1", + socketPath: "/tmp/herdr.sock", + source: "herdr:kimchi", + agent: "kimchi", + }) + + expect(() => reporter.reportState("working")).not.toThrow() + // Drain must resolve even when every send fails; allow plenty of + // microtask flushes for the retry path to complete. + for (let i = 0; i < 5; i++) await flushMicrotasks() + await expect(reporter.drain()).resolves.toBeUndefined() + expect(recordedRequests).toHaveLength(0) + }) + + it("release() prevents new reports from being enqueued", async () => { + const reporter = createHerdrReporter({ + paneId: "pane-1", + socketPath: "/tmp/herdr.sock", + source: "herdr:kimchi", + agent: "kimchi", + }) + + reporter.reportState("working") + await reporter.release() + reporter.reportState("idle") + + expect(recordedRequests).toHaveLength(1) + expect(recordedRequests[0].params.state).toBe("working") + }) + + it("updateSessionRef changes the agent_session stamp on subsequent state reports", async () => { + const reporter = createHerdrReporter({ + paneId: "pane-1", + socketPath: "/tmp/herdr.sock", + source: "herdr:kimchi", + agent: "kimchi", + }) + + reporter.updateSessionRef({ id: "sess-a" }) + reporter.reportState("working") + reporter.updateSessionRef({ id: "sess-b" }) + reporter.reportState("idle") + await reporter.drain() + + expect(recordedRequests[0].params.agent_session_id).toBe("sess-a") + expect(recordedRequests[1].params.agent_session_id).toBe("sess-b") + }) +}) + +// --------------------------------------------------------------------------- +// herdrReporterExtension lifecycle +// --------------------------------------------------------------------------- + +describe("herdrReporterExtension", () => { + beforeEach(() => { + process.env.HERDR_ENV = "1" + process.env.HERDR_SOCKET_PATH = "/tmp/herdr.sock" + process.env.HERDR_PANE_ID = "pane-1" + }) + + it("is a no-op when HERDR_ENV is unset", () => { + delete process.env.HERDR_ENV + const pi = makeFakePi() + herdrReporterExtension(pi.api) + expect(pi.handlers.size).toBe(0) + expect(pi.api.herdrReporter).toBeUndefined() + }) + + it("registers handlers and exposes the reporter on pi", () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + expect(pi.handlers.has("session_start")).toBe(true) + expect(pi.handlers.has("agent_start")).toBe(true) + expect(pi.handlers.has("agent_settled")).toBe(true) + expect(pi.handlers.has("session_shutdown")).toBe(true) + expect(pi.api.herdrReporter).toBeDefined() + }) + + it("session_start in TUI mode reports session + initial idle state", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + const handler = getHandler(pi, "session_start") + await handler({ reason: "startup" }, makeCtx({ isIdle: () => true })) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + + const methods = recordedRequests.map((r) => r.method) + expect(methods).toContain("pane.report_agent_session") + expect(methods).toContain("pane.report_agent") + const initialState = recordedRequests.find((r) => r.method === "pane.report_agent") + expect(initialState?.params.state).toBe("idle") + const sessionReport = recordedRequests.find((r) => r.method === "pane.report_agent_session") + expect(sessionReport?.params.session_start_source).toBe("startup") + }) + + it("session_start in TUI mode reports working when isIdle is false", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + const handler = getHandler(pi, "session_start") + await handler({ reason: "resume" }, makeCtx({ isIdle: () => false })) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + + const stateReports = recordedRequests.filter((r) => r.method === "pane.report_agent") + expect(stateReports[0].params.state).toBe("working") + }) + + it("session_start in non-TUI mode is ignored", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + const handler = getHandler(pi, "session_start") + await handler({ reason: "startup" }, makeCtx({ mode: "rpc" })) + + expect(recordedRequests).toHaveLength(0) + }) + + it("agent_start switches the state machine to working", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + // Anchor the root session. + await getHandler(pi, "session_start")({ reason: "startup" }, makeCtx({ isIdle: () => true })) + + recordedRequests.length = 0 + getHandler(pi, "agent_start")({}, makeCtx({ isIdle: () => false })) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + + const stateReports = recordedRequests.filter((r) => r.method === "pane.report_agent") + expect(stateReports.some((r) => r.params.state === "working")).toBe(true) + }) + + it("agent_settled when isIdle=true switches the state machine to idle", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + await getHandler(pi, "session_start")({ reason: "startup" }, makeCtx({ isIdle: () => false })) + + recordedRequests.length = 0 + getHandler(pi, "agent_settled")({}, makeCtx({ isIdle: () => true })) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + + const stateReports = recordedRequests.filter((r) => r.method === "pane.report_agent") + expect(stateReports[stateReports.length - 1].params.state).toBe("idle") + }) + + it("agent_settled when isIdle=false does not change the state", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + await getHandler(pi, "session_start")({ reason: "startup" }, makeCtx({ isIdle: () => true })) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + + recordedRequests.length = 0 + getHandler(pi, "agent_settled")({}, makeCtx({ isIdle: () => false })) + await reporter.drain() + + const stateReports = recordedRequests.filter((r) => r.method === "pane.report_agent") + expect(stateReports).toHaveLength(0) + }) + + it("herdr:blocked active:true switches to blocked and active:false returns to idle", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + await getHandler(pi, "session_start")({ reason: "startup" }, makeCtx({ isIdle: () => true })) + + recordedRequests.length = 0 + pi.events.emit("herdr:blocked", { active: true, label: "Permission: write" }) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + + const blockedReports = recordedRequests.filter( + (r) => r.method === "pane.report_agent" && r.params.state === "blocked", + ) + expect(blockedReports.length).toBeGreaterThan(0) + expect(blockedReports[0].params.message).toBe("Permission: write") + + recordedRequests.length = 0 + pi.events.emit("herdr:blocked", { active: false }) + await reporter.drain() + + const stateReports = recordedRequests.filter((r) => r.method === "pane.report_agent") + expect(stateReports[stateReports.length - 1].params.state).toBe("idle") + }) + + it("refcounts nested herdr:blocked activations correctly", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + await getHandler(pi, "session_start")({ reason: "startup" }, makeCtx({ isIdle: () => true })) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + + // Two nested activations. Use the same label so the state-change + // deduper doesn't republish on the inner activation. + pi.events.emit("herdr:blocked", { active: true, label: "Permission" }) + pi.events.emit("herdr:blocked", { active: true, label: "Permission" }) + await reporter.drain() + + recordedRequests.length = 0 + + // First deactivation: counter goes to 1 — still blocked, no idle emit. + pi.events.emit("herdr:blocked", { active: false }) + await reporter.drain() + const afterFirstDeactivate = recordedRequests.filter((r) => r.method === "pane.report_agent") + expect(afterFirstDeactivate.some((r) => r.params.state === "idle")).toBe(false) + + // Second deactivation: counter reaches 0 — back to idle. + recordedRequests.length = 0 + pi.events.emit("herdr:blocked", { active: false }) + await reporter.drain() + const afterLastDeactivate = recordedRequests.filter((r) => r.method === "pane.report_agent") + expect(afterLastDeactivate.some((r) => r.params.state === "idle")).toBe(true) + }) + + it("blocked state takes precedence over working state", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + await getHandler(pi, "session_start")({ reason: "startup" }, makeCtx({ isIdle: () => false })) + + recordedRequests.length = 0 + pi.events.emit("herdr:blocked", { active: true, label: "Permission: bash" }) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + + const stateReports = recordedRequests.filter((r) => r.method === "pane.report_agent") + expect(stateReports[stateReports.length - 1].params.state).toBe("blocked") + }) + + it("session_shutdown triggers release and enqueues a final idle state report", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + await getHandler(pi, "session_start")({ reason: "startup" }, makeCtx({ isIdle: () => false })) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + + const releaseSpy = vi.spyOn(reporter, "release") + + recordedRequests.length = 0 + getHandler(pi, "session_shutdown")({}, makeCtx()) + + // release is fire-and-forget (voided), so flush microtasks before + // inspecting the spy. + await flushMicrotasks() + await reporter.drain() + + expect(releaseSpy).toHaveBeenCalledTimes(1) + + // The shutdown MUST emit a final `idle` report so herdr does not + // keep showing the last working/blocked state. + const stateAfterShutdown = recordedRequests.filter((r) => r.method === "pane.report_agent") + expect(stateAfterShutdown).toHaveLength(1) + expect(stateAfterShutdown[0].params.state).toBe("idle") + expect(stateAfterShutdown[0].params.message).toBeUndefined() + + // After release, further reports are dropped. + reporter.reportState("working") + await reporter.drain() + const stateAfterRelease = recordedRequests.filter((r) => r.method === "pane.report_agent") + expect(stateAfterRelease).toHaveLength(1) // still the idle we already saw + }) + + it("session_shutdown enqueues a final idle report even from a blocked state", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + await getHandler(pi, "session_start")({ reason: "startup" }, makeCtx({ isIdle: () => true })) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + + // Force the state machine into `blocked` so we can assert the + // release overrides it. + pi.events.emit("herdr:blocked", { active: true, label: "Permission: write" }) + await reporter.drain() + + recordedRequests.length = 0 + getHandler(pi, "session_shutdown")({}, makeCtx()) + await flushMicrotasks() + await reporter.drain() + + const stateAfterShutdown = recordedRequests.filter((r) => r.method === "pane.report_agent") + expect(stateAfterShutdown).toHaveLength(1) + expect(stateAfterShutdown[0].params.state).toBe("idle") + expect(stateAfterShutdown[0].params.message).toBeUndefined() + }) + + it("session_shutdown returns a promise that resolves after the final idle report has drained", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + await getHandler(pi, "session_start")({ reason: "startup" }, makeCtx({ isIdle: () => false })) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + + recordedRequests.length = 0 + const shutdownHandler = getHandler(pi, "session_shutdown") + const ret = shutdownHandler({}, makeCtx()) + // The handler is synchronous-fire-and-forget but the underlying + // release promise is exposed so callers (and the beforeExit + // backstop) can await it. We don't await here — instead we drain + // the reporter afterwards and confirm the idle report landed. + expect(ret).toBeUndefined() + + await flushMicrotasks() + await reporter.drain() + + const stateAfterShutdown = recordedRequests.filter((r) => r.method === "pane.report_agent") + expect(stateAfterShutdown).toHaveLength(1) + expect(stateAfterShutdown[0].params.state).toBe("idle") + }) + + it("non-TUI session_start keeps the extension inert until TUI session_start arrives", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + // First, a non-TUI session_start (e.g. an RPC heartbeat). + await getHandler(pi, "session_start")({ reason: "startup" }, makeCtx({ mode: "rpc" })) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + expect(recordedRequests).toHaveLength(0) + + // Now a TUI session_start anchors the root session. + await getHandler(pi, "session_start")({ reason: "startup" }, makeCtx({ mode: "tui", isIdle: () => true })) + await reporter.drain() + + expect(recordedRequests.length).toBeGreaterThan(0) + const sessionReports = recordedRequests.filter((r) => r.method === "pane.report_agent_session") + expect(sessionReports).toHaveLength(1) + }) + + it("publishes working then blocked then working when blocked arrives mid-turn", async () => { + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + // Anchor root session in working state. + await getHandler(pi, "session_start")({ reason: "startup" }, makeCtx({ isIdle: () => false })) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + + recordedRequests.length = 0 + + // Block mid-turn. + pi.events.emit("herdr:blocked", { active: true, label: "Permission: write" }) + // Unblock back to working. + pi.events.emit("herdr:blocked", { active: false }) + + await reporter.drain() + + const stateSequence = recordedRequests.filter((r) => r.method === "pane.report_agent").map((r) => r.params.state) + expect(stateSequence).toEqual(["blocked", "working"]) + }) +}) + +// --------------------------------------------------------------------------- +// process-level beforeExit listener regression +// --------------------------------------------------------------------------- +// +// Each extension instance registers its `releaseReporter` callback with the +// module-level `beforeExitReleasers` Set. On `release()` (or +// `session_shutdown`) the instance must remove itself so repeated +// instantiations cannot accumulate process-level `beforeExit` listeners and +// trip Node's MaxListenersExceededWarning. + +describe("process-level beforeExit listener", () => { + beforeEach(() => { + process.env.HERDR_ENV = "1" + process.env.HERDR_SOCKET_PATH = "/tmp/herdr.sock" + process.env.HERDR_PANE_ID = "pane-regression" + }) + + it("does not leak process listeners or registry entries when multiple instances are created and released", async () => { + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}) + + try { + const initialListenerCount = process.listenerCount("beforeExit") + const initialRegistrySize = beforeExitReleasers.size + + const pi = makeFakePi() + herdrReporterExtension(pi.api) + + // Anchor a root session so the extension wires up its release + // callback against the module-level registry. + const sessionStart = getHandler(pi, "session_start") + await sessionStart({ reason: "startup" }, makeCtx({ isIdle: () => true })) + + const reporter = pi.api.herdrReporter as ReturnType + await reporter.drain() + + // One extension instance is registered. The module-level + // listener itself was registered once at import time; only + // instance callbacks should grow. + expect(beforeExitReleasers.size).toBe(initialRegistrySize + 1) + + // Release the first instance; the registry and process + // listener count must be restored. + const shutdown = getHandler(pi, "session_shutdown") + shutdown({}, makeCtx()) + await flushMicrotasks() + await reporter.drain() + + expect(beforeExitReleasers.size).toBe(initialRegistrySize) + expect(process.listenerCount("beforeExit")).toBe(initialListenerCount) + + // Create several more instances and release each one. The + // listener count must return to the original each time and + // the registry must never grow above the baseline. + const iterations = 5 + for (let i = 0; i < iterations; i++) { + const baselineForIteration = new Set(beforeExitReleasers) + const piN = makeFakePi() + herdrReporterExtension(piN.api) + await getHandler(piN, "session_start")({ reason: "startup" }, makeCtx({ isIdle: () => true })) + const reporterN = piN.api.herdrReporter as ReturnType + await reporterN.drain() + + expect(beforeExitReleasers.size).toBe(initialRegistrySize + 1) + + // Capture only the freshly-added release callback for + // this iteration. Prior tests in this file may have left + // their own entries in the module-level set; we filter to + // the delta so we can assert this instance's entry was + // specifically removed. + const newlyAdded = Array.from(beforeExitReleasers).filter((fn) => !baselineForIteration.has(fn)) + expect(newlyAdded).toHaveLength(1) + + getHandler(piN, "session_shutdown")({}, makeCtx()) + await flushMicrotasks() + await reporterN.drain() + + expect(beforeExitReleasers.size).toBe(initialRegistrySize) + expect(process.listenerCount("beforeExit")).toBe(initialListenerCount) + + // The specific callback this iteration registered must + // no longer be in the set after release. + for (const fn of newlyAdded) { + expect(beforeExitReleasers.has(fn)).toBe(false) + } + } + + // Final guardrail: no warning was emitted. + expect(warnSpy).not.toHaveBeenCalled() + } finally { + warnSpy.mockRestore() + } + }) + + it("beforeExit awaits every registered release promise and clears the registry", async () => { + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}) + + try { + // Snapshot the registry so we can identify the entries THIS + // test contributes. Earlier tests in this file may leave + // dangling extension instances whose release callbacks are + // still in the registry; that's fine — they're orthogonal to + // what we want to assert here. + const baseline = new Set(beforeExitReleasers) + + const piA = makeFakePi() + herdrReporterExtension(piA.api) + await getHandler(piA, "session_start")({ reason: "startup" }, makeCtx({ isIdle: () => false })) + const reporterA = piA.api.herdrReporter as ReturnType + const releaseASpy = vi.spyOn(reporterA, "release") + await reporterA.drain() + + const piB = makeFakePi() + herdrReporterExtension(piB.api) + await getHandler(piB, "session_start")({ reason: "startup" }, makeCtx({ isIdle: () => false })) + const reporterB = piB.api.herdrReporter as ReturnType + const releaseBSpy = vi.spyOn(reporterB, "release") + await reporterB.drain() + + const newlyAdded = Array.from(beforeExitReleasers).filter((fn) => !baseline.has(fn)) + expect(newlyAdded).toHaveLength(2) + + // Find the module-level async listener by inspecting its + // source for the registry it iterates. We do not invoke it + // for detection (invoking fires a real drain). + const asyncListener = process + .listeners("beforeExit") + .find((fn) => fn.toString().includes("beforeExitReleasers")) as (() => Promise) | undefined + expect(asyncListener).toBeDefined() + + // Core invariant: the listener is async — it returns a Promise + // that we can await. This is what keeps the event loop alive + // long enough for the final idle reports to flush before the + // process exits. + const result = asyncListener?.() + expect(result).toBeInstanceOf(Promise) + + await result + + // After the await, both reporters' release() must have been + // called AND completed (because `releaseReporter` awaits + // `reporter.release()` before resolving). + expect(releaseASpy).toHaveBeenCalledTimes(1) + expect(releaseBSpy).toHaveBeenCalledTimes(1) + + // The release callbacks this test registered must have + // self-removed from the registry. + for (const fn of newlyAdded) { + expect(beforeExitReleasers.has(fn)).toBe(false) + } + } finally { + warnSpy.mockRestore() + } + }) + + it("beforeExit tolerates non-promise release callbacks (legacy sync releases)", async () => { + const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}) + + try { + const initialRegistrySize = beforeExitReleasers.size + + // Inject a sync callback into the registry to confirm the + // handler does not throw when a release does not return a + // promise. Also inject a deliberately-throwing one to confirm + // it does not break sibling releases. + const syncRelease = vi.fn(() => { + // returns undefined on purpose + }) + const throwingRelease = vi.fn(() => { + throw new Error("synthetic") + }) + + beforeExitReleasers.add(syncRelease) + beforeExitReleasers.add(throwingRelease) + expect(beforeExitReleasers.size).toBe(initialRegistrySize + 2) + + const asyncListener = process + .listeners("beforeExit") + .find((fn) => fn.toString().includes("beforeExitReleasers")) as (() => Promise) | undefined + expect(asyncListener).toBeDefined() + + const result = asyncListener?.() + expect(result).toBeInstanceOf(Promise) + + await expect(result).resolves.toBeUndefined() + + expect(syncRelease).toHaveBeenCalledTimes(1) + expect(throwingRelease).toHaveBeenCalledTimes(1) + + // We left these entries in the registry on purpose — clean up + // so subsequent tests are unaffected. + beforeExitReleasers.delete(syncRelease) + beforeExitReleasers.delete(throwingRelease) + expect(beforeExitReleasers.size).toBe(initialRegistrySize) + } finally { + warnSpy.mockRestore() + } + }) +}) diff --git a/src/extensions/herdr-reporter.ts b/src/extensions/herdr-reporter.ts new file mode 100644 index 000000000..9f1a11f8f --- /dev/null +++ b/src/extensions/herdr-reporter.ts @@ -0,0 +1,541 @@ +/** + * herdr-reporter — kimchi → herdr agent-state reporter. + * + * herdr is a desktop app that renders per-pane agent status. This extension + * is the kimchi-side reporter that talks to the herdr control socket and + * surfaces agent state (idle / working / blocked) and session metadata so + * herdr can render the pane correctly. + * + * The reporter is best-effort and intentionally non-throwing — herdr is an + * out-of-process UI dependency, and a transient failure (herdr not running, + * socket disconnected, slow handshake) must never wedge a kimchi session. + * All socket errors are swallowed and retried once with a longer timeout; + * after the second failure the report is dropped silently. + * + * Activation is driven entirely by environment variables: + * + * - HERDR_ENV=1 master switch + * - HERDR_SOCKET_PATH path to the herdr control socket (or pipe name) + * - HERDR_PANE_ID pane identifier reported with every event + * - HERDR_BIN_PATH optional path to the herdr binary (read by the + * installer / daemon, not by this extension) + * + * When HERDR_ENV is unset, the extension is a no-op: no socket is opened + * and no reporter is exposed on `pi`. The wiring step (separate) decides + * what to do with `pi.herdrReporter`. + * + * Transport: + * + * - Unix domain socket at HERDR_SOCKET_PATH on non-Windows. + * - Named pipe `\\.\pipe\` on Windows. + * + * Wire format: JSON-RPC 2.0 over newline-delimited frames (one JSON object + * per line, terminated by `\n`). We do not currently consume responses — + * herdr treats writes as fire-and-forget state updates — but the framing + * matches JSON-RPC so a future version could add bidirectional control + * without changing the transport. + * + * Sequence numbers start at `Date.now() * 1000` (a monotonic microsecond-ish + * counter that is also unique across the process) and increment by 1 per + * report. herdr uses these to detect dropped or out-of-order events. + */ + +import net from "node:net" +import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent" + +// --------------------------------------------------------------------------- +// Public types +// --------------------------------------------------------------------------- + +export type HerdrAgentState = "idle" | "working" | "blocked" + +export interface HerdrSessionRef { + id?: string + path?: string +} + +export interface HerdrReporterOptions { + paneId: string + socketPath: string + source: string + agent: string +} + +export interface HerdrReporter { + reportState(state: HerdrAgentState, message?: string): void + reportSession(sessionRef: HerdrSessionRef, sessionStartSource?: string): void + /** Update the reporter's tracked session ref without emitting a report. */ + updateSessionRef(sessionRef: HerdrSessionRef): void + /** Stop accepting new reports and wait for the queue to drain. */ + release(): Promise + /** Wait for all pending reports to be sent (or dropped). */ + drain(): Promise +} + +interface HerdrRequest { + jsonrpc: "2.0" + id: string + method: string + params: Record +} + +// --------------------------------------------------------------------------- +// Tunables +// --------------------------------------------------------------------------- + +/** First-attempt timeout (ms). Short — we expect the herdr socket to be local. */ +const INITIAL_TIMEOUT_MS = 500 +/** Retry-attempt timeout (ms). Longer — gives herdr a chance to come back. */ +const RETRY_TIMEOUT_MS = 1500 + +// --------------------------------------------------------------------------- +// Process-level beforeExit backstop (singleton) +// --------------------------------------------------------------------------- +// +// A module-level beforeExit listener is registered exactly once per process. +// Each extension instance adds its `release` callback to `beforeExitReleasers` +// and removes it on shutdown. This avoids the per-instance listener +// accumulation that would otherwise trip Node's +// MaxListenersExceededWarning when the extension is loaded repeatedly +// (tests, extension reloads, multi-session hosts). + +// Exported so the regression test in `herdr-reporter.test.ts` can assert +// that each instance unregisters its release callback and the registry is +// restored to its prior size. The set itself is otherwise untouched. +export const beforeExitReleasers = new Set<() => void>() + +if (typeof process !== "undefined" && typeof process.on === "function") { + process.on("beforeExit", async () => { + // Snapshot to allow re-entry (a release callback must not mutate + // the set while we iterate). Collect every returned promise so + // `await Promise.all` keeps the event loop alive until all + // reporters have drained — otherwise the process can exit before + // the final state report has been written to herdr's socket. + const promises: Array> = [] + for (const release of Array.from(beforeExitReleasers)) { + try { + const result = release() as unknown + if (result && typeof (result as { then?: unknown }).then === "function") { + promises.push(result as Promise) + } + } catch { + // best-effort — a throwing release must not block siblings + } + } + await Promise.all(promises) + }) +} + +// --------------------------------------------------------------------------- +// Transport helpers +// --------------------------------------------------------------------------- + +function unrefTimer(t: NodeJS.Timeout): void { + // Some test/mock timers don't expose unref; ignore those. + const maybe = t as unknown as { unref?: () => void } + maybe.unref?.() +} + +function socketAddress(socketPath: string): string { + if (process.platform === "win32") { + // Named-pipe path — escape backslashes for the JS string literal so the + // runtime sees `\\.\pipe\`. + return `\\\\.\\pipe\\${socketPath}` + } + return socketPath +} + +/** + * Attempt a single send with the given timeout. Resolves once + * `socket.end(data, callback)` fires — that callback runs when the data + * has been flushed to the OS and FIN has been queued, which is the + * earliest point we can consider the write "done". The peer's half of + * the socket may stay open longer than our short timeouts, so we do + * not wait for a remote `close` event — that was the source of + * spurious "herdr send timeout" failures. Rejects on socket error or + * timeout. Never throws synchronously. + */ +function sendOnce(request: HerdrRequest, socketPath: string, timeoutMs: number): Promise { + return new Promise((resolve, reject) => { + const address = socketAddress(socketPath) + const socket = net.createConnection(address) + let settled = false + + const finish = (err?: Error) => { + if (settled) return + settled = true + clearTimeout(timer) + // Destroy the socket so its handle is released even if the + // remote has not yet closed its half. Idempotent and safe to + // call after `end` has flushed. + try { + socket.destroy() + } catch { + // best-effort + } + if (err) reject(err) + else resolve() + } + + const timer = setTimeout(() => finish(new Error("herdr send timeout")), timeoutMs) + unrefTimer(timer) + + socket.once("error", (err) => finish(err)) + socket.once("connect", () => { + try { + // `end(data, callback)` is equivalent to `write(data)` then + // `end()`, with the callback registered as a one-time + // `'finish'` listener. Resolving on `'finish'` is what gives + // us "data flushed / FIN sent" semantics without waiting for + // the far end to close. + socket.end(`${JSON.stringify(request)}\n`, () => { + finish() + }) + } catch (err) { + finish(err as Error) + } + }) + }) +} + +/** + * Send a request, retrying once with a longer timeout on failure. The + * second failure is logged (kimchi is best-effort but the user may want + * to know herdr stopped responding) and then swallowed — callers (the + * queue) must not surface errors. + */ +async function sendWithRetry(request: HerdrRequest, socketPath: string): Promise { + try { + await sendOnce(request, socketPath, INITIAL_TIMEOUT_MS) + return + } catch { + // first attempt failed — fall through to retry + } + try { + await sendOnce(request, socketPath, RETRY_TIMEOUT_MS) + } catch (err) { + // Final failure after retry — herdr socket unreachable. Log so the + // operator notices in long-lived sessions, then swallow. + console.warn( + `herdr reporter: dropping report (method=${request.method}) after retry — socket at ${socketPath} unreachable: ${ + err instanceof Error ? err.message : String(err) + }`, + ) + } +} + +// --------------------------------------------------------------------------- +// Reporter +// --------------------------------------------------------------------------- + +export function createHerdrReporter(options: HerdrReporterOptions): HerdrReporter { + const { paneId, socketPath, source, agent } = options + + // Start at wall-clock-microseconds so the first id is unique within the + // process and roughly tracks session start. Monotonic thereafter. + let nextSeq = Date.now() * 1000 + let queue: Promise = Promise.resolve() + let released = false + + const buildRequest = (method: string, params: Record): HerdrRequest => { + const seq = nextSeq++ + // herdr requires the JSON-RPC id field to be a string. We keep a + // monotonic numeric counter for ordering and stringify it for the + // framing id. + const id = String(seq) + // herdr uses the per-method `seq` counter to detect dropped or + // out-of-order events; mirror the JSON-RPC id so consumers see a + // monotonic sequence number alongside the framing id. + return { jsonrpc: "2.0", id, method, params: { ...params, seq } } + } + + // Tracked session ref shared across reports. Updated by + // `updateSessionRef`; stamped onto every state report via + // `withSessionRef` so herdr can correlate agent state with the + // session that produced it. + let currentSessionRef: HerdrSessionRef = {} + + function withSessionRef(params: Record): void { + // Prefer path when available, otherwise fall back to id — this + // matches herdr's preference for a stable on-disk handle. + if (currentSessionRef.path !== undefined) { + params.agent_session_path = currentSessionRef.path + } else if (currentSessionRef.id !== undefined) { + params.agent_session_id = currentSessionRef.id + } + } + + const enqueue = (request: HerdrRequest): void => { + // Chain onto the existing tail so reports are processed strictly in + // submission order with a single in-flight writer. + queue = queue + .then(() => sendWithRetry(request, socketPath)) + .catch(() => { + // Belt-and-braces — sendWithRetry already swallows, but the + // queue tail must never reject or every later enqueue would + // short-circuit. + }) + } + + return { + reportState(state, message) { + if (released) return + const params: Record = { + pane_id: paneId, + source, + agent, + state, + } + if (message !== undefined) params.message = message + withSessionRef(params) + enqueue(buildRequest("pane.report_agent", params)) + }, + + reportSession(sessionRef, sessionStartSource) { + if (released) return + const params: Record = { + pane_id: paneId, + source, + agent, + } + // Prefer path when available, otherwise fall back to id. + if (sessionRef.path !== undefined) { + params.agent_session_path = sessionRef.path + } else if (sessionRef.id !== undefined) { + params.agent_session_id = sessionRef.id + } + if (sessionStartSource !== undefined) { + params.session_start_source = sessionStartSource + } + enqueue(buildRequest("pane.report_agent_session", params)) + }, + + updateSessionRef(sessionRef) { + currentSessionRef = { + id: sessionRef.id, + path: sessionRef.path, + } + }, + + async release() { + released = true + await this.drain() + }, + + async drain() { + await queue + }, + } +} + +// --------------------------------------------------------------------------- +// Extension entry point +// --------------------------------------------------------------------------- + +interface HerdrReporterExtensionApi extends ExtensionAPI { + herdrReporter?: HerdrReporter +} + +/** + * Read the herdr environment and return the parsed view. Exported for + * the wiring step and for tests that need to assert activation logic + * without touching `process.env` directly. + */ +export function readHerdrEnv(): { + enabled: boolean + socketPath?: string + paneId?: string + binPath?: string +} { + const env = process.env.HERDR_ENV + const socketPath = process.env.HERDR_SOCKET_PATH + const paneId = process.env.HERDR_PANE_ID + const binPath = process.env.HERDR_BIN_PATH + + const enabled = env === "1" && Boolean(socketPath) && Boolean(paneId) + return { enabled, socketPath, paneId, binPath } +} + +export default function herdrReporterExtension(pi: ExtensionAPI): void { + const view = readHerdrEnv() + if (!view.enabled || !view.socketPath || !view.paneId) return + + const reporter = createHerdrReporter({ + paneId: view.paneId, + socketPath: view.socketPath, + source: "herdr:kimchi", + agent: "kimchi", + }) + + // Expose the reporter on `pi` so external consumers (e.g. tests, the + // daemon bridge) can introspect or send custom reports without having + // to re-read the environment or re-create the reporter. + const piWithReporter = pi as HerdrReporterExtensionApi + piWithReporter.herdrReporter = reporter + + // ------------------------------------------------------------------- + // State machine + // ------------------------------------------------------------------- + // + // We track three orthogonal bits of state and collapse them to one + // of three herdr-facing states: + // + // - blocked when at least one herdr:blocked activation is active + // - working when the agent loop is currently running + // - idle when nothing is happening + // + // blocked > working > idle — a prompt that opens while the agent is + // mid-turn still surfaces as blocked, because that's the state the + // user actually cares about. + // + // We only drive this state machine for the root TUI session. RPC and + // JSON modes may also emit session_start, but those runtimes drive + // their own UX and shouldn't pin a herdr pane. + let agentActive = false + let blockedCount = 0 + let blockedMessage: string | undefined + let lastState: HerdrAgentState | undefined + let lastMessage: string | undefined + let rootSession = false + let released = false + + function updateSessionRef(ctx: ExtensionContext | undefined): HerdrSessionRef { + const ref: HerdrSessionRef = {} + try { + const file = ctx?.sessionManager?.getSessionFile?.() + if (typeof file === "string" && file.length > 0) { + ref.path = file + } + } catch { + // best-effort — leave path unset + } + try { + const id = ctx?.sessionManager?.getSessionId?.() + if (typeof id === "string" && id.length > 0) { + ref.id = id + } + } catch { + // best-effort — leave id unset + } + // Sync the reporter's tracked ref so subsequent state reports pick + // up the latest session handle via `withSessionRef`. + reporter.updateSessionRef(ref) + return ref + } + + function desiredState(): { state: HerdrAgentState; message?: string } { + if (blockedCount > 0) { + return { state: "blocked", message: blockedMessage } + } + if (agentActive) { + return { state: "working" } + } + return { state: "idle" } + } + + function publishState(force = false): void { + const next = desiredState() + if (!force && next.state === lastState && next.message === lastMessage) { + return + } + lastState = next.state + lastMessage = next.message + reporter.reportState(next.state, next.message) + } + + function resetForRootSession(ctx: ExtensionContext, sessionStartSource?: string): void { + rootSession = true + const ref = updateSessionRef(ctx) + reporter.reportSession(ref, sessionStartSource) + agentActive = ctx?.isIdle?.() === false + publishState() + } + + // The very first session_start in TUI mode anchors the root session. + // `event.reason` carries the source (e.g. "startup", "resume") that + // herdr uses to render session badges. + pi.on("session_start", async (event, ctx) => { + if (ctx?.mode !== "tui") return + const reason = (event as { reason?: unknown } | undefined)?.reason + const sessionStartSource = typeof reason === "string" ? reason : undefined + resetForRootSession(ctx, sessionStartSource) + }) + + pi.on("agent_start", (_event, ctx) => { + if (!rootSession) return + const ref = updateSessionRef(ctx) + reporter.reportSession(ref) + agentActive = true + publishState() + }) + + pi.on("agent_settled", (_event, ctx) => { + if (!rootSession) return + if (ctx?.isIdle?.() === true) { + agentActive = false + publishState() + } + }) + + // Subscriptions to herdr:blocked refcount nested activations across + // every prompt surface (permissions, questionnaires, manual confirm). + // The bus emits activations BEFORE the prompt and deactivations in a + // finally — paired counts are the source of truth for "blocked". + pi.events.on("herdr:blocked", (data) => { + if (!rootSession) return + const payload = data as { active?: unknown; label?: unknown } | undefined + const active = payload?.active + const label = typeof payload?.label === "string" ? payload.label : undefined + + if (active === true) { + blockedCount += 1 + if (label !== undefined) blockedMessage = label + publishState() + return + } + + if (active === false) { + blockedCount = Math.max(0, blockedCount - 1) + if (blockedCount === 0) blockedMessage = undefined + publishState() + } + }) + + // Release lifecycle authority on session shutdown so the herdr pane + // can hand off to another integrator. beforeExit is a backstop for + // abnormal exits that bypass session_shutdown. + // + // Returns the promise from `reporter.release()` so callers + // (session_shutdown, the beforeExit backstop) can await the drain. + // The returned promise is also collected by the module-level + // `beforeExit` listener to keep the event loop alive until the + // final state report has been written. + const releaseReporter = (): Promise => { + if (released) return Promise.resolve() + // Force a final idle state report BEFORE marking released so the + // reporter still accepts the enqueue. Without this, herdr would + // keep whatever the last `working` / `blocked` state was — the + // pane would appear stuck after kimchi exits. + agentActive = false + blockedCount = 0 + blockedMessage = undefined + publishState(true) + released = true + // De-register from the process-level backstop so repeated + // extension instances (e.g. test reloads) do not accumulate + // beforeExit listeners and trip Node's + // MaxListenersExceededWarning. + beforeExitReleasers.delete(releaseReporter) + return reporter.release() + } + + // Register this instance's release with the process-level backstop. + beforeExitReleasers.add(releaseReporter) + + pi.on("session_shutdown", () => { + // Fire-and-forget — the module-level `beforeExit` handler awaits + // the returned promise so abnormal exits still drain the final + // report. + void releaseReporter() + }) +}