From cdc536d563b5a0800afddd72f2595f23e7c42566 Mon Sep 17 00:00:00 2001 From: AmbitiousRealism2025 Date: Wed, 27 May 2026 16:36:14 -0400 Subject: [PATCH 1/3] Add Pi provider integration --- .gitignore | 1 + ATTEMPTS.md | 241 ++++++++ PI_PARITY.md | 130 ++++ PLAN.md | 122 ++++ README.md | 43 +- RESEARCH.md | 189 ++++++ apps/server/src/provider/Drivers/PiDriver.ts | 117 ++++ apps/server/src/provider/Layers/PiAdapter.ts | 566 ++++++++++++++++++ apps/server/src/provider/Layers/PiProvider.ts | 365 +++++++++++ apps/server/src/provider/Layers/PiRpc.test.ts | 87 +++ apps/server/src/provider/Layers/PiRpc.ts | 356 +++++++++++ .../provider/Layers/ProviderRegistry.test.ts | 1 + apps/server/src/provider/builtInDrivers.ts | 5 +- .../src/textGeneration/PiTextGeneration.ts | 130 ++++ .../components/KeybindingsToast.browser.tsx | 5 + .../settings/AddProviderInstanceDialog.tsx | 7 +- .../components/settings/providerDriverMeta.ts | 9 +- packages/contracts/src/model.ts | 5 + packages/contracts/src/settings.ts | 35 ++ 19 files changed, 2403 insertions(+), 11 deletions(-) create mode 100644 ATTEMPTS.md create mode 100644 PI_PARITY.md create mode 100644 PLAN.md create mode 100644 RESEARCH.md create mode 100644 apps/server/src/provider/Drivers/PiDriver.ts create mode 100644 apps/server/src/provider/Layers/PiAdapter.ts create mode 100644 apps/server/src/provider/Layers/PiProvider.ts create mode 100644 apps/server/src/provider/Layers/PiRpc.test.ts create mode 100644 apps/server/src/provider/Layers/PiRpc.ts create mode 100644 apps/server/src/textGeneration/PiTextGeneration.ts diff --git a/.gitignore b/.gitignore index 5e941c7b9f0..3fac1e5a55b 100644 --- a/.gitignore +++ b/.gitignore @@ -24,5 +24,6 @@ __screenshots__/ squashfs-root/ .vercel .gstack/ +.pi/ dist-electron/ .electron-runtime/ diff --git a/ATTEMPTS.md b/ATTEMPTS.md new file mode 100644 index 00000000000..9a708bf7da4 --- /dev/null +++ b/ATTEMPTS.md @@ -0,0 +1,241 @@ +# Pi Provider Attempts + +Date: 2026-05-27 +Branch: `codex/pi-provider` + +## Loop 0: Repository And Pre-Edit Research + +Edit: + +- Cloned latest `pingdotgg/t3code` into `/Users/ambrealismwork/Desktop/coding-projects/pi-3-code-project`. +- Created local branch `codex/pi-provider`. +- Cloned Pi and the prior POC read-only into `/tmp/pi-provider-research`. +- Wrote `RESEARCH.md` and `PLAN.md` before implementation edits. + +Checks: + +- `git status --short --branch` +- `command -v pi` +- `/opt/homebrew/bin/pi --version` +- `/opt/homebrew/bin/pi --help` +- sanitized inspection of `~/.pi/agent/settings.json`, `models.json`, and auth provider names/types only +- Pi RPC probe for `get_available_models` and `get_commands` + +Result: + +- Current branch is `codex/pi-provider`. +- Local Pi executable exists at `/opt/homebrew/bin/pi`, version `0.75.5`. +- Existing Pi config is readable at `~/.pi/agent`. +- Pi RPC model discovery works locally. +- No implementation files were edited before `RESEARCH.md` and `PLAN.md`. + +Lesson: + +- Current T3 requires a provider-driver/provider-instance integration. The prior POC is useful for RPC/event behavior but stale for architecture and config handling. + +## Loop 1: Contracts And Client Provider Metadata + +Edit: + +- Added a `pi` driver kind to provider model defaults and display names. +- Added `PiSettings` with `enabled`, `binaryPath`, and hidden `customModels`. +- Added `providers.pi` to legacy provider settings hydration and settings patch schema. +- Added Pi to browser provider client definitions with the existing `PiAgentIcon`. +- Removed the disabled "Pi Agent" coming-soon entry from the add-provider dialog. + +Checks: + +- Pending with the server driver slice so typecheck sees a registered implementation. + +Result: + +- Pi is represented in contracts/settings and in client metadata. + +Lesson: + +- The first selectable-provider step is low-risk, but server registration must follow before meaningful runtime verification. + +## Loop 2: Server Driver, Snapshot, Adapter, And Typecheck + +Edit: + +- Added `PiDriver` and registered it in `BUILT_IN_DRIVERS`. +- Added a Pi provider snapshot probe that runs `pi --version`, reads local Pi settings for ordering, and uses Pi RPC `get_available_models` / `get_commands`. +- Added a Pi adapter that starts T3 provider sessions, stores Pi session files under T3 state, leaves Pi config/auth at the existing Pi default, and sends turns through `pi --mode rpc`. +- Added Pi text-generation support through one-shot Pi RPC prompts for titles, branch names, commit messages, and PR content. +- Added fixture support for `providers.pi`. + +Checks: + +- `bun install` +- `bun typecheck` + +Result: + +- First typecheck failed because dependencies were missing (`turbo: command not found`). +- `bun install` succeeded. +- Second typecheck found a missing web fixture `providers.pi` entry and several typed Effect/server diagnostics in the new Pi files. +- Patched the fixture, fixed Effect function annotations, mapped promise errors, removed a readonly turn-array mutation, and suppressed narrowly scoped diagnostics for unavoidable Node child-process/JSONL glue. +- Final `bun typecheck` passed: 13 successful tasks. + +Lesson: + +- T3's provider-driver contract accepted the Pi slice cleanly once the instance/settings shape and Effect error channels were made explicit. The remaining verification risk is runtime behavior, not static typing. + +## Loop 3: Formatting, Linting, And Tests + +Edit: + +- Ran formatting across the workspace. +- Updated the built-in provider registry test expectation to include the new Pi driver. +- Removed lint warnings introduced by the Pi glue code. + +Checks: + +- `bun fmt` +- `bun lint` +- `bun run test` +- `bunx vitest run src/provider/Layers/ProviderRegistry.test.ts` from `apps/server` +- `bun typecheck` + +Result: + +- `bun fmt` completed and formatted the edited files. +- `bun lint` passed. It still reports pre-existing warnings outside the Pi provider slice. +- First `bun run test` failed only because `ProviderRegistry.test.ts` expected the previous built-in driver set. +- Targeted provider registry test passed after the expectation update: 33 tests passed. +- Full `bun run test` passed after the patch: 13 successful tasks, including the server and web test suites. +- Final `bun typecheck` passed: 13 successful tasks. + +Lesson: + +- The only test expectation that needed updating was the explicit built-in provider list. Existing provider behavior remained covered by the surrounding suites. + +## Loop 4: Local Runtime Provider Snapshot + +Edit: + +- No implementation edit. +- Started T3 locally from `codex/pi-provider`. + +Checks: + +- `bun run dev` using the default T3 state. +- `T3CODE_HOME=/tmp/t3-pi-provider-verify T3CODE_DEV_INSTANCE=piverify bun run dev` +- WebSocket RPC `serverRefreshProviders` against the isolated dev server. + +Result: + +- The default `~/.t3` run hit an existing local migration/state issue unrelated to Pi: migration `24_BackfillProjectionThreadShellSummary` failed with `no such column: latest_turn_plan.implemented_at`. +- The isolated verification home started successfully on server port `15401` and web port `7361`. +- Provider refresh returned five built-in provider instances: `claudeAgent`, `codex`, `cursor`, `opencode`, and `pi`. +- Pi snapshot status was `ready`, installed/authenticated, version `0.75.5`. +- Pi exposed 63 model options and 71 slash commands. +- The first visible Pi models were ordered from local Pi preferences, starting with `openai-codex/gpt-5.5`. +- The provider message confirmed use of existing config at `/Users/ambrealismwork/.pi/agent`. + +Lesson: + +- The Pi provider integration works against a clean current T3 state. The existing default user-data migration error is a local-state blocker for that state directory, not a Pi integration blocker. + +## Loop 5: Basic Prompt Through T3 Into Pi + +Edit: + +- No implementation edit. +- Refreshed the isolated dev-server auth token after an initial stale-token WebSocket open failed. +- Added `.pi/` to `.gitignore` and removed generated local Pi runtime feed files after prompt verification. + +Checks: + +- WebSocket RPC `project.create`. +- WebSocket RPC `thread.turn.start` with `modelSelection.instanceId = "pi"` and model `openai-codex/gpt-5.5`. +- Polled the T3-owned Pi session file under `/tmp/t3-pi-provider-verify/dev/providers/pi/sessions`. + +Result: + +- The first WebSocket open used a stale token and failed before dispatch. +- That rejected socket also exposed an unrelated dev-server fragility: the watched server process threw an unhandled `ECONNRESET`. Restarting the isolated dev server fixed the verification environment. +- The second run dispatched successfully with sequence `7`. +- T3 created a Pi session file for the verification thread. +- The session file contained the exact assistant response text: `T3 Pi provider verification OK`. +- Pi also created project-local `.pi/` messenger feed files during verification; these are now ignored and were removed from the working tree. + +Lesson: + +- A basic prompt now travels through T3 orchestration with Pi selected and returns a Pi assistant response. Token freshness matters for the verification script because WebSocket tokens are short-lived. + +## Loop 6: Browser Smoke Attempt + +Edit: + +- No implementation edit. + +Checks: + +- Attempted to navigate the in-app browser tool to `http://localhost:7361/`. + +Result: + +- The browser tool returned an invalid-navigation error before loading localhost. +- The stronger local verification remains the successful live T3 WebSocket orchestration prompt and provider snapshot checks. + +Lesson: + +- The browser tool path was unavailable in this run, but it did not block the required provider, model, config, slash-command, and prompt verification gates. + +## Loop 7: Pi RPC Display Cleanup + +Edit: + +- Changed `extractAssistantText` to prefer final Pi assistant `message_end` or `agent_end` text before falling back to streamed `message_update` deltas. +- Added `PiRpc.test.ts` coverage for the exact leakage shape: streamed reasoning text plus JSON tool-call arguments followed by a clean final answer. + +Checks: + +- `bunx vitest run src/provider/Layers/PiRpc.test.ts` +- `bun typecheck` +- Confirmed the dev server is still listening on port `15401` and web is still listening on port `7361`. + +Result: + +- Targeted Pi RPC test passed: 2 tests passed. +- Typecheck passed: 13 successful tasks. +- The fix prevents Pi reasoning/tool-call JSON deltas such as `{"path":...}` and `{"command":...}` from being rendered as assistant text when Pi sends a final assistant message. + +Lesson: + +- Pi RPC emits useful intermediate deltas for its own TUI, but T3's buffered provider adapter should render the final assistant message as the chat answer and keep intermediate tool metadata out of visible prose. + +## Loop 8: Pi Assistant Text Streaming + +Edit: + +- Launched two read-only research subagents: + - Pi-side stream semantics: confirmed safe visible text is only `message_update.assistantMessageEvent.type === "text_delta"`. + - T3-side runtime contract: confirmed visible streaming should use `content.delta` with `streamKind: "assistant_text"`. +- Added a Pi RPC event callback so the adapter can receive JSONL events while the prompt is still running. +- Added `readPiAssistantTextDelta` to filter only Pi `text_delta` events. +- Updated `PiAdapter` to emit `item.started` and incremental `content.delta` events as Pi text deltas arrive. +- Kept final `message_end` extraction as fallback and avoided duplicate final output when streaming already emitted the text. +- Enabled `enableAssistantStreaming` in the isolated dev-server settings used for manual testing. + +Checks: + +- `bunx vitest run src/provider/Layers/PiRpc.test.ts` +- `bun typecheck` +- `bun fmt` +- Live adapter verification against local Pi and existing `~/.pi/agent` config. + +Result: + +- Targeted Pi RPC tests passed: 3 tests passed. +- Typecheck passed: 13 successful tasks. +- Formatting passed. +- Live adapter verification emitted six separate `content.delta` events before `turn.completed` for the prompt `alpha beta gamma delta epsilon`. +- The six visible deltas were `alpha`, ` beta`, ` gamma`, ` delta`, ` epsilon`, and `.`. +- The dev server was restarted cleanly after file-watcher churn and is running with the streaming patch loaded. + +Lesson: + +- Pi can stream cleanly through T3 as long as the adapter maps only `text_delta` to `assistant_text`; reasoning and tool-call deltas must remain hidden or be mapped to non-assistant surfaces later. diff --git a/PI_PARITY.md b/PI_PARITY.md new file mode 100644 index 00000000000..6cb6d15a571 --- /dev/null +++ b/PI_PARITY.md @@ -0,0 +1,130 @@ +# Pi Slash Command And Model Parity + +Date: 2026-05-27 +Branch: `codex/pi-provider` + +This document records the implemented Pi model and slash-command behavior for the +T3 provider integration. + +## Model List Behavior Target + +- Use existing Pi config from `~/.pi/agent` by default. +- Prefer live Pi RPC `get_available_models` for available model options. +- Use local Pi `settings.json` only for default and enabled-model ordering. +- Show the configured default model first when available. +- Show configured/frequently used models before all remaining models when available. +- Visual separator row is currently blocked by T3's `ServerProviderModel` contract unless a UI/schema change is added. + +## Slash Command Sources + +Pi built-in interactive slash commands observed from source: + +- `/settings` +- `/model` +- `/scoped-models` +- `/export` +- `/import` +- `/share` +- `/copy` +- `/name` +- `/session` +- `/changelog` +- `/hotkeys` +- `/fork` +- `/clone` +- `/tree` +- `/login` +- `/logout` +- `/new` +- `/compact` +- `/resume` +- `/reload` +- `/quit` + +Pi RPC dynamic command source: + +- `get_commands` returns extension commands, prompt templates, and skills. + +## Implementation Status + +Implemented as a provider snapshot plus best-effort pass-through behavior: + +- T3 discovers the Pi executable, then runs Pi RPC probes against the existing + local Pi config/auth state. +- T3 reads `~/.pi/agent/settings.json` for default and enabled-model ordering. +- T3 does not create a separate T3-only Pi config and does not set + `PI_CODING_AGENT_DIR`. +- Pi model options are populated from live Pi RPC `get_available_models`. +- Pi slash commands are populated from a built-in command list plus dynamic Pi + RPC `get_commands`. +- Slash command execution is not implemented as native T3 control actions. + Commands inserted/sent as chat text are passed to Pi through the prompt path, + which is the broadest feasible compatibility layer without embedding the Pi + interactive TUI. + +Verified snapshot on 2026-05-27: + +- Pi executable: `/opt/homebrew/bin/pi` +- Pi version: `0.75.5` +- Pi config root used by Pi: `/Users/ambrealismwork/.pi/agent` +- T3 provider status: `ready` +- Pi model count: 63 +- Pi slash command count: 71 +- First ordered models: + - `openai-codex/gpt-5.5` + - `anthropic/claude-opus-4-6` + - `anthropic/claude-sonnet-4-6` + - `anthropic/claude-haiku-4-5` + - `openai-codex/gpt-5.2` + - `openai-codex/gpt-5.3-codex` + - `openai-codex/gpt-5.3-codex-spark` + - `openai-codex/gpt-5.4` + +Visual separator status: + +- Preferred/default/enabled models are ordered first. +- A literal separator row is not implemented because the current + `ServerProviderModel` contract represents selectable models only and has no + non-selectable separator item shape. + +## Verified Commands + +- Built-in command names are exposed in the provider snapshot: + `/settings`, `/model`, `/scoped-models`, `/export`, `/import`, `/share`, + `/copy`, `/name`, `/session`, `/changelog`, `/hotkeys`, `/fork`, `/clone`, + `/tree`, `/login`, `/logout`, `/new`, `/compact`, `/resume`, `/reload`, + `/quit`. +- Dynamic command names from Pi RPC `get_commands` are exposed in the provider + snapshot, including extension and skill commands observed locally. +- T3 prompt delivery through Pi was verified with the model + `openai-codex/gpt-5.5`; the response contained the exact expected text + `T3 Pi provider verification OK`. + +## Partial Commands + +- `/model`: native model choice is handled by T3's model selector and passed to + Pi as `--model`; textual `/model` input is passed through as prompt text. +- `/new`: native new-chat behavior is handled by T3 thread creation; textual + `/new` input is passed through as prompt text. +- `/compact`: Pi has RPC command support for compaction, but this integration + does not wire it as a special T3 control action yet. Textual `/compact` input + is passed through as prompt text. +- `/fork` and `/clone`: Pi has RPC command definitions, but exact behavior needs + Pi-specific entry-id/session context that T3 does not currently expose as a + first-class slash-command UI parameter. +- `/name` and `/session`: T3 has its own thread title/session state. Exact Pi TUI + session mutation is not mapped to those T3 controls yet. +- Extension, prompt-template, and skill slash commands are exposed from + `get_commands` and can be passed to Pi as text, but exact parity depends on + how each Pi command expands inside Pi's prompt path. + +## Blocked Commands + +- Exact interactive TUI parity is blocked for commands whose primary behavior is + opening Pi UI panels or manipulating the Pi TUI process: + `/settings`, `/scoped-models`, `/export`, `/import`, `/share`, `/copy`, + `/session`, `/changelog`, `/hotkeys`, `/tree`, `/login`, `/logout`, + `/resume`, `/reload`, `/quit`. +- These commands are still visible as Pi-supported names where appropriate, but + T3 cannot reproduce their full interactive behavior without either embedding + Pi's TUI or adding dedicated T3 UI/control surfaces for each command. diff --git a/PLAN.md b/PLAN.md new file mode 100644 index 00000000000..f24ffce71ce --- /dev/null +++ b/PLAN.md @@ -0,0 +1,122 @@ +# Pi Provider Implementation Plan + +Date: 2026-05-27 +Branch: `codex/pi-provider` + +## Milestone 1: Pre-Edit Gate + +- Inspect current T3 Code provider architecture. +- Inspect current Pi executable, config, model, RPC, and slash-command behavior without printing secrets. +- Inspect prior POC for reusable and stale patterns. +- Write `RESEARCH.md` and this `PLAN.md` before implementation file edits. + +Status: complete. + +## Milestone 2: Contracts And Settings + +- Add `PI_DRIVER_KIND` defaults in `packages/contracts/src/model.ts`. +- Add `PiSettings` in `packages/contracts/src/settings.ts`. +- Add Pi to legacy `providers` settings hydration so a default `pi` provider instance can be auto-created. +- Add Pi patch schema support for settings updates. + +Verification: + +- Contract typecheck or targeted schema test. +- Existing provider defaults remain unchanged for Codex, Claude, Cursor, and OpenCode. + +Status: complete. + +## Milestone 3: Server Driver And Snapshot + +- Add `apps/server/src/provider/Drivers/PiDriver.ts`. +- Add a Pi provider snapshot layer/helper that: + - finds the Pi binary, + - runs `pi --version`, + - probes local Pi model state with RPC `get_available_models`, + - reads sanitized `~/.pi/agent/settings.json` only for ordering/defaults, + - exposes models through `ServerProvider.models`, + - exposes supported Pi slash commands through `ServerProvider.slashCommands`. +- Register Pi in `apps/server/src/provider/builtInDrivers.ts`. + +Verification: + +- Provider snapshots include Pi. +- Pi models appear when local Pi config is readable. +- Preferred/default/enabled models are ordered first when possible. + +Status: complete. + +## Milestone 4: Pi Adapter + +- Add a Pi adapter that satisfies `ProviderAdapterShape`. +- Use `pi --mode rpc` for prompt turns. +- Use T3-owned session files for T3 conversations while leaving Pi config/auth under `~/.pi/agent`. +- Map Pi JSONL events into canonical T3 runtime events. +- Support cancellation by sending RPC `abort` and terminating the child process when needed. +- Preserve behavior for all other providers. + +Verification: + +- A T3 session can start with provider instance `pi`. +- A basic prompt can be sent through Pi and returns assistant text. +- Interrupt/cleanup paths do not leak child processes in normal completion. + +Status: complete. + +## Milestone 5: Web Provider Option + +- Add Pi to client provider driver metadata. +- Remove or replace the disabled "Pi Agent" coming-soon entry. +- Confirm Pi appears as a selectable provider option in settings/model selection. + +Verification: + +- Local web app shows Pi provider option. +- Model selector receives Pi models from server snapshot. + +Status: complete. + +## Milestone 6: Slash Command Parity + +- Test dynamic Pi commands via RPC `get_commands`. +- Add best-effort slash command exposure and/or mappings. +- Document implemented, partial, and blocked commands in `PI_PARITY.md`. + +Verification: + +- Slash command menu shows Pi-supported entries where feasible. +- Unsupported interactive-only commands are explicitly documented, not silently claimed. + +Status: complete. + +## Milestone 7: Checks And Local Verification + +- Run fastest targeted checks after each coherent edit and record in `ATTEMPTS.md`. +- Run required project gates from `AGENTS.md`: + - `bun fmt` + - `bun lint` + - `bun typecheck` + - `bun run test` +- Start T3 locally from `codex/pi-provider`. +- Verify: + - Pi appears as a provider option. + - T3 reads existing `~/.pi/agent` config. + - Pi model options appear in selector/snapshot. +- A basic prompt reaches Pi and returns a response. +- Slash-command behavior is tested or manually verified. + +Status: complete. + +## Stop Criteria + +Stop only when T3 runs locally with Pi selectable, existing Pi config readable, Pi models visible, and a basic Pi prompt verified. + +Status: met using isolated verification state at `/tmp/t3-pi-provider-verify`. + +If blocked, stop with: + +- exact blocker, +- evidence, +- attempted fixes, +- current branch state, +- next decision needed. diff --git a/README.md b/README.md index c439743cea5..2a5247368aa 100644 --- a/README.md +++ b/README.md @@ -1,16 +1,47 @@ -# T3 Code +# T3 Code + Pi -T3 Code is a minimal web GUI for coding agents (currently Codex, Claude, and OpenCode, more coming soon). +T3 Code is a minimal web GUI for coding agents. This branch adds Pi as a local +CLI/TUI-backed provider alongside Codex, Claude, Cursor, and OpenCode. ## Installation > [!WARNING] -> T3 Code currently supports Codex, Claude, and OpenCode. +> T3 Code currently supports Codex, Claude, Cursor, OpenCode, and Pi. > Install and authenticate at least one provider before use: > > - Codex: install [Codex CLI](https://developers.openai.com/codex/cli) and run `codex login` > - Claude: install [Claude Code](https://claude.com/product/claude-code) and run `claude auth login` > - OpenCode: install [OpenCode](https://opencode.ai) and run `opencode auth login` +> - Pi: install the Pi CLI and authenticate/configure it normally. T3 reads Pi's existing `~/.pi/agent` config. + +## Pi Provider + +Pi support in this branch is an executable integration, not a generic API key +provider. T3 discovers the local `pi` binary, reads the existing Pi configuration +under `~/.pi/agent`, and uses Pi RPC mode for model discovery, prompts, and +session execution. + +Current Pi behavior: + +- Pi appears as a selectable provider in T3. +- Model options are populated from live Pi RPC discovery. +- The configured/default Pi model is ordered first when available. +- Assistant text streams from Pi `text_delta` events when T3 assistant streaming is enabled. +- Pi reasoning and tool-call RPC events are filtered out of visible assistant text. +- Pi slash commands are exposed on a best-effort basis; exact TUI-only command parity is documented in [PI_PARITY.md](./PI_PARITY.md). + +For local development against an existing Pi setup: + +```bash +bun install +bun run dev +``` + +If your default T3 state has old local migrations, use an isolated test home: + +```bash +T3CODE_HOME=/tmp/t3-pi-provider-verify T3CODE_DEV_INSTANCE=piverify bun run dev +``` ### Run without installing @@ -48,6 +79,12 @@ We are not accepting contributions yet. Observability guide: [docs/observability.md](./docs/observability.md) +Pi integration notes: + +- Research and implementation plan: [RESEARCH.md](./RESEARCH.md), [PLAN.md](./PLAN.md) +- Edit/test loop log: [ATTEMPTS.md](./ATTEMPTS.md) +- Slash-command and model parity: [PI_PARITY.md](./PI_PARITY.md) + ## If you REALLY want to contribute still.... read this first Before local development, prepare the environment and install dependencies: diff --git a/RESEARCH.md b/RESEARCH.md new file mode 100644 index 00000000000..1ce9bd5f55f --- /dev/null +++ b/RESEARCH.md @@ -0,0 +1,189 @@ +# Pi Provider Research + +Date: 2026-05-27 +Branch: `codex/pi-provider` +T3 Code base: `pingdotgg/t3code` at `4f0f24f0` (`fix: maintain reasoning selections for multiple providers (#2760)`) + +## Scope And Sources + +- Current T3 Code repository in `/Users/ambrealismwork/Desktop/coding-projects/pi-3-code-project`. +- Current Pi source cloned read-only at `/tmp/pi-provider-research/pi`. +- Prior proof-of-concept cloned read-only at `/tmp/pi-provider-research/poc`. +- Local Pi executable and sanitized local Pi config under `~/.pi/agent`. Secret values were not printed. +- Subagent read-only inspections covered current T3 provider architecture, Pi CLI/config/slash behavior, and prior POC reuse risk. + +## Current T3 Provider Architecture + +T3 Code now uses an open provider-driver and provider-instance architecture. Driver kinds are branded slugs, not a closed enum, and routing is by `ProviderInstanceId`. + +Key files: + +- `apps/server/src/provider/ProviderDriver.ts`: the driver SPI. A driver exposes `driverKind`, `metadata`, `configSchema`, `defaultConfig`, and `create()`. `create()` returns a scoped `ProviderInstance` containing `snapshot`, `adapter`, and `textGeneration`. +- `apps/server/src/provider/builtInDrivers.ts`: static built-in driver registry. Current entries are Codex, Claude, Cursor, and OpenCode. +- `apps/server/src/provider/Layers/ProviderInstanceRegistryHydration.ts`: hydrates configured instances from `settings.providerInstances` plus legacy `settings.providers.`. +- `apps/server/src/provider/Layers/ProviderInstanceRegistryLive.ts`: decodes config envelopes, calls `driver.create()`, owns per-instance scopes, and creates unavailable shadows for unknown or invalid drivers. +- `apps/server/src/provider/Layers/ProviderRegistry.ts`: aggregates per-instance snapshots for UI and status. +- `apps/server/src/provider/Layers/ProviderAdapterRegistry.ts`: resolves adapters dynamically by provider instance id. +- `apps/server/src/provider/Layers/ProviderService.ts`: cross-provider facade for session start, turn send, interrupt, rollback, and runtime event streaming. +- `apps/server/src/provider/Services/ProviderAdapter.ts`: adapter contract that Pi must satisfy. + +Model selection is instance-scoped. `packages/contracts/src/orchestration.ts` defines `ModelSelection` as `{ instanceId, model, options? }`; legacy `{ provider, model }` decodes into the default instance id. Runtime start/send paths pass `modelSelection` through `ProviderCommandReactor` into `ProviderService`. + +Provider snapshots use `packages/contracts/src/server.ts` `ServerProvider`, including: + +- `instanceId` +- `driver` +- `enabled`, `installed`, `version`, `status`, `auth` +- `models: ServerProviderModel[]` +- `slashCommands: ServerProviderSlashCommand[]` + +There is no native separator row in `ServerProviderModel`. Preferred model ordering is feasible; an actual visual separator would require a contract/UI change or an unsafe fake model entry. + +Provider settings currently live in `packages/contracts/src/settings.ts` while the migration to opaque per-driver config is incomplete. Existing provider schemas include executable paths and auth/config roots where needed. Adding Pi should follow this pattern with a small schema and no copied Pi secrets. + +## Current Pi CLI, Config, Models, And RPC + +Installed local Pi: + +- Executable: `/opt/homebrew/bin/pi` +- Version: `0.75.5` +- Symlink target: `/opt/homebrew/lib/node_modules/@earendil-works/pi-coding-agent/dist/cli.js` + +Current source package: + +- `/tmp/pi-provider-research/pi/packages/coding-agent/package.json` maps bin `pi` to `dist/cli.js`. +- Important CLI flags parsed in `/tmp/pi-provider-research/pi/packages/coding-agent/src/cli/args.ts`: + - `--mode text|json|rpc` + - `--print`, `-p` + - `--provider`, `--model`, `--models` + - `--session`, `--continue`, `--resume`, `--fork`, `--session-dir`, `--no-session` + - `--tools`, `--no-tools`, `--no-builtin-tools` + - `--extension`, `--skill`, `--prompt-template`, `--theme`, plus corresponding `--no-*` + - `--thinking off|minimal|low|medium|high|xhigh` + - `--list-models`, `--offline` + +Pi config: + +- Default config root: `~/.pi/agent`, overridable by `PI_CODING_AGENT_DIR`. +- Local settings file: `/Users/ambrealismwork/.pi/agent/settings.json`. +- Local model file: `/Users/ambrealismwork/.pi/agent/models.json`. +- Local auth file: `/Users/ambrealismwork/.pi/agent/auth.json`. +- Project settings may exist at `/.pi/settings.json` and override global settings. + +Sanitized local settings observed: + +- `defaultProvider: openai-codex` +- `defaultModel: gpt-5.5` +- `defaultThinkingLevel: low` +- `transport: websocket` +- `enabledModels`: 22 configured model patterns +- packages/extensions are configured + +Sanitized local auth providers observed by type only: + +- `openai-codex`: OAuth +- `anthropic`: OAuth +- `google-antigravity`: OAuth +- `opencode-go`: API key + +Pi model behavior: + +- Models are loaded from built-ins plus `~/.pi/agent/models.json`. +- Custom providers observed locally include `dashscope`, `kilo`, `openai-codex`, `qwen36`, `stepfun`, `zai-coding`, `zai-glm47-flash`, and `zai-glm5-turbo`. +- `pi --list-models` lists auth-available models. +- RPC `get_available_models` returns a structured live model list and was verified locally without printing secrets. +- Best T3 model source is Pi RPC `get_available_models`; fallback can parse `~/.pi/agent/settings.json` and `models.json` only if RPC discovery fails. + +Pi noninteractive modes: + +- `pi -p "prompt"` returns final text. +- `pi --mode json "prompt"` emits JSONL events. +- `pi --mode rpc` is bidirectional JSONL over stdin/stdout. This is the best adapter seam because it supports prompt, model changes, model discovery, session state, bash, compact, export, session switching, and command discovery. + +Pi RPC commands from source include: + +- `prompt`, `steer`, `follow_up`, `abort` +- `new_session`, `get_state`, `set_model`, `cycle_model`, `get_available_models` +- `set_thinking_level`, `cycle_thinking_level` +- `compact`, `bash`, `abort_bash` +- `get_session_stats`, `export_html`, `switch_session`, `fork`, `clone` +- `get_last_assistant_text`, `set_session_name`, `get_messages`, `get_commands` + +## Pi Slash Commands + +Pi built-in interactive slash command autocomplete includes: + +- `/settings` +- `/model` +- `/scoped-models` +- `/export` +- `/import` +- `/share` +- `/copy` +- `/name` +- `/session` +- `/changelog` +- `/hotkeys` +- `/fork` +- `/clone` +- `/tree` +- `/login` +- `/logout` +- `/new` +- `/compact` +- `/resume` +- `/reload` +- `/quit` + +Source findings: + +- Built-in interactive command handling lives in Pi interactive mode. +- Extension slash commands run before prompt expansion and can execute while streaming. +- Prompt templates expand `/template args`. +- Skill commands expand `/skill:name args`. +- RPC `get_commands` returns extension, prompt-template, and skill commands, but not the full interactive built-in command list. + +Integration implication: T3 can surface dynamic Pi commands discovered by RPC and can map some built-ins to RPC commands (`/model`, `/compact`, `/fork`, `/clone`, `/name`, `/session`, `/export`, bash-like commands). Exact interactive TUI parity is not guaranteed because several built-ins are UI-local or auth/session-manager affordances with no direct RPC equivalent. + +## Prior POC Findings + +Prior POC repository: `https://github.com/AmbitiousRealism2025/t3-code-atreides-pi-edition.git`. + +Reusable patterns: + +- `apps/server/src/provider/Layers/PiAdapter.ts` has useful JSONL RPC subprocess handling and event mapping. +- It correctly waits for Pi `agent_end` rather than finalizing on `turn_end`, because Pi can run multiple internal turns for one user request. +- It stores per-T3 session files and uses `--session` / `--continue` to preserve Pi conversation state. +- It maps Pi event stream concepts into T3 runtime events. +- `PiModelDiscoveryLive.ts` demonstrates the need for live model discovery, though its implementation path is stale. + +Stale or risky patterns: + +- The POC used an old provider manifest registry (`packages/contracts/src/providers/pi.ts`), which current T3 no longer uses. +- It used singleton provider-kind adapter routing, while current T3 routes by provider instance id. +- It added direct `/api/provider/pi/models` HTTP endpoints, while current T3 expects models through `ServerProvider.models` snapshots. +- It used stale start/send inputs (`model`, `modelOptions`) instead of current `modelSelection`. +- It depended on an older package name/version (`@mariozechner/pi-coding-agent`) while current Pi is `@earendil-works/pi-coding-agent` `0.75.5`. +- It copied or symlinked `auth.json`, `models.json`, and sanitized `settings.json` into a T3-owned Pi runtime directory. That conflicts with this task's requirement to read existing `~/.pi/` config instead of creating separate T3-only Pi config. +- It disabled extensions, skills, prompt templates, and themes for some paths, which would reduce slash-command parity. + +## Integration Risks + +- Pi is a CLI/TUI executable with structured RPC, not a generic hosted API provider. T3 needs a subprocess adapter, not a plain HTTP client. +- Pi config and auth must be read from the existing Pi config root. The adapter must not copy or print secrets. +- Local model lists may depend on auth state, custom model config, and project settings. +- Preferred-model grouping is partly supported by ordering. A visual separator is not supported by current T3 model contracts without UI/schema work. +- Extension UI requests, auth flows, and some interactive slash commands may not map cleanly to T3. +- Long-running provider subprocesses need cancellation and cleanup. +- Full verification may spend real local Pi account quota/tokens. + +## Proposed Current Integration Shape + +- Add a first-class `pi` `ProviderDriver`. +- Register it in `BUILT_IN_DRIVERS`. +- Add `PiSettings` with `enabled`, `binaryPath`, and hidden `customModels`; do not add a separate T3 Pi config root. +- Leave `PI_CODING_AGENT_DIR` unset by default so Pi reads its existing default `~/.pi/agent`; allow the provider instance environment to override it if a user has already configured that globally. +- Discover the Pi executable from settings first, then PATH. +- Build provider snapshots from live Pi RPC `get_available_models`, ordered by local Pi defaults and `enabledModels`. +- Implement prompt execution with `pi --mode rpc`, session files owned by T3 but config/auth owned by Pi. +- Surface Pi dynamic slash commands from RPC where available, add documented built-in mappings where safe, and document partial/blocked parity in `PI_PARITY.md`. diff --git a/apps/server/src/provider/Drivers/PiDriver.ts b/apps/server/src/provider/Drivers/PiDriver.ts new file mode 100644 index 00000000000..2bd3461dd79 --- /dev/null +++ b/apps/server/src/provider/Drivers/PiDriver.ts @@ -0,0 +1,117 @@ +import { PiSettings, ProviderDriverKind, type ServerProvider } from "@t3tools/contracts"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; + +import { ServerConfig } from "../../config.ts"; +import { makePiTextGeneration } from "../../textGeneration/PiTextGeneration.ts"; +import { ProviderDriverError } from "../Errors.ts"; +import { makePiAdapter } from "../Layers/PiAdapter.ts"; +import { checkPiProviderStatus, makePendingPiProvider } from "../Layers/PiProvider.ts"; +import { makeManagedServerProvider } from "../makeManagedServerProvider.ts"; +import { + defaultProviderContinuationIdentity, + type ProviderDriver, + type ProviderInstance, +} from "../ProviderDriver.ts"; +import type { ServerProviderDraft } from "../providerSnapshot.ts"; +import { mergeProviderInstanceEnvironment } from "../ProviderInstanceEnvironment.ts"; +import { makeManualOnlyProviderMaintenanceCapabilities } from "../providerMaintenance.ts"; + +const decodePiSettings = Schema.decodeSync(PiSettings); + +const DRIVER_KIND = ProviderDriverKind.make("pi"); +const SNAPSHOT_REFRESH_INTERVAL = Duration.minutes(5); + +export type PiDriverEnv = ServerConfig; + +const withInstanceIdentity = + (input: { + readonly instanceId: ProviderInstance["instanceId"]; + readonly displayName: string | undefined; + readonly accentColor: string | undefined; + readonly continuationGroupKey: string; + }) => + (snapshot: ServerProviderDraft): ServerProvider => ({ + ...snapshot, + instanceId: input.instanceId, + driver: DRIVER_KIND, + ...(input.displayName ? { displayName: input.displayName } : {}), + ...(input.accentColor ? { accentColor: input.accentColor } : {}), + continuation: { groupKey: input.continuationGroupKey }, + }); + +export const PiDriver: ProviderDriver = { + driverKind: DRIVER_KIND, + metadata: { + displayName: "Pi", + supportsMultipleInstances: true, + }, + configSchema: PiSettings, + defaultConfig: (): PiSettings => decodePiSettings({}), + create: ({ instanceId, displayName, accentColor, environment, enabled, config }) => + Effect.gen(function* () { + const serverConfig = yield* ServerConfig; + const processEnv = mergeProviderInstanceEnvironment(environment); + const continuationIdentity = defaultProviderContinuationIdentity({ + driverKind: DRIVER_KIND, + instanceId, + }); + const stampIdentity = withInstanceIdentity({ + instanceId, + displayName, + accentColor, + continuationGroupKey: continuationIdentity.continuationKey, + }); + const effectiveConfig = { ...config, enabled } satisfies PiSettings; + const maintenanceCapabilities = makeManualOnlyProviderMaintenanceCapabilities({ + provider: DRIVER_KIND, + packageName: "@earendil-works/pi-coding-agent", + }); + + const adapter = yield* makePiAdapter(effectiveConfig, { + instanceId, + environment: processEnv, + }); + const textGeneration = yield* makePiTextGeneration(effectiveConfig, processEnv); + const checkProvider = checkPiProviderStatus( + effectiveConfig, + serverConfig.cwd, + processEnv, + ).pipe(Effect.map(stampIdentity)); + + const snapshot = yield* makeManagedServerProvider({ + maintenanceCapabilities, + getSettings: Effect.succeed(effectiveConfig), + streamSettings: Stream.never, + haveSettingsChanged: () => false, + initialSnapshot: (settings) => + makePendingPiProvider(settings).pipe(Effect.map(stampIdentity)), + checkProvider, + refreshInterval: SNAPSHOT_REFRESH_INTERVAL, + }).pipe( + Effect.mapError( + (cause) => + new ProviderDriverError({ + driver: DRIVER_KIND, + instanceId, + detail: `Failed to build Pi snapshot: ${cause.message ?? String(cause)}`, + cause, + }), + ), + ); + + return { + instanceId, + driverKind: DRIVER_KIND, + continuationIdentity, + displayName, + accentColor, + enabled, + snapshot, + adapter, + textGeneration, + } satisfies ProviderInstance; + }), +}; diff --git a/apps/server/src/provider/Layers/PiAdapter.ts b/apps/server/src/provider/Layers/PiAdapter.ts new file mode 100644 index 00000000000..86e19a6f42f --- /dev/null +++ b/apps/server/src/provider/Layers/PiAdapter.ts @@ -0,0 +1,566 @@ +import { + EventId, + ProviderDriverKind, + ProviderInstanceId, + RuntimeItemId, + ThreadId, + TurnId, + type ChatAttachment, + type PiSettings, + type ProviderRuntimeEvent, + type ProviderSendTurnInput, + type ProviderSession, +} from "@t3tools/contracts"; +import { getModelSelectionStringOptionValue } from "@t3tools/shared/model"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Queue from "effect/Queue"; +import * as Random from "effect/Random"; +import * as Stream from "effect/Stream"; +import { randomUUID } from "node:crypto"; +// @effect-diagnostics-next-line nodeBuiltinImport:off +import { readFile } from "node:fs/promises"; +// @effect-diagnostics-next-line nodeBuiltinImport:off +import { mkdir, stat } from "node:fs/promises"; +// @effect-diagnostics-next-line nodeBuiltinImport:off +import { join } from "node:path"; + +import { resolveAttachmentPath } from "../../attachmentStore.ts"; +import { ServerConfig } from "../../config.ts"; +import { + ProviderAdapterProcessError, + ProviderAdapterRequestError, + ProviderAdapterSessionNotFoundError, + ProviderAdapterValidationError, +} from "../Errors.ts"; +import type { ProviderAdapterShape, ProviderThreadSnapshot } from "../Services/ProviderAdapter.ts"; +import { readPiAssistantTextDelta, runPiRpcPrompt, splitPiModelSlug } from "./PiRpc.ts"; + +const PROVIDER = ProviderDriverKind.make("pi"); + +type PiAdapterShape = ProviderAdapterShape< + | ProviderAdapterProcessError + | ProviderAdapterRequestError + | ProviderAdapterSessionNotFoundError + | ProviderAdapterValidationError +>; + +interface PiSessionContext { + session: ProviderSession; + readonly sessionPath: string; + activeAbort: AbortController | null; + turns: Array; +} + +const nowIso = Effect.map(DateTime.now, DateTime.formatIso); + +const sanitizePathPart = (value: string): string => value.replace(/[^a-zA-Z0-9._-]/g, "_"); + +function makePiArgs(input: { + readonly sessionPath: string; + readonly model: string | undefined; + readonly thinkingLevel: string | undefined; + readonly sessionExists: boolean; +}): ReadonlyArray { + const args = ["--session", input.sessionPath]; + if (input.sessionExists) args.push("--continue"); + if (input.model) args.push("--model", input.model); + if (input.thinkingLevel) args.push("--thinking", input.thinkingLevel); + return args; +} + +function errorDetail(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +export const makePiAdapter = Effect.fn("makePiAdapter")(function* ( + piSettings: PiSettings, + options: { + readonly instanceId?: ProviderInstanceId | undefined; + readonly environment?: NodeJS.ProcessEnv | undefined; + } = {}, +) { + const serverConfig = yield* ServerConfig; + const boundInstanceId = options.instanceId ?? ProviderInstanceId.make("pi"); + const environment = options.environment ?? process.env; + const runtimeEvents = yield* Queue.unbounded(); + const sessions = new Map(); + + const emit = (event: ProviderRuntimeEvent) => + Queue.offer(runtimeEvents, event).pipe(Effect.asVoid); + + const eventBase = (input: { + readonly threadId: ThreadId; + readonly turnId?: TurnId | undefined; + readonly itemId?: string | undefined; + }) => + Effect.gen(function* () { + const uuid = yield* Random.nextUUIDv4; + const createdAt = yield* nowIso; + return { + eventId: EventId.make(uuid), + provider: PROVIDER, + providerInstanceId: boundInstanceId, + threadId: input.threadId, + createdAt, + ...(input.turnId ? { turnId: input.turnId } : {}), + ...(input.itemId ? { itemId: RuntimeItemId.make(input.itemId) } : {}), + }; + }); + + const eventBaseSync = (input: { + readonly threadId: ThreadId; + readonly createdAt: string; + readonly turnId?: TurnId | undefined; + readonly itemId?: string | undefined; + }) => ({ + eventId: EventId.make(randomUUID()), + provider: PROVIDER, + providerInstanceId: boundInstanceId, + threadId: input.threadId, + createdAt: input.createdAt, + ...(input.turnId ? { turnId: input.turnId } : {}), + ...(input.itemId ? { itemId: RuntimeItemId.make(input.itemId) } : {}), + }); + + const requireSession = (threadId: ThreadId) => + Effect.sync(() => sessions.get(threadId)).pipe( + Effect.flatMap((session) => + session + ? Effect.succeed(session) + : Effect.fail(new ProviderAdapterSessionNotFoundError({ provider: PROVIDER, threadId })), + ), + ); + + const updateSession = (context: PiSessionContext, patch: Partial) => + Effect.gen(function* () { + const updatedAt = yield* nowIso; + context.session = { + ...context.session, + ...patch, + updatedAt, + }; + }); + + const resolveAttachment = (attachment: ChatAttachment) => + Effect.tryPromise({ + try: async () => { + const attachmentPath = resolveAttachmentPath({ + attachmentsDir: serverConfig.attachmentsDir, + attachment, + }); + if (!attachmentPath) { + throw new Error(`Invalid attachment id '${attachment.id}'.`); + } + const data = await readFile(attachmentPath); + return { + type: "image" as const, + data: data.toString("base64"), + mimeType: attachment.mimeType, + }; + }, + catch: (cause) => + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "prompt", + detail: errorDetail(cause), + cause, + }), + }); + + const startSession: PiAdapterShape["startSession"] = (input) => + Effect.gen(function* () { + if (input.provider !== undefined && input.provider !== PROVIDER) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "startSession", + issue: `Expected provider '${PROVIDER}' but received '${input.provider}'.`, + }); + } + const existing = sessions.get(input.threadId); + if (existing) { + existing.activeAbort?.abort(); + sessions.delete(input.threadId); + } + + const createdAt = yield* nowIso; + const sessionDir = join(serverConfig.stateDir, "providers", "pi", "sessions"); + yield* Effect.tryPromise(() => mkdir(sessionDir, { recursive: true })).pipe( + Effect.mapError( + (cause) => + new ProviderAdapterProcessError({ + provider: PROVIDER, + threadId: input.threadId, + detail: `Failed to create Pi session directory: ${errorDetail(cause)}`, + cause, + }), + ), + ); + const sessionPath = join(sessionDir, `${sanitizePathPart(input.threadId)}.json`); + const session: ProviderSession = { + provider: PROVIDER, + providerInstanceId: boundInstanceId, + status: "ready", + runtimeMode: input.runtimeMode, + cwd: input.cwd ?? serverConfig.cwd, + ...(input.modelSelection?.instanceId === boundInstanceId + ? { model: input.modelSelection.model } + : {}), + threadId: input.threadId, + createdAt, + updatedAt: createdAt, + }; + sessions.set(input.threadId, { + session, + sessionPath, + activeAbort: null, + turns: [], + }); + + yield* emit({ + ...(yield* eventBase({ threadId: input.threadId })), + type: "session.started", + payload: { message: "Pi session started" }, + }); + yield* emit({ + ...(yield* eventBase({ threadId: input.threadId })), + type: "thread.started", + payload: { providerThreadId: sessionPath }, + }); + + return session; + }); + + const sendTurn: PiAdapterShape["sendTurn"] = (input: ProviderSendTurnInput) => + Effect.gen(function* () { + const context = yield* requireSession(input.threadId); + const text = input.input?.trim(); + if (!text && (input.attachments ?? []).length === 0) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: "Pi turns require text input or at least one attachment.", + }); + } + if ( + input.modelSelection !== undefined && + input.modelSelection.instanceId !== boundInstanceId + ) { + return yield* new ProviderAdapterValidationError({ + provider: PROVIDER, + operation: "sendTurn", + issue: `Pi model selection is bound to instance '${input.modelSelection.instanceId}', expected '${boundInstanceId}'.`, + }); + } + + const turnId = TurnId.make(`pi-turn-${yield* Random.nextUUIDv4}`); + const itemId = `pi-assistant-${yield* Random.nextUUIDv4}`; + const abort = new AbortController(); + context.activeAbort = abort; + const model = input.modelSelection?.model ?? context.session.model; + const thinkingLevel = getModelSelectionStringOptionValue( + input.modelSelection, + "thinkingLevel", + ); + const parsedModel = splitPiModelSlug(model); + const modelArg = parsedModel ? `${parsedModel.provider}/${parsedModel.modelId}` : model; + const sessionExists = yield* Effect.promise(() => + stat(context.sessionPath).then( + () => true, + () => false, + ), + ); + const images = yield* Effect.forEach(input.attachments ?? [], resolveAttachment, { + concurrency: 1, + }); + + yield* updateSession(context, { + status: "running", + activeTurnId: turnId, + ...(model ? { model } : {}), + }); + yield* emit({ + ...(yield* eventBase({ threadId: input.threadId, turnId })), + type: "turn.started", + payload: { + ...(model ? { model } : {}), + ...(thinkingLevel ? { effort: thinkingLevel } : {}), + }, + }); + + let assistantItemStarted = false; + let streamedAssistantText = ""; + let streamingEmitQueue: Promise = Promise.resolve(); + const streamingCreatedAtFallback = yield* nowIso; + const runtimeContext = yield* Effect.context(); + const runPromise = Effect.runPromiseWith(runtimeContext); + const enqueueStreamingEmit = (effect: Effect.Effect) => { + streamingEmitQueue = streamingEmitQueue + .then(() => runPromise(effect)) + .catch(() => undefined); + }; + + const result = yield* Effect.tryPromise({ + try: () => + runPiRpcPrompt({ + binaryPath: piSettings.binaryPath, + args: makePiArgs({ + sessionPath: context.sessionPath, + model: modelArg, + thinkingLevel, + sessionExists, + }), + cwd: context.session.cwd ?? serverConfig.cwd, + environment, + message: text ?? "", + images, + timeoutMs: 180_000, + signal: abort.signal, + onEvent: (event) => { + const delta = readPiAssistantTextDelta(event); + if (delta.length === 0 || abort.signal.aborted) { + return; + } + const eventTimestamp = + typeof event.timestamp === "string" ? event.timestamp : streamingCreatedAtFallback; + streamedAssistantText += delta; + enqueueStreamingEmit( + Effect.gen(function* () { + if (!assistantItemStarted) { + assistantItemStarted = true; + yield* emit({ + ...eventBaseSync({ + threadId: input.threadId, + createdAt: eventTimestamp, + turnId, + itemId, + }), + type: "item.started", + payload: { + itemType: "assistant_message", + status: "inProgress", + title: "Pi response", + }, + }); + } + yield* emit({ + ...eventBaseSync({ + threadId: input.threadId, + createdAt: eventTimestamp, + turnId, + itemId, + }), + type: "content.delta", + payload: { + streamKind: "assistant_text", + delta, + }, + }); + }), + ); + }, + }), + catch: (cause) => + new ProviderAdapterProcessError({ + provider: PROVIDER, + threadId: input.threadId, + detail: errorDetail(cause), + cause, + }), + }).pipe( + Effect.tapError((error) => + Effect.gen(function* () { + context.activeAbort = null; + yield* updateSession(context, { + status: "ready", + activeTurnId: undefined, + lastError: error.message, + }); + yield* emit({ + ...(yield* eventBase({ threadId: input.threadId, turnId })), + type: "turn.completed", + payload: { + state: abort.signal.aborted ? "interrupted" : "failed", + errorMessage: error.message, + }, + }); + }), + ), + ); + + yield* Effect.promise(() => streamingEmitQueue); + + const assistantText = result.text.trim(); + if (assistantText.length > 0) { + if (!assistantItemStarted) { + assistantItemStarted = true; + yield* emit({ + ...(yield* eventBase({ threadId: input.threadId, turnId, itemId })), + type: "item.started", + payload: { + itemType: "assistant_message", + status: "inProgress", + title: "Pi response", + }, + }); + yield* emit({ + ...(yield* eventBase({ threadId: input.threadId, turnId, itemId })), + type: "content.delta", + payload: { + streamKind: "assistant_text", + delta: assistantText, + }, + }); + } else if (assistantText.startsWith(streamedAssistantText)) { + const finalRemainder = assistantText.slice(streamedAssistantText.length); + if (finalRemainder.length > 0) { + yield* emit({ + ...(yield* eventBase({ threadId: input.threadId, turnId, itemId })), + type: "content.delta", + payload: { + streamKind: "assistant_text", + delta: finalRemainder, + }, + }); + } + } + } + if (assistantItemStarted) { + yield* emit({ + ...(yield* eventBase({ threadId: input.threadId, turnId, itemId })), + type: "item.completed", + payload: { + itemType: "assistant_message", + status: "completed", + title: "Pi response", + }, + }); + } + yield* emit({ + ...(yield* eventBase({ threadId: input.threadId, turnId })), + type: "turn.completed", + payload: { + state: "completed", + stopReason: null, + }, + }); + + context.activeAbort = null; + context.turns.push({ id: turnId, items: result.events as unknown[] }); + yield* updateSession(context, { + status: "ready", + activeTurnId: undefined, + lastError: undefined, + }); + + return { + threadId: input.threadId, + turnId, + resumeCursor: { sessionPath: context.sessionPath }, + }; + }); + + const interruptTurn: PiAdapterShape["interruptTurn"] = (threadId) => + Effect.gen(function* () { + const context = yield* requireSession(threadId); + context.activeAbort?.abort(); + context.activeAbort = null; + yield* updateSession(context, { status: "ready", activeTurnId: undefined }); + }); + + const respondToRequest: PiAdapterShape["respondToRequest"] = (threadId, requestId) => + Effect.fail( + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "respondToRequest", + detail: `Pi adapter has no pending approval request '${requestId}' for thread '${threadId}'.`, + }), + ); + + const respondToUserInput: PiAdapterShape["respondToUserInput"] = (threadId, requestId) => + Effect.fail( + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "respondToUserInput", + detail: `Pi adapter has no pending user-input request '${requestId}' for thread '${threadId}'.`, + }), + ); + + const stopSession: PiAdapterShape["stopSession"] = (threadId) => + Effect.gen(function* () { + const context = yield* requireSession(threadId); + context.activeAbort?.abort(); + sessions.delete(threadId); + yield* emit({ + ...(yield* eventBase({ threadId })), + type: "session.exited", + payload: { + reason: "Pi session stopped", + recoverable: true, + exitKind: "graceful", + }, + }); + }); + + const listSessions: PiAdapterShape["listSessions"] = () => + Effect.sync(() => Array.from(sessions.values(), ({ session }) => ({ ...session }))); + + const hasSession: PiAdapterShape["hasSession"] = (threadId) => + Effect.sync(() => sessions.has(threadId)); + + const readThread: PiAdapterShape["readThread"] = (threadId) => + Effect.gen(function* () { + const context = yield* requireSession(threadId); + return { + threadId, + turns: [...context.turns], + }; + }); + + const rollbackThread: PiAdapterShape["rollbackThread"] = (threadId, numTurns) => + Effect.gen(function* () { + const context = yield* requireSession(threadId); + if (numTurns > 0) { + context.turns = context.turns.slice(0, Math.max(0, context.turns.length - numTurns)); + } + return { + threadId, + turns: [...context.turns], + }; + }); + + const stopAll: PiAdapterShape["stopAll"] = () => + Effect.forEach(Array.from(sessions.keys()), (threadId) => stopSession(threadId), { + discard: true, + }); + + yield* Effect.addFinalizer(() => + Effect.sync(() => { + for (const context of sessions.values()) { + context.activeAbort?.abort(); + } + sessions.clear(); + }).pipe(Effect.andThen(Queue.shutdown(runtimeEvents)), Effect.ignore), + ); + + return { + provider: PROVIDER, + capabilities: { + sessionModelSwitch: "in-session", + }, + startSession, + sendTurn, + interruptTurn, + respondToRequest, + respondToUserInput, + stopSession, + listSessions, + hasSession, + readThread, + rollbackThread, + stopAll, + get streamEvents() { + return Stream.fromQueue(runtimeEvents); + }, + } satisfies PiAdapterShape; +}); diff --git a/apps/server/src/provider/Layers/PiProvider.ts b/apps/server/src/provider/Layers/PiProvider.ts new file mode 100644 index 00000000000..2a330a8cdd5 --- /dev/null +++ b/apps/server/src/provider/Layers/PiProvider.ts @@ -0,0 +1,365 @@ +import { + type ModelCapabilities, + type PiSettings, + ProviderDriverKind, + type ServerProviderModel, + type ServerProviderSlashCommand, +} from "@t3tools/contracts"; +import { createModelCapabilities } from "@t3tools/shared/model"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; + +import { + buildSelectOptionDescriptor, + buildServerProvider, + detailFromResult, + parseGenericCliVersion, + providerModelsFromSettings, + type ServerProviderDraft, +} from "../providerSnapshot.ts"; +import { + piAgentDir, + readPiLocalSettings, + runPiCommand, + runPiRpcCommands, + type PiLocalSettings, + type PiRpcModel, + type PiRpcSlashCommand, +} from "./PiRpc.ts"; + +const PROVIDER = ProviderDriverKind.make("pi"); +const PI_PRESENTATION = { + displayName: "Pi", + showInteractionModeToggle: true, +} as const; + +const DEFAULT_PI_MODEL_CAPABILITIES: ModelCapabilities = createModelCapabilities({ + optionDescriptors: [], +}); + +const THINKING_LEVELS = ["off", "minimal", "low", "medium", "high", "xhigh"] as const; + +const BUILT_IN_SLASH_COMMANDS: ReadonlyArray = [ + { name: "settings", description: "Open or inspect Pi settings in Pi-native contexts" }, + { name: "model", description: "Change the active Pi model" }, + { name: "scoped-models", description: "Adjust Pi model cycling scope" }, + { name: "export", description: "Export the Pi session when supported by the runtime" }, + { name: "import", description: "Import Pi session content in Pi-native contexts" }, + { name: "share", description: "Share the Pi session in Pi-native contexts" }, + { name: "copy", description: "Copy Pi output in Pi-native contexts" }, + { name: "name", description: "Set the Pi session name", input: { hint: "name" } }, + { name: "session", description: "Show or switch Pi session information" }, + { name: "changelog", description: "Show Pi changelog in Pi-native contexts" }, + { name: "hotkeys", description: "Show Pi hotkeys in Pi-native contexts" }, + { name: "fork", description: "Fork the current Pi session when an entry id is available" }, + { name: "clone", description: "Clone the current Pi session" }, + { name: "tree", description: "Show Pi session tree in Pi-native contexts" }, + { name: "login", description: "Start Pi login in Pi-native contexts" }, + { name: "logout", description: "Start Pi logout in Pi-native contexts" }, + { name: "new", description: "Start a new Pi session" }, + { + name: "compact", + description: "Compact the current Pi session", + input: { hint: "instructions" }, + }, + { name: "resume", description: "Resume a Pi session in Pi-native contexts" }, + { name: "reload", description: "Reload Pi configuration in Pi-native contexts" }, + { name: "quit", description: "Quit Pi in Pi-native contexts" }, +]; + +function piModelCapabilities(model: PiRpcModel, settings: PiLocalSettings): ModelCapabilities { + if (model.reasoning !== true) return DEFAULT_PI_MODEL_CAPABILITIES; + const defaultThinking = settings.defaultThinkingLevel; + return createModelCapabilities({ + optionDescriptors: [ + buildSelectOptionDescriptor({ + id: "thinkingLevel", + label: "Thinking", + options: THINKING_LEVELS.map((level) => ({ + value: level, + label: level === "xhigh" ? "Extra High" : level[0]!.toUpperCase() + level.slice(1), + isDefault: defaultThinking === level, + })), + }), + ], + }); +} + +function modelSlug(model: PiRpcModel): string | null { + const provider = model.provider?.trim(); + const id = model.id?.trim(); + if (!provider || !id) return null; + return `${provider}/${id}`; +} + +function preferredIndex(model: PiRpcModel, settings: PiLocalSettings, slug: string): number | null { + const defaultSlug = + settings.defaultProvider && settings.defaultModel + ? `${settings.defaultProvider}/${settings.defaultModel}` + : null; + if (defaultSlug === slug || defaultSlug === model.id) return -1; + + const enabledModels = settings.enabledModels ?? []; + for (let index = 0; index < enabledModels.length; index += 1) { + const pattern = enabledModels[index]?.split(":")[0]?.trim(); + if (!pattern) continue; + if ( + pattern === slug || + pattern === model.id || + slug.endsWith(`/${pattern}`) || + model.name?.toLowerCase() === pattern.toLowerCase() + ) { + return index; + } + } + return null; +} + +export function piModelsToServerModels( + models: ReadonlyArray, + settings: PiLocalSettings, +): ReadonlyArray { + const mapped = models.flatMap((model) => { + const slug = modelSlug(model); + if (!slug) return []; + return [ + { + slug, + name: model.name?.trim() || model.id, + shortName: model.id, + subProvider: model.provider, + isCustom: false, + capabilities: piModelCapabilities(model, settings), + preferred: preferredIndex(model, settings, slug), + }, + ]; + }); + + return mapped + .toSorted((left, right) => { + const leftPreferred = left.preferred; + const rightPreferred = right.preferred; + if (leftPreferred !== null && rightPreferred !== null && leftPreferred !== rightPreferred) { + return leftPreferred - rightPreferred; + } + if (leftPreferred !== null && rightPreferred === null) return -1; + if (leftPreferred === null && rightPreferred !== null) return 1; + return `${left.subProvider ?? ""}/${left.name}`.localeCompare( + `${right.subProvider ?? ""}/${right.name}`, + ); + }) + .map(({ preferred: _preferred, ...model }) => model); +} + +function dedupeSlashCommands( + commands: ReadonlyArray, +): ReadonlyArray { + const byName = new Map(); + for (const command of commands) { + const name = command.name.trim(); + if (!name) continue; + const key = name.toLowerCase(); + if (!byName.has(key)) byName.set(key, { ...command, name }); + } + return Array.from(byName.values()).toSorted((left, right) => left.name.localeCompare(right.name)); +} + +function dynamicCommandsToSlashCommands( + commands: ReadonlyArray, +): ReadonlyArray { + return commands.flatMap((command) => { + const name = command.name?.trim().replace(/^\/+/, ""); + if (!name) return []; + return [ + { + name, + ...(command.description?.trim() ? { description: command.description.trim() } : {}), + }, + ]; + }); +} + +export const makePendingPiProvider = (piSettings: PiSettings): Effect.Effect => + Effect.gen(function* () { + const checkedAt = yield* Effect.map(DateTime.now, DateTime.formatIso); + return buildServerProvider({ + presentation: PI_PRESENTATION, + enabled: piSettings.enabled, + checkedAt, + models: providerModelsFromSettings( + [], + PROVIDER, + piSettings.customModels, + DEFAULT_PI_MODEL_CAPABILITIES, + ), + slashCommands: BUILT_IN_SLASH_COMMANDS, + probe: { + installed: false, + version: null, + status: "warning", + auth: { status: "unknown", type: "pi" }, + message: piSettings.enabled + ? "Pi provider status has not been checked in this session yet." + : "Pi is disabled in T3 Code settings.", + }, + }); + }); + +export const checkPiProviderStatus = Effect.fn("checkPiProviderStatus")(function* ( + piSettings: PiSettings, + cwd: string, + environment: NodeJS.ProcessEnv = process.env, +) { + const checkedAt = DateTime.formatIso(yield* DateTime.now); + const fallbackModels = providerModelsFromSettings( + [], + PROVIDER, + piSettings.customModels, + DEFAULT_PI_MODEL_CAPABILITIES, + ); + + if (!piSettings.enabled) { + return buildServerProvider({ + presentation: PI_PRESENTATION, + enabled: false, + checkedAt, + models: fallbackModels, + slashCommands: BUILT_IN_SLASH_COMMANDS, + probe: { + installed: false, + version: null, + status: "warning", + auth: { status: "unknown", type: "pi" }, + message: "Pi is disabled in T3 Code settings.", + }, + }); + } + + const versionExit = yield* Effect.exit( + Effect.tryPromise(() => + runPiCommand({ + binaryPath: piSettings.binaryPath, + args: ["--version"], + cwd, + environment, + timeoutMs: 10_000, + }), + ), + ); + if (versionExit._tag === "Failure") { + return buildServerProvider({ + presentation: PI_PRESENTATION, + enabled: true, + checkedAt, + models: fallbackModels, + slashCommands: BUILT_IN_SLASH_COMMANDS, + probe: { + installed: false, + version: null, + status: "error", + auth: { status: "unknown", type: "pi" }, + message: `Unable to run Pi at '${piSettings.binaryPath}'.`, + }, + }); + } + + const versionResult = versionExit.value; + const version = + parseGenericCliVersion(versionResult.stdout) ?? parseGenericCliVersion(versionResult.stderr); + if (versionResult.code !== 0) { + return buildServerProvider({ + presentation: PI_PRESENTATION, + enabled: true, + checkedAt, + models: fallbackModels, + slashCommands: BUILT_IN_SLASH_COMMANDS, + probe: { + installed: true, + version, + status: "error", + auth: { status: "unknown", type: "pi" }, + message: detailFromResult(versionResult) ?? "Pi version check failed.", + }, + }); + } + + const settings = yield* Effect.tryPromise(() => readPiLocalSettings(environment)).pipe( + Effect.catch(() => Effect.succeed({} satisfies PiLocalSettings)), + ); + + const inventoryExit = yield* Effect.exit( + Effect.tryPromise(() => + runPiRpcCommands({ + binaryPath: piSettings.binaryPath, + commands: [ + { id: "models", type: "get_available_models" }, + { id: "commands", type: "get_commands" }, + ], + cwd, + environment, + timeoutMs: 20_000, + }), + ), + ); + + if (inventoryExit._tag === "Failure") { + return buildServerProvider({ + presentation: PI_PRESENTATION, + enabled: true, + checkedAt, + models: fallbackModels, + slashCommands: BUILT_IN_SLASH_COMMANDS, + probe: { + installed: true, + version, + status: "warning", + auth: { status: "unknown", type: "pi" }, + message: `Pi is installed, but model discovery failed while reading ${piAgentDir(environment)}.`, + }, + }); + } + + const modelResponse = inventoryExit.value.responses.get("models"); + const commandResponse = inventoryExit.value.responses.get("commands"); + const piModels = + modelResponse?.data && + typeof modelResponse.data === "object" && + Array.isArray((modelResponse.data as { models?: unknown }).models) + ? ((modelResponse.data as { models: PiRpcModel[] }).models ?? []) + : []; + const dynamicCommands = + commandResponse?.data && + typeof commandResponse.data === "object" && + Array.isArray((commandResponse.data as { commands?: unknown }).commands) + ? ((commandResponse.data as { commands: PiRpcSlashCommand[] }).commands ?? []) + : []; + + const models = providerModelsFromSettings( + piModelsToServerModels(piModels, settings), + PROVIDER, + piSettings.customModels, + DEFAULT_PI_MODEL_CAPABILITIES, + ); + const slashCommands = dedupeSlashCommands([ + ...BUILT_IN_SLASH_COMMANDS, + ...dynamicCommandsToSlashCommands(dynamicCommands), + ]); + + return buildServerProvider({ + presentation: PI_PRESENTATION, + enabled: true, + checkedAt, + models, + slashCommands, + probe: { + installed: true, + version, + status: models.length > 0 ? "ready" : "warning", + auth: { status: models.length > 0 ? "authenticated" : "unknown", type: "pi" }, + message: + models.length > 0 + ? `Pi is available using existing config at ${piAgentDir(environment)}.` + : `Pi is installed, but no available models were reported from ${piAgentDir(environment)}.`, + }, + }); +}); diff --git a/apps/server/src/provider/Layers/PiRpc.test.ts b/apps/server/src/provider/Layers/PiRpc.test.ts new file mode 100644 index 00000000000..7ac553f0bcf --- /dev/null +++ b/apps/server/src/provider/Layers/PiRpc.test.ts @@ -0,0 +1,87 @@ +import { describe, expect, it } from "vitest"; + +import { extractAssistantText, readPiAssistantTextDelta, type PiRpcLine } from "./PiRpc.ts"; + +describe("extractAssistantText", () => { + it("prefers the final assistant message over streamed reasoning and tool-call deltas", () => { + const events: PiRpcLine[] = [ + { + type: "message_update", + assistantMessageEvent: { + type: "thinking_delta", + delta: "Considering scanning options\n", + }, + }, + { + type: "message_update", + assistantMessageEvent: { + type: "toolcall_delta", + delta: '{"command":"pwd; ls -la","timeout":10}', + }, + }, + { + type: "message_end", + message: { + role: "assistant", + content: [ + { + type: "toolCall", + name: "bash", + arguments: { command: "pwd; ls -la", timeout: 10 }, + }, + ], + }, + }, + { + type: "message_end", + message: { + role: "assistant", + content: [ + { + type: "text", + text: "This project folder is a fresh agent workspace.", + }, + ], + }, + }, + ]; + + expect(extractAssistantText(events)).toBe("This project folder is a fresh agent workspace."); + }); + + it("falls back to streamed deltas when no final assistant message is available", () => { + const events: PiRpcLine[] = [ + { + type: "message_update", + assistantMessageEvent: { type: "text_delta", delta: "partial " }, + }, + { + type: "message_update", + assistantMessageEvent: { type: "text_delta", delta: "answer" }, + }, + ]; + + expect(extractAssistantText(events)).toBe("partial answer"); + }); + + it("streams only text_delta events as visible assistant text", () => { + expect( + readPiAssistantTextDelta({ + type: "message_update", + assistantMessageEvent: { type: "text_delta", delta: "visible" }, + }), + ).toBe("visible"); + expect( + readPiAssistantTextDelta({ + type: "message_update", + assistantMessageEvent: { type: "thinking_delta", delta: "hidden" }, + }), + ).toBe(""); + expect( + readPiAssistantTextDelta({ + type: "message_update", + assistantMessageEvent: { type: "toolcall_delta", delta: '{"path":"SOUL.md"}' }, + }), + ).toBe(""); + }); +}); diff --git a/apps/server/src/provider/Layers/PiRpc.ts b/apps/server/src/provider/Layers/PiRpc.ts new file mode 100644 index 00000000000..2e5e3a14079 --- /dev/null +++ b/apps/server/src/provider/Layers/PiRpc.ts @@ -0,0 +1,356 @@ +// @effect-diagnostics-next-line nodeBuiltinImport:off +import { spawn } from "node:child_process"; +import { homedir } from "node:os"; +// @effect-diagnostics-next-line nodeBuiltinImport:off +import { join } from "node:path"; +// @effect-diagnostics-next-line nodeBuiltinImport:off +import { readFile } from "node:fs/promises"; + +export interface PiRpcModel { + readonly id: string; + readonly name?: string | undefined; + readonly provider?: string | undefined; + readonly reasoning?: boolean | undefined; + readonly input?: ReadonlyArray | undefined; +} + +export interface PiRpcSlashCommand { + readonly name: string; + readonly description?: string | undefined; + readonly source?: string | undefined; +} + +export interface PiLocalSettings { + readonly defaultProvider?: string | undefined; + readonly defaultModel?: string | undefined; + readonly defaultThinkingLevel?: string | undefined; + readonly enabledModels?: ReadonlyArray | undefined; +} + +export interface PiRpcLine { + readonly type?: string | undefined; + readonly id?: string | undefined; + readonly timestamp?: string | undefined; + readonly command?: string | undefined; + readonly success?: boolean | undefined; + readonly data?: unknown; + readonly error?: string | undefined; + readonly message?: unknown; + readonly assistantMessageEvent?: unknown; + readonly messages?: unknown; +} + +export interface PiRpcCommand { + readonly id: string; + readonly type: string; + readonly [key: string]: unknown; +} + +export interface PiRpcRunResult { + readonly responses: ReadonlyMap; + readonly events: ReadonlyArray; + readonly stderr: string; +} + +export interface PiPromptResult { + readonly text: string; + readonly events: ReadonlyArray; + readonly stderr: string; +} + +export type PiRpcEventHandler = (event: PiRpcLine) => void; + +export function piAgentDir(environment: NodeJS.ProcessEnv = process.env): string { + return environment.PI_CODING_AGENT_DIR?.trim() || join(homedir(), ".pi", "agent"); +} + +export async function readPiLocalSettings( + environment: NodeJS.ProcessEnv = process.env, +): Promise { + const settingsPath = join(piAgentDir(environment), "settings.json"); + const raw = await readFile(settingsPath, "utf8"); + const parsed = JSON.parse(raw) as Record; + return { + defaultProvider: + typeof parsed.defaultProvider === "string" ? parsed.defaultProvider : undefined, + defaultModel: typeof parsed.defaultModel === "string" ? parsed.defaultModel : undefined, + defaultThinkingLevel: + typeof parsed.defaultThinkingLevel === "string" ? parsed.defaultThinkingLevel : undefined, + enabledModels: Array.isArray(parsed.enabledModels) + ? parsed.enabledModels.filter((value): value is string => typeof value === "string") + : undefined, + }; +} + +export async function runPiCommand(input: { + readonly binaryPath: string; + readonly args: ReadonlyArray; + readonly cwd?: string | undefined; + readonly environment?: NodeJS.ProcessEnv | undefined; + readonly timeoutMs?: number | undefined; +}): Promise<{ stdout: string; stderr: string; code: number }> { + return await new Promise((resolve, reject) => { + const child = spawn(input.binaryPath, [...input.args], { + cwd: input.cwd, + env: input.environment ?? process.env, + shell: process.platform === "win32", + stdio: ["ignore", "pipe", "pipe"], + }); + let stdout = ""; + let stderr = ""; + let settled = false; + // @effect-diagnostics-next-line globalTimers:off + const timer = setTimeout(() => { + if (settled) return; + settled = true; + child.kill("SIGTERM"); + reject(new Error(`Timed out running ${input.binaryPath} ${input.args.join(" ")}`)); + }, input.timeoutMs ?? 10_000); + + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + child.stdout.on("data", (chunk) => { + stdout += chunk; + }); + child.stderr.on("data", (chunk) => { + stderr += chunk; + }); + child.on("error", (error) => { + if (settled) return; + settled = true; + clearTimeout(timer); + reject(error); + }); + child.on("close", (code) => { + if (settled) return; + settled = true; + clearTimeout(timer); + resolve({ stdout, stderr, code: code ?? 0 }); + }); + }); +} + +export async function runPiRpcCommands(input: { + readonly binaryPath: string; + readonly args?: ReadonlyArray | undefined; + readonly commands: ReadonlyArray; + readonly cwd?: string | undefined; + readonly environment?: NodeJS.ProcessEnv | undefined; + readonly timeoutMs?: number | undefined; + readonly waitForAgentEnd?: boolean | undefined; + readonly signal?: AbortSignal | undefined; + readonly onEvent?: PiRpcEventHandler | undefined; +}): Promise { + return await new Promise((resolve, reject) => { + const child = spawn(input.binaryPath, ["--mode", "rpc", ...(input.args ?? [])], { + cwd: input.cwd, + env: input.environment ?? process.env, + shell: process.platform === "win32", + stdio: ["pipe", "pipe", "pipe"], + }); + + const responses = new Map(); + const pending = new Set(input.commands.map((command) => command.id)); + const events: PiRpcLine[] = []; + let stderr = ""; + let stdoutBuffer = ""; + let agentEnded = false; + let requestedClose = false; + let settled = false; + + const finishIfReady = () => { + if (pending.size > 0) return; + if (input.waitForAgentEnd === true && !agentEnded) return; + if (requestedClose) return; + requestedClose = true; + child.stdin.end(); + }; + + const rejectOnce = (error: Error) => { + if (settled) return; + settled = true; + clearTimeout(timer); + child.kill("SIGTERM"); + reject(error); + }; + + // @effect-diagnostics-next-line globalTimers:off + const timer = setTimeout(() => { + rejectOnce(new Error(`Timed out waiting for Pi RPC. Stderr: ${stderr.trim()}`)); + }, input.timeoutMs ?? 120_000); + + input.signal?.addEventListener("abort", () => { + rejectOnce(new Error("Pi RPC request was aborted.")); + }); + + const handleLine = (line: string) => { + if (!line.trim()) return; + let parsed: PiRpcLine; + try { + parsed = JSON.parse(line) as PiRpcLine; + } catch { + events.push({ type: "parse_error", message: line }); + return; + } + + if (parsed.type === "response" && parsed.id) { + responses.set(parsed.id, parsed); + pending.delete(parsed.id); + if (parsed.success === false) { + rejectOnce( + new Error( + typeof parsed.error === "string" + ? parsed.error + : `Pi RPC command '${parsed.command ?? parsed.id}' failed.`, + ), + ); + return; + } + finishIfReady(); + return; + } + + events.push(parsed); + input.onEvent?.(parsed); + if (parsed.type === "agent_end") { + agentEnded = true; + finishIfReady(); + } + }; + + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + child.stdout.on("data", (chunk) => { + stdoutBuffer += chunk; + let newline = stdoutBuffer.indexOf("\n"); + while (newline >= 0) { + const line = stdoutBuffer.slice(0, newline).replace(/\r$/, ""); + stdoutBuffer = stdoutBuffer.slice(newline + 1); + handleLine(line); + newline = stdoutBuffer.indexOf("\n"); + } + }); + child.stderr.on("data", (chunk) => { + stderr += chunk; + }); + child.on("error", (error) => { + rejectOnce(error); + }); + child.on("close", () => { + if (settled) return; + if (stdoutBuffer.trim()) { + handleLine(stdoutBuffer.trim()); + } + if (pending.size > 0) { + rejectOnce(new Error(`Pi RPC exited before responses arrived. Stderr: ${stderr.trim()}`)); + return; + } + if (input.waitForAgentEnd === true && !agentEnded) { + rejectOnce(new Error(`Pi RPC exited before agent_end. Stderr: ${stderr.trim()}`)); + return; + } + settled = true; + clearTimeout(timer); + resolve({ responses, events, stderr }); + }); + + for (const command of input.commands) { + child.stdin.write(`${JSON.stringify(command)}\n`); + } + finishIfReady(); + }); +} + +export async function runPiRpcPrompt(input: { + readonly binaryPath: string; + readonly args?: ReadonlyArray | undefined; + readonly message: string; + readonly images?: ReadonlyArray<{ type: "image"; data: string; mimeType: string }> | undefined; + readonly cwd?: string | undefined; + readonly environment?: NodeJS.ProcessEnv | undefined; + readonly timeoutMs?: number | undefined; + readonly signal?: AbortSignal | undefined; + readonly onEvent?: PiRpcEventHandler | undefined; +}): Promise { + const result = await runPiRpcCommands({ + binaryPath: input.binaryPath, + args: input.args, + commands: [ + { + id: "prompt", + type: "prompt", + message: input.message, + ...(input.images && input.images.length > 0 ? { images: input.images } : {}), + }, + ], + cwd: input.cwd, + environment: input.environment, + timeoutMs: input.timeoutMs, + waitForAgentEnd: true, + signal: input.signal, + onEvent: input.onEvent, + }); + return { + text: extractAssistantText(result.events), + events: result.events, + stderr: result.stderr, + }; +} + +function readTextFromContent(content: unknown): string { + if (typeof content === "string") return content; + if (!Array.isArray(content)) return ""; + return content + .map((part) => { + if (!part || typeof part !== "object") return ""; + const record = part as Record; + if (typeof record.text === "string") return record.text; + if (typeof record.content === "string") return record.content; + return ""; + }) + .join(""); +} + +function readAssistantText(message: unknown): string { + if (!message || typeof message !== "object") return ""; + const record = message as Record; + if (record.role !== "assistant") return ""; + return readTextFromContent(record.content); +} + +export function readPiAssistantTextDelta(event: PiRpcLine): string { + if (event.type !== "message_update") return ""; + const assistantMessageEvent = event.assistantMessageEvent; + if (!assistantMessageEvent || typeof assistantMessageEvent !== "object") return ""; + const record = assistantMessageEvent as Record; + if (record.type !== "text_delta") return ""; + return typeof record.delta === "string" ? record.delta : ""; +} + +export function extractAssistantText(events: ReadonlyArray): string { + for (const event of events.toReversed()) { + if (event.type !== "message_end") continue; + const text = readAssistantText(event.message); + if (text.trim().length > 0) return text; + } + + for (const event of events.toReversed()) { + if (event.type !== "agent_end" || !Array.isArray(event.messages)) continue; + for (const candidate of event.messages.toReversed()) { + const text = readAssistantText(candidate); + if (text.trim().length > 0) return text; + } + } + + return events.map((event) => readPiAssistantTextDelta(event)).join(""); +} + +export function splitPiModelSlug( + slug: string | undefined, +): { provider: string; modelId: string } | null { + if (!slug) return null; + const [provider, ...modelParts] = slug.split("/"); + const modelId = modelParts.join("/"); + if (!provider || !modelId) return null; + return { provider, modelId: modelId.split(":")[0] ?? modelId }; +} diff --git a/apps/server/src/provider/Layers/ProviderRegistry.test.ts b/apps/server/src/provider/Layers/ProviderRegistry.test.ts index fb6eb3b443d..69ab1f1c571 100644 --- a/apps/server/src/provider/Layers/ProviderRegistry.test.ts +++ b/apps/server/src/provider/Layers/ProviderRegistry.test.ts @@ -1301,6 +1301,7 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsService.layerTest(), T "codex", "cursor", "opencode", + "pi", ]); assert.strictEqual(cursorProvider?.enabled, false); assert.strictEqual(cursorProvider?.status, "disabled"); diff --git a/apps/server/src/provider/builtInDrivers.ts b/apps/server/src/provider/builtInDrivers.ts index 5af56dc6b0e..0390e2c4f6b 100644 --- a/apps/server/src/provider/builtInDrivers.ts +++ b/apps/server/src/provider/builtInDrivers.ts @@ -24,6 +24,7 @@ import { ClaudeDriver, type ClaudeDriverEnv } from "./Drivers/ClaudeDriver.ts"; import { CodexDriver, type CodexDriverEnv } from "./Drivers/CodexDriver.ts"; import { CursorDriver, type CursorDriverEnv } from "./Drivers/CursorDriver.ts"; import { OpenCodeDriver, type OpenCodeDriverEnv } from "./Drivers/OpenCodeDriver.ts"; +import { PiDriver, type PiDriverEnv } from "./Drivers/PiDriver.ts"; import type { AnyProviderDriver } from "./ProviderDriver.ts"; /** @@ -35,7 +36,8 @@ export type BuiltInDriversEnv = | ClaudeDriverEnv | CodexDriverEnv | CursorDriverEnv - | OpenCodeDriverEnv; + | OpenCodeDriverEnv + | PiDriverEnv; /** * Ordered list of built-in drivers. Order matters only for tie-breaking in @@ -47,4 +49,5 @@ export const BUILT_IN_DRIVERS: ReadonlyArray line.trim()) + .find(Boolean) ?? "" + ); +} + +export const makePiTextGeneration = ( + piSettings: PiSettings, + environment: NodeJS.ProcessEnv = process.env, +): Effect.Effect => + Effect.succeed({ + generateCommitMessage: (input) => + Effect.tryPromise({ + try: async (): Promise => { + const prompt = [ + "Write a concise git commit message for these staged changes.", + "Return a subject line, then an optional body.", + input.includeBranch + ? "Also include a branch name on a final line starting with Branch:" + : "", + `Branch: ${input.branch ?? "unknown"}`, + `Summary:\n${input.stagedSummary}`, + `Patch:\n${input.stagedPatch}`, + ] + .filter(Boolean) + .join("\n\n"); + const result = await runPiRpcPrompt({ + binaryPath: piSettings.binaryPath, + args: ["--no-session", "--no-tools", "--no-builtin-tools"], + message: prompt, + cwd: input.cwd, + environment, + timeoutMs: 120_000, + }); + const lines = result.text.trim().split(/\r?\n/); + const subject = firstLine(result.text) || "Update files"; + const branchLine = lines.find((line) => line.toLowerCase().startsWith("branch:")); + const branch = branchLine?.replace(/^branch:\s*/i, "").trim(); + const body = lines + .filter((line) => line.trim() !== subject && line !== branchLine) + .join("\n") + .trim(); + return { + subject, + body, + ...(input.includeBranch && branch ? { branch } : {}), + }; + }, + catch: (cause) => textGenerationError("generateCommitMessage", cause), + }), + generatePrContent: (input) => + Effect.tryPromise({ + try: async (): Promise => { + const prompt = [ + "Write a pull request title and body for these changes.", + "Return the title on the first line, then the body.", + `Base branch: ${input.baseBranch}`, + `Head branch: ${input.headBranch}`, + `Commit summary:\n${input.commitSummary}`, + `Diff summary:\n${input.diffSummary}`, + `Patch:\n${input.diffPatch}`, + ].join("\n\n"); + const result = await runPiRpcPrompt({ + binaryPath: piSettings.binaryPath, + args: ["--no-session", "--no-tools", "--no-builtin-tools"], + message: prompt, + cwd: input.cwd, + environment, + timeoutMs: 120_000, + }); + const title = firstLine(result.text) || "Update"; + const body = result.text.replace(title, "").trim(); + return { title, body }; + }, + catch: (cause) => textGenerationError("generatePrContent", cause), + }), + generateBranchName: (input) => + Effect.tryPromise({ + try: async (): Promise => { + const result = await runPiRpcPrompt({ + binaryPath: piSettings.binaryPath, + args: ["--no-session", "--no-tools", "--no-builtin-tools"], + message: `Create a short kebab-case git branch name for this request. Return only the branch name.\n\n${input.message}`, + cwd: input.cwd, + environment, + timeoutMs: 60_000, + }); + return { branch: firstLine(result.text).replace(/[^a-zA-Z0-9/_-]/g, "") || "pi-update" }; + }, + catch: (cause) => textGenerationError("generateBranchName", cause), + }), + generateThreadTitle: (input) => + Effect.tryPromise({ + try: async (): Promise => { + const result = await runPiRpcPrompt({ + binaryPath: piSettings.binaryPath, + args: ["--no-session", "--no-tools", "--no-builtin-tools"], + message: `Create a concise chat title, five words or fewer. Return only the title.\n\n${input.message}`, + cwd: input.cwd, + environment, + timeoutMs: 60_000, + }); + return { title: firstLine(result.text) || "Pi Chat" }; + }, + catch: (cause) => textGenerationError("generateThreadTitle", cause), + }), + } satisfies TextGenerationShape); diff --git a/apps/web/src/components/KeybindingsToast.browser.tsx b/apps/web/src/components/KeybindingsToast.browser.tsx index 611eaf572d0..988f9fe6c55 100644 --- a/apps/web/src/components/KeybindingsToast.browser.tsx +++ b/apps/web/src/components/KeybindingsToast.browser.tsx @@ -130,6 +130,11 @@ function createBaseServerConfig(): ServerConfig { serverPassword: "", customModels: [], }, + pi: { + enabled: true, + binaryPath: "", + customModels: [], + }, }, }, }; diff --git a/apps/web/src/components/settings/AddProviderInstanceDialog.tsx b/apps/web/src/components/settings/AddProviderInstanceDialog.tsx index affa35ff260..bf3edd5dd0d 100644 --- a/apps/web/src/components/settings/AddProviderInstanceDialog.tsx +++ b/apps/web/src/components/settings/AddProviderInstanceDialog.tsx @@ -13,7 +13,7 @@ import { useSettings, useUpdateSettings } from "../../hooks/useSettings"; import { cn } from "../../lib/utils"; import { normalizeProviderAccentColor } from "../../providerInstances"; import { Button } from "../ui/button"; -import { ACPRegistryIcon, Gemini, GithubCopilotIcon, PiAgentIcon, type Icon } from "../Icons"; +import { ACPRegistryIcon, Gemini, GithubCopilotIcon, type Icon } from "../Icons"; import { Dialog, DialogDescription, @@ -86,11 +86,6 @@ const COMING_SOON_DRIVER_OPTIONS: readonly ComingSoonDriverOption[] = [ label: "ACP Registry", icon: ACPRegistryIcon, }, - { - value: ProviderDriverKind.make("piAgent"), - label: "Pi Agent", - icon: PiAgentIcon, - }, ]; /** diff --git a/apps/web/src/components/settings/providerDriverMeta.ts b/apps/web/src/components/settings/providerDriverMeta.ts index 8d3d7482f62..d73483d6191 100644 --- a/apps/web/src/components/settings/providerDriverMeta.ts +++ b/apps/web/src/components/settings/providerDriverMeta.ts @@ -3,10 +3,11 @@ import { CodexSettings, CursorSettings, OpenCodeSettings, + PiSettings, ProviderDriverKind, } from "@t3tools/contracts"; import type * as Schema from "effect/Schema"; -import { ClaudeAI, CursorIcon, type Icon, OpenAI, OpenCodeIcon } from "../Icons"; +import { ClaudeAI, CursorIcon, type Icon, OpenAI, OpenCodeIcon, PiAgentIcon } from "../Icons"; type ProviderSettingsSchema = { readonly fields: Readonly>; @@ -59,6 +60,12 @@ export const PROVIDER_CLIENT_DEFINITIONS: readonly ProviderClientDefinition[] = icon: OpenCodeIcon, settingsSchema: OpenCodeSettings, }, + { + value: ProviderDriverKind.make("pi"), + label: "Pi", + icon: PiAgentIcon, + settingsSchema: PiSettings, + }, ]; export const PROVIDER_CLIENT_DEFINITION_BY_VALUE: Partial< diff --git a/packages/contracts/src/model.ts b/packages/contracts/src/model.ts index 8e7daaa0c79..8092e84b54a 100644 --- a/packages/contracts/src/model.ts +++ b/packages/contracts/src/model.ts @@ -131,6 +131,7 @@ const CODEX_DRIVER_KIND = ProviderDriverKind.make("codex"); const CLAUDE_DRIVER_KIND = ProviderDriverKind.make("claudeAgent"); const CURSOR_DRIVER_KIND = ProviderDriverKind.make("cursor"); const OPENCODE_DRIVER_KIND = ProviderDriverKind.make("opencode"); +const PI_DRIVER_KIND = ProviderDriverKind.make("pi"); export const DEFAULT_MODEL = "gpt-5.4"; export const DEFAULT_GIT_TEXT_GENERATION_MODEL = "gpt-5.4-mini"; @@ -140,6 +141,7 @@ export const DEFAULT_MODEL_BY_PROVIDER: Partial> [CLAUDE_DRIVER_KIND]: "Claude", [CURSOR_DRIVER_KIND]: "Cursor", [OPENCODE_DRIVER_KIND]: "OpenCode", + [PI_DRIVER_KIND]: "Pi", }; diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 2d115eed98e..fbbdfcc43b4 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -331,6 +331,33 @@ export const OpenCodeSettings = makeProviderSettingsSchema( ); export type OpenCodeSettings = typeof OpenCodeSettings.Type; +export const PiSettings = makeProviderSettingsSchema( + { + enabled: Schema.Boolean.pipe( + Schema.withDecodingDefault(Effect.succeed(true)), + Schema.annotateKey({ providerSettingsForm: { hidden: true } }), + ), + binaryPath: makeBinaryPathSetting("pi").pipe( + Schema.annotateKey({ + title: "Binary path", + description: "Path to the Pi binary. Pi reads its existing ~/.pi config by default.", + providerSettingsForm: { + placeholder: "pi", + clearWhenEmpty: "omit", + }, + }), + ), + customModels: Schema.Array(Schema.String).pipe( + Schema.withDecodingDefault(Effect.succeed([])), + Schema.annotateKey({ providerSettingsForm: { hidden: true } }), + ), + }, + { + order: ["binaryPath"], + }, +); +export type PiSettings = typeof PiSettings.Type; + export const ObservabilitySettings = Schema.Struct({ otlpTracesUrl: TrimmedString.pipe(Schema.withDecodingDefault(Effect.succeed(""))), otlpMetricsUrl: TrimmedString.pipe(Schema.withDecodingDefault(Effect.succeed(""))), @@ -370,6 +397,7 @@ export const ServerSettings = Schema.Struct({ claudeAgent: ClaudeSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), cursor: CursorSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), opencode: OpenCodeSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), + pi: PiSettings.pipe(Schema.withDecodingDefault(Effect.succeed({}))), }).pipe(Schema.withDecodingDefault(Effect.succeed({}))), // New driver-agnostic instance map. Keyed by `ProviderInstanceId`; values // are `ProviderInstanceConfig` envelopes. The driver-specific config blob @@ -445,6 +473,12 @@ const OpenCodeSettingsPatch = Schema.Struct({ customModels: Schema.optionalKey(Schema.Array(Schema.String)), }); +const PiSettingsPatch = Schema.Struct({ + enabled: Schema.optionalKey(Schema.Boolean), + binaryPath: Schema.optionalKey(TrimmedString), + customModels: Schema.optionalKey(Schema.Array(Schema.String)), +}); + export const ServerSettingsPatch = Schema.Struct({ // Server settings enableAssistantStreaming: Schema.optionalKey(Schema.Boolean), @@ -464,6 +498,7 @@ export const ServerSettingsPatch = Schema.Struct({ claudeAgent: Schema.optionalKey(ClaudeSettingsPatch), cursor: Schema.optionalKey(CursorSettingsPatch), opencode: Schema.optionalKey(OpenCodeSettingsPatch), + pi: Schema.optionalKey(PiSettingsPatch), }), ), // Whole-map replacement for the new instance config. Patching individual From f0f0662e8290fa4ef644f3cf0e1419a17be86b7c Mon Sep 17 00:00:00 2001 From: AmbitiousRealism2025 Date: Wed, 27 May 2026 17:01:03 -0400 Subject: [PATCH 2/3] Use activity timeout for Pi RPC prompts --- ATTEMPTS.md | 25 ++++++++++++++++++++ apps/server/src/provider/Layers/PiAdapter.ts | 1 + apps/server/src/provider/Layers/PiRpc.ts | 23 ++++++++++++++++-- 3 files changed, 47 insertions(+), 2 deletions(-) diff --git a/ATTEMPTS.md b/ATTEMPTS.md index 9a708bf7da4..409b857f96c 100644 --- a/ATTEMPTS.md +++ b/ATTEMPTS.md @@ -239,3 +239,28 @@ Result: Lesson: - Pi can stream cleanly through T3 as long as the adapter maps only `text_delta` to `assistant_text`; reasoning and tool-call deltas must remain hidden or be mapped to non-assistant surfaces later. + +## Loop 9: Long Pi Turn Timeout Diagnosis + +Edit: + +- Investigated a desktop/userdata failure for thread `a3b619ce-023d-4afd-9bf7-1454aa52af77`. +- Found Pi ran for the full 180-second adapter timeout, executed multiple tool rounds, wrote output, and was killed before final `agent_end`. +- Changed Pi prompt turns to reset the RPC timeout on stdout/stderr activity. +- Kept the 180-second timeout as an idle timeout so genuinely stuck Pi processes still fail. + +Checks: + +- `bunx vitest run src/provider/Layers/PiRpc.test.ts` +- `bun typecheck` +- `bun fmt` + +Result: + +- Targeted Pi RPC tests passed: 3 tests passed. +- Typecheck passed: 13 successful tasks. +- Formatting passed. + +Lesson: + +- Long-running Pi work should be bounded by inactivity, not total wall-clock time. Multi-tool tasks can exceed three minutes while still making healthy progress. diff --git a/apps/server/src/provider/Layers/PiAdapter.ts b/apps/server/src/provider/Layers/PiAdapter.ts index 86e19a6f42f..760b88e5033 100644 --- a/apps/server/src/provider/Layers/PiAdapter.ts +++ b/apps/server/src/provider/Layers/PiAdapter.ts @@ -315,6 +315,7 @@ export const makePiAdapter = Effect.fn("makePiAdapter")(function* ( message: text ?? "", images, timeoutMs: 180_000, + resetTimeoutOnActivity: true, signal: abort.signal, onEvent: (event) => { const delta = readPiAssistantTextDelta(event); diff --git a/apps/server/src/provider/Layers/PiRpc.ts b/apps/server/src/provider/Layers/PiRpc.ts index 2e5e3a14079..14325152e94 100644 --- a/apps/server/src/provider/Layers/PiRpc.ts +++ b/apps/server/src/provider/Layers/PiRpc.ts @@ -140,6 +140,7 @@ export async function runPiRpcCommands(input: { readonly waitForAgentEnd?: boolean | undefined; readonly signal?: AbortSignal | undefined; readonly onEvent?: PiRpcEventHandler | undefined; + readonly resetTimeoutOnActivity?: boolean | undefined; }): Promise { return await new Promise((resolve, reject) => { const child = spawn(input.binaryPath, ["--mode", "rpc", ...(input.args ?? [])], { @@ -174,10 +175,20 @@ export async function runPiRpcCommands(input: { reject(error); }; + const timeoutMs = input.timeoutMs ?? 120_000; + let timer: ReturnType; + const resetTimer = () => { + clearTimeout(timer); + // @effect-diagnostics-next-line globalTimers:off + timer = setTimeout(() => { + rejectOnce(new Error(`Timed out waiting for Pi RPC. Stderr: ${stderr.trim()}`)); + }, timeoutMs); + }; + // @effect-diagnostics-next-line globalTimers:off - const timer = setTimeout(() => { + timer = setTimeout(() => { rejectOnce(new Error(`Timed out waiting for Pi RPC. Stderr: ${stderr.trim()}`)); - }, input.timeoutMs ?? 120_000); + }, timeoutMs); input.signal?.addEventListener("abort", () => { rejectOnce(new Error("Pi RPC request was aborted.")); @@ -185,6 +196,9 @@ export async function runPiRpcCommands(input: { const handleLine = (line: string) => { if (!line.trim()) return; + if (input.resetTimeoutOnActivity === true) { + resetTimer(); + } let parsed: PiRpcLine; try { parsed = JSON.parse(line) as PiRpcLine; @@ -232,6 +246,9 @@ export async function runPiRpcCommands(input: { }); child.stderr.on("data", (chunk) => { stderr += chunk; + if (input.resetTimeoutOnActivity === true) { + resetTimer(); + } }); child.on("error", (error) => { rejectOnce(error); @@ -271,6 +288,7 @@ export async function runPiRpcPrompt(input: { readonly timeoutMs?: number | undefined; readonly signal?: AbortSignal | undefined; readonly onEvent?: PiRpcEventHandler | undefined; + readonly resetTimeoutOnActivity?: boolean | undefined; }): Promise { const result = await runPiRpcCommands({ binaryPath: input.binaryPath, @@ -289,6 +307,7 @@ export async function runPiRpcPrompt(input: { waitForAgentEnd: true, signal: input.signal, onEvent: input.onEvent, + resetTimeoutOnActivity: input.resetTimeoutOnActivity, }); return { text: extractAssistantText(result.events), From 9c1c57e6a1fc5e802bd320c6323b89783d026ed8 Mon Sep 17 00:00:00 2001 From: AmbitiousRealism2025 Date: Wed, 27 May 2026 17:10:17 -0400 Subject: [PATCH 3/3] Stream Pi tool activity --- ATTEMPTS.md | 28 +++ apps/server/src/provider/Layers/PiAdapter.ts | 198 +++++++++++++++++- apps/server/src/provider/Layers/PiRpc.test.ts | 27 ++- apps/server/src/provider/Layers/PiRpc.ts | 16 ++ 4 files changed, 267 insertions(+), 2 deletions(-) diff --git a/ATTEMPTS.md b/ATTEMPTS.md index 409b857f96c..e9fef5edd7c 100644 --- a/ATTEMPTS.md +++ b/ATTEMPTS.md @@ -264,3 +264,31 @@ Result: Lesson: - Long-running Pi work should be bounded by inactivity, not total wall-clock time. Multi-tool tasks can exceed three minutes while still making healthy progress. + +## Loop 10: Pi Tool Activity Streaming + +Edit: + +- Compared current T3 runtime behavior with the prior Pi POC. +- Confirmed the prior POC used a separate Pi activity bridge and `tool.progress`, while current T3 already projects tool lifecycle `item.started`, `item.updated`, and `item.completed` events into the Work Log. +- Added Pi RPC fields for `tool_execution_start`, `tool_execution_update`, and `tool_execution_end`. +- Mapped Pi tool events into T3-native tool lifecycle events with item types for command execution, file changes, web search, and dynamic tools. +- Added Pi tool result text extraction for partial and final tool output. + +Checks: + +- `bunx vitest run src/provider/Layers/PiRpc.test.ts` +- `bun typecheck` +- `bun fmt` +- Live Pi RPC smoke against the existing local `~/.pi/agent` config using a harmless bash prompt. + +Result: + +- Targeted Pi RPC tests passed: 5 tests passed. +- Typecheck passed before formatting: 13 successful tasks. +- Formatting passed. +- Live Pi RPC smoke emitted `tool_execution_start`, two `tool_execution_update` events, and `tool_execution_end` for the `bash` tool before `agent_end`. + +Lesson: + +- Assistant token streaming and tool activity streaming are separate surfaces. Pi was doing real tool work, but the adapter only forwarded assistant `text_delta`; forwarding Pi tool execution events to T3 tool lifecycle events should make long Pi turns visibly active in the Work Log. diff --git a/apps/server/src/provider/Layers/PiAdapter.ts b/apps/server/src/provider/Layers/PiAdapter.ts index 760b88e5033..92612859f17 100644 --- a/apps/server/src/provider/Layers/PiAdapter.ts +++ b/apps/server/src/provider/Layers/PiAdapter.ts @@ -1,5 +1,6 @@ import { EventId, + type CanonicalItemType, ProviderDriverKind, ProviderInstanceId, RuntimeItemId, @@ -34,7 +35,13 @@ import { ProviderAdapterValidationError, } from "../Errors.ts"; import type { ProviderAdapterShape, ProviderThreadSnapshot } from "../Services/ProviderAdapter.ts"; -import { readPiAssistantTextDelta, runPiRpcPrompt, splitPiModelSlug } from "./PiRpc.ts"; +import { + readPiAssistantTextDelta, + readPiToolResultText, + runPiRpcPrompt, + splitPiModelSlug, + type PiRpcLine, +} from "./PiRpc.ts"; const PROVIDER = ProviderDriverKind.make("pi"); @@ -73,6 +80,105 @@ function errorDetail(error: unknown): string { return error instanceof Error ? error.message : String(error); } +function asRecord(value: unknown): Record | undefined { + return value && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : undefined; +} + +function asString(value: unknown): string | undefined { + return typeof value === "string" && value.trim().length > 0 ? value : undefined; +} + +function truncateDetail(value: string, maxLength = 1_200): string { + const trimmed = value.trim(); + if (trimmed.length <= maxLength) return trimmed; + return `${trimmed.slice(0, maxLength - 3).trimEnd()}...`; +} + +function piToolItemType(toolName: string | undefined): CanonicalItemType { + const normalized = toolName?.toLowerCase(); + if ( + normalized === "bash" || + normalized === "shell" || + normalized === "terminal" || + normalized === "execute" + ) { + return "command_execution"; + } + if ( + normalized === "write" || + normalized === "edit" || + normalized === "apply_patch" || + normalized === "patch" + ) { + return "file_change"; + } + if ( + normalized === "exa_search" || + normalized === "web_search" || + normalized === "search" || + normalized === "grep" || + normalized === "find" + ) { + return "web_search"; + } + return "dynamic_tool_call"; +} + +function piToolTitle(toolName: string | undefined, itemType: CanonicalItemType): string { + switch (itemType) { + case "command_execution": + return "Ran command"; + case "file_change": + return "Changed file"; + case "web_search": + return "Searched"; + default: + return toolName ?? "Pi tool"; + } +} + +function piToolArgsDetail(args: unknown, itemType: CanonicalItemType): string | undefined { + const record = asRecord(args); + if (!record) return undefined; + if (itemType === "command_execution") { + return asString(record.command); + } + return ( + asString(record.path) ?? + asString(record.filePath) ?? + asString(record.relativePath) ?? + asString(record.query) ?? + asString(record.pattern) + ); +} + +function piToolData(event: PiRpcLine, itemType: CanonicalItemType): Record { + const args = asRecord(event.args); + const output = + event.type === "tool_execution_update" + ? readPiToolResultText(event.partialResult) + : event.type === "tool_execution_end" + ? readPiToolResultText(event.result) + : ""; + return { + tool: event.toolName, + kind: + itemType === "command_execution" + ? "execute" + : itemType === "file_change" + ? "edit" + : itemType === "web_search" + ? "search" + : "tool", + toolCallId: event.toolCallId, + ...(args ? { args, rawInput: args } : {}), + ...(itemType === "command_execution" && args?.command ? { command: args.command } : {}), + ...(output.length > 0 ? { rawOutput: { content: output } } : {}), + }; +} + export const makePiAdapter = Effect.fn("makePiAdapter")(function* ( piSettings: PiSettings, options: { @@ -290,6 +396,7 @@ export const makePiAdapter = Effect.fn("makePiAdapter")(function* ( let assistantItemStarted = false; let streamedAssistantText = ""; + const startedPiToolIds = new Set(); let streamingEmitQueue: Promise = Promise.resolve(); const streamingCreatedAtFallback = yield* nowIso; const runtimeContext = yield* Effect.context(); @@ -299,6 +406,94 @@ export const makePiAdapter = Effect.fn("makePiAdapter")(function* ( .then(() => runPromise(effect)) .catch(() => undefined); }; + const enqueuePiToolEvent = (event: PiRpcLine) => { + if ( + event.type !== "tool_execution_start" && + event.type !== "tool_execution_update" && + event.type !== "tool_execution_end" + ) { + return; + } + const toolItemId = + event.toolCallId?.trim() || event.id?.trim() || `pi-tool-${randomUUID()}`; + const toolName = event.toolName?.trim() || undefined; + const itemType = piToolItemType(toolName); + const title = piToolTitle(toolName, itemType); + const eventTimestamp = + typeof event.timestamp === "string" ? event.timestamp : streamingCreatedAtFallback; + const outputText = + event.type === "tool_execution_update" + ? readPiToolResultText(event.partialResult) + : event.type === "tool_execution_end" + ? readPiToolResultText(event.result) + : ""; + const argsDetail = piToolArgsDetail(event.args, itemType); + const detail = truncateDetail(outputText || argsDetail || toolName || title); + const data = piToolData(event, itemType); + + enqueueStreamingEmit( + Effect.gen(function* () { + if (!startedPiToolIds.has(toolItemId)) { + startedPiToolIds.add(toolItemId); + yield* emit({ + ...eventBaseSync({ + threadId: input.threadId, + createdAt: eventTimestamp, + turnId, + itemId: toolItemId, + }), + type: "item.started", + payload: { + itemType, + status: "inProgress", + title, + detail, + data, + }, + }); + } + + if (event.type === "tool_execution_update") { + yield* emit({ + ...eventBaseSync({ + threadId: input.threadId, + createdAt: eventTimestamp, + turnId, + itemId: toolItemId, + }), + type: "item.updated", + payload: { + itemType, + status: "inProgress", + title, + detail, + data, + }, + }); + return; + } + + if (event.type === "tool_execution_end") { + yield* emit({ + ...eventBaseSync({ + threadId: input.threadId, + createdAt: eventTimestamp, + turnId, + itemId: toolItemId, + }), + type: "item.completed", + payload: { + itemType, + status: event.isError === true ? "failed" : "completed", + title, + detail, + data, + }, + }); + } + }), + ); + }; const result = yield* Effect.tryPromise({ try: () => @@ -318,6 +513,7 @@ export const makePiAdapter = Effect.fn("makePiAdapter")(function* ( resetTimeoutOnActivity: true, signal: abort.signal, onEvent: (event) => { + enqueuePiToolEvent(event); const delta = readPiAssistantTextDelta(event); if (delta.length === 0 || abort.signal.aborted) { return; diff --git a/apps/server/src/provider/Layers/PiRpc.test.ts b/apps/server/src/provider/Layers/PiRpc.test.ts index 7ac553f0bcf..079b6d64046 100644 --- a/apps/server/src/provider/Layers/PiRpc.test.ts +++ b/apps/server/src/provider/Layers/PiRpc.test.ts @@ -1,6 +1,11 @@ import { describe, expect, it } from "vitest"; -import { extractAssistantText, readPiAssistantTextDelta, type PiRpcLine } from "./PiRpc.ts"; +import { + extractAssistantText, + readPiAssistantTextDelta, + readPiToolResultText, + type PiRpcLine, +} from "./PiRpc.ts"; describe("extractAssistantText", () => { it("prefers the final assistant message over streamed reasoning and tool-call deltas", () => { @@ -85,3 +90,23 @@ describe("extractAssistantText", () => { ).toBe(""); }); }); + +describe("readPiToolResultText", () => { + it("extracts text from Pi tool result content arrays", () => { + expect( + readPiToolResultText({ + content: [ + { type: "text", text: "partial output" }, + { type: "text", text: " continued" }, + ], + details: { truncation: null }, + }), + ).toBe("partial output continued"); + }); + + it("supports common scalar output fields", () => { + expect(readPiToolResultText("raw output")).toBe("raw output"); + expect(readPiToolResultText({ stdout: "stdout output" })).toBe("stdout output"); + expect(readPiToolResultText({ output: "generic output" })).toBe("generic output"); + }); +}); diff --git a/apps/server/src/provider/Layers/PiRpc.ts b/apps/server/src/provider/Layers/PiRpc.ts index 14325152e94..df3106592e5 100644 --- a/apps/server/src/provider/Layers/PiRpc.ts +++ b/apps/server/src/provider/Layers/PiRpc.ts @@ -33,6 +33,12 @@ export interface PiRpcLine { readonly timestamp?: string | undefined; readonly command?: string | undefined; readonly success?: boolean | undefined; + readonly toolCallId?: string | undefined; + readonly toolName?: string | undefined; + readonly args?: unknown; + readonly partialResult?: unknown; + readonly result?: unknown; + readonly isError?: boolean | undefined; readonly data?: unknown; readonly error?: string | undefined; readonly message?: unknown; @@ -337,6 +343,16 @@ function readAssistantText(message: unknown): string { return readTextFromContent(record.content); } +export function readPiToolResultText(result: unknown): string { + if (typeof result === "string") return result; + if (!result || typeof result !== "object") return ""; + const record = result as Record; + if (typeof record.text === "string") return record.text; + if (typeof record.output === "string") return record.output; + if (typeof record.stdout === "string") return record.stdout; + return readTextFromContent(record.content); +} + export function readPiAssistantTextDelta(event: PiRpcLine): string { if (event.type !== "message_update") return ""; const assistantMessageEvent = event.assistantMessageEvent;