chat: rename the pi chat transport + model list to chat (Stage 4b/1)

Pure rename, no behavior change. Moves the misnamed "pi" chat harness into the chat namespace: - api/pi/{websocket,session-manager,types,logger,list-models} → api/chat/ - merge api/pi/rest.ts into api/chat/chat.ts (/pi/models → /chat/models, /pi/stt → /chat/stt); drop the piRestRouter mount - PiEvent → ChatEvent, piWebsocket → chatWebsocket, listPiModels → listChatModels - WS route /api/pi/chat/ws → /api/chat/ws, provider tag 'pi' → 'chat' - frontend: useChat/useAudioRecording URLs, usePiModels→useModels / useVisiblePiModels→useVisibleModels / useEnabledPiModels→useEnabledModels, 'PI_MODELS' query key → 'CHAT_MODELS', attachments provider 'pi-mono' → 'chat' The /pi-mono provider/harness settings router is renamed separately (next commit). Note: the WS route change requires the mobile app to point at /api/chat/ws. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-24 15:47:40 +00:00
parent 70555c8d71
commit 49df1c0b0c
31 changed files with 119 additions and 129 deletions
@@ -0,0 +1,449 @@
+import type { ServerWebSocket } from 'bun';
+import { randomUUID } from 'crypto';
+import type { ClientMessage, ServerMessage, Message, ChatEvent } from './types';
+import { sessionManager } from './session-manager';
+import { sendClaudeCodeStreaming } from '@@/channels/send-claude-code';
+import { ensureClaudeSessionsCwd } from './claude-sessions';
+import * as sidecar from '@@/sidecar-registry';
+import { join } from 'path';
+import { getHomeDirForRole } from '../../../servers/data-path';
+import { getUserSettings } from 'officerdb';
+import { logger } from './logger';
+
+// Default model when no user preference is set
+const DEFAULT_MODEL = 'claude-code';
+
+async function getUserDefaultModel(userId: number): Promise<string | null> {
+  try {
+    const settings = await getUserSettings(userId);
+    const chat = settings?.chat as Record<string, unknown> | undefined;
+    return (chat?.defaultModel as string) || null;
+  } catch (err) {
+    logger.error('Failed to read user settings for default model', { userId, error: String(err) });
+  }
+  return null;
+}
+
+type WSData = {
+  userId: number;
+  email: string;
+  username: string;
+  role: string;
+  sandboxed: boolean;
+  provider: string;
+};
+
+const IDLE_TIMEOUT_MS = 60 * 60 * 1000; // 1 hour
+
+const resolveCwd = (email: string, role: string, cwd?: string) => {
+  const root = getHomeDirForRole(email, role);
+  if (!cwd || cwd === '~') return root;
+  if (cwd.startsWith('~/')) return join(root, cwd.slice(2));
+  if (cwd.startsWith('/')) {
+    // Super Admin: trust absolute paths as-is
+    if (role === 'Super Admin') return cwd;
+    return join(root, cwd.slice(1));
+  }
+  return join(root, cwd);
+};
+
+export const resolveBaseCwd = (email: string, role: string, cwd?: string) => {
+  return resolveCwd(email, role, cwd);
+};
+
+const wsToSessionMap = new WeakMap<any, string>();
+
+// Per-connection heartbeat. Bun closes a WS idle for `idleTimeout` (60s), and its timer only resets
+// on frames *received* from the client — but during a chat turn the client only receives. So we ping
+// each connection every 25s; the client auto-pongs at the protocol level, which resets Bun's timer
+// (and keeps reverse proxies happy). Genuinely dead sockets still time out (no pong).
+const pingTimers = new WeakMap<ServerWebSocket<WSData>, ReturnType<typeof setInterval>>();
+const PING_INTERVAL_MS = 25_000;
+
+function sendToClient(ws: ServerWebSocket<WSData> | null, msg: ServerMessage): void {
+  if (ws?.readyState === 1) {
+    ws.send(JSON.stringify(msg));
+  }
+}
+
+export async function open(ws: ServerWebSocket<WSData>): Promise<void> {
+  const timer = setInterval(() => {
+    try {
+      ws.ping(); // client auto-pongs → resets Bun's idleTimeout
+    } catch {
+      /* socket already gone */
+    }
+  }, PING_INTERVAL_MS);
+  pingTimers.set(ws, timer);
+}
+
+export function message(ws: ServerWebSocket<WSData>, raw: string | Buffer): void {
+  const data = typeof raw === 'string' ? raw : raw.toString();
+
+  (async () => {
+    try {
+      const clientMsg = JSON.parse(data) as ClientMessage;
+
+      if (clientMsg.type === 'chat') {
+        await handleChat(ws, clientMsg);
+      } else if (clientMsg.type === 'resume') {
+        await handleResume(ws, clientMsg);
+      } else if (clientMsg.type === 'stop') {
+        await handleStop(ws);
+      }
+    } catch (err) {
+      logger.error('Error handling WebSocket message', { email: ws.data.email, error: String(err) });
+      sendToClient(ws, { type: 'error', message: 'Failed to process message' });
+    }
+  })();
+}
+
+export function close(ws: ServerWebSocket<WSData>): void {
+  const timer = pingTimers.get(ws);
+  if (timer) {
+    clearInterval(timer);
+    pingTimers.delete(ws);
+  }
+
+  const sessionId = wsToSessionMap.get(ws);
+  if (sessionId) {
+    sessionManager.detachWs(sessionId);
+    sessionManager.setIdleTimeout(sessionId, IDLE_TIMEOUT_MS);
+  }
+}
+
+function createEventHandler(sessionId: string, model: string, cwd: string) {
+  return async (event: ChatEvent): Promise<void> => {
+    const session = sessionManager.getSession(sessionId);
+    if (!session) return;
+
+    const ws = session.ws as ServerWebSocket<WSData> | null;
+
+    switch (event.type) {
+      case 'delta': {
+        sendToClient(ws, { type: 'assistant:delta', text: event.text });
+        session.streamBuffer += event.text;
+        break;
+      }
+
+      case 'text': {
+        // Flush streaming buffer as complete text
+        const text = event.text || session.streamBuffer;
+        if (text) {
+          sendToClient(ws, { type: 'assistant:text', text });
+
+          const assistantMsg: Message = {
+            id: randomUUID(),
+            timestamp: Date.now(),
+            role: 'assistant',
+            text,
+            model,
+          };
+          session.messages.push(assistantMsg);
+          session.meta.messageCount += 1;
+          session.streamBuffer = '';
+        }
+        break;
+      }
+
+      case 'tool:start': {
+        // Flush any pending streaming text first
+        if (session.streamBuffer) {
+          sendToClient(ws, { type: 'assistant:text', text: session.streamBuffer });
+
+          const assistantMsg: Message = {
+            id: randomUUID(),
+            timestamp: Date.now(),
+            role: 'assistant',
+            text: session.streamBuffer,
+            model,
+          };
+          session.messages.push(assistantMsg);
+          session.meta.messageCount += 1;
+          session.streamBuffer = '';
+        }
+
+        sendToClient(ws, {
+          type: 'tool:start',
+          toolCallId: event.toolCallId,
+          toolName: event.toolName,
+          toolInput: event.toolInput,
+        });
+
+        const toolMsg: Message = {
+          id: randomUUID(),
+          timestamp: Date.now(),
+          role: 'tool',
+          toolCallId: event.toolCallId,
+          toolName: event.toolName,
+          toolInput: event.toolInput,
+        };
+        session.messages.push(toolMsg);
+        session.meta.messageCount += 1;
+        break;
+      }
+
+      case 'tool:result': {
+        sendToClient(ws, {
+          type: 'tool:result',
+          toolCallId: event.toolCallId,
+          output: event.output,
+          isError: event.isError,
+        });
+
+        // Update existing tool message with output
+        for (let i = session.messages.length - 1; i >= 0; i--) {
+          const m = session.messages[i]!;
+          if (m.role === 'tool' && m.toolCallId === event.toolCallId) {
+            m.output = event.output;
+            m.isError = event.isError;
+            break;
+          }
+        }
+        break;
+      }
+
+      case 'result': {
+        // Flush any remaining streaming buffer
+        if (session.streamBuffer) {
+          sendToClient(ws, { type: 'assistant:text', text: session.streamBuffer });
+
+          const assistantMsg: Message = {
+            id: randomUUID(),
+            timestamp: Date.now(),
+            role: 'assistant',
+            text: session.streamBuffer,
+            model,
+            cost: event.cost,
+          };
+          session.messages.push(assistantMsg);
+          session.meta.messageCount += 1;
+          session.streamBuffer = '';
+        }
+
+        sendToClient(ws, { type: 'result', sessionId, cost: event.cost });
+
+        session.isGenerating = false;
+        session.meta.cost.inputTokens += event.cost.inputTokens;
+        session.meta.cost.outputTokens += event.cost.outputTokens;
+        session.meta.cost.totalUSD += event.cost.totalUSD;
+        session.meta.updatedAt = Date.now();
+        // No disk persistence — Claude's transcript is the record.
+        break;
+      }
+
+      case 'error': {
+        sendToClient(ws, { type: 'error', message: event.message });
+        session.isGenerating = false;
+        break;
+      }
+
+      case 'stopped': {
+        sendToClient(ws, { type: 'stopped' });
+        session.isGenerating = false;
+        break;
+      }
+    }
+  };
+}
+
+async function handleChat(
+  ws: ServerWebSocket<WSData>,
+  msg: {
+    prompt: string;
+    displayText?: string;
+    sessionId?: string;
+    model?: string;
+    cwd?: string;
+    cwdRoot?: string;
+    sandboxed?: boolean;
+    groupSlug?: string;
+    attachmentIds?: string[];
+    thinking?: string;
+    context?: string;
+    contextId?: string;
+    resumeSummary?: string;
+    resumeSessionId?: string;
+  },
+): Promise<void> {
+  const { userId } = ws.data;
+  const sessionId = msg.sessionId || randomUUID();
+
+  // Prepend resume summary to the prompt if present
+  const prompt = msg.resumeSummary
+    ? `Here is a summary of a previous conversation to continue from:\n\n${msg.resumeSummary}\n\n---\n\nUser's new message: ${msg.prompt}`
+    : msg.prompt;
+
+  // Use provided model, or fall back to user default, or the system default.
+  let model = msg.model || (await getUserDefaultModel(userId)) || DEFAULT_MODEL;
+
+  // Claude-only: coerce any legacy/non-Claude model preference to the Claude default so old saved
+  // settings (Pi/opencode/openrouter model ids) don't break chat.
+  if (!model.startsWith('claude-code')) {
+    logger.info('Coercing non-Claude model to Claude default', { sessionId, requested: model });
+    model = DEFAULT_MODEL;
+  }
+
+  logger.info('Model selected for chat', { sessionId, model, clientModel: msg.model || null });
+
+  return handleClaudeCodeChat(ws, sessionId, model, msg, prompt);
+}
+
+async function handleClaudeCodeChat(
+  ws: ServerWebSocket<WSData>,
+  sessionId: string,
+  model: string,
+  msg: {
+    prompt: string;
+    displayText?: string;
+    groupSlug?: string;
+    context?: string;
+    contextId?: string;
+    cwd?: string;
+    cwdRoot?: string;
+    sandboxed?: boolean;
+    resumeSessionId?: string;
+  },
+  effectivePrompt: string,
+): Promise<void> {
+  const { email, username, userId } = ws.data;
+
+  // The standalone /chat route runs from a chosen working directory (the pwd selector) or, by default,
+  // a dedicated `claude_sessions` dir — so transcripts form their own Claude "project" group per cwd.
+  // Other contexts (email/project panels) keep their own cwd.
+  const cwd =
+    msg.context === 'chat'
+      ? msg.cwd?.trim()
+        ? resolveCwd(email, ws.data.role, msg.cwd)
+        : ensureClaudeSessionsCwd(email)
+      : resolveCwd(email, ws.data.role, msg.cwd);
+
+  const groupSlug = msg.groupSlug || null;
+
+  const session = sessionManager.getOrCreate(sessionId, email, cwd, model, groupSlug, msg.context, msg.contextId);
+  session.userId = userId;
+  sessionManager.attachWs(sessionId, ws);
+  wsToSessionMap.set(ws as any, sessionId);
+
+  sendToClient(ws, {
+    type: 'session:init',
+    sessionId,
+    model,
+    cwd,
+    context: session.meta.context,
+    contextId: session.meta.contextId,
+  });
+
+  // Add user message to session
+  const userMsg: Message = {
+    id: randomUUID(),
+    timestamp: Date.now(),
+    role: 'user',
+    text: msg.prompt,
+  };
+  session.messages.push(userMsg);
+  session.meta.messageCount += 1;
+  session.meta.updatedAt = Date.now();
+
+  if (!session.meta.title) {
+    session.meta.title = (msg.displayText ?? msg.prompt).slice(0, 100);
+  }
+
+  session.isGenerating = true;
+
+  const onEvent = createEventHandler(sessionId, model, cwd);
+
+  try {
+    const handle = await sendClaudeCodeStreaming({
+      userId,
+      email,
+      username,
+      prompt: effectivePrompt,
+      sessionKey: sessionId,
+      cwd,
+      model,
+      role: ws.data.role,
+      resumeSessionId: msg.resumeSessionId,
+      onEvent,
+    });
+
+    // Store sentinel so handleStop can kill it via sidecar
+    session.piProcess = sessionId as any;
+    session._claudeKill = handle.kill;
+  } catch (err) {
+    logger.error('Failed to start Claude Code streaming', { sessionId, error: String(err) });
+    sendToClient(ws, { type: 'error', message: 'Failed to start Claude Code' });
+    session.isGenerating = false;
+  }
+}
+
+async function handleResume(
+  ws: ServerWebSocket<WSData>,
+  msg: { sessionId: string; cwd?: string; cwdRoot?: string },
+): Promise<void> {
+  const { sessionId } = msg;
+
+  try {
+    const session = sessionManager.getSession(sessionId);
+
+    if (!session) {
+      // Sessions live in memory for the connection's lifetime; there's no disk store to reload from.
+      sendToClient(ws, { type: 'error', message: 'Session not found', errorCode: 'SESSION_NOT_FOUND' });
+      return;
+    }
+
+    sessionManager.attachWs(sessionId, ws);
+    wsToSessionMap.set(ws as any, sessionId);
+
+    sendToClient(ws, {
+      type: 'session:init',
+      sessionId,
+      model: session.model,
+      cwd: session.cwd,
+      context: session.meta.context,
+      contextId: session.meta.contextId,
+    });
+
+    // Claude resumes lazily: the next chat prompt re-attaches via `--resume <sessionKey>`, so there's
+    // no long-lived process to spawn here — just replay the stored transcript to the client.
+    sendToClient(ws, {
+      type: 'sync:messages',
+      sessionId,
+      messages: session.messages,
+      isGenerating: session.isGenerating,
+      streamingText: session.streamBuffer,
+    });
+
+    logger.info('Session resumed successfully', { sessionId, messageCount: session.messages.length });
+  } catch (err) {
+    logger.error('Unexpected error in handleResume', { sessionId, error: String(err) });
+    sendToClient(ws, { type: 'error', message: 'Failed to resume session' });
+  }
+}
+
+async function handleStop(ws: ServerWebSocket<WSData>): Promise<void> {
+  const sessionId = wsToSessionMap.get(ws);
+
+  if (sessionId) {
+    const session = sessionManager.getSession(sessionId);
+
+    if (session?.piProcess) {
+      try {
+        sidecar.killClaude(sessionId, session.email);
+        logger.info('Killed Claude Code process via sidecar', { sessionId });
+        session.isGenerating = false;
+      } catch (err) {
+        logger.error('Failed to stop process', { sessionId, error: String(err) });
+      }
+    }
+  }
+
+  sendToClient(ws, { type: 'stopped' });
+}
+
+export const chatWebsocket = {
+  open,
+  message,
+  close,
+  drain() {},
+};