process sidecar: independent process manager for long-running work
Introduces a separate Bun process (port 5100) that owns all spawned processes and long-running work, so the API server can restart freely without disrupting active sessions. The sidecar owns: - Anthropic proxy (port 5051) with persisted secret across restarts - Claude Code process spawning and session tracking (--resume support) - Pi agent spawning and RPC lifecycle (prompt/abort/thinking) - Job queue engine (lane processing, retries, notifications) The API server becomes a thin client that forwards commands over a single WebSocket connection with auto-reconnect. send-claude-code.ts goes from 550 lines of spawn logic to 73 lines of sidecar delegation. State persisted to data/sidecar/state.json every 30s and on shutdown. Lockfile prevents duplicate instances. See SIDECAR.md for full docs and manual testing procedures. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.6
@@ -0,0 +1,347 @@
|
||||
# Process Sidecar
|
||||
|
||||
Independent Bun process that owns all long-running work so the API server can restart without disrupting active sessions.
|
||||
|
||||
## Problem
|
||||
|
||||
The API server (port 5000) previously owned all spawned processes: Pi agents, Claude Code sessions, the Anthropic proxy, and the job queue. Restarting the API server would:
|
||||
|
||||
- Kill active Pi and Claude Code conversations mid-response
|
||||
- Regenerate the Anthropic proxy secret, breaking any Claude Code sessions using it
|
||||
- Lose in-flight job progress (queue engine ran in-process)
|
||||
|
||||
## Solution
|
||||
|
||||
A separate Bun process ("process sidecar") on port 5100 that owns all spawned processes. The API server communicates with it over a single WebSocket connection. The sidecar is managed by pm2 and starts before the API server.
|
||||
|
||||
## Architecture
|
||||
|
||||
```
|
||||
┌─────────────────────────┐ ┌──────────────────────────────┐
|
||||
│ API Server (:5000) │ WS │ Process Sidecar (:5100) │
|
||||
│ │◄───────►│ │
|
||||
│ - Browser WS clients │ │ - Anthropic Proxy (:5051) │
|
||||
│ - REST API routes │ │ - Claude Code processes │
|
||||
│ - Channel bots │ │ - Pi agent processes │
|
||||
│ - sidecar-client.ts │ │ - Job queue engine │
|
||||
│ (auto-reconnect) │ │ - State persistence │
|
||||
└─────────────────────────┘ └──────────────────────────────┘
|
||||
```
|
||||
|
||||
### What the sidecar owns
|
||||
|
||||
| Concern | Previous location | Sidecar module |
|
||||
|---------|-------------------|----------------|
|
||||
| Anthropic proxy (port 5051) | `api/anthropic-proxy.ts` | `sidecar/proxy.ts` |
|
||||
| Claude Code spawn + session tracking | `channels/send-claude-code.ts` | `sidecar/claude-manager.ts` |
|
||||
| Pi agent spawn + RPC commands | `api/pi/pi-bridge.ts` | `sidecar/pi-manager.ts` |
|
||||
| Job queue engine + handlers | `queue/engine.ts` | `sidecar/queue-runner.ts` |
|
||||
|
||||
### What stays in the API server
|
||||
|
||||
- Browser WebSocket connections (ephemeral by nature)
|
||||
- REST API routes (now thin proxies to sidecar)
|
||||
- Channel bots (Discord/Telegram/WhatsApp — already reconnect gracefully)
|
||||
- Browser relay (CDP state is ephemeral)
|
||||
- PTY sidecar (already its own process, unchanged)
|
||||
|
||||
## Communication Protocol
|
||||
|
||||
Single WebSocket between API server and sidecar. JSON messages with `{ type, id?, ... }` envelopes.
|
||||
|
||||
**Request/response**: Commands include an `id` field. The sidecar responds with a message carrying the same `id`. The client correlates responses via this ID with configurable timeouts.
|
||||
|
||||
**Streaming events**: Pi and Claude Code output events are broadcast to all connected clients without a correlation ID. They carry a `sessionId` or `sessionKey` so the API server can route them to the correct browser WS.
|
||||
|
||||
### Command categories
|
||||
|
||||
```
|
||||
ping / pong — health check
|
||||
state:sync — full state snapshot on connect
|
||||
|
||||
proxy:secret — get persisted proxy secret
|
||||
|
||||
claude:spawn / claude:result — blocking Claude Code exec
|
||||
claude:spawn-streaming / claude:event — streaming Claude Code exec
|
||||
claude:kill / claude:clear-session — session management
|
||||
|
||||
pi:spawn / pi:prompt / pi:abort — Pi agent lifecycle
|
||||
pi:kill / pi:set-thinking — Pi session control
|
||||
|
||||
queue:enqueue / queue:cancel — job management
|
||||
queue:list / queue:get — job queries
|
||||
```
|
||||
|
||||
See `protocol.ts` for the full type definitions.
|
||||
|
||||
## State Persistence
|
||||
|
||||
File: `data/sidecar/state.json`
|
||||
|
||||
Written every 30 seconds (debounced) and on graceful shutdown (SIGTERM/SIGINT). Contains:
|
||||
|
||||
- **proxySecret** — generated once on first boot, reused forever. This is the key fix: the Anthropic proxy secret no longer changes on restart.
|
||||
- **claudeSessions** — map of `sessionKey → Claude Code session_id` for `--resume` support across restarts.
|
||||
- **piSessions** — session metadata with PIDs for liveness checking on restart.
|
||||
|
||||
### Lockfile
|
||||
|
||||
`data/sidecar/sidecar.lock` — contains the PID of the running sidecar. On startup, if the lock exists and the PID is alive, the sidecar exits. Stale locks (dead PID) are cleaned up automatically.
|
||||
|
||||
## File Structure
|
||||
|
||||
```
|
||||
src/servers/sidecar/
|
||||
index.ts — entry point: Bun.serve on :5100, WS dispatch, shutdown
|
||||
protocol.ts — shared message types (imported by both sides)
|
||||
proxy.ts — Anthropic proxy server (moved from anthropic-proxy.ts)
|
||||
claude-manager.ts — Claude Code blocking + streaming spawn, session map
|
||||
pi-manager.ts — Pi agent spawn, RPC (prompt/abort/thinking), event parsing
|
||||
queue-runner.ts — Job queue engine (lanes, retries, notifications)
|
||||
state.ts — File-backed state persistence + lockfile
|
||||
|
||||
src/servers/sidecar-client.ts — API server's WebSocket client (singleton)
|
||||
```
|
||||
|
||||
## API Server Integration
|
||||
|
||||
The API server connects to the sidecar on startup via `initSidecarClient()` in `server.tsx`. The client auto-reconnects with exponential backoff (200ms → 15s).
|
||||
|
||||
On connect, it sends `state:sync` to get the current proxy secret and live session info.
|
||||
|
||||
### Modified API server files
|
||||
|
||||
| File | Change |
|
||||
|------|--------|
|
||||
| `server.tsx` | `startAnthropicProxy()` → `initSidecarClient()` |
|
||||
| `bootstrap.ts` | Removed `initQueue()` (sidecar owns it) |
|
||||
| `channels/send-claude-code.ts` | 550 lines → 73 lines thin client |
|
||||
| `api/pi/websocket.ts` | Spawn/prompt/abort go through sidecar |
|
||||
| `api/pi/session-manager.ts` | Cleanup sidecar subscriptions on delete |
|
||||
| `api/queue/queue.ts` | Routes use sidecar client |
|
||||
| `channels/discord/handler.ts` | `enqueue` → `enqueueJob` via sidecar |
|
||||
| `channels/telegram/handler.ts` | Same |
|
||||
| `channels/whatsapp/handler.ts` | Same |
|
||||
|
||||
## PM2 Configuration
|
||||
|
||||
```js
|
||||
// ecosystem.config.cjs
|
||||
module.exports = {
|
||||
apps: [
|
||||
{
|
||||
name: 'officer-sidecar', // starts first
|
||||
script: 'bun',
|
||||
args: 'run src/servers/sidecar/index.ts',
|
||||
watch: false,
|
||||
},
|
||||
{
|
||||
name: 'officer',
|
||||
script: 'bun',
|
||||
args: 'start',
|
||||
watch: false,
|
||||
},
|
||||
],
|
||||
};
|
||||
```
|
||||
|
||||
The sidecar is listed first so pm2 starts it before the API server. The API server's sidecar client handles the case where the sidecar isn't ready yet (auto-reconnect with backoff).
|
||||
|
||||
## Environment Variables
|
||||
|
||||
| Variable | Default | Description |
|
||||
|----------|---------|-------------|
|
||||
| `SIDECAR_PORT` | `5100` | Sidecar HTTP/WS port |
|
||||
| `ANTHROPIC_PROXY_PORT` | `5051` | Anthropic proxy port (owned by sidecar) |
|
||||
| `DATA_PATH` | `./data` | Shared data directory |
|
||||
|
||||
## Manual Testing
|
||||
|
||||
### 1. Start the sidecar standalone
|
||||
|
||||
```bash
|
||||
# From monorepo root
|
||||
bun run src/servers/sidecar/index.ts
|
||||
```
|
||||
|
||||
Expected output:
|
||||
```
|
||||
[sidecar:proxy] listening on 127.0.0.1:5051
|
||||
[sidecar:queue] initialized
|
||||
[sidecar] listening on 127.0.0.1:5100
|
||||
```
|
||||
|
||||
### 2. Health check
|
||||
|
||||
```bash
|
||||
# Basic liveness
|
||||
curl http://127.0.0.1:5100/
|
||||
# → "process-sidecar"
|
||||
|
||||
# Detailed health
|
||||
curl http://127.0.0.1:5100/health
|
||||
# → {"status":"ok","uptime":12345,"piSessions":0,"claudeSessions":0}
|
||||
```
|
||||
|
||||
### 3. Verify Anthropic proxy is running
|
||||
|
||||
```bash
|
||||
# Should reject without valid secret
|
||||
curl -s http://127.0.0.1:5051/v1/messages
|
||||
# → {"error":"Unauthorized"}
|
||||
|
||||
# Check state file was created
|
||||
cat data/sidecar/state.json
|
||||
# → should show proxySecret, empty claudeSessions, empty piSessions
|
||||
```
|
||||
|
||||
### 4. Verify proxy secret persistence
|
||||
|
||||
```bash
|
||||
# Note the proxySecret from state.json
|
||||
cat data/sidecar/state.json | jq .proxySecret
|
||||
|
||||
# Stop and restart the sidecar
|
||||
# Kill the sidecar (Ctrl+C or kill)
|
||||
bun run src/servers/sidecar/index.ts
|
||||
|
||||
# Check the secret is the same
|
||||
cat data/sidecar/state.json | jq .proxySecret
|
||||
# → should be identical to before
|
||||
```
|
||||
|
||||
### 5. WebSocket communication test
|
||||
|
||||
```bash
|
||||
# In one terminal, start the sidecar
|
||||
bun run src/servers/sidecar/index.ts
|
||||
|
||||
# In another terminal, connect with websocat (or wscat)
|
||||
# Install: cargo install websocat OR npm install -g wscat
|
||||
websocat ws://127.0.0.1:5100
|
||||
|
||||
# Send a ping
|
||||
{"type":"ping","id":"test1"}
|
||||
# → should receive: {"type":"pong","id":"test1"}
|
||||
|
||||
# Send state sync
|
||||
{"type":"state:sync","id":"test2"}
|
||||
# → should receive: {"type":"state:sync","id":"test2","state":{...}}
|
||||
|
||||
# Get proxy secret
|
||||
{"type":"proxy:secret","id":"test3"}
|
||||
# → should receive: {"type":"proxy:secret","id":"test3","secret":"sk-ant-api03-..."}
|
||||
```
|
||||
|
||||
### 6. Test lockfile protection
|
||||
|
||||
```bash
|
||||
# Start sidecar in one terminal
|
||||
bun run src/servers/sidecar/index.ts
|
||||
|
||||
# Try starting another in a second terminal
|
||||
bun run src/servers/sidecar/index.ts
|
||||
# → should exit with: "[sidecar] another instance is already running"
|
||||
|
||||
# Check lockfile
|
||||
cat data/sidecar/sidecar.lock
|
||||
# → PID of the running sidecar
|
||||
```
|
||||
|
||||
### 7. Full integration test (sidecar + API server)
|
||||
|
||||
```bash
|
||||
# Start sidecar first
|
||||
bun run src/servers/sidecar/index.ts &
|
||||
|
||||
# Start API server
|
||||
bun start
|
||||
# → should see "[sidecar-client] connected" in logs
|
||||
|
||||
# Test via the dashboard:
|
||||
# 1. Open a chat panel, send a message with claude-code model
|
||||
# → should see streaming response (routed through sidecar)
|
||||
# 2. Open a chat with a Pi model
|
||||
# → should see streaming response (routed through sidecar)
|
||||
# 3. Restart the API server (kill + bun start)
|
||||
# → active Claude Code processes should NOT die
|
||||
# → sidecar should show "client disconnected" then "client connected"
|
||||
# → proxy secret should remain the same
|
||||
```
|
||||
|
||||
### 8. Test API server restart resilience
|
||||
|
||||
This is the key scenario that motivated the sidecar:
|
||||
|
||||
```bash
|
||||
# 1. Start sidecar + API server
|
||||
bun run src/servers/sidecar/index.ts &
|
||||
bun start &
|
||||
|
||||
# 2. Start a Claude Code streaming session in the dashboard
|
||||
|
||||
# 3. While Claude Code is running, kill the API server
|
||||
kill $(pgrep -f "bun start")
|
||||
|
||||
# 4. Restart the API server
|
||||
bun start
|
||||
|
||||
# 5. Check:
|
||||
# - The Claude Code process should still be running (check with ps)
|
||||
# - The proxy secret should be the same (check data/sidecar/state.json)
|
||||
# - The sidecar should show the reconnection in logs
|
||||
```
|
||||
|
||||
### 9. Test with pm2
|
||||
|
||||
```bash
|
||||
pm2 start ecosystem.config.cjs
|
||||
|
||||
# Check both are running
|
||||
pm2 status
|
||||
# → officer-sidecar: online
|
||||
# → officer: online
|
||||
|
||||
# Restart API server only
|
||||
pm2 restart officer
|
||||
|
||||
# Check sidecar is still running
|
||||
pm2 status
|
||||
curl http://127.0.0.1:5100/health
|
||||
|
||||
# Stop everything
|
||||
pm2 stop all
|
||||
```
|
||||
|
||||
### 10. Queue test
|
||||
|
||||
```bash
|
||||
# With sidecar running, queue a job via the API
|
||||
curl -X POST http://localhost:5000/api/queue/jobs \
|
||||
-H "Content-Type: application/json" \
|
||||
-H "Authorization: Bearer <your-token>" \
|
||||
-d '{"lane":"test","type":"gmail-sync"}'
|
||||
|
||||
# List jobs
|
||||
curl http://localhost:5000/api/queue/jobs \
|
||||
-H "Authorization: Bearer <your-token>"
|
||||
```
|
||||
|
||||
## Graceful Shutdown
|
||||
|
||||
On SIGTERM or SIGINT, the sidecar:
|
||||
1. Flushes pending state to `data/sidecar/state.json`
|
||||
2. Releases the lockfile
|
||||
3. Does **NOT** kill spawned processes — they are independent OS processes
|
||||
|
||||
On restart, the sidecar reads the persisted state and checks which PIDs are still alive.
|
||||
|
||||
## Risks & Mitigations
|
||||
|
||||
| Risk | Mitigation |
|
||||
|------|------------|
|
||||
| Sidecar dies with live sessions | pm2 auto-restart + state.json + processes are independent OS processes |
|
||||
| API↔sidecar WebSocket drops | Auto-reconnect with exponential backoff (200ms → 15s) |
|
||||
| Two sidecar instances running | Lockfile with PID liveness check on startup |
|
||||
| Sidecar needs DB access | Imports DB modules directly (same user, same filesystem) |
|
||||
| Queue handlers need server context | Handlers are self-contained modules imported by the sidecar |
|
||||
@@ -0,0 +1,355 @@
|
||||
import { join } from 'node:path';
|
||||
import type { Subprocess } from 'bun';
|
||||
import type { PiEvent, MessageCost } from '../api/pi/types';
|
||||
import type { ClaudeSpawnParams, ClaudeSpawnStreamingParams, ClaudeCodeResult } from './protocol';
|
||||
import { getState, setClaudeSession, clearClaudeSession, getClaudeSession } from './state';
|
||||
import { getProxySecret } from './proxy';
|
||||
|
||||
const SEND_TIMEOUT_MS = 5 * 60 * 1000;
|
||||
const PROXY_PORT = process.env.ANTHROPIC_PROXY_PORT ?? '5051';
|
||||
const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data');
|
||||
|
||||
const getHomeDir = (email: string) => join(DATA_PATH, email, 'home');
|
||||
|
||||
const toShellUsername = (username: string, email: string): string => {
|
||||
const raw = username || email.split('@')[0]!;
|
||||
return raw.replace(/@.*$/, '').replace(/[^a-zA-Z0-9._-]/g, '_').toLowerCase().slice(0, 32) || 'officer';
|
||||
};
|
||||
|
||||
async function hasOwnCredentials(homeDir: string): Promise<boolean> {
|
||||
try {
|
||||
return await Bun.file(join(homeDir, '.claude', '.credentials.json')).exists();
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
// Resolve absolute path to claude binary
|
||||
const CLAUDE_BIN = (() => {
|
||||
const result = Bun.spawnSync({ cmd: ['which', 'claude'], stdout: 'pipe', stderr: 'ignore' });
|
||||
return result.stdout.toString().trim() || 'claude';
|
||||
})();
|
||||
|
||||
// Active streaming processes
|
||||
const activeProcs = new Map<string, Subprocess>();
|
||||
|
||||
function buildAuthEnv(shellUsername: string, homeDir: string, isServiceUser: boolean, userHasCredentials: boolean): Record<string, string> {
|
||||
if (isServiceUser) return { HOME: process.env.HOME ?? '' };
|
||||
if (userHasCredentials) return { HOME: homeDir };
|
||||
return { ANTHROPIC_BASE_URL: `http://127.0.0.1:${PROXY_PORT}`, ANTHROPIC_API_KEY: getProxySecret() };
|
||||
}
|
||||
|
||||
// ── Blocking send ──
|
||||
|
||||
type ClaudeCodeOutput = {
|
||||
result: string;
|
||||
session_id: string;
|
||||
cost_usd: number;
|
||||
input_tokens: number;
|
||||
output_tokens: number;
|
||||
is_error: boolean;
|
||||
};
|
||||
|
||||
export async function spawnClaude(params: ClaudeSpawnParams): Promise<ClaudeCodeResult> {
|
||||
const { userId, email, username, prompt, sessionKey } = params;
|
||||
const homeDir = getHomeDir(email);
|
||||
const shellUsername = toShellUsername(username, email);
|
||||
|
||||
const claudeArgs = [CLAUDE_BIN, '-p', prompt, '--dangerously-skip-permissions', '--output-format', 'json'];
|
||||
|
||||
const subModel = params.model?.split('/')[1];
|
||||
if (subModel) claudeArgs.push('--model', subModel);
|
||||
|
||||
const existingSession = getClaudeSession(sessionKey);
|
||||
if (existingSession) {
|
||||
claudeArgs.push('--resume', existingSession);
|
||||
}
|
||||
|
||||
const isServiceUser = shellUsername === (process.env.USER ?? '');
|
||||
const userHasCredentials = !isServiceUser && (await hasOwnCredentials(homeDir));
|
||||
const authEnv = buildAuthEnv(shellUsername, homeDir, isServiceUser, userHasCredentials);
|
||||
|
||||
const env: Record<string, string> = {
|
||||
...authEnv,
|
||||
PATH: process.env.PATH ?? '',
|
||||
TERM: 'xterm-256color',
|
||||
};
|
||||
|
||||
const proc = isServiceUser
|
||||
? Bun.spawn(claudeArgs, { stdin: 'pipe', stdout: 'pipe', stderr: 'pipe', env: { ...process.env, ...env } })
|
||||
: Bun.spawn(
|
||||
['sudo', '-u', shellUsername, 'env', ...Object.entries(env).map(([k, v]) => `${k}=${v}`), ...claudeArgs],
|
||||
{ stdin: 'pipe', stdout: 'pipe', stderr: 'pipe' },
|
||||
);
|
||||
|
||||
const timeout = setTimeout(() => {
|
||||
try { proc.kill(); } catch { /* already dead */ }
|
||||
}, SEND_TIMEOUT_MS);
|
||||
|
||||
try {
|
||||
const [stdout, stderr] = await Promise.all([new Response(proc.stdout).text(), new Response(proc.stderr).text()]);
|
||||
const exitCode = await proc.exited;
|
||||
clearTimeout(timeout);
|
||||
|
||||
if (exitCode !== 0 && !stdout.trim()) {
|
||||
throw new Error(`claude exited with code ${exitCode}: ${stderr.trim().slice(0, 200)}`);
|
||||
}
|
||||
|
||||
let output: ClaudeCodeOutput;
|
||||
try {
|
||||
output = JSON.parse(stdout) as ClaudeCodeOutput;
|
||||
} catch {
|
||||
return {
|
||||
text: stdout.trim() || '(no response)',
|
||||
sessionId: sessionKey,
|
||||
model: 'claude-code',
|
||||
cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 },
|
||||
};
|
||||
}
|
||||
|
||||
if (output.is_error) {
|
||||
throw new Error(output.result || 'Claude Code returned an error');
|
||||
}
|
||||
|
||||
if (output.session_id) {
|
||||
setClaudeSession(sessionKey, output.session_id);
|
||||
}
|
||||
|
||||
return {
|
||||
text: output.result || '(no response)',
|
||||
sessionId: sessionKey,
|
||||
model: 'claude-code',
|
||||
cost: {
|
||||
inputTokens: output.input_tokens ?? 0,
|
||||
outputTokens: output.output_tokens ?? 0,
|
||||
totalUSD: output.cost_usd ?? 0,
|
||||
},
|
||||
};
|
||||
} catch (err) {
|
||||
clearTimeout(timeout);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
// ── Streaming send ──
|
||||
|
||||
export async function spawnClaudeStreaming(
|
||||
params: ClaudeSpawnStreamingParams,
|
||||
onEvent: (event: PiEvent) => void,
|
||||
): Promise<void> {
|
||||
const { userId, email, username, prompt, sessionKey, cwd } = params;
|
||||
const shellUsername = toShellUsername(username, email);
|
||||
const homeDir = getHomeDir(email);
|
||||
const workDir = cwd ?? homeDir;
|
||||
|
||||
const claudeArgs = [
|
||||
CLAUDE_BIN, '-p', prompt,
|
||||
'--dangerously-skip-permissions',
|
||||
'--output-format', 'stream-json',
|
||||
'--verbose', '--include-partial-messages',
|
||||
];
|
||||
|
||||
const subModel = params.model?.split('/')[1];
|
||||
if (subModel) claudeArgs.push('--model', subModel);
|
||||
|
||||
const existingSession = getClaudeSession(sessionKey);
|
||||
if (existingSession) {
|
||||
claudeArgs.push('--resume', existingSession);
|
||||
}
|
||||
|
||||
const { CLAUDECODE: _, ...cleanEnv } = process.env;
|
||||
|
||||
const isServiceUser = shellUsername === (process.env.USER ?? '');
|
||||
const userHasCredentials = !isServiceUser && (await hasOwnCredentials(homeDir));
|
||||
const authEnv = buildAuthEnv(shellUsername, homeDir, isServiceUser, userHasCredentials);
|
||||
|
||||
const env: Record<string, string> = {
|
||||
...authEnv,
|
||||
PATH: cleanEnv.PATH ?? '',
|
||||
TERM: 'xterm-256color',
|
||||
};
|
||||
|
||||
const proc = isServiceUser
|
||||
? Bun.spawn(claudeArgs, { cwd: workDir, stdin: 'ignore', stdout: 'pipe', stderr: 'pipe', env: { ...cleanEnv, ...env } })
|
||||
: Bun.spawn(
|
||||
['sudo', '-u', shellUsername, 'env', ...Object.entries(env).map(([k, v]) => `${k}=${v}`), ...claudeArgs],
|
||||
{ cwd: workDir, stdin: 'ignore', stdout: 'pipe', stderr: 'pipe' },
|
||||
);
|
||||
|
||||
activeProcs.set(sessionKey, proc);
|
||||
|
||||
const timeout = setTimeout(() => {
|
||||
try { proc.kill(); } catch { /* already dead */ }
|
||||
onEvent({ type: 'error', message: 'Claude Code timed out after 5 minutes' });
|
||||
}, SEND_TIMEOUT_MS);
|
||||
|
||||
// Process NDJSON stream
|
||||
try {
|
||||
const stdout = proc.stdout as ReadableStream<Uint8Array>;
|
||||
const reader = stdout.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = '';
|
||||
let textBuffer = '';
|
||||
let gotResult = false;
|
||||
|
||||
const processLine = (line: string) => {
|
||||
if (!line.trim()) return;
|
||||
try {
|
||||
const msg = JSON.parse(line) as Record<string, unknown>;
|
||||
const type = msg.type as string;
|
||||
|
||||
if (type === 'stream_event') {
|
||||
const event = msg.event as Record<string, unknown> | undefined;
|
||||
if (event?.type === 'content_block_delta') {
|
||||
const delta = event.delta as Record<string, unknown> | undefined;
|
||||
if (delta?.type === 'text_delta' && typeof delta.text === 'string') {
|
||||
textBuffer += delta.text;
|
||||
onEvent({ type: 'delta', text: delta.text });
|
||||
}
|
||||
}
|
||||
} else if (type === 'assistant') {
|
||||
const message = msg.message as Record<string, unknown> | undefined;
|
||||
const content = message?.content as Array<Record<string, unknown>> | undefined;
|
||||
if (Array.isArray(content)) {
|
||||
for (const block of content) {
|
||||
if (block.type === 'text' && typeof block.text === 'string') {
|
||||
onEvent({ type: 'text', text: block.text });
|
||||
textBuffer = '';
|
||||
} else if (block.type === 'tool_use') {
|
||||
if (textBuffer) {
|
||||
onEvent({ type: 'text', text: textBuffer });
|
||||
textBuffer = '';
|
||||
}
|
||||
onEvent({
|
||||
type: 'tool:start',
|
||||
toolCallId: (block.id as string) ?? '',
|
||||
toolName: (block.name as string) ?? 'unknown',
|
||||
toolInput: (block.input as Record<string, unknown>) ?? {},
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if (type === 'user') {
|
||||
const message = msg.message as Record<string, unknown> | undefined;
|
||||
const content = message?.content as Array<Record<string, unknown>> | undefined;
|
||||
if (Array.isArray(content)) {
|
||||
for (const block of content) {
|
||||
if (block.type === 'tool_result') {
|
||||
let output = '';
|
||||
if (typeof block.content === 'string') {
|
||||
output = block.content;
|
||||
} else if (Array.isArray(block.content)) {
|
||||
output = (block.content as Array<Record<string, unknown>>)
|
||||
.filter((c) => c.type === 'text')
|
||||
.map((c) => c.text as string)
|
||||
.join('\n');
|
||||
}
|
||||
onEvent({
|
||||
type: 'tool:result',
|
||||
toolCallId: (block.tool_use_id as string) ?? '',
|
||||
output,
|
||||
isError: (block.is_error as boolean) ?? false,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if (type === 'system' && msg.subtype === 'init') {
|
||||
const sessionId = msg.session_id as string | undefined;
|
||||
if (sessionId) {
|
||||
setClaudeSession(sessionKey, sessionId);
|
||||
}
|
||||
} else if (type === 'result') {
|
||||
gotResult = true;
|
||||
clearTimeout(timeout);
|
||||
|
||||
const isError = (msg.is_error as boolean) ?? false;
|
||||
const resultText = (msg.result as string) ?? '';
|
||||
|
||||
if (isError) {
|
||||
if (textBuffer) {
|
||||
onEvent({ type: 'text', text: textBuffer });
|
||||
textBuffer = '';
|
||||
}
|
||||
onEvent({ type: 'error', message: resultText || 'Claude Code returned an error' });
|
||||
return;
|
||||
}
|
||||
|
||||
if (textBuffer) {
|
||||
onEvent({ type: 'text', text: textBuffer });
|
||||
textBuffer = '';
|
||||
}
|
||||
|
||||
const usage = msg.usage as Record<string, number> | undefined;
|
||||
const cost: MessageCost = {
|
||||
inputTokens: usage?.input_tokens ?? 0,
|
||||
outputTokens: usage?.output_tokens ?? 0,
|
||||
totalUSD: (msg.total_cost_usd as number) ?? 0,
|
||||
};
|
||||
|
||||
const sessionId = msg.session_id as string | undefined;
|
||||
if (sessionId) {
|
||||
setClaudeSession(sessionKey, sessionId);
|
||||
}
|
||||
|
||||
onEvent({ type: 'result', cost });
|
||||
}
|
||||
} catch {
|
||||
// Skip malformed JSON lines
|
||||
}
|
||||
};
|
||||
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
const chunk = decoder.decode(value, { stream: true });
|
||||
buffer += chunk;
|
||||
const lines = buffer.split('\n');
|
||||
buffer = lines.pop()!;
|
||||
for (const line of lines) {
|
||||
processLine(line);
|
||||
}
|
||||
}
|
||||
|
||||
if (buffer.trim()) {
|
||||
processLine(buffer);
|
||||
}
|
||||
|
||||
clearTimeout(timeout);
|
||||
|
||||
if (!gotResult) {
|
||||
const exitCode = await proc.exited;
|
||||
const stderr = await new Response(proc.stderr as ReadableStream<Uint8Array>).text();
|
||||
if (textBuffer) {
|
||||
onEvent({ type: 'text', text: textBuffer });
|
||||
}
|
||||
if (exitCode !== 0) {
|
||||
onEvent({ type: 'error', message: `Claude Code exited with code ${exitCode}: ${stderr.trim().slice(0, 200)}` });
|
||||
} else {
|
||||
onEvent({ type: 'result', cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 } });
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
clearTimeout(timeout);
|
||||
onEvent({ type: 'error', message: String(err) });
|
||||
} finally {
|
||||
activeProcs.delete(sessionKey);
|
||||
}
|
||||
}
|
||||
|
||||
export function killClaudeSession(sessionKey: string): boolean {
|
||||
const proc = activeProcs.get(sessionKey);
|
||||
if (proc) {
|
||||
try { proc.kill(); } catch { /* already dead */ }
|
||||
activeProcs.delete(sessionKey);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
export function clearSession(sessionKey: string): void {
|
||||
clearClaudeSession(sessionKey);
|
||||
}
|
||||
|
||||
export function getActiveSessionKeys(): string[] {
|
||||
return Array.from(activeProcs.keys());
|
||||
}
|
||||
@@ -0,0 +1,259 @@
|
||||
import type { ServerWebSocket } from 'bun';
|
||||
import type { SidecarCommand, SidecarEvent, SidecarState } from './protocol';
|
||||
import { loadState, flushAndSave, acquireLock, releaseLock, getState } from './state';
|
||||
import { startAnthropicProxy, getProxySecret, ensureProxySecret } from './proxy';
|
||||
import * as claudeManager from './claude-manager';
|
||||
import * as piManager from './pi-manager';
|
||||
import * as queueRunner from './queue-runner';
|
||||
|
||||
const PORT = Number(process.env.SIDECAR_PORT ?? '5100');
|
||||
const startedAt = Date.now();
|
||||
|
||||
// ── Startup ──
|
||||
|
||||
if (!acquireLock()) {
|
||||
console.error('[sidecar] another instance is already running (lock file exists with live PID)');
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
loadState();
|
||||
ensureProxySecret();
|
||||
|
||||
// Start Anthropic proxy
|
||||
try {
|
||||
startAnthropicProxy();
|
||||
} catch (err) {
|
||||
console.error('[sidecar] failed to start Anthropic proxy:', err instanceof Error ? err.message : err);
|
||||
}
|
||||
|
||||
// Initialize queue
|
||||
queueRunner.initQueue().catch((err) => {
|
||||
console.error('[sidecar] failed to initialize queue:', err);
|
||||
});
|
||||
|
||||
// ── WebSocket connections ──
|
||||
|
||||
const clients = new Set<ServerWebSocket<unknown>>();
|
||||
|
||||
function broadcast(msg: SidecarEvent) {
|
||||
const data = JSON.stringify(msg);
|
||||
for (const ws of clients) {
|
||||
if (ws.readyState === 1) {
|
||||
ws.send(data);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function reply(ws: ServerWebSocket<unknown>, msg: SidecarEvent) {
|
||||
if (ws.readyState === 1) {
|
||||
ws.send(JSON.stringify(msg));
|
||||
}
|
||||
}
|
||||
|
||||
function buildState(): SidecarState {
|
||||
return {
|
||||
proxySecret: getProxySecret(),
|
||||
claudeSessions: { ...getState().claudeSessions },
|
||||
piSessions: piManager.getAllSessions(),
|
||||
uptime: Date.now() - startedAt,
|
||||
};
|
||||
}
|
||||
|
||||
// ── Command handlers ──
|
||||
|
||||
async function handleCommand(ws: ServerWebSocket<unknown>, cmd: SidecarCommand) {
|
||||
switch (cmd.type) {
|
||||
case 'ping':
|
||||
reply(ws, { type: 'pong', id: cmd.id });
|
||||
break;
|
||||
|
||||
case 'state:sync':
|
||||
reply(ws, { type: 'state:sync', id: cmd.id, state: buildState() });
|
||||
break;
|
||||
|
||||
case 'proxy:secret':
|
||||
reply(ws, { type: 'proxy:secret', id: cmd.id, secret: getProxySecret() });
|
||||
break;
|
||||
|
||||
// ── Claude Code ──
|
||||
|
||||
case 'claude:spawn': {
|
||||
try {
|
||||
const result = await claudeManager.spawnClaude(cmd.params);
|
||||
reply(ws, { type: 'claude:result', id: cmd.id, result });
|
||||
} catch (err) {
|
||||
reply(ws, { type: 'claude:error', id: cmd.id, error: err instanceof Error ? err.message : String(err) });
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
case 'claude:spawn-streaming': {
|
||||
reply(ws, { type: 'claude:spawned', id: cmd.id, sessionKey: cmd.params.sessionKey });
|
||||
|
||||
const onEvent = (event: import('../api/pi/types').PiEvent) => {
|
||||
broadcast({ type: 'claude:event', sessionKey: cmd.params.sessionKey, event });
|
||||
};
|
||||
|
||||
claudeManager.spawnClaudeStreaming(cmd.params, onEvent).catch((err) => {
|
||||
broadcast({
|
||||
type: 'claude:event',
|
||||
sessionKey: cmd.params.sessionKey,
|
||||
event: { type: 'error', message: err instanceof Error ? err.message : String(err) },
|
||||
});
|
||||
});
|
||||
break;
|
||||
}
|
||||
|
||||
case 'claude:kill':
|
||||
claudeManager.killClaudeSession(cmd.sessionKey);
|
||||
reply(ws, { type: 'claude:killed', id: cmd.id });
|
||||
break;
|
||||
|
||||
case 'claude:clear-session':
|
||||
claudeManager.clearSession(cmd.sessionKey);
|
||||
reply(ws, { type: 'claude:session-cleared', id: cmd.id });
|
||||
break;
|
||||
|
||||
// ── Pi ──
|
||||
|
||||
case 'pi:spawn': {
|
||||
try {
|
||||
const onEvent = (event: import('../api/pi/types').PiEvent) => {
|
||||
broadcast({ 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(ws, { type: 'pi:spawned', id: cmd.id, sessionId: cmd.params.sessionId });
|
||||
} catch (err) {
|
||||
reply(ws, { 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(ws, { type: 'pi:killed', id: cmd.id });
|
||||
break;
|
||||
|
||||
case 'pi:set-thinking':
|
||||
piManager.setThinkingLevel(cmd.sessionId, cmd.level);
|
||||
break;
|
||||
|
||||
// ── Queue ──
|
||||
|
||||
case 'queue:enqueue': {
|
||||
try {
|
||||
const job = await queueRunner.enqueue(cmd.params);
|
||||
reply(ws, { type: 'queue:enqueued', id: cmd.id, job });
|
||||
} catch (err) {
|
||||
reply(ws, { type: 'queue:error', id: cmd.id, error: err instanceof Error ? err.message : String(err) });
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
case 'queue:cancel': {
|
||||
try {
|
||||
const job = await queueRunner.cancelJob(cmd.jobId);
|
||||
reply(ws, { type: 'queue:cancelled', id: cmd.id, job });
|
||||
} catch (err) {
|
||||
reply(ws, { type: 'queue:error', id: cmd.id, error: err instanceof Error ? err.message : String(err) });
|
||||
}
|
||||
break;
|
||||
}
|
||||
|
||||
case 'queue:list': {
|
||||
const jobs = await queueRunner.listAllJobs();
|
||||
reply(ws, { type: 'queue:list', id: cmd.id, jobs });
|
||||
break;
|
||||
}
|
||||
|
||||
case 'queue:get': {
|
||||
const job = await queueRunner.readJob(cmd.jobId);
|
||||
reply(ws, { type: 'queue:get', id: cmd.id, job });
|
||||
break;
|
||||
}
|
||||
|
||||
default:
|
||||
reply(ws, { type: 'error', id: (cmd as SidecarCommand).id, error: `Unknown command type: ${(cmd as Record<string, unknown>).type}` });
|
||||
}
|
||||
}
|
||||
|
||||
// ── Server ──
|
||||
|
||||
const server = Bun.serve({
|
||||
port: PORT,
|
||||
hostname: '127.0.0.1',
|
||||
|
||||
routes: {
|
||||
'/': () => new Response('process-sidecar'),
|
||||
'/health': () => new Response(JSON.stringify({
|
||||
status: 'ok',
|
||||
uptime: Date.now() - startedAt,
|
||||
piSessions: piManager.getAllSessions().length,
|
||||
claudeSessions: claudeManager.getActiveSessionKeys().length,
|
||||
}), { headers: { 'Content-Type': 'application/json' } }),
|
||||
},
|
||||
|
||||
fetch(req, server) {
|
||||
if (req.headers.get('upgrade') === 'websocket') {
|
||||
const ok = server.upgrade(req);
|
||||
if (!ok) return new Response('Upgrade failed', { status: 500 });
|
||||
return;
|
||||
}
|
||||
return new Response('Not found', { status: 404 });
|
||||
},
|
||||
|
||||
websocket: {
|
||||
open(ws) {
|
||||
clients.add(ws);
|
||||
console.log(`[sidecar] client connected (${clients.size} total)`);
|
||||
},
|
||||
message(ws, raw) {
|
||||
try {
|
||||
const data = typeof raw === 'string' ? raw : raw.toString();
|
||||
const cmd = JSON.parse(data) as SidecarCommand;
|
||||
handleCommand(ws, cmd);
|
||||
} catch (err) {
|
||||
reply(ws, { type: 'error', error: `Invalid message: ${err instanceof Error ? err.message : String(err)}` });
|
||||
}
|
||||
},
|
||||
close(ws) {
|
||||
clients.delete(ws);
|
||||
console.log(`[sidecar] client disconnected (${clients.size} total)`);
|
||||
},
|
||||
drain() {},
|
||||
},
|
||||
});
|
||||
|
||||
console.log(`[sidecar] listening on 127.0.0.1:${PORT}`);
|
||||
|
||||
// ── Graceful shutdown ──
|
||||
|
||||
async function shutdown(signal: string) {
|
||||
console.log(`[sidecar] ${signal} received, saving state...`);
|
||||
await flushAndSave();
|
||||
releaseLock();
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
process.on('SIGTERM', () => shutdown('SIGTERM'));
|
||||
process.on('SIGINT', () => shutdown('SIGINT'));
|
||||
@@ -0,0 +1,416 @@
|
||||
import { join } from 'node:path';
|
||||
import { readdirSync, existsSync, mkdirSync, readFileSync } from 'node:fs';
|
||||
import type { Subprocess } from 'bun';
|
||||
import type { PiEvent, MessageCost } from '../api/pi/types';
|
||||
import type { PiSpawnParams, PiSessionInfo } from './protocol';
|
||||
import { isPidAlive } from './state';
|
||||
|
||||
const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data');
|
||||
const PI_CONFIG_DIR = join(require('node:os').homedir(), '.pi', 'agent');
|
||||
const SEED_PATH = join(import.meta.dir, '../../../seed');
|
||||
|
||||
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 getGlobalSkillsDir = () => join(DATA_PATH, 'skills');
|
||||
const getUserSkillsDir = (email: string) => join(DATA_PATH, email, 'skills');
|
||||
const getGlobalExtensionsDir = () => join(DATA_PATH, 'extensions');
|
||||
const getUserExtensionsDir = (email: string) => join(DATA_PATH, email, 'extensions');
|
||||
const getGlobalToolsDir = () => join(DATA_PATH, 'tools');
|
||||
const getUserToolsDir = (email: string) => join(DATA_PATH, email, 'tools');
|
||||
const getNativeResourcesDir = () => join(SEED_PATH, 'resources');
|
||||
const getGlobalResourcesDir = () => join(DATA_PATH, 'resources');
|
||||
|
||||
const toShellUsername = (username: string, email: string): string => {
|
||||
const raw = username || email.split('@')[0]!;
|
||||
return raw.replace(/@.*$/, '').replace(/[^a-zA-Z0-9._-]/g, '_').toLowerCase().slice(0, 32) || 'officer';
|
||||
};
|
||||
|
||||
// Resolve pi as [node, cli.js]
|
||||
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];
|
||||
})();
|
||||
|
||||
// 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 collectSkillFlags(email: string): string[] {
|
||||
const flags: string[] = [];
|
||||
const dirs = [getGlobalSkillsDir(), getUserSkillsDir(email)];
|
||||
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(email: string): string[] {
|
||||
const flags: string[] = [];
|
||||
const dirs = [getGlobalExtensionsDir(), getUserExtensionsDir(email)];
|
||||
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;
|
||||
}
|
||||
|
||||
function buildResourcesEnv(): string {
|
||||
const nativeDir = getNativeResourcesDir();
|
||||
const globalDir = getGlobalResourcesDir();
|
||||
const resourceDirs = new Map<string, string>();
|
||||
for (const dir of [nativeDir, globalDir]) {
|
||||
if (!existsSync(dir)) continue;
|
||||
for (const entry of readdirSync(dir, { withFileTypes: true })) {
|
||||
if (!entry.isDirectory()) continue;
|
||||
if (existsSync(join(dir, entry.name, 'RESOURCE.md')) || existsSync(join(dir, entry.name, 'config.json'))) {
|
||||
resourceDirs.set(entry.name, dir);
|
||||
}
|
||||
}
|
||||
}
|
||||
const result: Record<string, Record<string, string>> = {};
|
||||
for (const [name] of resourceDirs) {
|
||||
let nativeConfig: Record<string, string> = {};
|
||||
let globalConfig: Record<string, string> = {};
|
||||
try { nativeConfig = JSON.parse(readFileSync(join(nativeDir, name, 'config.json'), 'utf-8')); } catch {}
|
||||
try { globalConfig = JSON.parse(readFileSync(join(globalDir, name, 'config.json'), 'utf-8')); } catch {}
|
||||
const config: Record<string, string> = {};
|
||||
for (const key of Object.keys(nativeConfig)) config[key] = globalConfig[key] ?? nativeConfig[key]!;
|
||||
for (const key of Object.keys(globalConfig)) if (!(key in config)) config[key] = globalConfig[key]!;
|
||||
if (Object.values(config).some((v) => v !== '')) {
|
||||
result[name] = config;
|
||||
}
|
||||
}
|
||||
return JSON.stringify(result);
|
||||
}
|
||||
|
||||
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('[sidecar: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 skillFlags = collectSkillFlags(email);
|
||||
const extensionFlags = collectExtensionFlags(email);
|
||||
|
||||
// Generate resource skill
|
||||
const { generateResourceSkill } = await import('../api/pi/pi-bridge');
|
||||
const resourceSkillDir = generateResourceSkill(DATA_PATH);
|
||||
const resourceSkillFlags = resourceSkillDir ? ['--skill', resourceSkillDir] : [];
|
||||
|
||||
const piArgs = [
|
||||
...PI_CMD,
|
||||
'--mode', 'rpc',
|
||||
'--no-skills', '--no-prompt-templates', '--no-themes',
|
||||
...skillFlags,
|
||||
...extensionFlags,
|
||||
...resourceSkillFlags,
|
||||
];
|
||||
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 = [getGlobalToolsDir(), getUserToolsDir(email)].join(':');
|
||||
const shellUsername = username ? toShellUsername(username, email) : toShellUsername('', email);
|
||||
const isServiceUser = shellUsername === (process.env.USER ?? '');
|
||||
|
||||
const env: Record<string, string> = {
|
||||
HOME: isServiceUser ? (process.env.HOME ?? '') : homeDir,
|
||||
OFFICER_USER_HOME: homeDir,
|
||||
OFFICER_USER_ROOT: join(DATA_PATH, email),
|
||||
PI_CODING_AGENT_DIR: isServiceUser ? PI_CONFIG_DIR : join(homeDir, '.pi', 'agent'),
|
||||
PI_TOOLS_DIRS: toolsDirs,
|
||||
OFFICER_RESOURCES: buildResourcesEnv(),
|
||||
OFFICER_EMAIL_DB: join(DATA_PATH, email, 'emails.db'),
|
||||
TERM: 'xterm-256color',
|
||||
PATH: process.env.PATH ?? '',
|
||||
};
|
||||
|
||||
const proc = isServiceUser
|
||||
? Bun.spawn(piArgs, { cwd, stdin: 'pipe', stdout: 'pipe', stderr: 'pipe', env: { ...process.env, ...env } })
|
||||
: Bun.spawn(
|
||||
['sudo', '-u', shellUsername, 'env', ...Object.entries(env).map(([k, v]) => `${k}=${v}`), ...piArgs],
|
||||
{ cwd, stdin: 'pipe', stdout: 'pipe', stderr: 'pipe' },
|
||||
);
|
||||
|
||||
const session: PiSession = { sessionId, email, userId, model, cwd, proc, onEvent };
|
||||
sessions.set(sessionId, session);
|
||||
|
||||
console.log(`[sidecar: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(`[sidecar:pi:stderr] ${text.trim()}`);
|
||||
}
|
||||
} catch { /* process ended */ }
|
||||
})();
|
||||
|
||||
// Handle exit
|
||||
proc.exited.then((code) => {
|
||||
sessions.delete(sessionId);
|
||||
if (code !== 0) {
|
||||
console.error(`[sidecar: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),
|
||||
}));
|
||||
}
|
||||
@@ -0,0 +1,115 @@
|
||||
import type { MessageCost, PiEvent } from '../api/pi/types';
|
||||
import type { Job, EnqueueParams, JobProgress } from '../queue/types';
|
||||
|
||||
// ── Envelope ──
|
||||
|
||||
export type SidecarMessage = SidecarCommand | SidecarEvent;
|
||||
|
||||
// ── Commands (API server → sidecar) ──
|
||||
|
||||
export type SidecarCommand =
|
||||
| { type: 'ping'; id: string }
|
||||
| { type: 'state:sync'; id: string }
|
||||
// Proxy
|
||||
| { type: 'proxy:secret'; id: string }
|
||||
// Claude Code
|
||||
| { type: 'claude:spawn'; id: string; params: ClaudeSpawnParams }
|
||||
| { 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 }
|
||||
// Queue
|
||||
| { type: 'queue:enqueue'; id: string; params: EnqueueParams }
|
||||
| { type: 'queue:cancel'; id: string; jobId: string }
|
||||
| { type: 'queue:list'; id: string }
|
||||
| { type: 'queue:get'; id: string; jobId: string };
|
||||
|
||||
// ── Responses/Events (sidecar → API server) ──
|
||||
|
||||
export type SidecarEvent =
|
||||
| { type: 'pong'; id: string }
|
||||
| { type: 'state:sync'; id: string; state: SidecarState }
|
||||
| { type: 'proxy:secret'; id: string; secret: string }
|
||||
// Claude Code
|
||||
| { type: 'claude:spawned'; id: string; sessionKey: string }
|
||||
| { type: 'claude:event'; sessionKey: string; event: PiEvent }
|
||||
| { type: 'claude:result'; id: string; result: ClaudeCodeResult }
|
||||
| { 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 }
|
||||
// Queue
|
||||
| { type: 'queue:enqueued'; id: string; job: Job }
|
||||
| { type: 'queue:cancelled'; id: string; job: Job | null }
|
||||
| { type: 'queue:list'; id: string; jobs: Job[] }
|
||||
| { type: 'queue:get'; id: string; job: Job | null }
|
||||
| { type: 'queue:error'; id: string; error: string }
|
||||
// Generic
|
||||
| { type: 'error'; id?: string; error: string };
|
||||
|
||||
// ── Shared state snapshot ──
|
||||
|
||||
export type SidecarState = {
|
||||
proxySecret: string;
|
||||
claudeSessions: Record<string, string>; // sessionKey → Claude Code session_id
|
||||
piSessions: PiSessionInfo[];
|
||||
uptime: number;
|
||||
};
|
||||
|
||||
export type PiSessionInfo = {
|
||||
sessionId: string;
|
||||
email: string;
|
||||
userId: number;
|
||||
model: string;
|
||||
cwd: string;
|
||||
pid: number;
|
||||
alive: boolean;
|
||||
};
|
||||
|
||||
// ── Param types ──
|
||||
|
||||
export type ClaudeSpawnParams = {
|
||||
userId: number;
|
||||
email: string;
|
||||
username: string;
|
||||
prompt: string;
|
||||
sessionKey: string;
|
||||
model?: string;
|
||||
};
|
||||
|
||||
export type ClaudeSpawnStreamingParams = {
|
||||
userId: number;
|
||||
email: string;
|
||||
username: string;
|
||||
prompt: string;
|
||||
sessionKey: string;
|
||||
cwd?: string;
|
||||
model?: string;
|
||||
};
|
||||
|
||||
export type ClaudeCodeResult = {
|
||||
text: string;
|
||||
sessionId: string;
|
||||
model: string;
|
||||
cost: MessageCost;
|
||||
};
|
||||
|
||||
export type PiSpawnParams = {
|
||||
sessionId: string;
|
||||
email: string;
|
||||
userId: number;
|
||||
username: string;
|
||||
role: string;
|
||||
cwd: string;
|
||||
model: string;
|
||||
sessionFile?: string;
|
||||
};
|
||||
@@ -0,0 +1,104 @@
|
||||
import { join } from 'node:path';
|
||||
import { homedir } from 'node:os';
|
||||
import { getState, updateState } from './state';
|
||||
|
||||
const PROXY_PORT = Number(process.env.ANTHROPIC_PROXY_PORT ?? '5051');
|
||||
const ANTHROPIC_API_BASE = 'https://api.anthropic.com';
|
||||
const CREDENTIALS_PATH = join(homedir(), '.claude', '.credentials.json');
|
||||
|
||||
type CredentialsFile = {
|
||||
claudeAiOauth?: {
|
||||
accessToken?: string;
|
||||
};
|
||||
};
|
||||
|
||||
async function readOAuthToken(): Promise<string | null> {
|
||||
try {
|
||||
const file = Bun.file(CREDENTIALS_PATH);
|
||||
if (!(await file.exists())) return null;
|
||||
const data = (await file.json()) as CredentialsFile;
|
||||
return data.claudeAiOauth?.accessToken?.trim() || null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export function getProxySecret(): string {
|
||||
return getState().proxySecret;
|
||||
}
|
||||
|
||||
export function ensureProxySecret(): string {
|
||||
const state = getState();
|
||||
if (state.proxySecret) return state.proxySecret;
|
||||
|
||||
// Generate once, persist forever
|
||||
const secret = `sk-ant-api03-${crypto.randomUUID()}`;
|
||||
updateState({ proxySecret: secret });
|
||||
return secret;
|
||||
}
|
||||
|
||||
export function startAnthropicProxy() {
|
||||
const secret = ensureProxySecret();
|
||||
|
||||
Bun.serve({
|
||||
port: PROXY_PORT,
|
||||
hostname: '127.0.0.1',
|
||||
idleTimeout: 0,
|
||||
|
||||
async fetch(req) {
|
||||
const url = new URL(req.url);
|
||||
|
||||
// Validate proxy secret
|
||||
const incomingKey = req.headers.get('x-api-key');
|
||||
if (incomingKey !== secret) {
|
||||
return new Response(JSON.stringify({ error: 'Unauthorized' }), {
|
||||
status: 401,
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
});
|
||||
}
|
||||
|
||||
// Read fresh OAuth token
|
||||
const token = await readOAuthToken();
|
||||
if (!token) {
|
||||
return new Response(JSON.stringify({ error: 'No OAuth token available' }), {
|
||||
status: 502,
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
});
|
||||
}
|
||||
|
||||
// Build upstream URL
|
||||
const upstream = `${ANTHROPIC_API_BASE}${url.pathname}${url.search}`;
|
||||
|
||||
// Clone headers, replace proxy secret with real OAuth token
|
||||
const headers = new Headers(req.headers);
|
||||
headers.set('x-api-key', token);
|
||||
headers.delete('host');
|
||||
|
||||
// Read request body fully before forwarding
|
||||
const body = req.body ? await req.arrayBuffer() : null;
|
||||
|
||||
// Forward request
|
||||
const upstreamRes = await fetch(upstream, {
|
||||
method: req.method,
|
||||
headers,
|
||||
body,
|
||||
});
|
||||
|
||||
// Build clean response headers
|
||||
const resHeaders = new Headers();
|
||||
for (const [key, value] of upstreamRes.headers) {
|
||||
const lower = key.toLowerCase();
|
||||
if (lower === 'content-encoding' || lower === 'content-length' || lower === 'transfer-encoding') continue;
|
||||
resHeaders.set(key, value);
|
||||
}
|
||||
|
||||
return new Response(upstreamRes.body, {
|
||||
status: upstreamRes.status,
|
||||
statusText: upstreamRes.statusText,
|
||||
headers: resHeaders,
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
console.log(`[sidecar:proxy] listening on 127.0.0.1:${PROXY_PORT}`);
|
||||
}
|
||||
@@ -0,0 +1,266 @@
|
||||
import { type Job, type JobProgress, type EnqueueParams, type StepContext, PermanentError } from '../queue/types';
|
||||
import { readJob, writeJob, listAllJobs, ensureQueueDir } from '../queue/storage';
|
||||
import { getHandler } from '../queue/handler-registry';
|
||||
|
||||
// Import handlers to register them
|
||||
import '../queue/handlers';
|
||||
|
||||
const activeLanes = new Map<string, boolean>();
|
||||
const PROGRESS_THROTTLE_MS = 1000;
|
||||
|
||||
export async function initQueue() {
|
||||
await ensureQueueDir();
|
||||
await resumeInterruptedJobs();
|
||||
console.log('[sidecar:queue] initialized');
|
||||
}
|
||||
|
||||
export async function enqueue(params: EnqueueParams): Promise<Job> {
|
||||
const handler = getHandler(params.type);
|
||||
if (!handler) throw new Error(`No handler registered for job type: ${params.type}`);
|
||||
|
||||
const job: Job = {
|
||||
id: crypto.randomUUID(),
|
||||
lane: params.lane,
|
||||
type: params.type,
|
||||
userId: params.userId,
|
||||
status: 'queued',
|
||||
steps: handler.steps.map((s) => ({ name: s.name, status: 'pending' as const })),
|
||||
currentStep: 0,
|
||||
createdAt: Date.now(),
|
||||
meta: params.meta,
|
||||
notify: params.notify,
|
||||
};
|
||||
|
||||
await writeJob(job);
|
||||
console.log(`[sidecar:queue] enqueued job ${job.id} (${job.type}) in lane ${job.lane}`);
|
||||
kickLane(job.lane);
|
||||
return job;
|
||||
}
|
||||
|
||||
export async function cancelJob(id: string): Promise<Job | null> {
|
||||
const job = await readJob(id);
|
||||
if (!job) return null;
|
||||
if (job.status === 'completed' || job.status === 'failed' || job.status === 'cancelled') return job;
|
||||
|
||||
job.status = 'cancelled';
|
||||
job.completedAt = Date.now();
|
||||
for (const step of job.steps) {
|
||||
if (step.status === 'pending' || step.status === 'running') {
|
||||
step.status = 'failed';
|
||||
step.error = 'Cancelled';
|
||||
}
|
||||
}
|
||||
await writeJob(job);
|
||||
console.log(`[sidecar:queue] cancelled job ${job.id}`);
|
||||
return job;
|
||||
}
|
||||
|
||||
async function resumeInterruptedJobs() {
|
||||
const jobs = await listAllJobs();
|
||||
const lanesToKick = new Set<string>();
|
||||
|
||||
for (const job of jobs) {
|
||||
if (job.status === 'running') {
|
||||
job.status = 'queued';
|
||||
job.startedAt = undefined;
|
||||
for (const step of job.steps) {
|
||||
if (step.status === 'running') {
|
||||
step.status = 'pending';
|
||||
step.startedAt = undefined;
|
||||
}
|
||||
}
|
||||
await writeJob(job);
|
||||
console.log(`[sidecar:queue] reset interrupted job ${job.id} back to queued`);
|
||||
lanesToKick.add(job.lane);
|
||||
} else if (job.status === 'queued') {
|
||||
lanesToKick.add(job.lane);
|
||||
}
|
||||
}
|
||||
|
||||
for (const lane of lanesToKick) {
|
||||
kickLane(lane);
|
||||
}
|
||||
}
|
||||
|
||||
function kickLane(lane: string) {
|
||||
if (activeLanes.get(lane)) return;
|
||||
activeLanes.set(lane, true);
|
||||
processNextInLane(lane);
|
||||
}
|
||||
|
||||
function scheduleRetry(lane: string, delayMs: number) {
|
||||
setTimeout(() => kickLane(lane), delayMs);
|
||||
}
|
||||
|
||||
async function processNextInLane(lane: string) {
|
||||
try {
|
||||
const jobs = await listAllJobs();
|
||||
const now = Date.now();
|
||||
const next = jobs
|
||||
.filter((j) => j.lane === lane && j.status === 'queued' && (!j.retryAt || j.retryAt <= now))
|
||||
.sort((a, b) => a.createdAt - b.createdAt)[0];
|
||||
|
||||
if (!next) {
|
||||
activeLanes.set(lane, false);
|
||||
return;
|
||||
}
|
||||
|
||||
await runJob(next);
|
||||
} catch (err) {
|
||||
console.error(`[sidecar:queue] lane ${lane} processing error:`, err);
|
||||
} finally {
|
||||
const jobs = await listAllJobs();
|
||||
const hasMore = jobs.some(
|
||||
(j) => j.lane === lane && j.status === 'queued' && (!j.retryAt || j.retryAt <= Date.now()),
|
||||
);
|
||||
if (hasMore) {
|
||||
processNextInLane(lane);
|
||||
} else {
|
||||
activeLanes.set(lane, false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function runJob(job: Job) {
|
||||
const handler = getHandler(job.type);
|
||||
if (!handler) {
|
||||
job.status = 'failed';
|
||||
job.error = `No handler for type: ${job.type}`;
|
||||
job.completedAt = Date.now();
|
||||
await writeJob(job);
|
||||
return;
|
||||
}
|
||||
|
||||
job.status = 'running';
|
||||
job.startedAt = Date.now();
|
||||
job.retryAt = undefined;
|
||||
await writeJob(job);
|
||||
const isRetry = (job.retries ?? 0) > 0;
|
||||
console.log(
|
||||
`[sidecar:queue] ${isRetry ? 'resuming' : 'running'} job ${job.id} (${job.type})${isRetry ? ` retry ${job.retries}` : ''}`,
|
||||
);
|
||||
|
||||
const sharedMeta: Record<string, unknown> = { ...(job.meta ?? {}) };
|
||||
|
||||
for (let i = 0; i < handler.steps.length; i++) {
|
||||
const fresh = await readJob(job.id);
|
||||
if (!fresh || fresh.status === 'cancelled') {
|
||||
console.log(`[sidecar:queue] job ${job.id} was cancelled, stopping`);
|
||||
return;
|
||||
}
|
||||
|
||||
const handlerStep = handler.steps[i]!;
|
||||
const step = fresh.steps[i]!;
|
||||
|
||||
if (step.status === 'completed') continue;
|
||||
|
||||
fresh.currentStep = i;
|
||||
step.status = 'running';
|
||||
step.startedAt = Date.now();
|
||||
await writeJob(fresh);
|
||||
|
||||
let lastProgressWrite = 0;
|
||||
let pendingProgress: JobProgress | null = null;
|
||||
|
||||
const updateProgress = async (progress: JobProgress) => {
|
||||
step.progress = progress;
|
||||
const now = Date.now();
|
||||
if (now - lastProgressWrite >= PROGRESS_THROTTLE_MS) {
|
||||
lastProgressWrite = now;
|
||||
pendingProgress = null;
|
||||
await writeJob(fresh);
|
||||
} else {
|
||||
pendingProgress = progress;
|
||||
}
|
||||
};
|
||||
|
||||
const ctx: StepContext = { job: fresh, step, updateProgress, meta: sharedMeta };
|
||||
|
||||
try {
|
||||
await handlerStep.run(ctx);
|
||||
|
||||
if (pendingProgress) {
|
||||
step.progress = pendingProgress;
|
||||
}
|
||||
step.status = 'completed';
|
||||
step.completedAt = Date.now();
|
||||
await writeJob(fresh);
|
||||
} catch (err) {
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
step.status = 'failed';
|
||||
step.error = errorMessage;
|
||||
step.completedAt = Date.now();
|
||||
|
||||
const isPermanent = err instanceof PermanentError;
|
||||
const retries = (fresh.retries ?? 0) + 1;
|
||||
if (!isPermanent && handler.retry && retries <= handler.retry.maxRetries) {
|
||||
step.status = 'pending';
|
||||
step.error = undefined;
|
||||
step.startedAt = undefined;
|
||||
step.completedAt = undefined;
|
||||
step.progress = undefined;
|
||||
fresh.status = 'queued';
|
||||
fresh.error = undefined;
|
||||
fresh.completedAt = undefined;
|
||||
fresh.startedAt = undefined;
|
||||
fresh.retries = retries;
|
||||
fresh.retryAt = Date.now() + handler.retry.delayMs;
|
||||
await writeJob(fresh);
|
||||
console.log(
|
||||
`[sidecar:queue] job ${fresh.id} will retry (${retries}/${handler.retry.maxRetries}) in ${handler.retry.delayMs / 1000}s`,
|
||||
);
|
||||
scheduleRetry(fresh.lane, handler.retry.delayMs);
|
||||
return;
|
||||
}
|
||||
|
||||
fresh.status = 'failed';
|
||||
fresh.error = `Step "${step.name}" failed: ${errorMessage}`;
|
||||
fresh.completedAt = Date.now();
|
||||
fresh.meta = { ...fresh.meta, ...sharedMeta };
|
||||
await writeJob(fresh);
|
||||
console.error(`[sidecar:queue] job ${fresh.id} failed at step "${step.name}":`, errorMessage);
|
||||
await notifyFailure(fresh);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
const final = await readJob(job.id);
|
||||
if (final && final.status === 'running') {
|
||||
final.status = 'completed';
|
||||
final.completedAt = Date.now();
|
||||
final.meta = { ...final.meta, ...sharedMeta };
|
||||
await writeJob(final);
|
||||
console.log(`[sidecar:queue] job ${final.id} completed`);
|
||||
await notifyCompletion(final);
|
||||
}
|
||||
}
|
||||
|
||||
async function notifyCompletion(job: Job) {
|
||||
try {
|
||||
const { sendMail } = await import('emailer');
|
||||
await sendMail({
|
||||
template: 'JobCompleted',
|
||||
subject: `Job completed: ${job.type}`,
|
||||
to: job.userId,
|
||||
data: { job },
|
||||
});
|
||||
} catch {
|
||||
// SMTP might not be configured — non-fatal
|
||||
}
|
||||
}
|
||||
|
||||
async function notifyFailure(job: Job) {
|
||||
try {
|
||||
const { sendMail } = await import('emailer');
|
||||
await sendMail({
|
||||
template: 'JobFailed',
|
||||
subject: `Job failed: ${job.type}`,
|
||||
to: job.userId,
|
||||
data: { job },
|
||||
});
|
||||
} catch {
|
||||
// SMTP might not be configured — non-fatal
|
||||
}
|
||||
}
|
||||
|
||||
export { readJob, listAllJobs };
|
||||
@@ -0,0 +1,141 @@
|
||||
import { join } from 'node:path';
|
||||
import { mkdirSync, existsSync } from 'node:fs';
|
||||
|
||||
const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data');
|
||||
const STATE_DIR = join(DATA_PATH, 'sidecar');
|
||||
const STATE_FILE = join(STATE_DIR, 'state.json');
|
||||
const LOCK_FILE = join(STATE_DIR, 'sidecar.lock');
|
||||
|
||||
export type PersistedState = {
|
||||
proxySecret: string;
|
||||
claudeSessions: Record<string, string>; // sessionKey → Claude Code session_id
|
||||
piSessions: Array<{
|
||||
sessionId: string;
|
||||
email: string;
|
||||
userId: number;
|
||||
model: string;
|
||||
cwd: string;
|
||||
pid: number;
|
||||
}>;
|
||||
};
|
||||
|
||||
const DEFAULT_STATE: PersistedState = {
|
||||
proxySecret: '',
|
||||
claudeSessions: {},
|
||||
piSessions: [],
|
||||
};
|
||||
|
||||
let currentState: PersistedState = { ...DEFAULT_STATE };
|
||||
let saveTimer: Timer | null = null;
|
||||
|
||||
function ensureDir() {
|
||||
if (!existsSync(STATE_DIR)) {
|
||||
mkdirSync(STATE_DIR, { recursive: true });
|
||||
}
|
||||
}
|
||||
|
||||
export function loadState(): PersistedState {
|
||||
ensureDir();
|
||||
try {
|
||||
const raw = Bun.file(STATE_FILE);
|
||||
// Synchronous check — Bun.file doesn't have sync exists, use fs
|
||||
if (!existsSync(STATE_FILE)) {
|
||||
currentState = { ...DEFAULT_STATE };
|
||||
return currentState;
|
||||
}
|
||||
// We need to read synchronously at startup
|
||||
const text = require('node:fs').readFileSync(STATE_FILE, 'utf-8');
|
||||
currentState = { ...DEFAULT_STATE, ...JSON.parse(text) };
|
||||
return currentState;
|
||||
} catch {
|
||||
currentState = { ...DEFAULT_STATE };
|
||||
return currentState;
|
||||
}
|
||||
}
|
||||
|
||||
export async function saveState(): Promise<void> {
|
||||
ensureDir();
|
||||
await Bun.write(STATE_FILE, JSON.stringify(currentState, null, 2));
|
||||
}
|
||||
|
||||
export function getState(): PersistedState {
|
||||
return currentState;
|
||||
}
|
||||
|
||||
export function updateState(patch: Partial<PersistedState>): void {
|
||||
Object.assign(currentState, patch);
|
||||
scheduleSave();
|
||||
}
|
||||
|
||||
export function setClaudeSession(sessionKey: string, sessionId: string): void {
|
||||
currentState.claudeSessions[sessionKey] = sessionId;
|
||||
scheduleSave();
|
||||
}
|
||||
|
||||
export function clearClaudeSession(sessionKey: string): void {
|
||||
delete currentState.claudeSessions[sessionKey];
|
||||
scheduleSave();
|
||||
}
|
||||
|
||||
export function getClaudeSession(sessionKey: string): string | undefined {
|
||||
return currentState.claudeSessions[sessionKey];
|
||||
}
|
||||
|
||||
function scheduleSave() {
|
||||
if (saveTimer) return;
|
||||
saveTimer = setTimeout(async () => {
|
||||
saveTimer = null;
|
||||
await saveState();
|
||||
}, 30_000);
|
||||
}
|
||||
|
||||
export async function flushAndSave(): Promise<void> {
|
||||
if (saveTimer) {
|
||||
clearTimeout(saveTimer);
|
||||
saveTimer = null;
|
||||
}
|
||||
await saveState();
|
||||
}
|
||||
|
||||
// ── Lockfile ──
|
||||
|
||||
export function acquireLock(): boolean {
|
||||
ensureDir();
|
||||
try {
|
||||
if (existsSync(LOCK_FILE)) {
|
||||
const pidStr = require('node:fs').readFileSync(LOCK_FILE, 'utf-8').trim();
|
||||
const pid = Number(pidStr);
|
||||
if (pid && isProcessAlive(pid)) {
|
||||
return false; // another sidecar is running
|
||||
}
|
||||
// Stale lock — remove it
|
||||
}
|
||||
require('node:fs').writeFileSync(LOCK_FILE, String(process.pid));
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
export function releaseLock(): void {
|
||||
try {
|
||||
if (existsSync(LOCK_FILE)) {
|
||||
require('node:fs').unlinkSync(LOCK_FILE);
|
||||
}
|
||||
} catch {
|
||||
// best effort
|
||||
}
|
||||
}
|
||||
|
||||
function isProcessAlive(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
export function isPidAlive(pid: number): boolean {
|
||||
return isProcessAlive(pid);
|
||||
}
|
||||
Reference in New Issue
Block a user