From b7390f7920bf55886b28e5cc8e8362440b1e5b00 Mon Sep 17 00:00:00 2001 From: Denys Rafael Date: Mon, 31 Aug 2026 18:57:28 +0300 Subject: [PATCH 1/2] Add Pi process-local Autopilot execution --- README.md | 2 +- docs/prerequisites.md | 2 +- skills/autopilot/README.md | 4 +- skills/autopilot/SKILL.md | 8 +- ...continuity-evidence-implementation-plan.md | 6 +- ...8-31-cooperative-harness-execution-plan.md | 6 +- skills/autopilot/docs/README.md | 2 +- skills/autopilot/docs/architecture.md | 4 +- skills/autopilot/docs/getting-started.md | 6 +- skills/autopilot/docs/implementation-plan.md | 18 +- skills/autopilot/docs/runtime-cli.md | 10 + skills/autopilot/references/adapters.md | 10 +- skills/autopilot/references/recovery.md | 2 +- .../autopilot/runtime/adapters/pi/bridge.ts | 195 -------- .../runtime/adapters/pi/in-process.ts | 392 ++++++++++++++++ skills/autopilot/runtime/adapters/pi/index.ts | 76 +-- .../runtime/dist/adapters/pi/bridge.d.ts | 20 - .../runtime/dist/adapters/pi/bridge.js | 144 ------ .../runtime/dist/adapters/pi/in-process.d.ts | 27 ++ .../runtime/dist/adapters/pi/in-process.js | 334 +++++++++++++ .../runtime/dist/adapters/pi/index.js | 63 +-- skills/autopilot/runtime/dist/src/cli.d.ts | 11 + skills/autopilot/runtime/dist/src/cli.js | 61 ++- skills/autopilot/runtime/dist/src/doctor.js | 4 +- .../runtime/dist/src/pi-extension-entry.d.ts | 2 + .../runtime/dist/src/pi-extension-entry.js | 5 + .../runtime/dist/src/pi-extension.d.ts | 24 + .../runtime/dist/src/pi-extension.js | 117 +++++ .../runtime/dist/src/pi-subagents.d.ts | 5 + .../runtime/dist/src/pi-subagents.js | 31 ++ skills/autopilot/runtime/package-lock.json | 8 + skills/autopilot/runtime/package.json | 13 + skills/autopilot/runtime/src/cli.ts | 102 +++- skills/autopilot/runtime/src/doctor.ts | 4 +- skills/autopilot/runtime/src/pi-core.d.ts | 3 + .../runtime/src/pi-extension-entry.ts | 6 + skills/autopilot/runtime/src/pi-extension.ts | 182 +++++++ skills/autopilot/runtime/src/pi-subagents.ts | 37 ++ .../autopilot/runtime/test/packaging.test.ts | 4 + .../runtime/test/pi-subagents.test.ts | 444 +++++++++++++++--- 40 files changed, 1769 insertions(+), 625 deletions(-) delete mode 100644 skills/autopilot/runtime/adapters/pi/bridge.ts create mode 100644 skills/autopilot/runtime/adapters/pi/in-process.ts delete mode 100644 skills/autopilot/runtime/dist/adapters/pi/bridge.d.ts delete mode 100644 skills/autopilot/runtime/dist/adapters/pi/bridge.js create mode 100644 skills/autopilot/runtime/dist/adapters/pi/in-process.d.ts create mode 100644 skills/autopilot/runtime/dist/adapters/pi/in-process.js create mode 100644 skills/autopilot/runtime/dist/src/pi-extension-entry.d.ts create mode 100644 skills/autopilot/runtime/dist/src/pi-extension-entry.js create mode 100644 skills/autopilot/runtime/dist/src/pi-extension.d.ts create mode 100644 skills/autopilot/runtime/dist/src/pi-extension.js create mode 100644 skills/autopilot/runtime/src/pi-core.d.ts create mode 100644 skills/autopilot/runtime/src/pi-extension-entry.ts create mode 100644 skills/autopilot/runtime/src/pi-extension.ts diff --git a/README.md b/README.md index 553473c..24aa6e7 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,7 @@ integration tests, and application-context smoke tests pass; no Spring Boot 3.x artifacts remain; and the migration guide is updated. Open a PR or MR, but do not merge or deploy. ``` -Autopilot requires Node.js 24+ and Git. Optional harness integrations can provide delegated workers and live activity; Autopilot falls back to direct harness execution when they are unavailable. Remote delivery also requires `gh` or `glab` and explicit per-run grants. +Autopilot requires Node.js 24+ and Git. Pi can load the packaged Autopilot extension and an active `pi-subagents` owner for process-local structured execution; an unavailable owner selects the distinct direct Pi CLI fallback before admission. Autopilot never installs or enables these components. Remote delivery also requires `gh` or `glab` and explicit per-run grants. Use `/autopilot status` to check overnight progress, `/autopilot resume` to continue an interrupted nonterminal run, or `/autopilot stop` to end a run while preserving its work. `/autopilot address review comments` snapshots feedback from the exact open PR/MR, creates a sealed amendment successor, and resolves provider-resolvable threads only after the fix passes. After provider-confirmed merge, `/autopilot wrap up` performs guarded remote-branch, sibling-worktree, local-branch, and amendment-chain cleanup; use `/autopilot wrap up with handoff` to preserve project-local summaries. diff --git a/docs/prerequisites.md b/docs/prerequisites.md index dc6179b..62714e0 100644 --- a/docs/prerequisites.md +++ b/docs/prerequisites.md @@ -13,7 +13,7 @@ The [`autopilot`](../skills/autopilot) skill requires these tools: - Node.js 24 or newer - Git - At least one supported harness CLI: Claude Code, Codex, Pi, or OpenCode -- Optional: `pi-subagents` 0.53.0 or newer for delegated Pi workers and visible activity; direct Pi remains available as a fallback +- Optional: an installed and active `pi-subagents` 0.53.0 or newer for the process-local Pi backend; the packaged Autopilot Pi extension probes it before launch, and direct Pi remains a distinct fallback - `gh` for GitHub delivery or `glab` for GitLab delivery Autopilot checks these tools automatically before starting a run and reports anything missing. It never installs tools, downloads runtimes, or changes authentication. diff --git a/skills/autopilot/README.md b/skills/autopilot/README.md index 05674b2..d55bdb7 100644 --- a/skills/autopilot/README.md +++ b/skills/autopilot/README.md @@ -6,7 +6,7 @@ Autopilot is not a prompt loop. It seals your request into an immutable charter, ## Use it -You need Node.js 24+, Git, and at least one supported harness CLI. Remote delivery also needs authenticated `gh` or `glab`. Autopilot checks the environment before launch; it never installs tools or signs you in. +You need Node.js 24+, Git, and at least one supported harness CLI. Remote delivery also needs authenticated `gh` or `glab`. Pi can use the packaged Autopilot extension with an installed and active `pi-subagents` 0.53.0+ owner for process-local execution; otherwise it reports the distinct direct CLI fallback. Autopilot checks the environment before launch; it never installs tools, enables extensions, or signs you in. From the project repository, invoke the skill through your host: @@ -19,6 +19,8 @@ replace removed APIs, and preserve the existing HTTP and persistence behavior. D changes and rollback notes. Open a PR, but do not merge or deploy it. ``` +When using Pi's process-local backend, load `runtime/dist/src/pi-extension-entry.js` through Pi's documented extension or package mechanism, then start the sealed charter with `/autopilot-start `. The owning Pi process must remain uninterrupted; losing it makes the exact attempt unknown and never triggers a replacement. + Autopilot will: 1. turn the request into explicit work items, completion predicates, budgets, and grants; diff --git a/skills/autopilot/SKILL.md b/skills/autopilot/SKILL.md index 071c321..bdfe020 100644 --- a/skills/autopilot/SKILL.md +++ b/skills/autopilot/SKILL.md @@ -7,7 +7,7 @@ disable-model-invocation: true # Autopilot -Autopilot delegates bounded coding work to a fresh Claude Code, Codex, Pi, or OpenCode process. Its runtime owns lifecycle state, Git commits, verification, remote delivery, and completion decisions. +Autopilot delegates bounded coding work to a fresh Claude Code, Codex, Pi, or OpenCode execution. Pi implementations prefer the packaged process-local extension backend; other modes and the Pi fallback use their declared CLI boundaries. The runtime owns lifecycle state, Git commits, verification, remote delivery, and completion decisions. ## New-run preconditions @@ -64,13 +64,13 @@ Start the successor normally. The runtime revalidates the immutable feedback sna 7. Set `commitPolicy.preCommitHook` explicitly. Prefer `run` when the repository configures a project pre-commit hook; use `skip` only when the user approves bypassing that project policy. Inspect hook code without executing it and include its known outputs in `commitPolicy.writableRoots`, repository writable roots, and runtime `files.read`/`files.write` grants without widening the worker's roots. Ask when those effects cannot be bounded. 8. Save the proposed charter outside the repository or in an explicitly writable documentation path. 9. Before starting the foreground process, report the work titles, branches, delivery boundary, and known unverified boundaries. The skill can rediscover the run later; do not ask the user to record runtime paths or identifiers. -10. Start in the foreground: +10. Start in the foreground. For Pi, when the packaged Autopilot extension and compatible process-local `pi-subagents` owner are active, instruct the operator to run `/autopilot-start ` in that owning Pi session; do not start a second outer Pi process. For other harnesses, or when Pi reports the distinct direct fallback, run: ```bash node runtime/dist/src/cli.js start ``` -11. During Pi runs, surface the runtime's stderr activity lines instead of hiding the command output. When `pi-subagents` 0.53.0+ is available, these lines show the delegated worker's live tool, token, and terminal activity; otherwise report the direct-worker fallback. +11. During Pi runs, surface bounded stderr activity instead of hiding it. The process-local path remains visible through Pi's ordinary foreground subagent observability and binds completion to the exact extension instance. An unavailable owner is selected and reported as the direct fallback before admission; never infer one mode's assurance from the other. 12. Report the terminal result and any new unverified boundaries. After every recorded PR/MR is merged, use `wrap-up` only when the user wants Autopilot to remove its exact remote branches, sibling worktrees, local branches, and canonical run-state chain. ## Lifecycle commands @@ -86,6 +86,8 @@ node runtime/dist/src/cli.js [--state-dir ] [--handoff] wrap-up [run-id] node runtime/dist/src/cli.js doctor ``` +The packaged Pi extension additionally provides `/autopilot-start `, `/autopilot-resume [run-id]`, and `/autopilot-recover `. These commands invoke the same runtime core without giving the extension or worker journal ownership. A reload, session replacement, or whole-process loss during an admitted process-local implementation becomes `EXECUTION_STATE_UNKNOWN`. + These are internal and recovery commands; users normally invoke the skill forms above. Use the same `--state-dir` for every direct command addressing a run. Omitted-ID lifecycle commands discover unsuperseded runs for the current repository and mutate only one unambiguous candidate. `pause` enters nonterminal waiting only after active implementation is observed quiescent. `resume` continues an interrupted or paused nonterminal run. A stopped run requires a successor charter. `wrap-up` is destructive: without a run ID it proceeds only when discovery finds exactly one successful unsuperseded provider-delivered run; otherwise it lists candidates without mutation. `--handoff` writes optional Markdown and JSON summaries under `.autopilot/handoffs/` before cleanup. ## Safety rules diff --git a/skills/autopilot/docs/2026-08-30-continuity-evidence-implementation-plan.md b/skills/autopilot/docs/2026-08-30-continuity-evidence-implementation-plan.md index 385b2ff..ac0de7a 100644 --- a/skills/autopilot/docs/2026-08-30-continuity-evidence-implementation-plan.md +++ b/skills/autopilot/docs/2026-08-30-continuity-evidence-implementation-plan.md @@ -162,7 +162,7 @@ Add `test/attempt-context.test.ts` only if context assembly becomes a distinct p 5. Write an immutable, user-only context artifact under the existing attempt-report area before recording `ATTEMPT_STARTED`. A crash may leave an unreferenced artifact; it must not create a lifecycle transition. 6. Add the context hash and source journal sequence or hash to `ATTEMPT_STARTED`. Parse these fields compatibly for journals created before this change. 7. Render one worker prompt from the normalized context. Adapter-specific argument builders may change transport syntax, but they must not add authority or omit required semantics. -8. Continue to pass the rendered task through the Pi bridge. Do not teach the bridge a second context format unless Pi needs structured fields for a demonstrated capability. +8. Pass the same rendered task through Pi process-local structured delegation or the distinct direct fallback. Do not introduce a second context format unless a demonstrated capability requires structured fields. 9. On a replacement attempt, build a new context from reconciled state. Never reuse a stale lease, deadline, expected head, or prior context artifact. 10. Treat repository guidance and review findings as labeled data. They may appear only through sealed assumptions or normalized evidence fields and cannot alter grants or predicates. @@ -487,7 +487,7 @@ skills/autopilot/runtime/test/fault-injection.test.ts ## Phase 7: Finish documentation and packaging -**Result:** Implemented for Phases 0–6A. Phase 6B is not promoted under the recorded no-receiver decision and is not a current release blocker. The current validation baseline is 162 Node tests locally and a 93-file package dry run; the previous 137-test baseline passed on Ubuntu and Windows. +**Result:** Implemented for Phases 0–6A. Phase 6B is not promoted under the recorded no-receiver decision and is not a current release blocker. The current validation baseline is 200 Node tests locally, including the later Pi process-local backend; the earlier 162-test and 137-test baselines passed at their recorded revisions on Ubuntu and Windows. ### Files @@ -560,7 +560,7 @@ GitHub Actions run [33335872725](https://github.com/drafael/coding-harness/actio ### Direct Pi validation evidence -Pi 0.84.4 completed five isolated `--no-session --no-extensions` calls against a disposable Git repository. The matrix covered two sequential calls in one worktree, an explicit retry context, two additional calls in a sibling worktree under distinct run identities, and a read-only exact-tree review. Each implementation call created only its unique nonce file with the expected content; the review returned `AUTOPILOT_REVIEW_RESULT:{"verdict":"clean","findings":[]}`. No output contained the historical `ctx is stale` error or either captured-context invalidation phrase. This validates the current direct Pi path for repeated process-isolated calls; it does not change the structured `pi-subagents` worker path or claim compatibility for the retired `/run` slash bridge. +Pi 0.84.4 completed five isolated `--no-session --no-extensions` calls against a disposable Git repository. The matrix covered two sequential calls in one worktree, an explicit retry context, two additional calls in a sibling worktree under distinct run identities, and a read-only exact-tree review. Each implementation call created only its unique nonce file with the expected content; the review returned `AUTOPILOT_REVIEW_RESULT:{"verdict":"clean","findings":[]}`. No output contained the historical `ctx is stale` error or either captured-context invalidation phrase. This remains evidence for the distinct direct Pi fallback at that revision; the later process-local structured backend has its own exact-admission and continuity tests and does not rely on the retired `/run` slash bridge. ## Cross-cutting test matrix diff --git a/skills/autopilot/docs/2026-08-31-cooperative-harness-execution-plan.md b/skills/autopilot/docs/2026-08-31-cooperative-harness-execution-plan.md index bbab8e7..63b399f 100644 --- a/skills/autopilot/docs/2026-08-31-cooperative-harness-execution-plan.md +++ b/skills/autopilot/docs/2026-08-31-cooperative-harness-execution-plan.md @@ -1,6 +1,6 @@ # Cooperative harness execution implementation plan -- **Status:** Approved design; PR 1 decision/promotion shutdown, PR 2 execution assurance, and PR 3 fenced unknown-execution recovery are complete; harness integration remains pending +- **Status:** Approved design; PR 1 decision/promotion shutdown, PR 2 execution assurance, PR 3 fenced unknown-execution recovery, and PR 4 Pi process-local integration are complete; Windows native-path removal remains pending - **Date:** 2026-08-31 - **Audience:** Autopilot implementers and reviewers - **Related:** [Architecture](architecture.md), [continuity implementation plan](2026-08-30-continuity-evidence-implementation-plan.md), [durable event engine ADR](adr/0001-durable-event-engine.md) @@ -453,8 +453,8 @@ Pause and revisit the design if implementation shows any of the following: ## Remaining evidence gaps -- Live Pi fault-injection against extension reload, session replacement, whole-process loss, and late writes has not been run. +- Controlled Pi process-local tests cover exact admission, cancellation, terminal-before-shutdown precedence, reload/session invalidation, lost admission, late/mismatched result rejection, direct fallback, and runtime-core completion in one reused local repository fixture. Whole-process live fault evidence remains environment-specific and does not prove OS quiescence or provider parity. - No provider currently proves Windows process-tree quiescence through its public subagent contract. - Codex app-server live rejoin, OpenCode disconnect reconciliation, and Claude interruption behavior were researched but not exercised for this design. - Cooperative terminality does not prevent external effects performed by worker tools before terminal response. -- The exact Pi extension packaging and invocation surface must be chosen from complete Pi documentation during PR 4. +- The Pi entry point follows the documented package manifest at `runtime/dist/src/pi-extension-entry.js` and registers `/autopilot-start`, `/autopilot-resume`, and `/autopilot-recover`; callers must load it through Pi's normal package or extension mechanism. diff --git a/skills/autopilot/docs/README.md b/skills/autopilot/docs/README.md index 715e7a1..ae69881 100644 --- a/skills/autopilot/docs/README.md +++ b/skills/autopilot/docs/README.md @@ -39,4 +39,4 @@ Both figures use the vendored Diagram Design default profile. The PNG exports ar ## Verification status -The runtime has 162 Node tests locally covering deterministic attempt context, predicate evidence maps, exact-tree review fixtures, local Git lifecycle behavior, crash reconciliation, intentional pause, exact-subject provider waiting, hooks, queues, stacks, sealed review-feedback amendments, GitHub and GitLab provider contracts, sibling worktrees, descendant process cancellation, and wrap-up. Disposable exact-tree reviews passed with Pi 0.84.4, Codex 0.151.0, and OpenCode 1.18.25; Claude Code 2.1.251 remains unverified because its identity-linked API key requires an `ANTHROPIC_WORKSPACE_ID` that is not present in the validation environment. GitHub PR creation, marker reconciliation, exact review-thread resolution, exact-head amendment, merge, and wrap-up passed on authorized private disposable targets with `gh` 2.98.0. GitLab MR creation and reconciliation, exact discussion resolution, duplicate-status latest selection, exact-head amendment, merge, and wrap-up passed on authorized private disposable targets with `glab` 1.115.0. +The runtime has 200 Node tests locally covering deterministic attempt context, predicate evidence maps, exact-tree review fixtures, local Git lifecycle behavior, crash reconciliation, intentional pause, exact-subject provider waiting, hooks, queues, stacks, sealed review-feedback amendments, GitHub and GitLab provider contracts, sibling worktrees, descendant process cancellation, and wrap-up. Controlled Pi 0.84.4 with pi-subagents 0.60.0 process-local tests cover exact admission, cancellation, extension-context loss, direct fallback, and runtime-core completion in a reused local repository; disposable exact-tree reviews also passed with Pi 0.84.4, Codex 0.151.0, and OpenCode 1.18.25; Claude Code 2.1.251 remains unverified because its identity-linked API key requires an `ANTHROPIC_WORKSPACE_ID` that is not present in the validation environment. GitHub PR creation, marker reconciliation, exact review-thread resolution, exact-head amendment, merge, and wrap-up passed on authorized private disposable targets with `gh` 2.98.0. GitLab MR creation and reconciliation, exact discussion resolution, duplicate-status latest selection, exact-head amendment, merge, and wrap-up passed on authorized private disposable targets with `glab` 1.115.0. diff --git a/skills/autopilot/docs/architecture.md b/skills/autopilot/docs/architecture.md index 5815f55..62ab672 100644 --- a/skills/autopilot/docs/architecture.md +++ b/skills/autopilot/docs/architecture.md @@ -1,6 +1,6 @@ # Harness-agnostic Autopilot design -- **Status:** Developer-preview implementation available; POSIX attempt-scoped implementation-process reattachment and controlled-fixture sealed restack successors are packaged. Windows restart reattachment remains disabled while approved cooperative harness execution is implemented; notification wake is not promoted, and live restack mutation remains unverified without renewed disposable-target authority. +- **Status:** Developer-preview implementation available; POSIX attempt-scoped process reattachment, fenced unknown recovery, Pi process-local cooperative execution, and controlled-fixture sealed restack successors are packaged. Windows native-path removal remains pending; notification wake is not promoted, and live restack mutation remains unverified without renewed target authority. - **Date:** 2026-08-22 - **Audience:** Coding-harness maintainers and adapter authors - **Implementation plan:** [Autopilot implementation plan](implementation-plan.md) @@ -312,7 +312,7 @@ Capability degradation is explicit: The first adapters target Claude Code, Codex, Pi, and OpenCode. They share one conformance suite. -The Pi adapter prefers the installed `pi-subagents` 0.53.0+ public structured delegation API. Autopilot loads only that extension and its bundled bridge in a headless Pi process, delegates one item to the resolved `worker` role in the runtime-owned worktree, and projects bounded progress to stderr. If the compatible extension is absent, the adapter uses a direct Pi worker and records the fallback. The originating interactive Pi FleetView cannot own this subprocess because Pi's event bus and FleetView are process-local; the stderr projection preserves visible activity without transferring lifecycle authority away from Autopilot. +The packaged Pi extension invokes the same runtime core inside the owning Pi process and probes an installed `pi-subagents` 0.53.0+ owner before selecting process-local structured delegation. The runtime persists admission intent before the extension emits one request, then binds the exact request, logical node, subject, and extension-instance identity. Only one matching terminal response from that uninterrupted instance may proceed to repository verification. Extension reload, session replacement, stale context, process loss, or a missing exact response becomes `EXECUTION_STATE_UNKNOWN` and cannot launch a replacement. Pi's ordinary foreground subagent observability remains available, while bounded activity is also projected to stderr. If the compatible owner is absent or inactive, Autopilot selects and reports the distinct direct Pi CLI fallback before admission. Direct POSIX execution keeps process supervision; direct Windows execution remains session-scoped. Independent review remains a separate direct read-only Pi execution. ## Playbooks diff --git a/skills/autopilot/docs/getting-started.md b/skills/autopilot/docs/getting-started.md index 7894d7e..5cbfdb9 100644 --- a/skills/autopilot/docs/getting-started.md +++ b/skills/autopilot/docs/getting-started.md @@ -10,7 +10,7 @@ Install these yourself before launching a run: - Git - Claude Code, Codex, Pi, or OpenCode - `gh` for GitHub delivery or `glab` for GitLab delivery -- optionally, `pi-subagents` 0.53.0 or newer for structured Pi delegation +- optionally, an installed and active `pi-subagents` 0.53.0 or newer for Pi process-local structured delegation Autopilot checks these requirements before launch. It reports missing or unverified capabilities without installing dependencies, downloading runtimes, authenticating providers, or changing global configuration. @@ -36,7 +36,7 @@ and rollback notes. Open a PR, but do not merge or deploy it. Autopilot converts the request into a proposed charter. Review the charter when the skill asks about credentials, remote writes, merge authority, assumptions, waivers, or hook behavior. It never infers deployment, force-push, unrelated credentials, destructive cleanup, or weaker completion checks. -Leave the foreground process running. During Pi execution, progress is written to stderr while structured output remains machine-readable. +For the Pi process-local backend, load the packaged runtime extension through Pi's documented package or `--extension` mechanism and start the charter with `/autopilot-start `. The extension checks that the compatible `pi-subagents` owner is active in the same process; otherwise it reports and uses the distinct direct Pi CLI fallback. Autopilot never installs or enables either extension. Leave the owning Pi process running. Extension reload, session replacement, or process loss makes an admitted in-process execution unknown rather than launching a replacement. ## Understand the run @@ -70,7 +70,7 @@ Natural requests work too: `status` rebuilds progress from the sealed charter, hash-linked journal, Git identities, and receipts. It reports the last durable milestone, unmet predicate identities, normalized failure, remaining budgets, repeated no-change attempts, and next legal action. `pause` asks the live coordinator to cancel active implementation work, prove quiescence, retire the exact lease, and enter nonterminal waiting. A cancellation caused solely by pause remains auditable but does not consume an attempt. `resume` continues a paused or interrupted nonterminal run within its original limits. Verified items reconcile their checkpoint and effects without rerunning implementation. It does not restart a run that still has a live coordinator. `stop` asks a live coordinator to cancel active adapter work and record a durable terminal stop; if the coordinator is gone, Autopilot records the stop under the run lock. Branches, worktrees, receipts, and evidence remain intact. -A stopped run cannot be resumed. Changed authority, budgets, or objectives require a sealed successor. After coordinator loss on supported POSIX hosts, Autopilot reattaches built-in supervised implementation executions and waits for terminal process-tree evidence. Legacy attempts, review executions, and incomplete or mismatched supervisor artifacts record `EXECUTION_STATE_UNKNOWN` and refuse a replacement launch until quiescence can be proven. +A stopped run cannot be resumed. Changed authority, budgets, or objectives require a sealed successor. After coordinator loss on supported POSIX hosts, Autopilot reattaches built-in supervised implementation executions and waits for terminal process-tree evidence. Legacy attempts, review executions, and incomplete or mismatched supervisor artifacts record `EXECUTION_STATE_UNKNOWN` and refuse a replacement launch until quiescence can be proven. Pi in-process implementations also become unknown when their exact owning extension instance is lost. Resume them through `/autopilot-resume [run-id]` in a loaded Autopilot extension; use fenced `/autopilot-recover` or the runtime CLI to abandon, adopt, or stop an unknown attempt. If several runs match, Autopilot lists their title, short ID, state, progress, and last update. It changes nothing until you choose one, for example `resume 1` or `status spring-boot-4`. diff --git a/skills/autopilot/docs/implementation-plan.md b/skills/autopilot/docs/implementation-plan.md index f4c2374..eff41ed 100644 --- a/skills/autopilot/docs/implementation-plan.md +++ b/skills/autopilot/docs/implementation-plan.md @@ -1,6 +1,6 @@ # Harness-agnostic Autopilot implementation plan -- **Status:** Developer-preview implementation available; POSIX attempt-scoped implementation reattachment, versioned per-mode execution assurance, exact admitted-subject journaling, and controlled-fixture sealed restack successors are packaged. The reviewed Windows x64 Job Object artifact will not be packaged; Windows restart reattachment remains disabled while operator recovery and cooperative harness integration are implemented. +- **Status:** Developer-preview implementation available; POSIX attempt-scoped reattachment, versioned execution assurance, fenced unknown recovery, and the Pi process-local structured delegation backend are packaged. The reviewed Windows x64 Job Object artifact will not be packaged; its source and runtime path remain only until the ordered native-removal change. - **Date:** 2026-08-22 - **Audience:** Autopilot implementers and reviewers - **Governing design:** [Autopilot architecture](architecture.md) @@ -37,10 +37,11 @@ The completed first release must: - Keep harness and delivery adapters outside reducer and journal ownership. - Prefer Node built-ins and small direct modules. Evaluate any production dependency before adding it. - Never auto-install tools, download runtimes, or alter user authentication. +- Local tests may create temporary Git repositories. GitHub and GitLab live validation must reuse an existing authorized validation project (or create one persistent project once and reuse it); do not create a new remote repository per scenario. ## Developer-preview evidence -The implementation currently has 162 Node test cases and a clean-copy package smoke test. Generated attempt context, predicate-to-evidence reports, exact-tree independent review, intentional pause, and exact-subject provider waiting have controlled coverage. Disposable exact-tree review runs passed with Pi 0.84.4 through pi-subagents 0.60.0, Codex 0.151.0, and OpenCode 1.18.25. Claude Code 2.1.251 reached its adapter but reported no usable noninteractive credential source, so its edit and review flows remain unverified. The same suite passes in Node 24 CI on Ubuntu and Windows; Windows coverage includes locking, atomic writes, Git worktrees and governed hooks, provider fixtures, cancellation, and descendant process-tree termination. An authorized GitHub wrap-up was exercised against merged chat4j PR #69. Authorized private GitHub project `drafael/autopilot-amendment-validation` PR #1 exercised immutable feedback capture, exact-head successor adoption, fast-forward update, exact thread resolution, merge, and amendment-chain wrap-up with `gh` 2.98.0. Authorized private GitLab project `drafael/autopilot-amendment-validation` MR !2 exercised the equivalent complete amendment workflow with `glab` 1.115.0. +The implementation currently has 200 Node test cases and a clean-copy package smoke test. Generated attempt context, predicate-to-evidence reports, exact-tree independent review, intentional pause, and exact-subject provider waiting have controlled coverage. Disposable exact-tree review runs passed with Pi 0.84.4 through pi-subagents 0.60.0, Codex 0.151.0, and OpenCode 1.18.25. Claude Code 2.1.251 reached its adapter but reported no usable noninteractive credential source, so its edit and review flows remain unverified. The same suite passes in Node 24 CI on Ubuntu and Windows; Windows coverage includes locking, atomic writes, Git worktrees and governed hooks, provider fixtures, cancellation, and descendant process-tree termination. An authorized GitHub wrap-up was exercised against merged chat4j PR #69. Authorized private GitHub project `drafael/autopilot-amendment-validation` PR #1 exercised immutable feedback capture, exact-head successor adoption, fast-forward update, exact thread resolution, merge, and amendment-chain wrap-up with `gh` 2.98.0. Authorized private GitLab project `drafael/autopilot-amendment-validation` MR !2 exercised the equivalent complete amendment workflow with `glab` 1.115.0. ## Planned package boundary @@ -207,8 +208,8 @@ skills/autopilot/runtime/test/fixtures/adapter-events/ 2. Reject unknown protocol versions and malformed messages. 3. Bound line size, retained output, deadlines, and subprocess lifetime. 4. Forward cancellation to the complete process group where the platform supports it. -5. Implement the Pi adapter against Pi's public JSON-mode CLI, preferring the installed `pi-subagents` 0.53.0+ structured delegation API and retaining a direct-worker fallback. -6. Project bounded delegated-worker activity to stderr so JSON stdout remains machine-readable. +5. Implement the packaged Pi extension against the public process-local `pi-subagents` 0.53.0+ structured delegation API, retaining the direct Pi CLI as an explicitly different fallback. +6. Invoke the runtime core from the owning extension context, bind exact admission and terminal identity, and project bounded worker activity without transferring journal ownership. 7. Report whether restrictions are enforced or cooperative. 8. Run each attempt in a fresh session with only item-scoped context. 9. Build an adapter conformance suite that does not depend on Pi-specific event names. @@ -219,9 +220,10 @@ skills/autopilot/runtime/test/fixtures/adapter-events/ - Duplicate, reordered, truncated, oversized, and malicious event lines cannot advance state. - A silent adapter triggers the idle deadline and bounded recovery path. - A late Pi result with an expired lease is quarantined. -- One disposable local fixture completes through the real Pi adapter. -- A compatible `pi-subagents` fixture proves structured delegation, terminal-result validation, direct fallback behavior, and visible activity without allowing child output to claim completion. -- The fixture records the exact Pi and `pi-subagents` versions and exercised boundary; support remains unverified until that real end-to-end fixture passes. +- One reused local Git repository fixture completes through the process-local runtime entry point; no remote repository is required. +- Compatible `pi-subagents` fixtures prove exact admission, terminal-result validation, cancellation, extension-instance loss, direct fallback behavior, and visible activity without allowing child output to claim completion. +- The fixture records the exact Pi and `pi-subagents` versions and exercised boundary. Cooperative completion remains distinct from OS process-tree proof. +- Live local validation with Pi 0.84.4 and `pi-subagents` 0.60.0 reused one repository: normal completion reached `SUCCEEDED`; whole-process loss resumed as `WAITING/execution-unknown` with exactly one started attempt; and pause during a shell tool call produced an uncharged cancelled attempt and `WAITING/operator-pause`. ## Phase 5: Add Claude Code, Codex, and OpenCode adapters @@ -329,7 +331,7 @@ skills/autopilot/runtime/test/gitlab-delivery.test.ts - Unknown CI failures remain failures. - A named waiver applies only to its configured gate and failure signature. - A stack lands only through its contiguous verified prefix. -- All external mutation tests use disposable repositories. Organization-specific approval and merge policies remain unverified. +- External mutation tests use local disposable repositories or existing authorized reusable GitHub/GitLab validation projects; they do not create one-off remote repositories. Organization-specific approval and merge policies remain unverified. ## Phase 8: Complete the skill, playbooks, doctor, and packaging diff --git a/skills/autopilot/docs/runtime-cli.md b/skills/autopilot/docs/runtime-cli.md index c8bdb8e..6008ebf 100644 --- a/skills/autopilot/docs/runtime-cli.md +++ b/skills/autopilot/docs/runtime-cli.md @@ -29,6 +29,16 @@ The copied-skill entry point is: node "$AUTOPILOT_CLI" COMMAND ``` +When the packaged runtime is loaded as a Pi extension, it also registers: + +```text +/autopilot-start +/autopilot-resume [run-id] +/autopilot-recover +``` + +These commands invoke the same coordinator in the owning Pi process. They select process-local structured delegation only when a compatible installed `pi-subagents` owner answers the process-local probe; otherwise they report the distinct direct Pi CLI fallback before admission. + `status` returns the journal identity, last durable milestone, per-predicate evidence map, remaining budgets, and next legal action without rewriting coordinator-owned reports. Plain output is concise and omits state/worktree paths and full run IDs; `--json` returns the complete machine-readable report. `status`, `resume`, `pause`, and `stop` discover unsuperseded runs for the current repository when the run ID is omitted. `review-feedback` discovers successful `change-request-ready` leaf runs and returns an immutable-input snapshot of unresolved provider feedback for skill-driven amendment compilation. `wrap-up` separately discovers successful provider-delivered leaf runs. Mutating commands proceed only for one unambiguous candidate. A unique short run-ID prefix is accepted; an ambiguous prefix returns choices without mutation. `--json` keeps stdout machine-readable. Diagnostics and live adapter activity remain on stderr. diff --git a/skills/autopilot/references/adapters.md b/skills/autopilot/references/adapters.md index cce6841..a4672e0 100644 --- a/skills/autopilot/references/adapters.md +++ b/skills/autopilot/references/adapters.md @@ -6,7 +6,7 @@ Adapters start fresh noninteractive sessions and normalize observations. They ca | Adapter | Command surface | Assurance | Restart reattachment | Verification status | |---|---|---|---|---| -| Pi | Pi JSON mode plus `pi-subagents` structured delegation when version 0.53.0+ is installed; direct Pi fallback otherwise | Cooperative | Supervised implementation attempts on POSIX; session-scoped on Windows while the approved in-process cooperative backend is implemented | Local edit, exact-tree review, verification, commit, and provider amendment passed with Pi 0.84.4 through pi-subagents 0.60.0; the direct fallback, process-tree cancellation, and restart reattachment have controlled coverage | +| Pi | Owning Pi extension plus process-local `pi-subagents` structured delegation when version 0.53.0+ is installed and active; direct Pi CLI fallback otherwise | Cooperative | In-process implementations require the same uninterrupted extension instance; direct fallback retains supervised POSIX execution and session-scoped Windows execution | Local in-process runtime invocation, exact admission, cancellation, terminal response, continuity loss, direct fallback, and exact-tree verification have controlled coverage with Pi 0.84.4 and pi-subagents 0.60.0; whole-process live fault evidence remains bounded to the documented local matrix | | Claude Code | `claude --print --output-format stream-json --safe-mode ...` | Cooperative | Supervised implementation attempts on POSIX; session-scoped on Windows | The current 2.1.251 credential reaches the API but requires `ANTHROPIC_WORKSPACE_ID`; authenticated edit and review flows remain unverified | | Codex | `codex exec --json --ephemeral --sandbox workspace-write ...` | Cooperative overall; Codex enforces the workspace sandbox, while item-path restrictions are post-checked | Supervised implementation attempts on POSIX; session-scoped on Windows | Disposable local edit, exact-tree review, verification, and commit passed with 0.151.0 | | OpenCode | `opencode run --format json --pure --auto ...` | Cooperative | Supervised implementation attempts on POSIX; session-scoped on Windows | Disposable local edit, exact-tree review, verification, and commit passed with 1.18.25 | @@ -21,9 +21,11 @@ Before launch, `ATTEMPT_STARTED` persists the selected assurance and exact reque For a `review` gate, the runtime sends a separate role-scoped request with no writable roots or worker write/process grants. Claude Code receives only read/search tools, Codex uses its read-only sandbox, and direct Pi receives only its read tool. OpenCode and any ambient operating-system access remain cooperative. The adapter extracts exactly one structured review marker; missing, contradictory, malformed, truncated, timed-out, or inconclusive output is `UNVERIFIED`. The runtime compares the complete tree, HEAD, refs, and Git configuration before and after review and rejects any mutation. Version-pinned disposable exact-tree runs passed with Pi 0.84.4, Codex 0.151.0, and OpenCode 1.18.25. Claude Code 2.1.251 now reaches its API credential path, but the identity-linked key requires an `ANTHROPIC_WORKSPACE_ID` that is not present in the validation environment, so edit and review remain unverified. -For Pi, Autopilot checks the standard Pi package directory for `pi-subagents` 0.53.0 or newer. When present, it loads only that extension and Autopilot's bridge, delegates the item to the resolved `worker` role through the public structured delegation API, and keeps the worker in Autopilot's existing worktree. The read-only exact-tree review role runs directly and is not subjected to the worker-only subagent terminal envelope. Autopilot does not install or update the extension. An older or absent installation uses the direct Pi process and reports that fallback through `doctor` and adapter limitations. +For Pi, the packaged Autopilot extension probes the process-local `pi-subagents` owner before sealing an execution mode. A compatible installed and active owner receives one public structured delegation request only after `ATTEMPT_STARTED` is durable. Autopilot binds the exact `requestId`, run-scoped `ownerRunId`, attempt-and-context-derived `nodeId`, subject digest, and extension-instance identity in `ATTEMPT_EXECUTION_ADMITTED`. It accepts at most one matching terminal response. Reload, session replacement, stale context, missing admission, or missing terminal response becomes `EXECUTION_STATE_UNKNOWN`; cancellation is only a request until the same tuple returns terminal cancellation. -Delegated Pi activity is written to stderr while the CLI runs, leaving JSON stdout machine-readable. The stream reports worker start, current tool, tool and token counts, elapsed time, and terminal status without copying child output or tool arguments. This is an Autopilot CLI projection, not the originating Pi session's FleetView: Pi's event bus and FleetView are process-local, while Autopilot owns a separate subprocess. The integration follows pi-subagents' public [structured delegation API](https://github.com/nicobailon/pi-subagents/blob/main/docs/extension-api.md) and [observability contract](https://github.com/nicobailon/pi-subagents/blob/main/docs/observability.md), rather than importing internal runners or scraping rendered terminal output. +The in-process worker stays in Autopilot's dedicated worktree and appears through Pi's ordinary foreground subagent observability. Bounded activity is also written to stderr without copying child output or tool arguments. Independent review remains a separate direct read-only Pi execution. An absent, inactive, or old `pi-subagents` installation selects the distinct direct CLI fallback before admission; it cannot inherit the in-process backend's harness-owned same-instance assurance. Direct POSIX execution retains its proven process supervisor, while direct Windows execution remains session-scoped. Autopilot never installs, activates, or updates either extension. + +The process-local backend follows pi-subagents' public [structured delegation API](https://github.com/nicobailon/pi-subagents/blob/main/docs/extension-api.md) and [observability contract](https://github.com/nicobailon/pi-subagents/blob/main/docs/observability.md), rather than importing internal runners or scraping rendered terminal output. Cooperative restrictions do not form an OS sandbox: the process-local worker inherits the owning Pi process environment, and completion does not prove descendant quiescence or rollback external effects. Every harness needs adapter `network.access` and `credentials.use` grants because its model control plane may use authenticated network access. Those grants do not authorize the worker or delivery provider. @@ -42,7 +44,7 @@ Unit tests use controlled fake CLIs for command construction, changed-head denia - Queue execution becomes serial when an adapter reports concurrency one. - Missing required assurance or grants stops before edits. -- On POSIX hosts, supervised implementation processes reattach after coordinator loss. The reviewed Windows Job Object helper will not be packaged, so Windows restart reattachment remains disabled. The planned Pi cooperative backend will accept terminality only through the uninterrupted owning harness instance; harness loss will remain `EXECUTION_STATE_UNKNOWN`. Legacy attempts, reviews, and incomplete supervisor bootstraps fail closed. +- Pi in-process implementations accept terminality only through the uninterrupted owning extension instance. Harness loss remains `EXECUTION_STATE_UNKNOWN`; a fresh Pi instance never repeats admission. Direct CLI implementations retain supervised POSIX behavior, while direct Windows, legacy attempts, reviews, and incomplete supervisor bootstraps fail closed at their declared boundaries. - Late results from expired leases are quarantined. - Provider head changes block merge. - Review findings block the current attempt and enter the next deterministic attempt context as untrusted data. diff --git a/skills/autopilot/references/recovery.md b/skills/autopilot/references/recovery.md index cd705a7..c29abc6 100644 --- a/skills/autopilot/references/recovery.md +++ b/skills/autopilot/references/recovery.md @@ -34,7 +34,7 @@ node runtime/dist/src/cli.js --state-dir resume Resume acquires the coordinator lock, validates the sealed charter and journal, rebuilds projection state, verifies context artifacts, inspects existing worktrees and refs, and continues within the original limits. A paused unfinished item receives a fresh lease and newly hashed context; its pause-cancelled physical launch is not charged to the attempt budget. An item with a durable `ITEM_VERIFIED` checkpoint continues commit, push, change-request, check, thread, or merge reconciliation from fresh exact observations without launching another worker. Resume does not re-open `SUCCEEDED` or `STOPPED` runs. -On POSIX hosts, built-in harness adapters supervise implementation executions with a detached, attempt-scoped helper and a pre-established process-group watchdog. After coordinator loss, `resume` reconstructs the exact request from the journaled attempt and immutable context, reattaches to matching running or terminal artifacts, and observes process-group quiescence before permitting a replacement attempt. The reviewed Windows x64 Job Object helper will not be packaged, so Windows restart reattachment remains disabled while the cooperative harness backend is implemented. Current Windows execution and cancellation are session-scoped. Under the approved cooperative design, only an exact terminal response through the uninterrupted owning harness instance may proceed; harness, session, or exact-subject loss becomes `EXECUTION_STATE_UNKNOWN` and cannot launch a speculative replacement. Legacy attempts, review executions, mismatched requests, and incomplete supervisor artifacts also remain unknown. +On POSIX hosts, built-in harness adapters supervise implementation executions with a detached, attempt-scoped helper and a pre-established process-group watchdog. After coordinator loss, `resume` reconstructs the exact request from the journaled attempt and immutable context, reattaches to matching running or terminal artifacts, and observes process-group quiescence before permitting a replacement attempt. The reviewed Windows x64 Job Object helper will not be packaged. Pi process-local implementations now use cooperative terminality: only an exact terminal response through the uninterrupted owning extension instance may proceed, while harness, session, or exact-subject loss becomes `EXECUTION_STATE_UNKNOWN` and cannot launch a speculative replacement. The distinct direct Pi fallback retains process-supervised POSIX execution and session-scoped Windows execution. Other current Windows CLI execution and cancellation remain session-scoped. Legacy attempts, review executions, mismatched requests, and incomplete supervisor artifacts also remain unknown. ## Recover an unknown execution diff --git a/skills/autopilot/runtime/adapters/pi/bridge.ts b/skills/autopilot/runtime/adapters/pi/bridge.ts deleted file mode 100644 index 7b22ef7..0000000 --- a/skills/autopilot/runtime/adapters/pi/bridge.ts +++ /dev/null @@ -1,195 +0,0 @@ -import { randomUUID } from "node:crypto"; - -const STARTED_EVENT = "prompt-template:subagent:started"; -const UPDATE_EVENT = "prompt-template:subagent:update"; -const RESPONSE_EVENT = "prompt-template:subagent:response"; -const CANCEL_EVENT = "prompt-template:subagent:cancel"; -const REQUEST_EVENT = "prompt-template:subagent:request"; - -interface EventIdentity { - readonly requestId?: string; - readonly ownerRunId?: string; - readonly nodeId?: string; -} - -interface DelegationUpdate extends EventIdentity { - readonly currentTool?: string; - readonly toolCount?: number; - readonly tokens?: number; - readonly durationMs?: number; -} - -interface DelegationResponse extends EventIdentity { - readonly status?: string; - readonly error?: string; - readonly runId?: string; - readonly model?: string; - readonly result?: { readonly kind?: string; readonly text?: string }; -} - -interface BridgePayload { - readonly runId: string; - readonly itemId: string; - readonly task: string; - readonly timeoutMs: number; -} - -interface PiExtensionApi { - readonly events: { - on(event: string, handler: (value: unknown) => void): () => void; - emit(event: string, value: unknown): void; - }; - registerCommand(name: string, options: { - readonly description: string; - readonly handler: (arguments_: string, context: { readonly cwd: string }) => Promise; - }): void; - sendMessage(message: { - readonly customType: string; - readonly content: string; - readonly display: boolean; - readonly details: Readonly>; - }): void; -} - -function isRecord(value: unknown): value is Record { - return typeof value === "object" && value !== null && !Array.isArray(value); -} - -function matches(value: unknown, requestId: string, ownerRunId: string, nodeId: string): value is Record { - return isRecord(value) && value.requestId === requestId && value.ownerRunId === ownerRunId && value.nodeId === nodeId; -} - -function decodePayload(value: string): BridgePayload { - const parsed: unknown = JSON.parse(Buffer.from(value.trim(), "base64url").toString("utf8")); - if (typeof parsed !== "object" || parsed === null) { - throw new Error("Autopilot Pi bridge payload must be an object"); - } - const payload = parsed as Record; - if (typeof payload.runId !== "string" || typeof payload.itemId !== "string" || typeof payload.task !== "string" - || typeof payload.timeoutMs !== "number" || !Number.isSafeInteger(payload.timeoutMs) || payload.timeoutMs <= 0) { - throw new Error("Autopilot Pi bridge payload is invalid"); - } - return { - runId: payload.runId, - itemId: payload.itemId, - task: payload.task, - timeoutMs: payload.timeoutMs, - }; -} - -function activity(update: DelegationUpdate): string { - return [ - update.currentTool, - update.toolCount === undefined ? undefined : `${update.toolCount} tools`, - update.tokens === undefined ? undefined : `${update.tokens} tokens`, - ].filter((field) => field !== undefined).join(" · "); -} - -async function delegate(pi: PiExtensionApi, context: { readonly cwd: string }, payload: BridgePayload): Promise { - const requestId = randomUUID(); - const ownerRunId = payload.runId; - const nodeId = payload.itemId; - let lastActivity = ""; - let lastActivityAt = 0; - return await new Promise((resolvePromise, rejectPromise) => { - const cleanups: Array<() => void> = []; - const cleanup = (): void => cleanups.splice(0).forEach((dispose) => dispose()); - cleanups.push(pi.events.on(STARTED_EVENT, (value) => { - if (matches(value, requestId, ownerRunId, nodeId)) { - console.error(`[autopilot] Pi subagent worker started · ${nodeId}`); - } - })); - cleanups.push(pi.events.on(UPDATE_EVENT, (value) => { - if (!matches(value, requestId, ownerRunId, nodeId)) { - return; - } - const event: DelegationUpdate = { - requestId, - ownerRunId, - nodeId, - ...(typeof value.currentTool === "string" ? { currentTool: value.currentTool } : {}), - ...(typeof value.toolCount === "number" ? { toolCount: value.toolCount } : {}), - ...(typeof value.tokens === "number" ? { tokens: value.tokens } : {}), - ...(typeof value.durationMs === "number" ? { durationMs: value.durationMs } : {}), - }; - const currentActivity = activity(event); - const now = Date.now(); - if (currentActivity !== "" && (currentActivity !== lastActivity || now - lastActivityAt >= 5_000)) { - lastActivity = currentActivity; - lastActivityAt = now; - const elapsed = event.durationMs === undefined ? "" : ` · ${Math.round(event.durationMs / 1_000)}s`; - console.error(`[autopilot] Pi subagent worker · ${currentActivity}${elapsed}`); - } - })); - cleanups.push(pi.events.on(RESPONSE_EVENT, (value) => { - if (!isRecord(value) || value.requestId !== requestId - || (value.ownerRunId !== undefined && value.ownerRunId !== ownerRunId) - || (value.nodeId !== undefined && value.nodeId !== nodeId)) { - return; - } - const result = isRecord(value.result) && (value.result.kind === "text" || value.result.kind === "structured") - ? { kind: value.result.kind, ...(typeof value.result.text === "string" ? { text: value.result.text } : {}) } - : undefined; - const event: DelegationResponse = { - requestId, - ...(typeof value.ownerRunId === "string" ? { ownerRunId: value.ownerRunId } : {}), - ...(typeof value.nodeId === "string" ? { nodeId: value.nodeId } : {}), - ...(typeof value.status === "string" ? { status: value.status } : {}), - ...(typeof value.error === "string" ? { error: value.error } : {}), - ...(typeof value.runId === "string" ? { runId: value.runId } : {}), - ...(typeof value.model === "string" ? { model: value.model } : {}), - ...(result === undefined ? {} : { result }), - }; - cleanup(); - resolvePromise(event); - })); - const timer = setTimeout(() => { - pi.events.emit(CANCEL_EVENT, { requestId, ownerRunId, nodeId }); - cleanup(); - rejectPromise(new Error("pi-subagents delegation exceeded the Autopilot attempt deadline")); - }, payload.timeoutMs); - timer.unref(); - cleanups.push(() => clearTimeout(timer)); - pi.events.emit(REQUEST_EVENT, { - requestId, - ownerRunId, - nodeId, - agent: "worker", - task: payload.task, - context: "fresh", - cwd: context.cwd, - timeoutMs: payload.timeoutMs, - artifacts: true, - result: { kind: "text" }, - }); - }); -} - -export default function registerAutopilotPiBridge(pi: PiExtensionApi): void { - pi.registerCommand("autopilot-worker", { - description: "Run one bounded Autopilot work item through pi-subagents", - handler: async (arguments_, context) => { - let response: DelegationResponse; - try { - const payload = decodePayload(arguments_); - response = await delegate(pi, context, payload); - console.error(`[autopilot] Pi subagent worker ${response.status ?? "failed"} · ${payload.itemId}`); - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - console.error(`[autopilot] Pi subagent worker failed · ${message}`); - response = { status: "bridge_failed", error: message }; - } - const completed = response.status === "completed" && response.result?.kind === "text"; - pi.sendMessage({ - customType: "autopilot-subagent-result", - content: completed ? response.result?.text ?? "" : response.error ?? `pi-subagents ended with ${response.status ?? "unknown"}`, - display: true, - details: { - status: response.status ?? "unknown", - ...(response.runId === undefined ? {} : { runId: response.runId }), - ...(response.model === undefined ? {} : { model: response.model }), - }, - }); - }, - }); -} diff --git a/skills/autopilot/runtime/adapters/pi/in-process.ts b/skills/autopilot/runtime/adapters/pi/in-process.ts new file mode 100644 index 0000000..53b34f6 --- /dev/null +++ b/skills/autopilot/runtime/adapters/pi/in-process.ts @@ -0,0 +1,392 @@ +import { randomUUID } from "node:crypto"; +import { + type CancelResult, + type CapabilityManifest, + type ExecutionHandle, + type ExecutionObservation, + type ExecutionRequest, + type HarnessPort, +} from "../../src/adapter-protocol.js"; +import { renderAttemptContext } from "../../src/attempt-context.js"; +import { AutopilotError } from "../../src/errors.js"; +import { canonicalJson, isRecord, sha256 } from "../../src/json.js"; +import { boundUtf8 } from "../../src/process.js"; + +export const PI_SUBAGENT_REQUEST_EVENT = "prompt-template:subagent:request"; +export const PI_SUBAGENT_STARTED_EVENT = "prompt-template:subagent:started"; +export const PI_SUBAGENT_UPDATE_EVENT = "prompt-template:subagent:update"; +export const PI_SUBAGENT_RESPONSE_EVENT = "prompt-template:subagent:response"; +export const PI_SUBAGENT_CANCEL_EVENT = "prompt-template:subagent:cancel"; + +export interface PiEventBus { + on(event: string, handler: (value: unknown) => void): () => void; + emit(event: string, value: unknown): void; +} + +export interface PiInProcessAdapterOptions { + readonly events: PiEventBus; + readonly harnessInstanceId: string; + readonly harnessVersion: string; + readonly piSubagentsVersion: string; + readonly reviewAdapter: HarnessPort; + readonly onActivity?: (message: string) => void; +} + +interface DelegationIdentity { + readonly requestId: string; + readonly ownerRunId: string; + readonly nodeId: string; + readonly subjectId: string; +} + +interface PendingExecution { + readonly request: ExecutionRequest; + readonly identity: DelegationIdentity; + readonly startedAt: string; + readonly terminal: Promise; + readonly rejectStarted: (error: Error) => void; + readonly rejectTerminal: (error: Error) => void; + readonly isTerminalAccepted: () => boolean; + readonly disposeListeners: () => void; +} + +function deferred(): { + readonly promise: Promise; + readonly resolve: (value: T) => void; + readonly reject: (error: Error) => void; +} { + let resolvePromise: (value: T) => void = () => undefined; + let rejectPromise: (error: Error) => void = () => undefined; + const promise = new Promise((resolve, reject) => { + resolvePromise = resolve; + rejectPromise = reject; + }); + return { promise, resolve: resolvePromise, reject: rejectPromise }; +} + +function identityFor(request: ExecutionRequest): DelegationIdentity { + const requestId = randomUUID(); + const ownerRunId = request.runId; + const nodeId = `autopilot-${sha256(canonicalJson({ + runId: request.runId, + itemId: request.itemId, + attemptId: request.attemptId, + leaseEpoch: request.context.leaseEpoch, + contextHash: request.contextHash, + })).slice(0, 40)}`; + return { + requestId, + ownerRunId, + nodeId, + subjectId: sha256(canonicalJson({ requestId, ownerRunId, nodeId })), + }; +} + +function delegationTuple(identity: DelegationIdentity): Pick { + return { requestId: identity.requestId, ownerRunId: identity.ownerRunId, nodeId: identity.nodeId }; +} + +function exactIdentity(value: unknown, identity: DelegationIdentity): value is Record { + return isRecord(value) && value.requestId === identity.requestId + && value.ownerRunId === identity.ownerRunId && value.nodeId === identity.nodeId; +} + +function credentialEnvironmentNames(request: ExecutionRequest): ReadonlySet { + return new Set(request.grants + .filter(({ actor, family }) => actor === "adapter" && family === "credentials.use") + .flatMap(({ environmentNames }) => environmentNames ?? [])); +} + +function redactSecrets(text: string, request: ExecutionRequest): string { + const grantedNames = credentialEnvironmentNames(request); + return Object.entries(process.env).reduce((redacted, [name, value]) => { + const granted = grantedNames.has(name); + return value === undefined || value === "" + || (!granted && !/(TOKEN|KEY|SECRET|PASSWORD|COOKIE|AUTH)/iu.test(name)) + ? redacted + : redacted.replaceAll(value, "****"); + }, text); +} + +function terminalObservation( + request: ExecutionRequest, + identity: DelegationIdentity, + value: Record, +): ExecutionObservation { + const status = value.status; + const result = isRecord(value.result) ? value.result : undefined; + const text = result?.kind === "text" && typeof result.text === "string" ? result.text : undefined; + const error = typeof value.error === "string" ? value.error : ""; + const boundedText = boundUtf8(redactSecrets(text ?? "", request), request.maximumOutputBytes); + const boundedError = boundUtf8(redactSecrets(error, request), request.maximumOutputBytes); + const completed = status === "completed" && text !== undefined && !boundedText.truncated; + const observationStatus: ExecutionObservation["status"] = completed + ? "completed" + : status === "cancelled" || status === "interrupted" ? "cancelled" + : status === "timed_out" ? "timed-out" : "failed"; + const malformed = status === "completed" && text === undefined + ? "Pi structured delegation completed without the required text result" + : status === "completed" && boundedText.truncated + ? "Pi structured delegation result exceeded the Autopilot output bound" + : undefined; + return { + protocolVersion: 1, + adapterExecutionId: identity.requestId, + status: observationStatus, + exitCode: observationStatus === "completed" ? 0 : observationStatus === "timed-out" ? 124 + : observationStatus === "cancelled" ? 130 : 1, + completedAt: new Date().toISOString(), + stdout: completed ? boundedText.value : "", + stderr: malformed ?? (boundedError.value || "Pi structured delegation ended without a valid completion result"), + truncated: boundedText.truncated || boundedError.truncated, + }; +} + +export class PiInProcessAdapter implements HarnessPort { + readonly #options: PiInProcessAdapterOptions; + readonly #pending = new Map(); + readonly #reviewHandles = new Set(); + #active = true; + + constructor(options: PiInProcessAdapterOptions) { + this.#options = options; + } + + async describe(): Promise { + if (!this.#active) { + throw new AutopilotError("ADAPTER_UNSUPPORTED", "the owning Pi extension context is no longer active"); + } + return { + protocolVersion: 1, + adapterName: "pi", + adapterVersion: "2", + harnessVersion: this.#options.harnessVersion, + families: ["files.read", "files.write", "process.execute", "network.access", "credentials.use"], + assurance: "cooperative", + unattended: true, + maxConcurrency: 1, + eventStreaming: true, + cancellation: true, + restartReattachment: false, + executionAssurance: { + schemaVersion: 1, + implementation: { + schemaVersion: 1, + owner: "harness", + continuity: "same-harness-instance", + terminality: "cooperative", + admission: "single-shot", + }, + review: { + schemaVersion: 1, + owner: "runtime", + continuity: "session", + terminality: "cooperative", + admission: "single-shot", + }, + }, + restrictions: "cooperative", + limitations: [ + `Pi implementation workers use pi-subagents ${this.#options.piSubagentsVersion} process-local structured delegation.`, + "Completion proves an exact logical terminal response from the uninterrupted Pi extension context, not OS process-tree quiescence.", + "The process-local worker inherits the owning Pi process environment; grant restrictions remain cooperative.", + "Independent review uses the direct session-scoped Pi adapter.", + ], + }; + } + + async launch(request: ExecutionRequest): Promise { + if (request.role === "review") { + const handle = await this.#options.reviewAdapter.launch(request); + this.#reviewHandles.add(handle.adapterExecutionId); + return handle; + } + if (!this.#active) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "the owning Pi extension context is no longer active"); + } + const identity = identityFor(request); + const started = deferred(); + const terminal = deferred(); + void terminal.promise.catch(() => undefined); + let startedAccepted = false; + let terminalAccepted = false; + const cleanups: Array<() => void> = []; + const disposeListeners = (): void => cleanups.splice(0).forEach((dispose) => dispose()); + const fail = (message: string): void => { + if (terminalAccepted) { + return; + } + terminalAccepted = true; + const error = new AutopilotError("EXECUTION_STATE_UNKNOWN", message); + if (!startedAccepted) { + started.reject(error); + this.#pending.delete(identity.requestId); + } + terminal.reject(error); + disposeListeners(); + }; + cleanups.push(this.#options.events.on(PI_SUBAGENT_STARTED_EVENT, (value) => { + if (!exactIdentity(value, identity) || startedAccepted || terminalAccepted) { + return; + } + startedAccepted = true; + resetIdleTimer(); + started.resolve(); + this.#options.onActivity?.(`Pi worker started · ${request.itemId}`); + })); + cleanups.push(this.#options.events.on(PI_SUBAGENT_UPDATE_EVENT, (value) => { + if (!exactIdentity(value, identity) || terminalAccepted) { + return; + } + const currentTool = typeof value.currentTool === "string" + ? boundUtf8(redactSecrets(value.currentTool, request), 128).value : undefined; + const fields = [ + currentTool, + typeof value.toolCount === "number" && Number.isFinite(value.toolCount) && value.toolCount >= 0 + ? `${value.toolCount} tools` : undefined, + typeof value.tokens === "number" && Number.isFinite(value.tokens) && value.tokens >= 0 + ? `${value.tokens} tokens` : undefined, + typeof value.durationMs === "number" && Number.isFinite(value.durationMs) && value.durationMs >= 0 + ? `${Math.round(value.durationMs / 1_000)}s` : undefined, + ].filter((field) => field !== undefined); + resetIdleTimer(); + if (fields.length > 0) { + this.#options.onActivity?.(`Pi worker · ${fields.join(" · ")}`); + } + })); + cleanups.push(this.#options.events.on(PI_SUBAGENT_RESPONSE_EVENT, (value) => { + if (!exactIdentity(value, identity) || terminalAccepted) { + return; + } + if (!startedAccepted) { + fail("Pi structured delegation returned terminal state before exact admission was observed"); + return; + } + terminalAccepted = true; + terminal.resolve(terminalObservation(request, identity, value)); + disposeListeners(); + })); + const timeoutMs = Math.max(1, Date.parse(request.deadline) - Date.now()); + const cancelForUnknownState = (message: string): void => { + if (!terminalAccepted && this.#active) { + try { + this.#options.events.emit(PI_SUBAGENT_CANCEL_EVENT, delegationTuple(identity)); + } catch { + fail(`${message}; cancellation delivery also failed`); + return; + } + } + fail(message); + }; + let idleTimer: NodeJS.Timeout | undefined; + const resetIdleTimer = (): void => { + if (idleTimer !== undefined) { + clearTimeout(idleTimer); + } + idleTimer = setTimeout(() => { + cancelForUnknownState("Pi structured delegation exceeded the harness idle timeout without an exact terminal response"); + }, request.idleTimeoutMs); + idleTimer.unref(); + }; + resetIdleTimer(); + cleanups.push(() => { + if (idleTimer !== undefined) { + clearTimeout(idleTimer); + } + }); + const deadlineTimer = setTimeout(() => { + cancelForUnknownState("Pi structured delegation did not return an exact terminal response before the attempt deadline"); + }, timeoutMs); + deadlineTimer.unref(); + cleanups.push(() => clearTimeout(deadlineTimer)); + const pending: PendingExecution = { + request, + identity, + startedAt: new Date().toISOString(), + terminal: terminal.promise, + rejectStarted: started.reject, + rejectTerminal: terminal.reject, + isTerminalAccepted: () => terminalAccepted, + disposeListeners, + }; + this.#pending.set(identity.requestId, pending); + try { + this.#options.events.emit(PI_SUBAGENT_REQUEST_EVENT, { + requestId: identity.requestId, + ownerRunId: identity.ownerRunId, + nodeId: identity.nodeId, + agent: "worker", + task: renderAttemptContext(request.context), + context: "fresh", + cwd: request.worktreePath, + timeoutMs, + artifacts: true, + result: { kind: "text" }, + }); + } catch (error) { + fail(`Pi structured delegation request failed: ${error instanceof Error ? error.message : String(error)}`); + } + await started.promise; + return { + protocolVersion: 1, + adapterExecutionId: identity.requestId, + startedAt: pending.startedAt, + subject: { + schemaVersion: 1, + backendId: `pi-subagents-structured-v1@${this.#options.piSubagentsVersion}`, + subjectId: identity.subjectId, + harnessInstanceId: this.#options.harnessInstanceId, + }, + }; + } + + async observe(handle: ExecutionHandle): Promise { + if (this.#reviewHandles.delete(handle.adapterExecutionId)) { + return await this.#options.reviewAdapter.observe(handle); + } + const pending = this.#pending.get(handle.adapterExecutionId); + if (pending === undefined || handle.subject?.subjectId !== pending.identity.subjectId + || handle.subject.harnessInstanceId !== this.#options.harnessInstanceId) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "Pi execution is not attached to the exact owning extension instance"); + } + try { + return await pending.terminal; + } finally { + this.#pending.delete(handle.adapterExecutionId); + } + } + + async cancel(handle: ExecutionHandle): Promise { + if (this.#reviewHandles.has(handle.adapterExecutionId)) { + return await this.#options.reviewAdapter.cancel(handle); + } + const pending = this.#pending.get(handle.adapterExecutionId); + if (!this.#active || pending === undefined || handle.subject?.subjectId !== pending.identity.subjectId + || handle.subject.harnessInstanceId !== this.#options.harnessInstanceId) { + return { protocolVersion: 1, accepted: false }; + } + try { + this.#options.events.emit(PI_SUBAGENT_CANCEL_EVENT, delegationTuple(pending.identity)); + return { protocolVersion: 1, accepted: true }; + } catch { + return { protocolVersion: 1, accepted: false }; + } + } + + invalidate(reason: string): void { + if (!this.#active) { + return; + } + this.#active = false; + for (const pending of this.#pending.values()) { + if (pending.isTerminalAccepted()) { + continue; + } + const error = new AutopilotError("EXECUTION_STATE_UNKNOWN", reason); + pending.rejectStarted(error); + pending.rejectTerminal(error); + pending.disposeListeners(); + this.#pending.delete(pending.identity.requestId); + } + } +} diff --git a/skills/autopilot/runtime/adapters/pi/index.ts b/skills/autopilot/runtime/adapters/pi/index.ts index ffaa776..c88f7d6 100644 --- a/skills/autopilot/runtime/adapters/pi/index.ts +++ b/skills/autopilot/runtime/adapters/pi/index.ts @@ -1,15 +1,5 @@ -import { fileURLToPath } from "node:url"; import type { ExecutionRequest } from "../../src/adapter-protocol.js"; import { CliHarnessAdapter } from "../../src/adapter-process.js"; -import { isRecord } from "../../src/json.js"; -import { findPiSubagentsInstallation, type PiSubagentsInstallation } from "../../src/pi-subagents.js"; - -interface PiSubagentBridgePayload { - readonly runId: string; - readonly itemId: string; - readonly task: string; - readonly timeoutMs: number; -} function directArguments(request: ExecutionRequest, prompt: string): readonly string[] { return [ @@ -26,81 +16,21 @@ function directArguments(request: ExecutionRequest, prompt: string): readonly st ]; } -function subagentArguments( - request: ExecutionRequest, - prompt: string, - installation: PiSubagentsInstallation, -): readonly string[] { - const payload: PiSubagentBridgePayload = { - runId: request.runId, - itemId: request.itemId, - task: prompt, - timeoutMs: Math.min(2_147_483_647, Math.max(1, Date.parse(request.deadline) - Date.now())), - }; - const encoded = Buffer.from(JSON.stringify(payload), "utf8").toString("base64url"); - const bridgePath = fileURLToPath(new URL("./bridge.js", import.meta.url)); - return [ - "--mode", "json", - "--print", - "--no-session", - "--no-extensions", - "--extension", installation.extensionPath, - "--extension", bridgePath, - "--no-skills", - "--no-prompt-templates", - "--no-context-files", - "--no-tools", - "--approve", - `/autopilot-worker ${encoded}`, - ]; -} - -function validateSubagentResult(stdout: string): string | undefined { - for (const line of stdout.split("\n")) { - if (line === "") { - continue; - } - let value: unknown; - try { - value = JSON.parse(line) as unknown; - } catch { - continue; - } - if (!isRecord(value) || value.type !== "message_end" || !isRecord(value.message) - || value.message.customType !== "autopilot-subagent-result" || !isRecord(value.message.details)) { - continue; - } - const status = value.message.details.status; - return status === "completed" ? undefined : `pi-subagents worker ended with ${typeof status === "string" ? status : "unknown status"}`; - } - return "pi-subagents worker did not emit a terminal result"; -} - export function createPiAdapter(): CliHarnessAdapter { - const installation = findPiSubagentsInstallation(); - const usingSubagents = installation !== undefined; return new CliHarnessAdapter({ name: "pi", executable: "pi", versionArguments: ["--version"], - buildArguments: (request, prompt) => installation === undefined || request.role === "review" - ? directArguments(request, prompt) - : subagentArguments(request, prompt, installation), + buildArguments: directArguments, assurance: "cooperative", maxConcurrency: 1, cancellation: true, limitations: [ + "Pi runs through the direct CLI fallback because no owning process-local Autopilot extension backend was selected.", "Tool restrictions do not constrain commands executed through bash.", - "Implementation executions use the attempt-scoped supervisor for restart reattachment; review executions remain session-scoped.", - usingSubagents - ? `Pi workers use the pi-subagents ${installation.version} structured delegation API.` - : "pi-subagents 0.53.0 or newer was not found; Pi workers run directly without subagent activity integration.", + "POSIX implementation executions retain process-supervised terminality; Windows direct execution remains session-scoped.", ], expectsJsonLines: true, - ...(usingSubagents ? { - validateResult: (stdout: string, request: ExecutionRequest): string | undefined => - request.role === "review" ? undefined : validateSubagentResult(stdout), - } : {}), displayStderrActivity: true, }); } diff --git a/skills/autopilot/runtime/dist/adapters/pi/bridge.d.ts b/skills/autopilot/runtime/dist/adapters/pi/bridge.d.ts deleted file mode 100644 index f146d2b..0000000 --- a/skills/autopilot/runtime/dist/adapters/pi/bridge.d.ts +++ /dev/null @@ -1,20 +0,0 @@ -interface PiExtensionApi { - readonly events: { - on(event: string, handler: (value: unknown) => void): () => void; - emit(event: string, value: unknown): void; - }; - registerCommand(name: string, options: { - readonly description: string; - readonly handler: (arguments_: string, context: { - readonly cwd: string; - }) => Promise; - }): void; - sendMessage(message: { - readonly customType: string; - readonly content: string; - readonly display: boolean; - readonly details: Readonly>; - }): void; -} -export default function registerAutopilotPiBridge(pi: PiExtensionApi): void; -export {}; diff --git a/skills/autopilot/runtime/dist/adapters/pi/bridge.js b/skills/autopilot/runtime/dist/adapters/pi/bridge.js deleted file mode 100644 index d4dea36..0000000 --- a/skills/autopilot/runtime/dist/adapters/pi/bridge.js +++ /dev/null @@ -1,144 +0,0 @@ -import { randomUUID } from "node:crypto"; -const STARTED_EVENT = "prompt-template:subagent:started"; -const UPDATE_EVENT = "prompt-template:subagent:update"; -const RESPONSE_EVENT = "prompt-template:subagent:response"; -const CANCEL_EVENT = "prompt-template:subagent:cancel"; -const REQUEST_EVENT = "prompt-template:subagent:request"; -function isRecord(value) { - return typeof value === "object" && value !== null && !Array.isArray(value); -} -function matches(value, requestId, ownerRunId, nodeId) { - return isRecord(value) && value.requestId === requestId && value.ownerRunId === ownerRunId && value.nodeId === nodeId; -} -function decodePayload(value) { - const parsed = JSON.parse(Buffer.from(value.trim(), "base64url").toString("utf8")); - if (typeof parsed !== "object" || parsed === null) { - throw new Error("Autopilot Pi bridge payload must be an object"); - } - const payload = parsed; - if (typeof payload.runId !== "string" || typeof payload.itemId !== "string" || typeof payload.task !== "string" - || typeof payload.timeoutMs !== "number" || !Number.isSafeInteger(payload.timeoutMs) || payload.timeoutMs <= 0) { - throw new Error("Autopilot Pi bridge payload is invalid"); - } - return { - runId: payload.runId, - itemId: payload.itemId, - task: payload.task, - timeoutMs: payload.timeoutMs, - }; -} -function activity(update) { - return [ - update.currentTool, - update.toolCount === undefined ? undefined : `${update.toolCount} tools`, - update.tokens === undefined ? undefined : `${update.tokens} tokens`, - ].filter((field) => field !== undefined).join(" · "); -} -async function delegate(pi, context, payload) { - const requestId = randomUUID(); - const ownerRunId = payload.runId; - const nodeId = payload.itemId; - let lastActivity = ""; - let lastActivityAt = 0; - return await new Promise((resolvePromise, rejectPromise) => { - const cleanups = []; - const cleanup = () => cleanups.splice(0).forEach((dispose) => dispose()); - cleanups.push(pi.events.on(STARTED_EVENT, (value) => { - if (matches(value, requestId, ownerRunId, nodeId)) { - console.error(`[autopilot] Pi subagent worker started · ${nodeId}`); - } - })); - cleanups.push(pi.events.on(UPDATE_EVENT, (value) => { - if (!matches(value, requestId, ownerRunId, nodeId)) { - return; - } - const event = { - requestId, - ownerRunId, - nodeId, - ...(typeof value.currentTool === "string" ? { currentTool: value.currentTool } : {}), - ...(typeof value.toolCount === "number" ? { toolCount: value.toolCount } : {}), - ...(typeof value.tokens === "number" ? { tokens: value.tokens } : {}), - ...(typeof value.durationMs === "number" ? { durationMs: value.durationMs } : {}), - }; - const currentActivity = activity(event); - const now = Date.now(); - if (currentActivity !== "" && (currentActivity !== lastActivity || now - lastActivityAt >= 5_000)) { - lastActivity = currentActivity; - lastActivityAt = now; - const elapsed = event.durationMs === undefined ? "" : ` · ${Math.round(event.durationMs / 1_000)}s`; - console.error(`[autopilot] Pi subagent worker · ${currentActivity}${elapsed}`); - } - })); - cleanups.push(pi.events.on(RESPONSE_EVENT, (value) => { - if (!isRecord(value) || value.requestId !== requestId - || (value.ownerRunId !== undefined && value.ownerRunId !== ownerRunId) - || (value.nodeId !== undefined && value.nodeId !== nodeId)) { - return; - } - const result = isRecord(value.result) && (value.result.kind === "text" || value.result.kind === "structured") - ? { kind: value.result.kind, ...(typeof value.result.text === "string" ? { text: value.result.text } : {}) } - : undefined; - const event = { - requestId, - ...(typeof value.ownerRunId === "string" ? { ownerRunId: value.ownerRunId } : {}), - ...(typeof value.nodeId === "string" ? { nodeId: value.nodeId } : {}), - ...(typeof value.status === "string" ? { status: value.status } : {}), - ...(typeof value.error === "string" ? { error: value.error } : {}), - ...(typeof value.runId === "string" ? { runId: value.runId } : {}), - ...(typeof value.model === "string" ? { model: value.model } : {}), - ...(result === undefined ? {} : { result }), - }; - cleanup(); - resolvePromise(event); - })); - const timer = setTimeout(() => { - pi.events.emit(CANCEL_EVENT, { requestId, ownerRunId, nodeId }); - cleanup(); - rejectPromise(new Error("pi-subagents delegation exceeded the Autopilot attempt deadline")); - }, payload.timeoutMs); - timer.unref(); - cleanups.push(() => clearTimeout(timer)); - pi.events.emit(REQUEST_EVENT, { - requestId, - ownerRunId, - nodeId, - agent: "worker", - task: payload.task, - context: "fresh", - cwd: context.cwd, - timeoutMs: payload.timeoutMs, - artifacts: true, - result: { kind: "text" }, - }); - }); -} -export default function registerAutopilotPiBridge(pi) { - pi.registerCommand("autopilot-worker", { - description: "Run one bounded Autopilot work item through pi-subagents", - handler: async (arguments_, context) => { - let response; - try { - const payload = decodePayload(arguments_); - response = await delegate(pi, context, payload); - console.error(`[autopilot] Pi subagent worker ${response.status ?? "failed"} · ${payload.itemId}`); - } - catch (error) { - const message = error instanceof Error ? error.message : String(error); - console.error(`[autopilot] Pi subagent worker failed · ${message}`); - response = { status: "bridge_failed", error: message }; - } - const completed = response.status === "completed" && response.result?.kind === "text"; - pi.sendMessage({ - customType: "autopilot-subagent-result", - content: completed ? response.result?.text ?? "" : response.error ?? `pi-subagents ended with ${response.status ?? "unknown"}`, - display: true, - details: { - status: response.status ?? "unknown", - ...(response.runId === undefined ? {} : { runId: response.runId }), - ...(response.model === undefined ? {} : { model: response.model }), - }, - }); - }, - }); -} diff --git a/skills/autopilot/runtime/dist/adapters/pi/in-process.d.ts b/skills/autopilot/runtime/dist/adapters/pi/in-process.d.ts new file mode 100644 index 0000000..9a1d5a3 --- /dev/null +++ b/skills/autopilot/runtime/dist/adapters/pi/in-process.d.ts @@ -0,0 +1,27 @@ +import { type CancelResult, type CapabilityManifest, type ExecutionHandle, type ExecutionObservation, type ExecutionRequest, type HarnessPort } from "../../src/adapter-protocol.js"; +export declare const PI_SUBAGENT_REQUEST_EVENT = "prompt-template:subagent:request"; +export declare const PI_SUBAGENT_STARTED_EVENT = "prompt-template:subagent:started"; +export declare const PI_SUBAGENT_UPDATE_EVENT = "prompt-template:subagent:update"; +export declare const PI_SUBAGENT_RESPONSE_EVENT = "prompt-template:subagent:response"; +export declare const PI_SUBAGENT_CANCEL_EVENT = "prompt-template:subagent:cancel"; +export interface PiEventBus { + on(event: string, handler: (value: unknown) => void): () => void; + emit(event: string, value: unknown): void; +} +export interface PiInProcessAdapterOptions { + readonly events: PiEventBus; + readonly harnessInstanceId: string; + readonly harnessVersion: string; + readonly piSubagentsVersion: string; + readonly reviewAdapter: HarnessPort; + readonly onActivity?: (message: string) => void; +} +export declare class PiInProcessAdapter implements HarnessPort { + #private; + constructor(options: PiInProcessAdapterOptions); + describe(): Promise; + launch(request: ExecutionRequest): Promise; + observe(handle: ExecutionHandle): Promise; + cancel(handle: ExecutionHandle): Promise; + invalidate(reason: string): void; +} diff --git a/skills/autopilot/runtime/dist/adapters/pi/in-process.js b/skills/autopilot/runtime/dist/adapters/pi/in-process.js new file mode 100644 index 0000000..6ce09e8 --- /dev/null +++ b/skills/autopilot/runtime/dist/adapters/pi/in-process.js @@ -0,0 +1,334 @@ +import { randomUUID } from "node:crypto"; +import {} from "../../src/adapter-protocol.js"; +import { renderAttemptContext } from "../../src/attempt-context.js"; +import { AutopilotError } from "../../src/errors.js"; +import { canonicalJson, isRecord, sha256 } from "../../src/json.js"; +import { boundUtf8 } from "../../src/process.js"; +export const PI_SUBAGENT_REQUEST_EVENT = "prompt-template:subagent:request"; +export const PI_SUBAGENT_STARTED_EVENT = "prompt-template:subagent:started"; +export const PI_SUBAGENT_UPDATE_EVENT = "prompt-template:subagent:update"; +export const PI_SUBAGENT_RESPONSE_EVENT = "prompt-template:subagent:response"; +export const PI_SUBAGENT_CANCEL_EVENT = "prompt-template:subagent:cancel"; +function deferred() { + let resolvePromise = () => undefined; + let rejectPromise = () => undefined; + const promise = new Promise((resolve, reject) => { + resolvePromise = resolve; + rejectPromise = reject; + }); + return { promise, resolve: resolvePromise, reject: rejectPromise }; +} +function identityFor(request) { + const requestId = randomUUID(); + const ownerRunId = request.runId; + const nodeId = `autopilot-${sha256(canonicalJson({ + runId: request.runId, + itemId: request.itemId, + attemptId: request.attemptId, + leaseEpoch: request.context.leaseEpoch, + contextHash: request.contextHash, + })).slice(0, 40)}`; + return { + requestId, + ownerRunId, + nodeId, + subjectId: sha256(canonicalJson({ requestId, ownerRunId, nodeId })), + }; +} +function delegationTuple(identity) { + return { requestId: identity.requestId, ownerRunId: identity.ownerRunId, nodeId: identity.nodeId }; +} +function exactIdentity(value, identity) { + return isRecord(value) && value.requestId === identity.requestId + && value.ownerRunId === identity.ownerRunId && value.nodeId === identity.nodeId; +} +function credentialEnvironmentNames(request) { + return new Set(request.grants + .filter(({ actor, family }) => actor === "adapter" && family === "credentials.use") + .flatMap(({ environmentNames }) => environmentNames ?? [])); +} +function redactSecrets(text, request) { + const grantedNames = credentialEnvironmentNames(request); + return Object.entries(process.env).reduce((redacted, [name, value]) => { + const granted = grantedNames.has(name); + return value === undefined || value === "" + || (!granted && !/(TOKEN|KEY|SECRET|PASSWORD|COOKIE|AUTH)/iu.test(name)) + ? redacted + : redacted.replaceAll(value, "****"); + }, text); +} +function terminalObservation(request, identity, value) { + const status = value.status; + const result = isRecord(value.result) ? value.result : undefined; + const text = result?.kind === "text" && typeof result.text === "string" ? result.text : undefined; + const error = typeof value.error === "string" ? value.error : ""; + const boundedText = boundUtf8(redactSecrets(text ?? "", request), request.maximumOutputBytes); + const boundedError = boundUtf8(redactSecrets(error, request), request.maximumOutputBytes); + const completed = status === "completed" && text !== undefined && !boundedText.truncated; + const observationStatus = completed + ? "completed" + : status === "cancelled" || status === "interrupted" ? "cancelled" + : status === "timed_out" ? "timed-out" : "failed"; + const malformed = status === "completed" && text === undefined + ? "Pi structured delegation completed without the required text result" + : status === "completed" && boundedText.truncated + ? "Pi structured delegation result exceeded the Autopilot output bound" + : undefined; + return { + protocolVersion: 1, + adapterExecutionId: identity.requestId, + status: observationStatus, + exitCode: observationStatus === "completed" ? 0 : observationStatus === "timed-out" ? 124 + : observationStatus === "cancelled" ? 130 : 1, + completedAt: new Date().toISOString(), + stdout: completed ? boundedText.value : "", + stderr: malformed ?? (boundedError.value || "Pi structured delegation ended without a valid completion result"), + truncated: boundedText.truncated || boundedError.truncated, + }; +} +export class PiInProcessAdapter { + #options; + #pending = new Map(); + #reviewHandles = new Set(); + #active = true; + constructor(options) { + this.#options = options; + } + async describe() { + if (!this.#active) { + throw new AutopilotError("ADAPTER_UNSUPPORTED", "the owning Pi extension context is no longer active"); + } + return { + protocolVersion: 1, + adapterName: "pi", + adapterVersion: "2", + harnessVersion: this.#options.harnessVersion, + families: ["files.read", "files.write", "process.execute", "network.access", "credentials.use"], + assurance: "cooperative", + unattended: true, + maxConcurrency: 1, + eventStreaming: true, + cancellation: true, + restartReattachment: false, + executionAssurance: { + schemaVersion: 1, + implementation: { + schemaVersion: 1, + owner: "harness", + continuity: "same-harness-instance", + terminality: "cooperative", + admission: "single-shot", + }, + review: { + schemaVersion: 1, + owner: "runtime", + continuity: "session", + terminality: "cooperative", + admission: "single-shot", + }, + }, + restrictions: "cooperative", + limitations: [ + `Pi implementation workers use pi-subagents ${this.#options.piSubagentsVersion} process-local structured delegation.`, + "Completion proves an exact logical terminal response from the uninterrupted Pi extension context, not OS process-tree quiescence.", + "The process-local worker inherits the owning Pi process environment; grant restrictions remain cooperative.", + "Independent review uses the direct session-scoped Pi adapter.", + ], + }; + } + async launch(request) { + if (request.role === "review") { + const handle = await this.#options.reviewAdapter.launch(request); + this.#reviewHandles.add(handle.adapterExecutionId); + return handle; + } + if (!this.#active) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "the owning Pi extension context is no longer active"); + } + const identity = identityFor(request); + const started = deferred(); + const terminal = deferred(); + void terminal.promise.catch(() => undefined); + let startedAccepted = false; + let terminalAccepted = false; + const cleanups = []; + const disposeListeners = () => cleanups.splice(0).forEach((dispose) => dispose()); + const fail = (message) => { + if (terminalAccepted) { + return; + } + terminalAccepted = true; + const error = new AutopilotError("EXECUTION_STATE_UNKNOWN", message); + if (!startedAccepted) { + started.reject(error); + this.#pending.delete(identity.requestId); + } + terminal.reject(error); + disposeListeners(); + }; + cleanups.push(this.#options.events.on(PI_SUBAGENT_STARTED_EVENT, (value) => { + if (!exactIdentity(value, identity) || startedAccepted || terminalAccepted) { + return; + } + startedAccepted = true; + resetIdleTimer(); + started.resolve(); + this.#options.onActivity?.(`Pi worker started · ${request.itemId}`); + })); + cleanups.push(this.#options.events.on(PI_SUBAGENT_UPDATE_EVENT, (value) => { + if (!exactIdentity(value, identity) || terminalAccepted) { + return; + } + const currentTool = typeof value.currentTool === "string" + ? boundUtf8(redactSecrets(value.currentTool, request), 128).value : undefined; + const fields = [ + currentTool, + typeof value.toolCount === "number" && Number.isFinite(value.toolCount) && value.toolCount >= 0 + ? `${value.toolCount} tools` : undefined, + typeof value.tokens === "number" && Number.isFinite(value.tokens) && value.tokens >= 0 + ? `${value.tokens} tokens` : undefined, + typeof value.durationMs === "number" && Number.isFinite(value.durationMs) && value.durationMs >= 0 + ? `${Math.round(value.durationMs / 1_000)}s` : undefined, + ].filter((field) => field !== undefined); + resetIdleTimer(); + if (fields.length > 0) { + this.#options.onActivity?.(`Pi worker · ${fields.join(" · ")}`); + } + })); + cleanups.push(this.#options.events.on(PI_SUBAGENT_RESPONSE_EVENT, (value) => { + if (!exactIdentity(value, identity) || terminalAccepted) { + return; + } + if (!startedAccepted) { + fail("Pi structured delegation returned terminal state before exact admission was observed"); + return; + } + terminalAccepted = true; + terminal.resolve(terminalObservation(request, identity, value)); + disposeListeners(); + })); + const timeoutMs = Math.max(1, Date.parse(request.deadline) - Date.now()); + const cancelForUnknownState = (message) => { + if (!terminalAccepted && this.#active) { + try { + this.#options.events.emit(PI_SUBAGENT_CANCEL_EVENT, delegationTuple(identity)); + } + catch { + fail(`${message}; cancellation delivery also failed`); + return; + } + } + fail(message); + }; + let idleTimer; + const resetIdleTimer = () => { + if (idleTimer !== undefined) { + clearTimeout(idleTimer); + } + idleTimer = setTimeout(() => { + cancelForUnknownState("Pi structured delegation exceeded the harness idle timeout without an exact terminal response"); + }, request.idleTimeoutMs); + idleTimer.unref(); + }; + resetIdleTimer(); + cleanups.push(() => { + if (idleTimer !== undefined) { + clearTimeout(idleTimer); + } + }); + const deadlineTimer = setTimeout(() => { + cancelForUnknownState("Pi structured delegation did not return an exact terminal response before the attempt deadline"); + }, timeoutMs); + deadlineTimer.unref(); + cleanups.push(() => clearTimeout(deadlineTimer)); + const pending = { + request, + identity, + startedAt: new Date().toISOString(), + terminal: terminal.promise, + rejectStarted: started.reject, + rejectTerminal: terminal.reject, + isTerminalAccepted: () => terminalAccepted, + disposeListeners, + }; + this.#pending.set(identity.requestId, pending); + try { + this.#options.events.emit(PI_SUBAGENT_REQUEST_EVENT, { + requestId: identity.requestId, + ownerRunId: identity.ownerRunId, + nodeId: identity.nodeId, + agent: "worker", + task: renderAttemptContext(request.context), + context: "fresh", + cwd: request.worktreePath, + timeoutMs, + artifacts: true, + result: { kind: "text" }, + }); + } + catch (error) { + fail(`Pi structured delegation request failed: ${error instanceof Error ? error.message : String(error)}`); + } + await started.promise; + return { + protocolVersion: 1, + adapterExecutionId: identity.requestId, + startedAt: pending.startedAt, + subject: { + schemaVersion: 1, + backendId: `pi-subagents-structured-v1@${this.#options.piSubagentsVersion}`, + subjectId: identity.subjectId, + harnessInstanceId: this.#options.harnessInstanceId, + }, + }; + } + async observe(handle) { + if (this.#reviewHandles.delete(handle.adapterExecutionId)) { + return await this.#options.reviewAdapter.observe(handle); + } + const pending = this.#pending.get(handle.adapterExecutionId); + if (pending === undefined || handle.subject?.subjectId !== pending.identity.subjectId + || handle.subject.harnessInstanceId !== this.#options.harnessInstanceId) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "Pi execution is not attached to the exact owning extension instance"); + } + try { + return await pending.terminal; + } + finally { + this.#pending.delete(handle.adapterExecutionId); + } + } + async cancel(handle) { + if (this.#reviewHandles.has(handle.adapterExecutionId)) { + return await this.#options.reviewAdapter.cancel(handle); + } + const pending = this.#pending.get(handle.adapterExecutionId); + if (!this.#active || pending === undefined || handle.subject?.subjectId !== pending.identity.subjectId + || handle.subject.harnessInstanceId !== this.#options.harnessInstanceId) { + return { protocolVersion: 1, accepted: false }; + } + try { + this.#options.events.emit(PI_SUBAGENT_CANCEL_EVENT, delegationTuple(pending.identity)); + return { protocolVersion: 1, accepted: true }; + } + catch { + return { protocolVersion: 1, accepted: false }; + } + } + invalidate(reason) { + if (!this.#active) { + return; + } + this.#active = false; + for (const pending of this.#pending.values()) { + if (pending.isTerminalAccepted()) { + continue; + } + const error = new AutopilotError("EXECUTION_STATE_UNKNOWN", reason); + pending.rejectStarted(error); + pending.rejectTerminal(error); + pending.disposeListeners(); + this.#pending.delete(pending.identity.requestId); + } + } +} diff --git a/skills/autopilot/runtime/dist/adapters/pi/index.js b/skills/autopilot/runtime/dist/adapters/pi/index.js index 972ff4d..fd100bd 100644 --- a/skills/autopilot/runtime/dist/adapters/pi/index.js +++ b/skills/autopilot/runtime/dist/adapters/pi/index.js @@ -1,7 +1,4 @@ -import { fileURLToPath } from "node:url"; import { CliHarnessAdapter } from "../../src/adapter-process.js"; -import { isRecord } from "../../src/json.js"; -import { findPiSubagentsInstallation } from "../../src/pi-subagents.js"; function directArguments(request, prompt) { return [ "--mode", "json", @@ -16,75 +13,21 @@ function directArguments(request, prompt) { prompt, ]; } -function subagentArguments(request, prompt, installation) { - const payload = { - runId: request.runId, - itemId: request.itemId, - task: prompt, - timeoutMs: Math.min(2_147_483_647, Math.max(1, Date.parse(request.deadline) - Date.now())), - }; - const encoded = Buffer.from(JSON.stringify(payload), "utf8").toString("base64url"); - const bridgePath = fileURLToPath(new URL("./bridge.js", import.meta.url)); - return [ - "--mode", "json", - "--print", - "--no-session", - "--no-extensions", - "--extension", installation.extensionPath, - "--extension", bridgePath, - "--no-skills", - "--no-prompt-templates", - "--no-context-files", - "--no-tools", - "--approve", - `/autopilot-worker ${encoded}`, - ]; -} -function validateSubagentResult(stdout) { - for (const line of stdout.split("\n")) { - if (line === "") { - continue; - } - let value; - try { - value = JSON.parse(line); - } - catch { - continue; - } - if (!isRecord(value) || value.type !== "message_end" || !isRecord(value.message) - || value.message.customType !== "autopilot-subagent-result" || !isRecord(value.message.details)) { - continue; - } - const status = value.message.details.status; - return status === "completed" ? undefined : `pi-subagents worker ended with ${typeof status === "string" ? status : "unknown status"}`; - } - return "pi-subagents worker did not emit a terminal result"; -} export function createPiAdapter() { - const installation = findPiSubagentsInstallation(); - const usingSubagents = installation !== undefined; return new CliHarnessAdapter({ name: "pi", executable: "pi", versionArguments: ["--version"], - buildArguments: (request, prompt) => installation === undefined || request.role === "review" - ? directArguments(request, prompt) - : subagentArguments(request, prompt, installation), + buildArguments: directArguments, assurance: "cooperative", maxConcurrency: 1, cancellation: true, limitations: [ + "Pi runs through the direct CLI fallback because no owning process-local Autopilot extension backend was selected.", "Tool restrictions do not constrain commands executed through bash.", - "Implementation executions use the attempt-scoped supervisor for restart reattachment; review executions remain session-scoped.", - usingSubagents - ? `Pi workers use the pi-subagents ${installation.version} structured delegation API.` - : "pi-subagents 0.53.0 or newer was not found; Pi workers run directly without subagent activity integration.", + "POSIX implementation executions retain process-supervised terminality; Windows direct execution remains session-scoped.", ], expectsJsonLines: true, - ...(usingSubagents ? { - validateResult: (stdout, request) => request.role === "review" ? undefined : validateSubagentResult(stdout), - } : {}), displayStderrActivity: true, }); } diff --git a/skills/autopilot/runtime/dist/src/cli.d.ts b/skills/autopilot/runtime/dist/src/cli.d.ts index 1b3887b..99c31fc 100644 --- a/skills/autopilot/runtime/dist/src/cli.d.ts +++ b/skills/autopilot/runtime/dist/src/cli.d.ts @@ -1,2 +1,13 @@ #!/usr/bin/env node +import type { HarnessPort } from "./adapter-protocol.js"; +import { type UnknownRecoveryRequest } from "./execution-recovery.js"; +export type CoordinatorAdapterFactory = (name: string) => HarnessPort; +export interface CoordinatorInvocationOptions { + readonly stateDir?: string; + readonly repairJournal?: boolean; + readonly adapterFactory: CoordinatorAdapterFactory; +} +export declare function startCoordinator(charterFile: string, options: CoordinatorInvocationOptions): Promise; +export declare function resumeCoordinator(runId: string | undefined, options: CoordinatorInvocationOptions): Promise; +export declare function recoverCoordinator(runId: string, request: UnknownRecoveryRequest, options: CoordinatorInvocationOptions): Promise; export declare function main(arguments_?: readonly string[]): Promise; diff --git a/skills/autopilot/runtime/dist/src/cli.js b/skills/autopilot/runtime/dist/src/cli.js index 8f16c49..368d2f3 100755 --- a/skills/autopilot/runtime/dist/src/cli.js +++ b/skills/autopilot/runtime/dist/src/cli.js @@ -12,7 +12,7 @@ import { runDoctor } from "./doctor.js"; import { AutopilotEngine } from "./engine.js"; import { AutopilotError } from "./errors.js"; import { newEventId } from "./events.js"; -import { recoverUnknownExecution } from "./execution-recovery.js"; +import { recoverUnknownExecution, } from "./execution-recovery.js"; import { appendEvent, readJournal, repairTruncatedJournal, writeImmutableJson } from "./journal.js"; import { isRecord } from "./json.js"; import { acquireBranchOwnershipLock, acquireRunLock, requestRunPause, requestRunStop } from "./lock.js"; @@ -100,7 +100,7 @@ async function loadRun(runId, stateDirectory, repair) { } return { stateRoot, directory, charter, journal }; } -async function runEngine(engine, lock, runId) { +async function runEngine(engine, lock, runId, captureProcessSignals) { const interrupt = () => { interrupted = true; void engine.requestStop(); @@ -139,8 +139,10 @@ async function runEngine(engine, lock, runId) { const controlMonitor = setInterval(checkControlRequest, 100); controlMonitor.unref(); checkControlRequest(); - process.once("SIGINT", interrupt); - process.once("SIGTERM", interrupt); + if (captureProcessSignals) { + process.once("SIGINT", interrupt); + process.once("SIGTERM", interrupt); + } try { return await engine.run(); } @@ -148,11 +150,13 @@ async function runEngine(engine, lock, runId) { controlMonitorStopped = true; clearInterval(controlMonitor); await pendingControlCheck; - process.removeListener("SIGINT", interrupt); - process.removeListener("SIGTERM", interrupt); + if (captureProcessSignals) { + process.removeListener("SIGINT", interrupt); + process.removeListener("SIGTERM", interrupt); + } } } -async function start(charterFile, options) { +async function start(charterFile, options, adapterFactory = createAdapter, captureProcessSignals = true) { let proposed; try { proposed = JSON.parse(await readFile(charterFile, "utf8")); @@ -234,11 +238,11 @@ async function start(charterFile, options) { stateRoot, runDirectory: directory, charter, - adapter: createAdapter(charter.harnessAdapter), + adapter: adapterFactory(charter.harnessAdapter), records: journal.records, projection, }); - return await runEngine(engine, lock, charter.runId); + return await runEngine(engine, lock, charter.runId, captureProcessSignals); } finally { await lock.release(); @@ -305,7 +309,7 @@ async function status(runId, options) { const metadata = await loadReportMetadata(run.directory); return await writeReports(run.directory, run.charter, projection, run.journal.records, metadata.assurance, metadata.unverifiedBoundaries, false); } -async function resume(runId, options) { +async function resume(runId, options, adapterFactory = createAdapter, captureProcessSignals = true) { const selected = await selectLifecycleRun("resume", runId, options); if (typeof selected !== "string") { return selected; @@ -327,11 +331,11 @@ async function resume(runId, options) { stateRoot: run.stateRoot, runDirectory: run.directory, charter: run.charter, - adapter: createAdapter(run.charter.harnessAdapter), + adapter: adapterFactory(run.charter.harnessAdapter), records: run.journal.records, projection, }); - return await runEngine(engine, lock, run.charter.runId); + return await runEngine(engine, lock, run.charter.runId, captureProcessSignals); } finally { await lock.release(); @@ -341,7 +345,7 @@ async function resume(runId, options) { await ownershipLock?.release(); } } -async function recover(runId, options) { +async function recover(runId, options, adapterFactory = createAdapter, captureProcessSignals = true) { if (runId === undefined || options.recoveryAction === undefined || options.recoveryItem === undefined || options.recoveryAttempt === undefined || options.recoveryLeaseEpoch === undefined || options.recoveryAttestation === undefined) { @@ -375,11 +379,11 @@ async function recover(runId, options) { stateRoot: run.stateRoot, runDirectory: run.directory, charter: run.charter, - adapter: createAdapter(run.charter.harnessAdapter), + adapter: adapterFactory(run.charter.harnessAdapter), records: journal.records, projection: recovered, }); - return await runEngine(engine, lock, run.charter.runId); + return await runEngine(engine, lock, run.charter.runId, captureProcessSignals); } finally { await lock.release(); @@ -389,6 +393,31 @@ async function recover(runId, options) { await ownershipLock?.release(); } } +function coordinatorOptions(options) { + return { + json: true, + repairJournal: options.repairJournal ?? false, + handoff: false, + ...(options.stateDir === undefined ? {} : { stateDir: options.stateDir }), + }; +} +export async function startCoordinator(charterFile, options) { + return await start(charterFile, coordinatorOptions(options), options.adapterFactory, false); +} +export async function resumeCoordinator(runId, options) { + return await resume(runId, coordinatorOptions(options), options.adapterFactory, false); +} +export async function recoverCoordinator(runId, request, options) { + return await recover(runId, { + ...coordinatorOptions(options), + recoveryAction: request.action, + recoveryItem: request.itemId, + recoveryAttempt: request.attemptId, + recoveryLeaseEpoch: request.leaseEpoch, + recoveryAttestation: request.attestation, + ...(request.expectedTreeIdentity === undefined ? {} : { recoveryTree: request.expectedTreeIdentity }), + }, options.adapterFactory, false); +} async function reviewFeedback(runId, options) { const stateRoot = await resolveStateRoot(process.cwd(), options.stateDir); return await observeReviewFeedback(stateRoot, process.cwd(), runId); @@ -483,7 +512,7 @@ async function pause(runId, options) { projection, }); await engine.requestPause(); - return await runEngine(engine, lock, run.charter.runId); + return await runEngine(engine, lock, run.charter.runId, true); } finally { await lock.release(); diff --git a/skills/autopilot/runtime/dist/src/doctor.js b/skills/autopilot/runtime/dist/src/doctor.js index ed0f561..a8fd4b0 100644 --- a/skills/autopilot/runtime/dist/src/doctor.js +++ b/skills/autopilot/runtime/dist/src/doctor.js @@ -59,7 +59,7 @@ export async function runDoctor() { }]; const piSubagents = findPiSubagentsInstallation(); checks.push(await commandCheck("git", "git", ["--version"], "Install Git and make it available on PATH."), await commandCheck("pi", "pi", ["--version"], "Install Pi only if you plan to use the Pi adapter."), piSubagents === undefined - ? { name: "pi-subagents", status: "unverified", detail: "version 0.53.0 or newer was not found; the Pi adapter will use its direct fallback", setup: "Install pi-subagents through Pi to enable delegated worker activity; Autopilot never installs it." } - : { name: "pi-subagents", status: "ok", detail: `${piSubagents.version} at ${piSubagents.extensionPath}` }, await commandCheck("claude-code", "claude", ["--version"], "Install Claude Code only if you plan to use that adapter."), await commandCheck("codex", "codex", ["--version"], "Install Codex only if you plan to use that adapter."), await commandCheck("opencode", "opencode", ["--version"], "Install OpenCode only if you plan to use that adapter."), await commandCheck("github-cli", "gh", ["--version"], "Install gh only for GitHub delivery."), await commandCheck("gitlab-cli", "glab", ["--version"], "Install glab only for GitLab delivery."), await authenticationCheck("claude-auth-config", "claude", ["auth", "status"]), await authenticationCheck("codex-auth-config", "codex", ["login", "status"]), await authenticationCheck("opencode-auth-config", "opencode", ["providers", "list"]), await authenticationCheck("github-auth", "gh", ["auth", "status"]), await authenticationCheck("gitlab-auth", "glab", ["auth", "status"]), await filesystemCheck()); + ? { name: "pi-subagents", status: "unverified", detail: "version 0.53.0 or newer was not found; Pi will use its distinct direct CLI fallback", setup: "Install and enable pi-subagents through Pi to use the process-local backend; Autopilot never installs it." } + : { name: "pi-subagents", status: "ok", detail: `${piSubagents.version} at ${piSubagents.extensionPath}; process-local owner availability is checked by the Autopilot Pi extension before launch` }, await commandCheck("claude-code", "claude", ["--version"], "Install Claude Code only if you plan to use that adapter."), await commandCheck("codex", "codex", ["--version"], "Install Codex only if you plan to use that adapter."), await commandCheck("opencode", "opencode", ["--version"], "Install OpenCode only if you plan to use that adapter."), await commandCheck("github-cli", "gh", ["--version"], "Install gh only for GitHub delivery."), await commandCheck("gitlab-cli", "glab", ["--version"], "Install glab only for GitLab delivery."), await authenticationCheck("claude-auth-config", "claude", ["auth", "status"]), await authenticationCheck("codex-auth-config", "codex", ["login", "status"]), await authenticationCheck("opencode-auth-config", "opencode", ["providers", "list"]), await authenticationCheck("github-auth", "gh", ["auth", "status"]), await authenticationCheck("gitlab-auth", "glab", ["auth", "status"]), await filesystemCheck()); return checks; } diff --git a/skills/autopilot/runtime/dist/src/pi-extension-entry.d.ts b/skills/autopilot/runtime/dist/src/pi-extension-entry.d.ts new file mode 100644 index 0000000..816d621 --- /dev/null +++ b/skills/autopilot/runtime/dist/src/pi-extension-entry.d.ts @@ -0,0 +1,2 @@ +import { registerAutopilotPiExtension } from "./pi-extension.js"; +export default function register(pi: Parameters[0]): void; diff --git a/skills/autopilot/runtime/dist/src/pi-extension-entry.js b/skills/autopilot/runtime/dist/src/pi-extension-entry.js new file mode 100644 index 0000000..0235ec9 --- /dev/null +++ b/skills/autopilot/runtime/dist/src/pi-extension-entry.js @@ -0,0 +1,5 @@ +import { VERSION as PI_VERSION } from "@earendil-works/pi-coding-agent"; +import { registerAutopilotPiExtension } from "./pi-extension.js"; +export default function register(pi) { + registerAutopilotPiExtension(pi, { piVersion: PI_VERSION }); +} diff --git a/skills/autopilot/runtime/dist/src/pi-extension.d.ts b/skills/autopilot/runtime/dist/src/pi-extension.d.ts new file mode 100644 index 0000000..4a51c33 --- /dev/null +++ b/skills/autopilot/runtime/dist/src/pi-extension.d.ts @@ -0,0 +1,24 @@ +import { type PiEventBus } from "../adapters/pi/in-process.js"; +interface PiCommandContext { + readonly cwd: string; + readonly sessionManager: { + getSessionId(): string | undefined; + }; + readonly ui: { + notify(message: string, level: "info" | "warning" | "error"): void; + }; +} +interface PiExtensionApi { + readonly events: PiEventBus; + on(event: "session_shutdown", handler: (event: { + readonly reason: string; + }) => void): void; + registerCommand(name: string, options: { + readonly description: string; + readonly handler: (arguments_: string, context: PiCommandContext) => Promise; + }): void; +} +export declare function registerAutopilotPiExtension(pi: PiExtensionApi, options?: { + readonly piVersion?: string; +}): void; +export default registerAutopilotPiExtension; diff --git a/skills/autopilot/runtime/dist/src/pi-extension.js b/skills/autopilot/runtime/dist/src/pi-extension.js new file mode 100644 index 0000000..8392874 --- /dev/null +++ b/skills/autopilot/runtime/dist/src/pi-extension.js @@ -0,0 +1,117 @@ +import { randomUUID } from "node:crypto"; +import { resolve } from "node:path"; +import { createPiAdapter } from "../adapters/pi/index.js"; +import { PiInProcessAdapter } from "../adapters/pi/in-process.js"; +import { createAdapter } from "./adapters.js"; +import { recoverCoordinator, resumeCoordinator, startCoordinator, } from "./cli.js"; +import { isRecord } from "./json.js"; +import { findPiSubagentsInstallation, probePiSubagentsOwner } from "./pi-subagents.js"; +async function selectAdapter(pi, cwd, harnessInstanceId, piVersion, activeAdapters) { + const installation = findPiSubagentsInstallation(cwd); + const ownerAvailable = installation !== undefined && await probePiSubagentsOwner(pi.events); + if (!ownerAvailable || installation === undefined) { + return { factory: createAdapter, mode: "direct-fallback" }; + } + const adapter = new PiInProcessAdapter({ + events: pi.events, + harnessInstanceId, + harnessVersion: piVersion, + piSubagentsVersion: installation.version, + reviewAdapter: createPiAdapter(), + onActivity: (message) => process.stderr.write(`[autopilot] ${message}\n`), + }); + activeAdapters.add(adapter); + return { + mode: "in-process", + inProcess: adapter, + factory: (name) => name === "pi" ? adapter : createAdapter(name), + }; +} +function parseRecovery(arguments_) { + const separator = arguments_.search(/\s/u); + if (separator < 1) { + throw new Error("Usage: /autopilot-recover "); + } + const runId = arguments_.slice(0, separator); + const requestValue = JSON.parse(arguments_.slice(separator).trim()); + if (!isRecord(requestValue) || !["abandon", "adopt", "stop"].includes(String(requestValue.action)) + || typeof requestValue.itemId !== "string" || typeof requestValue.attemptId !== "string" + || typeof requestValue.leaseEpoch !== "number" || !Number.isSafeInteger(requestValue.leaseEpoch) + || requestValue.leaseEpoch < 1 || typeof requestValue.attestation !== "string" + || (requestValue.expectedTreeIdentity !== undefined && typeof requestValue.expectedTreeIdentity !== "string")) { + throw new Error("Autopilot recovery request JSON is invalid"); + } + return { + runId, + request: { + action: requestValue.action, + itemId: requestValue.itemId, + attemptId: requestValue.attemptId, + leaseEpoch: requestValue.leaseEpoch, + attestation: requestValue.attestation, + ...(requestValue.expectedTreeIdentity === undefined + ? {} + : { expectedTreeIdentity: requestValue.expectedTreeIdentity }), + }, + }; +} +async function runWithSelectedAdapter(pi, context, harnessInstanceId, piVersion, activeAdapters, operation) { + const selected = await selectAdapter(pi, context.cwd, harnessInstanceId, piVersion, activeAdapters); + context.ui.notify(selected.mode === "in-process" + ? "Autopilot is using Pi process-local structured delegation." + : "Autopilot is using the distinct direct Pi CLI fallback.", selected.mode === "in-process" ? "info" : "warning"); + try { + const result = await operation(selected.factory); + const summary = isRecord(result) && typeof result.runId === "string" && typeof result.state === "string" + ? `${result.runId} · ${result.state}` + : "result available in the canonical Autopilot report"; + process.stderr.write(`[autopilot] coordinator completed · ${summary}\n`); + } + finally { + if (selected.inProcess !== undefined) { + activeAdapters.delete(selected.inProcess); + selected.inProcess.invalidate("the owning Autopilot coordinator invocation ended"); + } + } +} +export function registerAutopilotPiExtension(pi, options = {}) { + const extensionInstanceId = randomUUID(); + const piVersion = options.piVersion ?? "unknown"; + const activeAdapters = new Set(); + pi.on("session_shutdown", (event) => { + const reason = `the owning Pi extension context ended during ${event.reason}`; + activeAdapters.forEach((adapter) => adapter.invalidate(reason)); + activeAdapters.clear(); + }); + pi.registerCommand("autopilot-start", { + description: "Start a sealed Autopilot charter using the owning Pi extension context", + handler: async (arguments_, context) => { + const charterFile = arguments_.trim(); + if (charterFile === "") { + throw new Error("Usage: /autopilot-start "); + } + const sessionId = context.sessionManager.getSessionId() ?? "ephemeral"; + const harnessInstanceId = `${sessionId}:${extensionInstanceId}`; + await runWithSelectedAdapter(pi, context, harnessInstanceId, piVersion, activeAdapters, async (adapterFactory) => await startCoordinator(resolve(context.cwd, charterFile), { adapterFactory })); + }, + }); + pi.registerCommand("autopilot-resume", { + description: "Resume an interrupted Autopilot run using the owning Pi extension context", + handler: async (arguments_, context) => { + const sessionId = context.sessionManager.getSessionId() ?? "ephemeral"; + const harnessInstanceId = `${sessionId}:${extensionInstanceId}`; + const runId = arguments_.trim() || undefined; + await runWithSelectedAdapter(pi, context, harnessInstanceId, piVersion, activeAdapters, async (adapterFactory) => await resumeCoordinator(runId, { adapterFactory })); + }, + }); + pi.registerCommand("autopilot-recover", { + description: "Apply fenced unknown-execution recovery in the owning Pi extension context", + handler: async (arguments_, context) => { + const recovery = parseRecovery(arguments_); + const sessionId = context.sessionManager.getSessionId() ?? "ephemeral"; + const harnessInstanceId = `${sessionId}:${extensionInstanceId}`; + await runWithSelectedAdapter(pi, context, harnessInstanceId, piVersion, activeAdapters, async (adapterFactory) => await recoverCoordinator(recovery.runId, recovery.request, { adapterFactory })); + }, + }); +} +export default registerAutopilotPiExtension; diff --git a/skills/autopilot/runtime/dist/src/pi-subagents.d.ts b/skills/autopilot/runtime/dist/src/pi-subagents.d.ts index a17243a..6c5bd17 100644 --- a/skills/autopilot/runtime/dist/src/pi-subagents.d.ts +++ b/skills/autopilot/runtime/dist/src/pi-subagents.d.ts @@ -2,4 +2,9 @@ export interface PiSubagentsInstallation { readonly extensionPath: string; readonly version: string; } +export interface ProcessLocalEventBus { + on(event: string, handler: (value: unknown) => void): () => void; + emit(event: string, value: unknown): void; +} export declare function findPiSubagentsInstallation(cwd?: string): PiSubagentsInstallation | undefined; +export declare function probePiSubagentsOwner(events: ProcessLocalEventBus, timeoutMs?: number): Promise; diff --git a/skills/autopilot/runtime/dist/src/pi-subagents.js b/skills/autopilot/runtime/dist/src/pi-subagents.js index 9e50e30..efd2179 100644 --- a/skills/autopilot/runtime/dist/src/pi-subagents.js +++ b/skills/autopilot/runtime/dist/src/pi-subagents.js @@ -1,7 +1,10 @@ +import { randomUUID } from "node:crypto"; import { existsSync, readFileSync } from "node:fs"; import { homedir } from "node:os"; import { dirname, join, resolve } from "node:path"; import { isRecord } from "./json.js"; +const RPC_REQUEST_EVENT = "subagents:rpc:v1:request"; +const RPC_REPLY_PREFIX = "subagents:rpc:v1:reply:"; const MINIMUM_MINOR_VERSION = 53; function supportedVersion(version) { const match = /^(\d+)\.(\d+)\.(\d+)(?:[-+].*)?$/u.exec(version); @@ -37,3 +40,31 @@ export function findPiSubagentsInstallation(cwd = process.cwd()) { ]; return candidates.map(installationAt).find((installation) => installation !== undefined); } +export async function probePiSubagentsOwner(events, timeoutMs = 1_000) { + const requestId = randomUUID(); + return await new Promise((resolvePromise) => { + let settled = false; + const finish = (available) => { + if (settled) { + return; + } + settled = true; + clearTimeout(timer); + unsubscribe(); + resolvePromise(available); + }; + const unsubscribe = events.on(`${RPC_REPLY_PREFIX}${requestId}`, (value) => { + const data = isRecord(value) && isRecord(value.data) ? value.data : undefined; + finish(isRecord(value) && value.version === 1 && value.requestId === requestId && value.success === true + && data?.version === 1 && Array.isArray(data.methods) && data.methods.includes("ping")); + }); + const timer = setTimeout(() => finish(false), timeoutMs); + timer.unref(); + try { + events.emit(RPC_REQUEST_EVENT, { version: 1, requestId, method: "ping", params: {} }); + } + catch { + finish(false); + } + }); +} diff --git a/skills/autopilot/runtime/package-lock.json b/skills/autopilot/runtime/package-lock.json index 5033195..16b27b6 100644 --- a/skills/autopilot/runtime/package-lock.json +++ b/skills/autopilot/runtime/package-lock.json @@ -16,6 +16,14 @@ }, "engines": { "node": ">=24" + }, + "peerDependencies": { + "@earendil-works/pi-coding-agent": "*" + }, + "peerDependenciesMeta": { + "@earendil-works/pi-coding-agent": { + "optional": true + } } }, "node_modules/@types/node": { diff --git a/skills/autopilot/runtime/package.json b/skills/autopilot/runtime/package.json index 3d9ec0a..3df1bd0 100644 --- a/skills/autopilot/runtime/package.json +++ b/skills/autopilot/runtime/package.json @@ -11,6 +11,11 @@ "bin": { "autopilot": "./dist/src/cli.js" }, + "pi": { + "extensions": [ + "./dist/src/pi-extension-entry.js" + ] + }, "files": [ "dist/", "schemas/" @@ -24,6 +29,14 @@ "test": "npm run build && tsc -p tsconfig.test.json && node scripts/copy-native-helper.mjs && node --test .test-dist/test/*.test.js", "test:coverage": "npm run build && tsc -p tsconfig.test.json && node scripts/copy-native-helper.mjs && node --test --experimental-test-coverage .test-dist/test/*.test.js" }, + "peerDependencies": { + "@earendil-works/pi-coding-agent": "*" + }, + "peerDependenciesMeta": { + "@earendil-works/pi-coding-agent": { + "optional": true + } + }, "devDependencies": { "@types/node": "^24.10.0", "typescript": "^5.9.3" diff --git a/skills/autopilot/runtime/src/cli.ts b/skills/autopilot/runtime/src/cli.ts index 5bfb11c..4b25577 100644 --- a/skills/autopilot/runtime/src/cli.ts +++ b/skills/autopilot/runtime/src/cli.ts @@ -6,13 +6,18 @@ import { join } from "node:path"; import { pathToFileURL } from "node:url"; import { parseArgs } from "node:util"; import { createAdapter } from "./adapters.js"; +import type { HarnessPort } from "./adapter-protocol.js"; import { loadAmendmentContext } from "./amendment.js"; import { sealCharter, type RunCharter } from "./charter.js"; import { runDoctor } from "./doctor.js"; import { AutopilotEngine } from "./engine.js"; import { AutopilotError } from "./errors.js"; import { newEventId } from "./events.js"; -import { recoverUnknownExecution, type UnknownRecoveryAction } from "./execution-recovery.js"; +import { + recoverUnknownExecution, + type UnknownRecoveryAction, + type UnknownRecoveryRequest, +} from "./execution-recovery.js"; import { appendEvent, readJournal, repairTruncatedJournal, writeImmutableJson } from "./journal.js"; import { isRecord } from "./json.js"; import { acquireBranchOwnershipLock, acquireRunLock, requestRunPause, requestRunStop, type RunLock } from "./lock.js"; @@ -34,6 +39,14 @@ import { discoverWrapUpRuns, wrapUpRun, type WrapUpDiscovery } from "./wrap-up.j const VERSION = "0.1.0"; let interrupted = false; +export type CoordinatorAdapterFactory = (name: string) => HarnessPort; + +export interface CoordinatorInvocationOptions { + readonly stateDir?: string; + readonly repairJournal?: boolean; + readonly adapterFactory: CoordinatorAdapterFactory; +} + interface CliOptions { readonly stateDir?: string; readonly json: boolean; @@ -127,7 +140,12 @@ async function loadRun(runId: string, stateDirectory: string | undefined, repair return { stateRoot, directory, charter, journal }; } -async function runEngine(engine: AutopilotEngine, lock: RunLock, runId: string): Promise { +async function runEngine( + engine: AutopilotEngine, + lock: RunLock, + runId: string, + captureProcessSignals: boolean, +): Promise { const interrupt = (): void => { interrupted = true; void engine.requestStop(); @@ -163,20 +181,29 @@ async function runEngine(engine: AutopilotEngine, lock: RunLock, runId: string): const controlMonitor = setInterval(checkControlRequest, 100); controlMonitor.unref(); checkControlRequest(); - process.once("SIGINT", interrupt); - process.once("SIGTERM", interrupt); + if (captureProcessSignals) { + process.once("SIGINT", interrupt); + process.once("SIGTERM", interrupt); + } try { return await engine.run(); } finally { controlMonitorStopped = true; clearInterval(controlMonitor); await pendingControlCheck; - process.removeListener("SIGINT", interrupt); - process.removeListener("SIGTERM", interrupt); + if (captureProcessSignals) { + process.removeListener("SIGINT", interrupt); + process.removeListener("SIGTERM", interrupt); + } } } -async function start(charterFile: string, options: CliOptions): Promise { +async function start( + charterFile: string, + options: CliOptions, + adapterFactory: CoordinatorAdapterFactory = createAdapter, + captureProcessSignals = true, +): Promise { let proposed: unknown; try { proposed = JSON.parse(await readFile(charterFile, "utf8")) as unknown; @@ -259,11 +286,11 @@ async function start(charterFile: string, options: CliOptions): Promise stateRoot, runDirectory: directory, charter, - adapter: createAdapter(charter.harnessAdapter), + adapter: adapterFactory(charter.harnessAdapter), records: journal.records, projection, }); - return await runEngine(engine, lock, charter.runId); + return await runEngine(engine, lock, charter.runId, captureProcessSignals); } finally { await lock.release(); } @@ -342,7 +369,12 @@ async function status(runId: string | undefined, options: CliOptions): Promise { +async function resume( + runId: string | undefined, + options: CliOptions, + adapterFactory: CoordinatorAdapterFactory = createAdapter, + captureProcessSignals = true, +): Promise { const selected = await selectLifecycleRun("resume", runId, options); if (typeof selected !== "string") { return selected; @@ -364,11 +396,11 @@ async function resume(runId: string | undefined, options: CliOptions): Promise { +async function recover( + runId: string | undefined, + options: CliOptions, + adapterFactory: CoordinatorAdapterFactory = createAdapter, + captureProcessSignals = true, +): Promise { if (runId === undefined || options.recoveryAction === undefined || options.recoveryItem === undefined || options.recoveryAttempt === undefined || options.recoveryLeaseEpoch === undefined || options.recoveryAttestation === undefined) { @@ -418,11 +455,11 @@ async function recover(runId: string | undefined, options: CliOptions): Promise< stateRoot: run.stateRoot, runDirectory: run.directory, charter: run.charter, - adapter: createAdapter(run.charter.harnessAdapter), + adapter: adapterFactory(run.charter.harnessAdapter), records: journal.records, projection: recovered, }); - return await runEngine(engine, lock, run.charter.runId); + return await runEngine(engine, lock, run.charter.runId, captureProcessSignals); } finally { await lock.release(); } @@ -431,6 +468,39 @@ async function recover(runId: string | undefined, options: CliOptions): Promise< } } +function coordinatorOptions(options: CoordinatorInvocationOptions): CliOptions { + return { + json: true, + repairJournal: options.repairJournal ?? false, + handoff: false, + ...(options.stateDir === undefined ? {} : { stateDir: options.stateDir }), + }; +} + +export async function startCoordinator(charterFile: string, options: CoordinatorInvocationOptions): Promise { + return await start(charterFile, coordinatorOptions(options), options.adapterFactory, false); +} + +export async function resumeCoordinator(runId: string | undefined, options: CoordinatorInvocationOptions): Promise { + return await resume(runId, coordinatorOptions(options), options.adapterFactory, false); +} + +export async function recoverCoordinator( + runId: string, + request: UnknownRecoveryRequest, + options: CoordinatorInvocationOptions, +): Promise { + return await recover(runId, { + ...coordinatorOptions(options), + recoveryAction: request.action, + recoveryItem: request.itemId, + recoveryAttempt: request.attemptId, + recoveryLeaseEpoch: request.leaseEpoch, + recoveryAttestation: request.attestation, + ...(request.expectedTreeIdentity === undefined ? {} : { recoveryTree: request.expectedTreeIdentity }), + }, options.adapterFactory, false); +} + async function reviewFeedback(runId: string | undefined, options: CliOptions): Promise { const stateRoot = await resolveStateRoot(process.cwd(), options.stateDir); return await observeReviewFeedback(stateRoot, process.cwd(), runId); @@ -543,7 +613,7 @@ async function pause(runId: string | undefined, options: CliOptions): Promise { await commandCheck("git", "git", ["--version"], "Install Git and make it available on PATH."), await commandCheck("pi", "pi", ["--version"], "Install Pi only if you plan to use the Pi adapter."), piSubagents === undefined - ? { name: "pi-subagents", status: "unverified", detail: "version 0.53.0 or newer was not found; the Pi adapter will use its direct fallback", setup: "Install pi-subagents through Pi to enable delegated worker activity; Autopilot never installs it." } - : { name: "pi-subagents", status: "ok", detail: `${piSubagents.version} at ${piSubagents.extensionPath}` }, + ? { name: "pi-subagents", status: "unverified", detail: "version 0.53.0 or newer was not found; Pi will use its distinct direct CLI fallback", setup: "Install and enable pi-subagents through Pi to use the process-local backend; Autopilot never installs it." } + : { name: "pi-subagents", status: "ok", detail: `${piSubagents.version} at ${piSubagents.extensionPath}; process-local owner availability is checked by the Autopilot Pi extension before launch` }, await commandCheck("claude-code", "claude", ["--version"], "Install Claude Code only if you plan to use that adapter."), await commandCheck("codex", "codex", ["--version"], "Install Codex only if you plan to use that adapter."), await commandCheck("opencode", "opencode", ["--version"], "Install OpenCode only if you plan to use that adapter."), diff --git a/skills/autopilot/runtime/src/pi-core.d.ts b/skills/autopilot/runtime/src/pi-core.d.ts new file mode 100644 index 0000000..004dc28 --- /dev/null +++ b/skills/autopilot/runtime/src/pi-core.d.ts @@ -0,0 +1,3 @@ +declare module "@earendil-works/pi-coding-agent" { + export const VERSION: string; +} diff --git a/skills/autopilot/runtime/src/pi-extension-entry.ts b/skills/autopilot/runtime/src/pi-extension-entry.ts new file mode 100644 index 0000000..817eb83 --- /dev/null +++ b/skills/autopilot/runtime/src/pi-extension-entry.ts @@ -0,0 +1,6 @@ +import { VERSION as PI_VERSION } from "@earendil-works/pi-coding-agent"; +import { registerAutopilotPiExtension } from "./pi-extension.js"; + +export default function register(pi: Parameters[0]): void { + registerAutopilotPiExtension(pi, { piVersion: PI_VERSION }); +} diff --git a/skills/autopilot/runtime/src/pi-extension.ts b/skills/autopilot/runtime/src/pi-extension.ts new file mode 100644 index 0000000..cc32888 --- /dev/null +++ b/skills/autopilot/runtime/src/pi-extension.ts @@ -0,0 +1,182 @@ +import { randomUUID } from "node:crypto"; +import { resolve } from "node:path"; +import { createPiAdapter } from "../adapters/pi/index.js"; +import { PiInProcessAdapter, type PiEventBus } from "../adapters/pi/in-process.js"; +import { createAdapter } from "./adapters.js"; +import type { HarnessPort } from "./adapter-protocol.js"; +import { + recoverCoordinator, + resumeCoordinator, + startCoordinator, + type CoordinatorAdapterFactory, +} from "./cli.js"; +import type { UnknownRecoveryAction, UnknownRecoveryRequest } from "./execution-recovery.js"; +import { isRecord } from "./json.js"; +import { findPiSubagentsInstallation, probePiSubagentsOwner } from "./pi-subagents.js"; + +interface PiCommandContext { + readonly cwd: string; + readonly sessionManager: { + getSessionId(): string | undefined; + }; + readonly ui: { + notify(message: string, level: "info" | "warning" | "error"): void; + }; +} + +interface PiExtensionApi { + readonly events: PiEventBus; + on(event: "session_shutdown", handler: (event: { readonly reason: string }) => void): void; + registerCommand(name: string, options: { + readonly description: string; + readonly handler: (arguments_: string, context: PiCommandContext) => Promise; + }): void; +} + +interface SelectedAdapter { + readonly factory: CoordinatorAdapterFactory; + readonly inProcess?: PiInProcessAdapter; + readonly mode: "in-process" | "direct-fallback"; +} + +async function selectAdapter( + pi: PiExtensionApi, + cwd: string, + harnessInstanceId: string, + piVersion: string, + activeAdapters: Set, +): Promise { + const installation = findPiSubagentsInstallation(cwd); + const ownerAvailable = installation !== undefined && await probePiSubagentsOwner(pi.events); + if (!ownerAvailable || installation === undefined) { + return { factory: createAdapter, mode: "direct-fallback" }; + } + const adapter = new PiInProcessAdapter({ + events: pi.events, + harnessInstanceId, + harnessVersion: piVersion, + piSubagentsVersion: installation.version, + reviewAdapter: createPiAdapter(), + onActivity: (message) => process.stderr.write(`[autopilot] ${message}\n`), + }); + activeAdapters.add(adapter); + return { + mode: "in-process", + inProcess: adapter, + factory: (name: string): HarnessPort => name === "pi" ? adapter : createAdapter(name), + }; +} + +function parseRecovery(arguments_: string): { readonly runId: string; readonly request: UnknownRecoveryRequest } { + const separator = arguments_.search(/\s/u); + if (separator < 1) { + throw new Error("Usage: /autopilot-recover "); + } + const runId = arguments_.slice(0, separator); + const requestValue: unknown = JSON.parse(arguments_.slice(separator).trim()); + if (!isRecord(requestValue) || !["abandon", "adopt", "stop"].includes(String(requestValue.action)) + || typeof requestValue.itemId !== "string" || typeof requestValue.attemptId !== "string" + || typeof requestValue.leaseEpoch !== "number" || !Number.isSafeInteger(requestValue.leaseEpoch) + || requestValue.leaseEpoch < 1 || typeof requestValue.attestation !== "string" + || (requestValue.expectedTreeIdentity !== undefined && typeof requestValue.expectedTreeIdentity !== "string")) { + throw new Error("Autopilot recovery request JSON is invalid"); + } + return { + runId, + request: { + action: requestValue.action as UnknownRecoveryAction, + itemId: requestValue.itemId, + attemptId: requestValue.attemptId, + leaseEpoch: requestValue.leaseEpoch, + attestation: requestValue.attestation, + ...(requestValue.expectedTreeIdentity === undefined + ? {} + : { expectedTreeIdentity: requestValue.expectedTreeIdentity }), + }, + }; +} + +async function runWithSelectedAdapter( + pi: PiExtensionApi, + context: PiCommandContext, + harnessInstanceId: string, + piVersion: string, + activeAdapters: Set, + operation: (factory: CoordinatorAdapterFactory) => Promise, +): Promise { + const selected = await selectAdapter(pi, context.cwd, harnessInstanceId, piVersion, activeAdapters); + context.ui.notify( + selected.mode === "in-process" + ? "Autopilot is using Pi process-local structured delegation." + : "Autopilot is using the distinct direct Pi CLI fallback.", + selected.mode === "in-process" ? "info" : "warning", + ); + try { + const result = await operation(selected.factory); + const summary = isRecord(result) && typeof result.runId === "string" && typeof result.state === "string" + ? `${result.runId} · ${result.state}` + : "result available in the canonical Autopilot report"; + process.stderr.write(`[autopilot] coordinator completed · ${summary}\n`); + } finally { + if (selected.inProcess !== undefined) { + activeAdapters.delete(selected.inProcess); + selected.inProcess.invalidate("the owning Autopilot coordinator invocation ended"); + } + } +} + +export function registerAutopilotPiExtension( + pi: PiExtensionApi, + options: { readonly piVersion?: string } = {}, +): void { + const extensionInstanceId = randomUUID(); + const piVersion = options.piVersion ?? "unknown"; + const activeAdapters = new Set(); + + pi.on("session_shutdown", (event) => { + const reason = `the owning Pi extension context ended during ${event.reason}`; + activeAdapters.forEach((adapter) => adapter.invalidate(reason)); + activeAdapters.clear(); + }); + + pi.registerCommand("autopilot-start", { + description: "Start a sealed Autopilot charter using the owning Pi extension context", + handler: async (arguments_, context) => { + const charterFile = arguments_.trim(); + if (charterFile === "") { + throw new Error("Usage: /autopilot-start "); + } + const sessionId = context.sessionManager.getSessionId() ?? "ephemeral"; + const harnessInstanceId = `${sessionId}:${extensionInstanceId}`; + await runWithSelectedAdapter(pi, context, harnessInstanceId, piVersion, activeAdapters, async (adapterFactory) => + await startCoordinator(resolve(context.cwd, charterFile), { adapterFactory }) + ); + }, + }); + + pi.registerCommand("autopilot-resume", { + description: "Resume an interrupted Autopilot run using the owning Pi extension context", + handler: async (arguments_, context) => { + const sessionId = context.sessionManager.getSessionId() ?? "ephemeral"; + const harnessInstanceId = `${sessionId}:${extensionInstanceId}`; + const runId = arguments_.trim() || undefined; + await runWithSelectedAdapter(pi, context, harnessInstanceId, piVersion, activeAdapters, async (adapterFactory) => + await resumeCoordinator(runId, { adapterFactory }) + ); + }, + }); + + pi.registerCommand("autopilot-recover", { + description: "Apply fenced unknown-execution recovery in the owning Pi extension context", + handler: async (arguments_, context) => { + const recovery = parseRecovery(arguments_); + const sessionId = context.sessionManager.getSessionId() ?? "ephemeral"; + const harnessInstanceId = `${sessionId}:${extensionInstanceId}`; + await runWithSelectedAdapter(pi, context, harnessInstanceId, piVersion, activeAdapters, async (adapterFactory) => + await recoverCoordinator(recovery.runId, recovery.request, { adapterFactory }) + ); + }, + }); +} + +export default registerAutopilotPiExtension; diff --git a/skills/autopilot/runtime/src/pi-subagents.ts b/skills/autopilot/runtime/src/pi-subagents.ts index 594a69a..2be3ca4 100644 --- a/skills/autopilot/runtime/src/pi-subagents.ts +++ b/skills/autopilot/runtime/src/pi-subagents.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import { existsSync, readFileSync } from "node:fs"; import { homedir } from "node:os"; import { dirname, join, resolve } from "node:path"; @@ -8,6 +9,14 @@ export interface PiSubagentsInstallation { readonly version: string; } +export interface ProcessLocalEventBus { + on(event: string, handler: (value: unknown) => void): () => void; + emit(event: string, value: unknown): void; +} + +const RPC_REQUEST_EVENT = "subagents:rpc:v1:request"; +const RPC_REPLY_PREFIX = "subagents:rpc:v1:reply:"; + const MINIMUM_MINOR_VERSION = 53; function supportedVersion(version: string): boolean { @@ -45,3 +54,31 @@ export function findPiSubagentsInstallation(cwd = process.cwd()): PiSubagentsIns ]; return candidates.map(installationAt).find((installation) => installation !== undefined); } + +export async function probePiSubagentsOwner(events: ProcessLocalEventBus, timeoutMs = 1_000): Promise { + const requestId = randomUUID(); + return await new Promise((resolvePromise) => { + let settled = false; + const finish = (available: boolean): void => { + if (settled) { + return; + } + settled = true; + clearTimeout(timer); + unsubscribe(); + resolvePromise(available); + }; + const unsubscribe = events.on(`${RPC_REPLY_PREFIX}${requestId}`, (value) => { + const data = isRecord(value) && isRecord(value.data) ? value.data : undefined; + finish(isRecord(value) && value.version === 1 && value.requestId === requestId && value.success === true + && data?.version === 1 && Array.isArray(data.methods) && data.methods.includes("ping")); + }); + const timer = setTimeout(() => finish(false), timeoutMs); + timer.unref(); + try { + events.emit(RPC_REQUEST_EVENT, { version: 1, requestId, method: "ping", params: {} }); + } catch { + finish(false); + } + }); +} diff --git a/skills/autopilot/runtime/test/packaging.test.ts b/skills/autopilot/runtime/test/packaging.test.ts index 8d85d53..e2fd342 100644 --- a/skills/autopilot/runtime/test/packaging.test.ts +++ b/skills/autopilot/runtime/test/packaging.test.ts @@ -24,6 +24,10 @@ test("compiled skill CLI starts from a clean copy without node_modules", async ( assert.equal(version.exitCode, 0); assert.equal(version.stdout.trim(), "0.1.0"); assert.equal(doctor.exitCode, 0); + const packageManifest: unknown = JSON.parse(await readFile(join(copyRoot, "package.json"), "utf8")); + assert.ok(isRecord(packageManifest) && isRecord(packageManifest.pi)); + assert.deepEqual(packageManifest.pi.extensions, ["./dist/src/pi-extension-entry.js"]); + assert.equal(existsSync(join(copyRoot, "dist", "src", "pi-extension-entry.js")), true); const checks: unknown = JSON.parse(doctor.stdout); assert.ok(Array.isArray(checks)); const node = checks.find((entry) => isRecord(entry) && entry.name === "node"); diff --git a/skills/autopilot/runtime/test/pi-subagents.test.ts b/skills/autopilot/runtime/test/pi-subagents.test.ts index 48c0663..7a01ef9 100644 --- a/skills/autopilot/runtime/test/pi-subagents.test.ts +++ b/skills/autopilot/runtime/test/pi-subagents.test.ts @@ -1,12 +1,28 @@ import assert from "node:assert/strict"; -import { mkdir, mkdtemp, readFile, writeFile } from "node:fs/promises"; +import { mkdir, mkdtemp, readFile, realpath, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { delimiter, join } from "node:path"; import { test } from "node:test"; import { createPiAdapter } from "../adapters/pi/index.js"; -import type { ExecutionRequest } from "../src/adapter-protocol.js"; -import { findPiSubagentsInstallation } from "../src/pi-subagents.js"; -import { attemptContextFixture, writeNodeExecutable } from "./helpers.js"; +import { + PiInProcessAdapter, + PI_SUBAGENT_CANCEL_EVENT, + PI_SUBAGENT_REQUEST_EVENT, + PI_SUBAGENT_RESPONSE_EVENT, + PI_SUBAGENT_STARTED_EVENT, + type PiEventBus, +} from "../adapters/pi/in-process.js"; +import { registerAutopilotPiExtension } from "../src/pi-extension.js"; +import type { + CancelResult, + CapabilityManifest, + ExecutionHandle, + ExecutionObservation, + ExecutionRequest, + HarnessPort, +} from "../src/adapter-protocol.js"; +import { findPiSubagentsInstallation, probePiSubagentsOwner } from "../src/pi-subagents.js"; +import { attemptContextFixture, createRepository, proposedCharter, writeNodeExecutable } from "./helpers.js"; async function fakeInstallation(root: string, version: string): Promise { const packageRoot = join(root, "npm", "node_modules", "pi-subagents"); @@ -17,7 +33,7 @@ async function fakeInstallation(root: string, version: string): Promise return extensionPath; } -function request(role: "implementation" | "review" = "implementation"): ExecutionRequest { +function request(role: "implementation" | "review" = "implementation", maximumOutputBytes = 262_144): ExecutionRequest { return { protocolVersion: 1, role, @@ -34,11 +50,110 @@ function request(role: "implementation" | "review" = "implementation"): Executio deadline: new Date(Date.now() + 30_000).toISOString(), idleTimeoutMs: 10_000, maximumLineBytes: 65_536, - maximumOutputBytes: 262_144, + maximumOutputBytes, ...(role === "review" ? { reviewFocus: "Review the exact tree." } : {}), }; } +class FakeEventBus implements PiEventBus { + readonly emitted: Array<{ readonly event: string; readonly value: unknown }> = []; + readonly #listeners = new Map void>>(); + onEmit?: (event: string, value: unknown) => void; + + on(event: string, handler: (value: unknown) => void): () => void { + const listeners = this.#listeners.get(event) ?? new Set(); + listeners.add(handler); + this.#listeners.set(event, listeners); + return () => listeners.delete(handler); + } + + emit(event: string, value: unknown): void { + this.emitted.push({ event, value }); + this.onEmit?.(event, value); + this.#listeners.get(event)?.forEach((handler) => handler(value)); + } +} + +interface FakeCommandContext { + readonly cwd: string; + readonly sessionManager: { getSessionId(): string }; + readonly ui: { notify(message: string, level: "info" | "warning" | "error"): void }; +} + +class FakePiExtension { + readonly events = new FakeEventBus(); + readonly commands = new Map Promise>(); + readonly shutdownHandlers: Array<(event: { readonly reason: string }) => void> = []; + + on(event: "session_shutdown", handler: (event: { readonly reason: string }) => void): void { + assert.equal(event, "session_shutdown"); + this.shutdownHandlers.push(handler); + } + + registerCommand(name: string, options: { + readonly description: string; + readonly handler: (arguments_: string, context: FakeCommandContext) => Promise; + }): void { + assert.notEqual(options.description, ""); + this.commands.set(name, options.handler); + } +} + +class FakeReviewAdapter implements HarnessPort { + launches = 0; + + async describe(): Promise { + throw new Error("review adapter describe is not used by the in-process adapter"); + } + + async launch(_request: ExecutionRequest): Promise { + this.launches += 1; + return { protocolVersion: 1, adapterExecutionId: "review-1", startedAt: new Date().toISOString() }; + } + + async observe(handle: ExecutionHandle): Promise { + return { + protocolVersion: 1, + adapterExecutionId: handle.adapterExecutionId, + status: "completed", + exitCode: 0, + completedAt: new Date().toISOString(), + stdout: "AUTOPILOT_REVIEW_RESULT:{\"verdict\":\"clean\",\"findings\":[]}", + stderr: "", + truncated: false, + reviewResult: { verdict: "clean", findings: [] }, + }; + } + + async cancel(_handle: ExecutionHandle): Promise { + return { protocolVersion: 1, accepted: true }; + } +} + +function inProcessAdapter(bus: FakeEventBus, reviewAdapter = new FakeReviewAdapter()): PiInProcessAdapter { + return new PiInProcessAdapter({ + events: bus, + harnessInstanceId: "session-1:extension-1", + harnessVersion: "0.84.4", + piSubagentsVersion: "0.60.0", + reviewAdapter, + }); +} + +function startOnRequest(bus: FakeEventBus): void { + bus.onEmit = (event, value) => { + if (event === PI_SUBAGENT_REQUEST_EVENT) { + bus.emit(PI_SUBAGENT_STARTED_EVENT, value); + } + }; +} + +function emittedRequest(bus: FakeEventBus): Record { + const value = bus.emitted.filter(({ event }) => event === PI_SUBAGENT_REQUEST_EVENT).at(-1)?.value; + assert.ok(typeof value === "object" && value !== null && !Array.isArray(value)); + return value as Record; +} + test("pi-subagents discovery requires the supported structured delegation version", async () => { const agentDirectory = await mkdtemp(join(tmpdir(), "autopilot-pi-agent-")); const previous = process.env.PI_CODING_AGENT_DIR; @@ -58,11 +173,30 @@ test("pi-subagents discovery requires the supported structured delegation versio } }); -test("Pi adapter uses the direct worker fallback when pi-subagents is unavailable", async () => { +test("process-local pi-subagents owner probe requires an exact RPC ping reply", async () => { + const available = new FakeEventBus(); + available.onEmit = (event, value) => { + if (event === "subagents:rpc:v1:request" && typeof value === "object" && value !== null) { + const requestValue = value as Record; + available.emit(`subagents:rpc:v1:reply:${String(requestValue.requestId)}`, { + version: 1, + requestId: requestValue.requestId, + success: true, + data: { version: 1, methods: ["ping", "status"] }, + }); + } + }; + + assert.equal(await probePiSubagentsOwner(available, 50), true); + assert.equal(await probePiSubagentsOwner(new FakeEventBus(), 10), false); +}); + +test("Pi adapter uses the direct worker fallback without loading another extension", async () => { const root = await mkdtemp(join(tmpdir(), "autopilot-pi-direct-")); const agentDirectory = join(root, "agent"); const bin = join(root, "bin"); const marker = join(root, "arguments.json"); + await fakeInstallation(agentDirectory, "0.60.0"); await mkdir(bin, { recursive: true }); await writeNodeExecutable(bin, "pi", `#!/usr/bin/env node import { writeFileSync } from "node:fs"; @@ -102,75 +236,253 @@ console.log(JSON.stringify({type:"agent_settled"})); } }); -test("Pi adapter delegates through an available pi-subagents installation", async () => { - const root = await mkdtemp(join(tmpdir(), "autopilot-pi-subagents-")); - const agentDirectory = join(root, "agent"); - const bin = join(root, "bin"); - const marker = join(root, "arguments.json"); - await fakeInstallation(agentDirectory, "0.53.0"); - await mkdir(bin, { recursive: true }); - const script = `#!/usr/bin/env node -import { writeFileSync } from "node:fs"; -const args = process.argv.slice(2); -if (args[0] === "--version") { - console.log("pi 0.84.2"); -} else { - writeFileSync(process.env.AUTOPILOT_PI_ARGUMENTS, JSON.stringify(args)); - const message = {role:"custom", customType:"autopilot-subagent-result", content:"done", display:true, details:{status:process.env.AUTOPILOT_PI_STATUS ?? "completed"}}; - console.log(JSON.stringify({type:"message_end", message})); -} -`; - await writeNodeExecutable(bin, "pi", script); +test("Pi in-process adapter binds exact admission and preserves terminal completion across shutdown", async () => { + const bus = new FakeEventBus(); + startOnRequest(bus); + const adapter = inProcessAdapter(bus); + const manifest = await adapter.describe(); + + const handle = await adapter.launch(request()); + const delegation = emittedRequest(bus); + assert.equal(manifest.executionAssurance?.implementation.owner, "harness"); + assert.equal(manifest.executionAssurance?.implementation.continuity, "same-harness-instance"); + assert.equal(handle.adapterExecutionId, delegation.requestId); + assert.equal(handle.subject?.harnessInstanceId, "session-1:extension-1"); + assert.notEqual(delegation.nodeId, "item-1"); + + bus.emit(PI_SUBAGENT_RESPONSE_EVENT, { + requestId: "other-request", + ownerRunId: delegation.ownerRunId, + nodeId: delegation.nodeId, + status: "completed", + result: { kind: "text", text: "wrong" }, + }); + bus.emit(PI_SUBAGENT_RESPONSE_EVENT, { + requestId: delegation.requestId, + ownerRunId: delegation.ownerRunId, + nodeId: delegation.nodeId, + status: "completed", + result: { kind: "text", text: "done" }, + }); + adapter.invalidate("extension reloaded after terminal response"); + + const observation = await adapter.observe(handle); + assert.equal(observation.status, "completed"); + assert.equal(observation.stdout, "done"); +}); + +test("Pi in-process adapter makes admission or observation loss unknown without another request", async () => { + const beforeAdmission = new FakeEventBus(); + const firstAdapter = inProcessAdapter(beforeAdmission); + const launch = firstAdapter.launch(request()); + firstAdapter.invalidate("extension reloaded before exact admission"); + await assert.rejects(launch, /before exact admission/); + assert.equal(beforeAdmission.emitted.filter(({ event }) => event === PI_SUBAGENT_REQUEST_EVENT).length, 1); + + const afterAdmission = new FakeEventBus(); + startOnRequest(afterAdmission); + const secondAdapter = inProcessAdapter(afterAdmission); + const handle = await secondAdapter.launch(request()); + secondAdapter.invalidate("session replaced before terminal response"); + await assert.rejects(secondAdapter.observe(handle), /not attached|session replaced/); + assert.equal(afterAdmission.emitted.filter(({ event }) => event === PI_SUBAGENT_REQUEST_EVENT).length, 1); + + const idleBus = new FakeEventBus(); + idleBus.onEmit = (event, value) => { + if (event === PI_SUBAGENT_REQUEST_EVENT) { + idleBus.emit(PI_SUBAGENT_STARTED_EVENT, value); + } else if (event === PI_SUBAGENT_CANCEL_EVENT) { + throw new Error("cancel listener failed"); + } + }; + const idleAdapter = inProcessAdapter(idleBus); + const idleHandle = await idleAdapter.launch({ ...request(), idleTimeoutMs: 5 }); + await assert.rejects(idleAdapter.observe(idleHandle), /idle timeout.*cancellation delivery also failed/); + assert.equal(idleBus.emitted.filter(({ event }) => event === PI_SUBAGENT_CANCEL_EVENT).length, 1); +}); + +test("Pi in-process cancellation emits the exact tuple and waits for terminal cancellation", async () => { + const bus = new FakeEventBus(); + startOnRequest(bus); + const adapter = inProcessAdapter(bus); + const handle = await adapter.launch(request()); + const delegation = emittedRequest(bus); + + assert.deepEqual(await adapter.cancel(handle), { protocolVersion: 1, accepted: true }); + const cancellation = bus.emitted.find(({ event }) => event === PI_SUBAGENT_CANCEL_EVENT)?.value; + assert.deepEqual(cancellation, { + requestId: delegation.requestId, + ownerRunId: delegation.ownerRunId, + nodeId: delegation.nodeId, + }); + + bus.emit(PI_SUBAGENT_RESPONSE_EVENT, { + requestId: delegation.requestId, + ownerRunId: delegation.ownerRunId, + nodeId: delegation.nodeId, + status: "cancelled", + }); + const observation = await adapter.observe(handle); + assert.equal(observation.status, "cancelled"); +}); + +test("Pi in-process adapter fails a bounded malformed result and keeps review direct", async () => { + const bus = new FakeEventBus(); + startOnRequest(bus); + const reviewAdapter = new FakeReviewAdapter(); + const adapter = inProcessAdapter(bus, reviewAdapter); + const handle = await adapter.launch(request("implementation", 4)); + const delegation = emittedRequest(bus); + bus.emit(PI_SUBAGENT_RESPONSE_EVENT, { + requestId: delegation.requestId, + ownerRunId: delegation.ownerRunId, + nodeId: delegation.nodeId, + status: "completed", + result: { kind: "text", text: "oversized" }, + }); + + const observation = await adapter.observe(handle); + assert.equal(observation.status, "failed"); + assert.match(observation.stderr, /output bound/); + + const previousSecret = process.env.AUTOPILOT_SECRET_TOKEN; + process.env.AUTOPILOT_SECRET_TOKEN = "x"; + try { + const failedHandle = await adapter.launch(request()); + const failedDelegation = emittedRequest(bus); + bus.emit(PI_SUBAGENT_RESPONSE_EVENT, { + requestId: failedDelegation.requestId, + ownerRunId: failedDelegation.ownerRunId, + nodeId: failedDelegation.nodeId, + status: "failed", + error: "failure included x", + }); + assert.equal((await adapter.observe(failedHandle)).stderr, "failure included ****"); + } finally { + if (previousSecret === undefined) { + delete process.env.AUTOPILOT_SECRET_TOKEN; + } else { + process.env.AUTOPILOT_SECRET_TOKEN = previousSecret; + } + } + + const reviewHandle = await adapter.launch(request("review")); + const reviewObservation = await adapter.observe(reviewHandle); + assert.equal(reviewAdapter.launches, 1); + assert.deepEqual(reviewObservation.reviewResult, { verdict: "clean", findings: [] }); +}); + +test("Pi extension invokes the runtime core and completes through process-local delegation", async () => { + const repository = await createRepository(); + const repositoryRoot = await realpath(repository.root); + const runId = "run-pi-extension"; + const charterPath = join(repositoryRoot, "charter.json"); + await writeFile(charterPath, JSON.stringify(proposedCharter(repositoryRoot, repository.baseCommit, "single", runId))); + const agentDirectory = await mkdtemp(join(tmpdir(), "autopilot-pi-extension-agent-")); + await fakeInstallation(agentDirectory, "0.60.0"); const previousAgentDirectory = process.env.PI_CODING_AGENT_DIR; - const previousPath = process.env.PATH; - const previousMarker = process.env.AUTOPILOT_PI_ARGUMENTS; - const previousStatus = process.env.AUTOPILOT_PI_STATUS; process.env.PI_CODING_AGENT_DIR = agentDirectory; - process.env.PATH = `${bin}${delimiter}${previousPath ?? ""}`; - process.env.AUTOPILOT_PI_ARGUMENTS = marker; + const pi = new FakePiExtension(); + const notifications: string[] = []; + let completeDelegations = true; + pi.events.onEmit = (event, value) => { + if (event === "subagents:rpc:v1:request" && typeof value === "object" && value !== null) { + const rpc = value as Record; + pi.events.emit(`subagents:rpc:v1:reply:${String(rpc.requestId)}`, { + version: 1, + requestId: rpc.requestId, + success: true, + data: { version: 1, methods: ["ping"] }, + }); + return; + } + if (event === PI_SUBAGENT_REQUEST_EVENT && typeof value === "object" && value !== null) { + const delegation = value as Record; + pi.events.emit(PI_SUBAGENT_STARTED_EVENT, delegation); + if (completeDelegations) { + void writeFile(join(String(delegation.cwd), "result.txt"), "done\n").then(() => { + pi.events.emit(PI_SUBAGENT_RESPONSE_EVENT, { + requestId: delegation.requestId, + ownerRunId: delegation.ownerRunId, + nodeId: delegation.nodeId, + status: "completed", + result: { kind: "text", text: "created result.txt" }, + }); + }); + } + } + }; try { - const adapter = createPiAdapter(); + registerAutopilotPiExtension(pi, { piVersion: "0.84.4" }); + const start = pi.commands.get("autopilot-start"); + assert.ok(start !== undefined); + await start(charterPath, { + cwd: repositoryRoot, + sessionManager: { getSessionId: () => "pi-session-1" }, + ui: { notify: (message) => notifications.push(message) }, + }); - const handle = await adapter.launch(request()); - const observation = await adapter.observe(handle); - const arguments_: unknown = JSON.parse(await readFile(marker, "utf8")); + const report: unknown = JSON.parse(await readFile( + join(repositoryRoot, ".git", "autopilot", "runs", runId, "reports", "final.json"), + "utf8", + )); + assert.ok(typeof report === "object" && report !== null); + assert.equal((report as Record).state, "SUCCEEDED"); + assert.match(notifications.join("\n"), /process-local structured delegation/); + const journal = await readFile( + join(repositoryRoot, ".git", "autopilot", "runs", runId, "events.jsonl"), + "utf8", + ); + assert.match(journal, /ATTEMPT_EXECUTION_ADMITTED/); + assert.match(journal, /pi-subagents-structured-v1@0\.60\.0/); - assert.equal(observation.status, "completed"); - assert.ok(Array.isArray(arguments_)); - assert.ok(arguments_.includes("--extension")); - assert.ok(arguments_.some((argument) => typeof argument === "string" && argument.startsWith("/autopilot-worker "))); - - const reviewScript = `#!/usr/bin/env node -console.log(JSON.stringify({type:"message", message:{role:"assistant", content:[{type:"text", text:'AUTOPILOT_REVIEW_RESULT:{"verdict":"clean","findings":[]}'}]}})); -`; - await writeNodeExecutable(bin, "pi", reviewScript); - const reviewHandle = await adapter.launch(request("review")); - const reviewObservation = await adapter.observe(reviewHandle); - assert.equal(reviewObservation.status, "completed"); - assert.deepEqual(reviewObservation.reviewResult, { verdict: "clean", findings: [] }); - - await writeNodeExecutable(bin, "pi", script); - process.env.AUTOPILOT_PI_STATUS = "failed"; - const failedHandle = await adapter.launch(request()); - const failedObservation = await adapter.observe(failedHandle); - assert.equal(failedObservation.status, "failed"); - assert.match(failedObservation.stderr, /pi-subagents worker ended with failed/); + completeDelegations = false; + const reloadRunId = "reload31-extension"; + const reloadCharterPath = join(repositoryRoot, "charter-reload.json"); + await writeFile( + reloadCharterPath, + JSON.stringify(proposedCharter(repositoryRoot, repository.baseCommit, "single", reloadRunId)), + ); + const reloadExecution = start(reloadCharterPath, { + cwd: repositoryRoot, + sessionManager: { getSessionId: () => "pi-session-1" }, + ui: { notify: (message) => notifications.push(message) }, + }); + const reloadJournalPath = join( + repositoryRoot, + ".git", + "autopilot", + "runs", + reloadRunId, + "events.jsonl", + ); + let admissionObserved = false; + for (let attempt = 0; attempt < 100 && !admissionObserved; attempt += 1) { + try { + admissionObserved = (await readFile(reloadJournalPath, "utf8")).includes("ATTEMPT_EXECUTION_ADMITTED"); + } catch { + // The coordinator has not published the run directory yet. + } + if (!admissionObserved) { + await new Promise((resolvePromise) => setTimeout(resolvePromise, 5)); + } + } + assert.equal(admissionObserved, true); + pi.shutdownHandlers[0]?.({ reason: "reload" }); + await reloadExecution; + const reloadReport: unknown = JSON.parse(await readFile( + join(repositoryRoot, ".git", "autopilot", "runs", reloadRunId, "reports", "status.json"), + "utf8", + )); + assert.ok(typeof reloadReport === "object" && reloadReport !== null); + assert.equal((reloadReport as Record).state, "WAITING"); + assert.match(JSON.stringify(reloadReport), /execution-unknown/); + assert.equal((await readFile(reloadJournalPath, "utf8")).match(/ATTEMPT_STARTED/g)?.length, 1); } finally { if (previousAgentDirectory === undefined) { delete process.env.PI_CODING_AGENT_DIR; } else { process.env.PI_CODING_AGENT_DIR = previousAgentDirectory; } - process.env.PATH = previousPath; - if (previousMarker === undefined) { - delete process.env.AUTOPILOT_PI_ARGUMENTS; - } else { - process.env.AUTOPILOT_PI_ARGUMENTS = previousMarker; - } - if (previousStatus === undefined) { - delete process.env.AUTOPILOT_PI_STATUS; - } else { - process.env.AUTOPILOT_PI_STATUS = previousStatus; - } } }); From ed7a4eb09d2df984e23a82a3627435183e93dff2 Mon Sep 17 00:00:00 2001 From: Denys Rafael Date: Mon, 31 Aug 2026 19:03:02 +0300 Subject: [PATCH 2/2] Stabilize Pi reload test on Windows --- skills/autopilot/runtime/test/pi-subagents.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/skills/autopilot/runtime/test/pi-subagents.test.ts b/skills/autopilot/runtime/test/pi-subagents.test.ts index 7a01ef9..7f5439b 100644 --- a/skills/autopilot/runtime/test/pi-subagents.test.ts +++ b/skills/autopilot/runtime/test/pi-subagents.test.ts @@ -457,14 +457,14 @@ test("Pi extension invokes the runtime core and completes through process-local "events.jsonl", ); let admissionObserved = false; - for (let attempt = 0; attempt < 100 && !admissionObserved; attempt += 1) { + for (let attempt = 0; attempt < 1_000 && !admissionObserved; attempt += 1) { try { admissionObserved = (await readFile(reloadJournalPath, "utf8")).includes("ATTEMPT_EXECUTION_ADMITTED"); } catch { // The coordinator has not published the run directory yet. } if (!admissionObserved) { - await new Promise((resolvePromise) => setTimeout(resolvePromise, 5)); + await new Promise((resolvePromise) => setTimeout(resolvePromise, 20)); } } assert.equal(admissionObserved, true);