From 0afdc40e68a717aef1540bad07ef867c3cbe80d4 Mon Sep 17 00:00:00 2001 From: Miya Date: Thu, 23 Jul 2026 14:01:59 +0200 Subject: [PATCH 1/5] feat: add agent relay bridge plugin --- .github/workflows/agent-relay-plugin.yml | 50 ++++ plugins/agent-relay/README.md | 42 +++ plugins/agent-relay/config.example.json | 6 + plugins/agent-relay/dist/bridge.mjs | 246 ++++++++++++++++++ plugins/agent-relay/dist/config.mjs | 55 ++++ plugins/agent-relay/dist/herdr-socket.mjs | 131 ++++++++++ plugins/agent-relay/dist/state.mjs | 51 ++++ plugins/agent-relay/herdr-plugin.toml | 15 ++ plugins/agent-relay/package-lock.json | 58 +++++ plugins/agent-relay/package.json | 16 ++ plugins/agent-relay/test/bridge.test.mjs | 303 ++++++++++++++++++++++ 11 files changed, 973 insertions(+) create mode 100644 .github/workflows/agent-relay-plugin.yml create mode 100644 plugins/agent-relay/README.md create mode 100644 plugins/agent-relay/config.example.json create mode 100644 plugins/agent-relay/dist/bridge.mjs create mode 100644 plugins/agent-relay/dist/config.mjs create mode 100644 plugins/agent-relay/dist/herdr-socket.mjs create mode 100644 plugins/agent-relay/dist/state.mjs create mode 100644 plugins/agent-relay/herdr-plugin.toml create mode 100644 plugins/agent-relay/package-lock.json create mode 100644 plugins/agent-relay/package.json create mode 100644 plugins/agent-relay/test/bridge.test.mjs diff --git a/.github/workflows/agent-relay-plugin.yml b/.github/workflows/agent-relay-plugin.yml new file mode 100644 index 0000000000..0bafbfa071 --- /dev/null +++ b/.github/workflows/agent-relay-plugin.yml @@ -0,0 +1,50 @@ +name: Agent Relay Plugin + +on: + pull_request: + paths: + - ".github/workflows/agent-relay-plugin.yml" + - "plugins/agent-relay/**" + push: + branches: [master] + paths: + - ".github/workflows/agent-relay-plugin.yml" + - "plugins/agent-relay/**" + +permissions: + contents: read + +concurrency: + group: agent-relay-plugin-${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }} + cancel-in-progress: true + +jobs: + test: + name: test (${{ matrix.os }}) + runs-on: ${{ matrix.os }} + timeout-minutes: 10 + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-latest, windows-latest] + defaults: + run: + working-directory: plugins/agent-relay + steps: + - name: Checkout + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 + with: + persist-credentials: false + + - name: Install Node.js 22 + uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0 + with: + node-version: 22 + cache: npm + cache-dependency-path: plugins/agent-relay/package-lock.json + + - name: Install production dependencies + run: npm ci --omit=dev + + - name: Run plugin tests + run: npm test diff --git a/plugins/agent-relay/README.md b/plugins/agent-relay/README.md new file mode 100644 index 0000000000..1d1b487ce9 --- /dev/null +++ b/plugins/agent-relay/README.md @@ -0,0 +1,42 @@ +# Agent Relay bridge plugin + +This optional Herdr plugin runs a Node 22+ sidecar in an explicitly opened tab. +It forwards agent status changes for configured workspaces to one Agent Relay +channel and exposes a read-only `herdr.session_summary` Relay action. + +The bridge never reads pane output, cwd, environment, or terminal titles. It +does not expose prompt, key-send, shell, or raw-socket controls. + +## Setup + +Install dependencies when authoring a linked checkout: + +```bash +npm ci --omit=dev +herdr plugin link /path/to/agent-relay +``` + +Herdr runs the same `npm ci --omit=dev` command automatically for GitHub +installs. The plugin requires Node 22 or newer. + +Copy `config.example.json` to the directory reported by: + +```bash +herdr plugin config-dir agent-relay.herdr-bridge +``` + +Name the copied file `agent-relay.json`, replace `workspaceKey`, and list only +the Herdr workspace IDs that may be forwarded. Then open the sidecar: + +```bash +herdr plugin pane open --plugin agent-relay.herdr-bridge --entrypoint bridge +``` + +Closing that pane stops the bridge. It does not run from a startup hook. + +## State and safety + +The Relay agent token and status-transition dedupe state are stored only in +`HERDR_PLUGIN_STATE_DIR` as a mode-0600 file. Subsequent starts reconnect with +that token rather than registering another Relay agent. The configuration file +belongs in `HERDR_PLUGIN_CONFIG_DIR`; do not commit a real workspace key. diff --git a/plugins/agent-relay/config.example.json b/plugins/agent-relay/config.example.json new file mode 100644 index 0000000000..b0a76b33b0 --- /dev/null +++ b/plugins/agent-relay/config.example.json @@ -0,0 +1,6 @@ +{ + "workspaceKey": "rk_live_replace_me", + "baseUrl": "https://gateway.relaycast.dev", + "channel": "#agent-status", + "workspaceIds": ["w1"] +} diff --git a/plugins/agent-relay/dist/bridge.mjs b/plugins/agent-relay/dist/bridge.mjs new file mode 100644 index 0000000000..24db74b1cb --- /dev/null +++ b/plugins/agent-relay/dist/bridge.mjs @@ -0,0 +1,246 @@ +import { createHash } from 'node:crypto'; +import { hostname } from 'node:os'; + +import { AgentRelay } from '@agent-relay/sdk'; +import { z } from 'zod'; + +import { loadBridgeConfig, pluginPaths } from './config.mjs'; +import { requestHerdr, subscribeToBridgeEvents } from './herdr-socket.mjs'; +import { loadBridgeState, saveBridgeState } from './state.mjs'; + +const SUMMARY_INPUT = z.object({}).strict(); +const STATUS_VALUES = new Set(['idle', 'working', 'blocked', 'done', 'unknown']); + +function hash(value) { + return createHash('sha256').update(value).digest('hex'); +} + +export function bridgeName(socketPath, host = hostname()) { + const safeHost = host.toLowerCase().replace(/[^a-z0-9-]+/g, '-').replace(/^-+|-+$/g, '') || 'host'; + return `herdr-${safeHost.slice(0, 30)}-${hash(socketPath).slice(0, 12)}`; +} + +export function sessionSnapshotFrom(response) { + const snapshot = response?.result?.snapshot; + if (!snapshot || !Array.isArray(snapshot.agents) || !Array.isArray(snapshot.panes)) { + throw new Error('Herdr returned an invalid session snapshot'); + } + return snapshot; +} + +export function summarizeSnapshot(snapshot, workspaceIds) { + const allowed = new Set(workspaceIds); + const statuses = { idle: 0, working: 0, blocked: 0, done: 0, unknown: 0 }; + let agents = 0; + for (const agent of snapshot.agents) { + if (!allowed.has(agent.workspace_id) || !STATUS_VALUES.has(agent.agent_status)) continue; + statuses[agent.agent_status] += 1; + agents += 1; + } + return { workspaceIds: [...allowed], agents, statuses }; +} + +export function formatSessionSummary(summary) { + const { statuses } = summary; + return `Herdr summary: ${summary.agents} agents across ${summary.workspaceIds.length} workspaces — working ${statuses.working}, blocked ${statuses.blocked}, idle ${statuses.idle}, done ${statuses.done}, unknown ${statuses.unknown}.`; +} + +export function normalizeStatusEvent(message, workspaceIds) { + const data = message?.data; + if ( + message?.event !== 'pane.agent_status_changed' || + !data || + typeof data.pane_id !== 'string' || + typeof data.workspace_id !== 'string' || + !STATUS_VALUES.has(data.agent_status) || + !workspaceIds.has(data.workspace_id) + ) { + return undefined; + } + return { + paneId: data.pane_id, + workspaceId: data.workspace_id, + agent: typeof data.agent === 'string' && data.agent ? data.agent : 'unknown', + status: data.agent_status, + }; +} + +export function prepareTransition(state, event) { + const fingerprint = [event.paneId, event.workspaceId, event.agent, event.status].join('\u0000'); + const previous = state.transitions[event.paneId]; + if (previous?.fingerprint === fingerprint) return undefined; + const sequence = (previous?.sequence ?? 0) + 1; + state.transitions[event.paneId] = { fingerprint, sequence }; + return { + ...event, + idempotencyKey: `herdr-status-${hash(`${event.paneId}\u0000${sequence}\u0000${fingerprint}`)}`, + }; +} + +export function formatStatusMessage(event) { + return `Herdr agent status: ${event.workspaceId}/${event.paneId} ${event.agent} is ${event.status}.`; +} + +function stateWriter(stateDir, state) { + let pending = Promise.resolve(); + return () => { + pending = pending.catch(() => undefined).then(() => saveBridgeState(stateDir, state)); + return pending; + }; +} + +function isPaneLifecycleEvent(message) { + return ['pane.created', 'pane.closed', 'pane.moved', 'workspace.closed'].includes(message?.event); +} + +function createSubscriptionManager({ socketPath, workspaceIds, onStatus, logger }) { + let active; + let stopped = false; + let pending = Promise.resolve(); + let reconnectTimer; + + const scheduleRebuild = () => { + if (stopped || reconnectTimer) return; + reconnectTimer = setTimeout(() => { + reconnectTimer = undefined; + void rebuild().catch(() => { + logger.warn('Agent Relay bridge could not reconnect to Herdr'); + scheduleRebuild(); + }); + }, 1_000); + }; + + const rebuild = () => { + pending = pending.catch(() => undefined).then(async () => { + if (stopped) return; + const snapshot = sessionSnapshotFrom(await requestHerdr(socketPath, 'session.snapshot')); + const paneIds = snapshot.panes + .filter((pane) => workspaceIds.has(pane.workspace_id)) + .map((pane) => pane.pane_id); + const next = await subscribeToBridgeEvents(socketPath, paneIds, (message) => { + if (isPaneLifecycleEvent(message)) { + void rebuild().catch(() => scheduleRebuild()); + return; + } + onStatus(message); + }); + if (stopped) { + next.close(); + return; + } + const previous = active; + active = next; + next.once('closed', () => { + if (active === next && !stopped) scheduleRebuild(); + }); + if (next.closed) scheduleRebuild(); + previous?.close(); + }); + return pending; + }; + + return { + async start() { + await rebuild(); + }, + stop() { + stopped = true; + if (reconnectTimer) clearTimeout(reconnectTimer); + active?.close(); + }, + }; +} + +async function connectRelay(config, state, socketPath, AgentRelayCtor) { + const options = { workspaceKey: config.workspaceKey }; + if (config.baseUrl) options.baseUrl = config.baseUrl; + const relay = new AgentRelayCtor(options); + if (state.apiToken) { + return { agent: await relay.workspace.reconnect({ apiToken: state.apiToken }) }; + } + const agent = await relay.workspace.register({ + name: bridgeName(socketPath), + type: 'agent', + metadata: { integration: 'herdr' }, + }); + if (!agent.token) throw new Error('Agent Relay registration did not return an agent token'); + state.apiToken = agent.token; + return { agent }; +} + +export function registerSessionSummaryAction(agent, socketPath, workspaceIds) { + return agent.registerAction({ + name: 'herdr.session_summary', + description: 'Return aggregate Herdr agent status counts for the configured workspaces.', + input: SUMMARY_INPUT, + handler: async ({ agent: caller }) => { + const snapshot = sessionSnapshotFrom(await requestHerdr(socketPath, 'session.snapshot')); + const summary = summarizeSnapshot(snapshot, workspaceIds); + const recipient = caller.handle ?? caller.name; + await agent.sendMessage({ to: `@${recipient}`, text: formatSessionSummary(summary) }); + return summary; + }, + }); +} + +export async function startBridge({ + environment = process.env, + AgentRelayCtor = AgentRelay, + logger = console, +} = {}) { + const { configDir, stateDir } = pluginPaths(environment); + const socketPath = environment.HERDR_SOCKET_PATH; + if (!socketPath) throw new Error('Herdr did not provide HERDR_SOCKET_PATH'); + + const config = await loadBridgeConfig(configDir); + const workspaceIds = new Set(config.workspaceIds); + const state = await loadBridgeState(stateDir); + const persistState = stateWriter(stateDir, state); + + await requestHerdr(socketPath, 'ping'); + + const { agent } = await connectRelay(config, state, socketPath, AgentRelayCtor); + await persistState(); + const action = registerSessionSummaryAction(agent, socketPath, config.workspaceIds); + const subscriptions = createSubscriptionManager({ + socketPath, + workspaceIds, + logger, + onStatus(message) { + const event = normalizeStatusEvent(message, workspaceIds); + if (!event) return; + const transition = prepareTransition(state, event); + if (!transition) return; + void agent + .sendMessage({ + to: config.channel, + text: formatStatusMessage(transition), + idempotencyKey: transition.idempotencyKey, + }) + .then(() => persistState()) + .catch(() => logger.warn('Agent Relay bridge could not forward a Herdr status update')); + }, + }); + await subscriptions.start(); + + return { + stop() { + action.unregister(); + subscriptions.stop(); + }, + }; +} + +async function main() { + const bridge = await startBridge(); + const stop = () => bridge.stop(); + process.once('SIGINT', stop); + process.once('SIGTERM', stop); +} + +if (import.meta.url === `file://${process.argv[1]}`) { + main().catch((error) => { + console.error(`Agent Relay bridge failed: ${error.message}`); + process.exitCode = 1; + }); +} diff --git a/plugins/agent-relay/dist/config.mjs b/plugins/agent-relay/dist/config.mjs new file mode 100644 index 0000000000..c570117fda --- /dev/null +++ b/plugins/agent-relay/dist/config.mjs @@ -0,0 +1,55 @@ +import { readFile } from 'node:fs/promises'; +import { join } from 'node:path'; + +import { z } from 'zod'; + +const channelName = z.string().trim().regex(/^#[A-Za-z0-9][A-Za-z0-9_-]*$/, { + message: 'channel must be a #channel-name', +}); + +export const BridgeConfigSchema = z + .object({ + workspaceKey: z.string().trim().min(1), + baseUrl: z.string().url().optional(), + channel: channelName, + workspaceIds: z + .array(z.string().trim().min(1)) + .min(1) + .refine((ids) => new Set(ids).size === ids.length, 'workspaceIds must not contain duplicates'), + }) + .strict(); + +export function pluginPaths(environment = process.env) { + const configDir = environment.HERDR_PLUGIN_CONFIG_DIR; + const stateDir = environment.HERDR_PLUGIN_STATE_DIR; + if (!configDir || !stateDir) { + throw new Error('Herdr did not provide plugin config and state directories'); + } + return { configDir, stateDir }; +} + +export function configPath(configDir) { + return join(configDir, 'agent-relay.json'); +} + +export async function loadBridgeConfig(configDir) { + let parsed; + try { + parsed = JSON.parse(await readFile(configPath(configDir), 'utf8')); + } catch (error) { + if (error instanceof SyntaxError) { + throw new Error('Agent Relay bridge configuration is not valid JSON'); + } + throw new Error(`Cannot read Agent Relay bridge configuration at ${configPath(configDir)}`); + } + + const result = BridgeConfigSchema.safeParse(parsed); + if (!result.success) { + throw new Error('Agent Relay bridge configuration is invalid'); + } + return result.data; +} + +export function redactConfig(config) { + return { ...config, workspaceKey: '[redacted]' }; +} diff --git a/plugins/agent-relay/dist/herdr-socket.mjs b/plugins/agent-relay/dist/herdr-socket.mjs new file mode 100644 index 0000000000..e40ea46602 --- /dev/null +++ b/plugins/agent-relay/dist/herdr-socket.mjs @@ -0,0 +1,131 @@ +import { EventEmitter } from 'node:events'; +import net from 'node:net'; + +let requestSequence = 0; + +export function localSocketTarget(socketPath, platform = process.platform) { + if (!socketPath) throw new Error('HERDR_SOCKET_PATH is required'); + if (platform !== 'win32') return socketPath; + if (/^\\\\[^\\]+\\pipe\\/i.test(socketPath)) return socketPath; + return `\\\\.\\pipe\\${socketPath}`; +} + +function nextRequestId() { + requestSequence += 1; + return `agent-relay-${process.pid}-${requestSequence}`; +} + +export class JsonLineSocket extends EventEmitter { + static async connect(socketPath, options = {}) { + const target = localSocketTarget(socketPath, options.platform); + const socket = net.createConnection(target); + await new Promise((resolve, reject) => { + const onError = (error) => { + socket.off('connect', onConnect); + reject(error); + }; + const onConnect = () => { + socket.off('error', onError); + resolve(); + }; + socket.once('error', onError); + socket.once('connect', onConnect); + }); + return new JsonLineSocket(socket); + } + + constructor(socket) { + super(); + this.socket = socket; + this.buffer = ''; + this.closed = false; + this.pending = new Map(); + socket.setEncoding('utf8'); + socket.on('data', (chunk) => this.#receive(chunk)); + socket.on('error', (error) => this.#rejectAll(error)); + const handleClosed = () => { + if (this.closed) return; + this.closed = true; + this.#rejectAll(new Error('Herdr socket closed')); + this.emit('closed'); + }; + socket.on('end', handleClosed); + socket.on('close', handleClosed); + } + + request(method, params) { + const id = nextRequestId(); + return new Promise((resolve, reject) => { + this.pending.set(id, { resolve, reject }); + this.socket.write(`${JSON.stringify({ id, method, params })}\n`, (error) => { + if (!error) return; + this.pending.delete(id); + reject(error); + }); + }); + } + + close() { + this.socket.end(); + this.socket.destroy(); + } + + #receive(chunk) { + this.buffer += chunk; + let newline; + while ((newline = this.buffer.indexOf('\n')) !== -1) { + const line = this.buffer.slice(0, newline).trim(); + this.buffer = this.buffer.slice(newline + 1); + if (!line) continue; + let message; + try { + message = JSON.parse(line); + } catch { + this.emit('protocolError'); + continue; + } + const pending = typeof message.id === 'string' ? this.pending.get(message.id) : undefined; + if (pending) { + this.pending.delete(message.id); + if (message.error) pending.reject(new Error('Herdr API request failed')); + else pending.resolve(message); + continue; + } + this.emit('event', message); + } + } + + #rejectAll(error) { + for (const { reject } of this.pending.values()) reject(error); + this.pending.clear(); + } +} + +export async function requestHerdr(socketPath, method, params = {}) { + const client = await JsonLineSocket.connect(socketPath); + try { + return await client.request(method, params); + } finally { + client.close(); + } +} + +export async function subscribeToBridgeEvents(socketPath, paneIds, onEvent) { + const client = await JsonLineSocket.connect(socketPath); + client.on('event', onEvent); + try { + await client.request('events.subscribe', { + subscriptions: [ + { type: 'pane.created' }, + { type: 'pane.closed' }, + { type: 'pane.moved' }, + { type: 'workspace.closed' }, + ...[...new Set(paneIds)].map((pane_id) => ({ type: 'pane.agent_status_changed', pane_id })), + ], + }); + return client; + } catch (error) { + client.close(); + throw error; + } +} diff --git a/plugins/agent-relay/dist/state.mjs b/plugins/agent-relay/dist/state.mjs new file mode 100644 index 0000000000..9e17148980 --- /dev/null +++ b/plugins/agent-relay/dist/state.mjs @@ -0,0 +1,51 @@ +import { chmod, mkdir, readFile, rename, writeFile } from 'node:fs/promises'; +import { join } from 'node:path'; + +import { z } from 'zod'; + +const TransitionSchema = z.object({ + fingerprint: z.string().min(1), + sequence: z.number().int().positive(), +}); + +const BridgeStateSchema = z + .object({ + apiToken: z.string().min(1).optional(), + transitions: z.record(z.string(), TransitionSchema).default({}), + }) + .strict(); + +export function statePath(stateDir) { + return join(stateDir, 'relay-state.json'); +} + +export function emptyBridgeState() { + return { transitions: {} }; +} + +export async function loadBridgeState(stateDir) { + let parsed; + try { + parsed = JSON.parse(await readFile(statePath(stateDir), 'utf8')); + } catch (error) { + if (error && error.code === 'ENOENT') return emptyBridgeState(); + if (error instanceof SyntaxError) { + throw new Error('Agent Relay bridge state is not valid JSON'); + } + throw new Error('Cannot read Agent Relay bridge state'); + } + + const result = BridgeStateSchema.safeParse(parsed); + if (!result.success) throw new Error('Agent Relay bridge state is invalid'); + return result.data; +} + +export async function saveBridgeState(stateDir, state) { + await mkdir(stateDir, { recursive: true, mode: 0o700 }); + const target = statePath(stateDir); + const temporary = `${target}.tmp-${process.pid}`; + await writeFile(temporary, `${JSON.stringify(state)}\n`, { mode: 0o600 }); + await chmod(temporary, 0o600); + await rename(temporary, target); + await chmod(target, 0o600); +} diff --git a/plugins/agent-relay/herdr-plugin.toml b/plugins/agent-relay/herdr-plugin.toml new file mode 100644 index 0000000000..3b8258d539 --- /dev/null +++ b/plugins/agent-relay/herdr-plugin.toml @@ -0,0 +1,15 @@ +id = "agent-relay.herdr-bridge" +name = "Agent Relay bridge" +version = "0.1.0" +min_herdr_version = "0.7.5" +description = "Forward selected Herdr agent status changes to Agent Relay." +platforms = ["linux", "macos", "windows"] + +[[build]] +command = ["npm", "ci", "--omit=dev"] + +[[panes]] +id = "bridge" +title = "Agent Relay bridge" +placement = "tab" +command = ["node", "dist/bridge.mjs"] diff --git a/plugins/agent-relay/package-lock.json b/plugins/agent-relay/package-lock.json new file mode 100644 index 0000000000..3ab2e6b248 --- /dev/null +++ b/plugins/agent-relay/package-lock.json @@ -0,0 +1,58 @@ +{ + "name": "@agentworkforce/herdr-agent-relay-plugin", + "version": "0.1.0", + "lockfileVersion": 3, + "requires": true, + "packages": { + "": { + "name": "@agentworkforce/herdr-agent-relay-plugin", + "version": "0.1.0", + "dependencies": { + "@agent-relay/sdk": "11.1.0", + "zod": "4.3.6" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@agent-relay/sdk": { + "version": "11.1.0", + "resolved": "https://registry.npmjs.org/@agent-relay/sdk/-/sdk-11.1.0.tgz", + "integrity": "sha512-Q5avUPmfw677Kr3aBJtEO0WP606CEcYPrh/9cS4am8Tnw/o0mCaqhmFRT7KT/vE5yQRnY6qDb0LuSjmXnKIbpg==", + "dependencies": { + "@relaycast/sdk": "^6.0.0", + "@relaycast/types": "^6.0.0", + "zod": "^4.3.6" + }, + "engines": { + "node": ">=22.0.0" + } + }, + "node_modules/@relaycast/sdk": { + "version": "6.2.0", + "resolved": "https://registry.npmjs.org/@relaycast/sdk/-/sdk-6.2.0.tgz", + "integrity": "sha512-7j/QGtBf6WNRa8tnnCT1rfDmcIl5QyhSQZQ//R3PePNKrRwQum0K1i3lR/dmd8DkO37Wgr8rBAh3i1xIHiLFeg==", + "dependencies": { + "@relaycast/types": "6.2.0", + "zod": "^4.3.6" + } + }, + "node_modules/@relaycast/types": { + "version": "6.2.0", + "resolved": "https://registry.npmjs.org/@relaycast/types/-/types-6.2.0.tgz", + "integrity": "sha512-9e5ywyO0HM3yNfGE2JvvGWMeyv1xwshuLNkyEPGOCPyyw/UlFFQLESGShEy2GMwowZPlw8sm5mVTzmJqRZ3CLQ==", + "dependencies": { + "zod": "^4.3.6" + } + }, + "node_modules/zod": { + "version": "4.3.6", + "resolved": "https://registry.npmjs.org/zod/-/zod-4.3.6.tgz", + "integrity": "sha512-rftlrkhHZOcjDwkGlnUtZZkvaPHCsDATp4pGpuOOMDaTdDDXF91wuVDJoWoPsKX/3YPQ5fHuF3STjcYyKr+Qhg==", + "license": "MIT", + "funding": { + "url": "https://github.com/sponsors/colinhacks" + } + } + } +} diff --git a/plugins/agent-relay/package.json b/plugins/agent-relay/package.json new file mode 100644 index 0000000000..fb084a83d5 --- /dev/null +++ b/plugins/agent-relay/package.json @@ -0,0 +1,16 @@ +{ + "name": "@agentworkforce/herdr-agent-relay-plugin", + "version": "0.1.0", + "private": true, + "type": "module", + "engines": { + "node": ">=22.0.0" + }, + "scripts": { + "test": "node --test" + }, + "dependencies": { + "@agent-relay/sdk": "11.1.0", + "zod": "4.3.6" + } +} diff --git a/plugins/agent-relay/test/bridge.test.mjs b/plugins/agent-relay/test/bridge.test.mjs new file mode 100644 index 0000000000..030b840b08 --- /dev/null +++ b/plugins/agent-relay/test/bridge.test.mjs @@ -0,0 +1,303 @@ +import assert from 'node:assert/strict'; +import { mkdir, mkdtemp, readFile, rm, stat, unlink, writeFile } from 'node:fs/promises'; +import net from 'node:net'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import test from 'node:test'; + +import { + bridgeName, + normalizeStatusEvent, + prepareTransition, + startBridge, + summarizeSnapshot, +} from '../dist/bridge.mjs'; +import { loadBridgeConfig, redactConfig } from '../dist/config.mjs'; +import { localSocketTarget } from '../dist/herdr-socket.mjs'; + +const snapshot = { + agents: [ + { workspace_id: 'w1', agent_status: 'working' }, + { workspace_id: 'w1', agent_status: 'blocked' }, + { workspace_id: 'w2', agent_status: 'idle' }, + ], + panes: [ + { pane_id: 'w1:p1', workspace_id: 'w1' }, + { pane_id: 'w2:p1', workspace_id: 'w2' }, + ], +}; + +async function eventually(assertion, attempts = 50) { + let lastError; + for (let attempt = 0; attempt < attempts; attempt += 1) { + try { + return await assertion(); + } catch (error) { + lastError = error; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + } + throw lastError; +} + +async function withTemporaryDirectory(run) { + const directory = await mkdtemp(join(tmpdir(), 'herdr-agent-relay-')); + try { + return await run(directory); + } finally { + await rm(directory, { recursive: true, force: true }); + } +} + +async function fakeHerdrServer(socketPath, { dynamicPane = true, disconnectFirstSubscription = false } = {}) { + const requests = []; + let currentSnapshot = structuredClone(snapshot); + let subscriptions = 0; + const server = net.createServer((socket) => { + socket.setEncoding('utf8'); + let buffer = ''; + socket.on('data', (chunk) => { + buffer += chunk; + let newline; + while ((newline = buffer.indexOf('\n')) !== -1) { + const request = JSON.parse(buffer.slice(0, newline)); + buffer = buffer.slice(newline + 1); + requests.push(request); + if (request.method === 'ping') { + socket.write(`${JSON.stringify({ id: request.id, result: { type: 'pong' } })}\n`); + } else if (request.method === 'session.snapshot') { + socket.write(`${JSON.stringify({ id: request.id, result: { snapshot: currentSnapshot } })}\n`); + } else if (request.method === 'events.subscribe') { + socket.write(`${JSON.stringify({ id: request.id, result: { type: 'subscribed' } })}\n`); + subscriptions += 1; + if (dynamicPane && subscriptions === 1) { + queueMicrotask(() => { + currentSnapshot = { + ...currentSnapshot, + panes: [...currentSnapshot.panes, { pane_id: 'w1:p2', workspace_id: 'w1' }], + }; + socket.write( + `${JSON.stringify({ + event: 'pane.created', + data: { pane: { pane_id: 'w1:p2', workspace_id: 'w1' } }, + })}\n` + ); + }); + } else if (dynamicPane && subscriptions === 2) { + queueMicrotask(() => { + socket.write( + `${JSON.stringify({ + event: 'pane.agent_status_changed', + data: { pane_id: 'w1:p2', workspace_id: 'w1', agent: 'codex', agent_status: 'working' }, + })}\n` + ); + }); + } else if (disconnectFirstSubscription && subscriptions === 1) { + queueMicrotask(() => socket.end()); + } + } + } + }); + }); + await new Promise((resolve) => server.listen(socketPath, resolve)); + return { requests, server }; +} + +async function writeConfig(configDir) { + await mkdir(configDir, { recursive: true }); + await writeFile( + join(configDir, 'agent-relay.json'), + JSON.stringify({ workspaceKey: 'rk_live_secret', channel: '#agent-status', workspaceIds: ['w1'] }) + ); +} + +test('validates configuration and redacts workspace credentials', async () => { + await withTemporaryDirectory(async (directory) => { + await writeConfig(directory); + const config = await loadBridgeConfig(directory); + assert.equal(redactConfig(config).workspaceKey, '[redacted]'); + assert.equal(config.channel, '#agent-status'); + }); +}); + +test('uses the interprocess-compatible Windows pipe name', () => { + assert.equal(localSocketTarget('C:\\Users\\me\\herdr.sock', 'win32'), '\\\\.\\pipe\\C:\\Users\\me\\herdr.sock'); + assert.equal(localSocketTarget('\\\\.\\pipe\\herdr.sock', 'win32'), '\\\\.\\pipe\\herdr.sock'); +}); + +test('deduplicates identical status events while preserving later repeated states', () => { + const state = { transitions: {} }; + const working = { paneId: 'w1:p1', workspaceId: 'w1', agent: 'codex', status: 'working' }; + const blocked = { ...working, status: 'blocked' }; + assert.match(prepareTransition(state, working).idempotencyKey, /^herdr-status-/); + assert.equal(prepareTransition(state, working), undefined); + assert.match(prepareTransition(state, blocked).idempotencyKey, /^herdr-status-/); + assert.match(prepareTransition(state, working).idempotencyKey, /^herdr-status-/); + assert.equal(state.transitions['w1:p1'].sequence, 3); +}); + +test('filters status events and summarizes only configured workspaces', () => { + const allowed = new Set(['w1']); + assert.equal( + normalizeStatusEvent( + { event: 'pane.agent_status_changed', data: { pane_id: 'w2:p1', workspace_id: 'w2', agent_status: 'working' } }, + allowed + ), + undefined + ); + assert.deepEqual(summarizeSnapshot(snapshot, ['w1']), { + workspaceIds: ['w1'], + agents: 2, + statuses: { idle: 0, working: 1, blocked: 1, done: 0, unknown: 0 }, + }); +}); + +test('rebuilds per-pane subscriptions after a pane is created and forwards its status', async () => { + await withTemporaryDirectory(async (directory) => { + const configDir = join(directory, 'config'); + const stateDir = join(directory, 'state'); + const socketPath = join(directory, 'herdr.sock'); + await writeConfig(configDir); + const { server, requests } = await fakeHerdrServer(socketPath); + const sent = []; + const actions = []; + let registrations = 0; + const client = { + token: 'at_live_bridge', + registerAction(definition) { + actions.push(definition); + return { unregister() {} }; + }, + async sendMessage(input) { + sent.push(input); + }, + }; + class FakeRelay { + constructor(options) { + assert.equal(options.workspaceKey, 'rk_live_secret'); + this.workspace = { + register: async () => { + registrations += 1; + return client; + }, + reconnect: async () => { + throw new Error('reconnect should not run without persisted state'); + }, + }; + } + } + + const bridge = await startBridge({ + environment: { HERDR_PLUGIN_CONFIG_DIR: configDir, HERDR_PLUGIN_STATE_DIR: stateDir, HERDR_SOCKET_PATH: socketPath }, + AgentRelayCtor: FakeRelay, + logger: { warn() {} }, + }); + await eventually(() => assert.equal(requests.filter((request) => request.method === 'events.subscribe').length, 2)); + await eventually(() => assert.equal(sent.length, 1)); + assert.equal(registrations, 1); + assert.deepEqual(requests.filter((request) => request.method === 'events.subscribe')[0].params.subscriptions, [ + { type: 'pane.created' }, + { type: 'pane.closed' }, + { type: 'pane.moved' }, + { type: 'workspace.closed' }, + { type: 'pane.agent_status_changed', pane_id: 'w1:p1' }, + ]); + assert.deepEqual(requests.filter((request) => request.method === 'events.subscribe')[1].params.subscriptions, [ + { type: 'pane.created' }, + { type: 'pane.closed' }, + { type: 'pane.moved' }, + { type: 'workspace.closed' }, + { type: 'pane.agent_status_changed', pane_id: 'w1:p1' }, + { type: 'pane.agent_status_changed', pane_id: 'w1:p2' }, + ]); + assert.equal(sent[0].to, '#agent-status'); + assert.match(sent[0].text, /w1\/w1:p2 codex is working/); + assert.match(sent[0].idempotencyKey, /^herdr-status-/); + + const result = await actions[0].handler({ input: {}, agent: { handle: 'viewer' } }); + assert.equal(result.agents, 2); + assert.equal(sent[1].to, '@viewer'); + await eventually(async () => { + const contents = await readFile(join(stateDir, 'relay-state.json'), 'utf8'); + assert.match(contents, /at_live_bridge/); + }); + const permissions = (await stat(join(stateDir, 'relay-state.json'))).mode & 0o777; + assert.equal(permissions, 0o600); + + bridge.stop(); + await new Promise((resolve) => server.close(resolve)); + await unlink(socketPath).catch(() => {}); + }); +}); + +test('reconnects from the persisted bridge token without registering again', async () => { + await withTemporaryDirectory(async (directory) => { + const configDir = join(directory, 'config'); + const stateDir = join(directory, 'state'); + const socketPath = join(directory, 'herdr.sock'); + await writeConfig(configDir); + await mkdir(stateDir, { recursive: true }); + await writeFile(join(stateDir, 'relay-state.json'), JSON.stringify({ apiToken: 'at_live_saved', transitions: {} })); + const { server } = await fakeHerdrServer(socketPath, { dynamicPane: false }); + const reconnects = []; + const client = { registerAction: () => ({ unregister() {} }), sendMessage: async () => {} }; + class ReconnectRelay { + constructor() { + this.workspace = { + register: async () => { + throw new Error('registration must not rotate a persisted token'); + }, + reconnect: async ({ apiToken }) => { + reconnects.push(apiToken); + return client; + }, + }; + } + } + + const bridge = await startBridge({ + environment: { HERDR_PLUGIN_CONFIG_DIR: configDir, HERDR_PLUGIN_STATE_DIR: stateDir, HERDR_SOCKET_PATH: socketPath }, + AgentRelayCtor: ReconnectRelay, + logger: { warn() {} }, + }); + assert.deepEqual(reconnects, ['at_live_saved']); + assert.equal(bridgeName('/tmp/herdr.sock'), bridgeName('/tmp/herdr.sock')); + assert.notEqual(bridgeName('/tmp/herdr.sock'), bridgeName('/tmp/other.sock')); + bridge.stop(); + await new Promise((resolve) => server.close(resolve)); + await unlink(socketPath).catch(() => {}); + }); +}); + +test('resubscribes after its Herdr subscription connection closes', async () => { + await withTemporaryDirectory(async (directory) => { + const configDir = join(directory, 'config'); + const stateDir = join(directory, 'state'); + const socketPath = join(directory, 'herdr.sock'); + await writeConfig(configDir); + const { requests, server } = await fakeHerdrServer(socketPath, { + dynamicPane: false, + disconnectFirstSubscription: true, + }); + const client = { + token: 'at_live_bridge', + registerAction: () => ({ unregister() {} }), + sendMessage: async () => {}, + }; + class ReconnectSocketRelay { + constructor() { + this.workspace = { register: async () => client }; + } + } + + const bridge = await startBridge({ + environment: { HERDR_PLUGIN_CONFIG_DIR: configDir, HERDR_PLUGIN_STATE_DIR: stateDir, HERDR_SOCKET_PATH: socketPath }, + AgentRelayCtor: ReconnectSocketRelay, + logger: { warn() {} }, + }); + await eventually(() => assert.equal(requests.filter((request) => request.method === 'events.subscribe').length, 2), 150); + bridge.stop(); + await new Promise((resolve) => server.close(resolve)); + await unlink(socketPath).catch(() => {}); + }); +}); From c6dde068f7db483871049b27db7a1ed1341eb0eb Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 23 Jul 2026 15:08:36 +0200 Subject: [PATCH 2/5] fix: harden agent relay bridge runtime --- plugins/agent-relay/.gitignore | 1 + plugins/agent-relay/README.md | 28 +++- plugins/agent-relay/dist/bridge.mjs | 169 +++++++++++++++++------ plugins/agent-relay/dist/config.mjs | 44 +++++- plugins/agent-relay/dist/state.mjs | 83 ++++++++++- plugins/agent-relay/test/bridge.test.mjs | 110 +++++++++++++-- 6 files changed, 366 insertions(+), 69 deletions(-) create mode 100644 plugins/agent-relay/.gitignore diff --git a/plugins/agent-relay/.gitignore b/plugins/agent-relay/.gitignore new file mode 100644 index 0000000000..c2658d7d1b --- /dev/null +++ b/plugins/agent-relay/.gitignore @@ -0,0 +1 @@ +node_modules/ diff --git a/plugins/agent-relay/README.md b/plugins/agent-relay/README.md index 1d1b487ce9..e65f4b96ba 100644 --- a/plugins/agent-relay/README.md +++ b/plugins/agent-relay/README.md @@ -19,6 +19,12 @@ herdr plugin link /path/to/agent-relay Herdr runs the same `npm ci --omit=dev` command automatically for GitHub installs. The plugin requires Node 22 or newer. +Install the plugin from this public fork with: + +```bash +herdr plugin install AgentWorkforce/herdr/plugins/agent-relay +``` + Copy `config.example.json` to the directory reported by: ```bash @@ -26,7 +32,16 @@ herdr plugin config-dir agent-relay.herdr-bridge ``` Name the copied file `agent-relay.json`, replace `workspaceKey`, and list only -the Herdr workspace IDs that may be forwarded. Then open the sidecar: +the Herdr workspace IDs that may be forwarded. The configured Relay channel +must already exist; the bridge joins it before forwarding any status. Keep the +workspace key private: + +```bash +chmod 600 "$(herdr plugin config-dir agent-relay.herdr-bridge)/agent-relay.json" +``` + +Custom Relay endpoints must use HTTPS, except for loopback addresses used in +local development. Then open the sidecar: ```bash herdr plugin pane open --plugin agent-relay.herdr-bridge --entrypoint bridge @@ -37,6 +52,11 @@ Closing that pane stops the bridge. It does not run from a startup hook. ## State and safety The Relay agent token and status-transition dedupe state are stored only in -`HERDR_PLUGIN_STATE_DIR` as a mode-0600 file. Subsequent starts reconnect with -that token rather than registering another Relay agent. The configuration file -belongs in `HERDR_PLUGIN_CONFIG_DIR`; do not commit a real workspace key. +`HERDR_PLUGIN_STATE_DIR`. The state file is mode 0600 on POSIX; on Windows it +inherits the account-scoped ACL of Herdr's plugin state directory. Subsequent +starts reconnect with that token rather than registering another Relay agent. +An exclusive state-directory lock prevents two bridge panes from racing to +rotate that token. If the bridge process crashes, verify that it has stopped +before removing the reported stale `relay-bridge.lock` file. +The configuration file belongs in `HERDR_PLUGIN_CONFIG_DIR`; do not commit a +real workspace key. diff --git a/plugins/agent-relay/dist/bridge.mjs b/plugins/agent-relay/dist/bridge.mjs index 24db74b1cb..98267d984d 100644 --- a/plugins/agent-relay/dist/bridge.mjs +++ b/plugins/agent-relay/dist/bridge.mjs @@ -6,9 +6,24 @@ import { z } from 'zod'; import { loadBridgeConfig, pluginPaths } from './config.mjs'; import { requestHerdr, subscribeToBridgeEvents } from './herdr-socket.mjs'; -import { loadBridgeState, saveBridgeState } from './state.mjs'; +import { acquireBridgeLock, loadBridgeState, saveBridgeState } from './state.mjs'; const SUMMARY_INPUT = z.object({}).strict(); +const SUMMARY_OUTPUT = z + .object({ + workspaceIds: z.array(z.string()), + agents: z.number().int().nonnegative(), + statuses: z + .object({ + idle: z.number().int().nonnegative(), + working: z.number().int().nonnegative(), + blocked: z.number().int().nonnegative(), + done: z.number().int().nonnegative(), + unknown: z.number().int().nonnegative(), + }) + .strict(), + }) + .strict(); const STATUS_VALUES = new Set(['idle', 'working', 'blocked', 'done', 'unknown']); function hash(value) { @@ -73,24 +88,41 @@ export function prepareTransition(state, event) { state.transitions[event.paneId] = { fingerprint, sequence }; return { ...event, + previousTransition: previous, + transitionFingerprint: fingerprint, + transitionSequence: sequence, idempotencyKey: `herdr-status-${hash(`${event.paneId}\u0000${sequence}\u0000${fingerprint}`)}`, }; } +export function rollbackTransition(state, transition) { + const current = state.transitions[transition.paneId]; + if ( + current?.fingerprint !== transition.transitionFingerprint || + current.sequence !== transition.transitionSequence + ) { + return; + } + if (transition.previousTransition) state.transitions[transition.paneId] = transition.previousTransition; + else delete state.transitions[transition.paneId]; +} + export function formatStatusMessage(event) { return `Herdr agent status: ${event.workspaceId}/${event.paneId} ${event.agent} is ${event.status}.`; } function stateWriter(stateDir, state) { let pending = Promise.resolve(); - return () => { + const write = () => { pending = pending.catch(() => undefined).then(() => saveBridgeState(stateDir, state)); return pending; }; + write.flush = () => pending; + return write; } function isPaneLifecycleEvent(message) { - return ['pane.created', 'pane.closed', 'pane.moved', 'workspace.closed'].includes(message?.event); + return ['pane_created', 'pane_closed', 'pane_moved', 'workspace_closed'].includes(message?.event); } function createSubscriptionManager({ socketPath, workspaceIds, onStatus, logger }) { @@ -155,16 +187,19 @@ async function connectRelay(config, state, socketPath, AgentRelayCtor) { const options = { workspaceKey: config.workspaceKey }; if (config.baseUrl) options.baseUrl = config.baseUrl; const relay = new AgentRelayCtor(options); + let agent; if (state.apiToken) { - return { agent: await relay.workspace.reconnect({ apiToken: state.apiToken }) }; + agent = await relay.workspace.reconnect({ apiToken: state.apiToken }); + } else { + agent = await relay.workspace.register({ + name: bridgeName(socketPath), + type: 'agent', + metadata: { integration: 'herdr' }, + }); + if (!agent.token) throw new Error('Agent Relay registration did not return an agent token'); + state.apiToken = agent.token; } - const agent = await relay.workspace.register({ - name: bridgeName(socketPath), - type: 'agent', - metadata: { integration: 'herdr' }, - }); - if (!agent.token) throw new Error('Agent Relay registration did not return an agent token'); - state.apiToken = agent.token; + await agent.channels.join(config.channel); return { agent }; } @@ -173,6 +208,7 @@ export function registerSessionSummaryAction(agent, socketPath, workspaceIds) { name: 'herdr.session_summary', description: 'Return aggregate Herdr agent status counts for the configured workspaces.', input: SUMMARY_INPUT, + output: SUMMARY_OUTPUT, handler: async ({ agent: caller }) => { const snapshot = sessionSnapshotFrom(await requestHerdr(socketPath, 'session.snapshot')); const summary = summarizeSnapshot(snapshot, workspaceIds); @@ -192,48 +228,91 @@ export async function startBridge({ const socketPath = environment.HERDR_SOCKET_PATH; if (!socketPath) throw new Error('Herdr did not provide HERDR_SOCKET_PATH'); - const config = await loadBridgeConfig(configDir); - const workspaceIds = new Set(config.workspaceIds); - const state = await loadBridgeState(stateDir); - const persistState = stateWriter(stateDir, state); - - await requestHerdr(socketPath, 'ping'); - - const { agent } = await connectRelay(config, state, socketPath, AgentRelayCtor); - await persistState(); - const action = registerSessionSummaryAction(agent, socketPath, config.workspaceIds); - const subscriptions = createSubscriptionManager({ - socketPath, - workspaceIds, - logger, - onStatus(message) { - const event = normalizeStatusEvent(message, workspaceIds); - if (!event) return; - const transition = prepareTransition(state, event); - if (!transition) return; - void agent - .sendMessage({ - to: config.channel, - text: formatStatusMessage(transition), - idempotencyKey: transition.idempotencyKey, - }) - .then(() => persistState()) - .catch(() => logger.warn('Agent Relay bridge could not forward a Herdr status update')); - }, - }); - await subscriptions.start(); + const lock = await acquireBridgeLock(stateDir); + let action; + let subscriptions; + let persistState; + const deliveries = new Set(); + const cleanup = async () => { + const errors = []; + for (const step of [ + () => action?.unregister(), + () => subscriptions?.stop(), + () => Promise.allSettled(deliveries), + () => persistState?.flush(), + () => lock.release(), + ]) { + try { + await step(); + } catch (error) { + errors.push(error); + } + } + if (errors.length) throw new AggregateError(errors, 'Agent Relay bridge cleanup failed'); + }; + try { + const config = await loadBridgeConfig(configDir); + const workspaceIds = new Set(config.workspaceIds); + const state = await loadBridgeState(stateDir); + persistState = stateWriter(stateDir, state); + + await requestHerdr(socketPath, 'ping'); + + const { agent } = await connectRelay(config, state, socketPath, AgentRelayCtor); + await persistState(); + action = registerSessionSummaryAction(agent, socketPath, config.workspaceIds); + subscriptions = createSubscriptionManager({ + socketPath, + workspaceIds, + logger, + onStatus(message) { + const event = normalizeStatusEvent(message, workspaceIds); + if (!event) return; + const transition = prepareTransition(state, event); + if (!transition) return; + const delivery = agent + .sendMessage({ + to: config.channel, + text: formatStatusMessage(transition), + idempotencyKey: transition.idempotencyKey, + }) + .then( + () => + persistState().catch(() => + logger.warn('Agent Relay bridge forwarded a status update but could not persist its dedupe state') + ), + () => { + rollbackTransition(state, transition); + logger.warn('Agent Relay bridge could not forward a Herdr status update'); + } + ); + deliveries.add(delivery); + void delivery.finally(() => deliveries.delete(delivery)); + }, + }); + await subscriptions.start(); + } catch (error) { + await cleanup().catch(() => logger.warn('Agent Relay bridge cleanup failed during startup')); + throw error; + } + let stopped = false; return { - stop() { - action.unregister(); - subscriptions.stop(); + async stop() { + if (stopped) return; + stopped = true; + await cleanup(); }, }; } async function main() { const bridge = await startBridge(); - const stop = () => bridge.stop(); + const stop = () => + void bridge.stop().catch((error) => { + console.error(`Agent Relay bridge failed to stop cleanly: ${error.message}`); + process.exitCode = 1; + }); process.once('SIGINT', stop); process.once('SIGTERM', stop); } diff --git a/plugins/agent-relay/dist/config.mjs b/plugins/agent-relay/dist/config.mjs index c570117fda..b089854dd9 100644 --- a/plugins/agent-relay/dist/config.mjs +++ b/plugins/agent-relay/dist/config.mjs @@ -1,4 +1,4 @@ -import { readFile } from 'node:fs/promises'; +import { readFile, stat } from 'node:fs/promises'; import { join } from 'node:path'; import { z } from 'zod'; @@ -7,10 +7,32 @@ const channelName = z.string().trim().regex(/^#[A-Za-z0-9][A-Za-z0-9_-]*$/, { message: 'channel must be a #channel-name', }); +const relayBaseUrl = z + .string() + .url() + .superRefine((value, context) => { + let url; + try { + url = new URL(value); + } catch { + return; + } + const loopback = ['localhost', '127.0.0.1', '[::1]'].includes(url.hostname); + if (!['http:', 'https:'].includes(url.protocol) || (url.protocol !== 'https:' && !loopback)) { + context.addIssue({ + code: 'custom', + message: 'baseUrl must use HTTPS (HTTP is allowed only for a loopback address)', + }); + } + if (url.username || url.password) { + context.addIssue({ code: 'custom', message: 'baseUrl must not contain credentials' }); + } + }); + export const BridgeConfigSchema = z .object({ workspaceKey: z.string().trim().min(1), - baseUrl: z.string().url().optional(), + baseUrl: relayBaseUrl.optional(), channel: channelName, workspaceIds: z .array(z.string().trim().min(1)) @@ -32,10 +54,26 @@ export function configPath(configDir) { return join(configDir, 'agent-relay.json'); } +async function assertPrivateConfigFile(path) { + const metadata = await stat(path); + if (!metadata.isFile()) throw new Error('Agent Relay bridge configuration must be a regular file'); + if (process.platform !== 'win32' && (metadata.mode & 0o077) !== 0) { + throw new Error('Agent Relay bridge configuration must not be accessible by group or other users'); + } +} + export async function loadBridgeConfig(configDir) { + const path = configPath(configDir); + try { + await assertPrivateConfigFile(path); + } catch (error) { + if (!error?.code) throw error; + throw new Error(`Cannot read Agent Relay bridge configuration at ${path}`); + } + let parsed; try { - parsed = JSON.parse(await readFile(configPath(configDir), 'utf8')); + parsed = JSON.parse(await readFile(path, 'utf8')); } catch (error) { if (error instanceof SyntaxError) { throw new Error('Agent Relay bridge configuration is not valid JSON'); diff --git a/plugins/agent-relay/dist/state.mjs b/plugins/agent-relay/dist/state.mjs index 9e17148980..2a48767882 100644 --- a/plugins/agent-relay/dist/state.mjs +++ b/plugins/agent-relay/dist/state.mjs @@ -1,4 +1,5 @@ -import { chmod, mkdir, readFile, rename, writeFile } from 'node:fs/promises'; +import { randomUUID } from 'node:crypto'; +import { chmod, mkdir, open, readFile, rename, stat, unlink, writeFile } from 'node:fs/promises'; import { join } from 'node:path'; import { z } from 'zod'; @@ -19,16 +20,90 @@ export function statePath(stateDir) { return join(stateDir, 'relay-state.json'); } +export function lockPath(stateDir) { + return join(stateDir, 'relay-bridge.lock'); +} + export function emptyBridgeState() { return { transitions: {} }; } +async function prepareStateDirectory(stateDir) { + await mkdir(stateDir, { recursive: true, mode: 0o700 }); + if (process.platform !== 'win32') await chmod(stateDir, 0o700); +} + +async function assertPrivateStateFile(path) { + const metadata = await stat(path); + if (!metadata.isFile()) throw new Error('Agent Relay bridge state must be a regular file'); + if (process.platform !== 'win32' && (metadata.mode & 0o077) !== 0) { + throw new Error('Agent Relay bridge state must not be accessible by group or other users'); + } +} + +export async function acquireBridgeLock(stateDir) { + await prepareStateDirectory(stateDir); + const target = lockPath(stateDir); + const nonce = randomUUID(); + let handle; + try { + handle = await open(target, 'wx', 0o600); + await handle.writeFile(`${JSON.stringify({ pid: process.pid, nonce })}\n`); + await handle.sync(); + } catch (error) { + const created = Boolean(handle); + await handle?.close().catch(() => undefined); + if (error?.code !== 'EEXIST') { + if (created) await unlink(target).catch(() => undefined); + throw error; + } + let owner; + try { + owner = JSON.parse(await readFile(target, 'utf8')); + } catch { + owner = undefined; + } + const suffix = Number.isInteger(owner?.pid) ? ` (PID ${owner.pid})` : ''; + throw new Error( + `Another Agent Relay bridge holds ${target}${suffix}; remove the lock only after verifying that bridge is stopped` + ); + } + + let released = false; + return { + async release() { + if (released) return; + released = true; + await handle.close(); + let owner; + try { + owner = JSON.parse(await readFile(target, 'utf8')); + } catch { + return; + } + if (owner?.nonce === nonce) { + await unlink(target).catch((error) => { + if (error?.code !== 'ENOENT') throw error; + }); + } + }, + }; +} + export async function loadBridgeState(stateDir) { + const target = statePath(stateDir); + try { + await assertPrivateStateFile(target); + } catch (error) { + if (error?.code === 'ENOENT') return emptyBridgeState(); + if (!error?.code) throw error; + throw new Error('Cannot read Agent Relay bridge state'); + } + let parsed; try { - parsed = JSON.parse(await readFile(statePath(stateDir), 'utf8')); + parsed = JSON.parse(await readFile(target, 'utf8')); } catch (error) { - if (error && error.code === 'ENOENT') return emptyBridgeState(); if (error instanceof SyntaxError) { throw new Error('Agent Relay bridge state is not valid JSON'); } @@ -41,7 +116,7 @@ export async function loadBridgeState(stateDir) { } export async function saveBridgeState(stateDir, state) { - await mkdir(stateDir, { recursive: true, mode: 0o700 }); + await prepareStateDirectory(stateDir); const target = statePath(stateDir); const temporary = `${target}.tmp-${process.pid}`; await writeFile(temporary, `${JSON.stringify(state)}\n`, { mode: 0o600 }); diff --git a/plugins/agent-relay/test/bridge.test.mjs b/plugins/agent-relay/test/bridge.test.mjs index 030b840b08..ef0a62d9b3 100644 --- a/plugins/agent-relay/test/bridge.test.mjs +++ b/plugins/agent-relay/test/bridge.test.mjs @@ -1,5 +1,5 @@ import assert from 'node:assert/strict'; -import { mkdir, mkdtemp, readFile, rm, stat, unlink, writeFile } from 'node:fs/promises'; +import { chmod, mkdir, mkdtemp, readFile, rm, stat, unlink, writeFile } from 'node:fs/promises'; import net from 'node:net'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; @@ -9,11 +9,13 @@ import { bridgeName, normalizeStatusEvent, prepareTransition, + rollbackTransition, startBridge, summarizeSnapshot, } from '../dist/bridge.mjs'; -import { loadBridgeConfig, redactConfig } from '../dist/config.mjs'; +import { BridgeConfigSchema, loadBridgeConfig, redactConfig } from '../dist/config.mjs'; import { localSocketTarget } from '../dist/herdr-socket.mjs'; +import { acquireBridgeLock, loadBridgeState, lockPath } from '../dist/state.mjs'; const snapshot = { agents: [ @@ -78,7 +80,7 @@ async function fakeHerdrServer(socketPath, { dynamicPane = true, disconnectFirst }; socket.write( `${JSON.stringify({ - event: 'pane.created', + event: 'pane_created', data: { pane: { pane_id: 'w1:p2', workspace_id: 'w1' } }, })}\n` ); @@ -99,7 +101,7 @@ async function fakeHerdrServer(socketPath, { dynamicPane = true, disconnectFirst } }); }); - await new Promise((resolve) => server.listen(socketPath, resolve)); + await new Promise((resolve) => server.listen(localSocketTarget(socketPath), resolve)); return { requests, server }; } @@ -107,8 +109,10 @@ async function writeConfig(configDir) { await mkdir(configDir, { recursive: true }); await writeFile( join(configDir, 'agent-relay.json'), - JSON.stringify({ workspaceKey: 'rk_live_secret', channel: '#agent-status', workspaceIds: ['w1'] }) + JSON.stringify({ workspaceKey: 'rk_live_secret', channel: '#agent-status', workspaceIds: ['w1'] }), + { mode: 0o600 } ); + if (process.platform !== 'win32') await chmod(join(configDir, 'agent-relay.json'), 0o600); } test('validates configuration and redacts workspace credentials', async () => { @@ -117,6 +121,56 @@ test('validates configuration and redacts workspace credentials', async () => { const config = await loadBridgeConfig(directory); assert.equal(redactConfig(config).workspaceKey, '[redacted]'); assert.equal(config.channel, '#agent-status'); + assert.equal( + BridgeConfigSchema.safeParse({ + workspaceKey: 'rk_live_secret', + baseUrl: 'http://relay.example.com', + channel: '#agent-status', + workspaceIds: ['w1'], + }).success, + false + ); + assert.equal( + BridgeConfigSchema.safeParse({ + workspaceKey: 'rk_live_secret', + baseUrl: 'http://127.0.0.1:3000', + channel: '#agent-status', + workspaceIds: ['w1'], + }).success, + true + ); + assert.equal( + BridgeConfigSchema.safeParse({ + workspaceKey: 'rk_live_secret', + baseUrl: 'not-a-url', + channel: '#agent-status', + workspaceIds: ['w1'], + }).success, + false + ); + }); +}); + +test('rejects exposed credential files and duplicate bridge processes', async () => { + await withTemporaryDirectory(async (directory) => { + const configDir = join(directory, 'config'); + const stateDir = join(directory, 'state'); + await writeConfig(configDir); + if (process.platform !== 'win32') { + await chmod(join(configDir, 'agent-relay.json'), 0o644); + await assert.rejects(loadBridgeConfig(configDir), /must not be accessible/); + await chmod(join(configDir, 'agent-relay.json'), 0o600); + + await mkdir(stateDir, { recursive: true }); + await writeFile(join(stateDir, 'relay-state.json'), JSON.stringify({ transitions: {} }), { mode: 0o644 }); + await assert.rejects(loadBridgeState(stateDir), /must not be accessible/); + await chmod(join(stateDir, 'relay-state.json'), 0o600); + } + + const first = await acquireBridgeLock(stateDir); + await assert.rejects(acquireBridgeLock(stateDir), /Another Agent Relay bridge holds/); + await first.release(); + await assert.rejects(stat(lockPath(stateDir)), { code: 'ENOENT' }); }); }); @@ -136,6 +190,20 @@ test('deduplicates identical status events while preserving later repeated state assert.equal(state.transitions['w1:p1'].sequence, 3); }); +test('rolls back a failed transition so the same status can be retried', () => { + const state = { transitions: {} }; + const event = { paneId: 'w1:p1', workspaceId: 'w1', agent: 'codex', status: 'working' }; + const transition = prepareTransition(state, event); + rollbackTransition(state, transition); + assert.deepEqual(state.transitions, {}); + assert.equal(prepareTransition(state, event).idempotencyKey, transition.idempotencyKey); + + const superseded = prepareTransition(state, { ...event, status: 'blocked' }); + rollbackTransition(state, transition); + assert.equal(state.transitions['w1:p1'].fingerprint, superseded.transitionFingerprint); + assert.equal(state.transitions['w1:p1'].sequence, superseded.transitionSequence); +}); + test('filters status events and summarizes only configured workspaces', () => { const allowed = new Set(['w1']); assert.equal( @@ -164,6 +232,11 @@ test('rebuilds per-pane subscriptions after a pane is created and forwards its s let registrations = 0; const client = { token: 'at_live_bridge', + channels: { + async join(channel) { + assert.equal(channel, '#agent-status'); + }, + }, registerAction(definition) { actions.push(definition); return { unregister() {} }; @@ -214,17 +287,21 @@ test('rebuilds per-pane subscriptions after a pane is created and forwards its s assert.match(sent[0].text, /w1\/w1:p2 codex is working/); assert.match(sent[0].idempotencyKey, /^herdr-status-/); - const result = await actions[0].handler({ input: {}, agent: { handle: 'viewer' } }); + const result = await actions[0].handler({ input: {}, agent: { name: 'viewer' } }); assert.equal(result.agents, 2); + assert.equal(actions[0].input.safeParse({ unexpected: true }).success, false); + assert.equal(actions[0].output.safeParse(result).success, true); assert.equal(sent[1].to, '@viewer'); await eventually(async () => { const contents = await readFile(join(stateDir, 'relay-state.json'), 'utf8'); assert.match(contents, /at_live_bridge/); }); - const permissions = (await stat(join(stateDir, 'relay-state.json'))).mode & 0o777; - assert.equal(permissions, 0o600); + if (process.platform !== 'win32') { + const permissions = (await stat(join(stateDir, 'relay-state.json'))).mode & 0o777; + assert.equal(permissions, 0o600); + } - bridge.stop(); + await bridge.stop(); await new Promise((resolve) => server.close(resolve)); await unlink(socketPath).catch(() => {}); }); @@ -237,10 +314,16 @@ test('reconnects from the persisted bridge token without registering again', asy const socketPath = join(directory, 'herdr.sock'); await writeConfig(configDir); await mkdir(stateDir, { recursive: true }); - await writeFile(join(stateDir, 'relay-state.json'), JSON.stringify({ apiToken: 'at_live_saved', transitions: {} })); + await writeFile(join(stateDir, 'relay-state.json'), JSON.stringify({ apiToken: 'at_live_saved', transitions: {} }), { + mode: 0o600, + }); const { server } = await fakeHerdrServer(socketPath, { dynamicPane: false }); const reconnects = []; - const client = { registerAction: () => ({ unregister() {} }), sendMessage: async () => {} }; + const client = { + channels: { join: async () => {} }, + registerAction: () => ({ unregister() {} }), + sendMessage: async () => {}, + }; class ReconnectRelay { constructor() { this.workspace = { @@ -263,7 +346,7 @@ test('reconnects from the persisted bridge token without registering again', asy assert.deepEqual(reconnects, ['at_live_saved']); assert.equal(bridgeName('/tmp/herdr.sock'), bridgeName('/tmp/herdr.sock')); assert.notEqual(bridgeName('/tmp/herdr.sock'), bridgeName('/tmp/other.sock')); - bridge.stop(); + await bridge.stop(); await new Promise((resolve) => server.close(resolve)); await unlink(socketPath).catch(() => {}); }); @@ -281,6 +364,7 @@ test('resubscribes after its Herdr subscription connection closes', async () => }); const client = { token: 'at_live_bridge', + channels: { join: async () => {} }, registerAction: () => ({ unregister() {} }), sendMessage: async () => {}, }; @@ -296,7 +380,7 @@ test('resubscribes after its Herdr subscription connection closes', async () => logger: { warn() {} }, }); await eventually(() => assert.equal(requests.filter((request) => request.method === 'events.subscribe').length, 2), 150); - bridge.stop(); + await bridge.stop(); await new Promise((resolve) => server.close(resolve)); await unlink(socketPath).catch(() => {}); }); From a25404a0cafb94c5ac6c958c2dea1376ab4b77d4 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 23 Jul 2026 15:20:19 +0200 Subject: [PATCH 3/5] fix: avoid lifecycle replay loops --- plugins/agent-relay/README.md | 2 + plugins/agent-relay/dist/bridge.mjs | 45 ++++++++++++----------- plugins/agent-relay/dist/herdr-socket.mjs | 11 ++---- plugins/agent-relay/test/bridge.test.mjs | 31 ++++++++-------- 4 files changed, 44 insertions(+), 45 deletions(-) diff --git a/plugins/agent-relay/README.md b/plugins/agent-relay/README.md index e65f4b96ba..ec21362b50 100644 --- a/plugins/agent-relay/README.md +++ b/plugins/agent-relay/README.md @@ -6,6 +6,8 @@ channel and exposes a read-only `herdr.session_summary` Relay action. The bridge never reads pane output, cwd, environment, or terminal titles. It does not expose prompt, key-send, shell, or raw-socket controls. +It discovers pane membership from periodic session snapshots rather than +replaying Herdr's retained lifecycle-event history. ## Setup diff --git a/plugins/agent-relay/dist/bridge.mjs b/plugins/agent-relay/dist/bridge.mjs index 98267d984d..460d176d26 100644 --- a/plugins/agent-relay/dist/bridge.mjs +++ b/plugins/agent-relay/dist/bridge.mjs @@ -121,23 +121,22 @@ function stateWriter(stateDir, state) { return write; } -function isPaneLifecycleEvent(message) { - return ['pane_created', 'pane_closed', 'pane_moved', 'workspace_closed'].includes(message?.event); -} - function createSubscriptionManager({ socketPath, workspaceIds, onStatus, logger }) { let active; + let activePaneKey; let stopped = false; let pending = Promise.resolve(); - let reconnectTimer; + let refreshTimer; - const scheduleRebuild = () => { - if (stopped || reconnectTimer) return; - reconnectTimer = setTimeout(() => { - reconnectTimer = undefined; + // Generic Herdr lifecycle subscriptions replay retained events from sequence + // zero. Snapshot polling avoids treating that history as a fresh pane change. + const scheduleRefresh = () => { + if (stopped || refreshTimer) return; + refreshTimer = setTimeout(() => { + refreshTimer = undefined; void rebuild().catch(() => { - logger.warn('Agent Relay bridge could not reconnect to Herdr'); - scheduleRebuild(); + logger.warn('Agent Relay bridge could not refresh its Herdr subscriptions'); + scheduleRefresh(); }); }, 1_000); }; @@ -148,25 +147,27 @@ function createSubscriptionManager({ socketPath, workspaceIds, onStatus, logger const snapshot = sessionSnapshotFrom(await requestHerdr(socketPath, 'session.snapshot')); const paneIds = snapshot.panes .filter((pane) => workspaceIds.has(pane.workspace_id)) - .map((pane) => pane.pane_id); - const next = await subscribeToBridgeEvents(socketPath, paneIds, (message) => { - if (isPaneLifecycleEvent(message)) { - void rebuild().catch(() => scheduleRebuild()); - return; - } - onStatus(message); - }); + .map((pane) => pane.pane_id) + .sort(); + const paneKey = paneIds.join('\u0000'); + if (active && !active.closed && paneKey === activePaneKey) { + scheduleRefresh(); + return; + } + const next = await subscribeToBridgeEvents(socketPath, paneIds, onStatus); if (stopped) { next.close(); return; } const previous = active; active = next; + activePaneKey = paneKey; next.once('closed', () => { - if (active === next && !stopped) scheduleRebuild(); + if (active === next && !stopped) scheduleRefresh(); }); - if (next.closed) scheduleRebuild(); + if (next.closed) scheduleRefresh(); previous?.close(); + scheduleRefresh(); }); return pending; }; @@ -177,7 +178,7 @@ function createSubscriptionManager({ socketPath, workspaceIds, onStatus, logger }, stop() { stopped = true; - if (reconnectTimer) clearTimeout(reconnectTimer); + if (refreshTimer) clearTimeout(refreshTimer); active?.close(); }, }; diff --git a/plugins/agent-relay/dist/herdr-socket.mjs b/plugins/agent-relay/dist/herdr-socket.mjs index e40ea46602..348a11f12e 100644 --- a/plugins/agent-relay/dist/herdr-socket.mjs +++ b/plugins/agent-relay/dist/herdr-socket.mjs @@ -115,13 +115,10 @@ export async function subscribeToBridgeEvents(socketPath, paneIds, onEvent) { client.on('event', onEvent); try { await client.request('events.subscribe', { - subscriptions: [ - { type: 'pane.created' }, - { type: 'pane.closed' }, - { type: 'pane.moved' }, - { type: 'workspace.closed' }, - ...[...new Set(paneIds)].map((pane_id) => ({ type: 'pane.agent_status_changed', pane_id })), - ], + subscriptions: [...new Set(paneIds)].map((pane_id) => ({ + type: 'pane.agent_status_changed', + pane_id, + })), }); return client; } catch (error) { diff --git a/plugins/agent-relay/test/bridge.test.mjs b/plugins/agent-relay/test/bridge.test.mjs index ef0a62d9b3..42bc439c8b 100644 --- a/plugins/agent-relay/test/bridge.test.mjs +++ b/plugins/agent-relay/test/bridge.test.mjs @@ -72,18 +72,20 @@ async function fakeHerdrServer(socketPath, { dynamicPane = true, disconnectFirst } else if (request.method === 'events.subscribe') { socket.write(`${JSON.stringify({ id: request.id, result: { type: 'subscribed' } })}\n`); subscriptions += 1; + queueMicrotask(() => { + socket.write( + `${JSON.stringify({ + event: 'pane_created', + data: { pane: { pane_id: 'historical:pane', workspace_id: 'w1' } }, + })}\n` + ); + }); if (dynamicPane && subscriptions === 1) { queueMicrotask(() => { currentSnapshot = { ...currentSnapshot, panes: [...currentSnapshot.panes, { pane_id: 'w1:p2', workspace_id: 'w1' }], }; - socket.write( - `${JSON.stringify({ - event: 'pane_created', - data: { pane: { pane_id: 'w1:p2', workspace_id: 'w1' } }, - })}\n` - ); }); } else if (dynamicPane && subscriptions === 2) { queueMicrotask(() => { @@ -220,7 +222,7 @@ test('filters status events and summarizes only configured workspaces', () => { }); }); -test('rebuilds per-pane subscriptions after a pane is created and forwards its status', async () => { +test('refreshes per-pane subscriptions without replaying historical lifecycle events', async () => { await withTemporaryDirectory(async (directory) => { const configDir = join(directory, 'config'); const stateDir = join(directory, 'state'); @@ -265,27 +267,24 @@ test('rebuilds per-pane subscriptions after a pane is created and forwards its s AgentRelayCtor: FakeRelay, logger: { warn() {} }, }); - await eventually(() => assert.equal(requests.filter((request) => request.method === 'events.subscribe').length, 2)); + await eventually( + () => assert.equal(requests.filter((request) => request.method === 'events.subscribe').length, 2), + 150 + ); await eventually(() => assert.equal(sent.length, 1)); assert.equal(registrations, 1); assert.deepEqual(requests.filter((request) => request.method === 'events.subscribe')[0].params.subscriptions, [ - { type: 'pane.created' }, - { type: 'pane.closed' }, - { type: 'pane.moved' }, - { type: 'workspace.closed' }, { type: 'pane.agent_status_changed', pane_id: 'w1:p1' }, ]); assert.deepEqual(requests.filter((request) => request.method === 'events.subscribe')[1].params.subscriptions, [ - { type: 'pane.created' }, - { type: 'pane.closed' }, - { type: 'pane.moved' }, - { type: 'workspace.closed' }, { type: 'pane.agent_status_changed', pane_id: 'w1:p1' }, { type: 'pane.agent_status_changed', pane_id: 'w1:p2' }, ]); assert.equal(sent[0].to, '#agent-status'); assert.match(sent[0].text, /w1\/w1:p2 codex is working/); assert.match(sent[0].idempotencyKey, /^herdr-status-/); + await new Promise((resolve) => setTimeout(resolve, 1_100)); + assert.equal(requests.filter((request) => request.method === 'events.subscribe').length, 2); const result = await actions[0].handler({ input: {}, agent: { name: 'viewer' } }); assert.equal(result.agents, 2); From b033ea3da241a01fc052e97fe8c700e848a81045 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 23 Jul 2026 15:23:00 +0200 Subject: [PATCH 4/5] test: scope lifecycle replay fixture --- plugins/agent-relay/test/bridge.test.mjs | 25 ++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/plugins/agent-relay/test/bridge.test.mjs b/plugins/agent-relay/test/bridge.test.mjs index 42bc439c8b..e8b9b8afbe 100644 --- a/plugins/agent-relay/test/bridge.test.mjs +++ b/plugins/agent-relay/test/bridge.test.mjs @@ -51,7 +51,10 @@ async function withTemporaryDirectory(run) { } } -async function fakeHerdrServer(socketPath, { dynamicPane = true, disconnectFirstSubscription = false } = {}) { +async function fakeHerdrServer( + socketPath, + { dynamicPane = true, disconnectFirstSubscription = false, replayLifecycle = false } = {} +) { const requests = []; let currentSnapshot = structuredClone(snapshot); let subscriptions = 0; @@ -72,14 +75,16 @@ async function fakeHerdrServer(socketPath, { dynamicPane = true, disconnectFirst } else if (request.method === 'events.subscribe') { socket.write(`${JSON.stringify({ id: request.id, result: { type: 'subscribed' } })}\n`); subscriptions += 1; - queueMicrotask(() => { - socket.write( - `${JSON.stringify({ - event: 'pane_created', - data: { pane: { pane_id: 'historical:pane', workspace_id: 'w1' } }, - })}\n` - ); - }); + if (replayLifecycle) { + queueMicrotask(() => { + socket.write( + `${JSON.stringify({ + event: 'pane_created', + data: { pane: { pane_id: 'historical:pane', workspace_id: 'w1' } }, + })}\n` + ); + }); + } if (dynamicPane && subscriptions === 1) { queueMicrotask(() => { currentSnapshot = { @@ -228,7 +233,7 @@ test('refreshes per-pane subscriptions without replaying historical lifecycle ev const stateDir = join(directory, 'state'); const socketPath = join(directory, 'herdr.sock'); await writeConfig(configDir); - const { server, requests } = await fakeHerdrServer(socketPath); + const { server, requests } = await fakeHerdrServer(socketPath, { replayLifecycle: true }); const sent = []; const actions = []; let registrations = 0; From c9d80761495514cdaf1b1c2fb0b84007a61a0d27 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Thu, 23 Jul 2026 18:45:25 +0200 Subject: [PATCH 5/5] fix(plugin): make relay bridge restarts reliable --- plugins/agent-relay/README.md | 4 +- plugins/agent-relay/dist/bridge.mjs | 47 ++++++++++++++---------- plugins/agent-relay/dist/state.mjs | 13 +++++++ plugins/agent-relay/test/bridge.test.mjs | 43 +++++++++++++++++++++- 4 files changed, 83 insertions(+), 24 deletions(-) diff --git a/plugins/agent-relay/README.md b/plugins/agent-relay/README.md index ec21362b50..693fa0c370 100644 --- a/plugins/agent-relay/README.md +++ b/plugins/agent-relay/README.md @@ -58,7 +58,7 @@ The Relay agent token and status-transition dedupe state are stored only in inherits the account-scoped ACL of Herdr's plugin state directory. Subsequent starts reconnect with that token rather than registering another Relay agent. An exclusive state-directory lock prevents two bridge panes from racing to -rotate that token. If the bridge process crashes, verify that it has stopped -before removing the reported stale `relay-bridge.lock` file. +rotate that token. A later start removes the lock automatically when its owner +PID is no longer running; live or unidentifiable lock owners still fail closed. The configuration file belongs in `HERDR_PLUGIN_CONFIG_DIR`; do not commit a real workspace key. diff --git a/plugins/agent-relay/dist/bridge.mjs b/plugins/agent-relay/dist/bridge.mjs index 460d176d26..38862a672a 100644 --- a/plugins/agent-relay/dist/bridge.mjs +++ b/plugins/agent-relay/dist/bridge.mjs @@ -55,11 +55,6 @@ export function summarizeSnapshot(snapshot, workspaceIds) { return { workspaceIds: [...allowed], agents, statuses }; } -export function formatSessionSummary(summary) { - const { statuses } = summary; - return `Herdr summary: ${summary.agents} agents across ${summary.workspaceIds.length} workspaces — working ${statuses.working}, blocked ${statuses.blocked}, idle ${statuses.idle}, done ${statuses.done}, unknown ${statuses.unknown}.`; -} - export function normalizeStatusEvent(message, workspaceIds) { const data = message?.data; if ( @@ -210,16 +205,19 @@ export function registerSessionSummaryAction(agent, socketPath, workspaceIds) { description: 'Return aggregate Herdr agent status counts for the configured workspaces.', input: SUMMARY_INPUT, output: SUMMARY_OUTPUT, - handler: async ({ agent: caller }) => { + handler: async () => { const snapshot = sessionSnapshotFrom(await requestHerdr(socketPath, 'session.snapshot')); - const summary = summarizeSnapshot(snapshot, workspaceIds); - const recipient = caller.handle ?? caller.name; - await agent.sendMessage({ to: `@${recipient}`, text: formatSessionSummary(summary) }); - return summary; + return summarizeSnapshot(snapshot, workspaceIds); }, }); } +export function installStopHandlers(target, stop) { + target.once('SIGINT', stop); + target.once('SIGTERM', stop); + if (target.platform !== 'win32') target.once('SIGHUP', stop); +} + export async function startBridge({ environment = process.env, AgentRelayCtor = AgentRelay, @@ -237,11 +235,13 @@ export async function startBridge({ const cleanup = async () => { const errors = []; for (const step of [ - () => action?.unregister(), () => subscriptions?.stop(), + // Herdr escalates pane shutdown quickly after SIGHUP. Release the local + // singleton lock before any remote unregister or delivery can delay it. + () => lock.release(), + () => action?.unregister(), () => Promise.allSettled(deliveries), () => persistState?.flush(), - () => lock.release(), ]) { try { await step(); @@ -308,14 +308,21 @@ export async function startBridge({ } async function main() { - const bridge = await startBridge(); - const stop = () => - void bridge.stop().catch((error) => { - console.error(`Agent Relay bridge failed to stop cleanly: ${error.message}`); - process.exitCode = 1; - }); - process.once('SIGINT', stop); - process.once('SIGTERM', stop); + const bridge = startBridge(); + let stopping = false; + const stop = () => { + if (stopping) return; + stopping = true; + void bridge.then((active) => active.stop()).then( + () => process.exit(0), + (error) => { + console.error(`Agent Relay bridge failed to stop cleanly: ${error.message}`); + process.exit(1); + } + ); + }; + installStopHandlers(process, stop); + await bridge; } if (import.meta.url === `file://${process.argv[1]}`) { diff --git a/plugins/agent-relay/dist/state.mjs b/plugins/agent-relay/dist/state.mjs index 2a48767882..e898c20ec5 100644 --- a/plugins/agent-relay/dist/state.mjs +++ b/plugins/agent-relay/dist/state.mjs @@ -41,6 +41,15 @@ async function assertPrivateStateFile(path) { } } +function processIsAlive(pid) { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return error?.code === 'EPERM'; + } +} + export async function acquireBridgeLock(stateDir) { await prepareStateDirectory(stateDir); const target = lockPath(stateDir); @@ -63,6 +72,10 @@ export async function acquireBridgeLock(stateDir) { } catch { owner = undefined; } + if (Number.isInteger(owner?.pid) && !processIsAlive(owner.pid)) { + await unlink(target); + return acquireBridgeLock(stateDir); + } const suffix = Number.isInteger(owner?.pid) ? ` (PID ${owner.pid})` : ''; throw new Error( `Another Agent Relay bridge holds ${target}${suffix}; remove the lock only after verifying that bridge is stopped` diff --git a/plugins/agent-relay/test/bridge.test.mjs b/plugins/agent-relay/test/bridge.test.mjs index e8b9b8afbe..c0054079a2 100644 --- a/plugins/agent-relay/test/bridge.test.mjs +++ b/plugins/agent-relay/test/bridge.test.mjs @@ -7,6 +7,7 @@ import test from 'node:test'; import { bridgeName, + installStopHandlers, normalizeStatusEvent, prepareTransition, rollbackTransition, @@ -29,6 +30,25 @@ const snapshot = { ], }; +test('installs the pane hangup stop handler on Unix', () => { + const installedSignals = (platform) => { + const signals = []; + installStopHandlers( + { + platform, + once(signal) { + signals.push(signal); + }, + }, + () => {} + ); + return signals; + }; + + assert.deepEqual(installedSignals('darwin'), ['SIGINT', 'SIGTERM', 'SIGHUP']); + assert.deepEqual(installedSignals('win32'), ['SIGINT', 'SIGTERM']); +}); + async function eventually(assertion, attempts = 50) { let lastError; for (let attempt = 0; attempt < attempts; attempt += 1) { @@ -181,6 +201,23 @@ test('rejects exposed credential files and duplicate bridge processes', async () }); }); +test('recovers a bridge lock whose owner process has stopped', async () => { + await withTemporaryDirectory(async (directory) => { + await mkdir(directory, { recursive: true }); + await writeFile( + lockPath(directory), + `${JSON.stringify({ pid: 2_147_483_647, nonce: 'stale' })}\n`, + { mode: 0o600 } + ); + + const lock = await acquireBridgeLock(directory); + const owner = JSON.parse(await readFile(lockPath(directory), 'utf8')); + assert.equal(owner.pid, process.pid); + await lock.release(); + await assert.rejects(stat(lockPath(directory)), { code: 'ENOENT' }); + }); +}); + test('uses the interprocess-compatible Windows pipe name', () => { assert.equal(localSocketTarget('C:\\Users\\me\\herdr.sock', 'win32'), '\\\\.\\pipe\\C:\\Users\\me\\herdr.sock'); assert.equal(localSocketTarget('\\\\.\\pipe\\herdr.sock', 'win32'), '\\\\.\\pipe\\herdr.sock'); @@ -291,11 +328,13 @@ test('refreshes per-pane subscriptions without replaying historical lifecycle ev await new Promise((resolve) => setTimeout(resolve, 1_100)); assert.equal(requests.filter((request) => request.method === 'events.subscribe').length, 2); - const result = await actions[0].handler({ input: {}, agent: { name: 'viewer' } }); + // Action results are returned through Relay's invocation output. Do not + // depend on the transport event's caller label being a routable DM handle. + const result = await actions[0].handler({ input: {}, agent: { name: 'node' } }); assert.equal(result.agents, 2); assert.equal(actions[0].input.safeParse({ unexpected: true }).success, false); assert.equal(actions[0].output.safeParse(result).success, true); - assert.equal(sent[1].to, '@viewer'); + assert.equal(sent.length, 1); await eventually(async () => { const contents = await readFile(join(stateDir, 'relay-state.json'), 'utf8'); assert.match(contents, /at_live_bridge/);