diff --git a/CHANGELOG.md b/CHANGELOG.md index 964919ed..db1b6eac 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,35 @@ # Changelog +## 0.118.0 + +### pi runs through the bridge, like every other harness + +BREAKING. The bespoke pi executor is gone. cli-bridge already routes pi generically — `matches(m) => m === 'pi' || m.startsWith('pi/')` — so a second pi-only code path was duplication. A pi worker is now `createExecutor({ backend: 'bridge', bridgeUrl, bridgeBearer, model: 'pi//' })`, the same call every other harness uses. + +Removed public exports: `piExecutor`, `PI_RUNTIME`, `piSeamKey`, `PiSeam`, `PiExecutorOutput`, `preparePiMcp`, `piMcpAdapterAvailable`, `buildPiMcpServers`, `PI_MCP_ADAPTER`, `PI_MCP_ADAPTER_ENV`, `PI_MCP_CONFIG_FLAG`, `PiMcpMount`, `PiMcpMountOptions`, `PiMcpPreparation`, `PiMcpReceipt`. The `{ backend: 'pi' }` arm of `ExecutorConfig` is removed, and the executor registry no longer pre-registers a `'pi'` runtime. + +What the bridge path does BETTER, and the old executor dropped outright: + +- `profile.model.reasoningEffort` is lowered to pi's `--thinking`. +- `profile.tools` is lowered to `--exclude-tools`. +- `profile.resources` (context / skills / commands / subagents / instructions) is materialized into the run directory. +- `profile.prompt.systemPrompt` is passed as a native `--system-prompt` file instead of being prepended to the task text. +- Consequently a pi backend is now held to `fullProfileMaterialization` in `workerFromBackend`, not the prompt-and-model-only contract. A profile that changes axes the old executor silently dropped is now honored rather than rejected. + +What the bridge path does NOT do today, stated so no consumer is surprised: + +- `Executor.progress()` and `Executor.traceSource()` are absent on `bridgeExecutor`. This is missing for every harness on the bridge, not only pi; tracked as #683. +- Mid-turn steering degrades. The bridge runs pi as `--print --mode json` with stdin closed, so a steer lands after the current run rather than at the next turn boundary, and an interrupt is a process-tree kill rather than pi's non-destructive injection. +- `PiSeam.args`, `PiSeam.env`, and a per-worker `bin` have no wire equivalent. A caller that pinned a specific pi binary or injected per-worker environment must configure it on the bridge server instead. +- `WORKER_TRACE_PROPAGATION` loses its `pi: true` row. The bridge has no environment channel to the worker, so a pi worker no longer inherits `TRACE_ID`/`PARENT_SPAN_ID` — honestly unpropagated rather than silently dropped, same as the other bridge-dispatched arms. +- MCP for pi is now the bridge's concern. agent-runtime no longer writes a `--mcp-config` file or checks for the `pi-mcp-adapter` extension. + +Known consumers to migrate, none broken until they upgrade: + +- `loops` — `src/pi-worker.ts` and `extensions/pi/loops.ts`. The largest migration: `pi-worker.ts` builds the executor directly, and steering is the part that changes behaviour rather than just call shape. Pinned at 0.111.0. +- `supervisor-lab` — `bench/deepswe/live.ts:193`. A single `backend: 'pi'` in a benchmark rig; a call-shape change. Pinned at 0.116.0. +- `agent-eval-runtime-run-reader` — two test files reference the pi backend. + ## 0.117.0 - Run every supervisor, including the root, from one complete `AgentProfile`, preserve exact profile/task/candidate identity through recursive delegation, and reject execution paths that would silently drop profile fields. diff --git a/docs/api/primitive-catalog.md b/docs/api/primitive-catalog.md index aba04424..15bed704 100644 --- a/docs/api/primitive-catalog.md +++ b/docs/api/primitive-catalog.md @@ -7,7 +7,7 @@ # Primitive catalog — the never-stale anti-reinvention inventory -> **GENERATED** from `@tangle-network/agent-runtime@0.117.0` and `@tangle-network/agent-eval@0.139.2` by `scripts/gen-primitive-catalog.mjs`. Do NOT hand-edit — run `pnpm run docs:api`. This is the mechanical companion to the JUDGMENT in `canonical-api.md` (§2 decision table + §1.5 AgentProfile law): that doc says WHICH primitive to reach for and what NOT to build; this catalog proves WHAT exists. Per-symbol signatures + `file:line` live in the per-module pages under `docs/api/`. +> **GENERATED** from `@tangle-network/agent-runtime@0.118.0` and `@tangle-network/agent-eval@0.139.2` by `scripts/gen-primitive-catalog.mjs`. Do NOT hand-edit — run `pnpm run docs:api`. This is the mechanical companion to the JUDGMENT in `canonical-api.md` (§2 decision table + §1.5 AgentProfile law): that doc says WHICH primitive to reach for and what NOT to build; this catalog proves WHAT exists. Per-symbol signatures + `file:line` live in the per-module pages under `docs/api/`. ## 1. agent-runtime — own public surface @@ -508,7 +508,7 @@ Import from `@tangle-network/agent-runtime/intelligence` — 166 exports. ### Execution kernel — recursive atom, supervision, executors, round-synchronous loop -Import from `@tangle-network/agent-runtime/kernel` — 700 exports. +Import from `@tangle-network/agent-runtime/kernel` — 685 exports. | Symbol | Kind | Summary | |---|---|---| @@ -530,7 +530,6 @@ Import from `@tangle-network/agent-runtime/kernel` — 700 exports. | `authorStrategy` | function | Author + load a strategy from losses. Throws when the author emits no loadable module; | | `bestSoFar` | function | The best-so-far fold — the ONE definition of "how good was the run after k results", shared by | | `breadthStrategy` | function | BREADTH: K independent rollouts (each own artifact), verifier picks the best. | -| `buildPiMcpServers` | function | Build the canonical `{ mcpServers }` body the adapter reads, from `AgentProfile.mcp`. | | `buildSteerContext` | function | Build the `SteerContext` a combinator reads to steer (its `loopUntil.until`, `widen` gate, any | | `canDisplace` | function | The repair keep-best guard: a challenger displaces the incumbent only when it is | | `canonicalizeAuthoredProfile` | function | Lift a profile the supervisor AUTHORED into the canonical shape every executor reads. | @@ -629,12 +628,10 @@ Import from `@tangle-network/agent-runtime/kernel` — 700 exports. | `pendingWaits` | function | The waits a journaled tree shows as ARMED but never woken — what a resumed run re-arms with the | | `pickBestDelivered` | function | The single argmax both the default finalizer and `finalizeBestDelivered` share: highest | | `pickChampion` | function | The champion pick over a means table. 'score' takes the best mean score (ties → | -| `piMcpAdapterAvailable` | function | True when pi can actually consume MCP config — i.e. `pi-mcp-adapter` is installed. Ported from | | `pipeline` | function | `pipeline(stages)` — run the stages in order, feeding each stage's `done` deliverable into the | | `plateau` | function | "The objective has stopped climbing." Fires when the best-so-far curve has risen by no more than | | `plateauLength` | function | How many trailing entries of a best-so-far curve are within `minDelta` of the curve's value | | `pollFor` | function | Build a bounded `poll` spec from a duration. | -| `preparePiMcp` | function | The one call `piExecutor` makes before spawning pi: decide what pi must be told, fail loud if it | | `printBenchmarkReport` | function | Pretty-print a report — the "free optimization" verdict, with the cost vector. | | `probeSandboxCapabilities` | function | Probe (and memoize per client) what the loop may rely on. A client without a | | `profileRichnessFinding` | function | Turn a {@link ProfileRichness} verdict into a bus-routable `AnalystFinding` (area `profile-quality`). | @@ -734,12 +731,6 @@ Import from `@tangle-network/agent-runtime/kernel` — 700 exports. | `EVIDENCE_MAX_CHARS` | const | Hard cap on one worker's evidence block so the brain's context cannot blow up. | | `mcpSecretEnvMetadataKey` | const | The `AgentProfileMcpServer.metadata` key the declarative secret-env map | | `NOTE_MAX_CHARS` | const | Cap on the worker's closing note inside the evidence block. | -| `PI_MCP_ADAPTER` | const | The pi extension that gives pi MCP at all. Everything here is gated on it being loadable. | -| `PI_MCP_ADAPTER_ENV` | const | Overrides adapter detection for installs whose path does not carry the adapter's name. | -| `PI_MCP_CONFIG_FLAG` | const | The adapter-registered flag that names this run's config file (`pi-mcp-adapter` `index.ts:252`). | -| `PI_RUNTIME` | const | The runtime name `piExecutor` registers under. | -| `piExecutor` | const | Build the `Executor` for one pi worker. Registered as runtime `'pi'`. | -| `piSeamKey` | const | Seam key the registry threads a `PiSeam` through (`ExecutorContext.seams['pi']`). | | `refine` | const | Built-in `Strategy`: attempt → `observe()` reads the trace → steer the next attempt → repeat (deepen one lineage). | | `runLoop` | const | Pre-rename name for {@link runAgentRounds}; identical function, kept so existing | | `sample` | const | Built-in `Strategy`: K independent attempts, keep the best-verifying (best-of-N / resample). | @@ -874,13 +865,7 @@ Import from `@tangle-network/agent-runtime/kernel` — 700 exports. | `Persona` | interface | The "act like X" record. A thin composition over the keystone's `AgentSpec`: it pairs the | | `PersonaContext` | interface | The persona context blob — who the loop is acting as. Open by intent: a persona names its | | `PersonaExecutors` | interface | How a persona supplies executor resolution. Either a pre-built registry (factories already | -| `PiExecutorOutput` | interface | What one pi run reports about the terminal assistant turn, plus any derived MCP mount. | -| `PiMcpMount` | interface | What the caller must call once pi has exited, and where the config landed. | -| `PiMcpMountOptions` | interface | Where one worker execution's private config directory is created, and what it is called. | -| `PiMcpPreparation` | interface | Everything `piExecutor` needs between "profile in hand" and "pi spawned". | -| `PiMcpReceipt` | interface | What pi was actually given, as opposed to what the profile declared. This is the observable that | | `PipelineStage` | interface | `pipeline(stages)` — sequential composition: each stage's `Outcome.deliverable` feeds the next | -| `PiSeam` | interface | How to launch pi in its out-of-process RPC mode, and how long to wait on it. | | `PriorCoordination` | interface | Coordination evidence loaded from prior processes of one durable supervised run. | | `ProfileRichness` | interface | Per-field verdict on one authored profile — the raw material the bench renders + scores. | | `ProfileRichnessThresholds` | interface | Thresholds below which a system prompt is treated as a thin stub. Tunable per call. | diff --git a/docs/api/runtime.md b/docs/api/runtime.md index 03e36d54..2580eff3 100644 --- a/docs/api/runtime.md +++ b/docs/api/runtime.md @@ -11045,192 +11045,6 @@ that command). Default `[]` — gate on no-op / secret / forbidden / diff-size o *** -### PiSeam - -How to launch pi in its out-of-process RPC mode, and how long to wait on it. - -#### Properties - -##### bin? - -> `optional` **bin?**: `string` - -The pi executable (default `'pi'`). Anything on PATH or an absolute path. - -##### args? - -> `optional` **args?**: readonly `string`[] - -Extra args appended after `--mode rpc`. `--provider` / `--model` are added from `model`. - -##### model? - -> `optional` **model?**: `string` - -`provider/model` or just `model` — split on the first `/` into pi's two flags. - -##### cwd? - -> `optional` **cwd?**: `string` - -##### env? - -> `optional` **env?**: `Record`\<`string`, `string`\> - -##### turnTimeoutMs? - -> `optional` **turnTimeoutMs?**: `number` - -Wall-clock ceiling for one `prompt` (the wait for `agent_settled`). Omit = no timeout. - -##### activityWindow? - -> `optional` **activityWindow?**: `number` - -Newest-last activity window `progress()` reports. Default 12. - -*** - -### PiExecutorOutput - -What one pi run reports about the terminal assistant turn, plus any derived MCP mount. - -#### Properties - -##### content - -> **content**: `string` - -##### turns - -> **turns**: `number` - -##### mcp? - -> `optional` **mcp?**: [`PiMcpReceipt`](#pimcpreceipt) - -Present only when `profile.mcp` declared at least one usable server. Records what pi was - actually given — including an extension this executor added that the profile did not list. - -*** - -### PiMcpMount - -What the caller must call once pi has exited, and where the config landed. - -#### Properties - -##### configPath - -> **configPath**: `string` - -Absolute path of the file passed to `--mcp-config`. Unique to this worker execution. - -##### serverNames - -> **serverNames**: `string`[] - -Server names actually written, post-filter — never the raw `profile.mcp` keys. - -#### Methods - -##### cleanup() - -> **cleanup**(): `void` - -Remove the private config directory. Idempotent; safe to call on any exit path. - -###### Returns - -`void` - -*** - -### PiMcpReceipt - -What pi was actually given, as opposed to what the profile declared. This is the observable that -makes rule 2 above honest: `adapterInjected` is `true` exactly when this module added an extension -the caller did not ask for, because without it the MCP servers the caller DID ask for could not -have mounted. - -#### Properties - -##### servers - -> **servers**: `string`[] - -Server names written into `configPath`, in declaration order. - -##### configPath - -> **configPath**: `string` - -Absolute path of the mounted config. - -##### extensions - -> **extensions**: `string`[] - -`extensions.pi.load` entries as pi received them, resolved to absolute entry files. - -##### adapterInjected - -> **adapterInjected**: `boolean` - -True when `pi-mcp-adapter` was absent from an explicit `load` array and was added here. - -*** - -### PiMcpPreparation - -Everything `piExecutor` needs between "profile in hand" and "pi spawned". - -#### Properties - -##### args - -> **args**: `string`[] - -Extra argv for `pi`, in flag order: extension flags first, then `--mcp-config `. - -##### mount - -> **mount**: [`PiMcpMount`](#pimcpmount) \| `null` - -Present only when at least one usable MCP server was declared. - -##### receipt - -> **receipt**: [`PiMcpReceipt`](#pimcpreceipt) \| `undefined` - -Present only when a mount happened — the derived-versus-declared record. - -*** - -### PiMcpMountOptions - -Where one worker execution's private config directory is created, and what it is called. - -#### Properties - -##### cwd? - -> `optional` **cwd?**: `string` - -The worker's own working directory (`PiSeam.cwd`). Omit when the seam names none: the config -then lands in the OS temp directory. It is NEVER written into `process.cwd()` — the operator's -own working directory is not a scratch space for a worker's config. - -##### runId - -> **runId**: `string` - -The executor's per-execution run id, folded into the directory name so a directory that somehow -survives names the worker that left it. Uniqueness is NOT taken from this — `mkdtemp` provides -it — because two workers built from one factory in the same millisecond share a run id stem. - -*** - ### ActivityNote The most recent activity the executor can name — one tool call, one turn, or a free-form note. @@ -13597,7 +13411,7 @@ breaker, or a recursive parent. ###### Inherited from -[`SupervisorNodeContext`](#supervisornodecontext).[`runId`](#runid-14) +[`SupervisorNodeContext`](#supervisornodecontext).[`runId`](#runid-13) ##### runNamespace @@ -18627,7 +18441,7 @@ The stores a supervised run needs, in-memory or file-backed. `InMemoryRunContext ### ExecutorConfig -> **ExecutorConfig** = `object` & [`RouterSeam`](#routerseam) \| `object` & [`RouterToolsSeam`](#routertoolsseam) \| `object` & [`BridgeSeam`](#bridgeseam) \| `object` & [`CliSeam`](#cliseam) \| `object` & [`CliWorktreeSeam`](#cliworktreeseam) \| `object` & [`ProviderSeam`](#providerseam) \| `object` & [`PiSeam`](#piseam) \| `object` & [`SandboxSeam`](#sandboxseam) +> **ExecutorConfig** = `object` & [`RouterSeam`](#routerseam) \| `object` & [`RouterToolsSeam`](#routertoolsseam) \| `object` & [`BridgeSeam`](#bridgeseam) \| `object` & [`CliSeam`](#cliseam) \| `object` & [`CliWorktreeSeam`](#cliworktreeseam) \| `object` & [`ProviderSeam`](#providerseam) \| `object` & [`SandboxSeam`](#sandboxseam) Config for [createExecutor](#createexecutor): the backend is DATA — the cost dial a profile, an experiment config, or a replay journal can name — not an import choice. Each @@ -19728,54 +19542,6 @@ an empty collection is a no-winner, not a winner wrapping `[]`. *** -### PI\_RUNTIME - -> `const` **PI\_RUNTIME**: [`Runtime`](#runtime-4) = `'pi'` - -The runtime name `piExecutor` registers under. - -*** - -### piSeamKey - -> `const` **piSeamKey**: `"pi"` = `'pi'` - -Seam key the registry threads a `PiSeam` through (`ExecutorContext.seams['pi']`). - -*** - -### piExecutor - -> `const` **piExecutor**: [`ExecutorFactory`](#executorfactory)\<`unknown`\> - -Build the `Executor` for one pi worker. Registered as runtime `'pi'`. - -*** - -### PI\_MCP\_ADAPTER - -> `const` **PI\_MCP\_ADAPTER**: `"pi-mcp-adapter"` = `'pi-mcp-adapter'` - -The pi extension that gives pi MCP at all. Everything here is gated on it being loadable. - -*** - -### PI\_MCP\_ADAPTER\_ENV - -> `const` **PI\_MCP\_ADAPTER\_ENV**: `"AGENT_RUNTIME_PI_MCP_ADAPTER"` = `'AGENT_RUNTIME_PI_MCP_ADAPTER'` - -Overrides adapter detection for installs whose path does not carry the adapter's name. - -*** - -### PI\_MCP\_CONFIG\_FLAG - -> `const` **PI\_MCP\_CONFIG\_FLAG**: `"--mcp-config"` = `'--mcp-config'` - -The adapter-registered flag that names this run's config file (`pi-mcp-adapter` `index.ts:252`). - -*** - ### DEFAULT\_STALL\_AFTER\_MS > `const` **DEFAULT\_STALL\_AFTER\_MS**: `180000` = `180_000` @@ -23625,116 +23391,6 @@ whether the patch is DELIVERED (the `valid` conjunction). *** -### piMcpAdapterAvailable() - -> **piMcpAdapterAvailable**(): `boolean` - -True when pi can actually consume MCP config — i.e. `pi-mcp-adapter` is installed. Ported from -cli-bridge `pi.ts:189`: an env override first (for vendored installs whose path does not carry the -adapter's name), then the npm install directory, then a scan of `settings.json` `packages` in -which a local-path entry is resolved by reading the target's `package.json` `name`. An -unreadable or absent `settings.json` reports "not detected"; it never throws and never assumes. - -#### Returns - -`boolean` - -*** - -### buildPiMcpServers() - -> **buildPiMcpServers**(`mcp`): `Record`\<`string`, `Record`\<`string`, `unknown`\>\> - -Build the canonical `{ mcpServers }` body the adapter reads, from `AgentProfile.mcp`. - -Every server is written with `directTools: true` (`pi-mcp-adapter` types.ts:351), which -registers its tools as NATIVE pi tools instead of hiding them behind the generic `mcp` tool. -Without it an agent must discover its own verbs before it can use them — connect to the server, -then describe each tool — and a measured run spent 58 turns and 639,632 input tokens doing that -before it could delegate once. A profile that declares an MCP server is asking for its tools, -not for a directory it has to browse. - -This is cli-bridge's `buildCanonicalMcpServers` -(`/home/drew/code/cli-bridge/src/backends/profile-support.ts:253`) reproduced field for field and -in the same key order — same drop of `enabled:false`, same stdio-versus-remote decision -(`isStdioMcpSpec`, `profile-support.ts:177`), same `{command, args, env}` and `{type, url, -headers}` bodies, same silent drop of an entry with nothing to connect to. For any server both -input types can express, the emitted JSON is byte-identical. - -The two input types are not the same type, so three differences remain and each is a field one -side cannot express: - -1. **`cwd` (emitted here, absent there).** `AgentProfileLocalMcpServer` has `cwd`; cli-bridge's - `McpServerSpec` (`src/backends/types.ts:53`) has no such field, so it can never emit one. The - adapter honors it (`pi-mcp-adapter` `types.ts:325`, `ServerEntry.cwd`), so dropping it would - silently discard a declared setting. It is emitted where cli-bridge emits `timeout`. -2. **`timeout` (emitted there, absent here).** `AgentProfileMcpServer` has no timeout field, so - there is nothing to emit. (The adapter's own knob is `requestTimeoutMs`, not `timeout`.) -3. **A `url` with no declared transport.** cli-bridge DROPS it — its remote branch requires an - explicit `spec.type` — even though its own `McpServerSpec` docstring says "http is implied if - `url` is set" (`types.ts:53-56`). `transport` is optional on `AgentProfileRemoteMcpServer`, so - dropping such a server here would be exactly the silent-drop bug this module exists to fix. - We honor the documented convention and emit `type: 'http'`. - -`sse` is ACCEPTED and forwarded verbatim, as cli-bridge forwards it. The adapter connects a URL -server by probing StreamableHTTP and falling back to `SSEClientTransport` when the probe fails -(`pi-mcp-adapter` `server-manager.ts:765`), so an SSE-only endpoint does connect. Its -`ServerEntry` (`types.ts:319`) carries no transport discriminator at all, which means the `type` -key is inert to the adapter either way — but it is what cli-bridge writes, and a config file that -round-trips the caller's declared transport is the one that stays readable by both. - -#### Parameters - -##### mcp - -`Record`\<`string`, `AgentProfileMcpServer`\> \| `undefined` - -#### Returns - -`Record`\<`string`, `Record`\<`string`, `unknown`\>\> - -*** - -### preparePiMcp() - -> **preparePiMcp**(`profile`, `options`): [`PiMcpPreparation`](#pimcppreparation) - -The one call `piExecutor` makes before spawning pi: decide what pi must be told, fail loud if it -cannot work, then mount. - -Order matters and is the point: - -1. Canonicalize `profile.mcp` first, so the adapter gate keys off servers that would ACTUALLY be - written — a block of `enabled:false` entries must not trip it. -2. If any server survives and `pi-mcp-adapter` is not installed, throw. No file has been written - and pi has not been spawned. -3. Lower `extensions.pi.load` (which may itself throw on an unresolvable extension) — still before - any file exists. -4. Only then create this execution's private config directory and emit `--mcp-config`. - -A profile with no usable MCP server is a pure no-op: no adapter check, no directory, no flags, no -throw. `extensions.pi.load` on its own still produces flags, because reproducible ablation arms -need them whether or not MCP is involved. - -The caller MUST call `mount.cleanup()` on every exit path — settle, throw, abort, timeout, and -spawn failure — so the directory does not outlive the worker that owns it. - -#### Parameters - -##### profile - -`AgentProfile` \| `undefined` - -##### options - -[`PiMcpMountOptions`](#pimcpmountoptions) - -#### Returns - -[`PiMcpPreparation`](#pimcppreparation) - -*** - ### createActivityLog() > **createActivityLog**(`limit?`): [`ActivityLog`](#activitylog) diff --git a/docs/architecture.md b/docs/architecture.md index 63c3b2a3..cb3896d3 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -442,11 +442,11 @@ Salience filtering and the cross-box durable mailbox are not built; see **§13.6 - **REAL** — a local external-harness supervisor runs automatically through a `bridge` `driverBackend ?? backend` with the live coordination MCP injected under one reserved alias. Its own tools, resources, MCP servers, hooks, subagents, permissions, modes, prompt, and model remain profile data sent to that backend (`supervise/supervise.ts`, `supervise/runtime.ts`). A pre-execution `materialized` journal event binds the authored-profile, effective-profile, and platform-attachment digests to the node that ran. -- **REAL** — a cli-bridge worker is observable mid-run like a pi one: `bridgeExecutor.progress()` reports its live turn count, tool activity, queued steers, and what it derived about the caller's declaration, and `bridgeExecutor.traceSource()` feeds the online detectors and the settle-time analysts (`supervise/runtime.ts`). +- **REAL** — a cli-bridge worker is observable mid-run: `bridgeExecutor.progress()` reports its live turn count, tool activity, queued steers, and what it derived about the caller's declaration, and `bridgeExecutor.traceSource()` feeds the online detectors and the settle-time analysts (`supervise/runtime.ts`). Its bridge-side run state (`GET /v1/runs/:id`) is refreshed OUT OF BAND — a read schedules the fetch and never awaits it — so an observability read can neither block nor fail a live run. - **LIMIT** — a bridge worker's tool spans are INSTANTS with no status: the OpenAI-shaped `tool_calls` wire reports the model's decision to call a tool and never reports the call finishing, so no duration or outcome exists to read. - Only `piExecutor` has true per-tool durations, because pi's RPC stream carries `tool_execution_start`/`_end`. - Order, counts, names, and arguments are the same on both; per-tool latency and per-tool error rate are pi-only. + A harness whose native protocol reports tool completion could carry true durations; this wire does not. + Order, counts, names, and arguments are all present; per-tool latency and per-tool error rate are not. - **LIMIT** — a remote sandbox cannot reach the loopback coordination server automatically. It needs an explicit `driveHarness` that provides a reachable relay or tunnel. - **LIMIT** — the in-process router arm has no environment in which to materialize profile resources, hooks, subagents, permissions, or modes. diff --git a/docs/canonical-api.md b/docs/canonical-api.md index 60145481..22447e0f 100644 --- a/docs/canonical-api.md +++ b/docs/canonical-api.md @@ -4,7 +4,7 @@ Generated signatures and the complete export list live in docs/api/. Run pnpm docs:freshness after editing this file. --> -> **Version 0.117.0.** +> **Version 0.118.0.** > [`docs/api/primitive-catalog.md`](./api/primitive-catalog.md) lists every export and import path. > `agent-eval` must satisfy `>=0.139.2 <0.140.0`. > `sandbox` must satisfy `>=0.15.0 <0.16.0`. diff --git a/package.json b/package.json index 08f85ed2..6db0dced 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@tangle-network/agent-runtime", - "version": "0.117.0", + "version": "0.118.0", "description": "Shared task-lifecycle skeleton for agents: a recursive loop kernel for chat turns, one-shot tasks, and multi-attempt loops, with trace capture and eval-gated self-improvement. Domain behavior lives in adapters; scoring and ship-gates in @tangle-network/agent-eval.", "homepage": "https://github.com/tangle-network/agent-runtime#readme", "repository": { diff --git a/src/runtime/index.ts b/src/runtime/index.ts index 4977ac7e..9ad84e47 100644 --- a/src/runtime/index.ts +++ b/src/runtime/index.ts @@ -637,36 +637,6 @@ export { // The mechanical patch gate as a generic DeliverableSpec over the worktree-CLI patch artifact: // no-op / always-on secret-path floor / forbidden-path / diff-size + required test/typecheck pass. export { type PatchDeliverableOptions, patchDelivered } from './supervise/patch-deliverable' -// pi WRAPPED, not forked: `piExecutor` speaks pi's own out-of-process RPC protocol, so its -// steering queue, session persistence, abort, and compaction stay upstream's. Registered as -// runtime `'pi'` through the documented `ExecutorRegistry.register` extension point. -export { - PI_RUNTIME, - type PiExecutorOutput, - type PiSeam, - piExecutor, - piSeamKey, -} from './supervise/pi-executor' -// pi has NO native MCP: it comes from the `pi-mcp-adapter` extension, which reads the canonical -// `{mcpServers}` object from the file its `--mcp-config` flag names. `piExecutor` writes that file -// from `profile.mcp` before spawn — one file per worker EXECUTION, never one per workspace, so the -// ordinary supervisor case of many children behind one shared `PiSeam.cwd` cannot collide. It -// refuses to start when the adapter is missing, and RECORDS on `PiExecutorOutput.mcp` (and on the -// live `progress().derived` channel, which survives a failed run) any extension it had to add to -// keep `--no-extensions` from suppressing the adapter. Exported so a caller can perform the same -// check ahead of a spawn. -export { - buildPiMcpServers, - PI_MCP_ADAPTER, - PI_MCP_ADAPTER_ENV, - PI_MCP_CONFIG_FLAG, - type PiMcpMount, - type PiMcpMountOptions, - type PiMcpPreparation, - type PiMcpReceipt, - piMcpAdapterAvailable, - preparePiMcp, -} from './supervise/pi-mcp' // The LIVE read-model of a RUNNING worker — last activity, idle time, derived stall, turns, // tokens so far, recent tool/file activity, unread steers. What `observe_agent` now returns // mid-flight, and the evidence a supervisor steers FROM. diff --git a/src/runtime/supervise/pi-executor.ts b/src/runtime/supervise/pi-executor.ts deleted file mode 100644 index ec767e43..00000000 --- a/src/runtime/supervise/pi-executor.ts +++ /dev/null @@ -1,875 +0,0 @@ -/** - * - * `piExecutor` — pi wrapped behind `Executor`, NOT forked. - * - * pi already implements everything a steerable worker needs, and implements it well: a queued - * steering channel delivered between turns, follow-ups, abort, compaction, session persistence - * with fork/branch, and an out-of-process line-delimited JSON protocol over stdin/stdout - * (`pi --mode rpc`). Reimplementing any of that here would mean owning a turn loop someone else - * already maintains. So this module is a thin protocol adapter, and every capability maps onto a - * verb pi already has: - * - * `execute` → `prompt`, draining pi's event stream until `agent_settled` - * `deliver` → `prompt` with `streamingBehavior` — pi owns the queue, we do not - * `teardown` → `abort`, then close stdin and reap the process - * `progress` → pi's `tool_execution_start`/`_end` + `turn_end` events, plus `get_state`'s - * `pendingMessageCount` mirrored locally so the read stays synchronous - * `traceSource` → the same tool events decoded into the shared `ToolSpan` currency - * `resultArtifact` → the terminal successful assistant turn - * - * It is registered through the DOCUMENTED extension point (`ExecutorRegistry.register('pi', …)`), - * so nothing in the resolver switches on it and a consumer can replace it wholesale. - * - * agent-runtime does NOT take a dependency on pi. The wire shapes (`RpcCommand`, `AgentEvent`) - * are read structurally off JSON lines, so a pi that adds commands stays compatible and a pi that - * is not installed simply fails loud at spawn instead of at import. - * - * Usage accounting: pi repeats the same assistant receipt in `message_end` and `turn_end`. - * Only `turn_end` is counted, once per model call. Pi reports cache traffic separately from fresh - * input, so all three input classes are folded into Runtime's two-field token total. Subscription - * usage whose dollar price is absent or zero stays explicitly unknown through the live usage - * stream and terminal artifact. - * - * ## What of the `AgentProfile` this executor honors, and what it does not - * - * This adapter is NOT a profile materializer. It reads a deliberately small part of - * `spec.profile`, and a caller who needs the rest must lower it before handing the profile over. - * Stated exhaustively so a silent drop is never a surprise: - * - * | Field | Status | - * | --- | --- | - * | `profile.name` | honored — trace `runId` label only, no behavioral effect | - * | `profile.prompt.systemPrompt` | honored — prepended to the task text (pi RPC takes no separate system-prompt channel) | - * | `profile.mcp` | honored — written to this execution's own file and passed as `--mcp-config` for `pi-mcp-adapter`; see `pi-mcp.ts` | - * | `profile.extensions.pi.load` | honored — lowered to `--no-extensions` + `--extension ` | - * | `profile.prompt.instructions` | honored — appended to the system prompt, one per line | - * | `profile.model.default` | honored — overrides the seam's `model`; the seam is the fallback for profiles that select none | - * | `profile.model.reasoningEffort` | DROPPED — no `--thinking` flag is emitted, so pi's configured `defaultThinkingLevel` applies | - * | `profile.tools` | DROPPED — no `--no-tools` / allow-deny mapping; pi runs its full builtin tool set | - * | `profile.permissions` | DROPPED | - * | `profile.resources` (context / skills / commands / subagents / instructions) | DROPPED — nothing is written into the run cwd | - * | `profile.hooks` | DROPPED | - * | `profile.subagents`, `profile.connections`, `profile.modes`, `profile.confidential` | DROPPED | - * | `profile.resources.failOnError` | not consulted — the MCP path above is unconditionally fail-closed, which is the stricter reading | - * | every other `profile.extensions.` | DROPPED — only the `pi` namespace is read | - * - * Closing those gaps belongs in `@tangle-network/agent-profile-materialize`, whose plan this - * executor would then apply, rather than in a second mapping grown here. - * - * @experimental - */ - -import { type ChildProcess, spawn } from 'node:child_process' -import { randomUUID } from 'node:crypto' -import type { AgentProfile } from '@tangle-network/agent-interface' -import { ValidationError } from '../../errors' -import { abortError, throwIfAborted } from '../util' -import { createInbox, type Inbox, type InboxMessage } from './inbox' -import { attestRuntimeOwnedExecutor, newExecutionAttemptId } from './materialization' -import { PI_MCP_ADAPTER, type PiMcpReceipt, preparePiMcp } from './pi-mcp' -import { - type ActivityLog, - createActivityLog, - describeToolArgs, - type ExecutorProgress, -} from './progress' -import { createPushTraceSource, type ToolStepInput, type TraceSource } from './trace-source' -import type { - ExecutorContext, - ExecutorFactory, - ExecutorResult, - Runtime, - Spend, - UsageEvent, -} from './types' -import { workerTraceEnv } from './worker-trace' - -/** The runtime name `piExecutor` registers under. */ -export const PI_RUNTIME: Runtime = 'pi' - -/** Seam key the registry threads a `PiSeam` through (`ExecutorContext.seams['pi']`). */ -export const piSeamKey = 'pi' - -/** How to launch pi in its out-of-process RPC mode, and how long to wait on it. */ -export interface PiSeam { - /** The pi executable (default `'pi'`). Anything on PATH or an absolute path. */ - bin?: string - /** Extra args appended after `--mode rpc`. `--provider` / `--model` are added from `model`. */ - args?: ReadonlyArray - /** `provider/model` or just `model` — split on the first `/` into pi's two flags. */ - model?: string - cwd?: string - env?: Record - /** Wall-clock ceiling for one `prompt` (the wait for `agent_settled`). Omit = no timeout. */ - turnTimeoutMs?: number - /** Newest-last activity window `progress()` reports. Default 12. */ - activityWindow?: number -} - -/** Structural read of Pi's stdout records: agent events plus correlated RPC responses. */ -interface PiEvent { - id?: string - type?: string - command?: string - success?: boolean - error?: string - toolCallId?: string - toolName?: string - args?: unknown - result?: unknown - isError?: boolean - message?: unknown -} - -interface PendingPiTool { - args: unknown - startedAt: number -} - -interface PiAssistantOutcome { - text: string - stopReason?: string - errorMessage?: string -} - -/** Build the `Executor` for one pi worker. Registered as runtime `'pi'`. */ -export const piExecutor: ExecutorFactory = (spec, ctx) => { - const configured = readPiSeam(ctx) - const seam: PiSeam = { - ...configured, - // The backend model is a fallback for profiles that do not select one. AgentProfile is the - // experiment-owned knob, so an ambient/default seam must never override the authored arm. - ...(spec.profile.model?.default ? { model: spec.profile.model.default } : {}), - } - // `TRACE_ID` / `PARENT_SPAN_ID` for this worker when the run records spans; `{}` otherwise, which - // leaves the spawn environment byte-identical to the untraced path. - const traceEnv = workerTraceEnv(ctx) - const inbox = createInbox() - const activity = createActivityLog(seam.activityWindow ?? 12) - // One id per worker EXECUTION, not per factory: it labels the trace and names this worker's - // private MCP config directory, and two workers built from one seam must not share either. - const runId = `pi-${spec.profile.name ?? 'worker'}-${Date.now()}` - const trace = createPushTraceSource({ runId }) - const executionId = ctx.node?.nodeId ?? `pi-run-${randomUUID()}` - const attemptId = ctx.node?.attemptId ?? newExecutionAttemptId(executionId) - // What this executor changed about what the caller declared. Unlike `recentActivity` this is - // never evicted, so a run that fails after the change still reports it. - const derived: string[] = [] - - const controller = new AbortController() - const cascade = () => controller.abort() - if (ctx.signal.aborted) controller.abort() - else ctx.signal.addEventListener('abort', cascade, { once: true }) - - const state = { - turns: 0, - lastText: '', - note: 'starting', - proc: undefined as ChildProcess | undefined, - artifact: undefined as ExecutorResult | undefined, - } - - const executor: ReturnType> = { - runtime: PI_RUNTIME, - // pi owns the queue; `deliver` only routes through its state-safe `prompt` command. Its - // streaming behavior chooses steer versus follow-up atomically in pi, rather than trusting - // this adapter's delayed view of whether the current run has already ended. - deliver: (m) => inbox.deliver(m), - progress: (): ExecutorProgress => ({ - turns: state.turns, - pendingMessages: inbox.pending(), - recentActivity: activity.read(), - ...(derived.length > 0 ? { derived: [...derived] } : {}), - note: state.note, - }), - traceSource: (): TraceSource => trace.source, - execute(task, signal): AsyncIterable { - return streamPiSession({ - task, - signal, - controller, - seam, - traceEnv, - spec, - runId, - inbox, - activity, - derived, - record: (step: ToolStepInput) => { - trace.record(step) - }, - state, - }) - }, - async teardown(grace): Promise<{ destroyed: boolean }> { - controller.abort() - const proc = state.proc - if (!proc || proc.exitCode !== null || proc.killed) return { destroyed: true } - // Ask pi to stop cleanly first — an `abort` lets it finalize its session JSONL, which is - // the whole reason to wrap pi rather than kill it. - sendCommand(proc, { type: 'abort' }) - return killPi(proc, grace) - }, - resultArtifact() { - if (!state.artifact) { - throw new ValidationError('piExecutor: resultArtifact() read before stream drained') - } - return state.artifact - }, - } - // Attestation binds the KERNEL-minted attempt id. An executor built outside a Scope (no - // `ctx.node`) has no kernel identity to bind, so it makes no attestation claim and a later - // scope spawn records honest unknown receipts instead of a mismatched binding. - if (ctx.node === undefined) return executor - return attestRuntimeOwnedExecutor( - executor, - { - effectiveProfile: spec.profile, - backend: 'pi', - model: seam.model - ? { status: 'known', id: seam.model } - : { status: 'unknown', reason: 'pi selected its configured default model' }, - execution: { kind: 'run', id: executionId }, - materializer: 'pi-rpc-agent-profile', - plan: { - kind: 'pi-rpc-session', - bin: seam.bin ?? 'pi', - args: seam.args ?? [], - cwd: seam.cwd ?? null, - model: seam.model ?? null, - turnTimeoutMs: seam.turnTimeoutMs ?? null, - }, - }, - { - attemptId, - binding: { - executionId, - bin: seam.bin ?? 'pi', - cwd: seam.cwd ?? null, - model: seam.model ?? null, - }, - descriptor: { kind: 'pi-rpc-run', transport: 'process', backend: 'pi' }, - }, - ) -} - -/** What one pi run reports about the terminal assistant turn, plus any derived MCP mount. */ -export interface PiExecutorOutput { - content: string - turns: number - /** Present only when `profile.mcp` declared at least one usable server. Records what pi was - * actually given — including an extension this executor added that the profile did not list. */ - mcp?: PiMcpReceipt -} - -interface StreamPiArgs { - task: unknown - signal: AbortSignal - controller: AbortController - seam: PiSeam - /** Inherited `TRACE_ID` / `PARENT_SPAN_ID` for the pi subprocess; empty when tracing is off. */ - traceEnv: Record - spec: { profile: AgentProfile } - /** This execution's id — labels the trace and names the private MCP config directory. */ - runId: string - inbox: Inbox - activity: ActivityLog - /** Append-only record of what this executor changed about the caller's declaration. */ - derived: string[] - record: (step: ToolStepInput) => void - state: { - turns: number - lastText: string - note: string - proc: ChildProcess | undefined - artifact: ExecutorResult | undefined - } -} - -/** - * One pi RPC session, run to `agent_settled`. Every steer delivered while the turn is in flight is - * forwarded to pi IMMEDIATELY (pi queues and delivers it at its own turn boundary — the queue we - * deliberately do not reimplement), and any steer still unread when pi settles re-prompts it. - * `agent_end` is deliberately not terminal: Pi may auto-retry or compact after emitting it. - */ -async function* streamPiSession(args: StreamPiArgs): AsyncIterable { - const { seam, inbox, activity, state } = args - const started = Date.now() - const tokens = { input: 0, output: 0 } - let usd = 0 - let usdKnown = true - let tokensKnown = true - throwIfAborted(args.signal) - throwIfAborted(args.controller.signal) - - // Profile-declared MCP servers become this execution's OWN `--mcp-config` file BEFORE pi starts, - // because `pi-mcp-adapter` reads that path at load time. The file is private to this worker, so - // two workers built from one seam never share it. This THROWS — before any file is written and - // before pi is spawned — when the adapter that gives pi MCP at all is missing, rather than - // starting a worker whose declared tools silently do not exist. - const piMcp = preparePiMcp(args.spec.profile, { - ...(seam.cwd !== undefined ? { cwd: seam.cwd } : {}), - runId: args.runId, - }) - if (piMcp.receipt) { - const injected = piMcp.receipt.adapterInjected ? ` (+${PI_MCP_ADAPTER} added)` : '' - activity.push({ - at: Date.now(), - kind: 'note', - label: 'mcp', - detail: `${piMcp.receipt.servers.join(', ')}${injected}`, - }) - // `recentActivity` is a bounded ring and `resultArtifact()` throws until the stream drains, so - // neither can answer "what was pi actually given?" for a run that failed on turn 40. This can. - args.derived.push( - `mcp: mounted ${piMcp.receipt.servers.join(', ')} via --mcp-config ${piMcp.receipt.configPath}`, - ) - if (piMcp.receipt.adapterInjected) { - args.derived.push( - `extensions: added ${PI_MCP_ADAPTER} to extensions.pi.load — the profile's own list would ` + - 'have run under --no-extensions and suppressed it, mounting zero servers', - ) - } - } - - let proc: ChildProcess - try { - proc = spawnPi(seam, piMcp.args, args.traceEnv) - } catch (spawnFailure) { - // The config directory was created before the spawn was attempted; a spawn that never happened - // still owes its removal. - piMcp.mount?.cleanup() - throw spawnFailure - } - state.proc = proc - state.note = 'connected' - - const events: PiEvent[] = [] - const pendingTools = new Map() - const awaitingPromptResponses = new Set() - let settled = false - let acceptedPromptSinceSettlement = false - let promptSequence = 0 - let lastAssistant: PiAssistantOutcome | undefined - let failure: Error | undefined - let abortDeadline: number | undefined - let exited = false - let exitCode: number | null | undefined - let wake: (() => void) | undefined - const notify = () => { - const w = wake - wake = undefined - w?.() - } - - const stdoutLines = readJsonLines(proc, (value) => { - const ev = value as PiEvent - events.push(ev) - notify() - }) - - proc.once('exit', (code) => { - exited = true - exitCode = code - notify() - }) - proc.once('error', (e) => { - failure = new ValidationError(`piExecutor: pi failed to start: ${e.message}`) - notify() - }) - - const abortAll = () => { - if (abortDeadline !== undefined) return - sendCommand(proc, { type: 'abort' }) - abortDeadline = Date.now() + PI_ABORT_RECEIPT_MS - state.note = 'aborting' - notify() - } - args.signal.addEventListener('abort', abortAll, { once: true }) - args.controller.signal.addEventListener('abort', abortAll, { once: true }) - - const system = [ - args.spec.profile.prompt?.systemPrompt, - ...(args.spec.profile.prompt?.instructions ?? []), - ] - .filter((line): line is string => typeof line === 'string' && line.trim().length > 0) - .join('\n') - const opening = system ? `${system}\n\n${taskText(args.task)}` : taskText(args.task) - const deadline = seam.turnTimeoutMs ? Date.now() + seam.turnTimeoutMs : undefined - const sendPrompt = (message: string, streamingBehavior?: 'steer' | 'followUp'): void => { - const id = `agent-runtime-prompt-${++promptSequence}` - awaitingPromptResponses.add(id) - sendCommand(proc, { - id, - type: 'prompt', - message, - ...(streamingBehavior ? { streamingBehavior } : {}), - }) - } - - try { - // Close the check→listener race without ever dispatching a cancelled task. - if (args.signal.aborted || args.controller.signal.aborted) throw abortError() - sendPrompt(opening) - state.note = 'turn 0' - - for (;;) { - // Forward anything the driver delivered — pi's own queue is the single source of truth - // for ordering, so this is a route, not a second queue. - if (!settled && abortDeadline === undefined) forwardPending(inbox, activity, sendPrompt) - - // Drain what pi has emitted so far, projecting usage + activity. - while (events.length > 0) { - const ev = events.shift() as PiEvent - if ( - ev.type === 'response' && - ev.command === 'prompt' && - typeof ev.id === 'string' && - awaitingPromptResponses.delete(ev.id) - ) { - if (ev.success === true) { - // A settlement that preceded this acceptance belongs to an older prompt. Require a - // later `agent_settled` before ending the session. - acceptedPromptSinceSettlement = true - } else { - failure = new ValidationError( - `piExecutor: Pi rejected prompt: ${ev.error ?? 'unknown RPC error'}`, - ) - } - } - if (ev.type === 'agent_start') settled = false - if (ev.type === 'agent_settled') { - settled = true - acceptedPromptSinceSettlement = false - } - const projected = projectPiEvent(ev, args, tokens, pendingTools) - if (projected.assistant) lastAssistant = projected.assistant - if (projected.tokensUnknown) tokensKnown = false - for (const usage of projected.events) { - if (usage.kind === 'cost') { - usd += usage.usd - if (usage.usdKnown === false) usdKnown = false - } - yield usage - } - } - - if (failure) throw failure - if (abortDeadline !== undefined && (settled || exited)) throw abortError() - if ( - abortDeadline === undefined && - exited && - (!settled || awaitingPromptResponses.size > 0 || acceptedPromptSinceSettlement) - ) { - throw new ValidationError( - `piExecutor: pi exited before agent_settled (code ${exitCode ?? 'unknown'})`, - ) - } - // Once cancellation starts, Pi's terminal receipt gets its own bounded drain window. - if (abortDeadline === undefined && deadline !== undefined && Date.now() > deadline) { - throw new ValidationError('piExecutor: turn exceeded turnTimeoutMs') - } - if (abortDeadline !== undefined && Date.now() > abortDeadline) { - const error = abortError() - error.message = `piExecutor: abort did not settle within ${PI_ABORT_RECEIPT_MS}ms` - throw error - } - - if (settled && awaitingPromptResponses.size === 0 && !acceptedPromptSinceSettlement) { - // A steer delivered in the settle gap starts a new Pi prompt. Pi has declared the prior - // session activity fully quiet, so no retry/compaction can race this transition. - const pending = inbox.drain() - if (pending.length === 0) break - settled = false - lastAssistant = undefined - sendPrompt(inbox.fold(pending)) - state.note = `turn ${state.turns}` - continue - } - - await new Promise((resolve) => { - wake = resolve - // A short fence keeps the loop responsive to a steer that arrives with no pi output. - const t = setTimeout(() => { - if (wake === resolve) { - wake = undefined - resolve() - } - }, 50) - if (typeof t.unref === 'function') t.unref() - }) - } - } finally { - stdoutLines() - state.note = 'settled' - args.signal.removeEventListener('abort', abortAll) - args.controller.signal.removeEventListener('abort', abortAll) - await killPi(proc, 2_000).catch(() => ({ destroyed: false })) - // Every exit path — settle, throw, abort, turn timeout — removes this execution's config - // directory. Only after pi is reaped, so a still-running pi can never observe it half-removed. - piMcp.mount?.cleanup() - } - - if (args.signal.aborted || args.controller.signal.aborted) throw abortError() - if (!lastAssistant) { - throw new ValidationError('piExecutor: agent_settled without an assistant turn') - } - if (lastAssistant.stopReason === 'aborted') throw abortError() - if (lastAssistant.stopReason === 'error') { - throw new ValidationError( - `piExecutor: Pi assistant failed: ${lastAssistant.errorMessage ?? 'unknown provider error'}`, - ) - } - state.lastText = lastAssistant.text - - const spent: Spend = { - iterations: state.turns, - tokens, - ...(tokensKnown ? {} : { tokensKnown: false }), - usd, - ...(usdKnown ? {} : { usdKnown: false }), - ms: Date.now() - started, - } - const out: PiExecutorOutput = { - content: state.lastText, - turns: state.turns, - ...(piMcp.receipt ? { mcp: piMcp.receipt } : {}), - } - state.artifact = { - outRef: `pi:${hash(state.lastText)}`, - out, - spent, - } -} - -/** Route messages through pi's state-safe prompt command; pi owns the queue and idle transition. */ -function forwardPending( - inbox: Inbox, - activity: ActivityLog, - sendPrompt: (message: string, streamingBehavior?: 'steer' | 'followUp') => void, -): void { - const pending = inbox.drain() - for (const m of pending) { - sendPrompt(renderOne(m), m.interrupt ? 'steer' : 'followUp') - activity.push({ - at: Date.now(), - kind: 'note', - label: m.interrupt ? 'steer' : 'follow-up', - detail: m.text.length > 80 ? `${m.text.slice(0, 77)}...` : m.text, - }) - } -} - -function renderOne(m: InboxMessage): string { - if (m.kind === 'answer') { - return `Answer to your question${m.questionId ? ` (${m.questionId})` : ''}: ${m.text}` - } - return `New instruction from your supervisor: ${m.text}` -} - -/** Project one pi event onto usage events, updating turn count / activity / trace as a side - * effect. pi's tool events are the live trace; its `turn_end` bumps the turn counter. */ -function projectPiEvent( - ev: PiEvent, - args: StreamPiArgs, - tokens: { input: number; output: number }, - pendingTools: Map, -): { events: UsageEvent[]; assistant?: PiAssistantOutcome; tokensUnknown?: true } { - const out: UsageEvent[] = [] - const at = Date.now() - if (ev.type === 'tool_execution_start' && typeof ev.toolName === 'string') { - args.activity.push({ at, kind: 'tool', label: ev.toolName, detail: describeToolArgs(ev.args) }) - if (typeof ev.toolCallId === 'string') { - pendingTools.set(ev.toolCallId, { args: ev.args ?? {}, startedAt: at }) - } - return { events: out } - } - if (ev.type === 'tool_execution_end' && typeof ev.toolName === 'string') { - const status = ev.isError === true ? 'error' : 'ok' - const callId = typeof ev.toolCallId === 'string' ? ev.toolCallId : undefined - const started = callId ? pendingTools.get(callId) : undefined - if (callId) pendingTools.delete(callId) - const error = status === 'error' ? describeToolError(ev.result) : undefined - args.activity.push({ at, kind: 'tool', label: ev.toolName, status }) - args.record({ - toolName: ev.toolName, - args: started?.args ?? {}, - ...(started ? {} : { argsCaptured: false }), - status, - ...(ev.result !== undefined ? { result: ev.result } : {}), - ...(error !== undefined ? { error } : {}), - ...(callId ? { callId } : {}), - ...(started ? { startedAt: started.startedAt } : {}), - endedAt: at, - }) - return { events: out } - } - // Pi emits this for user, assistant, and toolResult messages. `turn_end.message` is the - // authoritative assistant receipt; a generic message can never become the result artifact. - if (ev.type === 'message_end') return { events: out } - if (ev.type === 'turn_end') { - const usage = readUsage(ev.message) - if (usage && (usage.input || usage.output)) { - tokens.input += usage.input - tokens.output += usage.output - out.push({ kind: 'tokens', input: usage.input, output: usage.output }) - } - out.push( - usage?.usd !== undefined - ? { kind: 'cost', usd: usage.usd } - : { kind: 'cost', usd: 0, usdKnown: false }, - ) - args.state.turns += 1 - args.activity.push({ at, kind: 'turn', label: `turn ${args.state.turns}` }) - out.push({ kind: 'iteration' }) - const assistant = readAssistantOutcome(ev.message) - // A turn whose receipt named no token field at all did real work with an unreported count. - // The terminal artifact must carry that as `tokensKnown: false`, never as a silent zero. - const tokensUnknown = !usage || usage.tokensKnown === false - return { - events: out, - ...(assistant ? { assistant } : {}), - ...(tokensUnknown ? { tokensUnknown: true as const } : {}), - } - } - return { events: out } -} - -/** Pi's fresh input excludes cache reads and writes. Runtime's input channel includes all model - * input, so combine them once at the assistant receipt. A missing or zero price is unknown because - * subscription-backed providers report zero even when compute was not free. */ -function readUsage( - message: unknown, -): { input: number; output: number; usd?: number; tokensKnown: boolean } | undefined { - if (!message || typeof message !== 'object') return undefined - const usage = (message as { usage?: unknown }).usage - if (!usage || typeof usage !== 'object') return undefined - const u = usage as Record - const promptTokens = num(u.prompt_tokens) - const freshInput = num(u.input) ?? num(u.inputTokens) - const outputRaw = num(u.output) ?? num(u.outputTokens) ?? num(u.completion_tokens) - const input = - promptTokens ?? - (freshInput ?? 0) + - (num(u.cacheRead) ?? num(u.cache_read_input_tokens) ?? num(u.cacheReadInputTokens) ?? 0) + - (num(u.cacheWrite) ?? - num(u.cache_creation_input_tokens) ?? - num(u.cacheCreationInputTokens) ?? - 0) - const output = outputRaw ?? 0 - const costRaw = u.cost - const reportedUsd = - num(costRaw) ?? - (costRaw && typeof costRaw === 'object' - ? (num((costRaw as Record).total) ?? - num((costRaw as Record).totalCost)) - : undefined) - return { - input, - output, - ...(reportedUsd !== undefined && reportedUsd > 0 ? { usd: reportedUsd } : {}), - // A usage object that named NO token field is a receipt without a count, not a zero. - tokensKnown: promptTokens !== undefined || freshInput !== undefined || outputRaw !== undefined, - } -} - -function readAssistantOutcome(message: unknown): PiAssistantOutcome | undefined { - if (!message || typeof message !== 'object') return undefined - const value = message as Record - if (value.role !== 'assistant') return undefined - return { - text: readText(message) ?? '', - ...(typeof value.stopReason === 'string' ? { stopReason: value.stopReason } : {}), - ...(typeof value.errorMessage === 'string' ? { errorMessage: value.errorMessage } : {}), - } -} - -function readText(message: unknown): string | undefined { - if (!message || typeof message !== 'object') return undefined - const content = (message as { content?: unknown }).content - if (typeof content === 'string' && content.length > 0) return content - if (!Array.isArray(content)) return undefined - const text = content - .map((block) => - block && typeof block === 'object' && typeof (block as { text?: unknown }).text === 'string' - ? ((block as { text: string }).text ?? '') - : '', - ) - .join('') - return text.length > 0 ? text : undefined -} - -function num(v: unknown): number | undefined { - return typeof v === 'number' && Number.isFinite(v) && v >= 0 ? v : undefined -} - -function describeToolError(result: unknown): string | undefined { - if (typeof result === 'string' && result.length > 0) return result - if (result && typeof result === 'object') { - const value = result as Record - const direct = value.error ?? value.message - if (typeof direct === 'string' && direct.length > 0) return direct - if (Array.isArray(value.content)) { - const text = value.content - .map((block) => - block && - typeof block === 'object' && - typeof (block as { text?: unknown }).text === 'string' - ? (block as { text: string }).text - : '', - ) - .filter(Boolean) - .join('\n') - if (text.length > 0) return text - } - } - return undefined -} - -function taskText(task: unknown): string { - if (typeof task === 'string') return task - try { - return JSON.stringify(task) ?? String(task) - } catch { - return String(task) - } -} - -/** - * Launch `pi --mode rpc`, plus provider/model flags derived from the seam's `provider/model` and - * the profile-derived extension flags (`--no-extensions` / `--extension`). Seam args go LAST so an - * operator's explicit flag wins over a derived one under pi's last-flag-wins parsing; RPC mode has - * no positional prompt, so nothing here has to precede an argument. - * - * `traceEnv` is the inherited `TRACE_ID` / `PARENT_SPAN_ID` pair (empty when the run records no - * spans). It sits ABOVE the supervisor's ambient `process.env` — a supervisor that was itself - * launched as someone's worker holds ids describing ITS place in an outer trace, which are the - * wrong parent for this child — and BELOW `seam.env`, so an operator who sets either id explicitly - * still wins. See `worker-trace.ts` for the full precedence rule. - */ -function spawnPi( - seam: PiSeam, - profileArgs: ReadonlyArray = [], - traceEnv: Record = {}, -): ChildProcess { - const bin = seam.bin ?? 'pi' - const argv = ['--mode', 'rpc'] - if (seam.model) { - const slash = seam.model.indexOf('/') - if (slash > 0) { - argv.push('--provider', seam.model.slice(0, slash), '--model', seam.model.slice(slash + 1)) - } else { - argv.push('--model', seam.model) - } - } - argv.push(...profileArgs) - if (seam.args) argv.push(...seam.args) - return spawn(bin, argv, { - ...(seam.cwd ? { cwd: seam.cwd } : {}), - env: { ...process.env, ...traceEnv, ...(seam.env ?? {}) }, - stdio: ['pipe', 'pipe', 'pipe'], - }) -} - -/** Write one `RpcCommand` as a JSON line on pi's stdin. Best-effort: a dead process is not an - * error here (the stream loop already observes the exit and fails loud there). */ -function sendCommand(proc: ChildProcess, command: Record): void { - try { - proc.stdin?.write(`${JSON.stringify(command)}\n`) - } catch { - // The exit/error handlers own the failure path. - } -} - -/** Read newline-delimited JSON off pi's stdout; returns an unsubscribe. Malformed lines are - * ignored (pi writes only JSONL on stdout; anything else is noise from a wrapper script). */ -function readJsonLines(proc: ChildProcess, onValue: (value: unknown) => void): () => void { - let buffer = '' - const onData = (chunk: Buffer | string) => { - buffer += typeof chunk === 'string' ? chunk : chunk.toString('utf8') - for (;;) { - const nl = buffer.indexOf('\n') - if (nl < 0) break - const line = buffer.slice(0, nl).trim() - buffer = buffer.slice(nl + 1) - if (!line) continue - try { - onValue(JSON.parse(line)) - } catch { - // not JSON — ignore - } - } - } - proc.stdout?.on('data', onData) - return () => { - proc.stdout?.off('data', onData) - } -} - -/** The grace window pi gets to act on the `abort` it was just sent, before a signal is used. - * Without it the SIGTERM races the command down the pipe and pi never sees the abort at all — - * which defeats the reason to wrap pi rather than kill it (a clean abort finalizes its session). */ -const PI_ABORT_GRACE_MS = 500 -/** Maximum wait for Pi's aborted receipt and `agent_settled` before forced teardown. */ -const PI_ABORT_RECEIPT_MS = 2_000 - -async function killPi( - proc: ChildProcess, - grace: number | 'brutalKill' | 'infinity', -): Promise<{ destroyed: boolean }> { - if (proc.exitCode !== null || proc.killed) return { destroyed: true } - if (grace === 'brutalKill') { - endStdin(proc) - proc.kill('SIGKILL') - return { destroyed: true } - } - const total = grace === 'infinity' ? 10_000 : Math.max(0, grace) - // Phase 1: let the already-sent `abort` land and pi exit on its own terms. - if (await waitForExit(proc, Math.min(PI_ABORT_GRACE_MS, total))) return { destroyed: true } - // Phase 2: close its input and ask the OS politely. - endStdin(proc) - proc.kill('SIGTERM') - if (await waitForExit(proc, Math.max(0, total - PI_ABORT_GRACE_MS))) return { destroyed: true } - proc.kill('SIGKILL') - return { destroyed: true } -} - -function endStdin(proc: ChildProcess): void { - try { - proc.stdin?.end() - } catch { - // already closed - } -} - -function waitForExit(proc: ChildProcess, ms: number): Promise { - if (proc.exitCode !== null) return Promise.resolve(true) - return new Promise((resolve) => { - const timer = setTimeout(() => resolve(false), ms) - if (typeof timer.unref === 'function') timer.unref() - proc.once('exit', () => { - clearTimeout(timer) - resolve(true) - }) - }) -} - -function readPiSeam(ctx: ExecutorContext): PiSeam { - const seam = ctx.seams[piSeamKey] - if (seam === undefined) return {} - if (!seam || typeof seam !== 'object') { - throw new ValidationError(`piExecutor: seams['${piSeamKey}'] must be a PiSeam object`) - } - return seam as PiSeam -} - -/** FNV-1a over the terminal text — the same non-cryptographic content address the other leaf - * executors mint for their `outRef` dedup hint (the journal re-derives the canonical one). */ -function hash(value: string): string { - let h = 0x811c9dc5 - for (let i = 0; i < value.length; i += 1) { - h ^= value.charCodeAt(i) - h = Math.imul(h, 0x01000193) - } - return (h >>> 0).toString(16).padStart(8, '0') -} diff --git a/src/runtime/supervise/pi-mcp.ts b/src/runtime/supervise/pi-mcp.ts deleted file mode 100644 index ee565e66..00000000 --- a/src/runtime/supervise/pi-mcp.ts +++ /dev/null @@ -1,534 +0,0 @@ -/** - * - * MCP for the `pi` harness — the mount `piExecutor` performs before it spawns pi. - * - * pi ships NO MCP support of its own. MCP arrives through the `pi-mcp-adapter` extension, and the - * adapter takes its config from a path the CALLER names: it registers a `--mcp-config ` flag - * (`pi-mcp-adapter` `index.ts:252`, read back at `init.ts:101` and off raw argv at `utils.ts:55`). - * So "give pi an MCP server" means: write the canonical `{ mcpServers: … }` object to a path this - * run owns, pass that path with `--mcp-config`, make sure the adapter is loaded, and delete the - * file afterwards. This module does exactly that and nothing else. - * - * Three rules carry the whole design: - * - * 1. **Fail loud, never fail safe.** If the profile declares MCP servers and `pi-mcp-adapter` is not - * installed, `preparePiMcp` throws BEFORE any file is written and before pi is spawned. A run - * whose tools never existed must fail, not score zero for the wrong reason. - * 2. **`--no-extensions` is a trap.** When `extensions.pi.load` is an array, pi runs - * `--no-extensions`, which suppresses ALL auto-discovery — cwd `.pi/extensions`, global - * `~/.pi/agent/extensions`, AND the `settings.json` `packages` list that would otherwise have - * auto-loaded the adapter. So `load: ['pi-memory']` alongside `mcp` kills MCP dead even though - * the adapter IS installed. When MCP is declared, this module adds `pi-mcp-adapter` to the - * resolved load list and REPORTS that it did (`PiMcpReceipt.adapterInjected`). The caller's - * profile object is never mutated. - * 3. **One config file per worker EXECUTION, never one per workspace.** `PiSeam.cwd` comes from the - * `ExecutorConfig` and is therefore shared by every worker the factory builds, so a config - * discovered by cwd is a file two concurrent siblings fight over. `mountPiMcpConfig` instead - * creates a fresh `mkdtemp` directory — OS-guaranteed unique, so no counter, lock, merge or - * restore is needed — and hands its path to `--mcp-config`. Concurrent workers built from ONE - * seam get ONE config each and never collide. - * - * Why the flag is safe to emit here even though pi only lists it under `Extension CLI Flags`: it is - * emitted ONLY when the profile declares at least one usable MCP server, and that case has already - * passed the `piMcpAdapterAvailable()` gate above, so the extension that registers the flag is - * guaranteed loadable (auto-discovered, or force-loaded by the `--extension` this module appends). - * A machine without the adapter never reaches the flag — it fails at the gate with an install - * command instead. - * - * Where the run's servers sit in pi's config layering, stated plainly because it is a real - * behavioral edge: `--mcp-config` REPLACES pi's user-global override file (`config.ts:310`, - * `getPiGlobalConfigPath(overridePath)`), so the operator's own `~/.pi/agent/mcp.json` servers no - * longer leak into a worker — good. It does NOT suppress project discovery: the adapter still - * layers `/.mcp.json` and `/.pi/mcp.json` on top (`config.ts:352-373`), and those are - * later sources, so a workspace that ships a server under the SAME NAME overrides this run's entry. - * The consequence is worth naming: an A/B whose arms differ only in `profile.mcp` is confounded - * when the shared workspace carries its own MCP config. Deleting a file the run did not create - * would be worse — a run cannot quietly edit the repo it was pointed at. - * - * Upstream note: `@tangle-network/agent-profile-materialize` is the shared owner of profile → harness - * mapping and already declares `pi` a first-class `HarnessId`, but as of 0.9.4 its pi arm answers - * `unsupported('mcp', 'pi MCP requires a loaded MCP extension; the base Pi runtime has no MCP - * control')` and its pi extensions arm answers `unsupported('extensions', …)`. That refusal is now - * factually stale — the shape is exactly its `kimi-code` arm's (write a config file, push a config - * flag, because the harness has no cwd discovery worth trusting). Teaching the materializer these - * two rows is the durable home for this mapping; until it lands, this module is the only thing in - * the `backend: 'pi'` path that can honor `profile.mcp` at all. - * - * Verified against pi 0.83.0 and pi-mcp-adapter 2.15.0. - * - * @experimental - */ - -import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs' -import { homedir, tmpdir } from 'node:os' -import { isAbsolute, join, sep } from 'node:path' -import type { - AgentProfile, - AgentProfileConfigValue, - AgentProfileMcpServer, -} from '@tangle-network/agent-interface' -import { ConfigError, ValidationError } from '../../errors' - -/** The pi extension that gives pi MCP at all. Everything here is gated on it being loadable. */ -export const PI_MCP_ADAPTER = 'pi-mcp-adapter' - -/** Overrides adapter detection for installs whose path does not carry the adapter's name. */ -export const PI_MCP_ADAPTER_ENV = 'AGENT_RUNTIME_PI_MCP_ADAPTER' - -/** The adapter-registered flag that names this run's config file (`pi-mcp-adapter` `index.ts:252`). */ -export const PI_MCP_CONFIG_FLAG = '--mcp-config' - -/** What the caller must call once pi has exited, and where the config landed. */ -export interface PiMcpMount { - /** Absolute path of the file passed to `--mcp-config`. Unique to this worker execution. */ - configPath: string - /** Server names actually written, post-filter — never the raw `profile.mcp` keys. */ - serverNames: string[] - /** Remove the private config directory. Idempotent; safe to call on any exit path. */ - cleanup(): void -} - -/** - * What pi was actually given, as opposed to what the profile declared. This is the observable that - * makes rule 2 above honest: `adapterInjected` is `true` exactly when this module added an extension - * the caller did not ask for, because without it the MCP servers the caller DID ask for could not - * have mounted. - */ -export interface PiMcpReceipt { - /** Server names written into `configPath`, in declaration order. */ - servers: string[] - /** Absolute path of the mounted config. */ - configPath: string - /** `extensions.pi.load` entries as pi received them, resolved to absolute entry files. */ - extensions: string[] - /** True when `pi-mcp-adapter` was absent from an explicit `load` array and was added here. */ - adapterInjected: boolean -} - -/** Everything `piExecutor` needs between "profile in hand" and "pi spawned". */ -export interface PiMcpPreparation { - /** Extra argv for `pi`, in flag order: extension flags first, then `--mcp-config `. */ - args: string[] - /** Present only when at least one usable MCP server was declared. */ - mount: PiMcpMount | null - /** Present only when a mount happened — the derived-versus-declared record. */ - receipt: PiMcpReceipt | undefined -} - -/** pi's agent home. Holds `settings.json` and `npm/node_modules/`. */ -export function piAgentDir(): string { - return process.env.PI_CODING_AGENT_DIR ?? join(homedir(), '.pi', 'agent') -} - -/** `/npm`, whose `node_modules/` holds installed pi extensions. */ -export function piNpmRoot(): string { - return join(piAgentDir(), 'npm') -} - -/** - * True when pi can actually consume MCP config — i.e. `pi-mcp-adapter` is installed. Ported from - * cli-bridge `pi.ts:189`: an env override first (for vendored installs whose path does not carry the - * adapter's name), then the npm install directory, then a scan of `settings.json` `packages` in - * which a local-path entry is resolved by reading the target's `package.json` `name`. An - * unreadable or absent `settings.json` reports "not detected"; it never throws and never assumes. - */ -export function piMcpAdapterAvailable(): boolean { - const override = process.env[PI_MCP_ADAPTER_ENV] - if (override === '1' || override === 'true') return true - if (override === '0' || override === 'false') return false - const agentDir = piAgentDir() - if (existsSync(join(agentDir, 'npm', 'node_modules', PI_MCP_ADAPTER))) return true - try { - const settings = JSON.parse(readFileSync(join(agentDir, 'settings.json'), 'utf-8')) as { - packages?: unknown - } - if (!Array.isArray(settings.packages)) return false - return settings.packages.some((entry) => packageEntryIsAdapter(entry, agentDir)) - } catch { - // unreadable/absent settings — fall through to "not detected" - return false - } -} - -/** One `settings.json` `packages[]` entry: a name containing the adapter, or a local path to it. */ -function packageEntryIsAdapter(entry: unknown, agentDir: string): boolean { - if (typeof entry !== 'string') return false - if (entry.includes(PI_MCP_ADAPTER)) return true - // Local-path installs (`/some/dir`, `./rel`, `file:…`, `path:…`) may not carry the adapter's name - // in the path. Relative specs resolve against the agent dir (where settings.json lives), never the - // host process cwd. - const spec = entry.replace(/^(file|path):(\/\/)?/, '') - const windowsAbsolute = /^[A-Za-z]:[\\/]/.test(spec) - if (!isAbsolute(spec) && !windowsAbsolute && !spec.startsWith('.')) return false - const localPath = isAbsolute(spec) || windowsAbsolute ? spec : join(agentDir, spec) - return readPackageName(localPath) === PI_MCP_ADAPTER -} - -function readPackageName(packageDir: string): string | undefined { - try { - const manifest = JSON.parse(readFileSync(join(packageDir, 'package.json'), 'utf-8')) as { - name?: unknown - } - return typeof manifest.name === 'string' ? manifest.name : undefined - } catch { - return undefined - } -} - -/** - * Read one `AgentProfileConfigValue` as public bytes. A `secret-ref` is REFUSED rather than - * stringified: this executor has no secret provider, so the only alternatives are writing a - * placeholder into the config (an auth failure disguised as a tool bug) or writing the literal - * reference object (nonsense to the adapter). Plain strings, numbers and booleans are accepted - * because profiles authored as JSON commonly carry them where the type says `{kind:'public'}`. - */ -function publicConfigString( - value: AgentProfileConfigValue | string | number | boolean, - where: string, -): string { - if (typeof value === 'string') return value - if (typeof value === 'number' || typeof value === 'boolean') return String(value) - if (value && typeof value === 'object' && value.kind === 'public') return value.value - if (value && typeof value === 'object' && value.kind === 'secret-ref') { - throw new ConfigError( - `piExecutor: ${where} is a secret-ref ("${value.key}") and piExecutor has no secret provider; ` + - 'resolve it before handing the profile to the executor, or declare a public value', - ) - } - throw new ValidationError(`piExecutor: ${where} is not a public configuration value`) -} - -/** - * Build the canonical `{ mcpServers }` body the adapter reads, from `AgentProfile.mcp`. - * - * Every server is written with `directTools: true` (`pi-mcp-adapter` types.ts:351), which - * registers its tools as NATIVE pi tools instead of hiding them behind the generic `mcp` tool. - * Without it an agent must discover its own verbs before it can use them — connect to the server, - * then describe each tool — and a measured run spent 58 turns and 639,632 input tokens doing that - * before it could delegate once. A profile that declares an MCP server is asking for its tools, - * not for a directory it has to browse. - * - * This is cli-bridge's `buildCanonicalMcpServers` - * (`/home/drew/code/cli-bridge/src/backends/profile-support.ts:253`) reproduced field for field and - * in the same key order — same drop of `enabled:false`, same stdio-versus-remote decision - * (`isStdioMcpSpec`, `profile-support.ts:177`), same `{command, args, env}` and `{type, url, - * headers}` bodies, same silent drop of an entry with nothing to connect to. For any server both - * input types can express, the emitted JSON is byte-identical. - * - * The two input types are not the same type, so three differences remain and each is a field one - * side cannot express: - * - * 1. **`cwd` (emitted here, absent there).** `AgentProfileLocalMcpServer` has `cwd`; cli-bridge's - * `McpServerSpec` (`src/backends/types.ts:53`) has no such field, so it can never emit one. The - * adapter honors it (`pi-mcp-adapter` `types.ts:325`, `ServerEntry.cwd`), so dropping it would - * silently discard a declared setting. It is emitted where cli-bridge emits `timeout`. - * 2. **`timeout` (emitted there, absent here).** `AgentProfileMcpServer` has no timeout field, so - * there is nothing to emit. (The adapter's own knob is `requestTimeoutMs`, not `timeout`.) - * 3. **A `url` with no declared transport.** cli-bridge DROPS it — its remote branch requires an - * explicit `spec.type` — even though its own `McpServerSpec` docstring says "http is implied if - * `url` is set" (`types.ts:53-56`). `transport` is optional on `AgentProfileRemoteMcpServer`, so - * dropping such a server here would be exactly the silent-drop bug this module exists to fix. - * We honor the documented convention and emit `type: 'http'`. - * - * `sse` is ACCEPTED and forwarded verbatim, as cli-bridge forwards it. The adapter connects a URL - * server by probing StreamableHTTP and falling back to `SSEClientTransport` when the probe fails - * (`pi-mcp-adapter` `server-manager.ts:765`), so an SSE-only endpoint does connect. Its - * `ServerEntry` (`types.ts:319`) carries no transport discriminator at all, which means the `type` - * key is inert to the adapter either way — but it is what cli-bridge writes, and a config file that - * round-trips the caller's declared transport is the one that stays readable by both. - */ -export function buildPiMcpServers( - mcp: Record | undefined, -): Record> { - const out: Record> = {} - if (!mcp || typeof mcp !== 'object') return out - for (const [name, raw] of Object.entries(mcp)) { - if (!name || !raw || typeof raw !== 'object') continue - if (raw.enabled === false) continue - const transport = raw.transport - const command = typeof raw.command === 'string' && raw.command.length > 0 ? raw.command : null - const url = typeof raw.url === 'string' && raw.url.length > 0 ? raw.url : null - // cli-bridge `isStdioMcpSpec`: an explicit stdio transport still needs a command; an explicit - // remote transport is never stdio; with no transport declared, a command decides. - const stdio = transport === 'http' || transport === 'sse' ? false : command !== null - if (stdio && command !== null) { - const args = Array.isArray(raw.args) - ? raw.args.map((a, i) => publicConfigString(a, `mcp["${name}"].args[${i}]`)) - : [] - const env = readConfigRecord(raw.env, `mcp["${name}"].env`) - out[name] = { - command, - ...(args.length > 0 ? { args } : {}), - ...(Object.keys(env).length > 0 ? { env } : {}), - ...(typeof raw.cwd === 'string' && raw.cwd.length > 0 ? { cwd: raw.cwd } : {}), - directTools: true, - } - continue - } - if (url !== null && transport !== 'stdio') { - const headers = readConfigRecord(raw.headers, `mcp["${name}"].headers`) - out[name] = { - type: transport === 'sse' ? 'sse' : 'http', - url, - ...(Object.keys(headers).length > 0 ? { headers } : {}), - directTools: true, - } - } - // No command and no url — nothing to connect to. Dropped, exactly as cli-bridge drops it. - } - return out -} - -function readConfigRecord( - record: Record | undefined, - where: string, -): Record { - if (!record || typeof record !== 'object') return {} - const out: Record = {} - for (const [key, value] of Object.entries(record)) { - if (value === undefined) continue - out[key] = publicConfigString(value, `${where}["${key}"]`) - } - return out -} - -/** - * Resolve one `extensions.pi.load` entry to an absolute extension entry file. Ported from cli-bridge - * `resolvePiExtensionEntry` (`pi.ts:123`): an absolute path is used verbatim; a bare package name - * resolves to `/node_modules//`, `main` defaulting to `index.ts`. - * The resolution is required because pi's per-file loader does NOT resolve bare specifiers against - * `~/.pi/agent/npm/node_modules`. Fail-closed: an unreadable manifest or a missing entry file - * rejects the run rather than letting an ablation arm run without the extension it declared. - */ -export function resolvePiExtensionEntry(spec: string, npmRoot: string): string { - let entry: string - if (isAbsolute(spec)) { - entry = spec - } else { - const packageDir = join(npmRoot, 'node_modules', spec) - let main = 'index.ts' - try { - const manifest = JSON.parse(readFileSync(join(packageDir, 'package.json'), 'utf-8')) as { - main?: unknown - } - if (typeof manifest.main === 'string' && manifest.main) main = manifest.main - } catch (err) { - throw new ConfigError( - `piExecutor: cannot load pi extension "${spec}": ${join(packageDir, 'package.json')} is ` + - `not readable (install it with \`pi install npm:${spec}\`): ` + - `${err instanceof Error ? err.message : String(err)}`, - ) - } - entry = join(packageDir, main) - } - if (!existsSync(entry)) { - throw new ConfigError( - `piExecutor: cannot load pi extension "${spec}": resolved entry ${entry} does not exist`, - ) - } - return entry -} - -/** Does this `load` entry name the MCP adapter, before resolution? */ -function entryNamesAdapter(spec: string): boolean { - if (!isAbsolute(spec)) return spec === PI_MCP_ADAPTER - const normalized = spec.split(/[\\/]/) - if (normalized.includes(PI_MCP_ADAPTER)) return true - // An absolute path outside a `pi-mcp-adapter/` directory still counts when the package it belongs - // to declares that name — a vendored checkout under any directory name. - const parent = spec.slice(0, spec.lastIndexOf(sep) + 1) - return parent.length > 0 && readPackageName(parent) === PI_MCP_ADAPTER -} - -/** Outcome of lowering `extensions.pi.load` into argv. */ -export interface PiExtensionArgs { - args: string[] - /** Resolved absolute entry files, in the order pi receives them. */ - entries: string[] - adapterInjected: boolean -} - -/** - * Lower `AgentProfile.extensions.pi.load` into pi argv, honoring the `--no-extensions` trap. - * - * `load` absent (not an array) → no flags at all, so pi keeps its default discovery and the adapter - * auto-loads from `settings.json` `packages`. `load` present → `--no-extensions` plus one - * `--extension ` per listed extension, which is what makes an ablation arm - * reproducible and leak-free. - * - * When `mcpRequested` is true and an explicit `load` array does not already name the adapter, the - * adapter is APPENDED — otherwise `--no-extensions` would suppress the very extension that turns - * the `--mcp-config` file into tools. The caller's profile is not touched; the addition is reported - * through `adapterInjected` so it reaches the executor receipt and the live progress channel. - * - * A non-string or blank entry is a schema error here, not a silent skip: an ablation arm that - * quietly drops a typo'd extension is the same class of structural false-null this module exists to - * eliminate. - */ -export function derivePiExtensionArgs( - profile: { extensions?: Record | undefined> } | undefined, - mcpRequested: boolean, -): PiExtensionArgs { - const load = profile?.extensions?.pi?.load - if (!Array.isArray(load)) return { args: [], entries: [], adapterInjected: false } - - const declared: string[] = [] - for (const [index, entry] of load.entries()) { - if (typeof entry !== 'string' || !entry.trim()) { - throw new ValidationError( - `piExecutor: extensions.pi.load[${index}] must be a non-empty string ` + - `(got ${entry === null ? 'null' : typeof entry})`, - ) - } - declared.push(entry.trim()) - } - - const adapterInjected = mcpRequested && !declared.some(entryNamesAdapter) - const resolvedSpecs = adapterInjected ? [...declared, PI_MCP_ADAPTER] : declared - - const npmRoot = piNpmRoot() - const entries = resolvedSpecs.map((spec) => resolvePiExtensionEntry(spec, npmRoot)) - const args = ['--no-extensions'] - for (const entry of entries) args.push('--extension', entry) - return { args, entries, adapterInjected } -} - -/** Where one worker execution's private config directory is created, and what it is called. */ -export interface PiMcpMountOptions { - /** - * The worker's own working directory (`PiSeam.cwd`). Omit when the seam names none: the config - * then lands in the OS temp directory. It is NEVER written into `process.cwd()` — the operator's - * own working directory is not a scratch space for a worker's config. - */ - cwd?: string - /** - * The executor's per-execution run id, folded into the directory name so a directory that somehow - * survives names the worker that left it. Uniqueness is NOT taken from this — `mkdtemp` provides - * it — because two workers built from one factory in the same millisecond share a run id stem. - */ - runId: string -} - -/** Directory-name-safe form of a run id; the `mkdtemp` suffix supplies the uniqueness. */ -function runIdSlug(runId: string): string { - const slug = runId.replace(/[^A-Za-z0-9._-]+/g, '-').replace(/^-+|-+$/g, '') - return slug.length > 0 ? slug.slice(0, 60) : 'worker' -} - -/** - * Write this worker execution's MCP config to a directory only it owns, and hand back the removal. - * - * Every property of the old cwd-discovered mount — a cross-process lock, a merge under whatever the - * workspace already had, a verbatim byte restore, a symlink guard on `.pi` — existed because two - * workers sharing a `PiSeam` shared one file path. `mkdtemp` deletes the shared path, and with it - * the entire class of problem: the directory did not exist a moment ago, nothing else can name it, - * `cleanup()` removes it whole, and two siblings started in the same millisecond get two configs. - * - * Returns `null` when there are no usable servers, so the no-MCP path writes nothing at all. - */ -export function mountPiMcpConfig( - mcpServers: Record>, - options: PiMcpMountOptions, -): PiMcpMount | null { - const serverNames = Object.keys(mcpServers) - if (serverNames.length === 0) return null - - const slug = runIdSlug(options.runId) - // ALWAYS the OS temp directory, never the worker's workspace. The path is handed to pi by - // `--mcp-config`, so it does not need to sit anywhere discoverable — and a scratch directory - // inside the workspace is a file the worker did not create, in a tree it may commit, diff, or - // clean. A dot prefix hides it from `ls`, not from `git status`. - const base = tmpdir() - const prefix = `agent-runtime-pi-mcp-${slug}-` - - let dir: string - try { - dir = mkdtempSync(join(base, prefix)) - } catch (err) { - throw new ConfigError( - `piExecutor: cannot create a private MCP config directory under ${base}: ` + - `${err instanceof Error ? err.message : String(err)}`, - ) - } - - const configPath = join(dir, 'mcp.json') - try { - // `wx` because a fresh mkdtemp directory cannot already hold this file — if it does, something - // is wrong enough that overwriting would hide it. - writeFileSync(configPath, JSON.stringify({ mcpServers }, null, 2), { flag: 'wx', mode: 0o600 }) - } catch (err) { - rmSync(dir, { recursive: true, force: true }) - throw new ConfigError( - `piExecutor: failed to write the MCP config at ${configPath}: ` + - `${err instanceof Error ? err.message : String(err)}`, - ) - } - - let cleaned = false - return { - configPath, - serverNames, - cleanup: (): void => { - if (cleaned) return - cleaned = true - rmSync(dir, { recursive: true, force: true }) - }, - } -} - -/** - * The one call `piExecutor` makes before spawning pi: decide what pi must be told, fail loud if it - * cannot work, then mount. - * - * Order matters and is the point: - * - * 1. Canonicalize `profile.mcp` first, so the adapter gate keys off servers that would ACTUALLY be - * written — a block of `enabled:false` entries must not trip it. - * 2. If any server survives and `pi-mcp-adapter` is not installed, throw. No file has been written - * and pi has not been spawned. - * 3. Lower `extensions.pi.load` (which may itself throw on an unresolvable extension) — still before - * any file exists. - * 4. Only then create this execution's private config directory and emit `--mcp-config`. - * - * A profile with no usable MCP server is a pure no-op: no adapter check, no directory, no flags, no - * throw. `extensions.pi.load` on its own still produces flags, because reproducible ablation arms - * need them whether or not MCP is involved. - * - * The caller MUST call `mount.cleanup()` on every exit path — settle, throw, abort, timeout, and - * spawn failure — so the directory does not outlive the worker that owns it. - */ -export function preparePiMcp( - profile: AgentProfile | undefined, - options: PiMcpMountOptions, -): PiMcpPreparation { - const mcpServers = buildPiMcpServers(profile?.mcp) - const requested = Object.keys(mcpServers) - - if (requested.length > 0 && !piMcpAdapterAvailable()) { - throw new ConfigError( - `piExecutor: cannot mount MCP servers for pi (${requested.join(', ')}): the ` + - `${PI_MCP_ADAPTER} extension is not installed. pi ships no MCP support of its own, so ` + - 'these servers would never connect and the worker would run tool-less. Looked in ' + - `${join(piNpmRoot(), 'node_modules', PI_MCP_ADAPTER)} and in the \`packages\` list of ` + - `${join(piAgentDir(), 'settings.json')}. Install it with ` + - `\`pi install npm:${PI_MCP_ADAPTER}\`, or set ${PI_MCP_ADAPTER_ENV}=1 if it is vendored ` + - 'somewhere this check cannot see.', - ) - } - - const extensions = derivePiExtensionArgs(profile, requested.length > 0) - if (requested.length === 0) return { args: extensions.args, mount: null, receipt: undefined } - - const mount = mountPiMcpConfig(mcpServers, options) - if (!mount) return { args: extensions.args, mount: null, receipt: undefined } - return { - args: [...extensions.args, PI_MCP_CONFIG_FLAG, mount.configPath], - mount, - receipt: { - servers: mount.serverNames, - configPath: mount.configPath, - extensions: extensions.entries, - adapterInjected: extensions.adapterInjected, - }, - } -} diff --git a/src/runtime/supervise/runtime.ts b/src/runtime/supervise/runtime.ts index 0af6527e..f4a1fee0 100644 --- a/src/runtime/supervise/runtime.ts +++ b/src/runtime/supervise/runtime.ts @@ -72,7 +72,6 @@ import type { import { zeroTokenUsage } from '../util' import { createInbox, type Inbox } from './inbox' import { attestRuntimeOwnedExecutor, newExecutionAttemptId } from './materialization' -import { PI_RUNTIME, type PiSeam, piExecutor } from './pi-executor' import { type ActivityLog, createActivityLog, @@ -2092,8 +2091,8 @@ interface BridgeStreamChunk { * `statusCaptured: false` and carries no `startedAt`/`endedAt` — its span is an instant with no * status. Synthesising an end time would inject a fabricated 0ms latency, and defaulting to 'ok' * would count an unobserved call as a success; both would silently corrupt any downstream latency - * or error-rate analysis. An honest lower-fidelity span beats a fabricated one. `piExecutor` has - * true durations because pi's RPC stream carries `tool_execution_start`/`_end`; this wire does not. + * or error-rate analysis. An honest lower-fidelity span beats a fabricated one. A harness whose + * native protocol reports tool completion could carry true durations; this wire does not. * * cli-bridge emits each call complete in ONE delta (`{id, name, arguments}` together), so no * cross-delta argument-fragment assembly is needed; a frame that carries argument bytes without a @@ -2545,7 +2544,6 @@ export type ExecutorConfig = | ({ backend: 'cli' } & CliSeam) | ({ backend: 'cli-worktree' } & CliWorktreeSeam) | ({ backend: 'provider' } & ProviderSeam) - | ({ backend: 'pi' } & PiSeam) | ({ backend: 'sandbox'; harness?: BackendType } & SandboxSeam) /** Capture one public executor configuration at its call boundary. All data that selects policy, @@ -2611,7 +2609,6 @@ export function snapshotExecutorConfig(config: ExecutorConfig): ExecutorConfig { case 'router': case 'bridge': case 'cli': - case 'pi': return detachedSnapshot(config, `createExecutor ${config.backend} config`) } } @@ -2675,7 +2672,6 @@ export function bindReusableExecutorExecutionId( case 'router-tools': case 'cli': case 'provider': - case 'pi': case 'sandbox': return captured } @@ -2706,8 +2702,6 @@ export function createExecutor(config: ExecutorConfig): ExecutorFactory return cliExecutor(spec, seamed) case 'cli-worktree': return cliWorktreeExecutor(spec, seamed) - case 'pi': - return piExecutor(spec, seamed) case 'provider': { const providerSeam = readSeam(seamed, providerSeamKey, 'provider') const provider = resolveAgentEnvironmentProvider( @@ -2801,10 +2795,6 @@ export function createExecutorRegistry(): ExecutorRegistry { factories.set('inline', routerInlineExecutor) factories.set('sandbox', sandboxExecutor) factories.set('cli', cliExecutor) - // pi is wrapped, not forked: `piExecutor` speaks pi's own out-of-process RPC protocol, so its - // steering queue / session persistence / abort stay upstream's. Registered here through the - // documented extension point so a spec can select it by `AgentSpec.executor` or by name. - factories.set(PI_RUNTIME, piExecutor) return { register(runtime: Runtime, factory: ExecutorFactory): void { diff --git a/src/runtime/supervise/supervise.ts b/src/runtime/supervise/supervise.ts index 3d239fca..255144ab 100644 --- a/src/runtime/supervise/supervise.ts +++ b/src/runtime/supervise/supervise.ts @@ -184,7 +184,6 @@ function backendProfileMaterialization(backend: ExecutorConfig): ProfileMaterial return backend.bridge ? fullProfileMaterialization : worktreeCliProfileMaterialization case 'router': case 'router-tools': - case 'pi': return promptModelProfileMaterialization case 'cli': return controlProfileMaterialization diff --git a/src/runtime/supervise/worker-trace.ts b/src/runtime/supervise/worker-trace.ts index da23bdfe..dc2d1311 100644 --- a/src/runtime/supervise/worker-trace.ts +++ b/src/runtime/supervise/worker-trace.ts @@ -37,7 +37,7 @@ * PRECEDENCE, from lowest to highest: * 1. the supervisor process's own `process.env` (ambient inheritance), * 2. the trace context stamped here, - * 3. the caller's own seam env (`PiSeam.env`, `CliSeam.env`). + * 3. the caller's own seam env (`CliSeam.env`). * A caller who sets `TRACE_ID` / `PARENT_SPAN_ID` on a seam wins — theirs is a deliberate * declaration about the worker. Ambient `process.env` does NOT win: when the supervisor process was * itself launched as someone's worker, its inherited ids describe the SUPERVISOR's place in an @@ -48,7 +48,6 @@ * * WHICH BACKENDS PROPAGATE. Only a backend with a real environment channel to the worker can carry * this, and the ones that cannot say so here rather than dropping it silently: - * - `pi` YES — `PiSeam.env` → the `pi --mode rpc` subprocess. * - `cli` YES — `CliSeam.env` → the spawned subprocess. * - `sandbox` YES — `CreateSandboxOptions.env` on the box the worker runs in (single-shot * and steerable). This is the cross-MACHINE case the feature exists for. @@ -66,14 +65,13 @@ import type { ExecutorConfig } from './runtime' /** * The census above, as a value the compiler checks. `satisfies` against every `ExecutorConfig` - * discriminant means a NINTH backend arm cannot be added without classifying it here — the prose + * discriminant means an EIGHTH backend arm cannot be added without classifying it here — the prose * alone could go stale silently, which is the failure this table exists to prevent. * * `true` = the arm has a real environment channel to the worker and stamps the context. * `false` = it has none today; it is honestly unpropagated rather than silently dropping. */ export const WORKER_TRACE_PROPAGATION = { - pi: true, cli: true, sandbox: true, router: false, diff --git a/src/testing/fixtures/agent-improvement-proposal.json b/src/testing/fixtures/agent-improvement-proposal.json index d65eda83..8cb0ecc2 100644 --- a/src/testing/fixtures/agent-improvement-proposal.json +++ b/src/testing/fixtures/agent-improvement-proposal.json @@ -1,6 +1,6 @@ { "changedSurfaces": ["prompt"], - "digest": "sha256:a8e04889e298e6cd1dfe3d527722f43a14fd1e979023b6f020dee16b5f04842b", + "digest": "sha256:2db253a4fe022158540e2cad176191b3bb86ceeb192fb0f0b8574283bf4e2148", "evaluation": { "decision": { "contributingChecks": [ @@ -4810,7 +4810,7 @@ ], "metadata": { "fixture": "agent-improvement-proposal", - "runtimeVersion": "0.117.0" + "runtimeVersion": "0.118.0" }, "objectives": [ { @@ -4921,8 +4921,8 @@ "baselineContentHash": "sha256:5c21ee53e513fc604cb09754e21c392b24a424da0ef37dbf8f1ee4a8a0b08f09", "candidateContentHash": "sha256:60fcbb1c728194bd51d7d19cb732d1c3f1881dce7e0a6266b41c8b98cfd65693", "kind": "agent-eval-loop", - "recordDigest": "sha256:80ce3a9d5959a8c0183a1457bdf2eec95c9e4686b3548671ca021074ab29f43d", - "runId": "agent-runtime-0.117.0-proposal-fixture", + "recordDigest": "sha256:e88997c579a8a41ec72deae1db49c28758ef729479aa89a9f6816e7b0dec1164", + "runId": "agent-runtime-0.118.0-proposal-fixture", "schema": "agent-candidate-experiment" } }, @@ -4949,5 +4949,5 @@ ], "kind": "agent-improvement-proposal", "proposedAt": "2026-07-10T01:00:00.000Z", - "runId": "agent-runtime-0.117.0-proposal-fixture" + "runId": "agent-runtime-0.118.0-proposal-fixture" } diff --git a/src/testing/fixtures/agent-profile-improvement-proposal.json b/src/testing/fixtures/agent-profile-improvement-proposal.json index 8f465414..73b1360d 100644 --- a/src/testing/fixtures/agent-profile-improvement-proposal.json +++ b/src/testing/fixtures/agent-profile-improvement-proposal.json @@ -1,6 +1,6 @@ { "changedSurfaces": ["prompt", "skills"], - "digest": "sha256:b0d63ad56178709cfe403755993becbc8bf21635c5d95fbd884eda98a223d6f7", + "digest": "sha256:9721b548dc4d74c11d90f5ad421d34dd41f8a65aae4dacb881897f6961cdecbe", "evaluation": { "decision": { "contributingChecks": [ @@ -1715,7 +1715,7 @@ ], "metadata": { "fixture": "agent-profile-improvement-proposal", - "runtimeVersion": "0.117.0" + "runtimeVersion": "0.118.0" }, "objectives": [ { @@ -1826,7 +1826,7 @@ "baselineContentHash": "sha256:21c495a37c418c10bde64fbaa188beddeed31f1f051ea60a6a6582a9ee0db704", "candidateContentHash": "sha256:103f77bc8481601eef1ad5fe6ba84a40dffabc3a44f421f8c8559121edab84e9", "kind": "agent-eval-loop", - "recordDigest": "sha256:73b72fb671a9c72744c0bb5ac9a1a31f7c6d02bac6e9204865318f8d55c75e67", + "recordDigest": "sha256:099deb662846f036d55099d91cb4e3294f56d0618ec10fecb618a1e37d9471ae", "runId": "profile-improvement-1", "schema": "agent-profile-improvement-experiment" } diff --git a/tests/runtime/executor-config-snapshot.test.ts b/tests/runtime/executor-config-snapshot.test.ts index 16b32266..cefd2e4d 100644 --- a/tests/runtime/executor-config-snapshot.test.ts +++ b/tests/runtime/executor-config-snapshot.test.ts @@ -103,9 +103,16 @@ describe('createExecutor config intake', () => { runtime: 'provider-runtime', }, { - name: 'pi', - config: { backend: 'pi', bin: 'pi', args: ['--test'], model: 'provider/model' }, - runtime: 'pi', + // pi is not a backend of its own: it is a bridge wire id like every other harness, and + // the intake capture must hold for it exactly as it does for the generic bridge case. + name: 'pi-over-bridge', + config: { + backend: 'bridge', + bridgeUrl: 'http://bridge.test', + bridgeBearer: 'secret', + model: 'pi/tangle-router/gpt-5-mini', + }, + runtime: 'cli', }, { name: 'sandbox', diff --git a/tests/runtime/pi-executor.test.ts b/tests/runtime/pi-executor.test.ts deleted file mode 100644 index a65dd7c7..00000000 --- a/tests/runtime/pi-executor.test.ts +++ /dev/null @@ -1,685 +0,0 @@ -/** - * `piExecutor` — pi wrapped behind `Executor`, driven against a FAKE pi that speaks pi 0.83's - * real RPC wire (`--mode rpc`, JSON lines on stdin/stdout, `AgentEvent` shapes). - * - * The point of the wrapper is that it delegates rather than reimplements: a message becomes pi's - * own `prompt` command with a streaming behavior, teardown becomes `abort`. These tests assert - * exactly that — the COMMANDS the wrapper sends — plus that pi's tool/turn events become - * the live progress feed and the shared tool-span currency. - */ - -import { mkdtemp, rm, writeFile } from 'node:fs/promises' -import { tmpdir } from 'node:os' -import { join } from 'node:path' -import type { ToolSpan } from '@tangle-network/agent-eval' -import type { AgentProfile } from '@tangle-network/agent-interface' -import { afterAll, beforeAll, describe, expect, it } from 'vitest' -import { InMemoryResultBlobStore, InMemorySpawnJournal } from '../../src/durable/spawn-journal' -import { createBudgetPool } from '../../src/runtime/supervise/budget' -import { gateOnDeliverable } from '../../src/runtime/supervise/completion-gate' -import { piExecutor, piSeamKey } from '../../src/runtime/supervise/pi-executor' -import { createExecutorRegistry } from '../../src/runtime/supervise/runtime' -import { createScope } from '../../src/runtime/supervise/scope' -import type { - Agent, - AgentSpec, - ExecutorContext, - UsageEvent, -} from '../../src/runtime/supervise/types' - -let dir: string -let fakePi: string -let commandLog: string - -/** - * A stand-in for `pi --mode rpc` using Pi 0.83's actual session event order. In particular: - * user and tool-result messages both emit `message_end`; each model call emits `turn_end`; - * `agent_end` may be followed by an automatic retry; only `agent_settled` means the session is - * idle. The scenarios below pin those distinctions instead of approximating the wire. - */ -const FAKE_PI = `#!/usr/bin/env node -const fs = require('node:fs') -const log = process.env.PI_COMMAND_LOG -fs.appendFileSync(log, JSON.stringify({ type: 'argv', args: process.argv.slice(2) }) + '\\n') -const emit = (o) => process.stdout.write(JSON.stringify(o) + '\\n') -const result = (text) => ({ content: [{ type: 'text', text }], details: {} }) -const message = (m) => { - emit({ type: 'message_start', message: m }) - emit({ type: 'message_end', message: m }) -} -const pricedUsage = (input, output, cacheRead, cacheWrite, total) => ({ - input, - output, - cacheRead, - cacheWrite, - totalTokens: input + output + cacheRead + cacheWrite, - cost: { input: total / 2, output: total / 2, cacheRead: 0, cacheWrite: 0, total }, -}) -const freeUsage = (input, output, cacheRead, cacheWrite) => ({ - input, - output, - cacheRead, - cacheWrite, - totalTokens: input + output + cacheRead + cacheWrite, - cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, -}) -const assistant = (text, stopReason, usage, extra = []) => ({ - role: 'assistant', - content: [...(text ? [{ type: 'text', text }] : []), ...extra], - stopReason, - usage, -}) -const begin = (prompt) => { - const user = { role: 'user', content: [{ type: 'text', text: String(prompt) }] } - emit({ type: 'agent_start' }) - emit({ type: 'turn_start' }) - message(user) - return user -} -const finish = (messages, willRetry = false) => { - emit({ type: 'agent_end', messages, willRetry }) - if (!willRetry) emit({ type: 'agent_settled' }) -} -const finalTurn = (text, usage = pricedUsage(10, 4, 2, 1, 0.001)) => { - const final = assistant(text, 'stop', usage) - emit({ type: 'turn_start' }) - message(final) - emit({ type: 'turn_end', message: final, toolResults: [] }) - return final -} -let buf = '' -let turn = 0 -let pendingAbort -let abortScheduled = false -let exitOnAbort = false -process.stdin.on('data', (c) => { - buf += c.toString('utf8') - for (;;) { - const nl = buf.indexOf('\\n') - if (nl < 0) break - const line = buf.slice(0, nl).trim() - buf = buf.slice(nl + 1) - if (!line) continue - let cmd - try { cmd = JSON.parse(line) } catch { continue } - fs.appendFileSync(log, JSON.stringify(cmd) + '\\n') - - if (cmd.type === 'abort') { - emit({ id: cmd.id, type: 'response', command: 'abort', success: true }) - if (exitOnAbort) process.exit(0) - if (!pendingAbort || abortScheduled) continue - abortScheduled = true - const aborted = pendingAbort - pendingAbort = undefined - setTimeout(() => { - message(aborted) - emit({ type: 'turn_end', message: aborted, toolResults: [] }) - finish([aborted]) - }, 100) - continue - } - if (cmd.type !== 'prompt') continue - - const prompt = String(cmd.message) - if (prompt.includes('reject at preflight')) { - emit({ - id: cmd.id, - type: 'response', - command: 'prompt', - success: false, - error: 'No model selected', - }) - continue - } - emit({ id: cmd.id, type: 'response', command: 'prompt', success: true }) - const target = prompt.includes('right.ts') ? 'right.ts' : 'wrong.ts' - const myTurn = turn++ - const user = begin(prompt) - - if (prompt.includes('delayed abort receipt')) { - pendingAbort = assistant( - 'cancelled after billed work', - 'aborted', - freeUsage(40, 10, 60, 5), - ) - continue - } - - if (prompt.includes('exit on abort')) { - exitOnAbort = true - continue - } - - if (prompt.includes('retryable provider error')) { - const failed = { - ...assistant('', 'error', freeUsage(0, 0, 0, 0)), - errorMessage: 'Stream ended without finish_reason', - } - message(failed) - emit({ type: 'turn_end', message: failed, toolResults: [] }) - emit({ type: 'agent_end', messages: [user, failed], willRetry: true }) - emit({ type: 'auto_retry_start', attempt: 1, maxAttempts: 3, delayMs: 25 }) - setTimeout(() => { - emit({ type: 'agent_start' }) - emit({ type: 'turn_start' }) - const recovered = assistant('recovered', 'stop', pricedUsage(10, 4, 2, 1, 0.001)) - message(recovered) - emit({ type: 'turn_end', message: recovered, toolResults: [] }) - finish([recovered]) - }, 25) - continue - } - - if (prompt.includes('terminal provider error')) { - const failed = { - ...assistant('', 'error', freeUsage(0, 0, 0, 0)), - errorMessage: 'Stream ended without finish_reason', - } - message(failed) - emit({ type: 'turn_end', message: failed, toolResults: [] }) - finish([user, failed]) - continue - } - - if (prompt.includes('subscription usage')) { - const final = assistant('subscription result', 'stop', freeUsage(30, 12, 5, 3)) - message(final) - emit({ type: 'turn_end', message: final, toolResults: [] }) - finish([user, final]) - continue - } - - const parallel = prompt.includes('parallel tools') - const emptyFinal = prompt.includes('empty final assistant') - const calls = parallel - ? [ - { type: 'toolCall', id: 'read-' + myTurn, name: 'read', arguments: { path: 'alpha.ts' } }, - { type: 'toolCall', id: 'bash-' + myTurn, name: 'bash', arguments: { command: 'pnpm test' } }, - ] - : [{ type: 'toolCall', id: 'edit-' + myTurn, name: 'edit', arguments: { path: target } }] - const toolTurn = assistant('working', 'toolUse', pricedUsage(30, 12, 5, 3, 0.002), calls) - message(toolTurn) - - const toolResults = [] - if (parallel) { - emit({ type: 'tool_execution_start', toolCallId: 'read-' + myTurn, toolName: 'read', args: { path: 'alpha.ts' } }) - emit({ type: 'tool_execution_start', toolCallId: 'bash-' + myTurn, toolName: 'bash', args: { command: 'pnpm test' } }) - emit({ type: 'tool_execution_end', toolCallId: 'bash-' + myTurn, toolName: 'bash', result: result('tests passed'), isError: false }) - emit({ type: 'tool_execution_end', toolCallId: 'read-' + myTurn, toolName: 'read', result: result('permission denied'), isError: true }) - toolResults.push( - { role: 'toolResult', toolCallId: 'read-' + myTurn, toolName: 'read', content: result('permission denied').content, isError: true }, - { role: 'toolResult', toolCallId: 'bash-' + myTurn, toolName: 'bash', content: result('tests passed').content, isError: false }, - ) - } else { - emit({ type: 'tool_execution_start', toolCallId: 'edit-' + myTurn, toolName: 'edit', args: { path: target } }) - const toolText = emptyFinal ? 'TOOL_RESULT_MUST_NOT_BECOME_OUTPUT' : 'updated ' + target - emit({ type: 'tool_execution_end', toolCallId: 'edit-' + myTurn, toolName: 'edit', result: result(toolText), isError: false }) - toolResults.push({ role: 'toolResult', toolCallId: 'edit-' + myTurn, toolName: 'edit', content: result(toolText).content, isError: false }) - } - for (const toolResult of toolResults) message(toolResult) - emit({ type: 'turn_end', message: toolTurn, toolResults }) - const final = finalTurn(emptyFinal ? '' : 'edited ' + target) - finish([user, toolTurn, ...toolResults, final]) - } -}) -` - -beforeAll(async () => { - dir = await mkdtemp(join(tmpdir(), 'pi-exec-')) - fakePi = join(dir, 'fake-pi.cjs') - commandLog = join(dir, 'commands.jsonl') - await writeFile(fakePi, FAKE_PI, { mode: 0o755 }) - await writeFile(commandLog, '') -}) - -afterAll(async () => { - await rm(dir, { recursive: true, force: true }) -}) - -/** The fake is an executable with a `#!/usr/bin/env node` shebang, so it is launched exactly the - * way a real `pi` binary is: ` --mode rpc`, with the wrapper owning the flags. */ -function piCtx(): ExecutorContext { - return { - signal: new AbortController().signal, - seams: { [piSeamKey]: { bin: fakePi, env: { PI_COMMAND_LOG: commandLog } } }, - } -} - -const spec: AgentSpec = { - profile: { name: 'coder' } as AgentProfile, - harness: null, -} - -async function drain(iter: AsyncIterable): Promise { - const out: UsageEvent[] = [] - for await (const e of iter) out.push(e) - return out -} - -async function captureFailure(iter: AsyncIterable): Promise<{ - events: UsageEvent[] - error: Error | undefined -}> { - const events: UsageEvent[] = [] - try { - for await (const event of iter) events.push(event) - return { events, error: undefined } - } catch (error) { - return { events, error: error instanceof Error ? error : new Error(String(error)) } - } -} - -async function readCommands(): Promise>> { - const { readFile } = await import('node:fs/promises') - const text = await readFile(commandLog, 'utf8') - return text - .split('\n') - .filter((l) => l.length > 0) - .map((l) => JSON.parse(l) as Record) -} - -async function waitForCommand( - predicate: (command: Record) => boolean, -): Promise { - const deadline = Date.now() + 1_000 - while (Date.now() < deadline) { - if ((await readCommands()).some(predicate)) return - await new Promise((resolve) => setTimeout(resolve, 10)) - } - throw new Error('timed out waiting for fake Pi command') -} - -describe('piExecutor — pi wrapped, not forked', () => { - it('uses AgentProfile.model.default instead of the backend fallback', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const withFallback: ExecutorContext = { - ...ctx, - seams: { - ...ctx.seams, - [piSeamKey]: { - ...(ctx.seams[piSeamKey] as Record), - model: 'fallback/ignored-model', - }, - }, - } - const ex = piExecutor( - { - profile: { - name: 'profile-model', - model: { default: 'profile/selected-model' }, - }, - harness: null, - }, - withFallback, - ) - - await drain(ex.execute('make the change', withFallback.signal) as AsyncIterable) - const argv = (await readCommands()).find((command) => command.type === 'argv')?.args - - expect(argv).toEqual(['--mode', 'rpc', '--provider', 'profile', '--model', 'selected-model']) - await ex.teardown('brutalKill') - }) - - it('counts duplicated message_end + turn_end telemetry once from authoritative turn_end', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const ex = piExecutor(spec, ctx) - expect(ex.runtime).toBe('pi') - - const events = await drain( - ex.execute('make the change', ctx.signal) as AsyncIterable, - ) - - // REAL usage only — the numbers the fake pi reported, not a fabricated estimate. - expect(events.filter((event) => event.kind === 'tokens')).toEqual([ - { kind: 'tokens', input: 38, output: 12 }, - { kind: 'tokens', input: 13, output: 4 }, - ]) - expect(events).toContainEqual({ kind: 'cost', usd: 0.002 }) - expect(events).toContainEqual({ kind: 'cost', usd: 0.001 }) - expect(events.filter((e) => e.kind === 'iteration')).toHaveLength(2) - - const progress = ex.progress?.() - expect(progress?.turns).toBe(2) - expect(progress?.recentActivity?.some((a) => a.label === 'edit')).toBe(true) - - const spans = (await ex.traceSource?.()?.collect()) as ToolSpan[] - expect(spans).toHaveLength(1) - expect(spans[0]).toMatchObject({ - toolName: 'edit', - args: { path: 'wrong.ts' }, - status: 'ok', - result: { content: [{ type: 'text', text: 'updated wrong.ts' }], details: {} }, - }) - - const artifact = ex.resultArtifact() - expect(String((artifact.out as { content: string }).content)).toContain('edited wrong.ts') - expect(artifact.spent.tokens).toEqual({ input: 51, output: 16 }) - expect(artifact.spent.usd).toBe(0.003) - expect(artifact.spent).not.toHaveProperty('tokensKnown') - expect(artifact.spent).not.toHaveProperty('usdKnown') - await ex.teardown('brutalKill') - }) - - it('joins parallel Pi 0.83 tool completions to their starts by toolCallId', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const ex = piExecutor(spec, ctx) - - await drain(ex.execute('run parallel tools', ctx.signal) as AsyncIterable) - - const spans = (await ex.traceSource?.()?.collect()) as ToolSpan[] - expect(spans).toHaveLength(2) - expect(spans[0]).toMatchObject({ - toolName: 'bash', - args: { command: 'pnpm test' }, - status: 'ok', - result: { content: [{ type: 'text', text: 'tests passed' }], details: {} }, - }) - expect(spans[1]).toMatchObject({ - toolName: 'read', - args: { path: 'alpha.ts' }, - status: 'error', - error: 'permission denied', - result: { content: [{ type: 'text', text: 'permission denied' }], details: {} }, - }) - await ex.teardown('brutalKill') - }) - - it('marks subscription-priced Pi usage as unknown instead of known zero', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const ex = piExecutor(spec, ctx) - - const events = await drain( - ex.execute('subscription usage', ctx.signal) as AsyncIterable, - ) - - expect(events.filter((event) => event.kind === 'cost')).toEqual([ - { kind: 'cost', usd: 0, usdKnown: false }, - ]) - expect(ex.resultArtifact().spent).toMatchObject({ usd: 0, usdKnown: false }) - await ex.teardown('brutalKill') - }) - - it('waits for agent_settled so Pi can recover after agent_end requests a retry', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const ex = piExecutor(spec, ctx) - - const events = await drain( - ex.execute('retryable provider error', ctx.signal) as AsyncIterable, - ) - - expect(events.filter((event) => event.kind === 'iteration')).toHaveLength(2) - expect(events.filter((event) => event.kind === 'tokens')).toEqual([ - { kind: 'tokens', input: 13, output: 4 }, - ]) - expect(ex.resultArtifact()).toMatchObject({ - out: { content: 'recovered', turns: 2 }, - spent: { usdKnown: false }, - }) - await ex.teardown('brutalKill') - }) - - it('rejects a terminal Pi error instead of saving the user prompt as a result', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const ex = piExecutor(spec, ctx) - - const captured = await captureFailure( - ex.execute('terminal provider error', ctx.signal) as AsyncIterable, - ) - - expect(captured.error).toBeInstanceOf(Error) - expect(captured.error?.message).toContain('Stream ended without finish_reason') - expect(captured.events).toContainEqual({ kind: 'cost', usd: 0, usdKnown: false }) - expect(() => ex.resultArtifact()).toThrow(/before stream drained/) - await ex.teardown('brutalKill') - }) - - it('settles a terminal Pi error as down without checking or persisting a deliverable', async () => { - await writeFile(commandLog, '') - const root = 'pi-terminal-error' - const journal = new InMemorySpawnJournal() - await journal.beginTree(root, new Date(0).toISOString()) - const innerBlobs = new InMemoryResultBlobStore() - let blobWrites = 0 - const blobs = { - async put(outRef: string, artifact: unknown) { - blobWrites += 1 - await innerBlobs.put(outRef, artifact) - }, - get: (outRef: string) => innerBlobs.get(outRef), - } - let deliverableChecks = 0 - const executor = gateOnDeliverable(piExecutor(spec, piCtx()), { - check: () => { - deliverableChecks += 1 - return true - }, - }) - const agentSpec: AgentSpec = { ...spec, executor } - const agent = { - name: 'pi-error-worker', - act: async () => undefined, - executorSpec: agentSpec, - } as Agent & { executorSpec: AgentSpec } - const scope = createScope({ - parentId: root, - root, - pool: createBudgetPool({ maxIterations: 4, maxTokens: 100 }), - journal, - blobs, - executors: createExecutorRegistry(), - seams: {}, - depth: 0, - signal: new AbortController().signal, - now: () => 0, - }) - - const spawned = scope.spawn(agent, 'terminal provider error', { - budget: { maxIterations: 4, maxTokens: 100 }, - label: 'provider-error', - }) - expect(spawned.ok).toBe(true) - if (!spawned.ok) return - - const settled = await scope.next() - expect(settled).toMatchObject({ - kind: 'down', - reason: expect.stringContaining('Stream ended without finish_reason'), - }) - expect(deliverableChecks).toBe(0) - expect(blobWrites).toBe(0) - expect(scope.progress(spawned.handle.id)).toMatchObject({ - live: false, - status: 'failed', - turns: 1, - usd: 0, - usdKnown: false, - }) - const terminal = (await journal.loadTree(root))?.find((event) => event.kind === 'settled') - expect(terminal).toMatchObject({ - kind: 'settled', - status: 'down', - spent: { - iterations: 1, - tokens: { input: 0, output: 0 }, - usd: 0, - usdKnown: false, - }, - }) - expect(terminal).not.toHaveProperty('outRef') - }) - - it('uses only the terminal successful assistant turn as output, never a tool result', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const ex = piExecutor(spec, ctx) - - await drain(ex.execute('empty final assistant', ctx.signal) as AsyncIterable) - - expect(ex.resultArtifact().out).toEqual({ content: '', turns: 2 }) - await ex.teardown('brutalKill') - }) - - it('refuses a pre-aborted execution before spawning Pi or sending a prompt', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const ex = piExecutor(spec, ctx) - const controller = new AbortController() - controller.abort() - - const captured = await captureFailure( - ex.execute('must not run', controller.signal) as AsyncIterable, - ) - - expect(captured.error?.name).toBe('AbortError') - expect(captured.events).toEqual([]) - expect(await readCommands()).toEqual([]) - }) - - it('drains Pi’s delayed terminal receipt before surfacing an active abort', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const ex = piExecutor(spec, ctx) - const controller = new AbortController() - const pending = captureFailure( - ex.execute( - 'wait for a delayed abort receipt', - controller.signal, - ) as AsyncIterable, - ) - await waitForCommand( - (command) => - command.type === 'prompt' && String(command.message).includes('delayed abort receipt'), - ) - - controller.abort() - const captured = await pending - - expect(captured.error?.name).toBe('AbortError') - expect(captured.events).toEqual([ - { kind: 'tokens', input: 105, output: 10 }, - { kind: 'cost', usd: 0, usdKnown: false }, - { kind: 'iteration' }, - ]) - expect((await readCommands()).map((command) => command.type)).toContain('abort') - expect(() => ex.resultArtifact()).toThrow(/before stream drained/) - }) - - it('keeps an active cancellation typed as AbortError when Pi exits without a receipt', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const ex = piExecutor(spec, ctx) - const controller = new AbortController() - const pending = captureFailure( - ex.execute('exit on abort', controller.signal) as AsyncIterable, - ) - await waitForCommand( - (command) => command.type === 'prompt' && String(command.message).includes('exit on abort'), - ) - - controller.abort() - const captured = await pending - - expect(captured.error?.name).toBe('AbortError') - expect(captured.events).toEqual([]) - expect(() => ex.resultArtifact()).toThrow(/before stream drained/) - }) - - it('fails explicitly instead of hanging when Pi rejects a prompt at preflight', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const ex = piExecutor(spec, ctx) - - const captured = await captureFailure( - ex.execute('reject at preflight', ctx.signal) as AsyncIterable, - ) - - expect(captured.events).toEqual([]) - expect(captured.error).toMatchObject({ - name: 'ValidationError', - message: 'piExecutor: Pi rejected prompt: No model selected', - }) - expect(() => ex.resultArtifact()).toThrow(/before stream drained/) - }) - - it('uses pi’s state-safe prompt behavior for forceful and queued messages', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const ex = piExecutor(spec, ctx) - // Deliver BEFORE draining so both messages are pending when the loop first forwards them — - // the wrapper must not hold its own queue, it must hand them straight to pi. - ex.deliver?.({ steer: 'switch to right.ts', interrupt: true }) - ex.deliver?.({ steer: 'and add a test', interrupt: false }) - - await drain(ex.execute('make the change', ctx.signal) as AsyncIterable) - const commands = await readCommands() - - const steer = commands.find( - (command) => command.type === 'prompt' && command.streamingBehavior === 'steer', - ) - const followUp = commands.find( - (command) => command.type === 'prompt' && command.streamingBehavior === 'followUp', - ) - expect(String(steer?.message)).toContain('switch to right.ts') - expect(String(followUp?.message)).toContain('and add a test') - await ex.teardown('brutalKill') - }) - - it('re-prompts pi when a steer lands while it is idle, so it cannot settle unread', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const ex = piExecutor(spec, ctx) - const stream = ex.execute('make the change', ctx.signal) as AsyncIterable - - const seen: UsageEvent[] = [] - let steered = false - for await (const e of stream) { - seen.push(e) - // The first completed turn is the boundary: deliver there, and pi must run again. - if (!steered && e.kind === 'iteration') { - steered = true - ex.deliver?.({ steer: 'the change belongs in right.ts', interrupt: false }) - expect(ex.progress?.()?.pendingMessages).toBe(1) - } - } - // Each prompt contains a tool turn and a terminal assistant turn. - expect(seen.filter((e) => e.kind === 'iteration').length).toBe(4) - expect(seen.filter((e) => e.kind === 'tokens')).toEqual([ - { kind: 'tokens', input: 38, output: 12 }, - { kind: 'tokens', input: 13, output: 4 }, - { kind: 'tokens', input: 38, output: 12 }, - { kind: 'tokens', input: 13, output: 4 }, - ]) - const artifact = ex.resultArtifact() - expect(String((artifact.out as { content: string }).content)).toContain('edited right.ts') - expect(artifact.spent.tokens).toEqual({ input: 102, output: 32 }) - expect(artifact.spent.usd).toBe(0.006) - await ex.teardown('brutalKill') - }) - - it('teardown asks pi to `abort` before killing the process', async () => { - await writeFile(commandLog, '') - const ctx = piCtx() - const ex = piExecutor(spec, ctx) - const stream = ex.execute('make the change', ctx.signal) as AsyncIterable - // Start the process, then tear it down mid-stream. - const iterator = stream[Symbol.asyncIterator]() - await iterator.next() - const res = await ex.teardown(500) - expect(res.destroyed).toBe(true) - const commands = await readCommands() - expect(commands.map((c) => c.type)).toContain('abort') - }) - - it('accepts a seam-less context (all fields optional) without throwing at construction', () => { - expect(() => - piExecutor(spec, { signal: new AbortController().signal, seams: {} }), - ).not.toThrow() - expect(() => piExecutor(spec, piCtx())).not.toThrow() - }) -}) diff --git a/tests/runtime/pi-mcp.test.ts b/tests/runtime/pi-mcp.test.ts deleted file mode 100644 index 0d1102f3..00000000 --- a/tests/runtime/pi-mcp.test.ts +++ /dev/null @@ -1,840 +0,0 @@ -/** - * MCP for the `pi` harness, at the layer that actually spawns pi. - * - * pi ships no MCP of its own — it comes from the `pi-mcp-adapter` extension, which reads the file - * named by its `--mcp-config` flag. Before this path existed, a profile carrying `mcp` reached pi - * with nothing mounted and the worker ran tool-less while burning its whole budget hunting for - * tools that were never there. These tests pin the properties that failure needs: - * - * 1. a declared server reaches the exact file pi was told to read, with the canonical body; - * 2. every worker gets its OWN file — two workers built from ONE seam run concurrently without - * colliding, which is the ordinary supervisor case; - * 3. `--no-extensions` (from `extensions.pi.load`) can never suppress the adapter that makes the - * declared servers work — and when this executor adds it, the addition is REPORTED on a - * channel a FAILED run still exposes; - * 4. a missing adapter throws before any file is written and before pi is spawned; - * 5. a profile with no MCP changes nothing at all; - * 6. the caller's profile object is never mutated. - * - * The fake pi below is deliberately minimal: it records the argv it was launched with and the exact - * bytes of the file `--mcp-config` named — resolved the same way the real adapter resolves it - * (`pi-mcp-adapter` `utils.ts:55`, `process.argv.indexOf('--mcp-config')`). That is the only - * evidence that proves the mount landed where a real pi-mcp-adapter would look. It writes its - * record under its OWN pid so two concurrent workers cannot overwrite each other's evidence. - */ - -import { existsSync, readdirSync } from 'node:fs' -import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' -import { tmpdir } from 'node:os' -import { join } from 'node:path' -import type { AgentProfile } from '@tangle-network/agent-interface' -import { afterAll, afterEach, beforeAll, describe, expect, it } from 'vitest' -import { ConfigError, ValidationError } from '../../src/errors' -import { piExecutor, piSeamKey } from '../../src/runtime/supervise/pi-executor' -import { - buildPiMcpServers, - derivePiExtensionArgs, - mountPiMcpConfig, - PI_MCP_ADAPTER_ENV, - PI_MCP_CONFIG_FLAG, - piMcpAdapterAvailable, - preparePiMcp, -} from '../../src/runtime/supervise/pi-mcp' -import type { AgentSpec, ExecutorContext, UsageEvent } from '../../src/runtime/supervise/types' - -const FAKE_PI = `#!/usr/bin/env node -const fs = require('node:fs') -const path = require('node:path') -const argv = process.argv.slice(2) -// The real adapter resolves its config off raw argv (pi-mcp-adapter utils.ts:55). Same here. -const flagIndex = argv.indexOf('--mcp-config') -let mcp = null -if (flagIndex >= 0 && flagIndex + 1 < argv.length) { - try { - mcp = fs.readFileSync(argv[flagIndex + 1], 'utf8') - } catch (err) { - mcp = 'UNREADABLE:' + err.code - } -} -if (process.env.PI_LOG_DIR) { - fs.writeFileSync( - path.join(process.env.PI_LOG_DIR, process.pid + '.json'), - JSON.stringify({ argv, mcp }), - ) -} -const holdMs = Number(process.env.PI_HOLD_MS || 0) -const emit = (o) => process.stdout.write(JSON.stringify(o) + '\\n') -const answer = (id) => { - emit({ id, type: 'response', command: 'prompt', success: true }) - emit({ type: 'agent_start' }) - emit({ type: 'turn_start' }) - const final = { - role: 'assistant', - content: [{ type: 'text', text: 'done' }], - stopReason: 'stop', - usage: { input: 1, output: 1, cacheRead: 0, cacheWrite: 0, cost: { total: 0.001 } }, - } - emit({ type: 'message_end', message: final }) - emit({ type: 'turn_end', message: final, toolResults: [] }) - emit({ type: 'agent_end', messages: [final], willRetry: false }) - emit({ type: 'agent_settled' }) -} -let buf = '' -process.stdin.on('data', (chunk) => { - buf += chunk.toString('utf8') - for (;;) { - const nl = buf.indexOf('\\n') - if (nl < 0) break - const line = buf.slice(0, nl).trim() - buf = buf.slice(nl + 1) - if (!line) continue - let cmd - try { cmd = JSON.parse(line) } catch { continue } - if (cmd.type !== 'prompt') continue - // A hold makes two concurrent workers genuinely overlap: neither can settle before the other - // has spawned, read its own config, and recorded it. - if (holdMs > 0) setTimeout(() => answer(cmd.id), holdMs) - else answer(cmd.id) - } -}) -` - -let root: string -let fakePi: string -/** A pi agent home with `pi-mcp-adapter` (and one unrelated extension) genuinely installed. */ -let agentDirWithAdapter: string -/** A pi agent home with nothing installed — the "adapter missing" machine. */ -let agentDirBare: string -let adapterEntry: string -let memoryEntry: string - -const originalAgentDir = process.env.PI_CODING_AGENT_DIR -const originalOverride = process.env[PI_MCP_ADAPTER_ENV] - -async function installFakeExtension(agentDir: string, name: string): Promise { - const packageDir = join(agentDir, 'npm', 'node_modules', name) - await mkdir(packageDir, { recursive: true }) - await writeFile(join(packageDir, 'package.json'), JSON.stringify({ name, main: 'index.ts' })) - await writeFile(join(packageDir, 'index.ts'), 'export default {}\n') - return join(packageDir, 'index.ts') -} - -beforeAll(async () => { - root = await mkdtemp(join(tmpdir(), 'pi-mcp-')) - fakePi = join(root, 'fake-pi.cjs') - await writeFile(fakePi, FAKE_PI, { mode: 0o755 }) - - agentDirWithAdapter = join(root, 'agent-with-adapter') - adapterEntry = await installFakeExtension(agentDirWithAdapter, 'pi-mcp-adapter') - memoryEntry = await installFakeExtension(agentDirWithAdapter, 'pi-memory') - - agentDirBare = join(root, 'agent-bare') - await mkdir(join(agentDirBare, 'npm', 'node_modules'), { recursive: true }) -}) - -afterAll(async () => { - await rm(root, { recursive: true, force: true }) -}) - -afterEach(() => { - restoreEnv('PI_CODING_AGENT_DIR', originalAgentDir) - restoreEnv(PI_MCP_ADAPTER_ENV, originalOverride) -}) - -function restoreEnv(name: string, value: string | undefined): void { - if (value === undefined) delete process.env[name] - else process.env[name] = value -} - -function withAdapter(): void { - process.env.PI_CODING_AGENT_DIR = agentDirWithAdapter - delete process.env[PI_MCP_ADAPTER_ENV] -} - -function withoutAdapter(): void { - process.env.PI_CODING_AGENT_DIR = agentDirBare - delete process.env[PI_MCP_ADAPTER_ENV] -} - -let caseCounter = 0 -async function freshCwd(): Promise { - caseCounter += 1 - const cwd = join(root, `cwd-${caseCounter}`) - await mkdir(cwd, { recursive: true }) - return cwd -} - -/** Config directories this module creates inside a worker cwd, so a leak is visible. */ -function mcpDirsIn(cwd: string): string[] { - return readdirSync(cwd).filter((name) => name.startsWith('.pi-mcp-')) -} - -interface PiRunRecord { - argv: string[] - /** Bytes of the file `--mcp-config` named, or null when no such flag was passed. */ - mcp: string | null -} - -/** Everything the fake pi processes recorded during one test, newest-irrelevant (keyed by pid). */ -async function readRunRecords(logDir: string): Promise { - const names = existsSync(logDir) ? readdirSync(logDir) : [] - const records: PiRunRecord[] = [] - for (const name of names) { - records.push(JSON.parse(await readFile(join(logDir, name), 'utf8')) as PiRunRecord) - } - return records -} - -function configPathOf(argv: ReadonlyArray): string | undefined { - const index = argv.indexOf(PI_MCP_CONFIG_FLAG) - return index >= 0 ? argv[index + 1] : undefined -} - -/** A run: one pi launch in `cwd` with `profile`, returning everything pi observed. */ -async function runPi( - profile: AgentProfile, - cwd: string, -): Promise<{ - argv: string[] | undefined - mcpConfig: string | null | undefined - out: unknown - derived: ReadonlyArray | undefined - error: Error | undefined -}> { - const logDir = join(cwd, 'pi-logs') - await mkdir(logDir, { recursive: true }) - const ctx: ExecutorContext = { - signal: new AbortController().signal, - seams: { [piSeamKey]: { bin: fakePi, cwd, env: { PI_LOG_DIR: logDir } } }, - } - const spec: AgentSpec = { profile, harness: null } - const executor = piExecutor(spec, ctx) - let error: Error | undefined - try { - for await (const _ of executor.execute( - 'do the work', - ctx.signal, - ) as AsyncIterable) { - // drain - } - } catch (err) { - error = err instanceof Error ? err : new Error(String(err)) - } - await executor.teardown('brutalKill') - const records = await readRunRecords(logDir) - const record = records[0] - return { - argv: record?.argv, - mcpConfig: record?.mcp, - out: error ? undefined : executor.resultArtifact().out, - derived: executor.progress?.()?.derived, - error, - } -} - -describe('buildPiMcpServers — the canonical body pi-mcp-adapter reads', () => { - it('emits stdio and remote entries in the shape cli-bridge already proved, dropping disabled', () => { - const servers = buildPiMcpServers({ - coordination: { transport: 'http', url: 'http://127.0.0.1:7331/mcp' }, - local: { - command: 'node', - args: [{ kind: 'public', value: 'server.mjs' }], - env: { LOG: { kind: 'public', value: 'debug' } }, - cwd: '/srv/tools', - }, - off: { enabled: false }, - hollow: { command: '' } as never, - }) - - expect(servers).toEqual({ - coordination: { type: 'http', url: 'http://127.0.0.1:7331/mcp', directTools: true }, - local: { - command: 'node', - args: ['server.mjs'], - env: { LOG: 'debug' }, - cwd: '/srv/tools', - directTools: true, - }, - }) - }) - - it('is byte-identical to cli-bridge buildCanonicalMcpServers on the shared field set', () => { - // cli-bridge `profile-support.ts:253` emits stdio as `{command, args, env, timeout}` and remote - // as `{type, url, headers, timeout}`, in that key order. `AgentProfileMcpServer` has no - // `timeout`, so for any server BOTH input types can express, the bytes match exactly — key - // order included, because this file is read by two different writers — EXCEPT for the trailing - // `directTools`, which this writer adds deliberately and cli-bridge does not yet. It is pinned - // as the last key of every entry so the divergence is exactly one field, visible in the bytes. - const bytes = JSON.stringify( - buildPiMcpServers({ - local: { - transport: 'stdio', - command: 'node', - args: [{ kind: 'public', value: 'server.mjs' }], - env: { LOG: { kind: 'public', value: 'debug' } }, - }, - remote: { transport: 'http', url: 'https://api.example/mcp' }, - legacy: { - transport: 'sse', - url: 'https://api.example/sse', - headers: { 'X-Tenant': { kind: 'public', value: 'acme' } }, - }, - }), - ) - - expect(bytes).toBe( - '{"local":{"command":"node","args":["server.mjs"],"env":{"LOG":"debug"},"directTools":true},' + - '"remote":{"type":"http","url":"https://api.example/mcp","directTools":true},' + - '"legacy":{"type":"sse","url":"https://api.example/sse","headers":{"X-Tenant":"acme"},' + - '"directTools":true}}', - ) - - // Strip the one deliberate addition and cli-bridge parity is exact again. - const withoutDirect = bytes.replaceAll(',"directTools":true', '') - expect(withoutDirect).toBe( - '{"local":{"command":"node","args":["server.mjs"],"env":{"LOG":"debug"}},' + - '"remote":{"type":"http","url":"https://api.example/mcp"},' + - '"legacy":{"type":"sse","url":"https://api.example/sse","headers":{"X-Tenant":"acme"}}}', - ) - }) - - it('accepts `sse` and round-trips the declared transport verbatim', () => { - // The adapter connects a URL server by probing StreamableHTTP and falling back to - // SSEClientTransport when the probe fails (pi-mcp-adapter server-manager.ts:765), so an - // SSE-only endpoint does connect. Refusing it would reject a server pi can actually run. - expect(buildPiMcpServers({ feed: { transport: 'sse', url: 'https://x/sse' } })).toEqual({ - feed: { type: 'sse', url: 'https://x/sse', directTools: true }, - }) - }) - - it('treats a url with no declared transport as http rather than dropping it', () => { - // `transport` is optional on AgentProfileRemoteMcpServer; a silent drop here is the exact bug - // this module exists to fix. - expect(buildPiMcpServers({ remote: { url: 'https://x/mcp' } })).toEqual({ - remote: { type: 'http', url: 'https://x/mcp', directTools: true }, - }) - }) - - it('drops an explicit stdio entry that carries no command, exactly as cli-bridge drops it', () => { - expect( - buildPiMcpServers({ broken: { transport: 'stdio', url: 'https://x/mcp' } as never }), - ).toEqual({}) - }) - - it('refuses a secret-ref instead of writing a placeholder into the config', () => { - expect(() => - buildPiMcpServers({ - api: { - transport: 'http', - url: 'https://api.example/mcp', - headers: { Authorization: { kind: 'secret-ref', key: 'API_TOKEN', format: 'bearer' } }, - }, - }), - ).toThrow(ConfigError) - }) -}) - -describe('piMcpAdapterAvailable — detection, never assumption', () => { - it('finds an npm-installed adapter, and reports a bare agent home as missing', () => { - withAdapter() - expect(piMcpAdapterAvailable()).toBe(true) - withoutAdapter() - expect(piMcpAdapterAvailable()).toBe(false) - }) - - it('finds an adapter listed only in settings.json packages', async () => { - const agentDir = join(root, 'agent-settings-only') - await mkdir(agentDir, { recursive: true }) - await writeFile( - join(agentDir, 'settings.json'), - JSON.stringify({ packages: ['npm:pi-mcp-adapter'] }), - ) - process.env.PI_CODING_AGENT_DIR = agentDir - delete process.env[PI_MCP_ADAPTER_ENV] - expect(piMcpAdapterAvailable()).toBe(true) - }) - - it('honors the env override in both directions for vendored installs', () => { - withoutAdapter() - process.env[PI_MCP_ADAPTER_ENV] = '1' - expect(piMcpAdapterAvailable()).toBe(true) - withAdapter() - process.env[PI_MCP_ADAPTER_ENV] = '0' - expect(piMcpAdapterAvailable()).toBe(false) - }) -}) - -describe('derivePiExtensionArgs — the `--no-extensions` trap', () => { - it('emits nothing when the profile declares no load list, so pi keeps its own discovery', () => { - withAdapter() - expect(derivePiExtensionArgs({}, true)).toEqual({ - args: [], - entries: [], - adapterInjected: false, - }) - }) - - it('adds the adapter to a load array that omits it, and reports the addition', () => { - // THE defect this guards: `--no-extensions` suppresses the settings.json `packages` list too, - // so `load: ['pi-memory']` next to `mcp` would kill the adapter and mount zero servers even - // though the adapter is installed. - withAdapter() - const derived = derivePiExtensionArgs({ extensions: { pi: { load: ['pi-memory'] } } }, true) - - expect(derived.adapterInjected).toBe(true) - expect(derived.entries).toEqual([memoryEntry, adapterEntry]) - expect(derived.args).toEqual([ - '--no-extensions', - '--extension', - memoryEntry, - '--extension', - adapterEntry, - ]) - }) - - it('does not double-add an adapter the profile already listed', () => { - withAdapter() - const derived = derivePiExtensionArgs( - { extensions: { pi: { load: ['pi-mcp-adapter', 'pi-memory'] } } }, - true, - ) - expect(derived.adapterInjected).toBe(false) - expect(derived.entries).toEqual([adapterEntry, memoryEntry]) - }) - - it('recognizes the adapter given as an absolute path', () => { - withAdapter() - const derived = derivePiExtensionArgs({ extensions: { pi: { load: [adapterEntry] } } }, true) - expect(derived.adapterInjected).toBe(false) - expect(derived.entries).toEqual([adapterEntry]) - }) - - it('keeps `load: []` a clean zero-extension control when no MCP is declared', () => { - withAdapter() - expect(derivePiExtensionArgs({ extensions: { pi: { load: [] } } }, false)).toEqual({ - args: ['--no-extensions'], - entries: [], - adapterInjected: false, - }) - }) - - it('rejects an unresolvable extension rather than running an ablation arm without it', () => { - withAdapter() - expect(() => - derivePiExtensionArgs({ extensions: { pi: { load: ['pi-not-installed'] } } }, false), - ).toThrow(ConfigError) - }) - - it('rejects a blank or non-string load entry instead of silently dropping it', () => { - withAdapter() - expect(() => derivePiExtensionArgs({ extensions: { pi: { load: [' '] } } }, false)).toThrow( - ValidationError, - ) - expect(() => derivePiExtensionArgs({ extensions: { pi: { load: [7] } } }, false)).toThrow( - ValidationError, - ) - }) -}) - -describe('mountPiMcpConfig — one config file per worker execution', () => { - it('writes the config OUTSIDE the worker workspace and removes it whole on cleanup', async () => { - const cwd = await freshCwd() - const mount = mountPiMcpConfig( - { coordination: { type: 'http', url: 'http://x/mcp' } }, - { cwd, runId: 'pi-worker-1' }, - ) - - expect(mount).not.toBeNull() - expect(mount?.serverNames).toEqual(['coordination']) - // `--mcp-config` carries an absolute path, so the file never needs to sit in the workspace — - // and a scratch file the worker did not create is one it may commit, diff, or clean. - expect(mount?.configPath.startsWith(cwd)).toBe(false) - expect(mount?.configPath).toContain('pi-worker-1') - expect(JSON.parse(await readFile(mount?.configPath as string, 'utf8'))).toEqual({ - mcpServers: { coordination: { type: 'http', url: 'http://x/mcp' } }, - }) - - mount?.cleanup() - expect(existsSync(mount?.configPath as string)).toBe(false) - expect(mcpDirsIn(cwd)).toEqual([]) - }) - - it('gives two mounts from ONE cwd two different paths instead of refusing the second', async () => { - // The defect: `PiSeam.cwd` comes from the ExecutorConfig, so EVERY worker the factory builds - // shares it. A per-cwd config file makes the ordinary supervisor case a collision. - const cwd = await freshCwd() - const first = mountPiMcpConfig({ a: { command: 'a' } }, { cwd, runId: 'pi-worker-1' }) - const second = mountPiMcpConfig({ b: { command: 'b' } }, { cwd, runId: 'pi-worker-1' }) - - expect(first?.configPath).not.toBe(second?.configPath) - expect(JSON.parse(await readFile(first?.configPath as string, 'utf8'))).toEqual({ - mcpServers: { a: { command: 'a' } }, - }) - expect(JSON.parse(await readFile(second?.configPath as string, 'utf8'))).toEqual({ - mcpServers: { b: { command: 'b' } }, - }) - - first?.cleanup() - // One worker's cleanup must not touch the other's still-live config. - expect(existsSync(second?.configPath as string)).toBe(true) - second?.cleanup() - expect(mcpDirsIn(cwd)).toEqual([]) - }) - - it('is idempotent, so cleanup on both the throw path and the finally path is safe', async () => { - const cwd = await freshCwd() - const mount = mountPiMcpConfig({ a: { command: 'a' } }, { cwd, runId: 'r' }) - mount?.cleanup() - expect(() => mount?.cleanup()).not.toThrow() - expect(mcpDirsIn(cwd)).toEqual([]) - }) - - it('uses an OS temp directory — never process.cwd() — when the seam names no cwd', async () => { - const mount = mountPiMcpConfig({ a: { command: 'a' } }, { runId: 'pi-worker-2' }) - expect(mount?.configPath.startsWith(tmpdir())).toBe(true) - expect(mount?.configPath.startsWith(process.cwd())).toBe(false) - expect(existsSync(mount?.configPath as string)).toBe(true) - mount?.cleanup() - expect(existsSync(mount?.configPath as string)).toBe(false) - }) - - it('mounts even when the seam names a cwd that does not exist, because the config is not in it', async () => { - // The config lives in the OS temp directory and reaches pi by `--mcp-config`, so the seam's - // cwd has no bearing on whether the mount can be written. - const mount = mountPiMcpConfig( - { a: { command: 'a' } }, - { cwd: join(root, 'no-such-cwd'), runId: 'r' }, - ) - expect(mount?.configPath.startsWith(tmpdir())).toBe(true) - mount?.cleanup() - }) - - it('returns null for an empty server set so the no-MCP path writes nothing', async () => { - const cwd = await freshCwd() - expect(mountPiMcpConfig({}, { cwd, runId: 'r' })).toBeNull() - expect(mcpDirsIn(cwd)).toEqual([]) - }) -}) - -describe('preparePiMcp — fail closed before anything is written', () => { - it('throws with the adapter name, both lookup locations, and the install command', async () => { - withoutAdapter() - const cwd = await freshCwd() - let caught: Error | undefined - try { - preparePiMcp( - { mcp: { coordination: { transport: 'http', url: 'http://x/mcp' } } }, - { - cwd, - runId: 'r', - }, - ) - } catch (err) { - caught = err as Error - } - - expect(caught).toBeInstanceOf(ConfigError) - expect(caught?.message).toContain('pi-mcp-adapter') - expect(caught?.message).toContain(join(agentDirBare, 'npm', 'node_modules', 'pi-mcp-adapter')) - expect(caught?.message).toContain(join(agentDirBare, 'settings.json')) - expect(caught?.message).toContain('pi install npm:pi-mcp-adapter') - // Fail-closed means nothing was written on the way out. - expect(mcpDirsIn(cwd)).toEqual([]) - }) - - it('does not trip the adapter gate on a block of disabled servers', async () => { - withoutAdapter() - const cwd = await freshCwd() - const prepared = preparePiMcp({ mcp: { off: { enabled: false } } }, { cwd, runId: 'r' }) - expect(prepared).toEqual({ args: [], mount: null, receipt: undefined }) - expect(mcpDirsIn(cwd)).toEqual([]) - }) - - it('emits --mcp-config after the extension flags, pointing at the mounted file', async () => { - withAdapter() - const cwd = await freshCwd() - const prepared = preparePiMcp( - { - mcp: { coordination: { transport: 'http', url: 'http://x/mcp' } }, - extensions: { pi: { load: ['pi-memory'] } }, - }, - { cwd, runId: 'r' }, - ) - - expect(prepared.args).toEqual([ - '--no-extensions', - '--extension', - memoryEntry, - '--extension', - adapterEntry, - '--mcp-config', - prepared.mount?.configPath, - ]) - expect(prepared.receipt?.configPath).toBe(prepared.mount?.configPath) - prepared.mount?.cleanup() - }) - - it('never mutates the caller`s profile object', async () => { - // The adapter injection and the mount are DERIVED state. A profile that came back changed - // would silently rewrite the caller's own arm definition — an ablation arm that mutates the - // thing it is ablating measures nothing. - withAdapter() - const cwd = await freshCwd() - const profile: AgentProfile = { - name: 'ablation-arm', - mcp: { - coordination: { transport: 'http', url: 'http://x/mcp' }, - off: { enabled: false }, - }, - extensions: { pi: { load: ['pi-memory'] } }, - } - const before = structuredClone(profile) - deepFreeze(profile) - - const prepared = preparePiMcp(profile, { cwd, runId: 'r' }) - prepared.mount?.cleanup() - - expect(prepared.receipt?.adapterInjected).toBe(true) - expect(profile).toEqual(before) - // The load array specifically: the adapter was appended to what pi received, not to this list. - expect(profile.extensions?.pi?.load).toEqual(['pi-memory']) - }) -}) - -/** Freeze deeply, so an attempted mutation THROWS in strict mode rather than being compared away. */ -function deepFreeze(value: T): T { - if (value && typeof value === 'object' && !Object.isFrozen(value)) { - Object.freeze(value) - for (const inner of Object.values(value)) deepFreeze(inner) - } - return value -} - -describe('piExecutor — a profile`s MCP servers reach the real pi process', () => { - it('mounts the declared servers where pi-mcp-adapter looks, and cleans up after', async () => { - withAdapter() - const cwd = await freshCwd() - const run = await runPi( - { - name: 'worker', - mcp: { coordination: { transport: 'http', url: 'http://127.0.0.1:1/mcp' } }, - }, - cwd, - ) - - expect(run.error).toBeUndefined() - // The bytes are what pi itself read from the path it was told to read at startup. - expect(JSON.parse(run.mcpConfig ?? '{}')).toEqual({ - mcpServers: { - coordination: { type: 'http', url: 'http://127.0.0.1:1/mcp', directTools: true }, - }, - }) - const configPath = configPathOf(run.argv ?? []) - expect(run.out).toMatchObject({ - content: 'done', - mcp: { - servers: ['coordination'], - configPath, - extensions: [], - adapterInjected: false, - }, - }) - // No load array, so pi keeps its own discovery and the adapter auto-loads. - expect(run.argv).not.toContain('--no-extensions') - expect(mcpDirsIn(cwd)).toEqual([]) - }) - - it('gives two workers built from ONE seam their own config, concurrently', async () => { - // The ordinary supervisor case: one ExecutorConfig, one `PiSeam.cwd`, many children. A - // cwd-discovered config file made this hard-fail; a per-execution file makes it routine. - withAdapter() - const cwd = await freshCwd() - const logDir = join(cwd, 'pi-logs') - await mkdir(logDir, { recursive: true }) - const ctx: ExecutorContext = { - signal: new AbortController().signal, - seams: { - [piSeamKey]: { bin: fakePi, cwd, env: { PI_LOG_DIR: logDir, PI_HOLD_MS: '150' } }, - }, - } - - // BOTH children carry the SAME profile name — N replicas of one arm is the common shape, and - // it makes the executor's own `pi--` run id identical for two workers built - // in the same millisecond. Nothing but a per-execution path can separate them. - const build = (server: string) => - piExecutor( - { - profile: { - name: 'worker', - mcp: { [server]: { transport: 'http', url: `http://127.0.0.1:1/${server}` } }, - }, - harness: null, - } satisfies AgentSpec, - ctx, - ) - - const alpha = build('tools-alpha') - const beta = build('tools-beta') - const drain = async (e: ReturnType): Promise => { - for await (const _ of e.execute('do the work', ctx.signal) as AsyncIterable) { - // drain - } - } - await Promise.all([drain(alpha), drain(beta)]) - await Promise.all([alpha.teardown('brutalKill'), beta.teardown('brutalKill')]) - - const alphaOut = alpha.resultArtifact().out as { mcp?: { configPath: string } } - const betaOut = beta.resultArtifact().out as { mcp?: { configPath: string } } - // Two workers, two configs. Same cwd, no collision, no refusal. - expect(alphaOut.mcp?.configPath).toBeDefined() - expect(betaOut.mcp?.configPath).toBeDefined() - expect(alphaOut.mcp?.configPath).not.toBe(betaOut.mcp?.configPath) - - const records = await readRunRecords(logDir) - expect(records).toHaveLength(2) - const mounted = records - .map((r) => ({ - configPath: configPathOf(r.argv), - servers: Object.keys( - (JSON.parse(r.mcp ?? '{}') as { mcpServers?: Record }).mcpServers ?? {}, - ), - })) - .sort((l, r) => (l.servers[0] ?? '').localeCompare(r.servers[0] ?? '')) - // Each pi process read ITS OWN file and saw only its own server. - expect(mounted.map((m) => m.servers)).toEqual([['tools-alpha'], ['tools-beta']]) - expect(mounted[0]?.configPath).not.toBe(mounted[1]?.configPath) - expect(new Set([mounted[0]?.configPath, mounted[1]?.configPath])).toEqual( - new Set([alphaOut.mcp?.configPath, betaOut.mcp?.configPath]), - ) - expect(mcpDirsIn(cwd)).toEqual([]) - }) - - it('keeps `--no-extensions` from suppressing the adapter, and RECORDS the addition', async () => { - withAdapter() - const cwd = await freshCwd() - const run = await runPi( - { - name: 'ablation-arm', - mcp: { coordination: { transport: 'http', url: 'http://127.0.0.1:1/mcp' } }, - extensions: { pi: { load: ['pi-memory'] } }, - }, - cwd, - ) - - expect(run.error).toBeUndefined() - const configPath = configPathOf(run.argv ?? []) - expect(run.argv).toEqual([ - '--mode', - 'rpc', - '--no-extensions', - '--extension', - memoryEntry, - '--extension', - adapterEntry, - '--mcp-config', - configPath, - ]) - // The profile said `['pi-memory']`; pi got two extensions. That difference is on the receipt. - expect(run.out).toMatchObject({ - mcp: { adapterInjected: true, extensions: [memoryEntry, adapterEntry] }, - }) - }) - - it('reports the adapter injection on the progress channel even when the run FAILS', async () => { - // `resultArtifact()` throws until the stream drains, so a run that dies after the injection - // could never report it. `progress().derived` is the channel that still answers. - withAdapter() - const cwd = await freshCwd() - const ctx: ExecutorContext = { - signal: new AbortController().signal, - // A binary that does not exist: pi emits `error` and the stream throws. - seams: { [piSeamKey]: { bin: join(root, 'no-such-pi'), cwd, turnTimeoutMs: 2_000 } }, - } - const executor = piExecutor( - { - profile: { - name: 'doomed', - mcp: { coordination: { transport: 'http', url: 'http://127.0.0.1:1/mcp' } }, - extensions: { pi: { load: ['pi-memory'] } }, - }, - harness: null, - } satisfies AgentSpec, - ctx, - ) - - await expect(async () => { - for await (const _ of executor.execute('work', ctx.signal) as AsyncIterable) { - // drain - } - }).rejects.toThrow() - - // The artifact channel cannot answer at all. - expect(() => executor.resultArtifact()).toThrow(ValidationError) - const derived = executor.progress?.()?.derived ?? [] - expect(derived.some((line) => line.includes('--mcp-config'))).toBe(true) - expect(derived.some((line) => line.includes('pi-mcp-adapter'))).toBe(true) - // And the workspace is clean despite the failure. - expect(mcpDirsIn(cwd)).toEqual([]) - }) - - it('throws before spawning pi when the adapter is not installed', async () => { - withoutAdapter() - const cwd = await freshCwd() - const run = await runPi( - { - name: 'worker', - mcp: { coordination: { transport: 'http', url: 'http://127.0.0.1:1/mcp' } }, - }, - cwd, - ) - - expect(run.error).toBeInstanceOf(ConfigError) - expect(run.error?.message).toContain('pi install npm:pi-mcp-adapter') - // Never spawned: no record at all, and no config left in the workspace. - expect(run.argv).toBeUndefined() - expect(mcpDirsIn(cwd)).toEqual([]) - }) - - it('is a pure no-op for a profile with no MCP — no file, no flag, no receipt', async () => { - // The regression that matters most: this path must be indistinguishable from before the fix. - withoutAdapter() - const cwd = await freshCwd() - const run = await runPi({ name: 'worker' }, cwd) - - expect(run.error).toBeUndefined() - expect(run.mcpConfig).toBeNull() - expect(run.argv).toEqual(['--mode', 'rpc']) - expect(run.out).toEqual({ content: 'done', turns: 1 }) - expect(run.derived).toBeUndefined() - expect(mcpDirsIn(cwd)).toEqual([]) - }) - - it('removes the config even when the run fails mid-flight', async () => { - withAdapter() - const cwd = await freshCwd() - const ctx: ExecutorContext = { - signal: new AbortController().signal, - seams: { - [piSeamKey]: { - // A binary that does not exist: pi emits `error`, the stream throws, and the `finally` - // still owes the workspace its cleanup. - bin: join(root, 'no-such-pi'), - cwd, - turnTimeoutMs: 2_000, - }, - }, - } - const spec: AgentSpec = { - profile: { mcp: { coordination: { transport: 'http', url: 'http://127.0.0.1:1/mcp' } } }, - harness: null, - } - const executor = piExecutor(spec, ctx) - await expect(async () => { - for await (const _ of executor.execute('work', ctx.signal) as AsyncIterable) { - // drain - } - }).rejects.toThrow() - - expect(mcpDirsIn(cwd)).toEqual([]) - }) -})