diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 2087e7ffe..02c7214d5 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,7 @@ ## [Unreleased] +- Added built-in Herdr integration that reports agent lifecycle state to Herdr panes automatically, without requiring `herdr integration install pi`. - Changed Escape to interrupt active work with a visible abort notice, double Escape to open the session tree from an empty prompt or clear an idle draft, and `?` to show shortcuts ([ENG-4489](https://linear.app/primeintellect/issue/ENG-4489/rewire-prime-agent-shortcuts-to-match-claude-code-flow)). - Changed new-chat guidance to show concise shell, command, file, and shortcut hints, with Agents View first and `? for shortcuts` after the model and effort ([ENG-4489](https://linear.app/primeintellect/issue/ENG-4489/rewire-prime-agent-shortcuts-to-match-claude-code-flow)). - Changed `?` shortcut help to appear as a temporary compact panel below the transcript, while `/hotkeys` shows the full reference without Ctrl+Z ([ENG-4489](https://linear.app/primeintellect/issue/ENG-4489/rewire-prime-agent-shortcuts-to-match-claude-code-flow)). diff --git a/packages/coding-agent/src/core/agent-session-services.ts b/packages/coding-agent/src/core/agent-session-services.ts index ae053107e..fc0d28b44 100644 --- a/packages/coding-agent/src/core/agent-session-services.ts +++ b/packages/coding-agent/src/core/agent-session-services.ts @@ -7,6 +7,7 @@ import type { AgentObserveController } from "./agent-observe.js"; import { installAgentTraceUpload } from "./agent-traces.js"; import { AuthStorage } from "./auth-storage.js"; import type { AgentRlmHeartbeatController } from "./cron-jobs.js"; +import { createHerdrAgentStateExtension } from "./extensions/builtin/herdr-agent-state.js"; import type { SessionStartEvent, ToolDefinition } from "./extensions/index.js"; import { McpManager } from "./mcp/mcp-manager.js"; import { ModelRegistry } from "./model-registry.js"; @@ -43,6 +44,13 @@ export interface CreateAgentSessionServicesOptions { modelRegistry?: ModelRegistry; extensionFlagValues?: Map; resourceLoaderOptions?: Omit; + /** + * Skip the built-in Herdr reporter for these services. Set for RLM subagent + * runtimes: they inherit the parent's HERDR_* pane identity, so their own + * reporter would race the parent's on the same pane and a subagent quit + * would release the pane while the parent is still running. + */ + noBuiltinHerdrReporter?: boolean; } export interface AgentSessionCreationOptions { @@ -167,8 +175,21 @@ export async function createAgentSessionServices( // refresh() resets the OAuth registry to built-ins; re-add user MCP providers too. modelRegistry.setOnOAuthProvidersReset(() => mcpManager.registerUserProviders()); - const resourceLoader = new DefaultResourceLoader({ + const userExtensionFactories = options.resourceLoaderOptions?.extensionFactories ?? []; + // The built-in Herdr reporter defers to Herdr's own file-based integration + // when the loader actually loaded it; two reporters would race on the same + // pane. Deferral is late-bound to the loader's loaded paths (inline + // factories run after file extensions load), so a file that exists but is + // disabled or never discovered does not silence the built-in. + // noExtensions is a full opt-out: it disables the built-in reporter too, + // not just discovered extension files. + const skipHerdrReporter = options.noBuiltinHerdrReporter || options.resourceLoaderOptions?.noExtensions; + const builtinExtensionFactories = skipHerdrReporter + ? [] + : [createHerdrAgentStateExtension(() => resourceLoader.getLoadedExtensionPaths())]; + const resourceLoader: DefaultResourceLoader = new DefaultResourceLoader({ ...(options.resourceLoaderOptions ?? {}), + extensionFactories: [...builtinExtensionFactories, ...userExtensionFactories], cwd, agentDir, settingsManager, diff --git a/packages/coding-agent/src/core/extensions/builtin/herdr-agent-state.ts b/packages/coding-agent/src/core/extensions/builtin/herdr-agent-state.ts new file mode 100644 index 000000000..38b6e1efd --- /dev/null +++ b/packages/coding-agent/src/core/extensions/builtin/herdr-agent-state.ts @@ -0,0 +1,461 @@ +/** + * Built-in Herdr integration extension. + * + * Reports agent lifecycle state (working/idle/blocked) to the Herdr terminal + * workspace manager via its Unix socket. This is the in-tree equivalent of + * the extension that `herdr integration install pi` writes, so Prime Agent + * works inside Herdr panes out of the box without a manual install step. + * + * Unlike the file-based integration (re-evaluated per session load by jiti), + * this module is statically imported and evaluated once per process. All env + * capture and state therefore live inside the factory, which the resource + * loader invokes per session load — inside the daemon's client-env window — + * so each daemon session captures its own pane identity. + * + * The factory is a complete no-op when `HERDR_ENV` is not `"1"` (i.e. when + * not running inside a Herdr pane), so it is safe to always load. + */ + +import { createConnection } from "node:net"; +import { basename } from "node:path"; +import type { ExtensionAPI, ExtensionFactory } from "../types.js"; + +type AgentState = "working" | "blocked" | "idle"; + +/** + * True when Herdr's own file-based Pi integration (`herdr integration + * install pi`) is among the extension files the loader actually loaded this + * cycle. That extension reports with the same `herdr:pi` source but its own + * seq counter, so running the built-in alongside it would make the two + * reporters race on one pane. + * + * Loaded paths — not raw disk existence — are the deferral source of truth: + * a file that exists but never loads (settings `!` overrides, noExtensions, + * paths outside the discovery dirs such as the legacy `~/.pi/agent/`) never + * becomes an active reporter, and deferring to it would leave the pane with + * no reporter at all. + */ +export function hasFileBasedHerdrIntegration(loadedExtensionPaths: string[]): boolean { + return loadedExtensionPaths.some((path) => { + const base = basename(path); + return base === "herdr-agent-state.ts" || base === "herdr-agent-state.js"; + }); +} + +interface QueuedState { + state: AgentState; + message?: string; + seq: number; +} + +function parseDurationEnv(name: string, fallback: number): number { + const raw = process.env[name]; + if (!raw) { + return fallback; + } + const parsed = Number.parseInt(raw, 10); + if (!Number.isFinite(parsed) || parsed < 0) { + return fallback; + } + return parsed; +} + +function lastAssistantMessage(messages: unknown[]): any | undefined { + for (let i = messages.length - 1; i >= 0; i -= 1) { + const message = messages[i] as any; + if (message?.role === "assistant") { + return message; + } + } + return undefined; +} + +/** + * Error message of the turn's final assistant message, if it ended in error. + * + * The agent's auto-retry treats nearly every provider error as retryable and + * kicks in after extension agent_end fires, so any error end may be followed + * by a retry. Rather than second-guessing the agent's classification with a + * pattern list, hold "working" for every error through the retry grace + * window: if a retry starts, agent_start keeps the pane working; if none + * does, the hold settles to blocked with the error message. + */ +function errorHoldMessage(event: any): string | undefined { + const messages = Array.isArray(event?.messages) ? event.messages : []; + const assistant = lastAssistantMessage(messages); + if (assistant?.stopReason !== "error") { + return undefined; + } + return String(assistant.errorMessage ?? "") || "provider error"; +} + +// Monotonic across all extension instances in this process. Herdr guards +// pane.report_agent with a per-source seq and silently drops lower-seq +// reports, so a successor session instance (after /new, resume, fork, or +// reload) must never restart below a seq the previous instance already used — +// otherwise its idle reports are dropped and the pane sticks at "working". +let reportSeq = Date.now() * 1000; + +function nextReportSeq(): number { + reportSeq = Math.max(reportSeq + 1, Date.now() * 1000); + return reportSeq; +} + +/** + * Build the built-in Herdr reporter factory. `getLoadedExtensionPaths` + * returns the extension files the resource loader actually loaded in the + * current cycle; it is re-checked on every factory invocation (i.e. on every + * session load and `/reload`), so installing Herdr's own file-based + * integration and reloading hands the pane over to it without also keeping + * the built-in active — while a file that exists but never loads (settings + * overrides, legacy paths) does not silence the built-in. + */ +export function createHerdrAgentStateExtension(getLoadedExtensionPaths: () => string[]): ExtensionFactory { + return (pi: ExtensionAPI) => { + herdrAgentStateExtensionImpl(pi, getLoadedExtensionPaths); + }; +} + +/** Built-in reporter with no file-based deferral, for tests and embedders. */ +export const herdrAgentStateExtension: ExtensionFactory = (pi: ExtensionAPI) => { + herdrAgentStateExtensionImpl(pi, () => []); +}; + +function herdrAgentStateExtensionImpl(pi: ExtensionAPI, getLoadedExtensionPaths: () => string[]): void { + // Captured per factory invocation: the resource loader runs this during + // session load, inside the daemon's client-env window, so these reflect the + // session's own Herdr pane rather than the daemon's startup environment. + const socketPath = process.env.HERDR_SOCKET_PATH; + const paneId = process.env.HERDR_PANE_ID; + const enabled = process.env.HERDR_ENV === "1" && !!socketPath && !!paneId; + if (!enabled || hasFileBasedHerdrIntegration(getLoadedExtensionPaths())) { + return; + } + + const source = "herdr:pi"; + const agentLabel = "prime-agent"; + const idleDebounceMs = parseDurationEnv("HERDR_PI_IDLE_DEBOUNCE_MS", 250); + const retryGraceMs = parseDurationEnv("HERDR_PI_RETRY_GRACE_MS", 2500); + + let currentAgentSessionId: string | undefined; + let currentAgentSessionPath: string | undefined; + // Inline RLM child sessions reuse the parent's resource loader, so their + // extension runners re-bind these same handler closures. Bind the reporter + // to the first session that starts (the parent) and ignore events other + // sessions deliver, or a subagent's turns would flip the pane state and + // stomp the parent's session reference. + let boundSessionManager: unknown; + let sendInFlight = false; + let queuedState: QueuedState | undefined; + let activeDrain: Promise = Promise.resolve(); + let released = false; + + let agentActive = false; + let retryHoldActive = false; + let failureBlocked = false; + let failureMessage: string | undefined; + let blockedCount = 0; + let blockedMessage: string | undefined; + let lastState: AgentState | undefined; + let lastMessage: string | undefined; + let idleTimer: ReturnType | undefined; + let retryTimer: ReturnType | undefined; + + function sendRequest(request: unknown): Promise { + return new Promise((resolve) => { + let done = false; + const finish = () => { + if (done) return; + done = true; + socket.destroy(); + resolve(); + }; + + const socket = createConnection(socketPath!); + socket.on("error", finish); + socket.on("connect", () => socket.write(`${JSON.stringify(request)}\n`)); + socket.on("data", finish); + socket.on("end", finish); + const timeout = setTimeout(finish, 500); + timeout.unref?.(); + }); + } + + function isBoundSession(ctx: any): boolean { + if (boundSessionManager === undefined) { + return true; + } + return ctx?.sessionManager === boundSessionManager; + } + + function updateSessionRef(ctx: any): void { + try { + const file = ctx?.sessionManager?.getSessionFile?.(); + currentAgentSessionPath = typeof file === "string" && file.startsWith("/") ? file : undefined; + } catch { + currentAgentSessionPath = undefined; + } + + try { + const id = ctx?.sessionManager?.getSessionId?.(); + currentAgentSessionId = typeof id === "string" && id.length > 0 ? id : undefined; + } catch { + currentAgentSessionId = undefined; + } + } + + function withSessionRef(params: Record): Record { + if (currentAgentSessionPath) { + return { ...params, agent_session_path: currentAgentSessionPath }; + } + if (currentAgentSessionId) { + return { ...params, agent_session_id: currentAgentSessionId }; + } + return params; + } + + function sendState(state: AgentState, message?: string, seq = nextReportSeq()): Promise { + return sendRequest({ + id: `${source}:${Date.now()}:${Math.random().toString(36).slice(2)}`, + method: "pane.report_agent", + params: withSessionRef({ + pane_id: paneId, + source, + agent: agentLabel, + state, + message, + seq, + }), + }); + } + + function queueState(state: AgentState, message?: string): void { + if (released) { + // The pane was released on quit; a late report would reclaim it and + // leave Herdr showing an agent that already exited. + return; + } + queuedState = { state, message, seq: nextReportSeq() }; + if (!sendInFlight) { + activeDrain = drainStateQueue(); + } + } + + async function drainStateQueue(): Promise { + if (sendInFlight) { + return; + } + + sendInFlight = true; + try { + while (queuedState) { + const next = queuedState; + queuedState = undefined; + await sendState(next.state, next.message, next.seq); + } + } finally { + sendInFlight = false; + if (queuedState) { + activeDrain = drainStateQueue(); + } + } + } + + async function releaseAgent(): Promise { + // Stop new reports, drop anything still queued, and wait for the + // in-flight send to finish so the release is the last write on the wire; + // a report landing after the release would reclaim the pane. + released = true; + queuedState = undefined; + await activeDrain.catch(() => undefined); + return sendRequest({ + id: `${source}:release:${Date.now()}:${Math.random().toString(36).slice(2)}`, + method: "pane.release_agent", + params: { + pane_id: paneId, + source, + agent: agentLabel, + seq: nextReportSeq(), + }, + }); + } + + function clearTimer(timer: ReturnType | undefined) { + if (timer) { + clearTimeout(timer); + } + } + + function clearPendingTimers() { + clearTimer(idleTimer); + clearTimer(retryTimer); + idleTimer = undefined; + retryTimer = undefined; + } + + function clearFailureState() { + retryHoldActive = false; + failureBlocked = false; + failureMessage = undefined; + } + + function desiredState(): { state: AgentState; message?: string } { + if (blockedCount > 0) { + return { state: "blocked", message: blockedMessage }; + } + if (failureBlocked) { + return { state: "blocked", message: failureMessage }; + } + if (agentActive || retryHoldActive) { + return { state: "working", message: undefined }; + } + return { state: "idle", message: undefined }; + } + + function publishState(force = false) { + const next = desiredState(); + if (!force && next.state === lastState && next.message === lastMessage) { + return; + } + lastState = next.state; + lastMessage = next.message; + queueState(next.state, next.message); + } + + function scheduleIdle() { + clearPendingTimers(); + clearFailureState(); + idleTimer = setTimeout(() => { + idleTimer = undefined; + publishState(); + }, idleDebounceMs); + idleTimer.unref?.(); + } + + function holdForRetry(message: string) { + clearPendingTimers(); + retryHoldActive = true; + failureBlocked = false; + failureMessage = message; + publishState(); + + retryTimer = setTimeout(() => { + retryTimer = undefined; + retryHoldActive = false; + failureBlocked = true; + publishState(); + }, retryGraceMs); + retryTimer.unref?.(); + } + + pi.on("session_start", (_event, ctx) => { + if (!isBoundSession(ctx)) { + return; + } + if (boundSessionManager === undefined && ctx?.sessionManager !== undefined) { + boundSessionManager = ctx.sessionManager; + } + updateSessionRef(ctx); + // A reload can re-create this reporter mid-turn (daemon-driven reloads + // and extension ctx.reload() are not gated on idle). Seed the active + // flag from the session so the fresh instance does not report idle + // while the agent is still streaming, which would also make the guard + // in agent_end swallow the turn's real end transition. + if (typeof ctx?.isIdle === "function") { + try { + agentActive = !ctx.isIdle(); + } catch { + agentActive = false; + } + } + publishState(true); + }); + + const unsubscribeBlocked = pi.events.on("herdr:blocked", (data: any) => { + if (!data?.active) { + blockedCount = Math.max(0, blockedCount - 1); + if (blockedCount === 0) { + blockedMessage = undefined; + } + publishState(); + return; + } + + clearPendingTimers(); + // clearPendingTimers cancelled the retry timer; settle the hold the way + // the timer would have, or retryHoldActive keeps desiredState() at + // "working" forever once the block lifts. + if (retryHoldActive) { + retryHoldActive = false; + failureBlocked = true; + } + blockedCount += 1; + blockedMessage = data.label; + publishState(); + }); + + pi.on("agent_start", (_event, ctx) => { + if (!isBoundSession(ctx)) { + return; + } + clearPendingTimers(); + clearFailureState(); + agentActive = true; + publishState(); + }); + + pi.on("agent_end", (event, ctx) => { + if (!isBoundSession(ctx)) { + return; + } + if (!agentActive) { + // Duplicate/late end events can arrive while auto-retry is already + // holding the pane in Working. Do not let an unqualified duplicate end + // cancel the retry hold and publish a false Idle. + return; + } + + agentActive = false; + + const holdMessage = errorHoldMessage(event); + if (holdMessage) { + holdForRetry(holdMessage); + return; + } + + // Queued follow-up/steer messages start another loop right away; debounce + // so the pane does not flicker done -> working. With nothing queued, + // report idle immediately so Herdr flips to done as streaming finishes. + if (typeof ctx?.hasPendingMessages === "function" && ctx.hasPendingMessages()) { + scheduleIdle(); + return; + } + + clearPendingTimers(); + clearFailureState(); + publishState(); + }); + + pi.on("session_shutdown", async (event, ctx) => { + if (!isBoundSession(ctx)) { + return; + } + clearPendingTimers(); + // The event bus is shared across reloads and session replacements, so a + // listener left behind would keep this stale instance reporting with a + // captured (possibly wrong) pane identity forever. + unsubscribeBlocked(); + // On session replacement (new/resume/fork) or reload, a successor + // instance in this same pane re-reports immediately. Releasing here + // races that report: two independent socket writes with no ordering, + // and a release that lands after the successor's report clears the + // pane. Only a real quit should release; every shutdown silences this + // instance so no stale queued report lands around the successor's. + if (event?.reason !== "quit") { + released = true; + queuedState = undefined; + return; + } + await releaseAgent(); + }); +} diff --git a/packages/coding-agent/src/core/extensions/index.ts b/packages/coding-agent/src/core/extensions/index.ts index 0a508b403..820b6b47d 100644 --- a/packages/coding-agent/src/core/extensions/index.ts +++ b/packages/coding-agent/src/core/extensions/index.ts @@ -4,6 +4,12 @@ export type { SlashCommandInfo, SlashCommandSource } from "../slash-commands.js"; export type { SourceInfo } from "../source-info.js"; +// Built-in extensions +export { + createHerdrAgentStateExtension, + hasFileBasedHerdrIntegration, + herdrAgentStateExtension, +} from "./builtin/herdr-agent-state.js"; export { createExtensionRuntime, discoverAndLoadExtensions, diff --git a/packages/coding-agent/src/core/resource-loader.ts b/packages/coding-agent/src/core/resource-loader.ts index b5f019b19..5cb03a4bf 100644 --- a/packages/coding-agent/src/core/resource-loader.ts +++ b/packages/coding-agent/src/core/resource-loader.ts @@ -193,6 +193,7 @@ export class DefaultResourceLoader implements ResourceLoader { private appendSystemPromptOverride?: (base: string[]) => string[]; private extensionsResult: LoadExtensionsResult; + private loadedExtensionPaths: string[] = []; private skills: Skill[]; private skillDiagnostics: ResourceDiagnostic[]; private prompts: PromptTemplate[]; @@ -263,6 +264,11 @@ export class DefaultResourceLoader implements ResourceLoader { return this.extensionsResult; } + /** Extension file paths the last reload actually loaded (after settings overrides). */ + getLoadedExtensionPaths(): string[] { + return this.loadedExtensionPaths; + } + getSkills(): { skills: Skill[]; diagnostics: ResourceDiagnostic[] } { return { skills: this.skills, diagnostics: this.skillDiagnostics }; } @@ -405,6 +411,10 @@ export class DefaultResourceLoader implements ResourceLoader { : this.mergePaths(cliEnabledExtensions, enabledExtensions); const extensionsResult = await loadExtensions(extensionPaths, this.cwd, this.eventBus); + // Set before inline factories run so a factory can see which file-based + // extensions actually loaded this cycle (e.g. the built-in Herdr reporter + // defers to Herdr's own file-based integration only when it is active). + this.loadedExtensionPaths = extensionPaths; const inlineExtensions = await this.loadExtensionFactories(extensionsResult.runtime); extensionsResult.extensions.push(...inlineExtensions.extensions); extensionsResult.errors.push(...inlineExtensions.errors); diff --git a/packages/coding-agent/src/main.ts b/packages/coding-agent/src/main.ts index 0783701aa..cfbb715f0 100644 --- a/packages/coding-agent/src/main.ts +++ b/packages/coding-agent/src/main.ts @@ -733,6 +733,9 @@ async function prepareRuntimeServices(options: { agentDir: effectiveAgentDir, authStorage, extensionFlagValues: new Map(Object.entries(config.extensionFlagValues ?? {})), + // Subagents share the parent's Herdr pane; their own reporter would race + // the parent's and a subagent quit would release the still-active pane. + noBuiltinHerdrReporter: (options.sessionOptionsOverride?.rlmDepth ?? 0) > 0, resourceLoaderOptions: { additionalExtensionPaths: config.extensions, additionalSkillPaths: config.skills, diff --git a/packages/coding-agent/test/herdr-agent-state.test.ts b/packages/coding-agent/test/herdr-agent-state.test.ts new file mode 100644 index 000000000..41a2423cb --- /dev/null +++ b/packages/coding-agent/test/herdr-agent-state.test.ts @@ -0,0 +1,513 @@ +import { mkdirSync, rmSync } from "node:fs"; +import { createServer, type Server } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { + createHerdrAgentStateExtension, + hasFileBasedHerdrIntegration, + herdrAgentStateExtension, +} from "../src/core/extensions/builtin/herdr-agent-state.js"; +import type { ExtensionAPI } from "../src/core/extensions/types.js"; + +interface RecordedRequest { + method: string; + params: Record; +} + +function createMockPi() { + const handlers = new Map unknown>>(); + const busHandlers = new Map unknown>>(); + const pi = { + on(event: string, handler: (...args: unknown[]) => unknown) { + const list = handlers.get(event) ?? []; + list.push(handler); + handlers.set(event, list); + }, + events: { + on(event: string, handler: (...args: unknown[]) => unknown) { + const list = busHandlers.get(event) ?? []; + list.push(handler); + busHandlers.set(event, list); + return () => { + const current = busHandlers.get(event) ?? []; + const index = current.indexOf(handler); + if (index !== -1) { + current.splice(index, 1); + } + }; + }, + }, + } as unknown as ExtensionAPI; + return { pi, handlers, busHandlers }; +} + +async function startFakeHerdrServer(socketPath: string): Promise<{ + server: Server; + requests: RecordedRequest[]; + waitForRequests: (count: number, timeoutMs?: number) => Promise; +}> { + const requests: RecordedRequest[] = []; + const waiters: Array<{ count: number; resolve: () => void }> = []; + + const server = createServer((socket) => { + let buffer = ""; + socket.on("data", (chunk) => { + buffer += chunk.toString(); + let newlineIndex = buffer.indexOf("\n"); + while (newlineIndex >= 0) { + const line = buffer.slice(0, newlineIndex); + buffer = buffer.slice(newlineIndex + 1); + if (line.trim()) { + const parsed = JSON.parse(line); + requests.push({ method: parsed.method, params: parsed.params }); + socket.write(`${JSON.stringify({ id: parsed.id, result: { type: "ok" } })}\n`); + for (const waiter of [...waiters]) { + if (requests.length >= waiter.count) { + waiters.splice(waiters.indexOf(waiter), 1); + waiter.resolve(); + } + } + } + newlineIndex = buffer.indexOf("\n"); + } + }); + }); + + await new Promise((resolve, reject) => { + server.on("error", reject); + server.listen(socketPath, resolve); + }); + + const waitForRequests = (count: number, timeoutMs = 3000): Promise => { + if (requests.length >= count) { + return Promise.resolve(); + } + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error(`timed out waiting for ${count} herdr requests`)), timeoutMs); + waiters.push({ + count, + resolve: () => { + clearTimeout(timer); + resolve(); + }, + }); + }); + }; + + return { server, requests, waitForRequests }; +} + +describe("herdrAgentStateExtension", () => { + const cleanupPaths: string[] = []; + const cleanupServers: Server[] = []; + const savedEnv: Record = {}; + const envKeys = [ + "HERDR_ENV", + "HERDR_SOCKET_PATH", + "HERDR_PANE_ID", + "HERDR_PI_IDLE_DEBOUNCE_MS", + "HERDR_PI_RETRY_GRACE_MS", + "PRIME_AGENT_CODING_AGENT_DIR", + ]; + + for (const key of envKeys) { + savedEnv[key] = process.env[key]; + } + + afterEach(async () => { + for (const key of envKeys) { + if (savedEnv[key] === undefined) { + delete process.env[key]; + } else { + process.env[key] = savedEnv[key]; + } + } + while (cleanupServers.length > 0) { + const server = cleanupServers.pop(); + await new Promise((resolve) => server?.close(() => resolve())); + } + while (cleanupPaths.length > 0) { + const path = cleanupPaths.pop(); + if (path) { + rmSync(path, { recursive: true, force: true }); + } + } + }); + + it("registers no handlers when HERDR_ENV is not set", () => { + delete process.env.HERDR_ENV; + delete process.env.HERDR_SOCKET_PATH; + delete process.env.HERDR_PANE_ID; + + const { pi, handlers, busHandlers } = createMockPi(); + herdrAgentStateExtension(pi); + + expect(handlers.size).toBe(0); + expect(busHandlers.size).toBe(0); + }); + + it("detects the file-based herdr integration among loaded extension paths", () => { + expect(hasFileBasedHerdrIntegration(["/x/extensions/herdr-agent-state.ts"])).toBe(true); + expect(hasFileBasedHerdrIntegration(["/x/extensions/herdr-agent-state.js"])).toBe(true); + expect(hasFileBasedHerdrIntegration(["/x/extensions/other.ts"])).toBe(false); + expect(hasFileBasedHerdrIntegration([])).toBe(false); + }); + + it("defers only to a file-based integration that actually loaded", () => { + const tempDir = join(tmpdir(), `pi-herdr-defer-${Date.now()}-${Math.random().toString(36).slice(2)}`); + mkdirSync(tempDir, { recursive: true }); + cleanupPaths.push(tempDir); + + process.env.HERDR_ENV = "1"; + process.env.HERDR_SOCKET_PATH = join(tempDir, "h.sock"); + process.env.HERDR_PANE_ID = "w1:p1"; + + // The loader reports the file-based integration as loaded: defer. + let loadedPaths = [join(tempDir, "extensions", "herdr-agent-state.ts")]; + const factory = createHerdrAgentStateExtension(() => loadedPaths); + const { pi, handlers, busHandlers } = createMockPi(); + factory(pi); + expect(handlers.size).toBe(0); + expect(busHandlers.size).toBe(0); + + // Loaded paths are consulted per invocation, so a /reload after the + // file-based integration is removed or disabled re-enables the built-in. + loadedPaths = []; + const second = createMockPi(); + factory(second.pi); + expect(second.handlers.size).toBeGreaterThan(0); + }); + + it("ignores events from sessions other than the one it bound to", async () => { + const tempDir = join(tmpdir(), `hrd-${Math.random().toString(36).slice(2, 8)}`); + mkdirSync(tempDir, { recursive: true }); + cleanupPaths.push(tempDir); + const socketPath = join(tempDir, "h.sock"); + + const { server, requests, waitForRequests } = await startFakeHerdrServer(socketPath); + cleanupServers.push(server); + + process.env.HERDR_ENV = "1"; + process.env.HERDR_SOCKET_PATH = socketPath; + process.env.HERDR_PANE_ID = "w1:p1"; + + const { pi, handlers } = createMockPi(); + herdrAgentStateExtension(pi); + + const parentSessionManager = { getSessionFile: () => "/tmp/parent.jsonl", getSessionId: () => "parent" }; + const childSessionManager = { getSessionFile: () => "/tmp/child.jsonl", getSessionId: () => "child" }; + const parentCtx = { sessionManager: parentSessionManager }; + const childCtx = { sessionManager: childSessionManager }; + + handlers.get("session_start")?.[0]?.({ type: "session_start", reason: "startup" }, parentCtx); + await waitForRequests(1); + expect(requests[0]?.params.agent_session_path).toBe("/tmp/parent.jsonl"); + + // Inline RLM children rebind the same handlers with their own ctx; their + // events must not flip the pane state or release the parent's pane. + handlers.get("agent_start")?.[0]?.({ type: "agent_start" }, childCtx); + await handlers.get("session_shutdown")?.[0]?.({ type: "session_shutdown", reason: "quit" }, childCtx); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(requests).toHaveLength(1); + + // The bound parent still reports normally. + handlers.get("agent_start")?.[0]?.({ type: "agent_start" }, parentCtx); + await waitForRequests(2); + expect(requests[1]?.params.state).toBe("working"); + }); + + it("reports lifecycle state to the herdr socket", async () => { + // Unix socket paths must stay under ~104 chars; use a short base dir. + const tempDir = join(tmpdir(), `hrd-${Math.random().toString(36).slice(2, 8)}`); + mkdirSync(tempDir, { recursive: true }); + cleanupPaths.push(tempDir); + const socketPath = join(tempDir, "h.sock"); + + const { server, requests, waitForRequests } = await startFakeHerdrServer(socketPath); + cleanupServers.push(server); + + process.env.HERDR_ENV = "1"; + process.env.HERDR_SOCKET_PATH = socketPath; + process.env.HERDR_PANE_ID = "w1:p1"; + process.env.HERDR_PI_IDLE_DEBOUNCE_MS = "10"; + // Isolate from any real agent dir that may contain the file-based integration. + process.env.PRIME_AGENT_CODING_AGENT_DIR = tempDir; + + const { pi, handlers } = createMockPi(); + herdrAgentStateExtension(pi); + + expect(handlers.has("session_start")).toBe(true); + expect(handlers.has("agent_start")).toBe(true); + expect(handlers.has("agent_end")).toBe(true); + expect(handlers.has("session_shutdown")).toBe(true); + + const ctx = { + sessionManager: { + getSessionFile: () => "/tmp/session.jsonl", + getSessionId: () => "session-1", + }, + }; + + handlers.get("session_start")?.[0]?.({ type: "session_start", reason: "startup" }, ctx); + await waitForRequests(1); + expect(requests[0]?.method).toBe("pane.report_agent"); + expect(requests[0]?.params.agent).toBe("prime-agent"); + expect(requests[0]?.params.pane_id).toBe("w1:p1"); + expect(requests[0]?.params.state).toBe("idle"); + expect(requests[0]?.params.agent_session_path).toBe("/tmp/session.jsonl"); + + handlers.get("agent_start")?.[0]?.({ type: "agent_start" }, ctx); + await waitForRequests(2); + expect(requests[1]?.params.state).toBe("working"); + + handlers.get("agent_end")?.[0]?.({ type: "agent_end", messages: [] }, ctx); + await waitForRequests(3); + expect(requests[2]?.params.state).toBe("idle"); + + // Session replacement must not release: the successor session re-reports, + // and a racing release could clear the successor's fresh report. + await handlers.get("session_shutdown")?.[0]?.({ type: "session_shutdown", reason: "new" }, ctx); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(requests).toHaveLength(3); + + await handlers.get("session_shutdown")?.[0]?.({ type: "session_shutdown", reason: "quit" }, ctx); + await waitForRequests(4); + expect(requests[3]?.method).toBe("pane.release_agent"); + expect(requests[3]?.params.agent).toBe("prime-agent"); + }); + + it("unsubscribes the shared-bus herdr:blocked listener on shutdown", async () => { + const tempDir = join(tmpdir(), `hrd-${Math.random().toString(36).slice(2, 8)}`); + mkdirSync(tempDir, { recursive: true }); + cleanupPaths.push(tempDir); + const socketPath = join(tempDir, "h.sock"); + + const { server } = await startFakeHerdrServer(socketPath); + cleanupServers.push(server); + + process.env.HERDR_ENV = "1"; + process.env.HERDR_SOCKET_PATH = socketPath; + process.env.HERDR_PANE_ID = "w1:p1"; + + const { pi, handlers, busHandlers } = createMockPi(); + herdrAgentStateExtension(pi); + expect(busHandlers.get("herdr:blocked")).toHaveLength(1); + + const ctx = { sessionManager: { getSessionFile: () => undefined, getSessionId: () => "s" } }; + // Replacement shutdowns must also unsubscribe, or every /reload and /new + // stacks another live listener on the shared bus. + await handlers.get("session_shutdown")?.[0]?.({ type: "session_shutdown", reason: "new" }, ctx); + expect(busHandlers.get("herdr:blocked")).toHaveLength(0); + }); + + it("reports working when the session starts mid-turn (reload)", async () => { + const tempDir = join(tmpdir(), `hrd-${Math.random().toString(36).slice(2, 8)}`); + mkdirSync(tempDir, { recursive: true }); + cleanupPaths.push(tempDir); + const socketPath = join(tempDir, "h.sock"); + + const { server, requests, waitForRequests } = await startFakeHerdrServer(socketPath); + cleanupServers.push(server); + + process.env.HERDR_ENV = "1"; + process.env.HERDR_SOCKET_PATH = socketPath; + process.env.HERDR_PANE_ID = "w1:p1"; + process.env.HERDR_PI_IDLE_DEBOUNCE_MS = "10"; + + const { pi, handlers } = createMockPi(); + herdrAgentStateExtension(pi); + + // Reload mid-turn: the fresh reporter's session_start sees a busy session. + const ctx = { + sessionManager: { getSessionFile: () => undefined, getSessionId: () => "s" }, + isIdle: () => false, + }; + handlers.get("session_start")?.[0]?.({ type: "session_start", reason: "reload" }, ctx); + await waitForRequests(1); + expect(requests[0]?.params.state).toBe("working"); + + // The turn's real end must still be honored (agentActive was seeded). + handlers.get("agent_end")?.[0]?.({ type: "agent_end", messages: [] }, ctx); + await waitForRequests(2); + expect(requests[1]?.params.state).toBe("idle"); + }); + + it("settles the retry hold when a blocked event interrupts it", async () => { + const tempDir = join(tmpdir(), `hrd-${Math.random().toString(36).slice(2, 8)}`); + mkdirSync(tempDir, { recursive: true }); + cleanupPaths.push(tempDir); + const socketPath = join(tempDir, "h.sock"); + + const { server, requests, waitForRequests } = await startFakeHerdrServer(socketPath); + cleanupServers.push(server); + + process.env.HERDR_ENV = "1"; + process.env.HERDR_SOCKET_PATH = socketPath; + process.env.HERDR_PANE_ID = "w1:p1"; + process.env.HERDR_PI_IDLE_DEBOUNCE_MS = "10"; + + const { pi, handlers, busHandlers } = createMockPi(); + herdrAgentStateExtension(pi); + + const ctx = { sessionManager: { getSessionFile: () => undefined, getSessionId: () => "s" } }; + handlers.get("session_start")?.[0]?.({ type: "session_start", reason: "startup" }, ctx); + handlers.get("agent_start")?.[0]?.({ type: "agent_start" }, ctx); + // End with a retryable provider error: enters the retry hold (working). + handlers.get("agent_end")?.[0]?.( + { + type: "agent_end", + messages: [{ role: "assistant", stopReason: "error", errorMessage: "rate limit exceeded" }], + }, + ctx, + ); + // A blocked event lands during the hold, cancelling the retry timer... + const blocked = busHandlers.get("herdr:blocked")?.[0]; + blocked?.({ active: true, label: "permission needed" }); + // ...and when the block lifts, the pane must settle to blocked (failed + // retry), not stick at working forever. Intermediate states coalesce in + // the latest-wins queue, so assert on the settled final report. + blocked?.({ active: false }); + + await waitForRequests(2); + await new Promise((resolve) => setTimeout(resolve, 100)); + const finalState = requests.at(-1)?.params.state; + expect(finalState).toBe("blocked"); + expect(requests.at(-1)?.params.message).toContain("rate limit"); + }); + + it("holds working through any error end until the retry grace settles", async () => { + const tempDir = join(tmpdir(), `hrd-${Math.random().toString(36).slice(2, 8)}`); + mkdirSync(tempDir, { recursive: true }); + cleanupPaths.push(tempDir); + const socketPath = join(tempDir, "h.sock"); + + const { server, requests, waitForRequests } = await startFakeHerdrServer(socketPath); + cleanupServers.push(server); + + process.env.HERDR_ENV = "1"; + process.env.HERDR_SOCKET_PATH = socketPath; + process.env.HERDR_PANE_ID = "w1:p1"; + process.env.HERDR_PI_IDLE_DEBOUNCE_MS = "10"; + process.env.HERDR_PI_RETRY_GRACE_MS = "30"; + + const { pi, handlers } = createMockPi(); + herdrAgentStateExtension(pi); + + const ctx = { sessionManager: { getSessionFile: () => undefined, getSessionId: () => "s" } }; + handlers.get("session_start")?.[0]?.({ type: "session_start", reason: "startup" }, ctx); + handlers.get("agent_start")?.[0]?.({ type: "agent_start" }, ctx); + // The agent's auto-retry classifies almost every provider error as + // retryable, so an error the old pattern list missed must still hold + // working (not flip idle) while a retry may be pending. + handlers.get("agent_end")?.[0]?.( + { + type: "agent_end", + messages: [{ role: "assistant", stopReason: "error", errorMessage: "unexpected provider failure xyz" }], + }, + ctx, + ); + + // No retry arrives: after the grace window the pane settles to blocked + // with the error message, never reporting idle in between. + await waitForRequests(3); + expect(requests.map((r) => r.params.state)).toEqual(["idle", "working", "blocked"]); + expect(requests.at(-1)?.params.message).toContain("unexpected provider failure"); + }); + + it("sends no reports after quit release", async () => { + const tempDir = join(tmpdir(), `hrd-${Math.random().toString(36).slice(2, 8)}`); + mkdirSync(tempDir, { recursive: true }); + cleanupPaths.push(tempDir); + const socketPath = join(tempDir, "h.sock"); + + const { server, requests, waitForRequests } = await startFakeHerdrServer(socketPath); + cleanupServers.push(server); + + process.env.HERDR_ENV = "1"; + process.env.HERDR_SOCKET_PATH = socketPath; + process.env.HERDR_PANE_ID = "w1:p1"; + + const { pi, handlers } = createMockPi(); + herdrAgentStateExtension(pi); + + const ctx = { sessionManager: { getSessionFile: () => undefined, getSessionId: () => "s" } }; + handlers.get("session_start")?.[0]?.({ type: "session_start", reason: "startup" }, ctx); + handlers.get("agent_start")?.[0]?.({ type: "agent_start" }, ctx); + // Quit while the working report may still be in flight; the release must + // be the final write, with no report reclaiming the pane afterwards. + await handlers.get("session_shutdown")?.[0]?.({ type: "session_shutdown", reason: "quit" }, ctx); + handlers.get("agent_start")?.[0]?.({ type: "agent_start" }, ctx); + + await waitForRequests(2); + await new Promise((resolve) => setTimeout(resolve, 100)); + // Still-queued reports are dropped by the release, so the exact report + // count depends on send timing; the invariant is that the release is the + // last write and nothing reclaims the pane after it. + expect(requests.at(-1)?.method).toBe("pane.release_agent"); + expect(requests.filter((r) => r.method === "pane.release_agent")).toHaveLength(1); + }); + + it("silences a replaced instance without releasing the pane", async () => { + const tempDir = join(tmpdir(), `hrd-${Math.random().toString(36).slice(2, 8)}`); + mkdirSync(tempDir, { recursive: true }); + cleanupPaths.push(tempDir); + const socketPath = join(tempDir, "h.sock"); + + const { server, requests, waitForRequests } = await startFakeHerdrServer(socketPath); + cleanupServers.push(server); + + process.env.HERDR_ENV = "1"; + process.env.HERDR_SOCKET_PATH = socketPath; + process.env.HERDR_PANE_ID = "w1:p1"; + + const { pi, handlers } = createMockPi(); + herdrAgentStateExtension(pi); + + const ctx = { sessionManager: { getSessionFile: () => undefined, getSessionId: () => "s" } }; + handlers.get("session_start")?.[0]?.({ type: "session_start", reason: "startup" }, ctx); + await waitForRequests(1); + + // Replacement shutdown: no release (the successor re-reports), and the + // old instance must go quiet - a late report with its process-wide seq + // would outrank and stomp the successor's state. + await handlers.get("session_shutdown")?.[0]?.({ type: "session_shutdown", reason: "new" }, ctx); + handlers.get("agent_start")?.[0]?.({ type: "agent_start" }, ctx); + await new Promise((resolve) => setTimeout(resolve, 100)); + + expect(requests).toHaveLength(1); + expect(requests.some((r) => r.method === "pane.release_agent")).toBe(false); + }); + + it("keeps seq monotonically increasing across extension instances", async () => { + const tempDir = join(tmpdir(), `hrd-${Math.random().toString(36).slice(2, 8)}`); + mkdirSync(tempDir, { recursive: true }); + cleanupPaths.push(tempDir); + const socketPath = join(tempDir, "h.sock"); + + const { server, requests, waitForRequests } = await startFakeHerdrServer(socketPath); + cleanupServers.push(server); + + process.env.HERDR_ENV = "1"; + process.env.HERDR_SOCKET_PATH = socketPath; + process.env.HERDR_PANE_ID = "w1:p1"; + process.env.PRIME_AGENT_CODING_AGENT_DIR = tempDir; + + const ctx = { sessionManager: { getSessionFile: () => undefined, getSessionId: () => "s" } }; + + // Two instances, as after a session replacement: the successor's seq must + // exceed everything the predecessor sent, or herdr drops its reports. + const first = createMockPi(); + herdrAgentStateExtension(first.pi); + first.handlers.get("session_start")?.[0]?.({ type: "session_start", reason: "startup" }, ctx); + await waitForRequests(1); + + const second = createMockPi(); + herdrAgentStateExtension(second.pi); + second.handlers.get("session_start")?.[0]?.({ type: "session_start", reason: "new" }, ctx); + await waitForRequests(2); + + const seqs = requests.map((r) => r.params.seq as number); + expect(seqs[1]).toBeGreaterThan(seqs[0]); + }); +});