From ecd32905eeed103c0d3677b0083dcc2e1f726c15 Mon Sep 17 00:00:00 2001 From: kt Date: Thu, 9 Jul 2026 15:46:40 -0700 Subject: [PATCH] add built-in herdr integration (#360) * feat(coding-agent): add built-in herdr integration Herdr cannot detect Prime Agent panes: its process detection only knows the legacy "pi" binary name, and its "herdr integration install pi" writes to ~/.pi/agent/extensions which Prime Agent does not read. Ship the pi lifecycle-reporter as a built-in extension factory loaded for every session. It reports working/idle/blocked and session refs to Herdr's unix socket and releases the pane on shutdown, using the agent label "prime-agent" (verified accepted and displayed by herdr 0.7.0 against a live pane). The factory captures HERDR_* env per invocation, so daemon sessions pick up their own pane identity inside the client-env window, and it is a no-op outside Herdr or when the user has installed Herdr's own file-based integration. * fix(coding-agent): keep herdr seq monotonic across sessions and release only on quit Herdr guards pane.report_agent with a per-source monotonic seq and silently drops lower-seq reports (verified live: ok response, state unchanged). Seeding seq per factory instance meant a successor session after /new, resume, fork, or reload could restart below the predecessor's counter, so all its reports were dropped and the pane stuck at working. - move the seq counter to module scope and clamp it to max(prev + 1, now*1000) so it never regresses across instances - release the pane only on quit; on session replacement the successor re-reports immediately and a racing release could clear it - report idle immediately on agent_end when no messages are queued; keep the debounce only when queued follow-ups restart the loop * fix(coding-agent): address herdr integration review findings - unsubscribe the shared-bus herdr:blocked listener on session_shutdown; the event bus survives reloads and session replacements, so a leaked listener kept stale instances reporting with captured pane identity - move the file-based integration check into createAgentSessionServices using the same extension dirs the loader discovers from, so agentDir overrides are honored - stop treating the legacy ~/.pi/agent path as an active reporter; Prime Agent never loads extensions from there, so deferring to it left panes with no reporter at all * fix(coding-agent): scope herdr reporter to its own session and recheck deferral on reload - bind the reporter to the first session that starts and ignore events from other sessions sharing the closure; inline RLM children reuse the parent's resource loader, so a subagent's turns flipped the pane state and a subagent quit released the pane while the parent was running - skip the built-in reporter entirely for subagent runtimes (rlmDepth > 0), which share the parent's HERDR_* pane identity - re-check for Herdr's file-based integration on every factory invocation instead of once at service creation, so installing it and running /reload hands the pane over without both reporters active - also match herdr-agent-state.js for compiled installs * fix(coding-agent): make quit release the final write to the herdr socket releaseAgent now stops new reports, drops the queued state, and awaits the in-flight drain before sending pane.release_agent, so a late report cannot land after the release and reclaim the pane for an exited agent. * fix(coding-agent): seed working state when the reporter starts mid-turn Daemon-driven reloads and extension ctx.reload() are not gated on idle, so a fresh reporter's session_start could publish idle while the agent was still streaming, and the agentActive guard then swallowed the turn's real end transition. session_start now seeds agentActive from ctx.isIdle(). * fix(coding-agent): settle the retry hold when a blocked event cancels its timer A herdr:blocked event during the retry-hold window cancelled the retry timer but left retryHoldActive set, so after the block lifted the pane reported working forever. The blocked handler now settles the hold to failureBlocked, matching what the cancelled timer would have done. * fix(coding-agent): do not defer to file-based herdr reporter under noExtensions With noExtensions the loader never scans the extension dirs, so a file-based herdr-agent-state there never becomes an active reporter; deferring to it left the pane with no reporter at all. * fix(coding-agent): hold working for every error end instead of a pattern subset The agent's auto-retry treats nearly all provider errors as retryable and runs after extension agent_end, so an error the reporter's pattern list missed flipped the pane to idle while a retry was pending. Drop the pattern list and hold working for any error end; a retry's agent_start keeps the pane working, and with no retry the grace window settles it to blocked with the error message. * fix(coding-agent): silence replaced reporter instances on non-quit shutdown A replaced instance kept its report queue armed after reload/new/resume shutdowns; a late report carrying the process-wide seq could outrank and stomp the successor's state. Non-quit shutdowns now set released and drop the queued state so the old instance goes quiet. * fix(coding-agent): make noExtensions disable the built-in herdr reporter noExtensions is a full opt-out: inline factories deliberately bypass it for embedder-supplied extensions, but the built-in reporter should respect it so --no-extensions leaves nothing extension-driven running. * fix(coding-agent): defer to the file-based herdr reporter only when it loaded Deferral now consults the resource loader's actually-loaded extension paths (late-bound, per factory invocation) instead of raw disk existence, so a herdr-agent-state file disabled via settings overrides or outside the discovery dirs no longer silences the built-in reporter and leaves the pane with no reporter. --- packages/coding-agent/CHANGELOG.md | 1 + .../src/core/agent-session-services.ts | 23 +- .../extensions/builtin/herdr-agent-state.ts | 461 ++++++++++++++++ .../coding-agent/src/core/extensions/index.ts | 6 + .../coding-agent/src/core/resource-loader.ts | 10 + packages/coding-agent/src/main.ts | 3 + .../test/herdr-agent-state.test.ts | 513 ++++++++++++++++++ 7 files changed, 1016 insertions(+), 1 deletion(-) create mode 100644 packages/coding-agent/src/core/extensions/builtin/herdr-agent-state.ts create mode 100644 packages/coding-agent/test/herdr-agent-state.test.ts 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]); + }); +}); -- 2.54.0