From d88fe3cac72d43f488992155c3e8fef648c100f1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Fri, 6 Mar 2026 11:49:39 +0000 Subject: [PATCH] task logs: migrate from filesystem to postgresql; refactor sidecars into submodules Co-Authored-By: Claude Opus 4.6 --- scripts/setup-pty-sidecar.sh | 70 ------ scripts/setup.sh | 3 - .../Screens/Dashboard/TaskLogs/index.tsx | 194 ++++++++-------- src/server.tsx | 43 +++- src/servers/api/email/accounts.ts | 6 +- src/servers/api/queue/queue.ts | 10 +- src/servers/api/task-logger.ts | 42 ++-- src/servers/api/task-logs/task-logs.ts | 92 +++----- src/servers/channels/discord/handler.ts | 2 +- src/servers/channels/telegram/handler.ts | 2 +- src/servers/channels/whatsapp/handler.ts | 2 +- src/servers/data-path.ts | 2 - .../queue-runner.ts => queue/init.ts} | 47 ++-- .../sidecar/{ => claude}/claude-manager.ts | 55 ++++- src/servers/sidecar/claude/index.ts | 119 ++++++++++ src/servers/sidecar/{ => claude}/proxy.ts | 2 +- src/servers/sidecar/{ => claude}/state.ts | 23 +- src/servers/sidecar/{ => email}/email-cron.ts | 26 ++- src/servers/sidecar/email/index.ts | 100 ++++++++ src/servers/sidecar/index.ts | 216 ------------------ src/servers/sidecar/pi/index.ts | 88 +++++++ src/servers/sidecar/{ => pi}/pi-manager.ts | 22 +- 22 files changed, 621 insertions(+), 545 deletions(-) delete mode 100755 scripts/setup-pty-sidecar.sh rename src/servers/{sidecar/queue-runner.ts => queue/init.ts} (80%) rename src/servers/sidecar/{ => claude}/claude-manager.ts (92%) create mode 100644 src/servers/sidecar/claude/index.ts rename src/servers/sidecar/{ => claude}/proxy.ts (97%) rename src/servers/sidecar/{ => claude}/state.ts (82%) rename src/servers/sidecar/{ => email}/email-cron.ts (78%) create mode 100644 src/servers/sidecar/email/index.ts delete mode 100644 src/servers/sidecar/index.ts create mode 100644 src/servers/sidecar/pi/index.ts rename src/servers/sidecar/{ => pi}/pi-manager.ts (95%) diff --git a/scripts/setup-pty-sidecar.sh b/scripts/setup-pty-sidecar.sh deleted file mode 100755 index cb0ac1b8..00000000 --- a/scripts/setup-pty-sidecar.sh +++ /dev/null @@ -1,70 +0,0 @@ -#!/bin/bash -# Officer PTY Sidecar — systemd service setup -# Creates and enables a systemd service for the terminal PTY sidecar. -# Usage: bash scripts/setup-pty-sidecar.sh - -set -euo pipefail - -GREEN='\033[0;32m' -YELLOW='\033[1;33m' -RED='\033[0;31m' -NC='\033[0m' - -ok() { echo -e " ${GREEN}✓${NC} $1"; } -warn() { echo -e " ${YELLOW}!${NC} $1"; } -fail() { echo -e " ${RED}✗${NC} $1"; } -skip() { echo -e " - $1 (already set up)"; } - -SERVICE_NAME="officer-pty-sidecar" -SERVICE_FILE="/etc/systemd/system/${SERVICE_NAME}.service" -SERVICE_USER="$(whoami)" -SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" -PROJECT_DIR="$(dirname "$SCRIPT_DIR")" - -# Use system node (not nvm) for systemd service -NODE_BIN="/usr/bin/node" - -if [ ! -f "$NODE_BIN" ]; then - fail "System Node.js not found at $NODE_BIN" - fail "Install Node.js 22 via NodeSource:" - fail " curl -fsSL https://deb.nodesource.com/setup_22.x | sudo -E bash -" - fail " sudo apt-get install -y nodejs" - exit 1 -fi - -echo "" -echo "── PTY Sidecar (systemd service) ──" -echo " Node: $NODE_BIN ($("$NODE_BIN" -v))" - -sudo tee "$SERVICE_FILE" > /dev/null << EOF -[Unit] -Description=Officer PTY Sidecar -After=network.target - -[Service] -Type=simple -User=$SERVICE_USER -WorkingDirectory=$PROJECT_DIR -Environment=TERMINAL_PTY_PORT=5338 -ExecStart=$NODE_BIN $PROJECT_DIR/src/servers/api/terminal/pty-sidecar.mjs -Restart=on-failure -RestartSec=3 - -[Install] -WantedBy=multi-user.target -EOF - -sudo systemctl daemon-reload -sudo systemctl enable "$SERVICE_NAME" - -if systemctl is-active --quiet "$SERVICE_NAME" 2>/dev/null; then - sudo systemctl restart "$SERVICE_NAME" - ok "$SERVICE_NAME service updated and restarted" -else - sudo systemctl start "$SERVICE_NAME" - if systemctl is-active --quiet "$SERVICE_NAME"; then - ok "$SERVICE_NAME service installed and running" - else - warn "$SERVICE_NAME service failed to start — check 'journalctl -u $SERVICE_NAME'" - fi -fi diff --git a/scripts/setup.sh b/scripts/setup.sh index 0820eac4..ec20a03d 100755 --- a/scripts/setup.sh +++ b/scripts/setup.sh @@ -869,9 +869,6 @@ else esac fi -# ─── 18. PTY sidecar (systemd service) ────────────────────────────────────── -bash "$SCRIPT_DIR/setup-pty-sidecar.sh" - # ─── verification ───────────────────────────────────────────────────────────── echo "" echo "═══════════════════════════════════════════" diff --git a/src/apps/officer-web/Screens/Dashboard/TaskLogs/index.tsx b/src/apps/officer-web/Screens/Dashboard/TaskLogs/index.tsx index 3522d323..5dd77ed7 100644 --- a/src/apps/officer-web/Screens/Dashboard/TaskLogs/index.tsx +++ b/src/apps/officer-web/Screens/Dashboard/TaskLogs/index.tsx @@ -5,16 +5,16 @@ import { Card } from '@/components/Card'; import { MessageBubble, type ChatMessage } from 'officerdev'; type LogMetadata = { - filename: string; + id: number; taskName: string; taskDirName: string; entryName: string; - entryType: 'file' | 'directory'; + entryType: string; provider: string; model: string; + isError: boolean; startedAt: string; completedAt: string | null; - isError: boolean; }; type FullLog = LogMetadata & { @@ -37,7 +37,7 @@ const ProviderBadge = ({ provider }: { provider: string }) => ( export const TaskLogs = () => { const client = useClient(); const [logs, setLogs] = useState([]); - const [selectedFilename, setSelectedFilename] = useState(null); + const [selectedId, setSelectedId] = useState(null); const [showDetail, setShowDetail] = useState(false); const [selectedLog, setSelectedLog] = useState(null); const [search, setSearch] = useState(''); @@ -54,15 +54,15 @@ export const TaskLogs = () => { }, []); useEffect(() => { - if (!selectedFilename) { + if (!selectedId) { setSelectedLog(null); return; } client - .get(`/task-logs/${selectedFilename}`) + .get(`/task-logs/${selectedId}`) .then(setSelectedLog) .catch(() => setSelectedLog(null)); - }, [selectedFilename]); + }, [selectedId]); const filtered = search ? logs.filter((l) => { @@ -77,101 +77,103 @@ export const TaskLogs = () => { return (
- {/* Left panel: list */} - -
-
- - setSearch(ev.target.value)} - className="w-full pl-8 pr-3 py-1.5 text-sm rounded-md border border-duck-dark/15 bg-background/60 text-duck-dark placeholder:text-duck-dark/30 focus:outline-none focus:ring-1 focus:ring-duck-teal/40" - /> -
+ {/* Left panel: list */} + +
+
+ + setSearch(ev.target.value)} + className="w-full pl-8 pr-3 py-1.5 text-sm rounded-md border border-duck-dark/15 bg-background/60 text-duck-dark placeholder:text-duck-dark/30 focus:outline-none focus:ring-1 focus:ring-duck-teal/40" + />
+
-
- {isLoading && ( -
Loading...
- )} - {!isLoading && filtered.length === 0 && ( -
No logs found
- )} - {filtered.map((log) => ( - - ))} -
- - - {/* Right panel: log viewer */} - - {!selectedLog && ( -
- Select a log to view - -
+
+ {isLoading && ( +
Loading...
)} - {selectedLog && ( - <> -
- -
-
- {selectedLog.taskName} - -
-
- {selectedLog.entryName} · {selectedLog.model} · {formatDate(selectedLog.startedAt)} - {selectedLog.completedAt && ` — ${formatDate(selectedLog.completedAt)}`} -
-
- {selectedLog.isError && ( - Error + {!isLoading && filtered.length === 0 && ( +
No logs found
+ )} + {filtered.map((log) => ( + + ))} +
+ + + {/* Right panel: log viewer */} + + {!selectedLog && ( +
+ Select a log to view + +
+ )} + {selectedLog && ( + <> +
+ +
+
+ {selectedLog.taskName} + +
+
+ {selectedLog.entryName} · {selectedLog.model} · {formatDate(selectedLog.startedAt)} + {selectedLog.completedAt && ` — ${formatDate(selectedLog.completedAt)}`} +
+
+ {selectedLog.isError && ( + + Error + + )} +
+
+ {selectedLog.messages.map((msg, i) => ( + + ))} +
+ + )} +
); }; diff --git a/src/server.tsx b/src/server.tsx index f45d6ae5..a1b4e381 100644 --- a/src/server.tsx +++ b/src/server.tsx @@ -56,6 +56,12 @@ const sidecarWebsocket = { return; } + // Handle queue commands from sidecars (e.g., email sidecar enqueuing jobs) + if (typeof msg.type === 'string' && msg.type.startsWith('queue:') && msg.id) { + handleSidecarQueueCommand(ws, msg); + return; + } + const id = sidecarConnections.get(ws); if (id) { handleSidecarMessage(id, msg); @@ -74,6 +80,39 @@ const sidecarWebsocket = { drain() {}, }; +// Handle queue commands from sidecars (e.g., email sidecar enqueuing jobs) +async function handleSidecarQueueCommand(ws: ServerWebSocket, msg: Record) { + const id = msg.id as string; + try { + switch (msg.type) { + case 'queue:enqueue': { + const job = await queueEnqueue(msg.params as import('./servers/queue/types').EnqueueParams); + ws.send(JSON.stringify({ type: 'queue:enqueued', id, job })); + break; + } + case 'queue:cancel': { + const job = await queueCancel(msg.jobId as string); + ws.send(JSON.stringify({ type: 'queue:cancelled', id, job })); + break; + } + case 'queue:list': { + const jobs = await queueList(); + ws.send(JSON.stringify({ type: 'queue:list', id, jobs })); + break; + } + case 'queue:get': { + const job = await queueGet(msg.jobId as string); + ws.send(JSON.stringify({ type: 'queue:get', id, job })); + break; + } + default: + ws.send(JSON.stringify({ type: 'queue:error', id, error: `Unknown queue command: ${msg.type}` })); + } + } catch (err) { + ws.send(JSON.stringify({ type: 'queue:error', id, error: err instanceof Error ? err.message : String(err) })); + } +} + const handlers: Record = { terminal: terminalWebsocket, pi: piWebsocket, @@ -274,7 +313,9 @@ try { console.error('[browser-relay] failed to start:', err instanceof Error ? err.message : err); } -// Sidecars connect to us via /api/sidecar/register — no init needed +// Initialize queue engine in API server process +import { initQueue, enqueueJob as queueEnqueue, cancelJob as queueCancel, listAllJobs as queueList, readJob as queueGet } from './servers/queue/init'; +initQueue().catch((err) => console.error('[queue] failed to initialize:', err)); // Ensure PulseAudio is running with virtual sink for cliamp audio streaming (async () => { diff --git a/src/servers/api/email/accounts.ts b/src/servers/api/email/accounts.ts index 8a40401f..d6f333c4 100644 --- a/src/servers/api/email/accounts.ts +++ b/src/servers/api/email/accounts.ts @@ -9,7 +9,7 @@ import { updateEmailAccountStatus, } from 'officerdb'; import { validateImapConnection } from './imap-validate'; -import * as sidecar from '../../sidecar-registry'; +import { enqueueJob, listAllJobs } from '../../queue/init'; type CreateAccountBody = { provider: string; @@ -44,7 +44,7 @@ accountsRouter.get('/', async (ctx) => { if (hasActiveAccounts) { try { - const jobs = await sidecar.listJobs(); + const jobs = await listAllJobs(); activeJobAccountIds = new Set( jobs .filter((j) => j.type === 'email-sync' && (j.status === 'queued' || j.status === 'running')) @@ -148,7 +148,7 @@ accountsRouter.post('/:id/sync', async (ctx) => { // Set status immediately so the UI reflects the queued state await updateEmailAccountStatus(id, 'queued'); - const job = await sidecar.enqueueJob({ + const job = await enqueueJob({ lane: 'email', type: 'email-sync', userId: user.email, diff --git a/src/servers/api/queue/queue.ts b/src/servers/api/queue/queue.ts index c0fa7759..6c0dffec 100644 --- a/src/servers/api/queue/queue.ts +++ b/src/servers/api/queue/queue.ts @@ -1,5 +1,5 @@ import { createRouter } from '../../create-router'; -import * as sidecar from '../../sidecar-registry'; +import { enqueueJob, cancelJob, listAllJobs, readJob } from '../../queue/init'; import { NOT_FOUND } from '../../custom-errors'; export const queueRouter = createRouter(); @@ -10,7 +10,7 @@ queueRouter.get('/jobs', async (ctx) => { const type = ctx.req.query('type'); const status = ctx.req.query('status'); - let jobs = await sidecar.listJobs(); + let jobs = await listAllJobs(); jobs = jobs.filter((j) => j.userId === user.email); if (lane) jobs = jobs.filter((j) => j.lane === lane); @@ -21,7 +21,7 @@ queueRouter.get('/jobs', async (ctx) => { }); queueRouter.get('/jobs/:id', async (ctx) => { - const job = await sidecar.getJob(ctx.req.param('id')); + const job = await readJob(ctx.req.param('id')); if (!job) throw NOT_FOUND('Job not found'); return ctx.json(job); }); @@ -36,12 +36,12 @@ queueRouter.post('/jobs', async (ctx) => { notify?: boolean; }; - const job = await sidecar.enqueueJob({ lane, type, userId: user.email, meta, notify }); + const job = await enqueueJob({ lane, type, userId: user.email, meta, notify }); return ctx.json(job, 201); }); queueRouter.delete('/jobs/:id', async (ctx) => { - const job = await sidecar.cancelJob(ctx.req.param('id')); + const job = await cancelJob(ctx.req.param('id')); if (!job) throw NOT_FOUND('Job not found'); return ctx.json(job); }); diff --git a/src/servers/api/task-logger.ts b/src/servers/api/task-logger.ts index 25eb2025..59164a4d 100644 --- a/src/servers/api/task-logger.ts +++ b/src/servers/api/task-logger.ts @@ -1,6 +1,4 @@ -import { mkdir } from 'node:fs/promises'; -import { join } from 'node:path'; -import { getTaskLogsDir } from '@@/data-path'; +import { db, schema } from 'officerdb'; import type { TaskInfo } from '@@/api/chat-types'; type ChatMessage = @@ -18,46 +16,41 @@ type ChatMessage = | { role: 'error'; text: string }; type TaskLog = { + userId: number; taskName: string; taskDirName: string; entryName: string; entryType: 'file' | 'directory'; provider: string; model: string; - startedAt: string; - completedAt: string | null; + startedAt: Date; messages: ChatMessage[]; }; type LogEntry = { - email: string; - filePath: string; + userId: number; log: TaskLog; }; const activeLogs = new Map(); let logCounter = 0; -export function createTaskLog(email: string, taskInfo: TaskInfo, provider: string, model: string): string { +export function createTaskLog(userId: number, taskInfo: TaskInfo, provider: string, model: string): string { const logId = `log_${Date.now()}_${++logCounter}`; - const dir = getTaskLogsDir(email); - const timestamp = new Date().toISOString().replace(/[:.]/g, '-'); - const filename = `${timestamp}-${taskInfo.taskDirName}.json`; - const filePath = join(dir, filename); const log: TaskLog = { + userId, taskName: taskInfo.taskName, taskDirName: taskInfo.taskDirName, entryName: taskInfo.entryName, entryType: taskInfo.entryType, provider, model, - startedAt: new Date().toISOString(), - completedAt: null, + startedAt: new Date(), messages: [], }; - activeLogs.set(logId, { email, filePath, log }); + activeLogs.set(logId, { userId, log }); return logId; } @@ -82,11 +75,24 @@ export async function finalizeLog(logId: string) { const entry = activeLogs.get(logId); if (!entry) return; - entry.log.completedAt = new Date().toISOString(); + const { log } = entry; + const lastMessage = log.messages[log.messages.length - 1]; + const isError = lastMessage?.role === 'result' ? lastMessage.isError : lastMessage?.role === 'error'; try { - await mkdir(join(entry.filePath, '..'), { recursive: true }); - await Bun.write(entry.filePath, JSON.stringify(entry.log, null, 2)); + await db.insert(schema.taskLogs).values({ + userId: log.userId, + taskName: log.taskName, + taskDirName: log.taskDirName, + entryName: log.entryName, + entryType: log.entryType, + provider: log.provider, + model: log.model, + isError: !!isError, + messages: log.messages, + startedAt: log.startedAt, + completedAt: new Date(), + }); } catch (err) { console.error('[task-logger] Failed to write log:', err); } diff --git a/src/servers/api/task-logs/task-logs.ts b/src/servers/api/task-logs/task-logs.ts index 8ff71f71..12dfc35e 100644 --- a/src/servers/api/task-logs/task-logs.ts +++ b/src/servers/api/task-logs/task-logs.ts @@ -1,79 +1,47 @@ -import { readdir } from 'node:fs/promises'; -import { join } from 'node:path'; +import { desc, eq } from 'drizzle-orm'; +import { db, schema } from 'officerdb'; import { createRouter } from '../../create-router'; -import { getTaskLogsDir } from '../../data-path'; - -type LogMetadata = { - filename: string; - taskName: string; - taskDirName: string; - entryName: string; - entryType: 'file' | 'directory'; - provider: string; - model: string; - startedAt: string; - completedAt: string | null; - isError: boolean; -}; export const taskLogsRouter = createRouter(); -// GET / — list all log files (metadata only, no messages) +// GET / — list all logs (metadata only, no messages) taskLogsRouter.get('/', async (ctx) => { - const email = ctx.get('user').email; - const dir = getTaskLogsDir(email); + const userId = ctx.get('user').id; - let files: string[]; - try { - files = (await readdir(dir)).filter((f) => f.endsWith('.json')); - } catch { - return ctx.json([]); - } - - // Sort by filename descending (newest first since filenames start with timestamp) - files.sort((a, b) => b.localeCompare(a)); - - const logs: LogMetadata[] = []; - for (const filename of files) { - try { - const raw = await Bun.file(join(dir, filename)).json(); - const lastMessage = Array.isArray(raw.messages) ? raw.messages[raw.messages.length - 1] : null; - const isError = lastMessage?.role === 'result' ? lastMessage.isError : lastMessage?.role === 'error'; - logs.push({ - filename, - taskName: raw.taskName ?? '', - taskDirName: raw.taskDirName ?? '', - entryName: raw.entryName ?? '', - entryType: raw.entryType ?? 'file', - provider: raw.provider ?? '', - model: raw.model ?? '', - startedAt: raw.startedAt ?? '', - completedAt: raw.completedAt ?? null, - isError: !!isError, - }); - } catch { - // Skip unreadable files - } - } + const logs = await db + .select({ + id: schema.taskLogs.id, + taskName: schema.taskLogs.taskName, + taskDirName: schema.taskLogs.taskDirName, + entryName: schema.taskLogs.entryName, + entryType: schema.taskLogs.entryType, + provider: schema.taskLogs.provider, + model: schema.taskLogs.model, + isError: schema.taskLogs.isError, + startedAt: schema.taskLogs.startedAt, + completedAt: schema.taskLogs.completedAt, + }) + .from(schema.taskLogs) + .where(eq(schema.taskLogs.userId, userId)) + .orderBy(desc(schema.taskLogs.startedAt)); return ctx.json(logs); }); -// GET /:filename — return full log file content -taskLogsRouter.get('/:filename', async (ctx) => { - const email = ctx.get('user').email; - const filename = ctx.req.param('filename'); +// GET /:id — return full log with messages +taskLogsRouter.get('/:id', async (ctx) => { + const userId = ctx.get('user').id; + const id = Number(ctx.req.param('id')); - if (!filename.endsWith('.json') || filename.includes('/') || filename.includes('..')) { - return ctx.text('Invalid filename', 400); + if (Number.isNaN(id)) { + return ctx.text('Invalid id', 400); } - const filePath = join(getTaskLogsDir(email), filename); + const [log] = await db.select().from(schema.taskLogs).where(eq(schema.taskLogs.id, id)); - try { - const data = await Bun.file(filePath).json(); - return ctx.json(data); - } catch { + if (!log || log.userId !== userId) { return ctx.text('Not found', 404); } + + return ctx.json(log); }); diff --git a/src/servers/channels/discord/handler.ts b/src/servers/channels/discord/handler.ts index ae89c2c2..34d5df99 100644 --- a/src/servers/channels/discord/handler.ts +++ b/src/servers/channels/discord/handler.ts @@ -4,7 +4,7 @@ import { sendAndAwait, getSessionModel, setSessionModel } from '../send-and-awai import { consumePairingCode } from '../pairing'; import { chunkMessage } from './chunker'; import { listPiModels } from '@@/api/pi/list-models'; -import { enqueueJob } from '../../sidecar-registry'; +import { enqueueJob } from '../../queue/init'; import { readJob } from '@@/queue/storage'; import { openEmailDb } from '@@/api/email/email-db'; import type { ModelInfo } from '@@/api/pi/types'; diff --git a/src/servers/channels/telegram/handler.ts b/src/servers/channels/telegram/handler.ts index 4f4ea612..5e5c675e 100644 --- a/src/servers/channels/telegram/handler.ts +++ b/src/servers/channels/telegram/handler.ts @@ -5,7 +5,7 @@ import { consumePairingCode } from '../pairing'; import { chunkMessage } from './chunker'; import { getTelegramBot } from './bot'; import { listPiModels } from '@@/api/pi/list-models'; -import { enqueueJob } from '../../sidecar-registry'; +import { enqueueJob } from '../../queue/init'; import { readJob } from '@@/queue/storage'; import { openEmailDb } from '@@/api/email/email-db'; import type { ModelInfo } from '@@/api/pi/types'; diff --git a/src/servers/channels/whatsapp/handler.ts b/src/servers/channels/whatsapp/handler.ts index 64eaf362..c4fc2756 100644 --- a/src/servers/channels/whatsapp/handler.ts +++ b/src/servers/channels/whatsapp/handler.ts @@ -4,7 +4,7 @@ import { sendAndAwait, getSessionModel, setSessionModel } from '../send-and-awai import { consumePairingCode } from '../pairing'; import { getWhatsAppClient } from './bot'; import { listPiModels } from '@@/api/pi/list-models'; -import { enqueueJob } from '../../sidecar-registry'; +import { enqueueJob } from '../../queue/init'; import { readJob } from '@@/queue/storage'; import { openEmailDb } from '@@/api/email/email-db'; import type { ModelInfo } from '@@/api/pi/types'; diff --git a/src/servers/data-path.ts b/src/servers/data-path.ts index a045972a..cd3ce796 100644 --- a/src/servers/data-path.ts +++ b/src/servers/data-path.ts @@ -69,8 +69,6 @@ export const getGlobalProcessesDir = () => join(DATA_PATH, 'processes'); export const getUserProcessesDir = (email: string) => join(DATA_PATH, email, 'processes'); -export const getTaskLogsDir = (email: string) => join(DATA_PATH, email, 'logs', 'tasks'); - export const getTmpAttachmentsDir = (email: string) => join(DATA_PATH, email, 'chat_sessions', 'tmp_attachments'); export const getAttachmentsDir = (email: string, provider: 'claude' | 'pi-mono', sessionId: string) => diff --git a/src/servers/sidecar/queue-runner.ts b/src/servers/queue/init.ts similarity index 80% rename from src/servers/sidecar/queue-runner.ts rename to src/servers/queue/init.ts index d2811387..7f569715 100644 --- a/src/servers/sidecar/queue-runner.ts +++ b/src/servers/queue/init.ts @@ -1,9 +1,11 @@ -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 { type Job, type EnqueueParams } from './types'; +import { readJob, writeJob, listAllJobs, ensureQueueDir } from './storage'; +import { getHandler } from './handler-registry'; // Import handlers to register them -import '../queue/handlers'; +import './handlers'; + +// ── Queue engine (moved from sidecar/queue-runner.ts) ── function formatDuration(ms: number): string { const s = Math.floor(ms / 1000); @@ -22,10 +24,10 @@ const PROGRESS_THROTTLE_MS = 1000; export async function initQueue() { await ensureQueueDir(); await resumeInterruptedJobs(); - console.log('[sidecar:queue] initialized'); + console.log('[queue] initialized'); } -export async function enqueue(params: EnqueueParams): Promise { +export async function enqueueJob(params: EnqueueParams): Promise { const handler = getHandler(params.type); if (!handler) throw new Error(`No handler registered for job type: ${params.type}`); @@ -43,7 +45,7 @@ export async function enqueue(params: EnqueueParams): Promise { }; await writeJob(job); - console.log(`[sidecar:queue] enqueued job ${job.id} (${job.type}) in lane ${job.lane}`); + console.log(`[queue] enqueued job ${job.id} (${job.type}) in lane ${job.lane}`); kickLane(job.lane); return job; } @@ -62,7 +64,7 @@ export async function cancelJob(id: string): Promise { } } await writeJob(job); - console.log(`[sidecar:queue] cancelled job ${job.id}`); + console.log(`[queue] cancelled job ${job.id}`); return job; } @@ -81,14 +83,13 @@ async function resumeInterruptedJobs() { } } await writeJob(job); - console.log(`[sidecar:queue] reset interrupted job ${job.id} back to queued`); + console.log(`[queue] reset interrupted job ${job.id} back to queued`); lanesToKick.add(job.lane); } else if (job.status === 'queued') { - // Clear retry delay on restart — no reason to wait after a sidecar restart if (job.retryAt) { job.retryAt = undefined; await writeJob(job); - console.log(`[sidecar:queue] cleared retry delay for job ${job.id}`); + console.log(`[queue] cleared retry delay for job ${job.id}`); } lanesToKick.add(job.lane); } @@ -124,7 +125,7 @@ async function processNextInLane(lane: string) { await runJob(next); } catch (err) { - console.error(`[sidecar:queue] lane ${lane} processing error:`, err); + console.error(`[queue] lane ${lane} processing error:`, err); } finally { const jobs = await listAllJobs(); const hasMore = jobs.some( @@ -139,6 +140,7 @@ async function processNextInLane(lane: string) { } async function runJob(job: Job) { + const { PermanentError } = await import('./types'); const handler = getHandler(job.type); if (!handler) { job.status = 'failed'; @@ -155,7 +157,7 @@ async function runJob(job: Job) { const isRetry = (job.retries ?? 0) > 0; const startTime = Date.now(); console.log( - `[sidecar:queue] ▶ ${isRetry ? 'resuming' : 'running'} job ${job.id} (${job.type})${isRetry ? ` retry ${job.retries}` : ''}`, + `[queue] ▶ ${isRetry ? 'resuming' : 'running'} job ${job.id} (${job.type})${isRetry ? ` retry ${job.retries}` : ''}`, ); const sharedMeta: Record = { ...(job.meta ?? {}) }; @@ -163,7 +165,7 @@ async function runJob(job: Job) { 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`); + console.log(`[queue] job ${job.id} was cancelled, stopping`); return; } @@ -178,9 +180,9 @@ async function runJob(job: Job) { await writeJob(fresh); let lastProgressWrite = 0; - let pendingProgress: JobProgress | null = null; + let pendingProgress: import('./types').JobProgress | null = null; - const updateProgress = async (progress: JobProgress) => { + const updateProgress = async (progress: import('./types').JobProgress) => { step.progress = progress; const now = Date.now(); if (now - lastProgressWrite >= PROGRESS_THROTTLE_MS) { @@ -192,7 +194,7 @@ async function runJob(job: Job) { } }; - const ctx: StepContext = { job: fresh, step, updateProgress, meta: sharedMeta }; + const ctx: import('./types').StepContext = { job: fresh, step, updateProgress, meta: sharedMeta }; try { await handlerStep.run(ctx); @@ -205,7 +207,7 @@ async function runJob(job: Job) { await writeJob(fresh); } catch (err) { const errorMessage = err instanceof Error ? err.message : String(err); - console.error(`[sidecar:queue] step "${step.name}" failed: ${errorMessage}`); + console.error(`[queue] step "${step.name}" failed: ${errorMessage}`); step.status = 'failed'; step.error = errorMessage; step.completedAt = Date.now(); @@ -226,7 +228,7 @@ async function runJob(job: Job) { 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`, + `[queue] job ${fresh.id} will retry (${retries}/${handler.retry.maxRetries}) in ${handler.retry.delayMs / 1000}s`, ); scheduleRetry(fresh.lane, handler.retry.delayMs); return; @@ -237,7 +239,10 @@ async function runJob(job: Job) { fresh.completedAt = Date.now(); fresh.meta = { ...fresh.meta, ...sharedMeta }; await writeJob(fresh); - console.error(`[sidecar:queue] ✗ job ${fresh.id} failed at step "${step.name}" in ${formatDuration(Date.now() - startTime)}:`, errorMessage); + console.error( + `[queue] ✗ job ${fresh.id} failed at step "${step.name}" in ${formatDuration(Date.now() - startTime)}:`, + errorMessage, + ); await notifyFailure(fresh); return; } @@ -249,7 +254,7 @@ async function runJob(job: Job) { final.completedAt = Date.now(); final.meta = { ...final.meta, ...sharedMeta }; await writeJob(final); - console.log(`[sidecar:queue] ✓ job ${final.id} completed in ${formatDuration(Date.now() - startTime)}`); + console.log(`[queue] ✓ job ${final.id} completed in ${formatDuration(Date.now() - startTime)}`); await notifyCompletion(final); } } diff --git a/src/servers/sidecar/claude-manager.ts b/src/servers/sidecar/claude/claude-manager.ts similarity index 92% rename from src/servers/sidecar/claude-manager.ts rename to src/servers/sidecar/claude/claude-manager.ts index e605b9e4..115e9ec0 100644 --- a/src/servers/sidecar/claude-manager.ts +++ b/src/servers/sidecar/claude/claude-manager.ts @@ -1,7 +1,7 @@ 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 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'; @@ -13,7 +13,13 @@ 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'; + return ( + raw + .replace(/@.*$/, '') + .replace(/[^a-zA-Z0-9._-]/g, '_') + .toLowerCase() + .slice(0, 32) || 'officer' + ); }; async function hasOwnCredentials(homeDir: string): Promise { @@ -33,7 +39,12 @@ const CLAUDE_BIN = (() => { // Active streaming processes const activeProcs = new Map(); -function buildAuthEnv(shellUsername: string, homeDir: string, isServiceUser: boolean, userHasCredentials: boolean): Record { +function buildAuthEnv( + shellUsername: string, + homeDir: string, + isServiceUser: boolean, + userHasCredentials: boolean, +): Record { 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() }; @@ -83,7 +94,11 @@ export async function spawnClaude(params: ClaudeSpawnParams): Promise { - try { proc.kill(); } catch { /* already dead */ } + try { + proc.kill(); + } catch { + /* already dead */ + } }, SEND_TIMEOUT_MS); try { @@ -143,10 +158,14 @@ export async function spawnClaudeStreaming( const workDir = cwd ?? homeDir; const claudeArgs = [ - CLAUDE_BIN, '-p', prompt, + CLAUDE_BIN, + '-p', + prompt, '--dangerously-skip-permissions', - '--output-format', 'stream-json', - '--verbose', '--include-partial-messages', + '--output-format', + 'stream-json', + '--verbose', + '--include-partial-messages', ]; const subModel = params.model?.split('/')[1]; @@ -170,7 +189,13 @@ export async function spawnClaudeStreaming( }; const proc = isServiceUser - ? Bun.spawn(claudeArgs, { cwd: workDir, stdin: 'ignore', stdout: 'pipe', stderr: 'pipe', env: { ...cleanEnv, ...env } }) + ? 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' }, @@ -179,7 +204,11 @@ export async function spawnClaudeStreaming( activeProcs.set(sessionKey, proc); const timeout = setTimeout(() => { - try { proc.kill(); } catch { /* already dead */ } + try { + proc.kill(); + } catch { + /* already dead */ + } onEvent({ type: 'error', message: 'Claude Code timed out after 5 minutes' }); }, SEND_TIMEOUT_MS); @@ -339,7 +368,11 @@ export async function spawnClaudeStreaming( export function killClaudeSession(sessionKey: string): boolean { const proc = activeProcs.get(sessionKey); if (proc) { - try { proc.kill(); } catch { /* already dead */ } + try { + proc.kill(); + } catch { + /* already dead */ + } activeProcs.delete(sessionKey); return true; } diff --git a/src/servers/sidecar/claude/index.ts b/src/servers/sidecar/claude/index.ts new file mode 100644 index 00000000..09b8f244 --- /dev/null +++ b/src/servers/sidecar/claude/index.ts @@ -0,0 +1,119 @@ +import type { SidecarCommand, SidecarEvent } from '../protocol'; +import { loadState, flushAndSave, acquireLock, releaseLock, getState } from './state'; +import { startAnthropicProxy, getProxySecret, ensureProxySecret } from './proxy'; +import * as claudeManager from './claude-manager'; +import { createSidecarConnector } from '../connect'; + +const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${process.env.PORT ?? '5000'}`; + +// ── Startup ── + +if (!acquireLock()) { + console.error('[claude] 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('[claude] failed to start Anthropic proxy:', err instanceof Error ? err.message : err); +} + +// ── 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 'state:sync': + reply({ + type: 'state:sync', + id: cmd.id, + state: { + proxySecret: getProxySecret(), + claudeSessions: { ...getState().claudeSessions }, + }, + }); + break; + + case 'proxy:secret': + reply({ type: 'proxy:secret', id: cmd.id, secret: getProxySecret() }); + break; + + case 'claude:spawn': { + try { + const result = await claudeManager.spawnClaude(cmd.params); + reply({ type: 'claude:result', id: cmd.id, result }); + } catch (err) { + reply({ type: 'claude:error', id: cmd.id, error: err instanceof Error ? err.message : String(err) }); + } + break; + } + + case 'claude:spawn-streaming': { + reply({ type: 'claude:spawned', id: cmd.id, sessionKey: cmd.params.sessionKey }); + + const onEvent = (event: import('../../api/pi/types').PiEvent) => { + connection.send({ type: 'claude:event', sessionKey: cmd.params.sessionKey, event }); + }; + + claudeManager.spawnClaudeStreaming(cmd.params, onEvent).catch((err) => { + connection.send({ + 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({ type: 'claude:killed', id: cmd.id }); + break; + + case 'claude:clear-session': + claudeManager.clearSession(cmd.sessionKey); + reply({ type: 'claude:session-cleared', id: cmd.id }); + break; + + default: + reply({ + type: 'error', + id: (cmd as SidecarCommand).id, + error: `Unknown command type: ${(cmd as Record).type}`, + }); + } +} + +// ── Connect to API server ── + +const connection = createSidecarConnector({ + apiUrl: `${API_URL}/api/sidecar/register`, + name: 'claude', + capabilities: ['claude', 'proxy'], + onCommand(cmd, reply) { + handleCommand(cmd as SidecarCommand, reply as ReplyFn); + }, +}); + +// ── Graceful shutdown ── + +async function shutdown(signal: string) { + console.log(`[claude] ${signal} received, saving state...`); + connection.destroy(); + await flushAndSave(); + releaseLock(); + process.exit(0); +} + +process.on('SIGTERM', () => shutdown('SIGTERM')); +process.on('SIGINT', () => shutdown('SIGINT')); diff --git a/src/servers/sidecar/proxy.ts b/src/servers/sidecar/claude/proxy.ts similarity index 97% rename from src/servers/sidecar/proxy.ts rename to src/servers/sidecar/claude/proxy.ts index 8b037d23..f265490b 100644 --- a/src/servers/sidecar/proxy.ts +++ b/src/servers/sidecar/claude/proxy.ts @@ -100,5 +100,5 @@ export function startAnthropicProxy() { }, }); - console.log(`[sidecar:proxy] listening on 127.0.0.1:${PROXY_PORT}`); + console.log(`[claude:proxy] listening on 127.0.0.1:${PROXY_PORT}`); } diff --git a/src/servers/sidecar/state.ts b/src/servers/sidecar/claude/state.ts similarity index 82% rename from src/servers/sidecar/state.ts rename to src/servers/sidecar/claude/state.ts index c8e2d712..eb76f6b3 100644 --- a/src/servers/sidecar/state.ts +++ b/src/servers/sidecar/claude/state.ts @@ -3,26 +3,17 @@ 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'); +const STATE_FILE = join(STATE_DIR, 'claude-state.json'); +const LOCK_FILE = join(STATE_DIR, 'claude.lock'); export type PersistedState = { proxySecret: string; claudeSessions: Record; // 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 }; @@ -37,13 +28,10 @@ function ensureDir() { 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; @@ -106,9 +94,8 @@ export function acquireLock(): boolean { 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 + return false; } - // Stale lock — remove it } require('node:fs').writeFileSync(LOCK_FILE, String(process.pid)); return true; @@ -135,7 +122,3 @@ function isProcessAlive(pid: number): boolean { return false; } } - -export function isPidAlive(pid: number): boolean { - return isProcessAlive(pid); -} diff --git a/src/servers/sidecar/email-cron.ts b/src/servers/sidecar/email/email-cron.ts similarity index 78% rename from src/servers/sidecar/email-cron.ts rename to src/servers/sidecar/email/email-cron.ts index 1e6dd86f..2d17ddb8 100644 --- a/src/servers/sidecar/email-cron.ts +++ b/src/servers/sidecar/email/email-cron.ts @@ -1,16 +1,20 @@ +import type { Job, EnqueueParams } from '../../queue/types'; import { getAllSyncedAccounts, getUserById, getUserIntegration } from 'officerdb'; -import * as queueRunner from './queue-runner'; const INTERVAL_MS = 10 * 60 * 1000; // 10 minutes let timer: ReturnType | null = null; +let enqueueFn: ((params: EnqueueParams) => Promise) | null = null; +let listJobsFn: (() => Promise) | null = null; async function tick() { + if (!enqueueFn || !listJobsFn) return; + try { const accounts = await getAllSyncedAccounts(); if (accounts.length === 0) return; - const allJobs = await queueRunner.listAllJobs(); + const allJobs = await listJobsFn(); const activeEmailSyncIds = new Set( allJobs .filter((j) => j.type === 'email-sync' && (j.status === 'queued' || j.status === 'running')) @@ -40,7 +44,7 @@ async function tick() { } try { - await queueRunner.enqueue({ + await enqueueFn({ lane: 'email', type: 'email-sync', userId: user.email, @@ -63,7 +67,10 @@ async function tick() { }); console.log(`[email-cron] Enqueued incremental sync for ${account.email}`); } catch (err) { - console.error(`[email-cron] Failed to enqueue sync for ${account.email}:`, err instanceof Error ? err.message : err); + console.error( + `[email-cron] Failed to enqueue sync for ${account.email}:`, + err instanceof Error ? err.message : err, + ); } } } catch (err) { @@ -71,11 +78,18 @@ async function tick() { } } -export function initEmailCron() { +type EmailCronDeps = { + enqueue: (params: EnqueueParams) => Promise; + listJobs: () => Promise; +}; + +export function initEmailCron(deps: EmailCronDeps) { if (timer) return; + enqueueFn = deps.enqueue; + listJobsFn = deps.listJobs; console.log(`[email-cron] Starting email sync cron (every ${INTERVAL_MS / 60_000} min)`); timer = setInterval(tick, INTERVAL_MS); - // Run first tick after a short delay to let the queue initialize + // Run first tick after a short delay setTimeout(tick, 30_000); } diff --git a/src/servers/sidecar/email/index.ts b/src/servers/sidecar/email/index.ts new file mode 100644 index 00000000..ecde2cb3 --- /dev/null +++ b/src/servers/sidecar/email/index.ts @@ -0,0 +1,100 @@ +import type { SidecarEvent } from '../protocol'; +import type { Job, EnqueueParams } from '../../queue/types'; +import { initEmailCron, stopEmailCron } from './email-cron'; +import { createSidecarConnector } from '../connect'; + +const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${process.env.PORT ?? '5000'}`; + +// ── Queue access via WS ── + +let reqCounter = 0; +const pendingQueue = new Map void; reject: (e: Error) => void; timer: Timer }>(); + +function nextQueueId(): string { + return `eq_${Date.now()}_${++reqCounter}`; +} + +function sendQueueCommand(cmd: Record): Promise { + return new Promise((resolve, reject) => { + const id = cmd.id as string; + const timer = setTimeout(() => { + pendingQueue.delete(id); + reject(new Error(`Queue command ${cmd.type} timed out`)); + }, 30_000); + pendingQueue.set(id, { resolve, reject, timer }); + connection.send(cmd as SidecarEvent); + }); +} + +async function enqueueViaWs(params: EnqueueParams): Promise { + const res = (await sendQueueCommand({ type: 'queue:enqueue', id: nextQueueId(), params })) as Record; + if (res.type === 'queue:enqueued') return res.job as Job; + if (res.type === 'queue:error') throw new Error(res.error as string); + throw new Error('Unexpected response'); +} + +async function listJobsViaWs(): Promise { + const res = (await sendQueueCommand({ type: 'queue:list', id: nextQueueId() })) as Record; + if (res.type === 'queue:list') return res.jobs as Job[]; + throw new Error('Unexpected response'); +} + +// ── Command handlers ── + +type ReplyFn = (msg: SidecarEvent) => void; + +function handleCommand(cmd: Record, reply: ReplyFn) { + switch (cmd.type) { + case 'ping': + reply({ type: 'pong', id: cmd.id as string }); + break; + + default: + // Check if this is a queue response (from API server responding to our queue commands) + if ( + typeof cmd.type === 'string' && + cmd.type.startsWith('queue:') && + cmd.id && + pendingQueue.has(cmd.id as string) + ) { + const pending = pendingQueue.get(cmd.id as string)!; + pendingQueue.delete(cmd.id as string); + clearTimeout(pending.timer); + pending.resolve(cmd); + return; + } + + reply({ + type: 'error', + id: cmd.id as string, + error: `Unknown command type: ${cmd.type}`, + }); + } +} + +// ── Connect to API server ── + +const connection = createSidecarConnector({ + apiUrl: `${API_URL}/api/sidecar/register`, + name: 'email', + capabilities: ['email'], + onCommand(cmd, reply) { + handleCommand(cmd as Record, reply as ReplyFn); + }, + onConnected() { + // Start email cron once connected (so queue commands can reach API server) + // initEmailCron({ enqueue: enqueueViaWs, listJobs: listJobsViaWs }); // TODO: re-enable after testing + }, +}); + +// ── Graceful shutdown ── + +function shutdown(signal: string) { + console.log(`[email] ${signal} received, shutting down...`); + stopEmailCron(); + connection.destroy(); + process.exit(0); +} + +process.on('SIGTERM', () => shutdown('SIGTERM')); +process.on('SIGINT', () => shutdown('SIGINT')); diff --git a/src/servers/sidecar/index.ts b/src/servers/sidecar/index.ts deleted file mode 100644 index c1554096..00000000 --- a/src/servers/sidecar/index.ts +++ /dev/null @@ -1,216 +0,0 @@ -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'; -import { initEmailCron, stopEmailCron } from './email-cron'; -import { createSidecarConnector } from './connect'; - -const API_URL = process.env.API_URL ?? `ws://127.0.0.1:${process.env.PORT ?? '5000'}`; -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); -}); - -// Start email sync cron -// initEmailCron(); // TODO: re-enable after initial sync testing - -// ── State ── - -function buildState(): SidecarState { - return { - proxySecret: getProxySecret(), - claudeSessions: { ...getState().claudeSessions }, - piSessions: piManager.getAllSessions(), - uptime: Date.now() - startedAt, - }; -} - -// ── 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 'state:sync': - reply({ type: 'state:sync', id: cmd.id, state: buildState() }); - break; - - case 'proxy:secret': - reply({ type: 'proxy:secret', id: cmd.id, secret: getProxySecret() }); - break; - - // ── Claude Code ── - - case 'claude:spawn': { - try { - const result = await claudeManager.spawnClaude(cmd.params); - reply({ type: 'claude:result', id: cmd.id, result }); - } catch (err) { - reply({ type: 'claude:error', id: cmd.id, error: err instanceof Error ? err.message : String(err) }); - } - break; - } - - case 'claude:spawn-streaming': { - reply({ type: 'claude:spawned', id: cmd.id, sessionKey: cmd.params.sessionKey }); - - const onEvent = (event: import('../api/pi/types').PiEvent) => { - connection.send({ type: 'claude:event', sessionKey: cmd.params.sessionKey, event }); - }; - - claudeManager.spawnClaudeStreaming(cmd.params, onEvent).catch((err) => { - connection.send({ - 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({ type: 'claude:killed', id: cmd.id }); - break; - - case 'claude:clear-session': - claudeManager.clearSession(cmd.sessionKey); - reply({ type: 'claude:session-cleared', id: cmd.id }); - break; - - // ── Pi ── - - 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; - - // ── Queue ── - - case 'queue:enqueue': { - try { - const job = await queueRunner.enqueue(cmd.params); - reply({ type: 'queue:enqueued', id: cmd.id, job }); - } catch (err) { - reply({ 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({ type: 'queue:cancelled', id: cmd.id, job }); - } catch (err) { - reply({ type: 'queue:error', id: cmd.id, error: err instanceof Error ? err.message : String(err) }); - } - break; - } - - case 'queue:list': { - const jobs = await queueRunner.listAllJobs(); - reply({ type: 'queue:list', id: cmd.id, jobs }); - break; - } - - case 'queue:get': { - const job = await queueRunner.readJob(cmd.jobId); - reply({ type: 'queue:get', id: cmd.id, job }); - break; - } - - default: - reply({ - type: 'error', - id: (cmd as SidecarCommand).id, - error: `Unknown command type: ${(cmd as Record).type}`, - }); - } -} - -// ── Connect to API server ── - -const connection = createSidecarConnector({ - apiUrl: `${API_URL}/api/sidecar/register`, - name: 'process', - capabilities: ['claude', 'pi', 'queue', 'proxy'], - onCommand(cmd, reply) { - handleCommand(cmd as SidecarCommand, reply as ReplyFn); - }, -}); - -// ── Graceful shutdown ── - -async function shutdown(signal: string) { - console.log(`[sidecar] ${signal} received, saving state...`); - stopEmailCron(); - connection.destroy(); - await flushAndSave(); - releaseLock(); - process.exit(0); -} - -process.on('SIGTERM', () => shutdown('SIGTERM')); -process.on('SIGINT', () => shutdown('SIGINT')); diff --git a/src/servers/sidecar/pi/index.ts b/src/servers/sidecar/pi/index.ts new file mode 100644 index 00000000..2ce2383c --- /dev/null +++ b/src/servers/sidecar/pi/index.ts @@ -0,0 +1,88 @@ +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).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')); diff --git a/src/servers/sidecar/pi-manager.ts b/src/servers/sidecar/pi/pi-manager.ts similarity index 95% rename from src/servers/sidecar/pi-manager.ts rename to src/servers/sidecar/pi/pi-manager.ts index 206a6cec..ff146ac1 100644 --- a/src/servers/sidecar/pi-manager.ts +++ b/src/servers/sidecar/pi/pi-manager.ts @@ -1,9 +1,8 @@ import { join } 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 { isPidAlive } from './state'; +import type { PiEvent, MessageCost } from '../../api/pi/types'; +import type { PiSpawnParams, PiSessionInfo } from '../protocol'; const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data'); const PI_CONFIG_DIR = join(require('node:os').homedir(), '.pi', 'agent'); @@ -27,6 +26,15 @@ const toShellUsername = (username: string, email: string): string => { ); }; +function isPidAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + // Resolve pi as [node, cli.js] const PI_CMD = (() => { const whichResult = Bun.spawnSync({ cmd: ['which', 'pi'], stdout: 'pipe', stderr: 'ignore' }); @@ -216,7 +224,7 @@ function writeRpcCommand(proc: Subprocess, command: Record): vo writer.write(JSON.stringify(command) + '\n'); writer.flush(); } catch (err) { - console.error('[sidecar:pi] writeRpcCommand error:', err); + console.error('[pi] writeRpcCommand error:', err); } } @@ -288,7 +296,7 @@ export async function spawnPi(options: PiSpawnOptions): Promise { 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})`); + console.log(`[pi] spawned Pi for session ${sessionId} (model=${model}, pid=${proc.pid})`); // Read stdout JSON event stream const stdout = proc.stdout as ReadableStream; @@ -338,7 +346,7 @@ export async function spawnPi(options: PiSpawnOptions): Promise { 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()}`); + if (text.trim()) console.log(`[pi:stderr] ${text.trim()}`); } } catch { /* process ended */ @@ -349,7 +357,7 @@ export async function spawnPi(options: PiSpawnOptions): Promise { proc.exited.then((code) => { sessions.delete(sessionId); if (code !== 0) { - console.error(`[sidecar:pi] Pi process ${sessionId} exited with code ${code}`); + console.error(`[pi] Pi process ${sessionId} exited with code ${code}`); } }); }