chat: remove Pi runner — route all chat/pipeline/channels through Claude
Stage 1 of removing Pi (Claude-only). Cuts the non-Claude branches in the chat WS handler, pipeline executor, and channel send-and-await; deletes the Pi sidecar, its ecosystem entry, pi-bridge, and the Pi model-listing spawn (now a static Claude tier list). Adds a guard coercing any legacy non-claude-code model preference to the Claude default so old settings don't break chat or jobs. Removes the dead no-op session-save REST route and stale Pi docs. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.8
@@ -1,521 +0,0 @@
|
||||
# Chat with Pi - Flow Documentation
|
||||
|
||||
This document describes the full-stack architecture and data flow of the Chat app in Officer.dev, which uses **Pi** (an AI coding agent) as the backend.
|
||||
|
||||
---
|
||||
|
||||
## Architecture Overview
|
||||
|
||||
```
|
||||
┌─────────────────────────────────────────────────────────────────┐
|
||||
│ FRONTEND │
|
||||
│ ┌──────────────┐ ┌────────────────┐ ┌───────────────┐ │
|
||||
│ │ InputArea │───▶│ usePiChat │───▶│ useChatWebSocket│ │
|
||||
│ │ (user input) │ │ (hook) │ │ (WS connection)│ │
|
||||
│ └──────────────┘ └───────┬────────┘ └───────┬────────┘ │
|
||||
│ │ │ │
|
||||
│ ▼ ▼ │
|
||||
│ ┌────────────────┐ ┌───────────────┐ │
|
||||
│ │ MessageList │◀───│ handleMessage │ │
|
||||
│ │ (UI updates) │ │ (event parser)│ │
|
||||
│ └────────────────┘ └───────────────┘ │
|
||||
└─────────────────────────────────────────────────────────────────┘
|
||||
│ WebSocket
|
||||
▼
|
||||
┌─────────────────────────────────────────────────────────────────┐
|
||||
│ BACKEND │
|
||||
│ │
|
||||
│ ┌──────────────────────────────────────────────────────────┐ │
|
||||
│ │ server.tsx │ │
|
||||
│ │ '/api/pi/chat/ws' → upgradeWs() → piWebsocket handler │ │
|
||||
│ └──────────────────────────────────────────────────────────┘ │
|
||||
│ │ │
|
||||
│ ▼ │
|
||||
│ ┌──────────────────────────────────────────────────────────┐ │
|
||||
│ │ websocket.ts (piWebsocket) │ │
|
||||
│ │ │ │
|
||||
│ │ message() ──┬──▶ handleChat() ──▶ SessionManager │ │
|
||||
│ │ │ │ │ │
|
||||
│ │ │ ▼ │ │
|
||||
│ │ │ pi-bridge.ts │ │
|
||||
│ │ │ (spawn Pi subprocess) │ │
|
||||
│ │ │ │ │ │
|
||||
│ │ │ ▼ │ │
|
||||
│ │ │ Bun.spawn(['pi', '--mode', │ │
|
||||
│ │ │ 'rpc', ...]) │ │
|
||||
│ │ │ │ │ │
|
||||
│ │ │ ▼ │ │
|
||||
│ │ │ Parse JSON from stdout │ │
|
||||
│ │ │ │ │ │
|
||||
│ │ └────────────────────┘ │ │
|
||||
│ │ │ │ │
|
||||
│ │ ▼ │ │
|
||||
│ │ sendToClient() ──▶ WebSocket JSON │ │
|
||||
│ └──────────────────────────────────────────────────────────┘ │
|
||||
│ │
|
||||
│ ┌──────────────────────────────────────────────────────────┐ │
|
||||
│ │ storage.ts │ │
|
||||
│ │ Save to: ~/.pi-sessions/<sessionId>/ │ │
|
||||
│ │ - meta.json (session metadata) │ │
|
||||
│ │ - messages.json (message history) │ │
|
||||
│ └──────────────────────────────────────────────────────────┘ │
|
||||
└─────────────────────────────────────────────────────────────────┘
|
||||
▲
|
||||
│
|
||||
Pi subprocess
|
||||
(AI Agent)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Directory Structure
|
||||
|
||||
### Frontend
|
||||
|
||||
```
|
||||
src/workspaces/officerdev/src/
|
||||
├── apps/
|
||||
│ ├── Chat/
|
||||
│ │ ├── index.ts # Main entry, exports
|
||||
│ │ ├── types.ts # TypeScript types
|
||||
│ │ ├── ChatPanelWrapper.tsx # Main wrapper component
|
||||
│ │ ├── ChatList.tsx # Session list sidebar
|
||||
│ │ ├── ChatLauncher.tsx # Launch chat
|
||||
│ │ ├── EmbeddableChat.tsx # Embeddable version
|
||||
│ │ ├── components/
|
||||
│ │ │ ├── InputArea.tsx # User input
|
||||
│ │ │ ├── MessageList.tsx # Message display
|
||||
│ │ │ ├── MessageBubble.tsx # Individual message
|
||||
│ │ │ ├── ToolActivity.tsx # Tool execution UI
|
||||
│ │ │ ├── QuestionActivity.tsx # Question prompt UI
|
||||
│ │ │ ├── ModelSelector.tsx # Model dropdown
|
||||
│ │ │ ├── AttachButton.tsx # File attachment
|
||||
│ │ │ ├── AttachmentList.tsx # Attached files
|
||||
│ │ │ └── WebpageDialog.tsx # URL preview
|
||||
│ │ ├── hooks/
|
||||
│ │ │ └── useChat.ts # Main chat hook
|
||||
│ │ ├── useAttachments.ts # File attachment handling
|
||||
│ │ ├── useAudioRecording.ts # Voice input
|
||||
│ │ └── useSlashCommands.ts # /commands handling
|
||||
│ │
|
||||
│ └── ChatHistory/ # Session management UI
|
||||
│
|
||||
├── hooks/
|
||||
│ ├── usePiChat.ts # Main chat hook
|
||||
│ └── ...
|
||||
│
|
||||
└── state/
|
||||
└── useChatSessions.ts # Session state management
|
||||
```
|
||||
|
||||
### Backend
|
||||
|
||||
```
|
||||
src/
|
||||
├── server.tsx # Main server entry
|
||||
│ # - WebSocket upgrade
|
||||
│ # - Route registration
|
||||
│ # - Pi installation check
|
||||
│
|
||||
└── servers/
|
||||
├── hono.ts # Hono REST server
|
||||
│
|
||||
└── api/
|
||||
├── pi/
|
||||
│ ├── types.ts # TypeScript types
|
||||
│ ├── websocket.ts # WebSocket handler
|
||||
│ ├── session-manager.ts # Session lifecycle
|
||||
│ ├── pi-bridge.ts # Pi subprocess spawner
|
||||
│ ├── storage.ts # Disk persistence
|
||||
│ ├── rest.ts # REST API endpoints
|
||||
│ ├── logger.ts # Logging
|
||||
│ └── README.md # Original Pi API docs
|
||||
│
|
||||
└── ... (other API routes)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Type Definitions
|
||||
|
||||
### Client Messages (Frontend → Backend)
|
||||
|
||||
```typescript
|
||||
// Start new chat
|
||||
type ClientMessage =
|
||||
| { type: 'chat'; prompt: string; sessionId?: string; model?: string; cwd?: string; groupSlug?: string; attachmentIds?: string[] }
|
||||
| { type: 'resume'; sessionId: string } // Resume existing session
|
||||
| { type: 'stop' }; // Stop generation
|
||||
```
|
||||
|
||||
### Server Messages (Backend → Frontend)
|
||||
|
||||
```typescript
|
||||
type ServerMessage =
|
||||
| { type: 'session:init'; sessionId: string; model: string; cwd: string }
|
||||
| { type: 'assistant:delta'; text: string } // Streaming token
|
||||
| { type: 'assistant:text'; text: string } // Complete message
|
||||
| { type: 'tool:start'; toolCallId: string; toolName: string; toolInput: Record<string, unknown> }
|
||||
| { type: 'tool:result'; toolCallId: string; output: string; isError: boolean }
|
||||
| { type: 'result'; sessionId: string; cost: MessageCost }
|
||||
| { type: 'sync:messages'; sessionId: string; messages: Message[]; isGenerating: boolean; streamingText: string }
|
||||
| { type: 'error'; message: string; errorCode?: string }
|
||||
| { type: 'stopped' };
|
||||
```
|
||||
|
||||
### Internal Types
|
||||
|
||||
```typescript
|
||||
type Message = {
|
||||
id: string;
|
||||
timestamp: number;
|
||||
role: 'user' | 'assistant' | 'tool';
|
||||
text?: string;
|
||||
model?: string;
|
||||
cost?: MessageCost;
|
||||
toolCallId?: string;
|
||||
toolName?: string;
|
||||
toolInput?: Record<string, unknown>;
|
||||
output?: string;
|
||||
isError?: boolean;
|
||||
};
|
||||
|
||||
type SessionMeta = {
|
||||
id: string;
|
||||
title: string;
|
||||
model: string;
|
||||
cwd: string;
|
||||
createdAt: number;
|
||||
updatedAt: number;
|
||||
messageCount: number;
|
||||
cost: MessageCost;
|
||||
groupSlug?: string | null;
|
||||
};
|
||||
|
||||
type MessageCost = {
|
||||
inputTokens: number;
|
||||
outputTokens: number;
|
||||
totalUSD: number;
|
||||
};
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## REST API Endpoints
|
||||
|
||||
| Method | Endpoint | Description |
|
||||
|--------|----------|-------------|
|
||||
| GET | `/api/pi/models` | List available AI models |
|
||||
| POST | `/api/pi/sessions` | List all sessions |
|
||||
| GET | `/api/pi/sessions/:id` | Get session with messages |
|
||||
| PATCH | `/api/pi/sessions/:id` | Update session (rename) |
|
||||
| DELETE | `/api/pi/sessions/:id` | Delete session |
|
||||
| GET | `/api/pi/sessions/search?q=` | Search sessions |
|
||||
| POST | `/api/pi/groups` | Create group |
|
||||
| GET | `/api/pi/groups` | List groups |
|
||||
| PATCH | `/api/pi/groups/:slug` | Update group |
|
||||
| DELETE | `/api/pi/groups/:slug` | Delete group |
|
||||
| POST | `/api/pi/sessions/:id/move` | Move session between groups |
|
||||
|
||||
---
|
||||
|
||||
## Data Flow Details
|
||||
|
||||
### 1. Starting a New Chat
|
||||
|
||||
```
|
||||
User enters message
|
||||
↓
|
||||
usePiChat.sendPrompt()
|
||||
↓
|
||||
useChatWebSocket.send({ type: 'chat', prompt, ... })
|
||||
↓
|
||||
WebSocket connection to /api/pi/chat/ws
|
||||
↓
|
||||
server.tsx: upgradeWs() - JWT verification
|
||||
↓
|
||||
websocket.ts: handleChat()
|
||||
↓
|
||||
SessionManager.getOrCreate(sessionId, email, cwd, model)
|
||||
↓
|
||||
pi-bridge.ts: spawnPi(cwd, model, onEvent)
|
||||
↓
|
||||
Bun.spawn(['pi', '--mode', 'rpc', '--no-extensions', ...])
|
||||
↓
|
||||
send prompt to Pi via stdin
|
||||
↓
|
||||
Pi process streams JSON events to stdout
|
||||
↓
|
||||
pi-bridge parses events → onEvent(PiEvent)
|
||||
↓
|
||||
websocket.ts createEventHandler() → sendToClient()
|
||||
↓
|
||||
useChatWebSocket.onMessage() → usePiChat.handleMessage()
|
||||
↓
|
||||
React state update → UI renders
|
||||
```
|
||||
|
||||
### 2. Resuming a Session
|
||||
|
||||
```
|
||||
User navigates to /chat/:sessionId
|
||||
↓
|
||||
usePiChat mounts (with initialSessionId)
|
||||
↓
|
||||
useEffect triggers: getSession(initialSessionId)
|
||||
↓
|
||||
REST API: GET /api/pi/sessions/:sessionId
|
||||
↓
|
||||
storage.ts: loadSession() reads from disk
|
||||
↓
|
||||
Messages loaded into state
|
||||
↓
|
||||
WebSocket connects (for new messages)
|
||||
↓
|
||||
websocket.ts: handleResume()
|
||||
↓
|
||||
Spawns fresh Pi process
|
||||
↓
|
||||
Sends 'sync:messages' to client
|
||||
```
|
||||
|
||||
### 3. Pi Event Parsing
|
||||
|
||||
```
|
||||
Pi stdout JSON line:
|
||||
{
|
||||
"type": "message_update",
|
||||
"assistantMessageEvent": {
|
||||
"type": "text_delta",
|
||||
"delta": "Hello"
|
||||
}
|
||||
}
|
||||
↓
|
||||
pi-bridge.ts: parsePiEvent()
|
||||
↓
|
||||
Converted to PiEvent:
|
||||
{ type: 'delta', text: 'Hello' }
|
||||
↓
|
||||
Event handler:
|
||||
- 'delta' → append to streamBuffer, send 'assistant:delta'
|
||||
- 'text' → commit streamBuffer, send 'assistant:text'
|
||||
- 'tool:start' → send 'tool:start', add tool message
|
||||
- 'tool:result' → send 'tool:result', update tool message
|
||||
- 'result' → save session, send 'result'
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Session Storage
|
||||
|
||||
### Directory Structure
|
||||
|
||||
```
|
||||
~/.pi-sessions/
|
||||
├── <sessionId>/
|
||||
│ ├── meta.json # Session metadata
|
||||
│ └── messages.json # Message history
|
||||
│
|
||||
└── @<groupSlug>/ # Grouped sessions
|
||||
├── .group-meta.json
|
||||
├── <sessionId1>/
|
||||
│ ├── meta.json
|
||||
│ └── messages.json
|
||||
└── <sessionId2>/
|
||||
```
|
||||
|
||||
### meta.json
|
||||
|
||||
```json
|
||||
{
|
||||
"id": "session-uuid",
|
||||
"title": "First prompt...",
|
||||
"model": "opencode/big-pickle",
|
||||
"cwd": "/home/user",
|
||||
"createdAt": 1700000000000,
|
||||
"updatedAt": 1700000000000,
|
||||
"messageCount": 10,
|
||||
"cost": {
|
||||
"inputTokens": 5000,
|
||||
"outputTokens": 2000,
|
||||
"totalUSD": 0.05
|
||||
},
|
||||
"groupSlug": null
|
||||
}
|
||||
```
|
||||
|
||||
### messages.json
|
||||
|
||||
```json
|
||||
{
|
||||
"messages": [
|
||||
{ "id": "...", "timestamp": 1700000000000, "role": "user", "text": "Hello" },
|
||||
{ "id": "...", "timestamp": 1700000000001, "role": "assistant", "text": "Hi!" },
|
||||
{ "id": "...", "timestamp": 1700000000002, "role": "tool", "toolName": "Bash", "toolInput": {...}, "output": "..." }
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Authentication
|
||||
|
||||
### WebSocket Authentication
|
||||
|
||||
1. Client connects: `ws://host/api/pi/chat/ws?token=<JWT>`
|
||||
2. `server.tsx:upgradeWs()` extracts token from URL
|
||||
3. `jwt.ts:verify(token)` validates JWT
|
||||
4. Check token blacklist (logout/revocation)
|
||||
5. Extract user info: `{ userId, email, role }`
|
||||
6. Attach to WebSocket data: `ws.data = { userId, email, provider: 'pi', ... }`
|
||||
|
||||
### REST Authentication
|
||||
|
||||
- Bearer token in `Authorization` header
|
||||
- `hono.ts` uses `userMiddleware` to verify
|
||||
- User object attached to request context: `ctx.get('user')`
|
||||
|
||||
---
|
||||
|
||||
## Session Management
|
||||
|
||||
### SessionManager (in-memory)
|
||||
|
||||
```typescript
|
||||
class SessionManager {
|
||||
private sessions = new Map<string, UserSession>();
|
||||
private userSessions = new Map<string, string[]>(); // email → sessionIds
|
||||
|
||||
getOrCreate(sessionId, email, cwd, model, groupSlug): UserSession
|
||||
getSession(sessionId): UserSession | null
|
||||
attachWs(sessionId, ws): void
|
||||
detachWs(sessionId): void
|
||||
setIdleTimeout(sessionId, timeoutMs): void // Auto-cleanup after 1 hour
|
||||
deleteSession(sessionId): void
|
||||
}
|
||||
```
|
||||
|
||||
### UserSession Object
|
||||
|
||||
```typescript
|
||||
type UserSession = {
|
||||
sessionId: string;
|
||||
email: string;
|
||||
cwd: string;
|
||||
model: string;
|
||||
piProcess: Subprocess | null; // Bun subprocess
|
||||
ws: WebSocket | null; // Connected client
|
||||
lastActivity: number;
|
||||
idleTimer: Timer | null;
|
||||
streamBuffer: string; // Uncommitted streaming text
|
||||
isGenerating: boolean;
|
||||
systemContextSent: boolean;
|
||||
messages: Message[];
|
||||
meta: SessionMeta;
|
||||
};
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Pi Subprocess
|
||||
|
||||
### Spawn Command
|
||||
|
||||
```typescript
|
||||
const args = [
|
||||
'pi',
|
||||
'--mode', 'rpc',
|
||||
'--no-extensions',
|
||||
'--no-skills',
|
||||
'--no-prompt-templates',
|
||||
'--no-themes',
|
||||
'--model', model // e.g., 'opencode/big-pickle'
|
||||
];
|
||||
|
||||
Bun.spawn(args, {
|
||||
cwd,
|
||||
stdin: 'pipe',
|
||||
stdout: 'pipe', // JSON event stream
|
||||
stderr: 'pipe', // Debug logs
|
||||
});
|
||||
```
|
||||
|
||||
### RPC Protocol
|
||||
|
||||
```
|
||||
// Send prompt (stdin):
|
||||
{ "type": "prompt", "id": "request-uuid", "message": "Hello" }
|
||||
|
||||
{ "type": "abort", "id": "request-uuid" }
|
||||
|
||||
// Receive events (stdout):
|
||||
{ "type": "message_update", "assistantMessageEvent": { "type": "text_delta", "delta": "..." } }
|
||||
{ "type": "message_end" }
|
||||
{ "type": "tool_execution_start", "toolCallId": "...", "toolName": "Bash", "args": {...} }
|
||||
{ "type": "tool_execution_end", "toolCallId": "...", "result": "...", "isError": false }
|
||||
{ "type": "agent_end", "cost": {...} }
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Key Files Reference
|
||||
|
||||
| File | Purpose |
|
||||
|------|---------|
|
||||
| `server.tsx` | Main server, WebSocket upgrade, route setup |
|
||||
| `servers/api/pi/websocket.ts` | WebSocket message handling |
|
||||
| `servers/api/pi/session-manager.ts` | Session lifecycle in memory |
|
||||
| `servers/api/pi/pi-bridge.ts` | Spawn and communicate with Pi |
|
||||
| `servers/api/pi/storage.ts` | Disk persistence |
|
||||
| `servers/api/pi/rest.ts` | REST API endpoints |
|
||||
| `servers/api/pi/types.ts` | Shared types |
|
||||
| `workspaces/officerdev/src/hooks/usePiChat.ts` | Main frontend hook |
|
||||
| `workspaces/hooks/src/useChatWebSocket.ts` | WebSocket connection |
|
||||
| `workspaces/officerdev/src/apps/Chat/types.ts` | Frontend types |
|
||||
|
||||
---
|
||||
|
||||
## Configuration
|
||||
|
||||
### Environment Variables
|
||||
|
||||
- `PORT` - Server port (default: 5000)
|
||||
- `JWT_SECRET` - JWT signing secret
|
||||
|
||||
### Constants
|
||||
|
||||
```typescript
|
||||
// websocket.ts
|
||||
const IDLE_TIMEOUT_MS = 60 * 60 * 1000; // 1 hour
|
||||
|
||||
// storage.ts
|
||||
const PI_SESSIONS_DIR = '.pi-sessions';
|
||||
const GROUP_PREFIX = '@';
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Error Handling
|
||||
|
||||
- WebSocket errors → Logged and sent as `{ type: 'error', message: ... }`
|
||||
- Pi process errors → `{ type: 'error', message: errorMsg }`
|
||||
- Session not found → HTTP 404 or `SESSION_NOT_FOUND` error code
|
||||
- Token blacklist check → 401 Unauthorized
|
||||
|
||||
---
|
||||
|
||||
## Dependencies
|
||||
|
||||
### Backend
|
||||
- `bun` - Runtime
|
||||
- `hono` - HTTP framework
|
||||
- `drizzle-orm` - Database (for user auth)
|
||||
|
||||
### Frontend
|
||||
- `react` - UI framework
|
||||
- `@tanstack/react-query` - Data fetching
|
||||
- `lucide-react` - Icons
|
||||
|
||||
---
|
||||
|
||||
*Last updated: 2024*
|
||||
@@ -12,12 +12,6 @@ module.exports = {
|
||||
args: 'run src/servers/sidecar/claude/index.ts',
|
||||
watch: false,
|
||||
},
|
||||
{
|
||||
name: 'officer-pi',
|
||||
script: 'bun',
|
||||
args: 'run src/servers/sidecar/pi/index.ts',
|
||||
watch: false,
|
||||
},
|
||||
{
|
||||
name: 'officer-email',
|
||||
script: 'bun',
|
||||
|
||||
@@ -1,151 +0,0 @@
|
||||
# Pi Harness Integration Test
|
||||
|
||||
## Phase 5 Completion Checklist
|
||||
|
||||
### ✅ 1. WebSocket Integration
|
||||
- [x] WebSocket handler exported from `websocket.ts`
|
||||
- [x] Route `/api/pi/chat/ws` configured in `src/server.tsx`
|
||||
- [x] Authentication middleware applied via token query param
|
||||
- [x] Handler registered in handlers Map
|
||||
|
||||
### ✅ 2. REST API Integration
|
||||
- [x] REST router exported from `rest.ts`
|
||||
- [x] Router mounted in `src/servers/hono.ts` as `piRestRouter`
|
||||
- [x] Protected routes middleware applied
|
||||
- [x] All endpoints implemented:
|
||||
- GET `/api/pi/models` - List available models
|
||||
- POST `/api/pi/sessions` - List user sessions
|
||||
- GET `/api/pi/sessions/:sessionId` - Get session detail
|
||||
- PATCH `/api/pi/sessions/:sessionId` - Update session (rename)
|
||||
- DELETE `/api/pi/sessions/:sessionId` - Delete session
|
||||
- GET `/api/pi/sessions/search` - Search sessions
|
||||
|
||||
### ✅ 3. Type System
|
||||
- [x] All types defined in `src/servers/api/pi/types.ts`
|
||||
- [x] Legacy `chat-types.ts` updated to re-export new types
|
||||
- [x] Deprecation notice added to `chat-types.ts`
|
||||
- [x] No TypeScript compilation errors in Pi module
|
||||
|
||||
### ✅ 4. Architecture Verification
|
||||
```
|
||||
src/servers/api/pi/
|
||||
├── types.ts ✅ All message and session types
|
||||
├── storage.ts ✅ Session persistence to disk
|
||||
├── pi-bridge.ts ✅ Pi process spawning and RPC
|
||||
├── session-manager.ts ✅ In-memory session tracking
|
||||
├── websocket.ts ✅ WebSocket handler
|
||||
└── rest.ts ✅ REST endpoints
|
||||
```
|
||||
|
||||
## Manual Testing Steps
|
||||
|
||||
### Test 1: REST Endpoints
|
||||
```bash
|
||||
# Get JWT token
|
||||
TOKEN="your-jwt-token"
|
||||
|
||||
# List models
|
||||
curl -H "Authorization: Bearer $TOKEN" \
|
||||
http://localhost:5000/api/pi/models
|
||||
|
||||
# List sessions
|
||||
curl -X POST -H "Authorization: Bearer $TOKEN" \
|
||||
http://localhost:5000/api/pi/sessions
|
||||
|
||||
# Get session detail
|
||||
curl -H "Authorization: Bearer $TOKEN" \
|
||||
http://localhost:5000/api/pi/sessions/{sessionId}
|
||||
|
||||
# Update session title
|
||||
curl -X PATCH -H "Authorization: Bearer $TOKEN" \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"title": "New Title"}' \
|
||||
http://localhost:5000/api/pi/sessions/{sessionId}
|
||||
|
||||
# Delete session
|
||||
curl -X DELETE -H "Authorization: Bearer $TOKEN" \
|
||||
http://localhost:5000/api/pi/sessions/{sessionId}
|
||||
|
||||
# Search sessions
|
||||
curl -H "Authorization: Bearer $TOKEN" \
|
||||
"http://localhost:5000/api/pi/sessions/search?q=test"
|
||||
```
|
||||
|
||||
### Test 2: WebSocket Connection
|
||||
```javascript
|
||||
const token = "your-jwt-token";
|
||||
const ws = new WebSocket(`ws://localhost:5000/api/pi/chat/ws?token=${token}`);
|
||||
|
||||
ws.onopen = () => {
|
||||
console.log("Connected");
|
||||
// Start new chat
|
||||
ws.send(JSON.stringify({
|
||||
type: "chat",
|
||||
prompt: "Hello, world!",
|
||||
model: "gpt-4o",
|
||||
cwd: "/home/user/test"
|
||||
}));
|
||||
};
|
||||
|
||||
ws.onmessage = (event) => {
|
||||
const msg = JSON.parse(event.data);
|
||||
console.log("Received:", msg.type, msg);
|
||||
};
|
||||
```
|
||||
|
||||
### Test 3: Session Resume
|
||||
```javascript
|
||||
// First create a session (see Test 2)
|
||||
// Then close the WebSocket
|
||||
|
||||
// Reconnect and resume
|
||||
const ws2 = new WebSocket(`ws://localhost:5000/api/pi/chat/ws?token=${token}`);
|
||||
ws2.onopen = () => {
|
||||
ws2.send(JSON.stringify({
|
||||
type: "resume",
|
||||
sessionId: "previous-session-id"
|
||||
}));
|
||||
};
|
||||
|
||||
// Should receive sync:messages with full history
|
||||
```
|
||||
|
||||
### Test 4: Stop Generation
|
||||
```javascript
|
||||
ws.send(JSON.stringify({ type: "stop" }));
|
||||
// Should receive { type: "stopped" }
|
||||
```
|
||||
|
||||
## Expected Behaviors
|
||||
|
||||
### WebSocket Flow
|
||||
1. **Connect** → Authentication via token query param
|
||||
2. **Send chat** → Receive `session:init` → Stream of `assistant:delta` → `tool:*` events → `result`
|
||||
3. **Send resume** → Receive `sync:messages` with history
|
||||
4. **Send stop** → Generation stops → Receive `stopped`
|
||||
5. **Disconnect** → Idle timer starts (1 hour)
|
||||
|
||||
### Session Persistence
|
||||
- Sessions saved to `{cwd}/.pi-sessions/{sessionId}/`
|
||||
- `meta.json` - Session metadata
|
||||
- `messages.json` - Full conversation history
|
||||
- Updates written after each completed turn
|
||||
|
||||
### Error Handling
|
||||
- Invalid token → 401 Unauthorized
|
||||
- Invalid session ID → 404 Not Found
|
||||
- Pi process crash → Error message to client
|
||||
- Malformed JSON → Error message, connection stays open
|
||||
|
||||
## Integration Status
|
||||
|
||||
**Phase 5: COMPLETE ✅**
|
||||
|
||||
All components properly wired:
|
||||
- ✅ WebSocket handler integrated
|
||||
- ✅ REST endpoints integrated
|
||||
- ✅ Type system unified
|
||||
- ✅ No compilation errors
|
||||
- ✅ Architecture matches design document
|
||||
|
||||
**Next Steps**: Manual testing with real client to verify end-to-end functionality.
|
||||
@@ -1,656 +0,0 @@
|
||||
# Pi Harness — Architecture & Implementation Guide
|
||||
|
||||
**Version**: 1.0
|
||||
**Date**: February 20, 2026
|
||||
**Status**: Production
|
||||
|
||||
---
|
||||
|
||||
## Overview
|
||||
|
||||
The Pi harness is a unified WebSocket + REST API that manages long-running chat sessions with AI models through the Pi CLI tool. It replaces three legacy harnesses (Claude, OpenCode, Pi-Mono) with a single, streamlined implementation.
|
||||
|
||||
### Key Features
|
||||
|
||||
- **Session Management**: One Pi process per chat session, spawned on demand
|
||||
- **Resume from History**: Load conversations from disk and continue seamlessly
|
||||
- **Multi-Model Support**: Switch models per session (GPT, Claude, etc.)
|
||||
- **Concurrent Sessions**: Users can run multiple sessions simultaneously
|
||||
- **Idle Cleanup**: Automatic Pi process termination after 1 hour of inactivity
|
||||
- **Structured Logging**: Comprehensive logging with context for debugging
|
||||
|
||||
---
|
||||
|
||||
## Architecture
|
||||
|
||||
### Components
|
||||
|
||||
```
|
||||
┌─ Pi Harness
|
||||
│
|
||||
├─ websocket.ts WebSocket handler (chat lifecycle)
|
||||
├─ rest.ts REST API (sessions, models, search)
|
||||
├─ session-manager.ts In-memory session tracking
|
||||
├─ storage.ts Disk persistence (messages, metadata)
|
||||
├─ pi-bridge.ts Pi process spawning & RPC communication
|
||||
├─ types.ts TypeScript type definitions
|
||||
└─ logger.ts Structured logging utility
|
||||
```
|
||||
|
||||
### Data Flow
|
||||
|
||||
```
|
||||
Client
|
||||
↓ WebSocket (chat message)
|
||||
websocket.ts
|
||||
↓ Spawn Pi process if needed
|
||||
pi-bridge.ts → Pi CLI (RPC mode)
|
||||
↓ Stream events back
|
||||
session-manager.ts (track state)
|
||||
↓ Save to disk on completion
|
||||
storage.ts → {cwd}/.pi-sessions/{sessionId}/
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Session Lifecycle
|
||||
|
||||
### 1. New Chat
|
||||
|
||||
```typescript
|
||||
Client → { type: "chat", prompt: "Hello", model: "gpt-4o", cwd: "/home/user" }
|
||||
↓
|
||||
Server: Spawn Pi process
|
||||
Send prompt via RPC
|
||||
Stream responses (text, tool calls, results)
|
||||
Save session to disk
|
||||
↓
|
||||
Client ← { type: "session:init", sessionId: "...", model: "...", cwd: "..." }
|
||||
Client ← { type: "assistant:delta", text: "..." }
|
||||
Client ← { type: "result", sessionId: "...", cost: {...} }
|
||||
```
|
||||
|
||||
### 2. Resume Existing Chat
|
||||
|
||||
```typescript
|
||||
Client → { type: "resume", sessionId: "existing-session-id" }
|
||||
↓
|
||||
Server: Load session from disk
|
||||
Spawn fresh Pi process
|
||||
Inject conversation history as system context
|
||||
↓
|
||||
Client ← { type: "session:init", ... }
|
||||
Client ← { type: "sync:messages", messages: [...], isGenerating: false }
|
||||
```
|
||||
|
||||
### 3. WebSocket Disconnect
|
||||
|
||||
```typescript
|
||||
Client disconnects
|
||||
↓
|
||||
Server: Detach WebSocket from session
|
||||
Start 1-hour idle timer
|
||||
(Session and Pi process remain alive)
|
||||
```
|
||||
|
||||
### 4. Idle Timeout
|
||||
|
||||
```typescript
|
||||
1 hour passes without reconnection
|
||||
↓
|
||||
Server: Kill Pi process
|
||||
Save session to disk (if not already saved)
|
||||
Remove from memory
|
||||
```
|
||||
|
||||
### 5. Stop Generation
|
||||
|
||||
```typescript
|
||||
Client → { type: "stop" }
|
||||
↓
|
||||
Server: Send abort RPC to Pi process
|
||||
Set isGenerating = false
|
||||
↓
|
||||
Client ← { type: "stopped" }
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Wire Protocol
|
||||
|
||||
### Client → Server Messages
|
||||
|
||||
#### Chat Message
|
||||
```json
|
||||
{
|
||||
"type": "chat",
|
||||
"prompt": "Explain TypeScript generics",
|
||||
"sessionId": "uuid", // optional: omit for new chat
|
||||
"model": "gpt-4o", // optional: defaults to gpt-4o
|
||||
"cwd": "/home/user/project", // optional: defaults to user home
|
||||
"attachmentIds": ["file-1"] // optional: file references
|
||||
}
|
||||
```
|
||||
|
||||
#### Resume Session
|
||||
```json
|
||||
{
|
||||
"type": "resume",
|
||||
"sessionId": "existing-uuid"
|
||||
}
|
||||
```
|
||||
|
||||
#### Stop Generation
|
||||
```json
|
||||
{
|
||||
"type": "stop"
|
||||
}
|
||||
```
|
||||
|
||||
### Server → Client Messages
|
||||
|
||||
#### Session Initialized
|
||||
```json
|
||||
{
|
||||
"type": "session:init",
|
||||
"sessionId": "uuid",
|
||||
"model": "gpt-4o",
|
||||
"cwd": "/home/user/project"
|
||||
}
|
||||
```
|
||||
|
||||
#### Assistant Text (Complete Block)
|
||||
```json
|
||||
{
|
||||
"type": "assistant:text",
|
||||
"text": "TypeScript generics allow..."
|
||||
}
|
||||
```
|
||||
|
||||
#### Assistant Text (Streaming Delta)
|
||||
```json
|
||||
{
|
||||
"type": "assistant:delta",
|
||||
"text": "you to"
|
||||
}
|
||||
```
|
||||
|
||||
#### Tool Execution Started
|
||||
```json
|
||||
{
|
||||
"type": "tool:start",
|
||||
"toolCallId": "call-uuid",
|
||||
"toolName": "bash",
|
||||
"toolInput": { "command": "ls -la" }
|
||||
}
|
||||
```
|
||||
|
||||
#### Tool Execution Result
|
||||
```json
|
||||
{
|
||||
"type": "tool:result",
|
||||
"toolCallId": "call-uuid",
|
||||
"output": "file1.txt\nfile2.txt",
|
||||
"isError": false
|
||||
}
|
||||
```
|
||||
|
||||
#### Generation Completed
|
||||
```json
|
||||
{
|
||||
"type": "result",
|
||||
"sessionId": "uuid",
|
||||
"cost": {
|
||||
"inputTokens": 500,
|
||||
"outputTokens": 300,
|
||||
"totalUSD": 0.012
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### Full Sync (Resume)
|
||||
```json
|
||||
{
|
||||
"type": "sync:messages",
|
||||
"sessionId": "uuid",
|
||||
"messages": [
|
||||
{ "id": "msg-1", "role": "user", "text": "Hello", "timestamp": 1708396000000 },
|
||||
{ "id": "msg-2", "role": "assistant", "text": "Hi!", "timestamp": 1708396001000, "model": "gpt-4o" }
|
||||
],
|
||||
"isGenerating": false,
|
||||
"streamingText": ""
|
||||
}
|
||||
```
|
||||
|
||||
#### Error
|
||||
```json
|
||||
{
|
||||
"type": "error",
|
||||
"message": "Failed to spawn Pi process",
|
||||
"errorCode": "SPAWN_ERROR" // optional
|
||||
}
|
||||
```
|
||||
|
||||
#### Generation Stopped
|
||||
```json
|
||||
{
|
||||
"type": "stopped"
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Session Storage
|
||||
|
||||
### Directory Structure
|
||||
|
||||
```
|
||||
{userCwd}/.pi-sessions/
|
||||
└── {sessionId}/
|
||||
├── meta.json (metadata: title, model, cwd, timestamps, cost)
|
||||
└── messages.json (full conversation history)
|
||||
```
|
||||
|
||||
### meta.json
|
||||
|
||||
```json
|
||||
{
|
||||
"id": "session-uuid",
|
||||
"title": "First 100 characters of initial prompt",
|
||||
"model": "gpt-4o",
|
||||
"cwd": "/home/user/project",
|
||||
"createdAt": 1708396000000,
|
||||
"updatedAt": 1708396123000,
|
||||
"messageCount": 15,
|
||||
"cost": {
|
||||
"inputTokens": 5000,
|
||||
"outputTokens": 3000,
|
||||
"totalUSD": 0.15
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### messages.json
|
||||
|
||||
```json
|
||||
{
|
||||
"messages": [
|
||||
{
|
||||
"id": "msg-uuid-1",
|
||||
"timestamp": 1708396000000,
|
||||
"role": "user",
|
||||
"text": "Hello, can you help me?"
|
||||
},
|
||||
{
|
||||
"id": "msg-uuid-2",
|
||||
"timestamp": 1708396001000,
|
||||
"role": "assistant",
|
||||
"text": "Of course! What do you need?",
|
||||
"model": "gpt-4o",
|
||||
"cost": {
|
||||
"inputTokens": 100,
|
||||
"outputTokens": 20,
|
||||
"totalUSD": 0.001
|
||||
}
|
||||
},
|
||||
{
|
||||
"id": "msg-uuid-3",
|
||||
"timestamp": 1708396002000,
|
||||
"role": "tool",
|
||||
"toolCallId": "call-uuid",
|
||||
"toolName": "bash",
|
||||
"output": "file1.txt\nfile2.txt",
|
||||
"isError": false
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## REST API Endpoints
|
||||
|
||||
### List Sessions
|
||||
```http
|
||||
POST /api/pi/sessions
|
||||
Authorization: Bearer {token}
|
||||
|
||||
Response: { sessions: SessionMeta[] }
|
||||
```
|
||||
|
||||
### Get Session Detail
|
||||
```http
|
||||
GET /api/pi/sessions/{sessionId}
|
||||
Authorization: Bearer {token}
|
||||
|
||||
Response: { session: { ...meta, messages: Message[] } }
|
||||
```
|
||||
|
||||
### Update Session (Rename)
|
||||
```http
|
||||
PATCH /api/pi/sessions/{sessionId}
|
||||
Authorization: Bearer {token}
|
||||
Body: { title: "New Title" }
|
||||
|
||||
Response: { success: true, session: SessionMeta }
|
||||
```
|
||||
|
||||
### Delete Session
|
||||
```http
|
||||
DELETE /api/pi/sessions/{sessionId}
|
||||
Authorization: Bearer {token}
|
||||
|
||||
Response: { success: true }
|
||||
```
|
||||
|
||||
### Search Sessions
|
||||
```http
|
||||
GET /api/pi/sessions/search?q={query}
|
||||
Authorization: Bearer {token}
|
||||
|
||||
Response: { results: SessionMeta[] }
|
||||
```
|
||||
|
||||
### List Models
|
||||
```http
|
||||
GET /api/pi/models
|
||||
Authorization: Bearer {token}
|
||||
|
||||
Response: { models: ModelInfo[] }
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Error Handling
|
||||
|
||||
### Edge Cases & Solutions
|
||||
|
||||
#### 1. Corrupted messages.json
|
||||
**Problem**: JSON parse error when loading session
|
||||
**Solution**: Catch error, return `SESSION_NOT_FOUND` to client, allow deletion
|
||||
|
||||
```typescript
|
||||
try {
|
||||
const { meta, messages } = await storage.loadSession(homeDir, sessionId);
|
||||
// ...
|
||||
} catch (err) {
|
||||
logger.error('Failed to load session from disk', { sessionId, error: String(err) });
|
||||
ws.send({ type: 'error', message: 'Session not found', errorCode: 'SESSION_NOT_FOUND' });
|
||||
}
|
||||
```
|
||||
|
||||
#### 2. Pi Process Crash During Streaming
|
||||
**Problem**: Pi process exits unexpectedly while generating
|
||||
**Solution**: `readEvents` generator exits naturally, session marked as `isGenerating: false`
|
||||
|
||||
```typescript
|
||||
for await (const event of piBridge.readEvents(session.piProcess)) {
|
||||
// Handle events...
|
||||
}
|
||||
// If process dies, loop exits and session.isGenerating remains false
|
||||
```
|
||||
|
||||
#### 3. WebSocket Disconnect During Generation
|
||||
**Problem**: User closes browser while AI is generating
|
||||
**Solution**: Session continues in background until Pi completes, then auto-saves
|
||||
|
||||
```typescript
|
||||
// In close handler:
|
||||
sessionManager.detachWs(sessionId);
|
||||
sessionManager.setIdleTimeout(sessionId, IDLE_TIMEOUT_MS);
|
||||
// Pi process keeps running, will clean up after 1 hour
|
||||
```
|
||||
|
||||
#### 4. Concurrent WebSocket Connections to Same Session
|
||||
**Problem**: User opens same session in two tabs
|
||||
**Solution**: `attachWs` overwrites previous WebSocket reference, old tab receives no updates
|
||||
|
||||
```typescript
|
||||
// Only the most recent WebSocket receives updates
|
||||
sessionManager.attachWs(sessionId, ws);
|
||||
```
|
||||
|
||||
#### 5. CWD Resolution When Not Provided
|
||||
**Problem**: Client doesn't send `cwd` parameter
|
||||
**Solution**: Default to user's home directory from email
|
||||
|
||||
```typescript
|
||||
const cwd = msg.cwd || getHomeDir(email);
|
||||
```
|
||||
|
||||
#### 6. Session Save Failure
|
||||
**Problem**: Disk write fails (permissions, disk full)
|
||||
**Solution**: Log error but continue, session remains in memory
|
||||
|
||||
```typescript
|
||||
try {
|
||||
await storage.saveSession(cwd, sessionId, session.meta, session.messages);
|
||||
} catch (err) {
|
||||
logger.error('Failed to save session', { sessionId, error: String(err) });
|
||||
// Continue — session still in memory
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Logging
|
||||
|
||||
### Structured Logging Format
|
||||
|
||||
```typescript
|
||||
logger.info('Session saved to disk', { sessionId, messageCount: 15 });
|
||||
// Output:
|
||||
// [2026-02-20T19:49:00.000Z] [Pi] [INFO] Session saved to disk {"sessionId":"...","messageCount":15}
|
||||
```
|
||||
|
||||
### Log Levels
|
||||
|
||||
- **DEBUG**: Detailed flow information (currently unused)
|
||||
- **INFO**: Normal operations (spawns, saves, connections)
|
||||
- **WARN**: Recoverable issues (currently unused)
|
||||
- **ERROR**: Failures requiring attention (parse errors, spawn failures)
|
||||
|
||||
### Key Log Points
|
||||
|
||||
- WebSocket open/close
|
||||
- Session creation/resumption
|
||||
- Pi process spawn/kill
|
||||
- Session save/load
|
||||
- Errors in message handling, streaming, RPC communication
|
||||
|
||||
---
|
||||
|
||||
## Performance Considerations
|
||||
|
||||
### Memory Usage
|
||||
|
||||
- Each active session holds:
|
||||
- Full message history (in-memory)
|
||||
- Pi process (separate OS process)
|
||||
- WebSocket connection (if attached)
|
||||
|
||||
**Mitigation**: Idle timeout (1 hour) cleans up inactive sessions
|
||||
|
||||
### Disk I/O
|
||||
|
||||
- Sessions saved synchronously after each AI response
|
||||
- File writes are fast (JSON serialization)
|
||||
- No buffering — each message persisted immediately
|
||||
|
||||
**Optimization**: Could batch writes or use async queue (not currently needed)
|
||||
|
||||
### Pi Process Overhead
|
||||
|
||||
- Each session spawns one `pi` CLI process
|
||||
- Process runs in `--mode rpc --no-extensions --no-skills`
|
||||
- Minimal resource usage when idle
|
||||
|
||||
**Limitation**: Max concurrent sessions limited by system resources
|
||||
|
||||
---
|
||||
|
||||
## Development Guide
|
||||
|
||||
### Adding a New Message Type
|
||||
|
||||
1. Add type to `types.ts`:
|
||||
```typescript
|
||||
export type ServerMessage =
|
||||
| { type: 'new-type'; data: string }
|
||||
| ...;
|
||||
```
|
||||
|
||||
2. Handle in `websocket.ts`:
|
||||
```typescript
|
||||
if (event.type === 'new-type') {
|
||||
const msg: ServerMessage = { type: 'new-type', data: event.data };
|
||||
ws.send(JSON.stringify(msg));
|
||||
}
|
||||
```
|
||||
|
||||
3. Update Pi bridge if needed (`pi-bridge.ts`)
|
||||
|
||||
### Adding a New REST Endpoint
|
||||
|
||||
1. Add route in `rest.ts`:
|
||||
```typescript
|
||||
piRestRouter.get('/pi/new-endpoint', async (ctx: Context) => {
|
||||
const email = ctx.get('email');
|
||||
// Implementation
|
||||
return ctx.json({ result: '...' });
|
||||
});
|
||||
```
|
||||
|
||||
2. Wire into `hono.ts`:
|
||||
```typescript
|
||||
protectedRouter.route('/', piRestRouter);
|
||||
```
|
||||
|
||||
### Debugging Tips
|
||||
|
||||
- Check logs for session lifecycle events
|
||||
- Verify Pi process spawned: `ps aux | grep "pi --mode rpc"`
|
||||
- Inspect session files: `cat ~/.pi-sessions/{sessionId}/meta.json`
|
||||
- Test WebSocket manually: `wscat -c ws://localhost:3000/api/pi/chat/ws`
|
||||
|
||||
---
|
||||
|
||||
## Migration from Legacy Harnesses
|
||||
|
||||
### Differences from Claude/OpenCode/Pi-Mono
|
||||
|
||||
| Feature | Legacy Harnesses | New Pi Harness |
|
||||
|---------|-----------------|----------------|
|
||||
| Processes | 3 separate handlers | Single unified handler |
|
||||
| Session Storage | Per-provider directories | Unified `.pi-sessions/` |
|
||||
| Resume | Limited support | Full history injection |
|
||||
| Models | Provider-specific | Any model via Pi CLI |
|
||||
| WebSocket Protocol | Different per provider | Unified wire protocol |
|
||||
| REST API | Scattered endpoints | Centralized `/api/pi/*` |
|
||||
|
||||
### Migration Checklist
|
||||
|
||||
- [x] Delete old harness code (`src/servers/api/claude`, etc.)
|
||||
- [x] Remove old routes from `hono.ts` and `server.tsx`
|
||||
- [x] Update frontend to use new WebSocket protocol
|
||||
- [x] Migrate old session storage to new format (manual or script)
|
||||
- [x] Update user settings to reference only 'pi' provider
|
||||
|
||||
---
|
||||
|
||||
## Testing
|
||||
|
||||
### Manual Testing
|
||||
|
||||
```bash
|
||||
# 1. Start server
|
||||
bun dev
|
||||
|
||||
# 2. Test WebSocket (in separate terminal)
|
||||
wscat -c "ws://localhost:3000/api/pi/chat/ws?token=your-jwt-token"
|
||||
|
||||
# Send chat message
|
||||
> {"type":"chat","prompt":"Hello!","model":"gpt-4o"}
|
||||
|
||||
# 3. Test REST API
|
||||
curl -H "Authorization: Bearer your-jwt-token" \
|
||||
http://localhost:3000/api/pi/models
|
||||
|
||||
curl -H "Authorization: Bearer your-jwt-token" \
|
||||
http://localhost:3000/api/pi/sessions
|
||||
```
|
||||
|
||||
### Integration Tests (TODO)
|
||||
|
||||
- Session creation and resumption
|
||||
- Concurrent sessions per user
|
||||
- Idle timeout cleanup
|
||||
- WebSocket disconnect/reconnect
|
||||
- Error handling (corrupt files, Pi crashes)
|
||||
|
||||
---
|
||||
|
||||
## Future Improvements
|
||||
|
||||
### Potential Enhancements
|
||||
|
||||
1. **Streaming Buffer Optimization**: Batch small deltas to reduce WebSocket overhead
|
||||
2. **Session Compression**: Gzip old messages to save disk space
|
||||
3. **Model Auto-Detection**: Dynamically discover available models from Pi CLI
|
||||
4. **Cost Tracking Dashboard**: Aggregate cost data across sessions
|
||||
5. **Session Export**: Export conversations to Markdown or JSON
|
||||
6. **Real-time Collaboration**: Multiple users in same session
|
||||
7. **Attachment Support**: Full implementation of file attachments
|
||||
|
||||
---
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Common Issues
|
||||
|
||||
**Issue**: "Failed to spawn Pi process"
|
||||
**Solution**: Verify `pi` CLI is installed and in PATH
|
||||
**Check**: `which pi` → should return path
|
||||
|
||||
**Issue**: "Session not found" on resume
|
||||
**Solution**: Check session files exist in `{cwd}/.pi-sessions/{sessionId}/`
|
||||
**Check**: `ls ~/.pi-sessions/`
|
||||
|
||||
**Issue**: WebSocket disconnects immediately
|
||||
**Solution**: Verify JWT token is valid and not expired
|
||||
**Check**: Decode token, check `exp` claim
|
||||
|
||||
**Issue**: High memory usage
|
||||
**Solution**: Check idle timeout is working, verify old sessions cleaned up
|
||||
**Check**: `ps aux | grep pi` → should show minimal processes
|
||||
|
||||
---
|
||||
|
||||
## Appendix
|
||||
|
||||
### Type Definitions
|
||||
|
||||
See `types.ts` for complete type definitions:
|
||||
- `ClientMessage`
|
||||
- `ServerMessage`
|
||||
- `Message`
|
||||
- `SessionMeta`
|
||||
- `UserSession`
|
||||
- `PiEvent`
|
||||
- `ModelInfo`
|
||||
- `MessageCost`
|
||||
|
||||
### Configuration
|
||||
|
||||
| Setting | Value | Environment Variable |
|
||||
|---------|-------|---------------------|
|
||||
| Idle Timeout | 1 hour | N/A (hardcoded) |
|
||||
| Session Directory | `{cwd}/.pi-sessions/` | N/A |
|
||||
| Default Model | `gpt-4o` | N/A |
|
||||
| Pi RPC Mode | `--mode rpc --no-extensions --no-skills` | N/A |
|
||||
|
||||
---
|
||||
|
||||
**Last Updated**: February 20, 2026
|
||||
**Maintainer**: Pi Harness Team
|
||||
**Questions?** Check logs, read code, ask the team.
|
||||
@@ -1,133 +0,0 @@
|
||||
# Snap Node Compatibility Issue
|
||||
|
||||
## Problem
|
||||
|
||||
When using Officer with **snap node** (`/snap/bin/node`), the Pi harness fails with the following error:
|
||||
|
||||
```
|
||||
[Pi] [INFO] Pi process exited
|
||||
code=1
|
||||
```
|
||||
|
||||
This occurs on the first chat message, before Pi even processes the command.
|
||||
|
||||
## Root Cause
|
||||
|
||||
The snap version of Node.js has an incompatibility with how Bun's `spawn()` function sets up piped stdin file descriptors. When Officer tries to spawn a Pi process with `stdin: 'pipe'`, the process immediately exits with code 1, preventing the RPC communication from working.
|
||||
|
||||
This does **NOT** happen with:
|
||||
- System node installed via apt/package manager
|
||||
- NodeSource node
|
||||
- Homebrew node (on macOS)
|
||||
- Any non-snap Node.js installation
|
||||
|
||||
## Solution
|
||||
|
||||
**Uninstall snap node and install a system-managed version instead:**
|
||||
|
||||
### Step 1: Remove snap node
|
||||
|
||||
```bash
|
||||
sudo snap remove node
|
||||
```
|
||||
|
||||
### Step 2: Install Node.js via apt (recommended)
|
||||
|
||||
```bash
|
||||
# Add NodeSource repository for Node 20 LTS
|
||||
curl -fsSL https://deb.nodesource.com/setup_20.x | sudo -E bash -
|
||||
|
||||
# Install Node.js
|
||||
sudo apt-get install -y nodejs
|
||||
|
||||
# Verify installation
|
||||
node --version
|
||||
which node # Should be /usr/bin/node (NOT /snap/bin/node)
|
||||
```
|
||||
|
||||
### Alternative: Using system apt repository
|
||||
|
||||
If NodeSource is unavailable in your region:
|
||||
|
||||
```bash
|
||||
sudo apt-get update
|
||||
sudo apt-get install -y nodejs npm
|
||||
```
|
||||
|
||||
### Alternative: Using nvm (Node Version Manager)
|
||||
|
||||
For more control over Node.js versions:
|
||||
|
||||
```bash
|
||||
# Install nvm
|
||||
curl -o- https://raw.githubusercontent.com/nvm-sh/nvm/v0.39.0/install.sh | bash
|
||||
|
||||
# Install Node LTS
|
||||
nvm install --lts
|
||||
nvm use --lts
|
||||
|
||||
# Verify
|
||||
which node # Should be ~/.nvm/versions/node/*/bin/node
|
||||
```
|
||||
|
||||
## Verification
|
||||
|
||||
After installing a non-snap Node.js:
|
||||
|
||||
```bash
|
||||
# Verify node is not from snap
|
||||
which node
|
||||
# Output should NOT contain "/snap/"
|
||||
|
||||
# Verify node works
|
||||
node --version
|
||||
|
||||
# Clear npm cache
|
||||
npm cache clean --force
|
||||
npm install -g pi-coding-agent
|
||||
```
|
||||
|
||||
Then restart Officer and try the chat functionality again - it should work!
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Still seeing the error after reinstalling Node?
|
||||
|
||||
1. **Restart Officer service** (if running as a service):
|
||||
```bash
|
||||
sudo systemctl restart officer
|
||||
# or
|
||||
bun dev # if running locally
|
||||
```
|
||||
|
||||
2. **Verify Bun can find the correct node**:
|
||||
```bash
|
||||
bun run "which node"
|
||||
which node
|
||||
# Both should show the same path, not /snap/bin/node
|
||||
```
|
||||
|
||||
3. **Check Pi installation**:
|
||||
```bash
|
||||
pi --version
|
||||
pi --list-models
|
||||
```
|
||||
|
||||
### Error logs to look for
|
||||
|
||||
If you still see the error, check Officer logs for:
|
||||
|
||||
```
|
||||
Pi process exited with code 1
|
||||
nodeVersion: ...
|
||||
nodeExePath: /snap/bin/node
|
||||
isSnapNode: true
|
||||
```
|
||||
|
||||
This confirms snap node is the issue.
|
||||
|
||||
## Why snap node has this issue
|
||||
|
||||
The snap package environment isolates certain system calls and file descriptor handling, which conflicts with Bun's pipe setup mechanism. The snap version of Node.js doesn't properly inherit file descriptor flags when pipes are created by Bun, causing the process to fail on startup.
|
||||
|
||||
This is a known incompatibility and not a bug in Officer or Pi itself.
|
||||
@@ -1,118 +1,16 @@
|
||||
import type { ModelInfo } from './types';
|
||||
import { PI_CONFIG_DIR } from '../../data-path';
|
||||
import { logger } from './logger';
|
||||
|
||||
// Claude-only. The runner is the `claude` CLI, so the model list is a fixed set of Claude tiers.
|
||||
const CLAUDE_CODE_MODELS: ModelInfo[] = [
|
||||
{ id: 'claude-code/opus', name: 'opus', provider: 'claude-code', contextWindow: 200000, maxTokens: 16000, reasoning: true, images: true },
|
||||
{ id: 'claude-code/sonnet', name: 'sonnet', provider: 'claude-code', contextWindow: 200000, maxTokens: 16000, reasoning: true, images: true },
|
||||
{ id: 'claude-code/haiku', name: 'haiku', provider: 'claude-code', contextWindow: 200000, maxTokens: 8192, reasoning: false, images: true },
|
||||
];
|
||||
|
||||
const CACHE_TTL_MS = 60_000;
|
||||
|
||||
let cachedModels: ModelInfo[] | null = null;
|
||||
let cacheTimestamp = 0;
|
||||
|
||||
export function invalidateModelCache(): void {
|
||||
cachedModels = null;
|
||||
cacheTimestamp = 0;
|
||||
// No-op: the Claude model list is static. Kept for call-site compatibility.
|
||||
}
|
||||
|
||||
const parseSize = (s?: string): number => {
|
||||
if (!s) return 128000;
|
||||
const match = s.match(/^([\d.]+)([KMG])?$/i);
|
||||
if (!match) return 128000;
|
||||
const num = parseFloat(match[1]!);
|
||||
const unit = (match[2] ?? '').toUpperCase();
|
||||
if (unit === 'K') return Math.round(num * 1000);
|
||||
if (unit === 'M') return Math.round(num * 1000000);
|
||||
if (unit === 'G') return Math.round(num * 1000000000);
|
||||
return Math.round(num);
|
||||
};
|
||||
|
||||
export async function listPiModels(): Promise<ModelInfo[]> {
|
||||
if (cachedModels && Date.now() - cacheTimestamp < CACHE_TTL_MS) {
|
||||
return [...cachedModels, ...CLAUDE_CODE_MODELS];
|
||||
}
|
||||
|
||||
try {
|
||||
// Resolve absolute path to pi binary (PATH may differ under pm2/systemd)
|
||||
const piBin = (() => {
|
||||
const r = Bun.spawnSync({ cmd: ['which', 'pi'], stdout: 'pipe', stderr: 'ignore' });
|
||||
return r.stdout.toString().trim() || 'pi';
|
||||
})();
|
||||
|
||||
// Use spawnSync — Bun.spawn (async) loses stdout under pm2
|
||||
const proc = Bun.spawnSync({
|
||||
cmd: [piBin, '--list-models'],
|
||||
stdout: 'pipe',
|
||||
stderr: 'pipe',
|
||||
env: { ...process.env, PI_CODING_AGENT_DIR: PI_CONFIG_DIR },
|
||||
});
|
||||
|
||||
const output = proc.stdout.toString() || proc.stderr.toString();
|
||||
|
||||
if (proc.exitCode !== 0) {
|
||||
const stderrText = proc.stderr.toString();
|
||||
logger.error('pi --list-models failed', { exitCode: proc.exitCode, stderr: stderrText.trim(), piBin });
|
||||
return [...CLAUDE_CODE_MODELS];
|
||||
}
|
||||
|
||||
const lines = output.trim().split('\n');
|
||||
if (lines.length < 2) return [...CLAUDE_CODE_MODELS];
|
||||
|
||||
// Parse fixed-width table: provider, model, context, max-out, thinking, images
|
||||
const header = lines[0]!;
|
||||
const colStarts = [
|
||||
header.indexOf('provider'),
|
||||
header.indexOf('model'),
|
||||
header.indexOf('context'),
|
||||
header.indexOf('max-out'),
|
||||
header.indexOf('thinking'),
|
||||
header.indexOf('images'),
|
||||
];
|
||||
|
||||
const extractCol = (line: string, colIdx: number): string => {
|
||||
const start = colStarts[colIdx]!;
|
||||
const end = colIdx < colStarts.length - 1 ? colStarts[colIdx + 1]! : line.length;
|
||||
return line.slice(start, end).trim();
|
||||
};
|
||||
|
||||
const models: ModelInfo[] = [];
|
||||
const seen = new Set<string>();
|
||||
|
||||
for (let i = 1; i < lines.length; i++) {
|
||||
const line = lines[i]!;
|
||||
if (!line.trim()) continue;
|
||||
|
||||
const provider = extractCol(line, 0);
|
||||
const model = extractCol(line, 1);
|
||||
const context = extractCol(line, 2);
|
||||
const maxOut = extractCol(line, 3);
|
||||
const thinking = extractCol(line, 4);
|
||||
const images = extractCol(line, 5);
|
||||
|
||||
const dedupeKey = `${provider}/${model}`;
|
||||
if (seen.has(dedupeKey)) continue;
|
||||
seen.add(dedupeKey);
|
||||
|
||||
models.push({
|
||||
id: `${provider}/${model}`,
|
||||
name: model,
|
||||
provider,
|
||||
contextWindow: parseSize(context),
|
||||
maxTokens: parseSize(maxOut),
|
||||
reasoning: thinking === 'yes',
|
||||
images: images === 'yes',
|
||||
});
|
||||
}
|
||||
|
||||
logger.info('pi --list-models returned', { count: models.length });
|
||||
cachedModels = models;
|
||||
cacheTimestamp = Date.now();
|
||||
return [...models, ...CLAUDE_CODE_MODELS];
|
||||
} catch (err) {
|
||||
logger.error('Failed to run pi --list-models', { error: String(err) });
|
||||
return [...CLAUDE_CODE_MODELS];
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,401 +0,0 @@
|
||||
import { join } from 'path';
|
||||
import { readdirSync, existsSync, mkdirSync } from 'node:fs';
|
||||
import type { Subprocess } from 'bun';
|
||||
import type { PiEvent, MessageCost } from './types';
|
||||
import { PI_CONFIG_DIR, DATA_PATH, getHomeDirForRole, itemsDir } from '../../data-path';
|
||||
import { logger } from './logger';
|
||||
|
||||
// Resolve pi as [node, cli.js] — Bun.spawn async pipes break with shebang scripts under pm2
|
||||
const PI_CMD = (() => {
|
||||
const whichResult = Bun.spawnSync({ cmd: ['which', 'pi'], stdout: 'pipe', stderr: 'ignore' });
|
||||
const piBin = whichResult.stdout.toString().trim() || 'pi';
|
||||
// Follow symlink to get the actual .js file, then invoke via node directly
|
||||
const readlinkResult = Bun.spawnSync({ cmd: ['readlink', '-f', piBin], stdout: 'pipe', stderr: 'ignore' });
|
||||
const realPath = readlinkResult.stdout.toString().trim();
|
||||
const nodeResult = Bun.spawnSync({ cmd: ['which', 'node'], stdout: 'pipe', stderr: 'ignore' });
|
||||
const nodeBin = nodeResult.stdout.toString().trim() || 'node';
|
||||
if (realPath && realPath.endsWith('.js')) {
|
||||
return [nodeBin, realPath];
|
||||
}
|
||||
// Fallback: use pi binary directly (works for non-pm2 environments)
|
||||
return [piBin];
|
||||
})();
|
||||
|
||||
export type PiEventHandler = (event: PiEvent) => void;
|
||||
|
||||
function collectSkillFlags(): string[] {
|
||||
const flags: string[] = [];
|
||||
const dirs = [itemsDir('skills')];
|
||||
|
||||
for (const dir of dirs) {
|
||||
if (!existsSync(dir)) continue;
|
||||
for (const entry of readdirSync(dir, { withFileTypes: true })) {
|
||||
if (!entry.isDirectory()) continue;
|
||||
if (existsSync(join(dir, entry.name, 'SKILL.md'))) {
|
||||
flags.push('--skill', `${dir}/${entry.name}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return flags;
|
||||
}
|
||||
|
||||
function collectExtensionFlags(): string[] {
|
||||
const flags: string[] = [];
|
||||
const dirs = [itemsDir('extensions')];
|
||||
|
||||
for (const dir of dirs) {
|
||||
if (!existsSync(dir)) continue;
|
||||
for (const entry of readdirSync(dir, { withFileTypes: true })) {
|
||||
if (!entry.isDirectory()) continue;
|
||||
if (existsSync(join(dir, entry.name, 'index.ts'))) {
|
||||
flags.push('--extension', `${dir}/${entry.name}/index.ts`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return flags;
|
||||
}
|
||||
|
||||
async function resolveApiKeyForModel(model: string): Promise<string | null> {
|
||||
const provider = model.split('/')[0];
|
||||
if (!provider) return null;
|
||||
try {
|
||||
const authFile = Bun.file(join(PI_CONFIG_DIR, 'auth.json'));
|
||||
if (!(await authFile.exists())) return null;
|
||||
const auth = (await authFile.json()) as Record<string, { key?: string }>;
|
||||
return auth[provider]?.key?.trim() || null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
type SpawnPiOptions = {
|
||||
sessionFile?: string;
|
||||
role?: string;
|
||||
};
|
||||
|
||||
export async function spawnPi(
|
||||
cwd: string,
|
||||
model: string,
|
||||
userId: number,
|
||||
email: string,
|
||||
onEvent: PiEventHandler,
|
||||
options?: SpawnPiOptions,
|
||||
): Promise<Subprocess> {
|
||||
const skillFlags = collectSkillFlags();
|
||||
const extensionFlags = collectExtensionFlags();
|
||||
|
||||
const piArgs = [
|
||||
...PI_CMD,
|
||||
'--mode',
|
||||
'rpc',
|
||||
'--no-skills',
|
||||
'--no-prompt-templates',
|
||||
'--no-themes',
|
||||
...skillFlags,
|
||||
...extensionFlags,
|
||||
];
|
||||
if (model) piArgs.push('--model', model);
|
||||
if (options?.sessionFile) piArgs.push('--session', options.sessionFile);
|
||||
|
||||
const apiKey = await resolveApiKeyForModel(model);
|
||||
if (apiKey) piArgs.push('--api-key', apiKey);
|
||||
|
||||
if (!existsSync(cwd)) {
|
||||
mkdirSync(cwd, { recursive: true });
|
||||
}
|
||||
|
||||
const homeDir = getHomeDirForRole(email, options?.role ?? null);
|
||||
const toolsDirs = itemsDir('tools');
|
||||
|
||||
const env: Record<string, string> = {
|
||||
HOME: process.env.HOME ?? '',
|
||||
OFFICER_USER_HOME: homeDir,
|
||||
OFFICER_USER_ROOT: join(DATA_PATH, email),
|
||||
PI_CODING_AGENT_DIR: PI_CONFIG_DIR,
|
||||
PI_TOOLS_DIRS: toolsDirs,
|
||||
OFFICER_EMAIL_DB: join(DATA_PATH, email, 'emails.db'),
|
||||
TERM: 'xterm-256color',
|
||||
PATH: process.env.PATH ?? '',
|
||||
};
|
||||
|
||||
const proc = Bun.spawn(piArgs, {
|
||||
cwd,
|
||||
stdin: 'pipe',
|
||||
stdout: 'pipe',
|
||||
stderr: 'pipe',
|
||||
env: { ...process.env, ...env },
|
||||
});
|
||||
|
||||
logger.info('Spawned Pi', {
|
||||
model,
|
||||
skills: skillFlags.filter((f) => f !== '--skill').length,
|
||||
extensions: extensionFlags.filter((f) => f !== '--extension').length,
|
||||
});
|
||||
|
||||
// Read stdout JSON event stream (runs in background)
|
||||
const stdout = proc.stdout as ReadableStream<Uint8Array>;
|
||||
const reader = stdout.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = '';
|
||||
let streamBuffer = '';
|
||||
|
||||
(async () => {
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
const lines = buffer.split('\n');
|
||||
buffer = lines.pop() ?? '';
|
||||
|
||||
for (const line of lines) {
|
||||
if (!line.trim()) continue;
|
||||
try {
|
||||
const event = JSON.parse(line) as Record<string, unknown>;
|
||||
const piEvents = parsePiEvent(event, streamBuffer);
|
||||
for (const piEvent of piEvents) {
|
||||
if (piEvent.type === 'delta') {
|
||||
streamBuffer += piEvent.text;
|
||||
} else if (piEvent.type === 'text' || piEvent.type === 'tool:start') {
|
||||
streamBuffer = '';
|
||||
}
|
||||
onEvent(piEvent);
|
||||
}
|
||||
} catch {
|
||||
// Skip unparseable lines
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// Process ended
|
||||
}
|
||||
})();
|
||||
|
||||
// Stderr → debug log
|
||||
const stderr = proc.stderr as ReadableStream<Uint8Array>;
|
||||
const stderrReader = stderr.getReader();
|
||||
const stderrDecoder = new TextDecoder();
|
||||
let stderrOutput = '';
|
||||
(async () => {
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await stderrReader.read();
|
||||
if (done) break;
|
||||
const text = stderrDecoder.decode(value, { stream: true });
|
||||
stderrOutput += text;
|
||||
if (text.trim()) logger.info('Pi stderr', { text: text.trim() });
|
||||
}
|
||||
} catch {
|
||||
// Process ended
|
||||
}
|
||||
})();
|
||||
|
||||
// Handle process exit
|
||||
proc.exited.then((code) => {
|
||||
if (code === 1) {
|
||||
// Exit code 1 often indicates a startup issue, possibly snap node + piped stdin incompatibility
|
||||
logger.error('Pi process exited with code 1', {
|
||||
nodeVersion: process.version,
|
||||
nodeExePath: process.execPath,
|
||||
isSnapNode: process.execPath?.includes('/snap/'),
|
||||
hint: 'If node is from snap (/snap/bin/node), uninstall snap node and install via apt instead',
|
||||
});
|
||||
} else if (code !== 0) {
|
||||
logger.error('Pi process exited with error code', { code });
|
||||
} else {
|
||||
logger.info('Pi process exited normally');
|
||||
}
|
||||
});
|
||||
|
||||
return proc;
|
||||
}
|
||||
|
||||
function parseErrorMessage(raw: string): string {
|
||||
try {
|
||||
const parsed = JSON.parse(raw.replace(/^\d+\s*/, ''));
|
||||
const inner = parsed?.error;
|
||||
if (inner?.message) return inner.message;
|
||||
} catch {
|
||||
// not JSON
|
||||
}
|
||||
return raw;
|
||||
}
|
||||
|
||||
function extractMessageError(msg: Record<string, unknown>): string | null {
|
||||
if (msg.stopReason !== 'error') return null;
|
||||
const raw = msg.errorMessage as string | undefined;
|
||||
if (!raw) return null;
|
||||
return parseErrorMessage(raw);
|
||||
}
|
||||
|
||||
function parsePiEvent(event: Record<string, unknown>, currentStreamBuffer: string): PiEvent[] {
|
||||
const type = event.type as string;
|
||||
|
||||
// Handle response (success/failure for commands)
|
||||
if (type === 'response') {
|
||||
if (event.command === 'prompt' && !event.success) {
|
||||
const errorMsg = (event.error as string) ?? 'Prompt failed';
|
||||
return [{ type: 'error', message: errorMsg }];
|
||||
}
|
||||
return [];
|
||||
}
|
||||
|
||||
switch (type) {
|
||||
case 'agent_start':
|
||||
return [];
|
||||
|
||||
case 'message_update': {
|
||||
const ame = event.assistantMessageEvent as Record<string, unknown> | undefined;
|
||||
if (ame?.type === 'text_delta') {
|
||||
const delta = ame.delta as string;
|
||||
return [{ type: 'delta', text: delta }];
|
||||
}
|
||||
return [];
|
||||
}
|
||||
|
||||
case 'message_end': {
|
||||
const events: PiEvent[] = [];
|
||||
if (currentStreamBuffer) {
|
||||
events.push({ type: 'text', text: currentStreamBuffer });
|
||||
}
|
||||
const msg = event.message as Record<string, unknown> | undefined;
|
||||
if (msg) {
|
||||
const errorText = extractMessageError(msg);
|
||||
if (errorText) events.push({ type: 'error', message: errorText });
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
case 'tool_execution_start': {
|
||||
const toolCallId = (event.toolCallId as string) ?? '';
|
||||
const toolName = (event.toolName as string) ?? 'unknown';
|
||||
const args = (event.args as Record<string, unknown>) ?? {};
|
||||
|
||||
return [
|
||||
{
|
||||
type: 'tool:start',
|
||||
toolCallId,
|
||||
toolName,
|
||||
toolInput: args,
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
case 'tool_execution_end': {
|
||||
const toolCallId = (event.toolCallId as string) ?? '';
|
||||
const result = event.result;
|
||||
let resultObj: Record<string, unknown> | null = null;
|
||||
if (typeof result === 'object' && result !== null) {
|
||||
resultObj = result as Record<string, unknown>;
|
||||
} else if (typeof result === 'string') {
|
||||
try {
|
||||
resultObj = JSON.parse(result);
|
||||
} catch {
|
||||
/* not JSON */
|
||||
}
|
||||
}
|
||||
const isError = (event.isError as boolean) ?? (resultObj?.isError as boolean) ?? false;
|
||||
const output = result != null ? (typeof result === 'string' ? result : JSON.stringify(result)) : '';
|
||||
|
||||
return [
|
||||
{
|
||||
type: 'tool:result',
|
||||
toolCallId,
|
||||
output,
|
||||
isError,
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
case 'agent_end': {
|
||||
const cost: MessageCost = { inputTokens: 0, outputTokens: 0, totalUSD: 0 };
|
||||
const events: PiEvent[] = [];
|
||||
const messages = event.messages as Array<Record<string, unknown>> | undefined;
|
||||
if (messages) {
|
||||
for (const msg of messages) {
|
||||
const usage = msg.usage as Record<string, unknown> | undefined;
|
||||
if (usage) {
|
||||
cost.inputTokens += (usage.input as number) ?? 0;
|
||||
cost.outputTokens += (usage.output as number) ?? 0;
|
||||
const usageCost = usage.cost as Record<string, unknown> | undefined;
|
||||
if (usageCost) cost.totalUSD += (usageCost.total as number) ?? 0;
|
||||
}
|
||||
const errorText = extractMessageError(msg);
|
||||
if (errorText) events.push({ type: 'error', message: errorText });
|
||||
}
|
||||
}
|
||||
events.push({ type: 'result', cost });
|
||||
return events;
|
||||
}
|
||||
|
||||
case 'extension_ui_request':
|
||||
return [];
|
||||
|
||||
default:
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
function writeRpcCommand(proc: Subprocess, command: Record<string, unknown>): void {
|
||||
const stdin = proc.stdin;
|
||||
if (!stdin || typeof stdin === 'number') return;
|
||||
try {
|
||||
const writer = stdin as { write(data: string): void; flush(): void };
|
||||
writer.write(JSON.stringify(command) + '\n');
|
||||
writer.flush();
|
||||
} catch (err) {
|
||||
logger.error('writeRpcCommand error', { error: String(err) });
|
||||
}
|
||||
}
|
||||
|
||||
export function setThinkingLevel(process: Subprocess, level: string): void {
|
||||
writeRpcCommand(process, {
|
||||
type: 'set_thinking_level',
|
||||
level,
|
||||
});
|
||||
}
|
||||
|
||||
export function sendPrompt(process: Subprocess, prompt: string, requestId: string): void {
|
||||
writeRpcCommand(process, {
|
||||
type: 'prompt',
|
||||
id: requestId,
|
||||
message: prompt,
|
||||
});
|
||||
}
|
||||
|
||||
export function abort(process: Subprocess, requestId: string): void {
|
||||
writeRpcCommand(process, {
|
||||
type: 'abort',
|
||||
id: requestId,
|
||||
});
|
||||
}
|
||||
|
||||
export function cancelExtensionUi(process: Subprocess, id: unknown): void {
|
||||
writeRpcCommand(process, {
|
||||
type: 'extension_ui_response',
|
||||
id,
|
||||
cancelled: true,
|
||||
});
|
||||
}
|
||||
|
||||
export function killPi(process: Subprocess): void {
|
||||
try {
|
||||
process.kill();
|
||||
} catch {
|
||||
// Already dead
|
||||
}
|
||||
}
|
||||
|
||||
/** Build env vars needed by Officer tools when running on the host. */
|
||||
export async function buildHostToolEnv(email: string, role?: string): Promise<Record<string, string>> {
|
||||
const homeDir = getHomeDirForRole(email, role ?? null);
|
||||
|
||||
return {
|
||||
HOME: homeDir,
|
||||
OFFICER_USER_HOME: homeDir,
|
||||
OFFICER_USER_ROOT: join(DATA_PATH, email),
|
||||
OFFICER_EMAIL_DB: join(DATA_PATH, email, 'emails.db'),
|
||||
};
|
||||
}
|
||||
@@ -1,7 +1,6 @@
|
||||
import type { Context } from 'hono';
|
||||
import { createRouter } from '../../create-router';
|
||||
import * as storage from './storage';
|
||||
import { readLocalProviders } from '../server-settings/pi-mono';
|
||||
import { readSttConfig } from '../server-settings/stt';
|
||||
import { listPiModels } from './list-models';
|
||||
import { getHomeDirForRole } from '../../data-path';
|
||||
@@ -12,19 +11,12 @@ import { getUserSettings } from 'officerdb';
|
||||
export const piRestRouter = createRouter();
|
||||
|
||||
/**
|
||||
* GET /api/pi/models
|
||||
* GET /api/pi/models — Claude tiers only.
|
||||
*/
|
||||
piRestRouter.get('/pi/models', async (ctx: Context) => {
|
||||
try {
|
||||
const models = await listPiModels();
|
||||
|
||||
const providerNames: Record<string, string> = { 'claude-code': 'Claude Code' };
|
||||
const localProviders = await readLocalProviders();
|
||||
for (const lp of localProviders) {
|
||||
providerNames[`officer-local-${lp.id}`] = lp.name;
|
||||
}
|
||||
|
||||
logger.info('Models endpoint', { count: models.length, providers: [...new Set(models.map((m) => m.provider))] });
|
||||
return ctx.json({ models, providerNames, hostHome: process.env.HOME ?? '' });
|
||||
} catch (err) {
|
||||
logger.error('Failed to list models', { error: String(err) });
|
||||
@@ -57,14 +49,6 @@ piRestRouter.get('/pi/sessions/:sessionId', async (ctx: Context) => {
|
||||
}
|
||||
});
|
||||
|
||||
/**
|
||||
* PUT /api/pi/sessions/:sessionId/messages
|
||||
* No-op — sessions are persisted server-side via WebSocket events
|
||||
*/
|
||||
piRestRouter.put('/pi/sessions/:sessionId/messages', async (ctx: Context) => {
|
||||
return ctx.json({ success: true });
|
||||
});
|
||||
|
||||
/**
|
||||
* POST /api/pi/stt
|
||||
*/
|
||||
|
||||
@@ -1,5 +1,4 @@
|
||||
import type { UserSession } from "./types";
|
||||
import * as piBridge from "./pi-bridge";
|
||||
import { logger } from "./logger";
|
||||
|
||||
class SessionManager {
|
||||
|
||||
+11
-159
@@ -3,7 +3,6 @@ import { randomUUID } from 'crypto';
|
||||
import type { ClientMessage, ServerMessage, Message, PiEvent } from './types';
|
||||
import { sessionManager } from './session-manager';
|
||||
import * as storage from './storage';
|
||||
import * as piBridge from './pi-bridge';
|
||||
import { sendClaudeCodeStreaming } from '@@/channels/send-claude-code';
|
||||
import * as sidecar from '@@/sidecar-registry';
|
||||
import { join } from 'path';
|
||||
@@ -255,7 +254,7 @@ async function handleChat(
|
||||
resumeSummary?: string;
|
||||
},
|
||||
): Promise<void> {
|
||||
const { email, username, userId } = ws.data;
|
||||
const { userId } = ws.data;
|
||||
const sessionId = msg.sessionId || randomUUID();
|
||||
|
||||
// Prepend resume summary to the prompt if present
|
||||
@@ -263,121 +262,21 @@ async function handleChat(
|
||||
? `Here is a summary of a previous conversation to continue from:\n\n${msg.resumeSummary}\n\n---\n\nUser's new message: ${msg.prompt}`
|
||||
: msg.prompt;
|
||||
|
||||
// Use provided model, or fall back to user default, or use system default
|
||||
let model = msg.model;
|
||||
let modelSource = 'client-provided';
|
||||
let userDefault = null;
|
||||
if (!model) {
|
||||
userDefault = await getUserDefaultModel(userId);
|
||||
if (userDefault) {
|
||||
model = userDefault;
|
||||
modelSource = 'user-settings';
|
||||
} else {
|
||||
// Use provided model, or fall back to user default, or the system default.
|
||||
let model = msg.model || (await getUserDefaultModel(userId)) || DEFAULT_MODEL;
|
||||
|
||||
// Claude-only: coerce any legacy/non-Claude model preference to the Claude default so old saved
|
||||
// settings (Pi/opencode/openrouter model ids) don't break chat.
|
||||
if (!model.startsWith('claude-code')) {
|
||||
logger.info('Coercing non-Claude model to Claude default', { sessionId, requested: model });
|
||||
model = DEFAULT_MODEL;
|
||||
modelSource = 'system-default';
|
||||
}
|
||||
}
|
||||
|
||||
logger.info('Model selected for chat', {
|
||||
sessionId,
|
||||
model,
|
||||
modelSource,
|
||||
clientModel: msg.model || null,
|
||||
userDefault,
|
||||
});
|
||||
logger.info('Model selected for chat', { sessionId, model, clientModel: msg.model || null });
|
||||
|
||||
if (model.startsWith('claude-code')) {
|
||||
return handleClaudeCodeChat(ws, sessionId, model, msg, prompt);
|
||||
}
|
||||
|
||||
const homeDir = getHomeDirForRole(email, ws.data.role);
|
||||
const cwd = resolveCwd(email, ws.data.role, msg.cwd);
|
||||
const groupSlug = msg.groupSlug || null;
|
||||
const session = sessionManager.getOrCreate(sessionId, email, cwd, model, groupSlug, msg.context, msg.contextId);
|
||||
session.userId = userId;
|
||||
sessionManager.attachWs(sessionId, ws);
|
||||
wsToSessionMap.set(ws as any, sessionId);
|
||||
|
||||
sendToClient(ws, {
|
||||
type: 'session:init',
|
||||
sessionId,
|
||||
model,
|
||||
cwd,
|
||||
context: session.meta.context,
|
||||
contextId: session.meta.contextId,
|
||||
});
|
||||
if (!session.piProcess) {
|
||||
try {
|
||||
const onEvent = createEventHandler(sessionId, model, cwd, homeDir);
|
||||
|
||||
// If session has history, save to disk and pass --session for context replay
|
||||
let sessionFile: string | undefined;
|
||||
if (session.messages.length > 0) {
|
||||
await storage.saveSession(homeDir, sessionId, session.meta, session.messages);
|
||||
const hostPath = await storage.getSessionFilePath(homeDir, sessionId);
|
||||
if (hostPath) sessionFile = hostPath;
|
||||
}
|
||||
|
||||
// Subscribe to sidecar events for this session
|
||||
const unsub = sidecar.onPiEvent((evtSessionId, event) => {
|
||||
if (evtSessionId === sessionId) onEvent(event);
|
||||
});
|
||||
|
||||
await sidecar.spawnPi({
|
||||
sessionId,
|
||||
email,
|
||||
userId,
|
||||
username,
|
||||
role: ws.data.role,
|
||||
cwd,
|
||||
model,
|
||||
sessionFile,
|
||||
});
|
||||
|
||||
// Mark session as having a live process (use sessionId as sentinel)
|
||||
session.piProcess = sessionId as any;
|
||||
session._sidecarUnsub = unsub;
|
||||
|
||||
logger.info('Spawned Pi process via sidecar', {
|
||||
sessionId,
|
||||
model,
|
||||
cwd,
|
||||
hasSessionFile: !!sessionFile,
|
||||
});
|
||||
} catch (err) {
|
||||
logger.error('Failed to spawn Pi process', { sessionId, model, error: String(err) });
|
||||
sendToClient(ws, { type: 'error', message: 'Failed to start Pi process' });
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Add user message to session
|
||||
const userMsg: Message = {
|
||||
id: randomUUID(),
|
||||
timestamp: Date.now(),
|
||||
role: 'user',
|
||||
text: msg.prompt,
|
||||
};
|
||||
session.messages.push(userMsg);
|
||||
session.meta.messageCount += 1;
|
||||
session.meta.updatedAt = Date.now();
|
||||
|
||||
if (!session.meta.title) {
|
||||
session.meta.title = (msg.displayText ?? msg.prompt).slice(0, 100);
|
||||
}
|
||||
|
||||
// Set thinking level if provided
|
||||
console.log(`[pi] model: ${msg.model ?? 'default'}, thinking: ${msg.thinking ?? 'not set'}`);
|
||||
if (msg.thinking) {
|
||||
sidecar.setPiThinking(sessionId, msg.thinking);
|
||||
}
|
||||
|
||||
// Send prompt to Pi via sidecar
|
||||
const requestId = randomUUID();
|
||||
session.isGenerating = true;
|
||||
sidecar.sendPiPrompt(sessionId, prompt, requestId);
|
||||
}
|
||||
|
||||
async function handleClaudeCodeChat(
|
||||
ws: ServerWebSocket<WSData>,
|
||||
sessionId: string,
|
||||
@@ -497,50 +396,8 @@ async function handleResume(
|
||||
contextId: session.meta.contextId,
|
||||
});
|
||||
|
||||
// Spawn fresh Pi process if needed
|
||||
if (!session.piProcess) {
|
||||
try {
|
||||
const homeDir = getHomeDirForRole(email, ws.data.role);
|
||||
const onEvent = createEventHandler(sessionId, session.model, session.cwd, homeDir);
|
||||
|
||||
let sessionFile: string | undefined;
|
||||
if (session.messages.length > 0) {
|
||||
await storage.saveSession(homeDir, sessionId, session.meta, session.messages);
|
||||
const hostPath = await storage.getSessionFilePath(homeDir, sessionId);
|
||||
if (hostPath) sessionFile = hostPath;
|
||||
}
|
||||
|
||||
// Subscribe to sidecar events for this session
|
||||
const unsub = sidecar.onPiEvent((evtSessionId, event) => {
|
||||
if (evtSessionId === sessionId) onEvent(event);
|
||||
});
|
||||
|
||||
await sidecar.spawnPi({
|
||||
sessionId,
|
||||
email,
|
||||
userId: session.userId!,
|
||||
username: ws.data.username,
|
||||
role: ws.data.role,
|
||||
cwd: session.cwd,
|
||||
model: session.model,
|
||||
sessionFile,
|
||||
});
|
||||
|
||||
session.piProcess = sessionId as any;
|
||||
session._sidecarUnsub = unsub;
|
||||
|
||||
logger.info('Spawned fresh Pi process via sidecar for resumed session', {
|
||||
sessionId,
|
||||
model: session.model,
|
||||
hasSessionFile: !!sessionFile,
|
||||
});
|
||||
} catch (err) {
|
||||
logger.error('Failed to spawn Pi process for resume', { sessionId, error: String(err) });
|
||||
sendToClient(ws, { type: 'error', message: 'Failed to start Pi process' });
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// Claude resumes lazily: the next chat prompt re-attaches via `--resume <sessionKey>`, so there's
|
||||
// no long-lived process to spawn here — just replay the stored transcript to the client.
|
||||
sendToClient(ws, {
|
||||
type: 'sync:messages',
|
||||
sessionId,
|
||||
@@ -564,13 +421,8 @@ async function handleStop(ws: ServerWebSocket<WSData>): Promise<void> {
|
||||
|
||||
if (session?.piProcess) {
|
||||
try {
|
||||
if (session.model.startsWith('claude-code')) {
|
||||
sidecar.killClaude(sessionId, session.email);
|
||||
logger.info('Killed Claude Code process via sidecar', { sessionId });
|
||||
} else {
|
||||
sidecar.abortPi(sessionId, randomUUID());
|
||||
logger.info('Sent abort to Pi process via sidecar', { sessionId });
|
||||
}
|
||||
session.isGenerating = false;
|
||||
} catch (err) {
|
||||
logger.error('Failed to stop process', { sessionId, error: String(err) });
|
||||
|
||||
@@ -8,7 +8,6 @@ import { getTaskByDirName } from './task-files';
|
||||
import { getHomeDirForRole, getHomeDir } from '../../data-path';
|
||||
import { resolveBaseCwd } from '../pi/websocket';
|
||||
import { SANDBOX_HOME } from '../../sidecar/sandbox';
|
||||
import * as sidecar from '../../sidecar-registry';
|
||||
import { sendClaudeCodeStreaming } from '../../channels/send-claude-code';
|
||||
import type { PiEvent, MessageCost } from '../pi/types';
|
||||
import * as jobManager from './pipeline-job-manager';
|
||||
@@ -169,7 +168,6 @@ async function runAgenticStep({ userId, email, username, role, taskDirName, prom
|
||||
}, WAITING_INTERVAL_MS);
|
||||
|
||||
try {
|
||||
if (isClaudeCode) {
|
||||
const handle = await sendClaudeCodeStreaming({
|
||||
userId,
|
||||
email,
|
||||
@@ -182,19 +180,6 @@ async function runAgenticStep({ userId, email, username, role, taskDirName, prom
|
||||
onEvent,
|
||||
});
|
||||
cleanup = handle.kill;
|
||||
} else {
|
||||
const unsub = sidecar.onPiEvent((evtSessionId, event) => {
|
||||
if (evtSessionId === sessionId) onEvent(event);
|
||||
});
|
||||
cleanup = () => {
|
||||
unsub();
|
||||
sidecar.killPi(sessionId);
|
||||
};
|
||||
|
||||
await sidecar.spawnPi({ sessionId, email, userId, username, role, cwd, model });
|
||||
sidecar.sendPiPrompt(sessionId, prompt, randomUUID());
|
||||
}
|
||||
|
||||
} catch (err) {
|
||||
settle(() => { cleanup?.(); reject(err); });
|
||||
}
|
||||
@@ -489,7 +474,9 @@ export async function executePipeline({ userId, email, username, role, taskDirNa
|
||||
}
|
||||
|
||||
const baseCwd = resolveBaseCwd(email, role, cwd);
|
||||
const model = modelOverride || (await resolveModel(userId));
|
||||
let model = modelOverride || (await resolveModel(userId));
|
||||
// Claude-only: coerce any legacy non-Claude task-model preference to the Claude default.
|
||||
if (!model.startsWith('claude-code')) model = DEFAULT_MODEL;
|
||||
const totalCost: MessageCost = { inputTokens: 0, outputTokens: 0, totalUSD: 0 };
|
||||
|
||||
// Resolve concurrency from user input (default 1)
|
||||
|
||||
@@ -1,6 +1,4 @@
|
||||
import { mkdirSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { homedir } from 'node:os';
|
||||
import { DATA_PATH, ensureItemDirs } from './data-path';
|
||||
import { ensureToolLoader } from './ensure-tool-loader';
|
||||
// Queue is now owned by the sidecar process
|
||||
@@ -11,71 +9,7 @@ import { startWhatsAppBotIfConfigured } from './channels/whatsapp/bot';
|
||||
mkdirSync(DATA_PATH, { recursive: true });
|
||||
ensureItemDirs();
|
||||
|
||||
/** Check common locations for the Pi package directory. */
|
||||
async function findPiPackageDir(): Promise<string | null> {
|
||||
const candidates = [
|
||||
join(homedir(), '.npm-global', 'lib', 'node_modules', '@mariozechner', 'pi-coding-agent'),
|
||||
'/usr/local/lib/node_modules/@mariozechner/pi-coding-agent',
|
||||
];
|
||||
for (const dir of candidates) {
|
||||
if (await Bun.file(join(dir, 'package.json')).exists()) return dir;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
async function ensurePiInstalled(): Promise<boolean> {
|
||||
// If the `pi` binary is already on hand (e.g. a newer, manually-installed @earendil-works build),
|
||||
// treat Pi as installed. Otherwise the old @mariozechner package gets reinstalled over it every boot
|
||||
// and fails with EEXIST — noisy, and the "model discovery will not work" warning is a false alarm.
|
||||
const localPi = join(homedir(), '.local', 'bin', 'pi');
|
||||
if (Bun.which('pi') || (await Bun.file(localPi).exists())) {
|
||||
console.log('[bootstrap] Pi already available — skipping install');
|
||||
return true;
|
||||
}
|
||||
|
||||
const dir = await findPiPackageDir();
|
||||
if (!dir) return false;
|
||||
try {
|
||||
const pkg = await Bun.file(join(dir, 'package.json')).json();
|
||||
if (pkg.version) {
|
||||
console.log(`[bootstrap] Pi found: ${pkg.version}`);
|
||||
return true;
|
||||
}
|
||||
} catch {}
|
||||
return false;
|
||||
}
|
||||
|
||||
async function installPi(): Promise<boolean> {
|
||||
console.log('[bootstrap] Pi not found, installing...');
|
||||
try {
|
||||
const proc = Bun.spawn(['npm', 'install', '-g', '@mariozechner/pi-coding-agent'], {
|
||||
stdout: 'pipe',
|
||||
stderr: 'pipe',
|
||||
});
|
||||
const stderr = await new Response(proc.stderr).text();
|
||||
await proc.exited;
|
||||
if (proc.exitCode !== 0) {
|
||||
console.error('[bootstrap] Pi installation failed:', stderr.trim());
|
||||
return false;
|
||||
}
|
||||
console.log('[bootstrap] Pi installed successfully');
|
||||
return true;
|
||||
} catch (err) {
|
||||
console.error('[bootstrap] Pi installation error:', err);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
(async () => {
|
||||
const installed = await ensurePiInstalled();
|
||||
if (!installed) {
|
||||
const ok = await installPi();
|
||||
if (!ok) {
|
||||
console.error('[bootstrap] Could not install Pi — model discovery will not work');
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
ensureToolLoader();
|
||||
// Queue is initialized by the sidecar process
|
||||
|
||||
|
||||
@@ -1,17 +1,9 @@
|
||||
import { randomUUID } from 'crypto';
|
||||
import { join } from 'path';
|
||||
import type { PiEvent, MessageCost, Message } from '@@/api/pi/types';
|
||||
import { sessionManager } from '@@/api/pi/session-manager';
|
||||
import * as storage from '@@/api/pi/storage';
|
||||
import * as piBridge from '@@/api/pi/pi-bridge';
|
||||
import { getHomeDirForRole } from '@@/data-path';
|
||||
import type { MessageCost } from '@@/api/pi/types';
|
||||
import { getUserSettings } from 'officerdb';
|
||||
import { logger } from '@@/api/pi/logger';
|
||||
import { sendClaudeCode, clearClaudeCodeSession } from './send-claude-code';
|
||||
|
||||
const DEFAULT_MODEL = 'anthropic/claude-sonnet-4-20250514';
|
||||
const IDLE_TIMEOUT_MS = 60 * 60 * 1000;
|
||||
const SEND_TIMEOUT_MS = 5 * 60 * 1000;
|
||||
const DEFAULT_MODEL = 'claude-code';
|
||||
|
||||
type SendAndAwaitParams = {
|
||||
userId: number;
|
||||
@@ -34,10 +26,6 @@ type SendAndAwaitResult = {
|
||||
// Per-session mutex to serialize concurrent prompts
|
||||
const sessionLocks = new Map<string, Promise<void>>();
|
||||
|
||||
// Per-session callback — swapped each time a new prompt is sent
|
||||
type EventCallback = (event: PiEvent) => void;
|
||||
const sessionCallbacks = new Map<string, EventCallback>();
|
||||
|
||||
// Channel model overrides — survive session eviction/recreation
|
||||
const channelModelOverrides = new Map<string, string>();
|
||||
|
||||
@@ -56,292 +44,49 @@ async function getUserDefaultModel(userId: number): Promise<string | null> {
|
||||
}
|
||||
|
||||
export function getSessionModel(context: string, userId: number, contextId: string): string | null {
|
||||
const sessionId = buildSessionId(context, userId, contextId);
|
||||
const session = sessionManager.getSession(sessionId);
|
||||
return session?.model ?? channelModelOverrides.get(sessionId) ?? null;
|
||||
return channelModelOverrides.get(buildSessionId(context, userId, contextId)) ?? null;
|
||||
}
|
||||
|
||||
export function setSessionModel(context: string, userId: number, contextId: string, model: string): void {
|
||||
const sessionId = buildSessionId(context, userId, contextId);
|
||||
// Store override independently of session — survives idle eviction
|
||||
channelModelOverrides.set(sessionId, model);
|
||||
// Reset the Claude session so the next prompt starts fresh under the new model.
|
||||
clearClaudeCodeSession(sessionId);
|
||||
|
||||
const session = sessionManager.getSession(sessionId);
|
||||
if (session) {
|
||||
session.model = model;
|
||||
session.meta.model = model;
|
||||
// Kill existing PI process so it respawns with the new model
|
||||
if (session.piProcess) {
|
||||
piBridge.killPi(session.piProcess);
|
||||
session.piProcess = null;
|
||||
}
|
||||
logger.info('Channel model switched', { sessionId, model, killedProcess: true });
|
||||
} else {
|
||||
logger.info('Channel model override stored (no active session)', { sessionId, model });
|
||||
}
|
||||
logger.info('Channel model override stored', { sessionId, model });
|
||||
}
|
||||
|
||||
export async function sendAndAwait(params: SendAndAwaitParams): Promise<SendAndAwaitResult> {
|
||||
const { userId, context, contextId } = params;
|
||||
const sessionId = buildSessionId(context, userId, contextId);
|
||||
|
||||
// Serialize per session — if two messages arrive at once, second waits for first
|
||||
// Serialize per session — if two messages arrive at once, the second waits for the first.
|
||||
const existing = sessionLocks.get(sessionId) ?? Promise.resolve();
|
||||
let releaseLock: () => void;
|
||||
const lockPromise = new Promise<void>((resolve) => {
|
||||
releaseLock = resolve;
|
||||
});
|
||||
sessionLocks.set(
|
||||
sessionId,
|
||||
existing.then(() => lockPromise),
|
||||
);
|
||||
const chained = existing.then(() => lockPromise);
|
||||
sessionLocks.set(sessionId, chained);
|
||||
|
||||
await existing;
|
||||
|
||||
try {
|
||||
// Resolve model early to check for claude-code routing
|
||||
const override = channelModelOverrides.get(sessionId);
|
||||
const resolvedModel = params.model ?? override ?? (await getUserDefaultModel(params.userId)) ?? DEFAULT_MODEL;
|
||||
let model = params.model ?? override ?? (await getUserDefaultModel(userId)) ?? DEFAULT_MODEL;
|
||||
// Claude-only: coerce any legacy non-Claude model preference to the Claude default.
|
||||
if (!model.startsWith('claude-code')) model = DEFAULT_MODEL;
|
||||
|
||||
if (resolvedModel.startsWith('claude-code')) {
|
||||
return await sendClaudeCode({
|
||||
userId: params.userId,
|
||||
email: params.email,
|
||||
username: params.username,
|
||||
prompt: params.prompt,
|
||||
sessionKey: sessionId,
|
||||
model: resolvedModel,
|
||||
model,
|
||||
role: params.role,
|
||||
});
|
||||
}
|
||||
|
||||
return await doSend(sessionId, params);
|
||||
} finally {
|
||||
releaseLock!();
|
||||
if (sessionLocks.get(sessionId) === existing.then(() => lockPromise)) {
|
||||
sessionLocks.delete(sessionId);
|
||||
if (sessionLocks.get(sessionId) === chained) sessionLocks.delete(sessionId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Persistent event dispatcher — registered once at spawn time, delegates to current callback
|
||||
function createDispatcher(sessionId: string): (event: PiEvent) => void {
|
||||
return (event: PiEvent) => {
|
||||
const cb = sessionCallbacks.get(sessionId);
|
||||
if (cb) cb(event);
|
||||
};
|
||||
}
|
||||
|
||||
async function doSend(sessionId: string, params: SendAndAwaitParams): Promise<SendAndAwaitResult> {
|
||||
const { userId, email, username, prompt, context, contextId } = params;
|
||||
const homeDir = getHomeDirForRole(email, params.role ?? null);
|
||||
const cwd = homeDir;
|
||||
|
||||
// Resolve model: explicit param > !model override > user default > existing session > system default
|
||||
const existingSession = sessionManager.getSession(sessionId);
|
||||
const override = channelModelOverrides.get(sessionId);
|
||||
let model = params.model ?? override;
|
||||
if (!model) {
|
||||
const userDefault = await getUserDefaultModel(userId);
|
||||
model = userDefault ?? existingSession?.model ?? DEFAULT_MODEL;
|
||||
}
|
||||
|
||||
const session = sessionManager.getOrCreate(sessionId, email, cwd, model, null, context, contextId);
|
||||
session.model = model;
|
||||
session.meta.model = model;
|
||||
session.userId = userId;
|
||||
logger.info('Channel doSend', { sessionId, model, hasProcess: !!session.piProcess, userId, email });
|
||||
|
||||
return new Promise<SendAndAwaitResult>((resolve, reject) => {
|
||||
let resultText = '';
|
||||
const cost: MessageCost = { inputTokens: 0, outputTokens: 0, totalUSD: 0 };
|
||||
let settled = false;
|
||||
|
||||
const timeout = setTimeout(() => {
|
||||
if (!settled) {
|
||||
settled = true;
|
||||
session.isGenerating = false;
|
||||
sessionCallbacks.delete(sessionId);
|
||||
reject(new Error('sendAndAwait timed out after 5 minutes'));
|
||||
}
|
||||
}, SEND_TIMEOUT_MS);
|
||||
|
||||
const settle = () => {
|
||||
sessionCallbacks.delete(sessionId);
|
||||
clearTimeout(timeout);
|
||||
settled = true;
|
||||
};
|
||||
|
||||
// Register per-prompt callback — the persistent dispatcher will call this
|
||||
sessionCallbacks.set(sessionId, (event: PiEvent) => {
|
||||
if (settled) return;
|
||||
|
||||
switch (event.type) {
|
||||
case 'delta': {
|
||||
session.streamBuffer += event.text;
|
||||
break;
|
||||
}
|
||||
|
||||
case 'text': {
|
||||
const text = event.text || session.streamBuffer;
|
||||
if (text) {
|
||||
resultText += (resultText ? '\n\n' : '') + text;
|
||||
const assistantMsg: Message = {
|
||||
id: randomUUID(),
|
||||
timestamp: Date.now(),
|
||||
role: 'assistant',
|
||||
text,
|
||||
model,
|
||||
};
|
||||
session.messages.push(assistantMsg);
|
||||
session.meta.messageCount += 1;
|
||||
session.streamBuffer = '';
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
case 'tool:start': {
|
||||
if (session.streamBuffer) {
|
||||
resultText += (resultText ? '\n\n' : '') + session.streamBuffer;
|
||||
const assistantMsg: Message = {
|
||||
id: randomUUID(),
|
||||
timestamp: Date.now(),
|
||||
role: 'assistant',
|
||||
text: session.streamBuffer,
|
||||
model,
|
||||
};
|
||||
session.messages.push(assistantMsg);
|
||||
session.meta.messageCount += 1;
|
||||
session.streamBuffer = '';
|
||||
}
|
||||
const toolMsg: Message = {
|
||||
id: randomUUID(),
|
||||
timestamp: Date.now(),
|
||||
role: 'tool',
|
||||
toolCallId: event.toolCallId,
|
||||
toolName: event.toolName,
|
||||
toolInput: event.toolInput,
|
||||
};
|
||||
session.messages.push(toolMsg);
|
||||
session.meta.messageCount += 1;
|
||||
break;
|
||||
}
|
||||
|
||||
case 'tool:result': {
|
||||
for (let i = session.messages.length - 1; i >= 0; i--) {
|
||||
const m = session.messages[i]!;
|
||||
if (m.role === 'tool' && m.toolCallId === event.toolCallId) {
|
||||
m.output = event.output;
|
||||
m.isError = event.isError;
|
||||
break;
|
||||
}
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
case 'result': {
|
||||
if (session.streamBuffer) {
|
||||
resultText += (resultText ? '\n\n' : '') + session.streamBuffer;
|
||||
const assistantMsg: Message = {
|
||||
id: randomUUID(),
|
||||
timestamp: Date.now(),
|
||||
role: 'assistant',
|
||||
text: session.streamBuffer,
|
||||
model,
|
||||
cost: event.cost,
|
||||
};
|
||||
session.messages.push(assistantMsg);
|
||||
session.meta.messageCount += 1;
|
||||
session.streamBuffer = '';
|
||||
}
|
||||
|
||||
session.isGenerating = false;
|
||||
session.meta.cost.inputTokens += event.cost.inputTokens;
|
||||
session.meta.cost.outputTokens += event.cost.outputTokens;
|
||||
session.meta.cost.totalUSD += event.cost.totalUSD;
|
||||
session.meta.updatedAt = Date.now();
|
||||
|
||||
cost.inputTokens = event.cost.inputTokens;
|
||||
cost.outputTokens = event.cost.outputTokens;
|
||||
cost.totalUSD = event.cost.totalUSD;
|
||||
|
||||
storage.saveSession(homeDir, sessionId, session.meta, session.messages).catch((err) => {
|
||||
logger.error('Failed to save channel session', { sessionId, error: String(err) });
|
||||
});
|
||||
|
||||
sessionManager.setIdleTimeout(sessionId, IDLE_TIMEOUT_MS);
|
||||
|
||||
settle();
|
||||
resolve({ text: resultText || '(no response)', sessionId, model: model!, cost });
|
||||
break;
|
||||
}
|
||||
|
||||
case 'error': {
|
||||
session.isGenerating = false;
|
||||
settle();
|
||||
reject(new Error(event.message));
|
||||
break;
|
||||
}
|
||||
|
||||
case 'stopped': {
|
||||
session.isGenerating = false;
|
||||
settle();
|
||||
resolve({ text: resultText || '(stopped)', sessionId, model: model!, cost });
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Spawn Pi process if not running
|
||||
(async () => {
|
||||
try {
|
||||
if (!session.piProcess) {
|
||||
const role = params.role;
|
||||
let spawnOptions: { sessionFile?: string; username?: string; role?: string } | undefined;
|
||||
if (session.messages.length > 0) {
|
||||
await storage.saveSession(homeDir, sessionId, session.meta, session.messages);
|
||||
const hostPath = await storage.getSessionFilePath(homeDir, sessionId);
|
||||
if (hostPath) {
|
||||
spawnOptions = { sessionFile: hostPath, username, role };
|
||||
}
|
||||
}
|
||||
if (!spawnOptions) spawnOptions = { username, role };
|
||||
|
||||
const dispatcher = createDispatcher(sessionId);
|
||||
session.piProcess = await piBridge.spawnPi(cwd, model!, userId, email, dispatcher, spawnOptions);
|
||||
|
||||
const proc = session.piProcess;
|
||||
proc.exited.then(() => {
|
||||
if (session.piProcess === proc) {
|
||||
session.piProcess = null;
|
||||
logger.info('Channel Pi process exited', { sessionId });
|
||||
}
|
||||
});
|
||||
|
||||
logger.info('Spawned Pi for channel session', { sessionId, model, context, userId, email });
|
||||
}
|
||||
|
||||
// Add user message
|
||||
const userMsg: Message = {
|
||||
id: randomUUID(),
|
||||
timestamp: Date.now(),
|
||||
role: 'user',
|
||||
text: prompt,
|
||||
};
|
||||
session.messages.push(userMsg);
|
||||
session.meta.messageCount += 1;
|
||||
session.meta.updatedAt = Date.now();
|
||||
|
||||
if (!session.meta.title) {
|
||||
session.meta.title = prompt.slice(0, 100);
|
||||
}
|
||||
|
||||
session.isGenerating = true;
|
||||
piBridge.sendPrompt(session.piProcess, prompt, randomUUID());
|
||||
} catch (err) {
|
||||
settle();
|
||||
reject(err);
|
||||
}
|
||||
})();
|
||||
});
|
||||
}
|
||||
|
||||
@@ -8,7 +8,6 @@ import type {
|
||||
ClaudeSpawnParams,
|
||||
ClaudeSpawnStreamingParams,
|
||||
ClaudeCodeResult,
|
||||
PiSpawnParams,
|
||||
PtyCommand,
|
||||
PtyEvent,
|
||||
VncStartParams,
|
||||
@@ -351,42 +350,6 @@ export function onClaudeEvent(handler: (sessionKey: string, event: PiEvent) => v
|
||||
});
|
||||
}
|
||||
|
||||
// ── Pi ──
|
||||
|
||||
export async function spawnPi(params: PiSpawnParams): Promise<void> {
|
||||
const res = await sendCommand('pi', { type: 'pi:spawn', id: nextId(), params });
|
||||
if (res.type === 'pi:spawned') return;
|
||||
if (res.type === 'pi:error') throw new Error(res.error);
|
||||
throw new Error('Unexpected response');
|
||||
}
|
||||
|
||||
export function sendPiPrompt(sessionId: string, prompt: string, requestId: string): void {
|
||||
sendFire('pi', { type: 'pi:prompt', id: nextId(), sessionId, prompt, requestId });
|
||||
}
|
||||
|
||||
export function abortPi(sessionId: string, requestId: string): void {
|
||||
sendFire('pi', { type: 'pi:abort', id: nextId(), sessionId, requestId });
|
||||
}
|
||||
|
||||
export function killPi(sessionId: string): void {
|
||||
sendFire('pi', { type: 'pi:kill', id: nextId(), sessionId });
|
||||
}
|
||||
|
||||
export function setPiThinking(sessionId: string, level: string): void {
|
||||
sendFire('pi', { type: 'pi:set-thinking', id: nextId(), sessionId, level });
|
||||
}
|
||||
|
||||
export function onPiEvent(handler: (sessionId: string, event: PiEvent) => void): () => void {
|
||||
return on('pi:event', (msg) => {
|
||||
if (msg.type === 'pi:event') {
|
||||
handler(
|
||||
(msg as SidecarEvent & { type: 'pi:event' }).sessionId,
|
||||
(msg as SidecarEvent & { type: 'pi:event' }).event,
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// ── Terminal (PTY sidecar) ──
|
||||
|
||||
export function sendPtyCommand(cmd: PtyCommand): void {
|
||||
|
||||
@@ -1,88 +0,0 @@
|
||||
import type { SidecarCommand, SidecarEvent } from '../protocol';
|
||||
import * as piManager from './pi-manager';
|
||||
import { createSidecarConnector } from '../connect';
|
||||
|
||||
const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${process.env.PORT ?? '5000'}`;
|
||||
|
||||
// ── Command handlers ──
|
||||
|
||||
type ReplyFn = (msg: SidecarEvent) => void;
|
||||
|
||||
async function handleCommand(cmd: SidecarCommand, reply: ReplyFn) {
|
||||
switch (cmd.type) {
|
||||
case 'ping':
|
||||
reply({ type: 'pong', id: cmd.id });
|
||||
break;
|
||||
|
||||
case 'pi:spawn': {
|
||||
try {
|
||||
const onEvent = (event: import('../../api/pi/types').PiEvent) => {
|
||||
connection.send({ type: 'pi:event', sessionId: cmd.params.sessionId, event });
|
||||
};
|
||||
|
||||
await piManager.spawnPi({
|
||||
sessionId: cmd.params.sessionId,
|
||||
email: cmd.params.email,
|
||||
userId: cmd.params.userId,
|
||||
username: cmd.params.username,
|
||||
role: cmd.params.role,
|
||||
cwd: cmd.params.cwd,
|
||||
model: cmd.params.model,
|
||||
sessionFile: cmd.params.sessionFile,
|
||||
onEvent,
|
||||
});
|
||||
|
||||
reply({ type: 'pi:spawned', id: cmd.id, sessionId: cmd.params.sessionId });
|
||||
} catch (err) {
|
||||
reply({ type: 'pi:error', id: cmd.id, error: err instanceof Error ? err.message : String(err) });
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
case 'pi:prompt':
|
||||
piManager.sendPrompt(cmd.sessionId, cmd.prompt, cmd.requestId);
|
||||
break;
|
||||
|
||||
case 'pi:abort':
|
||||
piManager.abort(cmd.sessionId, cmd.requestId);
|
||||
break;
|
||||
|
||||
case 'pi:kill':
|
||||
piManager.killPiSession(cmd.sessionId);
|
||||
reply({ type: 'pi:killed', id: cmd.id });
|
||||
break;
|
||||
|
||||
case 'pi:set-thinking':
|
||||
piManager.setThinkingLevel(cmd.sessionId, cmd.level);
|
||||
break;
|
||||
|
||||
default:
|
||||
reply({
|
||||
type: 'error',
|
||||
id: (cmd as SidecarCommand).id,
|
||||
error: `Unknown command type: ${(cmd as Record<string, unknown>).type}`,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// ── Connect to API server ──
|
||||
|
||||
const connection = createSidecarConnector({
|
||||
apiUrl: `${API_URL}/api/sidecar/register`,
|
||||
name: 'pi',
|
||||
capabilities: ['pi'],
|
||||
onCommand(cmd, reply) {
|
||||
handleCommand(cmd as SidecarCommand, reply as ReplyFn);
|
||||
},
|
||||
});
|
||||
|
||||
// ── Graceful shutdown ──
|
||||
|
||||
function shutdown(signal: string) {
|
||||
console.log(`[pi] ${signal} received, shutting down...`);
|
||||
connection.destroy();
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
process.on('SIGTERM', () => shutdown('SIGTERM'));
|
||||
process.on('SIGINT', () => shutdown('SIGINT'));
|
||||
@@ -1,459 +0,0 @@
|
||||
import { join, dirname } from 'node:path';
|
||||
import { readdirSync, existsSync, mkdirSync } from 'node:fs';
|
||||
import type { Subprocess } from 'bun';
|
||||
import type { PiEvent, MessageCost } from '../../api/pi/types';
|
||||
import type { PiSpawnParams, PiSessionInfo } from '../protocol';
|
||||
import { sign } from '../../jwt';
|
||||
import {
|
||||
buildSandboxPrefix,
|
||||
buildRunuserSuffix,
|
||||
SANDBOX_DATA,
|
||||
SANDBOX_GLOBAL_EXTENSIONS,
|
||||
SANDBOX_GLOBAL_SKILLS,
|
||||
SANDBOX_GLOBAL_TOOLS,
|
||||
SANDBOX_HOME,
|
||||
} from '../sandbox';
|
||||
|
||||
const OFFICER_API_URL = process.env.OFFICER_API_URL ?? `http://127.0.0.1:${process.env.PORT ?? '9010'}`;
|
||||
|
||||
const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data');
|
||||
const PI_CONFIG_DIR = join(require('node:os').homedir(), '.pi', 'agent');
|
||||
const getHomeDir = (email: string) => join(DATA_PATH, email, 'home');
|
||||
const getHomeDirForRole = (email: string, role: string | null): string =>
|
||||
role === 'Super Admin' && process.env.HOME_DIR ? process.env.HOME_DIR : getHomeDir(email);
|
||||
const OFFICER_ITEMS_DIR = process.env.OFFICER_ITEMS_DIR ?? join(process.cwd(), 'officer-items');
|
||||
const itemsDir = (type: 'skills' | 'tools' | 'extensions') => join(OFFICER_ITEMS_DIR, type);
|
||||
|
||||
function isPidAlive(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
// Resolve pi as [node, cli.js] — the real .js path lives under ~/.local/lib/node_modules
|
||||
// which is ro-mounted in the sandbox. Node is at /usr/bin/node (under /usr ro-bind).
|
||||
const PI_CMD = (() => {
|
||||
const whichResult = Bun.spawnSync({ cmd: ['which', 'pi'], stdout: 'pipe', stderr: 'ignore' });
|
||||
const piBin = whichResult.stdout.toString().trim() || 'pi';
|
||||
const readlinkResult = Bun.spawnSync({ cmd: ['readlink', '-f', piBin], stdout: 'pipe', stderr: 'ignore' });
|
||||
const realPath = readlinkResult.stdout.toString().trim();
|
||||
const nodeResult = Bun.spawnSync({ cmd: ['which', 'node'], stdout: 'pipe', stderr: 'ignore' });
|
||||
const nodeBin = nodeResult.stdout.toString().trim() || 'node';
|
||||
if (realPath && realPath.endsWith('.js')) {
|
||||
return [nodeBin, realPath];
|
||||
}
|
||||
return [piBin];
|
||||
})();
|
||||
|
||||
// Pi's nested node_modules — needed for NODE_PATH so extensions can resolve Pi's dependencies
|
||||
// (e.g. @sinclair/typebox used by the tool-loader extension)
|
||||
const PI_NODE_MODULES = (() => {
|
||||
const whichResult = Bun.spawnSync({ cmd: ['which', 'pi'], stdout: 'pipe', stderr: 'ignore' });
|
||||
const piBin = whichResult.stdout.toString().trim() || 'pi';
|
||||
const readlinkResult = Bun.spawnSync({ cmd: ['readlink', '-f', piBin], stdout: 'pipe', stderr: 'ignore' });
|
||||
const realPath = readlinkResult.stdout.toString().trim();
|
||||
// cli.js is at .../pi-coding-agent/dist/cli.js — node_modules is at .../pi-coding-agent/node_modules
|
||||
if (realPath) {
|
||||
const pkgDir = join(dirname(realPath), '..');
|
||||
const nm = join(pkgDir, 'node_modules');
|
||||
if (existsSync(nm)) return nm;
|
||||
}
|
||||
return null;
|
||||
})();
|
||||
|
||||
// Active Pi processes
|
||||
type PiSession = {
|
||||
sessionId: string;
|
||||
email: string;
|
||||
userId: number;
|
||||
model: string;
|
||||
cwd: string;
|
||||
proc: Subprocess;
|
||||
onEvent: (event: PiEvent) => void;
|
||||
};
|
||||
|
||||
const sessions = new Map<string, PiSession>();
|
||||
|
||||
// ── Helpers ──
|
||||
|
||||
function collectSkillFlagsFromDir(scanDir: string, targetDir: string): string[] {
|
||||
const flags: string[] = [];
|
||||
if (!existsSync(scanDir)) return flags;
|
||||
|
||||
for (const entry of readdirSync(scanDir, { withFileTypes: true })) {
|
||||
if (!entry.isDirectory()) continue;
|
||||
if (existsSync(join(scanDir, entry.name, 'SKILL.md'))) {
|
||||
flags.push('--skill', `${targetDir}/${entry.name}`);
|
||||
}
|
||||
}
|
||||
|
||||
return flags;
|
||||
}
|
||||
|
||||
function collectSkillFlags(): string[] {
|
||||
return collectSkillFlagsFromDir(itemsDir('skills'), itemsDir('skills'));
|
||||
}
|
||||
|
||||
function collectExtensionFlagsFromDir(scanDir: string, targetDir: string): string[] {
|
||||
const flags: string[] = [];
|
||||
if (!existsSync(scanDir)) return flags;
|
||||
|
||||
for (const entry of readdirSync(scanDir, { withFileTypes: true })) {
|
||||
if (!entry.isDirectory()) continue;
|
||||
if (existsSync(join(scanDir, entry.name, 'index.ts'))) {
|
||||
flags.push('--extension', `${targetDir}/${entry.name}/index.ts`);
|
||||
}
|
||||
}
|
||||
|
||||
return flags;
|
||||
}
|
||||
|
||||
function collectExtensionFlags(): string[] {
|
||||
return collectExtensionFlagsFromDir(itemsDir('extensions'), itemsDir('extensions'));
|
||||
}
|
||||
|
||||
async function resolveApiKeyForModel(model: string): Promise<string | null> {
|
||||
const provider = model.split('/')[0];
|
||||
if (!provider) return null;
|
||||
try {
|
||||
const authFile = Bun.file(join(PI_CONFIG_DIR, 'auth.json'));
|
||||
if (!(await authFile.exists())) return null;
|
||||
const auth = (await authFile.json()) as Record<string, { key?: string }>;
|
||||
return auth[provider]?.key?.trim() || null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
// ── Event parsing (mirrors pi-bridge.ts) ──
|
||||
|
||||
function parseErrorMessage(raw: string): string {
|
||||
try {
|
||||
const parsed = JSON.parse(raw.replace(/^\d+\s*/, ''));
|
||||
const inner = parsed?.error;
|
||||
if (inner?.message) return inner.message;
|
||||
} catch {
|
||||
/* not JSON */
|
||||
}
|
||||
return raw;
|
||||
}
|
||||
|
||||
function extractMessageError(msg: Record<string, unknown>): string | null {
|
||||
if (msg.stopReason !== 'error') return null;
|
||||
const raw = msg.errorMessage as string | undefined;
|
||||
if (!raw) return null;
|
||||
return parseErrorMessage(raw);
|
||||
}
|
||||
|
||||
function parsePiEvent(event: Record<string, unknown>, currentStreamBuffer: string): PiEvent[] {
|
||||
const type = event.type as string;
|
||||
|
||||
if (type === 'response') {
|
||||
if (event.command === 'prompt' && !event.success) {
|
||||
return [{ type: 'error', message: (event.error as string) ?? 'Prompt failed' }];
|
||||
}
|
||||
return [];
|
||||
}
|
||||
|
||||
switch (type) {
|
||||
case 'agent_start':
|
||||
return [];
|
||||
|
||||
case 'message_update': {
|
||||
const ame = event.assistantMessageEvent as Record<string, unknown> | undefined;
|
||||
if (ame?.type === 'text_delta') {
|
||||
return [{ type: 'delta', text: ame.delta as string }];
|
||||
}
|
||||
return [];
|
||||
}
|
||||
|
||||
case 'message_end': {
|
||||
const events: PiEvent[] = [];
|
||||
if (currentStreamBuffer) {
|
||||
events.push({ type: 'text', text: currentStreamBuffer });
|
||||
}
|
||||
const msg = event.message as Record<string, unknown> | undefined;
|
||||
if (msg) {
|
||||
const errorText = extractMessageError(msg);
|
||||
if (errorText) events.push({ type: 'error', message: errorText });
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
case 'tool_execution_start':
|
||||
return [
|
||||
{
|
||||
type: 'tool:start',
|
||||
toolCallId: (event.toolCallId as string) ?? '',
|
||||
toolName: (event.toolName as string) ?? 'unknown',
|
||||
toolInput: (event.args as Record<string, unknown>) ?? {},
|
||||
},
|
||||
];
|
||||
|
||||
case 'tool_execution_end': {
|
||||
const toolCallId = (event.toolCallId as string) ?? '';
|
||||
const result = event.result;
|
||||
let resultObj: Record<string, unknown> | null = null;
|
||||
if (typeof result === 'object' && result !== null) {
|
||||
resultObj = result as Record<string, unknown>;
|
||||
} else if (typeof result === 'string') {
|
||||
try {
|
||||
resultObj = JSON.parse(result);
|
||||
} catch {
|
||||
/* not JSON */
|
||||
}
|
||||
}
|
||||
const isError = (event.isError as boolean) ?? (resultObj?.isError as boolean) ?? false;
|
||||
const output = result != null ? (typeof result === 'string' ? result : JSON.stringify(result)) : '';
|
||||
return [{ type: 'tool:result', toolCallId, output, isError }];
|
||||
}
|
||||
|
||||
case 'agent_end': {
|
||||
const cost: MessageCost = { inputTokens: 0, outputTokens: 0, totalUSD: 0 };
|
||||
const events: PiEvent[] = [];
|
||||
const messages = event.messages as Array<Record<string, unknown>> | undefined;
|
||||
if (messages) {
|
||||
for (const msg of messages) {
|
||||
const usage = msg.usage as Record<string, unknown> | undefined;
|
||||
if (usage) {
|
||||
cost.inputTokens += (usage.input as number) ?? 0;
|
||||
cost.outputTokens += (usage.output as number) ?? 0;
|
||||
const usageCost = usage.cost as Record<string, unknown> | undefined;
|
||||
if (usageCost) cost.totalUSD += (usageCost.total as number) ?? 0;
|
||||
}
|
||||
const errorText = extractMessageError(msg);
|
||||
if (errorText) events.push({ type: 'error', message: errorText });
|
||||
}
|
||||
}
|
||||
events.push({ type: 'result', cost });
|
||||
return events;
|
||||
}
|
||||
|
||||
default:
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
function writeRpcCommand(proc: Subprocess, command: Record<string, unknown>): void {
|
||||
const stdin = proc.stdin;
|
||||
if (!stdin || typeof stdin === 'number') return;
|
||||
try {
|
||||
const writer = stdin as { write(data: string): void; flush(): void };
|
||||
writer.write(JSON.stringify(command) + '\n');
|
||||
writer.flush();
|
||||
} catch (err) {
|
||||
console.error('[pi] writeRpcCommand error:', err);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Public API ──
|
||||
|
||||
export type PiSpawnOptions = {
|
||||
sessionId: string;
|
||||
email: string;
|
||||
userId: number;
|
||||
username: string;
|
||||
role: string;
|
||||
cwd: string;
|
||||
model: string;
|
||||
sessionFile?: string;
|
||||
onEvent: (event: PiEvent) => void;
|
||||
};
|
||||
|
||||
export async function spawnPi(options: PiSpawnOptions): Promise<void> {
|
||||
const { sessionId, email, userId, username, role, cwd, model, sessionFile, onEvent } = options;
|
||||
|
||||
const isSuperAdmin = role === 'Super Admin';
|
||||
const skillFlags = isSuperAdmin
|
||||
? collectSkillFlags()
|
||||
: collectSkillFlagsFromDir(itemsDir('skills'), SANDBOX_GLOBAL_SKILLS);
|
||||
const extensionFlags = isSuperAdmin
|
||||
? collectExtensionFlags()
|
||||
: collectExtensionFlagsFromDir(itemsDir('extensions'), SANDBOX_GLOBAL_EXTENSIONS);
|
||||
|
||||
const piArgs = [
|
||||
...PI_CMD,
|
||||
'--mode',
|
||||
'rpc',
|
||||
'--no-skills',
|
||||
'--no-prompt-templates',
|
||||
'--no-themes',
|
||||
...skillFlags,
|
||||
...extensionFlags,
|
||||
];
|
||||
if (model) piArgs.push('--model', model);
|
||||
if (sessionFile) piArgs.push('--session', sessionFile);
|
||||
|
||||
const apiKey = await resolveApiKeyForModel(model);
|
||||
if (apiKey) piArgs.push('--api-key', apiKey);
|
||||
|
||||
if (!existsSync(cwd)) {
|
||||
mkdirSync(cwd, { recursive: true });
|
||||
}
|
||||
|
||||
const homeDir = getHomeDirForRole(email, role);
|
||||
const toolsDirs = itemsDir('tools');
|
||||
|
||||
// Per-session JWT so tools (e.g. the gmail proxy) can call back to dev-platform
|
||||
// as the owning user. Mirrors the signin payload shape so userMiddleware accepts it.
|
||||
const officerAuthToken = await sign({ id: userId, email, username, role }, '24h');
|
||||
|
||||
let proc: Subprocess;
|
||||
|
||||
if (isSuperAdmin) {
|
||||
// Super Admin: run directly with host env, no sandbox
|
||||
const env: Record<string, string> = {
|
||||
...(process.env as Record<string, string>),
|
||||
HOME: process.env.HOME ?? homeDir,
|
||||
OFFICER_USER_HOME: homeDir,
|
||||
OFFICER_USER_ROOT: join(DATA_PATH, email),
|
||||
PI_CODING_AGENT_DIR: PI_CONFIG_DIR,
|
||||
PI_TOOLS_DIRS: toolsDirs,
|
||||
OFFICER_EMAIL_DB: join(DATA_PATH, email, 'emails.db'),
|
||||
OFFICER_API_URL,
|
||||
OFFICER_AUTH_TOKEN: officerAuthToken,
|
||||
TERM: 'xterm-256color',
|
||||
};
|
||||
if (PI_NODE_MODULES) env.NODE_PATH = PI_NODE_MODULES;
|
||||
|
||||
proc = Bun.spawn(piArgs, { cwd, stdin: 'pipe', stdout: 'pipe', stderr: 'pipe', env });
|
||||
} else {
|
||||
// Non-admin: run inside bwrap sandbox
|
||||
const sandboxToolsDirs = SANDBOX_GLOBAL_TOOLS;
|
||||
const prefix = buildSandboxPrefix(email);
|
||||
|
||||
// Pi-specific env vars
|
||||
prefix.push('--setenv', 'OFFICER_USER_HOME', SANDBOX_HOME);
|
||||
prefix.push('--setenv', 'OFFICER_USER_ROOT', SANDBOX_DATA);
|
||||
prefix.push('--setenv', 'PI_CODING_AGENT_DIR', `${SANDBOX_HOME}/.pi/agent`);
|
||||
prefix.push('--setenv', 'PI_TOOLS_DIRS', sandboxToolsDirs);
|
||||
prefix.push('--setenv', 'OFFICER_EMAIL_DB', `${SANDBOX_DATA}/emails.db`);
|
||||
prefix.push('--setenv', 'OFFICER_API_URL', OFFICER_API_URL);
|
||||
prefix.push('--setenv', 'OFFICER_AUTH_TOKEN', officerAuthToken);
|
||||
prefix.push('--setenv', 'TERM', 'xterm-256color');
|
||||
if (PI_NODE_MODULES) prefix.push('--setenv', 'NODE_PATH', PI_NODE_MODULES);
|
||||
|
||||
const sandboxArgs = [...prefix, ...buildRunuserSuffix()];
|
||||
proc = Bun.spawn([...sandboxArgs, ...piArgs], { stdin: 'pipe', stdout: 'pipe', stderr: 'pipe' });
|
||||
}
|
||||
|
||||
const session: PiSession = { sessionId, email, userId, model, cwd, proc, onEvent };
|
||||
sessions.set(sessionId, session);
|
||||
|
||||
console.log(`[pi] spawned Pi for session ${sessionId} (model=${model}, pid=${proc.pid})`);
|
||||
|
||||
// Read stdout JSON event stream
|
||||
const stdout = proc.stdout as ReadableStream<Uint8Array>;
|
||||
const reader = stdout.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = '';
|
||||
let streamBuffer = '';
|
||||
|
||||
(async () => {
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
const lines = buffer.split('\n');
|
||||
buffer = lines.pop() ?? '';
|
||||
for (const line of lines) {
|
||||
if (!line.trim()) continue;
|
||||
try {
|
||||
const event = JSON.parse(line) as Record<string, unknown>;
|
||||
const piEvents = parsePiEvent(event, streamBuffer);
|
||||
for (const piEvent of piEvents) {
|
||||
if (piEvent.type === 'delta') {
|
||||
streamBuffer += piEvent.text;
|
||||
} else if (piEvent.type === 'text' || piEvent.type === 'tool:start') {
|
||||
streamBuffer = '';
|
||||
}
|
||||
onEvent(piEvent);
|
||||
}
|
||||
} catch {
|
||||
/* skip */
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
/* process ended */
|
||||
}
|
||||
})();
|
||||
|
||||
// Stderr → log
|
||||
const stderr = proc.stderr as ReadableStream<Uint8Array>;
|
||||
const stderrReader = stderr.getReader();
|
||||
const stderrDecoder = new TextDecoder();
|
||||
(async () => {
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await stderrReader.read();
|
||||
if (done) break;
|
||||
const text = stderrDecoder.decode(value, { stream: true });
|
||||
if (text.trim()) console.log(`[pi:stderr] ${text.trim()}`);
|
||||
}
|
||||
} catch {
|
||||
/* process ended */
|
||||
}
|
||||
})();
|
||||
|
||||
// Handle exit
|
||||
proc.exited.then((code) => {
|
||||
sessions.delete(sessionId);
|
||||
if (code !== 0) {
|
||||
console.error(`[pi] Pi process ${sessionId} exited with code ${code}`);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
export function sendPrompt(sessionId: string, prompt: string, requestId: string): boolean {
|
||||
const session = sessions.get(sessionId);
|
||||
if (!session) return false;
|
||||
writeRpcCommand(session.proc, { type: 'prompt', id: requestId, message: prompt });
|
||||
return true;
|
||||
}
|
||||
|
||||
export function abort(sessionId: string, requestId: string): boolean {
|
||||
const session = sessions.get(sessionId);
|
||||
if (!session) return false;
|
||||
writeRpcCommand(session.proc, { type: 'abort', id: requestId });
|
||||
return true;
|
||||
}
|
||||
|
||||
export function setThinkingLevel(sessionId: string, level: string): boolean {
|
||||
const session = sessions.get(sessionId);
|
||||
if (!session) return false;
|
||||
writeRpcCommand(session.proc, { type: 'set_thinking_level', level });
|
||||
return true;
|
||||
}
|
||||
|
||||
export function killPiSession(sessionId: string): boolean {
|
||||
const session = sessions.get(sessionId);
|
||||
if (!session) return false;
|
||||
try {
|
||||
session.proc.kill();
|
||||
} catch {
|
||||
/* already dead */
|
||||
}
|
||||
sessions.delete(sessionId);
|
||||
return true;
|
||||
}
|
||||
|
||||
export function getSession(sessionId: string): PiSession | undefined {
|
||||
return sessions.get(sessionId);
|
||||
}
|
||||
|
||||
export function getAllSessions(): PiSessionInfo[] {
|
||||
return Array.from(sessions.values()).map((s) => ({
|
||||
sessionId: s.sessionId,
|
||||
email: s.email,
|
||||
userId: s.userId,
|
||||
model: s.model,
|
||||
cwd: s.cwd,
|
||||
pid: s.proc.pid,
|
||||
alive: isPidAlive(s.proc.pid),
|
||||
}));
|
||||
}
|
||||
@@ -16,12 +16,6 @@ export type SidecarCommand =
|
||||
| { type: 'claude:spawn-streaming'; id: string; params: ClaudeSpawnStreamingParams }
|
||||
| { type: 'claude:kill'; id: string; sessionKey: string }
|
||||
| { type: 'claude:clear-session'; id: string; sessionKey: string }
|
||||
// Pi
|
||||
| { type: 'pi:spawn'; id: string; params: PiSpawnParams }
|
||||
| { type: 'pi:prompt'; id: string; sessionId: string; prompt: string; requestId: string }
|
||||
| { type: 'pi:abort'; id: string; sessionId: string; requestId: string }
|
||||
| { type: 'pi:kill'; id: string; sessionId: string }
|
||||
| { type: 'pi:set-thinking'; id: string; sessionId: string; level: string }
|
||||
// VNC
|
||||
| { type: 'vnc:start'; id: string; params: VncStartParams }
|
||||
| { type: 'vnc:stop'; id: string; email: string }
|
||||
@@ -40,11 +34,6 @@ export type SidecarEvent =
|
||||
| { type: 'claude:error'; id: string; error: string }
|
||||
| { type: 'claude:killed'; id: string }
|
||||
| { type: 'claude:session-cleared'; id: string }
|
||||
// Pi
|
||||
| { type: 'pi:spawned'; id: string; sessionId: string }
|
||||
| { type: 'pi:event'; sessionId: string; event: PiEvent }
|
||||
| { type: 'pi:error'; id: string; error: string }
|
||||
| { type: 'pi:killed'; id: string }
|
||||
// VNC
|
||||
| { type: 'vnc:started'; id: string; port: number; display: number }
|
||||
| { type: 'vnc:stopped'; id: string }
|
||||
@@ -62,16 +51,6 @@ export type ClaudeState = {
|
||||
claudeSessions: Record<string, string>; // sessionKey → Claude Code session_id
|
||||
};
|
||||
|
||||
export type PiSessionInfo = {
|
||||
sessionId: string;
|
||||
email: string;
|
||||
userId: number;
|
||||
model: string;
|
||||
cwd: string;
|
||||
pid: number;
|
||||
alive: boolean;
|
||||
};
|
||||
|
||||
// ── Param types ──
|
||||
|
||||
export type ClaudeSpawnParams = {
|
||||
@@ -102,17 +81,6 @@ export type ClaudeCodeResult = {
|
||||
cost: MessageCost;
|
||||
};
|
||||
|
||||
export type PiSpawnParams = {
|
||||
sessionId: string;
|
||||
email: string;
|
||||
userId: number;
|
||||
username: string;
|
||||
role: string;
|
||||
cwd: string;
|
||||
model: string;
|
||||
sessionFile?: string;
|
||||
};
|
||||
|
||||
// ── VNC types ──
|
||||
|
||||
export type VncStartParams = {
|
||||
|
||||
Reference in New Issue
Block a user