From 2718d75cd618155731f737d87641d88c1c7423a9 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Fri, 7 Aug 2026 12:42:16 +0200 Subject: [PATCH] chore: retire agent relay fork copy (cherry picked from commit 7b657a6d6aa2b170e73c290b5e8adb66191247b3) --- .github/workflows/agent-relay-plugin.yml | 50 --- plugins/agent-relay/.gitignore | 1 - plugins/agent-relay/README.md | 64 ---- plugins/agent-relay/config.example.json | 6 - plugins/agent-relay/dist/bridge.mjs | 333 ----------------- plugins/agent-relay/dist/config.mjs | 93 ----- plugins/agent-relay/dist/herdr-socket.mjs | 128 ------- plugins/agent-relay/dist/state.mjs | 139 ------- 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 | 430 ---------------------- 12 files changed, 1333 deletions(-) delete mode 100644 .github/workflows/agent-relay-plugin.yml delete mode 100644 plugins/agent-relay/.gitignore delete mode 100644 plugins/agent-relay/README.md delete mode 100644 plugins/agent-relay/config.example.json delete mode 100644 plugins/agent-relay/dist/bridge.mjs delete mode 100644 plugins/agent-relay/dist/config.mjs delete mode 100644 plugins/agent-relay/dist/herdr-socket.mjs delete mode 100644 plugins/agent-relay/dist/state.mjs delete mode 100644 plugins/agent-relay/herdr-plugin.toml delete mode 100644 plugins/agent-relay/package-lock.json delete mode 100644 plugins/agent-relay/package.json delete 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 deleted file mode 100644 index 0bafbfa071..0000000000 --- a/.github/workflows/agent-relay-plugin.yml +++ /dev/null @@ -1,50 +0,0 @@ -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/.gitignore b/plugins/agent-relay/.gitignore deleted file mode 100644 index c2658d7d1b..0000000000 --- a/plugins/agent-relay/.gitignore +++ /dev/null @@ -1 +0,0 @@ -node_modules/ diff --git a/plugins/agent-relay/README.md b/plugins/agent-relay/README.md deleted file mode 100644 index 693fa0c370..0000000000 --- a/plugins/agent-relay/README.md +++ /dev/null @@ -1,64 +0,0 @@ -# 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. -It discovers pane membership from periodic session snapshots rather than -replaying Herdr's retained lifecycle-event history. - -## 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. - -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 -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. 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 -``` - -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`. 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. 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/config.example.json b/plugins/agent-relay/config.example.json deleted file mode 100644 index b0a76b33b0..0000000000 --- a/plugins/agent-relay/config.example.json +++ /dev/null @@ -1,6 +0,0 @@ -{ - "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 deleted file mode 100644 index 38862a672a..0000000000 --- a/plugins/agent-relay/dist/bridge.mjs +++ /dev/null @@ -1,333 +0,0 @@ -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 { 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) { - 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 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, - 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(); - const write = () => { - pending = pending.catch(() => undefined).then(() => saveBridgeState(stateDir, state)); - return pending; - }; - write.flush = () => pending; - return write; -} - -function createSubscriptionManager({ socketPath, workspaceIds, onStatus, logger }) { - let active; - let activePaneKey; - let stopped = false; - let pending = Promise.resolve(); - let refreshTimer; - - // 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 refresh its Herdr subscriptions'); - scheduleRefresh(); - }); - }, 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) - .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) scheduleRefresh(); - }); - if (next.closed) scheduleRefresh(); - previous?.close(); - scheduleRefresh(); - }); - return pending; - }; - - return { - async start() { - await rebuild(); - }, - stop() { - stopped = true; - if (refreshTimer) clearTimeout(refreshTimer); - 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); - let agent; - if (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; - } - await agent.channels.join(config.channel); - 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, - output: SUMMARY_OUTPUT, - handler: async () => { - const snapshot = sessionSnapshotFrom(await requestHerdr(socketPath, 'session.snapshot')); - 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, - 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 lock = await acquireBridgeLock(stateDir); - let action; - let subscriptions; - let persistState; - const deliveries = new Set(); - const cleanup = async () => { - const errors = []; - for (const step of [ - () => 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(), - ]) { - 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 { - async stop() { - if (stopped) return; - stopped = true; - await cleanup(); - }, - }; -} - -async function main() { - 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]}`) { - 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 deleted file mode 100644 index b089854dd9..0000000000 --- a/plugins/agent-relay/dist/config.mjs +++ /dev/null @@ -1,93 +0,0 @@ -import { readFile, stat } 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', -}); - -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: relayBaseUrl.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'); -} - -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(path, '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 deleted file mode 100644 index 348a11f12e..0000000000 --- a/plugins/agent-relay/dist/herdr-socket.mjs +++ /dev/null @@ -1,128 +0,0 @@ -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: [...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 deleted file mode 100644 index e898c20ec5..0000000000 --- a/plugins/agent-relay/dist/state.mjs +++ /dev/null @@ -1,139 +0,0 @@ -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'; - -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 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'); - } -} - -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); - 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; - } - 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` - ); - } - - 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(target, 'utf8')); - } catch (error) { - 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 prepareStateDirectory(stateDir); - 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 deleted file mode 100644 index 3b8258d539..0000000000 --- a/plugins/agent-relay/herdr-plugin.toml +++ /dev/null @@ -1,15 +0,0 @@ -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 deleted file mode 100644 index 3ab2e6b248..0000000000 --- a/plugins/agent-relay/package-lock.json +++ /dev/null @@ -1,58 +0,0 @@ -{ - "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 deleted file mode 100644 index fb084a83d5..0000000000 --- a/plugins/agent-relay/package.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "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 deleted file mode 100644 index c0054079a2..0000000000 --- a/plugins/agent-relay/test/bridge.test.mjs +++ /dev/null @@ -1,430 +0,0 @@ -import assert from 'node:assert/strict'; -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'; -import test from 'node:test'; - -import { - bridgeName, - installStopHandlers, - normalizeStatusEvent, - prepareTransition, - rollbackTransition, - startBridge, - summarizeSnapshot, -} from '../dist/bridge.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: [ - { 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' }, - ], -}; - -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) { - 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, replayLifecycle = 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 (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 = { - ...currentSnapshot, - panes: [...currentSnapshot.panes, { pane_id: 'w1:p2', workspace_id: 'w1' }], - }; - }); - } 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(localSocketTarget(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'] }), - { mode: 0o600 } - ); - if (process.platform !== 'win32') await chmod(join(configDir, 'agent-relay.json'), 0o600); -} - -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'); - 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' }); - }); -}); - -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'); -}); - -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('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( - 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('refreshes per-pane subscriptions without replaying historical lifecycle events', 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, { replayLifecycle: true }); - const sent = []; - const actions = []; - 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() {} }; - }, - 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), - 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.agent_status_changed', pane_id: 'w1:p1' }, - ]); - assert.deepEqual(requests.filter((request) => request.method === 'events.subscribe')[1].params.subscriptions, [ - { 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); - - // 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.length, 1); - await eventually(async () => { - const contents = await readFile(join(stateDir, 'relay-state.json'), 'utf8'); - assert.match(contents, /at_live_bridge/); - }); - if (process.platform !== 'win32') { - const permissions = (await stat(join(stateDir, 'relay-state.json'))).mode & 0o777; - assert.equal(permissions, 0o600); - } - - await 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: {} }), { - mode: 0o600, - }); - const { server } = await fakeHerdrServer(socketPath, { dynamicPane: false }); - const reconnects = []; - const client = { - channels: { join: async () => {} }, - 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')); - await 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', - channels: { join: async () => {} }, - 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); - await bridge.stop(); - await new Promise((resolve) => server.close(resolve)); - await unlink(socketPath).catch(() => {}); - }); -});