This commit is contained in:
2026-02-19 18:05:50 +00:00
parent 9870fa7ae8
commit 6a83342013
38 changed files with 1459 additions and 267 deletions
+54
View File
@@ -0,0 +1,54 @@
import { Hono } from 'hono';
import type { HonoVariables } from '@@/create-router';
export const piMonoModelsRouter = new Hono<{ Variables: HonoVariables }>();
// Hardcoded fallback models — pi supports many providers but these are the most common
const FALLBACK_MODELS = [
{ id: 'claude-sonnet-4-5-20250514', name: 'Claude Sonnet 4.5', provider: 'anthropic', providerId: 'anthropic' },
{ id: 'claude-opus-4-20250918', name: 'Claude Opus 4', provider: 'anthropic', providerId: 'anthropic' },
{ id: 'gpt-4.1', name: 'GPT-4.1', provider: 'openai', providerId: 'openai' },
{ id: 'o3', name: 'o3', provider: 'openai', providerId: 'openai' },
{ id: 'gemini-2.5-pro', name: 'Gemini 2.5 Pro', provider: 'google', providerId: 'google' },
{ id: 'gemini-2.5-flash', name: 'Gemini 2.5 Flash', provider: 'google', providerId: 'google' },
];
piMonoModelsRouter.get('/pi-mono/models', async (ctx) => {
// Spawn a short-lived pi process to query available models
try {
const proc = Bun.spawn(['pi', '--list-models', '--mode', 'json'], {
stdout: 'pipe',
stderr: 'pipe',
env: { ...process.env },
});
const output = await new Response(proc.stdout).text();
await proc.exited;
if (proc.exitCode !== 0) return ctx.json(FALLBACK_MODELS);
// Parse the output — pi --list-models outputs model info
const lines = output.trim().split('\n').filter(Boolean);
const models: { id: string; name: string; provider: string; providerId: string }[] = [];
for (const line of lines) {
try {
const data = JSON.parse(line);
if (data.id && data.provider) {
models.push({
id: data.id,
name: data.name ?? data.id,
provider: data.provider,
providerId: data.provider,
});
}
} catch {
// skip non-JSON lines
}
}
return ctx.json(models.length > 0 ? models : FALLBACK_MODELS);
} catch {
return ctx.json(FALLBACK_MODELS);
}
});
+448
View File
@@ -0,0 +1,448 @@
import type { ServerWebSocket } from 'bun';
import type { Subprocess } from 'bun';
import { mkdir, rename } from 'node:fs/promises';
import { join } from 'node:path';
import { homedir } from 'node:os';
import {
getPiMonoSessionDir,
getTmpAttachmentsDir,
getAttachmentsDir,
getHomeDir,
getNativeSkillsDir,
getGlobalSkillsDir,
getUserSkillsDir,
} from '@@/data-path';
import { readSkillDirs, parseFrontmatter } from '@@/api/skills/skills';
import type { ClientMessage, ServerMessage, ImageData, TaskInfo } from '@@/api/chat-types';
import { createTaskLog, appendToLog, finalizeLog } from '@@/api/task-logger';
type WSData = { userId: number; email: string };
type ConnectionState = {
piProcess: Subprocess | null;
sessionId: string | null;
pendingTitle: string | null;
selectedModel: string | null;
pendingAttachmentIds: string[];
cwd: string | null;
resourceChatDir: string | null;
logId: string | null;
fullText: string;
rpcReady: boolean;
};
const connections = new Map<ServerWebSocket<WSData>, ConnectionState>();
function send(ws: ServerWebSocket<WSData>, msg: ServerMessage) {
if (ws.readyState === 1) ws.send(JSON.stringify(msg));
}
function resolveRootDir(email: string, root?: string): string {
if (!root || root === 'home') return getHomeDir(email);
if (root === '~') return homedir();
if (root === 'officer.dev') return join(process.cwd(), '..');
return getHomeDir(email);
}
async function buildSkillsPrompt(email: string): Promise<string> {
const nativeSkills = await readSkillDirs(getNativeSkillsDir());
const globalSkills = await readSkillDirs(getGlobalSkillsDir());
const userSkills = await readSkillDirs(getUserSkillsDir(email));
const merged = new Map(nativeSkills);
for (const [name, path] of globalSkills) merged.set(name, path);
for (const [name, path] of userSkills) merged.set(name, path);
if (merged.size === 0) return '';
const lines = await Promise.all(
Array.from(merged.entries()).map(async ([dirName, filePath]) => {
const raw = await Bun.file(filePath).text();
const { frontmatter } = parseFrontmatter(raw);
const name = frontmatter.name || dirName;
return `- ${name}: ${frontmatter.description} (read ${filePath} for full instructions)`;
}),
);
return `\n\nYou have access to the following skills. When a user's request matches a skill, read its SKILL.md file for detailed instructions before proceeding.\n\nAvailable skills:\n${lines.join('\n')}`;
}
function writeRpcCommand(proc: Subprocess, command: Record<string, unknown>) {
const writer = proc.stdin as WritableStream;
const textEncoder = new TextEncoder();
const w = writer.getWriter();
w.write(textEncoder.encode(JSON.stringify(command) + '\n'));
w.releaseLock();
}
function spawnPiProcess(ws: ServerWebSocket<WSData>, state: ConnectionState, workingDir: string) {
const args = ['pi', '--mode', 'rpc', '--no-extensions', '--no-skills', '--no-prompt-templates', '--no-themes'];
if (state.selectedModel) {
args.push('--model', state.selectedModel);
}
const proc = Bun.spawn(args, {
cwd: workingDir,
stdin: 'pipe',
stdout: 'pipe',
stderr: 'pipe',
env: { ...process.env },
});
state.piProcess = proc;
// Read stdout line-by-line for JSON events
const reader = proc.stdout.getReader();
const decoder = new TextDecoder();
let buffer = '';
const readLoop = 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);
handlePiEvent(ws, state, event);
} catch {
// skip unparseable lines
}
}
}
} catch {
// process ended
}
};
readLoop();
// Read stderr for debugging
const stderrReader = proc.stderr.getReader();
const stderrDecoder = new TextDecoder();
const readStderr = async () => {
try {
while (true) {
const { done, value } = await stderrReader.read();
if (done) break;
const text = stderrDecoder.decode(value, { stream: true });
if (text.trim()) console.log('[pi-mono-ws] stderr:', text.trim());
}
} catch {
// process ended
}
};
readStderr();
// Handle process exit
proc.exited.then((code) => {
console.log(`[pi-mono-ws] pi process exited with code ${code}`);
if (state.piProcess === proc) {
state.piProcess = null;
}
});
}
function handlePiEvent(ws: ServerWebSocket<WSData>, state: ConnectionState, event: Record<string, unknown>) {
const type = event.type as string;
// RPC responses (type === 'response')
if (type === 'response') {
const command = event.command as string;
if (command === 'get_available_models' && event.success) {
// Models are handled by the REST endpoint, not here
}
if (command === 'prompt' && !event.success) {
send(ws, { type: 'error', message: (event.error as string) ?? 'Prompt failed' });
}
return;
}
switch (type) {
case 'agent_start':
state.fullText = '';
break;
case 'message_update': {
// message_update contains assistantMessageEvent with content deltas
const ame = event.assistantMessageEvent as Record<string, unknown> | undefined;
if (!ame) break;
const ameType = ame.type as string;
if (ameType === 'text_delta') {
const delta = ame.delta as string;
state.fullText += delta;
send(ws, { type: 'assistant:partial', text: delta });
}
break;
}
case 'message_end': {
// Full assistant message complete
if (state.fullText) {
send(ws, { type: 'assistant:text', text: state.fullText });
if (state.logId) appendToLog(state.logId, { role: 'assistant', text: state.fullText });
state.fullText = '';
}
break;
}
case 'tool_execution_start': {
const toolCallId = (event.toolCallId as string) ?? '';
const toolName = (event.toolName as string) ?? 'unknown';
const args = (event.args as Record<string, unknown>) ?? {};
// Commit any streaming text before tool use
if (state.fullText) {
send(ws, { type: 'assistant:text', text: state.fullText });
if (state.logId) appendToLog(state.logId, { role: 'assistant', text: state.fullText });
state.fullText = '';
}
send(ws, { type: 'tool:use', toolName, toolInput: args, toolUseId: toolCallId });
if (state.logId) appendToLog(state.logId, { role: 'tool', toolName, toolInput: args, toolUseId: toolCallId });
break;
}
case 'tool_execution_end': {
const toolCallId = (event.toolCallId as string) ?? '';
const result = event.result;
const isError = (event.isError as boolean) ?? false;
const output = result != null ? (typeof result === 'string' ? result : JSON.stringify(result)) : '';
send(ws, { type: 'tool:result', toolUseId: toolCallId, output, isError });
if (state.logId)
appendToLog(state.logId, {
role: 'tool',
toolName: '',
toolInput: {},
toolUseId: toolCallId,
output,
isError,
});
break;
}
case 'agent_end': {
// Commit any remaining streaming text
if (state.fullText) {
send(ws, { type: 'assistant:text', text: state.fullText });
if (state.logId) appendToLog(state.logId, { role: 'assistant', text: state.fullText });
state.fullText = '';
}
send(ws, { type: 'result', costUsd: 0, durationMs: 0, numTurns: 0, isError: false });
if (state.logId) {
appendToLog(state.logId, { role: 'result', costUsd: 0, durationMs: 0, numTurns: 0, isError: false });
finalizeLog(state.logId);
state.logId = null;
}
break;
}
case 'extension_ui_request': {
// Auto-cancel extension UI requests since we don't support them
if (state.piProcess && event.id) {
writeRpcCommand(state.piProcess, { type: 'extension_ui_response', id: event.id, cancelled: true });
}
break;
}
}
}
type HandleChatParams = {
ws: ServerWebSocket<WSData>;
prompt: string;
sessionId?: string;
model?: string;
cwd?: { root?: string; path: string };
attachmentIds?: string[];
images?: ImageData[];
resourceChatDir?: string;
taskInfo?: TaskInfo;
};
async function handleChat({
ws,
prompt,
sessionId,
model,
cwd,
attachmentIds,
images,
resourceChatDir,
taskInfo,
}: HandleChatParams) {
const state = connections.get(ws);
if (!state) return;
if (taskInfo && !state.logId) {
state.logId = createTaskLog(ws.data.email, taskInfo, 'pi-mono', model ?? 'unknown');
appendToLog(state.logId, { role: 'user', text: prompt });
}
if (resourceChatDir) state.resourceChatDir = resourceChatDir;
if (model) state.selectedModel = model;
if (!sessionId && !state.sessionId) {
// New session — generate our own sessionId for officer tracking
const newSessionId = crypto.randomUUID();
state.sessionId = newSessionId;
state.pendingTitle = prompt.slice(0, 100);
if (attachmentIds?.length) state.pendingAttachmentIds = attachmentIds;
send(ws, { type: 'session:init', sessionId: newSessionId, model: model ?? 'pi-mono' });
if (state.resourceChatDir) {
const chatDir = join(state.resourceChatDir, 'chat');
const meta = { id: newSessionId, model: model ?? 'pi-mono' };
mkdir(chatDir, { recursive: true })
.then(() => Bun.write(join(chatDir, 'meta.json'), JSON.stringify(meta)))
.catch(() => {});
} else {
const dir = getPiMonoSessionDir(ws.data.email, newSessionId);
const meta = {
id: newSessionId,
title: state.pendingTitle ?? 'New chat',
createdAt: Date.now(),
model: model ?? 'pi-mono',
};
mkdir(dir, { recursive: true })
.then(() => Bun.write(join(dir, 'meta.json'), JSON.stringify(meta)))
.catch(() => {});
// Move tmp attachments to session dir
if (state.pendingAttachmentIds.length > 0) {
const tmpDir = getTmpAttachmentsDir(ws.data.email);
const destDir = getAttachmentsDir(ws.data.email, 'pi-mono', newSessionId);
mkdir(destDir, { recursive: true })
.then(() =>
Promise.all(
state.pendingAttachmentIds.map((id) => rename(join(tmpDir, id), join(destDir, id)).catch(() => {})),
),
)
.catch(() => {});
state.pendingAttachmentIds = [];
}
}
state.pendingTitle = null;
} else if (sessionId && !state.sessionId) {
state.sessionId = sessionId;
}
// Ensure pi process is running
const homeDir = getHomeDir(ws.data.email);
if (cwd) state.cwd = join(resolveRootDir(ws.data.email, cwd.root), cwd.path);
const workingDir = state.cwd ?? homeDir;
if (!state.piProcess) {
spawnPiProcess(ws, state, workingDir);
// Give pi a moment to initialize
await new Promise((r) => setTimeout(r, 500));
}
if (!state.piProcess) {
send(ws, { type: 'error', message: 'Failed to start pi process' });
return;
}
// Set model if specified
if (model && model !== state.selectedModel) {
state.selectedModel = model;
// Model is set via CLI args on spawn, would need a new process to change
}
// Build context and send prompt
const skillsAppend = await buildSkillsPrompt(ws.data.email);
const contextAppend = `\n\nThe user's home directory is: ${homeDir}` + skillsAppend;
send(ws, { type: 'system:prompt', text: contextAppend });
const fullPrompt = `<system>${contextAppend}</system>\n\n${prompt}`;
const rpcCommand: Record<string, unknown> = {
type: 'prompt',
id: `req_${Date.now()}`,
message: fullPrompt,
};
writeRpcCommand(state.piProcess, rpcCommand);
}
function handleStop(ws: ServerWebSocket<WSData>) {
const state = connections.get(ws);
if (!state?.piProcess) return;
writeRpcCommand(state.piProcess, { type: 'abort', id: `abort_${Date.now()}` });
send(ws, { type: 'stopped' });
}
function killPiProcess(state: ConnectionState) {
if (state.piProcess) {
try {
state.piProcess.kill();
} catch {
// already dead
}
state.piProcess = null;
}
}
export const piMonoWebsocket = {
open(ws: ServerWebSocket<WSData>) {
connections.set(ws, {
piProcess: null,
sessionId: null,
pendingTitle: null,
selectedModel: null,
pendingAttachmentIds: [],
cwd: null,
resourceChatDir: null,
logId: null,
fullText: '',
rpcReady: false,
});
},
message(ws: ServerWebSocket<WSData>, raw: string | Buffer) {
let msg: ClientMessage;
try {
msg = JSON.parse(typeof raw === 'string' ? raw : raw.toString()) as ClientMessage;
} catch {
send(ws, { type: 'error', message: 'Invalid JSON' });
return;
}
if (msg.type === 'chat') {
handleChat({
ws,
prompt: msg.prompt,
sessionId: msg.sessionId,
model: typeof msg.model === 'string' ? msg.model : undefined,
cwd: msg.cwd,
attachmentIds: msg.attachmentIds,
images: msg.images,
resourceChatDir: msg.resourceChatDir,
taskInfo: msg.taskInfo,
});
} else if (msg.type === 'stop') {
handleStop(ws);
}
},
close(ws: ServerWebSocket<WSData>) {
const state = connections.get(ws);
if (state) {
killPiProcess(state);
}
connections.delete(ws);
},
drain() {},
};
+1 -1
View File
@@ -33,7 +33,7 @@ scrapeRouter.post('/', async (ctx) => {
const { url, sessionId, provider } = ctx.get('body') as {
url: string;
sessionId?: string;
provider?: 'claude' | 'opencode';
provider?: 'claude' | 'opencode' | 'pi-mono';
};
if (!url) return ctx.json({ error: 'url is required' }, 400);
@@ -0,0 +1,54 @@
import { createRouter } from '../../create-router';
export const piMonoRouter = createRouter();
const GLOBAL_DIRS = ['/usr/local/bin', '/usr/bin'];
const getPaths = async () => {
try {
const proc = Bun.spawn(['which', '-a', 'pi'], { stdout: 'pipe', stderr: 'pipe' });
const output = await new Response(proc.stdout).text();
await proc.exited;
if (proc.exitCode !== 0) return { path: null, globalPath: null };
const paths = [...new Set(output.trim().split('\n'))];
const path = paths[0] ?? null;
const globalPath = paths.find((p) => GLOBAL_DIRS.some((dir) => p.startsWith(dir))) ?? null;
return { path, globalPath };
} catch {
return { path: null, globalPath: null };
}
};
piMonoRouter.get('/version', async (ctx) => {
try {
const proc = Bun.spawn(['pi', '--version'], { stdout: 'pipe', stderr: 'pipe' });
const output = await new Response(proc.stdout).text();
await proc.exited;
if (proc.exitCode !== 0) return ctx.json({ version: null, path: null, globalPath: null });
const { path, globalPath } = await getPaths();
return ctx.json({ version: output.trim(), path, globalPath });
} catch {
return ctx.json({ version: null, path: null, globalPath: null });
}
});
piMonoRouter.post('/install', async (ctx) => {
try {
const proc = Bun.spawn(['npm', 'install', '-g', '@mariozechner/pi-coding-agent'], {
stdout: 'pipe',
stderr: 'pipe',
});
await proc.exited;
if (proc.exitCode !== 0) {
const stderr = await new Response(proc.stderr).text();
return ctx.json({ version: null, path: null, globalPath: null, error: stderr.trim() }, 500);
}
const versionProc = Bun.spawn(['pi', '--version'], { stdout: 'pipe', stderr: 'pipe' });
const output = await new Response(versionProc.stdout).text();
await versionProc.exited;
const { path, globalPath } = await getPaths();
return ctx.json({ version: output.trim(), path, globalPath });
} catch {
return ctx.json({ version: null, path: null, globalPath: null, error: 'Installation failed' }, 500);
}
});
@@ -6,6 +6,7 @@ import { join } from 'node:path';
import { officerdb, count, Users } from 'officerdb';
import { claudeCodeRouter } from './claude-code';
import { opencodeRouter } from './opencode';
import { piMonoRouter } from './pi-mono';
import { applicationsRouter } from './applications';
import { resourcesRouter } from './resources';
@@ -22,6 +23,7 @@ export const serverSettingsRouter = createRouter();
serverSettingsRouter.route('/claude-code', claudeCodeRouter);
serverSettingsRouter.route('/opencode', opencodeRouter);
serverSettingsRouter.route('/pi-mono', piMonoRouter);
serverSettingsRouter.route('/applications', applicationsRouter);
serverSettingsRouter.route('/resources', resourcesRouter);
+59 -7
View File
@@ -1,7 +1,7 @@
import { Hono } from 'hono';
import { mkdir, readdir, rename, rm } from 'node:fs/promises';
import { join } from 'node:path';
import { getClaudeDir, getSessionDir, getArchivedSessionDir, getOpencodeDir, getOpencodeSessionDir } from '@@/data-path';
import { getClaudeDir, getSessionDir, getArchivedSessionDir, getOpencodeDir, getOpencodeSessionDir, getPiMonoDir, getPiMonoSessionDir } from '@@/data-path';
import type { HonoVariables } from '@@/create-router';
const OPENCODE_PORT = process.env.OPENCODE_PORT ?? '10006';
@@ -14,9 +14,13 @@ export const sessionsRouter = new Hono<{ Variables: HonoVariables }>();
sessionsRouter.get('/sessions', async (ctx) => {
const { email } = ctx.get('user');
const [claudeSessions, opencodeSessions] = await Promise.all([fetchClaudeSessions(email), fetchOpencodeSessions(email)]);
const [claudeSessions, opencodeSessions, piMonoSessions] = await Promise.all([
fetchClaudeSessions(email),
fetchOpencodeSessions(email),
fetchPiMonoSessions(email),
]);
const merged = [...claudeSessions, ...opencodeSessions].sort((a, b) => b.createdAt - a.createdAt);
const merged = [...claudeSessions, ...opencodeSessions, ...piMonoSessions].sort((a, b) => b.createdAt - a.createdAt);
return ctx.json(merged);
});
@@ -37,6 +41,12 @@ sessionsRouter.get('/sessions/:provider/:id/messages', async (ctx) => {
return ctx.json(await fetchOpencodeMessages(id));
}
if (provider === 'pi-mono') {
const file = Bun.file(join(getPiMonoSessionDir(email, id), 'messages.json'));
if (!(await file.exists())) return ctx.json([]);
return ctx.json(await file.json());
}
return ctx.json({ error: 'invalid provider' }, 400);
});
@@ -46,10 +56,10 @@ sessionsRouter.put('/sessions/:provider/:id/messages', async (ctx) => {
const id = ctx.req.param('id');
if (provider === 'opencode') return ctx.json({ error: 'opencode sessions are read-only' }, 400);
if (provider !== 'claude') return ctx.json({ error: 'invalid provider' }, 400);
if (provider !== 'claude' && provider !== 'pi-mono') return ctx.json({ error: 'invalid provider' }, 400);
const messages = ctx.get('body');
const dir = getSessionDir(email, id);
const dir = provider === 'pi-mono' ? getPiMonoSessionDir(email, id) : getSessionDir(email, id);
await Bun.write(join(dir, 'messages.json'), JSON.stringify(messages));
return ctx.json({ ok: true });
});
@@ -92,6 +102,16 @@ sessionsRouter.put('/sessions/:provider/:id', async (ctx) => {
return ctx.json({ ok: true });
}
if (provider === 'pi-mono') {
const dir = getPiMonoSessionDir(email, id);
const metaFile = Bun.file(join(dir, 'meta.json'));
if (!(await metaFile.exists())) return ctx.json({ error: 'session not found' }, 404);
const meta = await metaFile.json();
meta.title = body.title.slice(0, 200);
await Bun.write(join(dir, 'meta.json'), JSON.stringify(meta));
return ctx.json({ ok: true });
}
return ctx.json({ error: 'invalid provider' }, 400);
});
@@ -124,6 +144,16 @@ sessionsRouter.delete('/sessions/:provider/:id', async (ctx) => {
return ctx.json({ ok: true });
}
if (provider === 'pi-mono') {
const dir = getPiMonoSessionDir(email, id);
try {
await rm(dir, { recursive: true });
} catch {
// dir may not exist
}
return ctx.json({ ok: true });
}
return ctx.json({ error: 'invalid provider' }, 400);
});
@@ -134,7 +164,7 @@ sessionsRouter.post('/sessions/:provider/:id/archive', async (ctx) => {
const provider = ctx.req.param('provider');
const id = ctx.req.param('id');
if (provider === 'opencode') return ctx.json({ error: 'opencode sessions cannot be archived' }, 400);
if (provider === 'opencode' || provider === 'pi-mono') return ctx.json({ error: `${provider} sessions cannot be archived` }, 400);
if (provider !== 'claude') return ctx.json({ error: 'invalid provider' }, 400);
const src = getSessionDir(email, id);
@@ -150,7 +180,7 @@ type SessionMeta = {
id: string;
title: string;
createdAt: number;
provider: 'claude' | 'opencode';
provider: 'claude' | 'opencode' | 'pi-mono';
model?: string | null;
};
@@ -200,6 +230,28 @@ async function fetchOpencodeSessions(email: string): Promise<SessionMeta[]> {
}
}
async function fetchPiMonoSessions(email: string): Promise<SessionMeta[]> {
const dir = getPiMonoDir(email);
try {
const entries = await readdir(dir);
const sessions = await Promise.all(
entries.map(async (id) => {
try {
const metaFile = Bun.file(join(dir, id, 'meta.json'));
if (!(await metaFile.exists())) return null;
const meta = await metaFile.json();
return { ...meta, provider: 'pi-mono' as const };
} catch {
return null;
}
}),
);
return sessions.filter((s): s is SessionMeta => s !== null);
} catch {
return [];
}
}
async function fetchOpencodeMessages(id: string) {
try {
const res = await fetch(`${OPENCODE_BASE}/session/${id}/message`);
+1 -1
View File
@@ -13,7 +13,7 @@ uploadRouter.post('/', async (ctx) => {
const file = body.file as File | null;
const sessionId = (body.sessionId as string) || null;
const provider = (body.provider as 'claude' | 'opencode') || null;
const provider = (body.provider as 'claude' | 'opencode' | 'pi-mono') || null;
if (!file || !(file instanceof File)) {
return ctx.json({ error: 'file is required' }, 400);