From aa456c13436895076a467755a9fbdd1e33951f80 Mon Sep 17 00:00:00 2001 From: Dan Lynch Date: Fri, 25 Sep 2026 00:44:31 +0000 Subject: [PATCH 1/2] refactor(agentic): machine-runner is a remote control; Claude/Codex adapters move above it into agent-cli MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirrors constructive-io/constructive-db#3855. The runner enrolls, runs allow-listed commands in a pty or pipes, streams bytes, resizes, signals, detaches/reattaches — and knows nothing about agents. Claude/Codex adapters, the agent stdio contract and approval decisions live in the new @constructive-db/agent-cli (constructive-agent-cli). The protocol gains the bounded LineSplitter and approval-decision line parser. Runner/agent-cli end-to-end proofs need the relay and stay in constructive-db; unit tests ship here. --- .github/workflows/run-tests.yaml | 2 +- agentic/agent-cli/LICENSE | 21 ++ agentic/agent-cli/README.md | 39 +++ .../__tests__/adapters.test.ts} | 8 + .../__tests__/fixtures/bin/claude | 0 .../agent-cli/__tests__/fixtures/bin/codex | 20 ++ agentic/agent-cli/jest.config.js | 9 + agentic/agent-cli/package.json | 43 +++ .../src/adapters.ts} | 13 +- agentic/agent-cli/src/cli.ts | 81 ++++++ agentic/agent-cli/src/index.ts | 12 + agentic/agent-cli/src/run.ts | 42 +++ agentic/agent-cli/src/session.ts | 253 +++++++++++++++++ agentic/agent-cli/tsconfig.esm.json | 10 + agentic/agent-cli/tsconfig.json | 9 + agentic/machine-protocol/README.md | 22 +- .../machine-protocol/__tests__/codec.test.ts | 102 ++++--- agentic/machine-protocol/src/index.ts | 200 +++++++++---- agentic/machine-runner/README.md | 62 ++-- .../__tests__/fixtures/bin/codex | 14 - .../fixtures/bin/constructive-agent-host | 53 ---- agentic/machine-runner/src/cli-session.ts | 268 ------------------ agentic/machine-runner/src/config.ts | 19 -- .../machine-runner/src/headless-session.ts | 151 ---------- agentic/machine-runner/src/index.ts | 28 +- agentic/machine-runner/src/policy.ts | 60 ++-- agentic/machine-runner/src/process.ts | 115 ++++++++ agentic/machine-runner/src/runner.ts | 170 ++--------- pnpm-lock.yaml | 7 + 29 files changed, 983 insertions(+), 850 deletions(-) create mode 100644 agentic/agent-cli/LICENSE create mode 100644 agentic/agent-cli/README.md rename agentic/{machine-runner/__tests__/agent-cli.test.ts => agent-cli/__tests__/adapters.test.ts} (95%) rename agentic/{machine-runner => agent-cli}/__tests__/fixtures/bin/claude (100%) create mode 100755 agentic/agent-cli/__tests__/fixtures/bin/codex create mode 100644 agentic/agent-cli/jest.config.js create mode 100644 agentic/agent-cli/package.json rename agentic/{machine-runner/src/agent-cli.ts => agent-cli/src/adapters.ts} (95%) create mode 100644 agentic/agent-cli/src/cli.ts create mode 100644 agentic/agent-cli/src/index.ts create mode 100644 agentic/agent-cli/src/run.ts create mode 100644 agentic/agent-cli/src/session.ts create mode 100644 agentic/agent-cli/tsconfig.esm.json create mode 100644 agentic/agent-cli/tsconfig.json delete mode 100755 agentic/machine-runner/__tests__/fixtures/bin/codex delete mode 100755 agentic/machine-runner/__tests__/fixtures/bin/constructive-agent-host delete mode 100644 agentic/machine-runner/src/cli-session.ts delete mode 100644 agentic/machine-runner/src/headless-session.ts create mode 100644 agentic/machine-runner/src/process.ts diff --git a/.github/workflows/run-tests.yaml b/.github/workflows/run-tests.yaml index 0ccc696b2a..eda7905997 100644 --- a/.github/workflows/run-tests.yaml +++ b/.github/workflows/run-tests.yaml @@ -115,7 +115,7 @@ jobs: - batch: graphile-unit packages: 'graphile/graphile-plugin-utils graphile/graphile-realtime-subscriptions graphile/graphile-sql-expression-validator graphile/graphile-upload-plugin graphile/graphile-storage-registry' - batch: agentic - packages: 'agentic/protocol agentic/agentic-kit agentic/agent agentic/harness agentic/chat agentic/cli agentic/db-tools agentic/pi agentic/dsh agentic/react agentic/agentic-server agentic/anthropic agentic/openai agentic/ollama agentic/run-log agentic/metering agentic/agent-conversation agentic/pi-host agentic/run-log-client agentic/run-log-gate agentic/machine-protocol agentic/machine-runner' + packages: 'agentic/protocol agentic/agentic-kit agentic/agent agentic/harness agentic/chat agentic/cli agentic/db-tools agentic/pi agentic/dsh agentic/react agentic/agentic-server agentic/anthropic agentic/openai agentic/ollama agentic/run-log agentic/metering agentic/agent-conversation agentic/pi-host agentic/run-log-client agentic/run-log-gate agentic/machine-protocol agentic/machine-runner agentic/agent-cli' - batch: pgpm-unit packages: 'pgpm/types pgpm/naming-spec pgpm/diff pgpm/import pgpm/slice pgpm/transform' - batch: pglite diff --git a/agentic/agent-cli/LICENSE b/agentic/agent-cli/LICENSE new file mode 100644 index 0000000000..d4ce902d22 --- /dev/null +++ b/agentic/agent-cli/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Interweb, Inc. + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/agentic/agent-cli/README.md b/agentic/agent-cli/README.md new file mode 100644 index 0000000000..6ae43bc1f1 --- /dev/null +++ b/agentic/agent-cli/README.md @@ -0,0 +1,39 @@ +# @constructive-db/agent-cli + +`constructive-agent-cli`: a coding-agent CLI (`claude`, `codex`) adapted to the +machine protocol's **agent stdio contract**, so that a machine session bound to +an agent run can run one as an ordinary command. + +``` +constructive-agent-cli [--resume ] + [--approval-timeout-ms ] [--on-timeout deny|allow] [-- ] +``` + +- **stdin** — the prompt, one line; then further prompts (turns, where the CLI + supports them) or `{"kind":"approval_decision",...}` JSON lines answering a + tool approval the CLI asked for. +- **stdout** — `AgentEvent` JSON lines, nothing else. +- **stderr** — everything the CLI says that is not its protocol. +- **exit** — the CLI's own status. + +`--approval-timeout-ms` and `--on-timeout` default from +`CONSTRUCTIVE_AGENT_CLI_APPROVAL_TIMEOUT_MS` / `CONSTRUCTIVE_AGENT_CLI_ON_TIMEOUT`, +so a machine owner sets them once, in the runner policy's `env.set`. + +## Where it sits + +The **machine runner** (`@constructive-db/machine-runner`) is a remote control: +it runs the commands its policy allows, on pipes or in a pty, and streams bytes. +It knows nothing about agents. This program is what gives a session its agent +shape — the relay asks the runner to run `constructive-agent-cli claude` for a +`cli`-bound session, reads the events off its stdout, puts an approval on the +run's log, and writes the decision back as a stdin line. The runner reads +neither direction. + +For this program to run on a machine, its policy must allow the command +(`"allowedCommands": ["constructive-agent-cli", ...]`) and the CLI itself must +be on the `PATH` that policy projects. The runner never runs `claude` or +`codex` directly. + +The adapters (`ClaudeCodeAdapter`, `CodexExecAdapter`) and the session loop +(`runAgentCliSession`) are exported for use in-process. diff --git a/agentic/machine-runner/__tests__/agent-cli.test.ts b/agentic/agent-cli/__tests__/adapters.test.ts similarity index 95% rename from agentic/machine-runner/__tests__/agent-cli.test.ts rename to agentic/agent-cli/__tests__/adapters.test.ts index 116d5e52a1..866b075575 100644 --- a/agentic/machine-runner/__tests__/agent-cli.test.ts +++ b/agentic/agent-cli/__tests__/adapters.test.ts @@ -168,8 +168,16 @@ describe('CodexExecAdapter', () => { '--json', 'resume', '01a0a38b-442f-7f91-a250-72f305c839f7', + '--', 'hello' ] }); }); + + it('hands the prompt over as a positional, never as an option', () => { + expect(codex.spawnArgs('--dangerously-bypass-approvals-and-sandbox rm -rf /')).toEqual({ + command: 'codex', + args: ['exec', '--json', '--', '--dangerously-bypass-approvals-and-sandbox rm -rf /'] + }); + }); }); diff --git a/agentic/machine-runner/__tests__/fixtures/bin/claude b/agentic/agent-cli/__tests__/fixtures/bin/claude similarity index 100% rename from agentic/machine-runner/__tests__/fixtures/bin/claude rename to agentic/agent-cli/__tests__/fixtures/bin/claude diff --git a/agentic/agent-cli/__tests__/fixtures/bin/codex b/agentic/agent-cli/__tests__/fixtures/bin/codex new file mode 100755 index 0000000000..3681047b01 --- /dev/null +++ b/agentic/agent-cli/__tests__/fixtures/bin/codex @@ -0,0 +1,20 @@ +#!/usr/bin/env node + +const args = process.argv.slice(2); +const resumeIndex = args.indexOf('resume'); +const sessionId = resumeIndex >= 0 ? args[resumeIndex + 1] : '01a0a38b-442f-7f91-a250-72f305c839f7'; +// Like the real CLI: the prompt is the positional after `--`. +const separator = args.indexOf('--'); +if (separator < 0 || separator === args.length - 1) { + process.stderr.write('codex fixture: prompt must follow --\n'); + process.exit(2); +} +const prompt = args[separator + 1]; +const emit = value => process.stdout.write(`${JSON.stringify(value)}\n`); + +emit({ type: 'thread.started', thread_id: sessionId }); +emit({ + type: 'item.completed', + item: { id: 'msg_fake', type: 'agent_message', text: `echo:${prompt}` } +}); +emit({ type: 'turn.completed', usage: {} }); diff --git a/agentic/agent-cli/jest.config.js b/agentic/agent-cli/jest.config.js new file mode 100644 index 0000000000..42c5ef9ac0 --- /dev/null +++ b/agentic/agent-cli/jest.config.js @@ -0,0 +1,9 @@ +/** @type {import('ts-jest').JestConfigWithTsJest} */ +module.exports = { + preset: 'ts-jest', + testEnvironment: 'node', + testMatch: ['**/__tests__/**/*.test.ts'], + // A built package carries a copy of its own manifest, which jest's module map + // reads as a second package of the same name. + modulePathIgnorePatterns: ['/dist/'] +}; diff --git a/agentic/agent-cli/package.json b/agentic/agent-cli/package.json new file mode 100644 index 0000000000..a0c810c775 --- /dev/null +++ b/agentic/agent-cli/package.json @@ -0,0 +1,43 @@ +{ + "name": "@constructive-db/agent-cli", + "version": "0.1.0", + "description": "constructive-agent-cli: a coding-agent CLI (claude, codex) adapted to the machine protocol's agent stdio contract — prompts and approval decisions in, AgentEvent JSON lines out. Runs on the user's own machine as an ordinary command under the machine runner, which knows nothing about it.", + "author": "Constructive ", + "license": "MIT", + "homepage": "https://github.com/constructive-io/constructive", + "repository": { + "type": "git", + "url": "https://github.com/constructive-io/constructive", + "directory": "agentic/agent-cli" + }, + "bugs": { + "url": "https://github.com/constructive-io/constructive/issues" + }, + "keywords": [ + "agent", + "claude", + "codex", + "remote-control" + ], + "main": "index.js", + "module": "esm/index.js", + "types": "index.d.ts", + "bin": { + "constructive-agent-cli": "run.js" + }, + "publishConfig": { + "access": "public", + "directory": "dist" + }, + "scripts": { + "build": "makage build && cp LICENSE dist/LICENSE", + "build:dev": "makage build --dev && cp LICENSE dist/LICENSE", + "clean": "makage clean", + "lint": "eslint . --fix", + "start": "node dist/run.js", + "test": "jest" + }, + "dependencies": { + "@constructive-db/machine-protocol": "workspace:*" + } +} diff --git a/agentic/machine-runner/src/agent-cli.ts b/agentic/agent-cli/src/adapters.ts similarity index 95% rename from agentic/machine-runner/src/agent-cli.ts rename to agentic/agent-cli/src/adapters.ts index f9c5830fec..bd90786910 100644 --- a/agentic/machine-runner/src/agent-cli.ts +++ b/agentic/agent-cli/src/adapters.ts @@ -177,7 +177,16 @@ export class CodexExecAdapter implements AgentCliAdapter { spawnArgs(prompt: string, resume?: string): AgentCliSpawn { return { command: 'codex', - args: ['exec', '--json', ...this.extraArgs, ...(resume ? ['resume', resume] : []), prompt] + // `--` ends option parsing, so a prompt is a prompt even when it starts + // with a dash. + args: [ + 'exec', + '--json', + ...this.extraArgs, + ...(resume ? ['resume', resume] : []), + '--', + prompt + ] }; } @@ -248,6 +257,6 @@ export function adapterForCommand(command: string, extraArgs: string[]): AgentCl case 'codex': return new CodexExecAdapter(extraArgs); default: - throw new Error(`machine-runner: no CLI adapter for command '${command}'`); + throw new Error(`agent-cli: no CLI adapter for command '${command}'`); } } diff --git a/agentic/agent-cli/src/cli.ts b/agentic/agent-cli/src/cli.ts new file mode 100644 index 0000000000..1e60c8e8b4 --- /dev/null +++ b/agentic/agent-cli/src/cli.ts @@ -0,0 +1,81 @@ +// The command line: `constructive-agent-cli [--resume ] +// [--approval-timeout-ms ] [--on-timeout deny|allow] [-- ]`. +// Which CLI is the one thing the caller must say; everything after `--` is +// handed to that CLI untouched. The approval flags default from the +// environment (`CONSTRUCTIVE_AGENT_CLI_APPROVAL_TIMEOUT_MS`, +// `CONSTRUCTIVE_AGENT_CLI_ON_TIMEOUT`): how long a machine waits on a question +// is the machine owner's setting, projected into the command by their policy. + +import type { ApprovalDecision } from '@constructive-db/machine-protocol'; + +export interface AgentCliArgs { + cli: 'claude' | 'codex'; + resume?: string; + approvalTimeoutMs?: number; + onTimeout?: ApprovalDecision; + extraArgs: string[]; +} + +export const USAGE = + 'usage: constructive-agent-cli [--resume ] ' + + '[--approval-timeout-ms ] [--on-timeout deny|allow] [-- ]'; + +export class UsageError extends Error { + constructor(message: string) { + super(`${message}\n${USAGE}`); + this.name = 'UsageError'; + } +} + +export const APPROVAL_TIMEOUT_ENV = 'CONSTRUCTIVE_AGENT_CLI_APPROVAL_TIMEOUT_MS'; +export const ON_TIMEOUT_ENV = 'CONSTRUCTIVE_AGENT_CLI_ON_TIMEOUT'; + +export function parseArgs( + argv: readonly string[], + env: Readonly> = {} +): AgentCliArgs { + const cli = argv[0]; + if (cli !== 'claude' && cli !== 'codex') { + throw new UsageError(cli === undefined ? 'a CLI name is required' : `unknown CLI '${cli}'`); + } + const values: Partial> = {}; + let extraArgs: string[] = []; + for (let i = 1; i < argv.length; i++) { + const arg = argv[i]; + if (arg === '--') { + extraArgs = argv.slice(i + 1); + break; + } + const eq = arg.indexOf('='); + const flag = arg.startsWith('--') ? arg.slice(2, eq === -1 ? undefined : eq) : undefined; + if (flag !== 'resume' && flag !== 'approval-timeout-ms' && flag !== 'on-timeout') { + throw new UsageError(`unexpected argument '${arg}'`); + } + const value = eq === -1 ? argv[++i] : arg.slice(eq + 1); + if (value === undefined || value.length === 0 || (eq === -1 && value.startsWith('--'))) { + throw new UsageError(`--${flag} needs a value`); + } + if (values[flag] !== undefined) throw new UsageError(`--${flag} given twice`); + values[flag] = value; + } + const args: AgentCliArgs = { cli, extraArgs }; + if (values.resume !== undefined) args.resume = values.resume; + const timeout = values['approval-timeout-ms'] ?? env[APPROVAL_TIMEOUT_ENV]; + if (timeout !== undefined) { + const ms = Number(timeout); + if (!Number.isFinite(ms) || ms <= 0) { + throw new UsageError( + `--approval-timeout-ms (or ${APPROVAL_TIMEOUT_ENV}) must be a positive number of milliseconds` + ); + } + args.approvalTimeoutMs = ms; + } + const onTimeout = values['on-timeout'] ?? env[ON_TIMEOUT_ENV]; + if (onTimeout !== undefined) { + if (onTimeout !== 'deny' && onTimeout !== 'allow') { + throw new UsageError(`--on-timeout (or ${ON_TIMEOUT_ENV}) must be 'deny' or 'allow'`); + } + args.onTimeout = onTimeout; + } + return args; +} diff --git a/agentic/agent-cli/src/index.ts b/agentic/agent-cli/src/index.ts new file mode 100644 index 0000000000..7d09fc8014 --- /dev/null +++ b/agentic/agent-cli/src/index.ts @@ -0,0 +1,12 @@ +export type { AgentCliAdapter, AgentCliSpawn } from './adapters'; +export { adapterForCommand, ClaudeCodeAdapter, CodexExecAdapter } from './adapters'; +export type { AgentCliArgs } from './cli'; +export { APPROVAL_TIMEOUT_ENV, ON_TIMEOUT_ENV, parseArgs, USAGE, UsageError } from './cli'; +export type { AgentCliExit, AgentCliIo, AgentCliSessionOptions } from './session'; +export { + APPROVAL_AUTO_DENY_REASON, + APPROVAL_EXIT_REASON, + APPROVAL_TIMEOUT_REASON, + DEFAULT_APPROVAL_TIMEOUT_MS, + runAgentCliSession +} from './session'; diff --git a/agentic/agent-cli/src/run.ts b/agentic/agent-cli/src/run.ts new file mode 100644 index 0000000000..7c6cc249dc --- /dev/null +++ b/agentic/agent-cli/src/run.ts @@ -0,0 +1,42 @@ +#!/usr/bin/env node +// `constructive-agent-cli`: a coding-agent CLI on the agent stdio contract. +// SIGTERM/SIGINT stop the CLI; its own exit status is this program's. + +import { adapterForCommand } from './adapters'; +import { parseArgs, UsageError } from './cli'; +import { runAgentCliSession } from './session'; + +async function main(): Promise { + const args = parseArgs(process.argv.slice(2), process.env); + const stop = new AbortController(); + process.once('SIGTERM', () => stop.abort()); + process.once('SIGINT', () => stop.abort()); + const exit = await runAgentCliSession({ + adapter: adapterForCommand(args.cli, args.extraArgs), + resume: args.resume, + approvalTimeoutMs: args.approvalTimeoutMs, + onTimeout: args.onTimeout, + io: { stdin: process.stdin, stdout: process.stdout, stderr: process.stderr }, + abort: stop.signal + }); + if (exit.signal) { + process.kill(process.pid, exit.signal); + return 128; + } + return exit.exitCode; +} + +// The CLI's exit ends this program even while the caller still holds its +// stdin open (a one-shot `codex exec` under a session that has not been +// closed); stdout is flushed before the exit takes effect. +const exit = (code: number): void => { + process.stdout.write('', () => process.exit(code)); +}; + +main().then(exit, (err: unknown) => { + const message = err instanceof Error ? err.message : String(err); + const prefixed = err instanceof UsageError || message.startsWith('agent-cli: '); + process.stderr.write(`${prefixed ? '' : 'agent-cli: '}${message}\n`, () => { + exit(err instanceof UsageError ? 2 : 1); + }); +}); diff --git a/agentic/agent-cli/src/session.ts b/agentic/agent-cli/src/session.ts new file mode 100644 index 0000000000..444712cccb --- /dev/null +++ b/agentic/agent-cli/src/session.ts @@ -0,0 +1,253 @@ +// One coding-agent CLI session, adapted to the machine protocol's agent stdio +// contract. This is the program the relay opens — on pipes, through a machine +// runner that treats it as any other command — when a session is bound to a +// run in `cli` mode: +// +// stdin the prompt, one line; then further prompts (turns) or +// `approval_decision` JSON lines +// stdout `AgentEvent` JSON lines, nothing else +// stderr everything the CLI says that is not its protocol +// exit the CLI's own status +// +// The CLI itself (`claude`, `codex`) is spawned from PATH: this program runs on +// the user's own machine, next to the CLI they installed, and is itself the +// command the machine's policy allows. + +import type { AgentEvent, ApprovalDecision } from '@constructive-db/machine-protocol'; +import { LineSplitter, parseApprovalDecisionLine } from '@constructive-db/machine-protocol'; +import { spawn as spawnChild } from 'child_process'; +import type { Readable, Writable } from 'stream'; + +import type { AgentCliAdapter } from './adapters'; + +export const APPROVAL_AUTO_DENY_REASON = + 'approval round-trip is not available in this session; denied automatically'; +export const APPROVAL_TIMEOUT_REASON = 'no decision arrived before the approval timeout'; +export const APPROVAL_EXIT_REASON = 'the session ended while the approval was pending'; + +/** Five minutes: long enough for a human to look, short enough to notice. */ +export const DEFAULT_APPROVAL_TIMEOUT_MS = 5 * 60_000; + +export interface AgentCliIo { + stdin: Readable; + stdout: Writable; + stderr: Writable; +} + +export interface AgentCliSessionOptions { + adapter: AgentCliAdapter; + /** The CLI's own session to continue, if any. */ + resume?: string; + io: AgentCliIo; + /** How long a tool approval waits for a decision line before it is settled here. */ + approvalTimeoutMs?: number; + /** What a timed-out approval is settled as. */ + onTimeout?: ApprovalDecision; + /** Where the CLI runs; defaults to this process's directory. */ + cwd?: string; + env?: NodeJS.ProcessEnv; + /** Stop: the CLI is sent SIGTERM and the session resolves with its exit. */ + abort?: AbortSignal; +} + +export interface AgentCliExit { + exitCode: number; + signal?: NodeJS.Signals; +} + +/** + * Run the session to the CLI's exit. Resolves with the CLI's exit, rejects when + * the CLI could not be started or broke its own protocol. + */ +export function runAgentCliSession(options: AgentCliSessionOptions): Promise { + const { adapter, io } = options; + const timeoutMs = options.approvalTimeoutMs ?? DEFAULT_APPROVAL_TIMEOUT_MS; + const onTimeout = options.onTimeout ?? 'deny'; + + return new Promise((resolve, reject) => { + let child: ReturnType | null = null; + let cliStdinEnded = false; + let exited = false; + const pending = new Map(); + const cliLines = new LineSplitter(); + const inputLines = new LineSplitter(); + + const event = (value: AgentEvent): void => { + io.stdout.write(`${JSON.stringify(value)}\n`); + }; + const note = (message: string): void => { + io.stderr.write(`agent-cli: ${message}\n`); + }; + const fail = (err: Error): void => { + if (exited) return; + exited = true; + for (const timer of pending.values()) clearTimeout(timer); + pending.clear(); + child?.kill(); + reject(err); + }; + + /** + * Answer the CLI and record the answer as an `approval_resolved` event, + * whoever decided. False when nothing was pending under that id: a second + * answer, or one arriving after the timeout spoke, is not written into the + * CLI as a second control response. + */ + const settle = (requestId: string, decision: ApprovalDecision, reason?: string): boolean => { + const timer = pending.get(requestId); + if (timer === undefined) return false; + clearTimeout(timer); + pending.delete(requestId); + if (!child || !adapter.encodeApproval) { + throw new Error(`agent-cli: ${adapter.name} approval '${requestId}' cannot be answered`); + } + if (cliStdinEnded || exited) { + note(`${adapter.name} approval '${requestId}' ${decision} arrived after the CLI's stdin closed`); + } else { + child.stdin.write(adapter.encodeApproval(requestId, decision, reason)); + } + event({ + kind: 'approval_resolved', + requestId, + decision, + ...(reason === undefined ? {} : { reason }) + }); + return true; + }; + const settleAll = (reason: string): void => { + for (const requestId of [...pending.keys()]) settle(requestId, 'deny', reason); + }; + + const onEvent = (value: AgentEvent): void => { + // The request is an event like any other — it is what the ledger and a + // watching client see. Only its answer travels the other way. + event(value); + if (value.kind !== 'approval_requested') return; + if (!adapter.encodeApproval) { + // The adapter can surface the question but has no way to answer it; + // say so in the record rather than leaving a request open forever. + event({ + kind: 'approval_resolved', + requestId: value.requestId, + decision: 'deny', + reason: APPROVAL_AUTO_DENY_REASON + }); + return; + } + if (pending.has(value.requestId)) { + throw new Error(`agent-cli: ${adapter.name} asked for approval '${value.requestId}' twice`); + } + const timer = setTimeout(() => { + settle(value.requestId, onTimeout, APPROVAL_TIMEOUT_REASON); + }, timeoutMs); + pending.set(value.requestId, timer); + }; + + const onCliLine = (line: string): void => { + // Only the parse is guarded: a line that is not the adapter's protocol is + // the CLI talking, and goes to stderr. What a parsed event then does is + // not guarded, so a fault there (a duplicated approval id) fails the run. + let events: AgentEvent[] | null; + try { + events = adapter.parseEvent(line); + } catch { + io.stderr.write(`${line}\n`); + return; + } + if (events) for (const value of events) onEvent(value); + }; + + const start = (prompt: string): void => { + const spawn = adapter.spawnArgs(prompt, options.resume); + child = spawnChild(spawn.command, spawn.args, { + cwd: options.cwd, + env: options.env ?? process.env, + stdio: ['pipe', 'pipe', 'pipe'] + }); + child.stdout.setEncoding('utf8').on('data', (data: string) => { + try { + for (const line of cliLines.push(data)) onCliLine(line); + } catch (err) { + fail(err instanceof Error ? err : new Error(String(err))); + } + }); + child.stderr.setEncoding('utf8').on('data', (data: string) => io.stderr.write(data)); + child.on('error', err => fail(new Error(`agent-cli: ${spawn.command} failed: ${err.message}`, { cause: err }))); + child.on('exit', (code, signal) => { + if (exited) return; + try { + for (const line of cliLines.flush()) onCliLine(line); + settleAll(APPROVAL_EXIT_REASON); + } catch (err) { + fail(err instanceof Error ? err : new Error(String(err))); + return; + } + exited = true; + resolve({ exitCode: code ?? -1, ...(signal ? { signal } : {}) }); + }); + // The CLI closing its stdin early (a one-shot `codex exec`) makes the + // next write EPIPE. Its exit is the report; the write that missed is noted. + child.stdin.on('error', err => { + if (!exited) note(`${spawn.command} stdin: ${err.message}`); + }); + if (spawn.stdinPrompt !== undefined) child.stdin.write(spawn.stdinPrompt); + if (!adapter.encodeTurn) { + child.stdin.end(); + cliStdinEnded = true; + } + }; + + const onInputLine = (line: string): void => { + if (line.length === 0) return; + const decision = parseApprovalDecisionLine(line); + if (decision) { + if (!settle(decision.requestId, decision.decision, decision.reason)) { + note(`no pending approval '${decision.requestId}'; the decision arrived after it was settled`); + } + return; + } + if (!child) { + start(line); + return; + } + if (adapter.encodeTurn && !cliStdinEnded) { + child.stdin.write(adapter.encodeTurn(line)); + return; + } + note(`${adapter.name} takes a single prompt; open a new session with cliSessionId to continue`); + }; + + options.abort?.addEventListener('abort', () => { + if (exited) return; + if (child) { + child.kill('SIGTERM'); + return; + } + exited = true; + resolve({ exitCode: -1, signal: 'SIGTERM' }); + }); + + io.stdin.setEncoding('utf8'); + io.stdin.on('data', (data: string) => { + try { + for (const line of inputLines.push(data)) onInputLine(line); + } catch (err) { + fail(err instanceof Error ? err : new Error(String(err))); + } + }); + io.stdin.on('end', () => { + try { + for (const line of inputLines.flush()) onInputLine(line); + } catch (err) { + fail(err instanceof Error ? err : new Error(String(err))); + return; + } + // No prompt ever came: there is nothing to run. + if (!child) { + exited = true; + resolve({ exitCode: 0 }); + } + }); + io.stdin.on('error', err => fail(new Error(`agent-cli: stdin failed: ${err.message}`, { cause: err }))); + }); +} diff --git a/agentic/agent-cli/tsconfig.esm.json b/agentic/agent-cli/tsconfig.esm.json new file mode 100644 index 0000000000..92d7545ebb --- /dev/null +++ b/agentic/agent-cli/tsconfig.esm.json @@ -0,0 +1,10 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "moduleResolution": "bundler", + "outDir": "dist/esm", + "module": "es2022", + "rootDir": "src/", + "declaration": false + } +} diff --git a/agentic/agent-cli/tsconfig.json b/agentic/agent-cli/tsconfig.json new file mode 100644 index 0000000000..1a9d5696cb --- /dev/null +++ b/agentic/agent-cli/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.json", + "compilerOptions": { + "outDir": "dist", + "rootDir": "src/" + }, + "include": ["src/**/*.ts"], + "exclude": ["dist", "node_modules", "**/*.spec.*", "**/*.test.*"] +} diff --git a/agentic/machine-protocol/README.md b/agentic/machine-protocol/README.md index bd5a33ba7b..de86043d1c 100644 --- a/agentic/machine-protocol/README.md +++ b/agentic/machine-protocol/README.md @@ -1,7 +1,7 @@ # @constructive-db/machine-protocol -The wire protocol shared by the machine relay (`compute/services/machine-relay`) -and the machine runner (`compute/lib/machine-runner`): the JSON frames a remote +The wire protocol shared by the machine relay (`compute/services/machine-relay` in constructive-db) +and the machine runner (`agentic/machine-runner`): the JSON frames a remote session is made of — `open`, `input`, `resize`, `close` from the client side; `output`, `exit`, `error` from the runner side; `attached` from the relay — and the codec that encodes them and refuses malformed ones. @@ -11,3 +11,21 @@ The relay never interprets terminal bytes: `input` flows client → runner and session lifecycle. The package has no dependencies so the relay can carry it into the functions image without dragging the runner's native pty bindings along. + +## Two legs, two vocabularies + +The runner leg — what a runner sends and receives — is the generic subset: +`open` (`command`, `args`, `interactive`, `cols`, `rows`, `cwd`), `input`, +`resize`, `signal`, `detach`, `reattach`, `close` in; `output` (with an +optional `stream: 'stdout' | 'stderr'` for a process on pipes), `exit`, +`error`, `scrollback` out; plus enrollment and credential exchange. A runner +never sees anything else, and never parses `output`. + +The agent vocabulary is the client leg's, spoken between a client and the +relay: an `open` may carry a binding (`runId`, `agentMode`, `cliSessionId`, +resolved by `agentBinding`), and the relay turns a bound program's stdout +lines into `agent_event` frames (`assertAgentEvent`). An approval decision is +not a frame at all on the runner side: the relay writes it to the program's +stdin as one JSON line (`encodeApprovalDecisionLine` / +`parseApprovalDecisionLine`), which `constructive-agent-cli` reads. The runner +carried bytes. diff --git a/agentic/machine-protocol/__tests__/codec.test.ts b/agentic/machine-protocol/__tests__/codec.test.ts index 7ebb29c227..92d2095436 100644 --- a/agentic/machine-protocol/__tests__/codec.test.ts +++ b/agentic/machine-protocol/__tests__/codec.test.ts @@ -1,4 +1,16 @@ -import { agentBinding, assertAgentEvent, decodeFrame, encodeFrame, Frame, isAgentEventLike, OpenFrame } from '../src'; +import { + agentBinding, + assertAgentEvent, + decodeFrame, + encodeApprovalDecisionLine, + encodeFrame, + Frame, + isAgentEventLike, + LineSplitter, + MAX_LINE_LENGTH, + OpenFrame, + parseApprovalDecisionLine +} from '../src'; describe('machine-protocol codec', () => { it('round-trips every frame type', () => { @@ -75,23 +87,8 @@ describe('machine-protocol codec', () => { sessionId: 's1', event: { kind: 'result', ok: true, summary: 'done', usage: { input: 1 }, costUsd: 0.1 } }, - { - type: 'approval_request', - sessionId: 's1', - requestId: 'request-1', - tool: 'Bash', - input: { command: 'rm -rf build' }, - reason: 'destructive' - }, - { type: 'approval_request', sessionId: 's1', requestId: 'request-2', tool: 'Edit', input: null }, - { type: 'approval_decision', sessionId: 's1', requestId: 'request-1', decision: 'allow' }, - { - type: 'approval_decision', - sessionId: 's1', - requestId: 'request-2', - decision: 'deny', - reason: 'not now' - }, + { type: 'output', sessionId: 's1', data: '{"kind":"text","text":"hi"}\n', stream: 'stdout' }, + { type: 'output', sessionId: 's1', data: 'warning: slow\n', stream: 'stderr' }, { type: 'open', sessionId: 's1', @@ -188,39 +185,43 @@ describe('machine-protocol codec', () => { ).toThrow(/missing ok/); }); - it('validates approval frames', () => { - const request = { type: 'approval_request', sessionId: 's', requestId: 'r', tool: 'Bash', input: {} }; - expect(() => decodeFrame(JSON.stringify({ ...request, requestId: undefined }))).toThrow( - /missing requestId/ - ); - expect(() => decodeFrame(JSON.stringify({ ...request, requestId: '' }))).toThrow(/missing requestId/); - expect(() => decodeFrame(JSON.stringify({ ...request, tool: undefined }))).toThrow(/missing tool/); - expect(() => decodeFrame(JSON.stringify({ ...request, reason: 7 }))).toThrow( - /reason must be a string/ - ); - expect(() => decodeFrame(JSON.stringify({ ...request, sessionId: undefined }))).toThrow( - /missing sessionId/ - ); + // An approval decision is not a frame the runner sees: it travels to the + // program as one line on its stdin, and the program reads it by this codec. + it('encodes and parses approval decision lines', () => { + const allow = { kind: 'approval_decision' as const, requestId: 'r', decision: 'allow' as const }; + expect(encodeApprovalDecisionLine(allow)).toBe(`${JSON.stringify(allow)}\n`); + expect(parseApprovalDecisionLine(JSON.stringify(allow))).toEqual(allow); + expect( + parseApprovalDecisionLine( + JSON.stringify({ kind: 'approval_decision', requestId: 'r', decision: 'deny', reason: 'not now' }) + ) + ).toEqual({ kind: 'approval_decision', requestId: 'r', decision: 'deny', reason: 'not now' }); + + // Anything that is not a decision is the program's to read: a prompt, an + // unrelated JSON line, an empty line. + expect(parseApprovalDecisionLine('hello')).toBeNull(); + expect(parseApprovalDecisionLine('{"kind":"text"}')).toBeNull(); + expect(parseApprovalDecisionLine('')).toBeNull(); - const decision = { type: 'approval_decision', sessionId: 's', requestId: 'r', decision: 'allow' }; - expect(() => decodeFrame(JSON.stringify({ ...decision, requestId: undefined }))).toThrow( + const decision = { kind: 'approval_decision', requestId: 'r', decision: 'allow' }; + expect(() => parseApprovalDecisionLine(JSON.stringify({ ...decision, requestId: undefined }))).toThrow( /missing requestId/ ); - expect(() => decodeFrame(JSON.stringify({ ...decision, decision: 'maybe' }))).toThrow( + expect(() => parseApprovalDecisionLine(JSON.stringify({ ...decision, decision: 'maybe' }))).toThrow( /invalid decision/ ); - expect(() => decodeFrame(JSON.stringify({ ...decision, decision: undefined }))).toThrow( + expect(() => parseApprovalDecisionLine(JSON.stringify({ ...decision, decision: undefined }))).toThrow( /invalid decision/ ); - expect(() => decodeFrame(JSON.stringify({ ...decision, reason: {} }))).toThrow( + expect(() => parseApprovalDecisionLine(JSON.stringify({ ...decision, reason: {} }))).toThrow( /reason must be a string/ ); }); - // A headless agent process prints events one per line; the runner relaying - // it tells an event line from any other by the same rule the codec applies - // to an `agent_event` frame, and a line that claims to be one but is not is - // a protocol fault, not output. + // A bound program prints events one per line; the relay reading its stdout + // tells an event line from any other by the same rule the codec applies to + // an `agent_event` frame, and a line that claims to be one but is not is a + // protocol fault, not output. The runner itself never reads the line. it('tells an agent event line from output and checks its shape', () => { expect(isAgentEventLike({ kind: 'text', text: 'hi' })).toBe(true); expect(isAgentEventLike({ kind: 'nope' })).toBe(false); @@ -233,6 +234,27 @@ describe('machine-protocol codec', () => { expect(() => assertAgentEvent(null)).toThrow(/'agent_event' frame is missing event/); }); + it('splits lines across chunks and refuses one that never ends', () => { + const lines = new LineSplitter(); + expect(lines.push('one\r\ntw')).toEqual(['one']); + expect(lines.push('o\n\nthree')).toEqual(['two', '']); + expect(lines.flush()).toEqual(['three']); + expect(lines.flush()).toEqual([]); + + // A line up to the cap is buffered; the byte past it is the fault, and the + // reader is left empty rather than holding on to what it refused. + const capped = new LineSplitter(8); + expect(capped.push('12345678')).toEqual([]); + expect(() => capped.push('9')).toThrow(/line exceeds 8 characters without a newline/); + expect(capped.flush()).toEqual([]); + expect(capped.push('ok\n')).toEqual(['ok']); + + expect(MAX_LINE_LENGTH).toBe(1024 * 1024); + const wide = new LineSplitter(); + expect(wide.push('x'.repeat(MAX_LINE_LENGTH))).toEqual([]); + expect(() => wide.push('x')).toThrow(/line exceeds 1048576 characters/); + }); + it('validates the open frame cwd', () => { const open = { type: 'open', sessionId: 's', command: 'constructive-agent-host' }; expect(() => decodeFrame(JSON.stringify({ ...open, cwd: '' }))).toThrow(/cwd/); diff --git a/agentic/machine-protocol/src/index.ts b/agentic/machine-protocol/src/index.ts index 8d50747c16..e82a2b47b6 100644 --- a/agentic/machine-protocol/src/index.ts +++ b/agentic/machine-protocol/src/index.ts @@ -14,6 +14,19 @@ // a decision. That is what makes a reattach from a different client safe — // nothing about the attach that opened the session carries over. // +// Two vocabularies live here, and the line between them is the line between the +// runner and everything above it: +// +// - The *machine* frames (`open` … `exit`, plus enrollment) are the remote +// control. A runner speaks only these: it runs an allow-listed command in a +// pty or on pipes, moves bytes, reports the exit. It has no notion of what the +// command is. +// - The *agent* vocabulary (`AgentBinding` on an `open`, `AgentEvent`, the +// `agent_event` frame, the stdio contract below) is between a client, the +// relay and an agent program running *under* the runner. The runner never +// reads or writes any of it; to the runner an agent program is a command on +// pipes whose stdout happens to be JSON lines. +// // This package is vocabulary, never policy. Everything a socket puts on the wire // — its database, its machine id — is a *coordinate*: it says which row the // socket is asking about, and the relay answers by asking that database's SQL. @@ -106,10 +119,13 @@ export type AgentMode = 'cli' | 'embedded'; export const AGENT_MODES: readonly AgentMode[] = ['cli', 'embedded']; /** - * The agent-run binding an `open` frame may carry. A bound session is headless: - * its process is the agent's, driven by the run rather than by a keyboard, so - * it never gets a pty. Like every coordinate on this wire the run id is a - * request — the tenant's SQL decides whether the opener may bind to that run. + * The agent-run binding an `open` frame may carry, client → relay. A bound + * session is headless: its process is the agent's, driven by the run rather + * than by a keyboard, so it never gets a pty. Like every coordinate on this + * wire the run id is a request — the tenant's SQL decides whether the opener + * may bind to that run. The relay keeps the binding; what reaches the runner + * is a plain `open` for the agent program, with the binding folded into its + * arguments. */ export interface AgentBinding { runId: string; @@ -127,10 +143,11 @@ export interface OpenFrame { cols?: number; rows?: number; /** - * Ask for a terminal rather than a pipe. An interactive session gets a pty, + * Ask for a terminal rather than pipes. An interactive session gets a pty, * so its output is opaque terminal bytes (escape sequences and all) and its * geometry is part of the session — which is what makes `vim` render and - * reflow. Command mode (phase 1) is the default and unchanged. + * reflow. The default is a command on pipes: stdin in, stdout and stderr out + * (each `output` frame says which), no echo, no geometry. */ interactive?: boolean; /** Agent run this session executes locally. Requires `agentMode`. */ @@ -140,8 +157,8 @@ export interface OpenFrame { /** The CLI's own session identifier; only with `agentMode: 'cli'`. */ cliSessionId?: string; /** - * Working directory the session runs in, as a request: the runner's policy - * decides whether it is honoured, and a plain command session ignores it. + * Working directory the session runs in, as a request: resolved against the + * runner's policy root, and refused when it would leave it. */ cwd?: string; } @@ -227,6 +244,13 @@ export interface OutputFrame { type: 'output'; sessionId: string; data: string; + /** + * Which pipe the bytes came from, on a non-interactive session. A terminal + * has one stream, so a pty session's output carries none; a program on + * pipes has two, and a consumer parsing stdout line by line (an agent's + * events) must not have stderr spliced into it. + */ + stream?: 'stdout' | 'stderr'; } export type AgentEvent = @@ -267,31 +291,115 @@ export interface AgentEventFrame { export type ApprovalDecision = 'allow' | 'deny'; +// --------------------------------------------------------------------------- +// The agent program contract. An agent runs under the runner as an ordinary +// non-interactive command; this is what it speaks on its pipes, and the relay +// is the peer that reads and writes it — not the runner, which sees bytes. +// +// stdin — one prompt per line, and {@link ApprovalDecisionLine}s as JSON +// lines where the relay answers a question the program asked; +// stdout — {@link AgentEvent}s as JSON lines; any other line is plain output; +// stderr — plain output; +// exit — the program's own. +// +// `constructive-agent-host` (`@constructive-db/agent-host`) speaks it natively. +// `constructive-agent-cli` (`@constructive-db/agent-cli`) speaks it on behalf +// of a coding-agent CLI (`claude`, `codex`) that does not. +// --------------------------------------------------------------------------- + /** - * Runner → relay: a bound CLI agent asked whether it may use a tool, and the - * runner is holding the turn until it hears back. The relay answers by writing - * the request into the bound run's log, where any authorized client resolves - * it, and then sends an {@link ApprovalDecisionFrame}. Carries no identity: - * who may answer is the run's RLS, asked under the attaching client's claims. + * The program the relay runs for a `cli` binding: it wraps the CLI the opener + * named (`claude`, `codex`) and translates its stream to this contract. The + * machine's policy must allow it by this exact name. */ -export interface ApprovalRequestFrame { - type: 'approval_request'; - sessionId: string; - requestId: string; - tool: string; - input: unknown; - reason?: string; -} +export const AGENT_CLI_COMMAND = 'constructive-agent-cli'; -/** Relay → runner: the answer to an {@link ApprovalRequestFrame}. */ -export interface ApprovalDecisionFrame { - type: 'approval_decision'; - sessionId: string; +/** + * Relay → agent program (a line on its stdin): the answer to an + * `approval_requested` event the program emitted. Carries no identity: who + * answered was the run's RLS, asked on the bound run's log. + */ +export interface ApprovalDecisionLine { + kind: 'approval_decision'; requestId: string; decision: ApprovalDecision; reason?: string; } +export function encodeApprovalDecisionLine(line: ApprovalDecisionLine): string { + return `${JSON.stringify(line)}\n`; +} + +/** The decision a stdin line carries, or null when the line is a prompt. */ +export function parseApprovalDecisionLine(line: string): ApprovalDecisionLine | null { + if (!line.startsWith('{')) return null; + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch { + return null; + } + if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) return null; + const value = parsed as Record; + if (value.kind !== 'approval_decision') return null; + if (typeof value.requestId !== 'string' || value.requestId.length === 0) { + throw new Error('machine-protocol: approval_decision line is missing requestId'); + } + if (value.decision !== 'allow' && value.decision !== 'deny') { + throw new Error('machine-protocol: approval_decision line has invalid decision'); + } + const reason = value.reason; + if (reason !== undefined && typeof reason !== 'string') { + throw new Error('machine-protocol: approval_decision line reason must be a string'); + } + const decoded: ApprovalDecisionLine = { + kind: 'approval_decision', + requestId: value.requestId, + decision: value.decision + }; + if (typeof reason === 'string') decoded.reason = reason; + return decoded; +} + +/** The longest line a {@link LineSplitter} buffers before giving up on it. */ +export const MAX_LINE_LENGTH = 1024 * 1024; + +/** + * Split a stream into the complete lines it has delivered so far. Keeps the + * unterminated tail for the next chunk; `flush` hands it over at the end. A + * tail that outgrows `maxLineLength` is dropped and reported, so a program + * that never writes a newline cannot grow the reader without bound. + */ +export class LineSplitter { + private tail = ''; + + constructor(private readonly maxLineLength = MAX_LINE_LENGTH) {} + + push(chunk: string): string[] { + this.tail += chunk; + const lines = this.tail.split('\n'); + this.tail = lines.pop() ?? ''; + if (this.tail.length > this.maxLineLength) { + this.tail = ''; + throw new Error( + `machine-protocol: line exceeds ${this.maxLineLength} characters without a newline` + ); + } + return lines.map(stripCarriageReturn); + } + + flush(): string[] { + if (!this.tail) return []; + const line = stripCarriageReturn(this.tail); + this.tail = ''; + return [line]; + } +} + +function stripCarriageReturn(line: string): string { + return line.endsWith('\r') ? line.slice(0, -1) : line; +} + /** Runner → client: the process ended. */ export interface ExitFrame { type: 'exit'; @@ -391,8 +499,6 @@ export type RunnerToClientFrame = export type Frame = | ClientToRunnerFrame | RunnerToClientFrame - | ApprovalRequestFrame - | ApprovalDecisionFrame | AttachedFrame | DetachedFrame | ReattachedFrame @@ -408,8 +514,6 @@ const FRAME_TYPES: ReadonlySet = new Set([ 'close', 'output', 'agent_event', - 'approval_request', - 'approval_decision', 'exit', 'error', 'scrollback', @@ -430,20 +534,12 @@ const SESSION_FRAME_TYPES: ReadonlySet = new Set([ 'close', 'output', 'agent_event', - 'approval_request', - 'approval_decision', 'exit', 'scrollback', 'detached', 'reattached' ]); -/** Frame types that carry a `requestId` pairing a request with its answer. */ -const REQUEST_FRAME_TYPES: ReadonlySet = new Set([ - 'approval_request', - 'approval_decision' -]); - const SIGNALS: ReadonlySet = new Set(SIGNAL_NAMES); /** @@ -599,12 +695,6 @@ export function decodeFrame(raw: string): Frame { if (SESSION_FRAME_TYPES.has(type) && typeof frame.sessionId !== 'string') { throw new Error(`machine-protocol: '${type}' frame is missing sessionId`); } - if ( - REQUEST_FRAME_TYPES.has(type) && - (typeof frame.requestId !== 'string' || frame.requestId.length === 0) - ) { - throw new Error(`machine-protocol: '${type}' frame is missing requestId`); - } switch (type) { case 'open': if (typeof frame.command !== 'string' || frame.command.length === 0) { @@ -624,26 +714,18 @@ export function decodeFrame(raw: string): Frame { if (typeof frame.data !== 'string') { throw new Error(`machine-protocol: '${type}' frame is missing data`); } + if ( + type === 'output' && + frame.stream !== undefined && + frame.stream !== 'stdout' && + frame.stream !== 'stderr' + ) { + throw new Error("machine-protocol: 'output' frame stream must be 'stdout' or 'stderr'"); + } break; case 'agent_event': assertAgentEvent(frame.event); break; - case 'approval_request': - if (typeof frame.tool !== 'string' || frame.tool.length === 0) { - throw new Error("machine-protocol: 'approval_request' frame is missing tool"); - } - if (frame.reason !== undefined && typeof frame.reason !== 'string') { - throw new Error("machine-protocol: 'approval_request' frame reason must be a string"); - } - break; - case 'approval_decision': - if (frame.decision !== 'allow' && frame.decision !== 'deny') { - throw new Error("machine-protocol: 'approval_decision' frame has invalid decision"); - } - if (frame.reason !== undefined && typeof frame.reason !== 'string') { - throw new Error("machine-protocol: 'approval_decision' frame reason must be a string"); - } - break; case 'resize': if (!validDimension(frame.cols) || !validDimension(frame.rows)) { throw new Error( diff --git a/agentic/machine-runner/README.md b/agentic/machine-runner/README.md index a7c8480e41..251d7a080f 100644 --- a/agentic/machine-runner/README.md +++ b/agentic/machine-runner/README.md @@ -1,11 +1,20 @@ # @constructive-db/machine-runner The machine runner: the Node daemon a user installs on their own machine for -Remote Control (constructive-planning#1690, phase 0). It dials **out** to a -machine relay over WebSocket — it never listens on a port — spawns -policy-checked processes in a pty (`node-pty`), and streams their bytes back -as `@constructive-db/machine-protocol` frames. When the relay connection -drops, it redials with exponential backoff and re-registers the machine. +Remote Control (constructive-planning#1690). It is a **remote control and +nothing else**: it dials **out** to a machine relay over WebSocket — it never +listens on a port — runs the allow-listed commands the relay hands it (in a pty +for a terminal, on pipes otherwise), streams their bytes back as +`@constructive-db/machine-protocol` frames, and stops them when told. When the +relay connection drops, it redials with exponential backoff and re-registers +the machine. + +The runner knows nothing about what it runs. It has no notion of agents, +Claude, Codex, runs, approvals or events; it never reads a byte of a +command's output. Everything agentic sits *above* it, as ordinary commands +(`constructive-agent-cli` from `@constructive-db/agent-cli`, +`constructive-agent-host` from `@constructive-db/agent-host`) that the relay +asks it to run and whose stdout the relay interprets. Keep it that way. ## Install @@ -63,32 +72,29 @@ start in the policy's `cwd`, and the spawned environment contains only the allow-listed pass-through variables plus the explicit `set` map — the runner's own environment (tokens included) never reaches a session. -## Sessions bound to an agent run - -A session opened with a `runId` is the run's process rather than a terminal: -it runs on pipes, never in a pty, and its stdout lines that are machine-protocol -`AgentEvent`s are relayed as `agent_event` frames (the relay ledgers them -structurally); every other line is ordinary output. The runner spawns and -relays — it holds no harness, no run log and no model credentials, and depends -on nothing from `@agentic-kit/*`. - -- `agentMode: 'cli'` — the command is a coding-agent CLI (`claude`, `codex`) - whose stream-JSON the runner adapts; its tool approvals are asked through the - relay and answered on the bound run's log. -- `agentMode: 'embedded'` — the command is an agent host, normally - `constructive-agent-host` from `@constructive-db/agent-host`. To the runner it - is **an ordinary allow-listed command**: it must appear in `allowedCommands`, - it is spawned through the same policy path as everything else (policy `cwd`, - allow-listed environment), and the runner adds only the binding as - arguments — `--run ` and, when the client asked for one, - `--cwd ` — after the opener's own (`--harness `, …). Which harness - runs, where the run's log lives, how approvals are gated and how the host - finds its platform credential are the host's concerns, documented in - `compute/lib/agent-host`. +## What a session is + +An `open` frame names a command, its arguments, whether it wants a terminal, +and optionally a directory. The runner: + +- refuses the command unless it is in `allowedCommands`, and refuses a `cwd` + that resolves outside the policy's `cwd` (a sibling such as + `/home/me/work-other` is outside `/home/me/work`); +- spawns it in a pty (`interactive: true`, with `cols`/`rows`) or on pipes, + in that directory, with the projected environment; +- streams `output` frames (pipe output tagged `stdout`/`stderr`; pty output is + the terminal's byte stream), writes `input` frames to its stdin, applies + `resize` and `signal`, and reports `exit` or `error`; +- keeps a detached interactive session alive with a bounded scrollback until + a client reattaches or the session is closed. + +That is the whole contract. An agent session is one of these whose command +happens to be an agent program — the relay composes that command and reads +its output; the runner ran an allow-listed command: ```json "policy": { - "allowedCommands": ["bash", "git", "constructive-agent-host"], + "allowedCommands": ["bash", "git", "constructive-agent-cli", "constructive-agent-host"], "cwd": "/home/me/work", "env": { "allow": ["PATH", "HOME"] } } diff --git a/agentic/machine-runner/__tests__/fixtures/bin/codex b/agentic/machine-runner/__tests__/fixtures/bin/codex deleted file mode 100755 index f986c8fd5f..0000000000 --- a/agentic/machine-runner/__tests__/fixtures/bin/codex +++ /dev/null @@ -1,14 +0,0 @@ -#!/usr/bin/env node - -const args = process.argv.slice(2); -const promptIndex = args.indexOf('resume'); -const sessionId = promptIndex >= 0 ? args[promptIndex + 1] : '01a0a38b-442f-7f91-a250-72f305c839f7'; -const prompt = promptIndex >= 0 ? args[promptIndex + 2] : args[args.length - 1]; -const emit = value => process.stdout.write(`${JSON.stringify(value)}\n`); - -emit({ type: 'thread.started', thread_id: sessionId }); -emit({ - type: 'item.completed', - item: { id: 'msg_fake', type: 'agent_message', text: `echo:${prompt}` } -}); -emit({ type: 'turn.completed', usage: {} }); diff --git a/agentic/machine-runner/__tests__/fixtures/bin/constructive-agent-host b/agentic/machine-runner/__tests__/fixtures/bin/constructive-agent-host deleted file mode 100755 index c161e891ad..0000000000 --- a/agentic/machine-runner/__tests__/fixtures/bin/constructive-agent-host +++ /dev/null @@ -1,53 +0,0 @@ -#!/usr/bin/env node -// A stand-in for the agent host the runner spawns for an `embedded` session. -// It knows nothing of any harness: it reports the arguments it was started -// with, echoes each prompt line back as agent events, and exits when its -// stdin closes — which is everything the runner needs to prove about the -// relationship. The real host (`@constructive-db/agent-host`) is tested on its -// own, against a fake harness. - -const readline = require('readline'); - -const argv = process.argv.slice(2); -const flag = name => { - const index = argv.indexOf(`--${name}`); - return index >= 0 ? argv[index + 1] : undefined; -}; - -const emit = value => process.stdout.write(`${JSON.stringify(value)}\n`); - -// Plain output first: the runner must relay what is not an event as output. -process.stdout.write('fake host starting\n'); -// No `session` event: a resumable CLI session id is a `cli`-mode fact, and the -// ledger's check constraint keeps `agent_session_ref` null on an embedded row. -emit({ - kind: 'text', - text: `started:${JSON.stringify({ argv, cwd: process.cwd(), runId: flag('run'), harness: flag('harness'), dir: flag('cwd') })}` -}); - -const rl = readline.createInterface({ input: process.stdin }); -rl.on('line', line => { - const prompt = line.trim(); - if (prompt === 'fail') { - process.stderr.write('fake host: failing on request\n'); - process.exit(3); - } - if (prompt === 'malformed') { - // Names an event kind but is not one: a protocol breach the runner must - // refuse rather than ledger. - emit({ kind: 'text' }); - return; - } - if (prompt === 'approve') { - // The host settles its own approvals — through its run-log gate, in the - // real one — and only reports them; the runner has no part in the answer. - emit({ kind: 'approval_requested', requestId: 'req-1', tool: 'bash', input: { command: 'rm -rf build' } }); - emit({ kind: 'approval_resolved', requestId: 'req-1', decision: 'deny', reason: 'fake host says no' }); - emit({ kind: 'tool_result', id: 'req-1', output: 'blocked: fake host says no', isError: true }); - } - emit({ kind: 'tool_call', id: 'call-1', name: 'echo', input: { prompt } }); - emit({ kind: 'tool_result', id: 'call-1', output: prompt }); - emit({ kind: 'text', text: `echo:${prompt}` }); - emit({ kind: 'result', ok: true, summary: `answered ${prompt}` }); -}); -rl.on('close', () => process.exit(0)); diff --git a/agentic/machine-runner/src/cli-session.ts b/agentic/machine-runner/src/cli-session.ts deleted file mode 100644 index 5d38f36c4b..0000000000 --- a/agentic/machine-runner/src/cli-session.ts +++ /dev/null @@ -1,268 +0,0 @@ -import type { AgentEvent, ApprovalDecision } from '@constructive-db/machine-protocol'; -import type { Logger } from '@pgpmjs/logger'; -import { spawn as spawnChild } from 'child_process'; -import os from 'os'; - -import type { AgentCliAdapter } from './agent-cli'; -import type { RunnerPolicy } from './policy'; -import { resolveApprovalPolicy, resolveSpawn } from './policy'; -import type { ApprovalRequest, SessionProcess } from './runner'; - -export const APPROVAL_AUTO_DENY_REASON = - 'approval round-trip is not available in this session; denied automatically'; -export const APPROVAL_TIMEOUT_REASON = 'no decision arrived before the approval timeout'; -export const APPROVAL_DISCONNECT_REASON = - 'the runner lost its relay while the approval was pending; denied'; -export const APPROVAL_EXIT_REASON = 'the session ended while the approval was pending'; - -export interface CliProcessOptions { - adapter: AgentCliAdapter; - policy: RunnerPolicy; - command: string; - resume?: string; - logger: Logger; -} - -export function cliProcess(options: CliProcessOptions): SessionProcess { - // Validate the opener's command before the session row reaches the runner. - resolveSpawn(options.policy, options.command, []); - - let child: ReturnType | null = null; - let firstWrite = true; - let stdinEnded = false; - let exited = false; - let stdoutTail = ''; - const approvalPolicy = resolveApprovalPolicy(options.policy); - const pendingApprovals = new Map(); - const dataListeners: Array<(data: string) => void> = []; - const eventListeners: Array<(event: AgentEvent) => void> = []; - const approvalListeners: Array<(request: ApprovalRequest) => void> = []; - const warningListeners: Array<(message: string) => void> = []; - const exitListeners: Array<(event: { exitCode: number; signal?: number }) => void> = []; - const errorListeners: Array<(err: Error) => void> = []; - - const emitData = (data: string): void => { - for (const listener of dataListeners) listener(data); - }; - const emitWarning = (message: string): void => { - for (const listener of warningListeners) listener(message); - }; - /** - * Answer the CLI. Returns false when nothing was pending under that id: an - * answer that arrives twice, or after a timeout already spoke, is ignored - * rather than written into the CLI's stdin as a second control response. - */ - const answer = (requestId: string, decision: ApprovalDecision, reason?: string): boolean => { - const timer = pendingApprovals.get(requestId); - if (timer === undefined) return false; - clearTimeout(timer); - pendingApprovals.delete(requestId); - if (!child || !options.adapter.encodeApproval) { - throw new Error( - `machine-runner: ${options.adapter.name} approval '${requestId}' cannot be answered without a process` - ); - } - if (stdinEnded || exited) { - options.logger.warn( - `machine-runner: ${options.adapter.name} approval '${requestId}' ${decision} arrived after stdin closed` - ); - return true; - } - child.stdin.write(options.adapter.encodeApproval(requestId, decision, reason)); - options.logger.info( - `machine-runner: ${options.adapter.name} approval '${requestId}' ${decision}` + - (reason ? `: ${reason}` : '') - ); - return true; - }; - /** - * Answer the CLI and report it as an `approval_resolved` event, whoever - * decided: the event stream is the session's record, and a request in it - * without its answer reads as still open. - */ - const settle = (requestId: string, decision: ApprovalDecision, reason?: string): boolean => { - if (!answer(requestId, decision, reason)) return false; - for (const listener of eventListeners) { - listener({ - kind: 'approval_resolved', - requestId, - decision, - ...(reason === undefined ? {} : { reason }) - }); - } - return true; - }; - const emitEvent = (event: AgentEvent): void => { - // The request itself is an event like any other: it is what the ledger - // and a watching client see. Only its answer is routed differently. - for (const listener of eventListeners) listener(event); - if (event.kind !== 'approval_requested') return; - if (!options.adapter.encodeApproval) { - // The adapter can surface the question but has no way to answer it; - // say so in the record rather than leaving a request open forever. - options.logger.warn( - `machine-runner: ${options.adapter.name} approval '${event.requestId}' cannot be answered: ${APPROVAL_AUTO_DENY_REASON}` - ); - for (const listener of eventListeners) { - listener({ - kind: 'approval_resolved', - requestId: event.requestId, - decision: 'deny', - reason: APPROVAL_AUTO_DENY_REASON - }); - } - return; - } - if (pendingApprovals.has(event.requestId)) { - throw new Error( - `machine-runner: ${options.adapter.name} asked for approval '${event.requestId}' twice` - ); - } - const timer = setTimeout(() => { - settle(event.requestId, approvalPolicy.onTimeout, APPROVAL_TIMEOUT_REASON); - }, approvalPolicy.timeoutMs); - pendingApprovals.set(event.requestId, timer); - const request: ApprovalRequest = { - requestId: event.requestId, - tool: event.tool, - input: event.input, - ...(event.reason !== undefined ? { reason: event.reason } : {}) - }; - if (approvalListeners.length === 0) { - // Nobody is carrying the question upstream, so nobody will ever answer - // it: the old automatic denial, made explicit. - settle(event.requestId, 'deny', APPROVAL_AUTO_DENY_REASON); - return; - } - for (const listener of approvalListeners) listener(request); - }; - const settleAll = (reason: string): void => { - for (const requestId of [...pendingApprovals.keys()]) settle(requestId, 'deny', reason); - }; - const parseLine = (line: string): void => { - // Only the parse is guarded: a line that is not the adapter's protocol is - // plain output. What a parsed event then does is not, so a fault there - // (a duplicated approval id) is the process's error, not stray stdout. - let events: AgentEvent[] | null; - try { - events = options.adapter.parseEvent(line); - } catch (err) { - const message = err instanceof Error ? err.message : String(err); - options.logger.warn( - `machine-runner: ${options.adapter.name} emitted non-JSON stdout: ${message}` - ); - emitData(`${line}\n`); - return; - } - if (events) for (const event of events) emitEvent(event); - }; - const consumeStdout = (chunk: string): void => { - stdoutTail += chunk; - const lines = stdoutTail.split('\n'); - stdoutTail = lines.pop() ?? ''; - for (const line of lines) parseLine(line.endsWith('\r') ? line.slice(0, -1) : line); - }; - const flushStdout = (): void => { - if (!stdoutTail) return; - const line = stdoutTail.endsWith('\r') ? stdoutTail.slice(0, -1) : stdoutTail; - stdoutTail = ''; - parseLine(line); - }; - const signalNumber = (signal: NodeJS.Signals | string | undefined): number | undefined => { - if (!signal) return undefined; - const value = os.constants.signals[signal as NodeJS.Signals]; - return value; - }; - const start = (prompt: string): void => { - const spawn = options.adapter.spawnArgs(prompt, options.resume); - const spec = resolveSpawn(options.policy, spawn.command, spawn.args); - child = spawnChild(spec.command, spec.args, { - cwd: spec.cwd, - env: spec.env, - stdio: ['pipe', 'pipe', 'pipe'] - }); - child.stdout.setEncoding('utf8').on('data', (data: string) => { - try { - consumeStdout(data); - } catch (err) { - // The same door a spawn failure comes through: the session fails - // visibly rather than the stream handler taking the runner down. - for (const listener of errorListeners) listener(err instanceof Error ? err : new Error(String(err))); - } - }); - child.stderr.setEncoding('utf8').on('data', (data: string) => emitData(data)); - child.on('error', err => { - for (const listener of errorListeners) listener(err); - }); - child.on('exit', (code, signal) => { - if (exited) return; - exited = true; - flushStdout(); - settleAll(APPROVAL_EXIT_REASON); - for (const listener of exitListeners) { - listener({ exitCode: code ?? -1, ...(signal ? { signal: signalNumber(signal) } : {}) }); - } - }); - if (spawn.stdinPrompt !== undefined) child.stdin.write(spawn.stdinPrompt); - if (!options.adapter.encodeTurn) { - child.stdin.end(); - stdinEnded = true; - } - }; - - return { - write(data: string): void { - const prompt = data.endsWith('\n') ? data.slice(0, -1) : data; - if (firstWrite) { - firstWrite = false; - start(prompt); - return; - } - if (options.adapter.encodeTurn && child && !stdinEnded) { - child.stdin.write(options.adapter.encodeTurn(prompt)); - return; - } - if (!options.adapter.encodeTurn) { - emitWarning( - `${options.adapter.name} takes a single prompt; open a new session with cliSessionId to continue` - ); - } - }, - resolveApproval(requestId, decision, reason): boolean { - return settle(requestId, decision, reason); - }, - denyPendingApprovals(reason = APPROVAL_DISCONNECT_REASON): void { - settleAll(reason); - }, - onApprovalRequest(listener): void { - approvalListeners.push(listener); - }, - kill(signal): void { - if (!child) { - if (exited) return; - exited = true; - const number = signalNumber(signal ?? 'SIGTERM'); - for (const listener of exitListeners) { - listener({ exitCode: -1, ...(number !== undefined ? { signal: number } : {}) }); - } - return; - } - child.kill(signal); - }, - onData(listener): void { - dataListeners.push(listener); - }, - onEvent(listener): void { - eventListeners.push(listener); - }, - onWarning(listener): void { - warningListeners.push(listener); - }, - onExit(listener): void { - exitListeners.push(listener); - }, - onError(listener): void { - errorListeners.push(listener); - } - }; -} diff --git a/agentic/machine-runner/src/config.ts b/agentic/machine-runner/src/config.ts index 5c65ef8de1..d4dc3381ce 100644 --- a/agentic/machine-runner/src/config.ts +++ b/agentic/machine-runner/src/config.ts @@ -65,25 +65,6 @@ export function parseRunnerConfig(raw: unknown): RunnerConfig { if (typeof policyObj.cwd !== 'string' || policyObj.cwd.length === 0) { throw new Error("machine-runner config: 'policy.cwd' is required"); } - if (policyObj.approvals !== undefined) { - if (typeof policyObj.approvals !== 'object' || policyObj.approvals === null) { - throw new Error("machine-runner config: 'policy.approvals' must be an object"); - } - const approvals = policyObj.approvals as Record; - if ( - approvals.timeoutMs !== undefined && - (typeof approvals.timeoutMs !== 'number' || !(approvals.timeoutMs > 0)) - ) { - throw new Error("machine-runner config: 'policy.approvals.timeoutMs' must be a positive number"); - } - if ( - approvals.onTimeout !== undefined && - approvals.onTimeout !== 'deny' && - approvals.onTimeout !== 'allow' - ) { - throw new Error("machine-runner config: 'policy.approvals.onTimeout' must be 'deny' or 'allow'"); - } - } return { enrollments, policy: policyObj as unknown as RunnerPolicy diff --git a/agentic/machine-runner/src/headless-session.ts b/agentic/machine-runner/src/headless-session.ts deleted file mode 100644 index c41df98c49..0000000000 --- a/agentic/machine-runner/src/headless-session.ts +++ /dev/null @@ -1,151 +0,0 @@ -// A bound session in `embedded` mode is an ordinary command on pipes: the -// runner spawns whatever the open frame names (an agent host, typically — -// `constructive-agent-host`), under the same allow-list and environment policy -// as any other command, and relays its stdio. What the runner adds is the -// binding, as arguments: `--run ` names the run the session executes, -// and `--cwd ` the directory the client asked for, both for the command -// to interpret and enforce. The runner knows nothing about what runs inside — -// not the harness, not its log, not its credentials. -// -// The one thing the runner reads on the way through is the machine protocol's -// own vocabulary: a stdout line that is an `AgentEvent` is relayed as one, so -// the relay can ledger it structurally, and every other line is output. - -import type { AgentEvent } from '@constructive-db/machine-protocol'; -import { assertAgentEvent, isAgentEventLike } from '@constructive-db/machine-protocol'; -import { spawn as spawnChild } from 'child_process'; -import os from 'os'; - -import type { RunnerPolicy } from './policy'; -import { resolveSpawn } from './policy'; -import type { SessionProcess } from './runner'; - -export interface HeadlessProcessOptions { - policy: RunnerPolicy; - command: string; - args: string[]; - /** The run the session is bound to; handed to the command as `--run`. */ - runId: string; - /** The working directory the client asked for; handed on as `--cwd`. */ - cwd?: string; -} - -/** The arguments the binding adds to the opener's own. */ -export function bindingArgs(runId: string, cwd?: string): string[] { - return ['--run', runId, ...(cwd !== undefined ? ['--cwd', cwd] : [])]; -} - -export function headlessProcess(options: HeadlessProcessOptions): SessionProcess { - const spec = resolveSpawn(options.policy, options.command, [ - ...options.args, - ...bindingArgs(options.runId, options.cwd) - ]); - - let exited = false; - let stdoutTail = ''; - const dataListeners: Array<(data: string) => void> = []; - const eventListeners: Array<(event: AgentEvent) => void> = []; - const exitListeners: Array<(event: { exitCode: number; signal?: number }) => void> = []; - const errorListeners: Array<(err: Error) => void> = []; - - const emitData = (data: string): void => { - for (const listener of dataListeners) listener(data); - }; - const fail = (err: Error): void => { - for (const listener of errorListeners) listener(err); - }; - const signalNumber = (signal: NodeJS.Signals | string | undefined): number | undefined => { - if (!signal) return undefined; - return os.constants.signals[signal as NodeJS.Signals]; - }; - /** - * A line is an event when it parses as one; anything else the process - * prints is output. A line that names an event kind but is malformed is the - * process breaking the protocol, and that fails the session rather than - * passing as output — the relay would otherwise ledger a lie. - */ - const parseLine = (line: string): void => { - if (!line.startsWith('{')) { - emitData(`${line}\n`); - return; - } - let parsed: unknown; - try { - parsed = JSON.parse(line); - } catch { - // Not JSON after all; the brace was the program's own output. - emitData(`${line}\n`); - return; - } - if (!isAgentEventLike(parsed)) { - emitData(`${line}\n`); - return; - } - const event = assertAgentEvent(parsed, `'${options.command}' stdout line`); - for (const listener of eventListeners) listener(event); - }; - const consumeStdout = (chunk: string): void => { - stdoutTail += chunk; - const lines = stdoutTail.split('\n'); - stdoutTail = lines.pop() ?? ''; - for (const line of lines) parseLine(line.endsWith('\r') ? line.slice(0, -1) : line); - }; - const flushStdout = (): void => { - if (!stdoutTail) return; - const line = stdoutTail.endsWith('\r') ? stdoutTail.slice(0, -1) : stdoutTail; - stdoutTail = ''; - parseLine(line); - }; - - const child = spawnChild(spec.command, spec.args, { - cwd: spec.cwd, - env: spec.env, - stdio: ['pipe', 'pipe', 'pipe'] - }); - child.stdout.setEncoding('utf8').on('data', (data: string) => { - try { - consumeStdout(data); - } catch (err) { - // A protocol fault on stdout ends the session the way a spawn failure - // does: visibly, with the process taken down rather than left talking to - // nobody. The error is the session's one terminal report; the exit the - // kill provokes must not follow it as a second. - exited = true; - fail(err instanceof Error ? err : new Error(String(err))); - child.kill('SIGTERM'); - } - }); - child.stderr.setEncoding('utf8').on('data', (data: string) => emitData(data)); - child.on('error', err => fail(err)); - child.on('exit', (code, signal) => { - if (exited) return; - exited = true; - flushStdout(); - for (const listener of exitListeners) { - listener({ exitCode: code ?? -1, ...(signal ? { signal: signalNumber(signal) } : {}) }); - } - }); - - return { - write(data: string): void { - // Input is line-oriented on a headless session: a prompt is a line. - if (exited || child.stdin.destroyed) return; - child.stdin.write(data.endsWith('\n') ? data : `${data}\n`); - }, - kill(signal): void { - child.kill(signal ?? 'SIGTERM'); - }, - onData(listener): void { - dataListeners.push(listener); - }, - onEvent(listener): void { - eventListeners.push(listener); - }, - onExit(listener): void { - exitListeners.push(listener); - }, - onError(listener): void { - errorListeners.push(listener); - } - }; -} diff --git a/agentic/machine-runner/src/index.ts b/agentic/machine-runner/src/index.ts index 4ee156bd97..ab09061e12 100644 --- a/agentic/machine-runner/src/index.ts +++ b/agentic/machine-runner/src/index.ts @@ -1,9 +1,3 @@ -export type { AgentCliAdapter, AgentCliSpawn } from './agent-cli'; -export { - adapterForCommand, - ClaudeCodeAdapter, - CodexExecAdapter -} from './agent-cli'; export type { CliIo } from './cli'; export { defaultConfigPath, @@ -14,25 +8,11 @@ export { TOKEN_ENV, writeEnrollment } from './cli'; -export { - APPROVAL_AUTO_DENY_REASON, - APPROVAL_DISCONNECT_REASON, - APPROVAL_EXIT_REASON, - APPROVAL_TIMEOUT_REASON, - cliProcess -} from './cli-session'; export type { Enrollment, RunnerConfig } from './config'; export { loadRunnerConfig, parseRunnerConfig } from './config'; -export type { HeadlessProcessOptions } from './headless-session'; -export { bindingArgs, headlessProcess } from './headless-session'; -export type { ApprovalPolicy, RunnerPolicy, SpawnSpec } from './policy'; -export { - DEFAULT_APPROVAL_TIMEOUT_MS, - DEFAULT_ENV_ALLOW, - PolicyViolationError, - resolveApprovalPolicy, - resolveSpawn -} from './policy'; +export type { RunnerPolicy, SpawnSpec } from './policy'; +export { DEFAULT_ENV_ALLOW, PolicyViolationError, resolveCwd, resolveSpawn } from './policy'; +export type { ProcessExit, SessionProcess } from './process'; +export { pipeProcess, ptyProcess } from './process'; export type { EnrollmentRunnerOptions, MachineRunnerOptions } from './runner'; -export type { ApprovalRequest, SessionProcess } from './runner'; export { EnrollmentRunner, MachineRunner } from './runner'; diff --git a/agentic/machine-runner/src/policy.ts b/agentic/machine-runner/src/policy.ts index 7356d32cde..3e25852aa4 100644 --- a/agentic/machine-runner/src/policy.ts +++ b/agentic/machine-runner/src/policy.ts @@ -4,11 +4,15 @@ // decided here. import os from 'os'; +import path from 'path'; export interface RunnerPolicy { /** Commands (argv[0], exact match) a session may start. */ allowedCommands: string[]; - /** Working directory every session starts in. */ + /** + * The root every session runs under: where a session starts by default, and + * the directory a requested `cwd` must stay inside. + */ cwd: string; /** * Environment the spawned process sees: `allow` names process.env vars to @@ -19,40 +23,6 @@ export interface RunnerPolicy { allow?: string[]; set?: Record; }; - /** - * How long a bound CLI agent's tool approval may wait for an answer from - * the relay before the runner settles it itself. The CLI is never left - * hanging: past `timeoutMs` the runner answers with `onTimeout`, which is a - * denial unless the machine's owner says otherwise. - */ - approvals?: { - timeoutMs?: number; - onTimeout?: 'deny' | 'allow'; - }; -} - -/** Five minutes: long enough for a human to look, short enough to notice. */ -export const DEFAULT_APPROVAL_TIMEOUT_MS = 5 * 60_000; - -export interface ApprovalPolicy { - timeoutMs: number; - onTimeout: 'deny' | 'allow'; -} - -export function resolveApprovalPolicy(policy: RunnerPolicy): ApprovalPolicy { - const timeoutMs = policy.approvals?.timeoutMs ?? DEFAULT_APPROVAL_TIMEOUT_MS; - if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) { - throw new PolicyViolationError( - `policy.approvals.timeoutMs must be a positive number of milliseconds, got ${timeoutMs}` - ); - } - const onTimeout = policy.approvals?.onTimeout ?? 'deny'; - if (onTimeout !== 'deny' && onTimeout !== 'allow') { - throw new PolicyViolationError( - `policy.approvals.onTimeout must be 'deny' or 'allow', got ${String(onTimeout)}` - ); - } - return { timeoutMs, onTimeout }; } /** Pass-through vars a pty session cannot reasonably run without. */ @@ -72,10 +42,26 @@ export interface SpawnSpec { env: Record; } +/** + * The directory a session runs in: the policy root, or a requested directory + * resolved against it — and refused when it would land outside. The request + * is the client's; where the root is, is the machine owner's. + */ +export function resolveCwd(policy: RunnerPolicy, requested?: string): string { + const root = path.resolve(policy.cwd || os.homedir()); + if (requested === undefined) return root; + const resolved = path.resolve(root, requested); + if (resolved !== root && !resolved.startsWith(root + path.sep)) { + throw new PolicyViolationError(`cwd '${requested}' is outside the policy root`); + } + return resolved; +} + export function resolveSpawn( policy: RunnerPolicy, command: string, - args: string[] = [] + args: string[] = [], + cwd?: string ): SpawnSpec { if (!policy.allowedCommands.includes(command)) { throw new PolicyViolationError(`command '${command}' is not in the allowed command list`); @@ -92,7 +78,7 @@ export function resolveSpawn( return { command, args, - cwd: policy.cwd || os.homedir(), + cwd: resolveCwd(policy, cwd), env }; } diff --git a/agentic/machine-runner/src/process.ts b/agentic/machine-runner/src/process.ts new file mode 100644 index 0000000000..e4325ce35b --- /dev/null +++ b/agentic/machine-runner/src/process.ts @@ -0,0 +1,115 @@ +// The two ways a session's process runs: in a pty, when the session is a +// terminal, or on pipes, when it is a command. Both are driven through the +// same small interface; the runner does not care which it holds beyond +// whether it can be resized. + +import type { SignalName } from '@constructive-db/machine-protocol'; +import { spawn as spawnChild } from 'child_process'; +import * as pty from 'node-pty'; +import os from 'os'; + +import type { SpawnSpec } from './policy'; + +export interface ProcessExit { + exitCode: number; + signal?: number; +} + +/** What the runner drives, whether it is a pty or a pair of pipes. */ +export interface SessionProcess { + write(data: string): void; + /** Absent where there is no terminal to resize. */ + resize?(cols: number, rows: number): void; + kill(signal?: SignalName): void; + /** + * Bytes the process wrote. A pty has one stream; a process on pipes says + * which of its two a chunk came from. + */ + onData(listener: (data: string, stream?: 'stdout' | 'stderr') => void): void; + onExit(listener: (event: ProcessExit) => void): void; + /** The process failed to run at all (a pty reports that by throwing from spawn). */ + onError(listener: (err: Error) => void): void; +} + +export function ptyProcess(spec: SpawnSpec, cols: number, rows: number): SessionProcess { + const proc = pty.spawn(spec.command, spec.args, { + name: 'xterm-256color', + cols, + rows, + cwd: spec.cwd, + env: spec.env + }); + return { + write: data => proc.write(data), + resize: (c, r) => proc.resize(c, r), + kill: signal => proc.kill(signal), + onData: listener => proc.onData(data => listener(data)), + onExit: listener => proc.onExit(listener), + // A pty has no asynchronous failure: `pty.spawn` throws, and everything + // after that is an exit. + onError: () => {} + }; +} + +function signalNumber(signal: NodeJS.Signals | string | null | undefined): number | undefined { + if (!signal) return undefined; + return os.constants.signals[signal as NodeJS.Signals]; +} + +export function pipeProcess(spec: SpawnSpec): SessionProcess { + const dataListeners: Array<(data: string, stream: 'stdout' | 'stderr') => void> = []; + const exitListeners: Array<(event: ProcessExit) => void> = []; + const errorListeners: Array<(err: Error) => void> = []; + let exited = false; + + const child = spawnChild(spec.command, spec.args, { + cwd: spec.cwd, + env: spec.env, + stdio: ['pipe', 'pipe', 'pipe'] + }); + child.stdout.setEncoding('utf8').on('data', (data: string) => { + for (const listener of dataListeners) listener(data, 'stdout'); + }); + child.stderr.setEncoding('utf8').on('data', (data: string) => { + for (const listener of dataListeners) listener(data, 'stderr'); + }); + child.on('error', err => { + for (const listener of errorListeners) listener(err); + }); + child.on('exit', (code, signal) => { + if (exited) return; + exited = true; + const number = signalNumber(signal); + // A signal death has no code of its own; report it as a shell would. + const exitCode = code ?? (number !== undefined ? 128 + number : -1); + for (const listener of exitListeners) { + listener({ exitCode, ...(number !== undefined ? { signal: number } : {}) }); + } + }); + // A child that closed its stdin early makes the next write EPIPE. While it + // runs that is the session's failure; once it has exited, the exit already + // said everything and there is nobody left to tell. + child.stdin.on('error', err => { + if (exited) return; + for (const listener of errorListeners) listener(err); + }); + + return { + write(data) { + if (!exited && child.stdin.writable) child.stdin.write(data); + }, + kill(signal) { + if (exited) return; + child.kill(signal); + }, + onData(listener) { + dataListeners.push(listener); + }, + onExit(listener) { + exitListeners.push(listener); + }, + onError(listener) { + errorListeners.push(listener); + } + }; +} diff --git a/agentic/machine-runner/src/runner.ts b/agentic/machine-runner/src/runner.ts index a203549534..faf165c74e 100644 --- a/agentic/machine-runner/src/runner.ts +++ b/agentic/machine-runner/src/runner.ts @@ -1,9 +1,17 @@ // One outbound connection per enrollment. The runner dials the relay, keeps // the socket alive with a reconnect loop, and serves sessions: an `open` -// frame spawns a pty (after the local policy says yes), `input`/`resize`/ +// frame spawns a process (after the local policy says yes), `input`/`resize`/ // `signal` drive it, and its output and exit stream back as frames. Nothing // here ever listens on a port. // +// The runner is a remote control and nothing more. It runs the command it is +// handed — in a pty when the session is a terminal, on pipes when it is a +// command — moves the bytes, and reports the exit. It does not know what the +// command is. A coding agent, an agent host, a build: each is an allow-listed +// program on pipes whose stdout the runner forwards without reading. Whatever +// vocabulary such a program speaks is between it and the relay's clients; +// none of it is here, and none of it may be added here. +// // A pty outlives the socket that asked for it. An interactive session is a // terminal on this machine, not a state of the connection, so `detach` (and a // dropped relay) stop the *streaming* and leave the process running; the bytes @@ -11,21 +19,8 @@ // `reattach` — from whichever client the tenant then authorizes — replays that // ring so the program repaints. The runner is the only place those bytes are // held in memory, and it holds a bounded window of them, never a transcript. -// -// A session bound to an agent run is headless: it is the run's process, not a -// terminal anyone types into, so it never gets a pty. It runs on pipes, the way -// a non-interactive command does. In `cli` mode the command is a coding-agent -// CLI whose stream-JSON the runner adapts; in `embedded` mode it is an agent -// host (`constructive-agent-host`) that speaks the machine protocol's events -// itself, and the runner relays it like any other command. Either way the -// runner spawns an allow-listed program and moves bytes: what runs inside — -// which harness, whose run log, which credentials — is that program's concern. import { - AgentBinding, - agentBinding, - AgentEvent, - ApprovalDecision, decodeFrame, encodeFrame, Frame, @@ -36,14 +31,11 @@ import { SignalName } from '@constructive-db/machine-protocol'; import { Logger } from '@pgpmjs/logger'; -import * as pty from 'node-pty'; import { WebSocket } from 'ws'; -import { adapterForCommand } from './agent-cli'; -import { cliProcess } from './cli-session'; import { Enrollment } from './config'; -import { headlessProcess } from './headless-session'; -import { PolicyViolationError, resolveSpawn, RunnerPolicy, SpawnSpec } from './policy'; +import { PolicyViolationError, resolveSpawn, RunnerPolicy } from './policy'; +import { pipeProcess, ptyProcess, SessionProcess } from './process'; /** Stop presenting a credential this long before it expires. */ const CREDENTIAL_SKEW_MS = 30_000; @@ -67,62 +59,10 @@ const CONTROL_CHARACTERS: Partial> = { SIGQUIT: '\x1c' }; -/** - * What the runner drives, whether it is a pty or a pair of pipes. `resize` is - * absent where there is no terminal to resize. - */ -export interface SessionProcess { - write(data: string): void; - resize?(cols: number, rows: number): void; - kill(signal?: SignalName): void; - onData(listener: (data: string) => void): void; - onExit(listener: (event: { exitCode: number; signal?: number }) => void): void; - /** The process failed to run at all (a pty reports that by throwing from spawn). */ - onError(listener: (err: Error) => void): void; - onEvent?(listener: (event: AgentEvent) => void): void; - onWarning?(listener: (message: string) => void): void; - /** - * A bound CLI agent asked whether it may use a tool. The process holds the - * turn until {@link resolveApproval} answers, or until its own timeout does. - */ - onApprovalRequest?(listener: (request: ApprovalRequest) => void): void; - /** Answer a pending request. False when nothing is pending under that id. */ - resolveApproval?(requestId: string, decision: ApprovalDecision, reason?: string): boolean; - /** Deny everything still pending, with the reason the CLI will be shown. */ - denyPendingApprovals?(reason?: string): void; -} - -export interface ApprovalRequest { - requestId: string; - tool: string; - input: unknown; - reason?: string; -} - -function ptyProcess(spec: SpawnSpec, cols: number, rows: number): SessionProcess { - const proc = pty.spawn(spec.command, spec.args, { - name: 'xterm-256color', - cols, - rows, - cwd: spec.cwd, - env: spec.env - }); - return { - write: data => proc.write(data), - resize: (c, r) => proc.resize(c, r), - kill: signal => proc.kill(signal), - onData: listener => proc.onData(listener), - onExit: listener => proc.onExit(listener), - onError: () => {} - }; -} - interface RunnerSession { proc: SessionProcess; /** A terminal the tenant may detach from and come back to. */ interactive: boolean; - /** The agent run this session executes, or null for a plain session. */ - binding: AgentBinding | null; cols: number; rows: number; /** False while nobody is attached: output is buffered, not sent. */ @@ -202,11 +142,6 @@ export class EnrollmentRunner { return [...this.sessions.keys()]; } - /** The agent-run binding a live session carries, for tests and PR-4 execution. */ - sessionBinding(sessionId: string): AgentBinding | null | undefined { - return this.sessions.get(sessionId)?.binding; - } - /** * What to present on this dial: the credential from the last exchange while it * still verifies, else the enrollment token. The enrollment token buys an @@ -273,9 +208,6 @@ export class EnrollmentRunner { // nothing to come back for, so it ends with its connection. for (const [sessionId, session] of this.sessions) { session.streaming = false; - // A question nobody can answer any more is answered here, before the - // process is torn down, so the CLI hears a denial and not silence. - session.proc.denyPendingApprovals?.(); if (session.interactive) continue; session.proc.kill(); this.sessions.delete(sessionId); @@ -311,7 +243,7 @@ export class EnrollmentRunner { encodeFrame({ type: 'error', sessionId: frame.sessionId, - message: 'headless session has no terminal to resize' + message: 'command session has no terminal to resize' }) ); return; @@ -371,28 +303,6 @@ export class EnrollmentRunner { } return; } - case 'approval_decision': { - const session = this.require(ws, frame.sessionId); - if (!session) return; - if (!session.proc.resolveApproval) { - this.logger.warn( - `machine-runner: session '${frame.sessionId}' is not an agent session and takes no ` + - `approval decisions; '${frame.requestId}' ignored` - ); - return; - } - if (!session.proc.resolveApproval(frame.requestId, frame.decision, frame.reason)) { - // A decision for a request this side already settled (its timeout ran - // out first, or the relay dropped and came back). An `error` frame is a - // session failure to the relay, and this is not one — the CLI was - // answered; the late verdict is only worth a line here. - this.logger.warn( - `machine-runner: session '${frame.sessionId}' has no pending approval ` + - `'${frame.requestId}'; the decision arrived after it was settled` - ); - } - return; - } case 'enrolled': // The exchange's result: the tenant resolved this machine to a principal // and minted a credential to dial with next time. Its value never reaches @@ -446,49 +356,16 @@ export class EnrollmentRunner { const { sessionId, command } = frame; const args = frame.args ?? []; const interactive = frame.interactive === true; - const binding = agentBinding(frame); if (this.sessions.has(sessionId)) { ws.send(encodeFrame({ type: 'error', sessionId, message: 'session id already in use' })); return; } - if (binding && interactive) { - // The codec refuses this shape already; a relay that sends it anyway is - // asking for a terminal on a session that is by definition headless. - ws.send( - encodeFrame({ - type: 'error', - sessionId, - message: 'session bound to an agent run cannot be interactive' - }) - ); - return; - } const cols = frame.cols ?? 80; const rows = frame.rows ?? 24; let proc: SessionProcess; try { - if (binding?.agentMode === 'embedded') { - // The agent host is a command like any other: allow-listed, spawned on - // pipes, its stdio relayed. The binding reaches it as arguments. - proc = headlessProcess({ - policy: this.policy, - command, - args, - runId: binding.runId, - cwd: frame.cwd - }); - } else if (binding?.agentMode === 'cli') { - const adapter = adapterForCommand(command, args); - proc = cliProcess({ - adapter, - policy: this.policy, - command, - resume: binding.cliSessionId, - logger: this.logger - }); - } else { - proc = ptyProcess(resolveSpawn(this.policy, command, args), cols, rows); - } + const spec = resolveSpawn(this.policy, command, args, frame.cwd); + proc = interactive ? ptyProcess(spec, cols, rows) : pipeProcess(spec); } catch (err) { // A rejected command is the policy answering "no": the requester learns // why, the machine's log records it, and the connection stays up. @@ -504,7 +381,6 @@ export class EnrollmentRunner { const session: RunnerSession = { proc, interactive, - binding, cols, rows, streaming: true, @@ -512,24 +388,14 @@ export class EnrollmentRunner { truncated: false }; this.sessions.set(sessionId, session); - proc.onData(data => { + proc.onData((data, stream) => { // Buffered for an interactive session whether or not anyone is watching: // that buffer is the repaint a reattach needs. Command mode streams and // keeps nothing — its transcript is the ledger. if (session.interactive) this.remember(session, data); - if (session.streaming) this.send({ type: 'output', sessionId, data }); - }); - proc.onEvent?.(event => { - if (session.streaming) this.send({ type: 'agent_event', sessionId, event }); - }); - proc.onApprovalRequest?.(request => { - // Not gated on `streaming`: a headless session has nobody to stream to - // but the question still needs an answer, and the relay is who asks. - this.send({ type: 'approval_request', sessionId, ...request }); - }); - proc.onWarning?.(message => { - this.logger.warn(`machine-runner: session '${sessionId}': ${message}`); - if (session.streaming) this.send({ type: 'output', sessionId, data: `${message}\n` }); + if (session.streaming) { + this.send({ type: 'output', sessionId, data, ...(stream ? { stream } : {}) }); + } }); proc.onExit(({ exitCode, signal }) => { this.sessions.delete(sessionId); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 9564d4f1e7..504efdd1e5 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -111,6 +111,13 @@ importers: version: link:../openai/dist publishDirectory: dist + agentic/agent-cli: + dependencies: + '@constructive-db/machine-protocol': + specifier: workspace:* + version: link:../machine-protocol/dist + publishDirectory: dist + agentic/agent-conversation: publishDirectory: dist From d5b9c278ae93c127e53b84a0d68d9d8c8716c277 Mon Sep 17 00:00:00 2001 From: Dan Lynch Date: Fri, 25 Sep 2026 01:04:24 +0000 Subject: [PATCH 2/2] fix(agentic): symlink-proof cwd confinement, terminal spawn failures, no CLI spawn after the session concludes, complete lines survive an oversized tail --- agentic/agent-cli/__tests__/session.test.ts | 62 +++++++++++++++++++ agentic/agent-cli/src/session.ts | 36 +++++++---- .../machine-protocol/__tests__/codec.test.ts | 10 +++ agentic/machine-protocol/src/index.ts | 21 ++++++- .../machine-runner/__tests__/policy.test.ts | 27 +++++++- .../machine-runner/__tests__/process.test.ts | 24 +++++++ agentic/machine-runner/src/policy.ts | 27 +++++++- agentic/machine-runner/src/process.ts | 5 ++ agentic/machine-runner/src/runner.ts | 4 +- 9 files changed, 196 insertions(+), 20 deletions(-) create mode 100644 agentic/agent-cli/__tests__/session.test.ts create mode 100644 agentic/machine-runner/__tests__/process.test.ts diff --git a/agentic/agent-cli/__tests__/session.test.ts b/agentic/agent-cli/__tests__/session.test.ts new file mode 100644 index 0000000000..940a81d0fb --- /dev/null +++ b/agentic/agent-cli/__tests__/session.test.ts @@ -0,0 +1,62 @@ +import path from 'path'; +import { PassThrough } from 'stream'; + +import { ClaudeCodeAdapter, runAgentCliSession } from '../src'; + +const fixtures = path.join(__dirname, 'fixtures', 'bin'); +const env = { ...process.env, PATH: `${fixtures}:${process.env.PATH}` }; + +const collect = (stream: PassThrough): string[] => { + const chunks: string[] = []; + stream.on('data', chunk => chunks.push(String(chunk))); + return chunks; +}; + +describe('runAgentCliSession', () => { + it('runs the prompt through the CLI and reports its exit', async () => { + const stdin = new PassThrough(); + const stdout = new PassThrough(); + const stderr = new PassThrough(); + const out = collect(stdout); + collect(stderr); + const abort = new AbortController(); + const done = runAgentCliSession({ + adapter: new ClaudeCodeAdapter(), + io: { stdin, stdout, stderr }, + env, + abort: abort.signal + }); + stdin.write('hello\n'); + await new Promise(resolve => { + stdout.on('data', () => { + if (out.join('').includes('"kind":"result"')) resolve(); + }); + }); + abort.abort(); + expect((await done).signal).toBe('SIGTERM'); + const kinds = out.join('').trim().split('\n').map(line => JSON.parse(line).kind); + expect(kinds).toContain('session'); + expect(kinds).toContain('text'); + }); + + it('starts nothing once the session has concluded', async () => { + const stdin = new PassThrough(); + const stdout = new PassThrough(); + const stderr = new PassThrough(); + const out = collect(stdout); + collect(stderr); + const abort = new AbortController(); + const done = runAgentCliSession({ + adapter: new ClaudeCodeAdapter(), + io: { stdin, stdout, stderr }, + env, + abort: abort.signal + }); + abort.abort(); + expect(await done).toEqual({ exitCode: -1, signal: 'SIGTERM' }); + stdin.write('hello\n'); + stdin.end(); + await new Promise(resolve => setTimeout(resolve, 300)); + expect(out).toEqual([]); + }); +}); diff --git a/agentic/agent-cli/src/session.ts b/agentic/agent-cli/src/session.ts index 444712cccb..28662d7cb5 100644 --- a/agentic/agent-cli/src/session.ts +++ b/agentic/agent-cli/src/session.ts @@ -78,9 +78,16 @@ export function runAgentCliSession(options: AgentCliSessionOptions): Promise { io.stderr.write(`agent-cli: ${message}\n`); }; + /** The session has spoken; nothing that arrives on stdin afterwards may start or feed a CLI. */ + const conclude = (): void => { + exited = true; + io.stdin.removeListener('data', onStdinData); + io.stdin.removeListener('end', onStdinEnd); + io.stdin.removeListener('error', onStdinError); + }; const fail = (err: Error): void => { if (exited) return; - exited = true; + conclude(); for (const timer of pending.values()) clearTimeout(timer); pending.clear(); child?.kill(); @@ -182,7 +189,7 @@ export function runAgentCliSession(options: AgentCliSessionOptions): Promise { - if (line.length === 0) return; + if (exited || line.length === 0) return; const decision = parseApprovalDecisionLine(line); if (decision) { if (!settle(decision.requestId, decision.decision, decision.reason)) { @@ -223,19 +230,18 @@ export function runAgentCliSession(options: AgentCliSessionOptions): Promise { + function onStdinData(data: string): void { try { for (const line of inputLines.push(data)) onInputLine(line); } catch (err) { fail(err instanceof Error ? err : new Error(String(err))); } - }); - io.stdin.on('end', () => { + } + function onStdinEnd(): void { try { for (const line of inputLines.flush()) onInputLine(line); } catch (err) { @@ -243,11 +249,17 @@ export function runAgentCliSession(options: AgentCliSessionOptions): Promise fail(new Error(`agent-cli: stdin failed: ${err.message}`, { cause: err }))); + } + function onStdinError(err: Error): void { + fail(new Error(`agent-cli: stdin failed: ${err.message}`, { cause: err })); + } + io.stdin.setEncoding('utf8'); + io.stdin.on('data', onStdinData); + io.stdin.on('end', onStdinEnd); + io.stdin.on('error', onStdinError); }); } diff --git a/agentic/machine-protocol/__tests__/codec.test.ts b/agentic/machine-protocol/__tests__/codec.test.ts index 92d2095436..506fbca519 100644 --- a/agentic/machine-protocol/__tests__/codec.test.ts +++ b/agentic/machine-protocol/__tests__/codec.test.ts @@ -249,6 +249,16 @@ describe('machine-protocol codec', () => { expect(capped.flush()).toEqual([]); expect(capped.push('ok\n')).toEqual(['ok']); + // Complete lines ahead of an oversized tail in one chunk are delivered; + // the fault is raised by whatever the reader does next. + const mixed = new LineSplitter(8); + expect(mixed.push('one\ntwo\n123456789')).toEqual(['one', 'two']); + expect(() => mixed.push('x\n')).toThrow(/line exceeds 8 characters/); + expect(mixed.push('x\n')).toEqual(['x']); + const mixedFlush = new LineSplitter(8); + expect(mixedFlush.push('one\n123456789')).toEqual(['one']); + expect(() => mixedFlush.flush()).toThrow(/line exceeds 8 characters/); + expect(MAX_LINE_LENGTH).toBe(1024 * 1024); const wide = new LineSplitter(); expect(wide.push('x'.repeat(MAX_LINE_LENGTH))).toEqual([]); diff --git a/agentic/machine-protocol/src/index.ts b/agentic/machine-protocol/src/index.ts index e82a2b47b6..07d04114c4 100644 --- a/agentic/machine-protocol/src/index.ts +++ b/agentic/machine-protocol/src/index.ts @@ -375,20 +375,35 @@ export class LineSplitter { constructor(private readonly maxLineLength = MAX_LINE_LENGTH) {} + /** + * Complete lines are delivered before an oversized tail in the same chunk is + * reported: the failure is raised on the next call instead of losing them. + */ + private overflowed = false; + push(chunk: string): string[] { + this.raiseOverflow(); this.tail += chunk; const lines = this.tail.split('\n'); this.tail = lines.pop() ?? ''; if (this.tail.length > this.maxLineLength) { this.tail = ''; - throw new Error( - `machine-protocol: line exceeds ${this.maxLineLength} characters without a newline` - ); + this.overflowed = true; + if (lines.length === 0) this.raiseOverflow(); } return lines.map(stripCarriageReturn); } + private raiseOverflow(): void { + if (!this.overflowed) return; + this.overflowed = false; + throw new Error( + `machine-protocol: line exceeds ${this.maxLineLength} characters without a newline` + ); + } + flush(): string[] { + this.raiseOverflow(); if (!this.tail) return []; const line = stripCarriageReturn(this.tail); this.tail = ''; diff --git a/agentic/machine-runner/__tests__/policy.test.ts b/agentic/machine-runner/__tests__/policy.test.ts index dfd4330296..9583aacb8d 100644 --- a/agentic/machine-runner/__tests__/policy.test.ts +++ b/agentic/machine-runner/__tests__/policy.test.ts @@ -1,4 +1,8 @@ -import { DEFAULT_ENV_ALLOW, PolicyViolationError, resolveSpawn, RunnerPolicy } from '../src'; +import fs from 'fs'; +import os from 'os'; +import path from 'path'; + +import { DEFAULT_ENV_ALLOW, PolicyViolationError, resolveCwd, resolveSpawn, RunnerPolicy } from '../src'; const policy: RunnerPolicy = { allowedCommands: ['echo', 'ls'], @@ -39,3 +43,24 @@ describe('runner policy', () => { expect(spec.env.MACHINE_SESSION).toBe('1'); }); }); + +describe('cwd confinement', () => { + let root: string; + beforeEach(() => { + root = fs.mkdtempSync(path.join(fs.realpathSync.native(os.tmpdir()), 'runner-policy-')); + fs.mkdirSync(path.join(root, 'inside')); + fs.symlinkSync(os.tmpdir(), path.join(root, 'escape')); + }); + afterEach(() => fs.rmSync(root, { recursive: true, force: true })); + + it('keeps a session inside the root, following symlinks', () => { + const confined: RunnerPolicy = { allowedCommands: ['ls'], cwd: root }; + expect(resolveCwd(confined)).toBe(root); + expect(resolveCwd(confined, 'inside')).toBe(path.join(root, 'inside')); + expect(resolveCwd(confined, 'inside/not-yet-created')).toBe(path.join(root, 'inside/not-yet-created')); + expect(() => resolveCwd(confined, '..')).toThrow(PolicyViolationError); + expect(() => resolveCwd(confined, '/etc')).toThrow(PolicyViolationError); + expect(() => resolveCwd(confined, 'escape')).toThrow(PolicyViolationError); + expect(() => resolveCwd(confined, 'escape/anything')).toThrow(PolicyViolationError); + }); +}); diff --git a/agentic/machine-runner/__tests__/process.test.ts b/agentic/machine-runner/__tests__/process.test.ts new file mode 100644 index 0000000000..f52a7b9a75 --- /dev/null +++ b/agentic/machine-runner/__tests__/process.test.ts @@ -0,0 +1,24 @@ +import { pipeProcess } from '../src/process'; + +describe('pipeProcess', () => { + it('reports a program that cannot start once: an error, then the exit it never had', async () => { + const proc = pipeProcess({ command: '/nonexistent/program', args: [], cwd: process.cwd(), env: {} }); + const errors: string[] = []; + proc.onError(err => errors.push(err.message)); + const exit = await new Promise(resolve => proc.onExit(resolve)); + expect(errors).toHaveLength(1); + expect(errors[0]).toMatch(/ENOENT/); + expect(exit).toEqual({ exitCode: -1 }); + // Nothing left to write to or kill; neither may throw or report again. + proc.write('late\n'); + proc.kill('SIGTERM'); + await new Promise(resolve => setTimeout(resolve, 20)); + expect(errors).toHaveLength(1); + }); + + it('reports a normal exit with the code the program chose', async () => { + const proc = pipeProcess({ command: process.execPath, args: ['-e', 'process.exit(3)'], cwd: process.cwd(), env: {} }); + const exit = await new Promise(resolve => proc.onExit(resolve)); + expect(exit).toEqual({ exitCode: 3 }); + }); +}); diff --git a/agentic/machine-runner/src/policy.ts b/agentic/machine-runner/src/policy.ts index 3e25852aa4..483fe66fc5 100644 --- a/agentic/machine-runner/src/policy.ts +++ b/agentic/machine-runner/src/policy.ts @@ -3,6 +3,7 @@ // travels over the wire — a compromised relay can ask, but the answer is // decided here. +import fs from 'fs'; import os from 'os'; import path from 'path'; @@ -42,16 +43,36 @@ export interface SpawnSpec { env: Record; } +/** + * Where a path really points: symlinks resolved through the deepest ancestor + * that exists, the rest appended as written. A path that does not exist yet is + * judged by where it would be created. + */ +function realPath(target: string): string { + let existing = target; + const rest: string[] = []; + while (!fs.existsSync(existing)) { + const parent = path.dirname(existing); + if (parent === existing) return target; + rest.unshift(path.basename(existing)); + existing = parent; + } + return path.join(fs.realpathSync.native(existing), ...rest); +} + /** * The directory a session runs in: the policy root, or a requested directory - * resolved against it — and refused when it would land outside. The request - * is the client's; where the root is, is the machine owner's. + * resolved against it — and refused when it would land outside, symlinks + * included. The request is the client's; where the root is, is the machine + * owner's. */ export function resolveCwd(policy: RunnerPolicy, requested?: string): string { const root = path.resolve(policy.cwd || os.homedir()); if (requested === undefined) return root; const resolved = path.resolve(root, requested); - if (resolved !== root && !resolved.startsWith(root + path.sep)) { + const realRoot = realPath(root); + const real = realPath(resolved); + if (real !== realRoot && !real.startsWith(realRoot + path.sep)) { throw new PolicyViolationError(`cwd '${requested}' is outside the policy root`); } return resolved; diff --git a/agentic/machine-runner/src/process.ts b/agentic/machine-runner/src/process.ts index e4325ce35b..3436d6a070 100644 --- a/agentic/machine-runner/src/process.ts +++ b/agentic/machine-runner/src/process.ts @@ -73,8 +73,13 @@ export function pipeProcess(spec: SpawnSpec): SessionProcess { child.stderr.setEncoding('utf8').on('data', (data: string) => { for (const listener of dataListeners) listener(data, 'stderr'); }); + // A program that could not start (ENOENT, EACCES) reports once, as an + // error and then as the exit it never had; nothing follows. child.on('error', err => { + if (exited) return; + exited = true; for (const listener of errorListeners) listener(err); + for (const listener of exitListeners) listener({ exitCode: -1 }); }); child.on('exit', (code, signal) => { if (exited) return; diff --git a/agentic/machine-runner/src/runner.ts b/agentic/machine-runner/src/runner.ts index faf165c74e..086c0a58ee 100644 --- a/agentic/machine-runner/src/runner.ts +++ b/agentic/machine-runner/src/runner.ts @@ -401,10 +401,12 @@ export class EnrollmentRunner { this.sessions.delete(sessionId); if (session.streaming) this.send({ type: 'exit', sessionId, exitCode, signal }); }); + // The exit that follows — the process's own, or the one a failed start + // reports — is what ends the session. proc.onError(err => { - this.sessions.delete(sessionId); this.logger.error(`machine-runner: session '${sessionId}' process failed: ${err.message}`); this.send({ type: 'error', sessionId, message: `process failed: ${err.message}` }); + proc.kill('SIGTERM'); }); }