platform/src/servers/api/chat/websocket.ts

import type { ServerWebSocket } from 'bun';
import { randomUUID } from 'crypto';
import type {
  ClientMessage,
  ServerMessage,
  Message,
  PromptImage,
  RunningTask,
  TurnMessage,
  UserSession,
} from './types';
import { sessionManager } from './session-manager';
import { rememberOpenCodePrompt } from './opencode/state';
import { sendClaudeCodeStreaming } from '@@/channels/send-claude-code';
import { sendOpenCodeStreaming } from '@@/channels/send-opencode';
import * as sidecar from '@@/sidecar-registry';
import { join } from 'path';
import { getOwnerHomeDir, getEmailAccountsDir } from '../../../servers/data-path';
import { getUserSettings, getEmailAccounts, getChatEventsSince, appendChatEvent, getUserById } from 'officerdb';
import { resolveHomeDir } from '@@/user-home';
import { claudeLoginState } from '@@/os-user-claude';
import { mkdirSync } from 'node:fs';
import { logger } from './logger';

// Default model when no user preference is set
const DEFAULT_MODEL = 'claude-code';

// Harness selection: the `claude-code` provider runs through the Claude sidecar; every other provider
// (opencode/anthropic/openai/… — all `providerID/modelID` ids) runs through the OpenCode server.
const isClaudeModel = (model: string): boolean => model.startsWith('claude-code');

/**
 * Whose identity a turn runs as.
 *
 * A three-way answer rather than a nullable one, because `undefined` downstream means **the owner** — their
 * binary, their `~/.claude` credential, their HOME, and their MCP config carrying `OFFICER_AUTH_TOKEN`. A
 * nullable return collapsed three inputs into that: the caller genuinely being the owner, `resolveHomeDir`
 * failing, and a member whose `osUser` is null. The last two mean "I could not determine whose this is", and
 * answering them with the owner's identity is the one wrong answer this whole feature exists to prevent.
 *
 * Case three is not hypothetical: `provisionOsAccount` is non-fatal at every stage and records the account
 * either way, so a member whose Linux provisioning failed exists as a row with no `osUser`. On the night this
 * was written, provisioning failed three separate ways on a real member while the account continued to exist.
 *
 * The property: the owner's identity is reachable only by positively establishing that the caller IS the
 * owner, never by failing to establish anything else. `resolveHomeDir` already reports `isOwner` as a positive
 * fact — the old funnel through `undefined` was the only thing throwing it away.
 */
type TurnIdentity =
  | { kind: 'owner' }
  | { kind: 'member'; run: { osUser: string; home: string } }
  | { kind: 'refuse'; reason: string };

async function resolveTurnIdentity(userId: number): Promise<TurnIdentity> {
  const resolved = await resolveHomeDir(userId);
  if (!resolved.ok) return { kind: 'refuse', reason: resolved.reason };
  if (resolved.isOwner) return { kind: 'owner' };

  const row = await getUserById(userId);
  if (!row?.osUser) {
    return { kind: 'refuse', reason: 'your Linux account is not provisioned yet, so an agent cannot run as you' };
  }
  return { kind: 'member', run: { osUser: row.osUser, home: resolved.home } };
}

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;
  provider: string;
};

const IDLE_TIMEOUT_MS = 60 * 60 * 1000; // 1 hour

/**
 * Where a turn runs, relative to the caller's own home.
 *
 * `root` used to be `getOwnerHomeDir(email)`, which ignores its argument whenever HOME_DIR is set — so every
 * `~` expanded to the OWNER'S home regardless of who asked, and the comment here said "the server owner is
 * the only account" as though that were a property rather than an assumption.
 *
 * An absolute path is still passed through unchanged. That is not a hole: a member's turn runs as their Linux
 * account, so the kernel decides what it can open, and containment is `resolveUserPath`'s job in the file
 * browser rather than a string check here. But it is worth knowing it is the kernel doing the work.
 */
const resolveCwd = (home: string, cwd?: string) => {
  if (!cwd || cwd === '~') return home;
  if (cwd.startsWith('~/')) return join(home, cwd.slice(2));
  if (cwd.startsWith('/')) return cwd;
  return join(home, cwd);
};

/**
 * @param home an absolute filesystem path — NOT an email.
 *
 * It took an email until `95951fb`, resolved its own root, and both parameters are `string`, so the change of
 * meaning was invisible to the compiler and to every caller outside that diff. Three of them kept passing an
 * email and silently began building relative paths out of an address. If a fourth caller ever appears, this
 * line is the warning it gets.
 */
export const resolveBaseCwd = (home: string, cwd?: string) => resolveCwd(home, cwd);

// The email chat runs from the selected account's storage dir:
//   DATA_PATH/<owner>/email_accounts/<accountEmail>
// `accountEmail` will come from the account selector (msg.contextId) later; for now default to the
// owner's first enabled account. Falls back to the email_accounts root if there are no accounts.
async function resolveEmailCwd(userId: number, ownerEmail: string, accountEmail?: string): Promise<string> {
  let account = accountEmail?.trim();
  if (!account) {
    try {
      const accounts = await getEmailAccounts(userId);
      account = (accounts.find((a) => a.enabled) ?? accounts[0])?.email;
    } catch (err) {
      logger.error('Failed to resolve email account for chat cwd', { userId, error: String(err) });
    }
  }
  const dir = account ? join(getEmailAccountsDir(ownerEmail), account) : getEmailAccountsDir(ownerEmail);
  mkdirSync(dir, { recursive: true });
  return dir;
}

// The working directory a chat turn runs in, by context: email → the account dir; /chat → a chosen
// pwd or the caller's own home; everything else (browser/project/dashboard) → the given cwd.
async function resolveChatCwd(
  msg: { context?: string; contextId?: string; cwd?: string },
  email: string,
  userId: number,
  home: string,
): Promise<string> {
  if (msg.context === 'email') return resolveEmailCwd(userId, email, msg.contextId);
  if (msg.context === 'chat') {
    if (msg.cwd?.trim()) return resolveCwd(home, msg.cwd);
    // The caller's own home, for everyone.
    //
    // This used to be `DATA_PATH/<email>/general_chat_sessions`, a dedicated directory so /chat sessions
    // formed their own Claude project group and did not clutter the home. That is a sibling of the home,
    // and `confineUserTree` makes every sibling the platform's at 0700 because the others are `attachments`
    // and `email_accounts` — so it was unreachable for a member. The first live member turn ran there and
    // every Bash call failed on its own working directory before doing anything.
    //
    // A per-member copy inside each home would have worked and would have left two rules to remember. The
    // owner chose one: a chat with no chosen directory runs in the account's own home, whoever they are.
    // The cost is that /chat sessions now share a project group with anything else run from that home,
    // which was the reason the dedicated directory existed and is a trade the owner made knowingly.
    return home;
  }
  return resolveCwd(home, msg.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, seq?: number): void {
  if (ws?.readyState === 1) {
    ws.send(JSON.stringify(seq === undefined ? msg : { ...msg, seq }));
  }
}

// Nothing in officer writes to chat_session_events any more. Both harnesses commit their own turn
// output in the sidecar that produced it (`sidecar/claude/session-log.ts`), which is the whole point:
// the durable record does not travel over the socket between the two processes, so officer can restart
// mid-turn without losing it. Officer reads the table on `resume` (getChatEventsSince) and relays what
// the sidecars send. The old `emitToSession` used to live here.

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);
      } else if (clientMsg.type === 'disconnect') {
        await handleDisconnect(ws);
      } else if (clientMsg.type === 'resume-cursor') {
        await handleResumeCursor(ws, clientMsg);
      } else if (clientMsg.type === 'attach') {
        await handleAttach(ws, clientMsg);
      }
    } 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, ws);
    // Only once nothing is watching. Another pane or another device still attached means the
    // conversation is live, and arming the idle GC here would collect it out from under them.
    if (!sessionManager.hasSockets(sessionId)) sessionManager.setIdleTimeout(sessionId, IDLE_TIMEOUT_MS);
  }
}

// ── Claude Code: relay, don't rebuild ──

// The agent sidecar owns Claude's turn output end to end — it translates the parser stream, commits each
// message to chat_session_events and hands us a finished message plus its cursor id. Officer relays it.
// That is what makes a restart survivable: the durable record no longer travels over the socket between
// the two processes, so if this one is down the output is already written and the client replays it.
//
// Officer keeps only the in-memory transcript, which exists to answer a `resume` with sync:messages —
// Claude's own transcript is the real record — so it is folded from the same messages, not rebuilt.
function foldIntoSession(session: UserSession, msg: TurnMessage, model: string): void {
  switch (msg.type) {
    case 'assistant:delta':
      // A subagent's deltas are not the main agent typing; appending them here spliced its sentences into
      // whatever the agent you are talking to was mid-way through saying.
      if (!msg.parentToolUseId) session.streamBuffer += msg.text;
      break;

    case 'assistant:text':
      session.messages.push({
        id: randomUUID(),
        timestamp: Date.now(),
        role: 'assistant',
        text: msg.text,
        model,
        parentToolUseId: msg.parentToolUseId,
      });
      session.meta.messageCount += 1;
      // Only the main agent's own stream feeds the buffer a resume replays as `streamingText`.
      if (!msg.parentToolUseId) session.streamBuffer = '';
      break;

    case 'tool:start':
      session.messages.push({
        id: randomUUID(),
        timestamp: Date.now(),
        role: 'tool',
        toolCallId: msg.toolCallId,
        toolName: msg.toolName,
        toolInput: msg.toolInput,
        parentToolUseId: msg.parentToolUseId,
      });
      session.meta.messageCount += 1;
      break;

    case 'tool:result':
      for (let i = session.messages.length - 1; i >= 0; i--) {
        const m = session.messages[i]!;
        if (m.role === 'tool' && m.toolCallId === msg.toolCallId) {
          m.output = msg.output;
          m.isError = msg.isError;
          break;
        }
      }
      break;

    case 'result': {
      session.isGenerating = false;
      session.meta.cost.inputTokens += msg.cost.inputTokens;
      session.meta.cost.outputTokens += msg.cost.outputTokens;
      session.meta.cost.totalUSD += msg.cost.totalUSD;
      session.meta.updatedAt = Date.now();
      // The turn's cost belongs to the assistant message it paid for (as it did when officer built these).
      // The turn's cost is the main agent's, so skip past any subagent tail.
      const last = [...session.messages].reverse().find((m) => !m.parentToolUseId);
      if (last?.role === 'assistant' && !last.cost) last.cost = msg.cost;
      break;
    }

    case 'error':
    case 'stopped':
      session.isGenerating = false;
      break;
  }
}

// Shared by both harnesses now: whichever sidecar ran the turn has already translated and committed it.
function createMessageHandler(sessionId: string, model: string) {
  return (msg: TurnMessage, seq?: number): void => {
    const session = sessionManager.getSession(sessionId);
    if (!session) return;
    foldIntoSession(session, msg, model);
    for (const socket of session.sockets) sendToClient(socket as ServerWebSocket<WSData>, msg, seq);
  };
}

async function handleChat(
  ws: ServerWebSocket<WSData>,
  msg: {
    prompt: string;
    displayText?: string;
    sessionId?: string;
    model?: string;
    cwd?: string;
    cwdRoot?: string;
    groupSlug?: string;
    attachmentIds?: string[];
    images?: PromptImage[];
    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.
  const model = msg.model || (await getUserDefaultModel(userId)) || DEFAULT_MODEL;

  logger.info('Model selected for chat', { sessionId, model, clientModel: msg.model || null });

  // Route by harness: claude-code → Claude sidecar; anything else → OpenCode server.
  return isClaudeModel(model)
    ? handleClaudeCodeChat(ws, sessionId, model, msg, prompt)
    : handleOpenCodeChat(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;
    images?: PromptImage[];
    resumeSessionId?: string;
  },
  effectivePrompt: string,
): Promise<void> {
  const { email, username, userId } = ws.data;

  // Identity first: it decides both whose home `~` expands against and whose account the turn runs as, and
  // those must be the same answer. Resolving the cwd first would expand `~` before knowing whose it was.
  const identity = await resolveTurnIdentity(userId);
  if (identity.kind === 'refuse') {
    sendToClient(ws, { type: 'error', message: identity.reason });
    return;
  }

  // A member has to sign `claude` in themselves, once, with their own Anthropic account — the platform cannot
  // do it for them without lending them the owner's credential, which is the thing this whole feature exists
  // to avoid. Without this check their turn spawns, `claude` exits on an auth error, and it surfaces as "the
  // agent is broken" — the exact confusion `/agent-status` was built to prevent, arriving through a different
  // door. The refusal carries the instruction so the answer is the same whether the UI asked or not.
  if (identity.kind === 'member') {
    const state = await claudeLoginState({ email, osUser: identity.run.osUser });
    if (!state.installed) {
      sendToClient(ws, {
        type: 'error',
        message: 'Claude is not installed in your home yet — ask the server owner to reprovision your account.',
      });
      return;
    }
    if (!state.loggedIn) {
      sendToClient(ws, {
        type: 'error',
        message:
          'Open a terminal and run `claude` once to sign in with your own Anthropic account. It stays signed in.',
      });
      return;
    }
  }

  const home = identity.kind === 'member' ? identity.run.home : getOwnerHomeDir(email);

  const cwd = await resolveChatCwd(msg, email, userId, home);

  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 onMessage = createMessageHandler(sessionId, model);

  try {
    if (!session._claudeKill) {
      // First turn of this session: open the persistent session + a SESSION-scoped event subscription
      // (survives turn-end so background task:notifications keep flowing). `kill` tears both down for an
      // explicit disconnect; `detach` drops only the listener, which is what the idle GC uses so an
      // absent browser stops taking a live agent with it.
      const handle = await sendClaudeCodeStreaming({
        userId,
        email,
        username,
        prompt: effectivePrompt,
        images: msg.images,
        sessionKey: sessionId,
        cwd,
        model,
        member: identity.kind === 'member' ? identity.run : undefined,
        resumeSessionId: msg.resumeSessionId,
        onMessage,
      });
      session.piProcess = sessionId as any;
      session._claudeKill = handle.kill;
      session._sidecarUnsub = handle.detach;
    } else {
      // Session already live: push this turn onto the existing persistent session (no new subscription).
      await sidecar.spawnClaudeStreaming({
        userId,
        email,
        username,
        prompt: effectivePrompt,
        images: msg.images,
        sessionKey: sessionId,
        cwd,
        model,
        resumeSessionId: msg.resumeSessionId,
      });
    }
  } 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 handleOpenCodeChat(
  ws: ServerWebSocket<WSData>,
  sessionId: string,
  model: string,
  msg: {
    prompt: string;
    displayText?: string;
    groupSlug?: string;
    context?: string;
    contextId?: string;
    cwd?: string;
    cwdRoot?: string;
    resumeSessionId?: string;
    // The whole of B4 lived in this omission. The browser sent images, the bubble rendered them, and
    // they stopped at this signature — so they were never passed on and never reached the model, with
    // nothing anywhere reporting a loss.
    images?: PromptImage[];
  },
  effectivePrompt: string,
): Promise<void> {
  const { email, username, userId } = ws.data;

  // The owner's home: opencode receives no identity at all (`TODO.md` → Multi-user), so this path is
  // owner-only and resolving anything else here would imply an isolation it does not have.
  const cwd = await resolveChatCwd(msg, email, userId, getOwnerHomeDir(email));

  // Names this session in the Live panel until OpenCode gets round to titling it. First turn only.
  rememberOpenCodePrompt(sessionId, msg.displayText || msg.prompt);

  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,
  });

  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 onMessage = createMessageHandler(sessionId, model);

  try {
    // Drop the PREVIOUS turn's listener before opening the next one.
    //
    // This deliberately does not mirror the Claude guard above. Claude keeps one persistent session and
    // skips re-subscribing; OpenCode runs a fresh `opencode run` subprocess per turn, so a new
    // subscription each time is correct. What was wrong is that the old handle was overwritten without
    // being detached, leaving the previous session-scoped listener attached — so every turn after the
    // first delivered doubled, tripled, and so on, for any termination that is not result/error/stopped.
    session._sidecarUnsub?.();
    session._sidecarUnsub = undefined;

    const handle = await sendOpenCodeStreaming({
      userId,
      email,
      username,
      prompt: effectivePrompt,
      sessionKey: sessionId,
      cwd,
      model,
      resumeSessionId: msg.resumeSessionId,
      images: msg.images,
      onMessage,
    });

    // Store the abort handle so handleStop can end the turn (OpenCode is aborted via this handle).
    session.piProcess = sessionId as any;
    session._claudeKill = handle.kill;
    session._sidecarUnsub = handle.detach;
  } catch (err) {
    logger.error('Failed to start OpenCode streaming', { sessionId, error: String(err) });
    sendToClient(ws, { type: 'error', message: 'Failed to start OpenCode' });
    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 {
        if (isClaudeModel(session.model)) {
          // Interrupt the current turn but KEEP the persistent session alive (background tasks + the
          // warm worker survive). Full teardown is 'disconnect' → deleteSession → _claudeKill.
          void sidecar.interruptClaude(sessionId, ws.data.userId);
          logger.info('Interrupted Claude Code turn via sidecar (session stays warm)', { sessionId });
        } else {
          session._claudeKill?.(); // OpenCode: abort the turn via the stored handle
          logger.info('Aborted OpenCode turn', { sessionId });
        }
        session.isGenerating = false;
      } catch (err) {
        logger.error('Failed to stop process', { sessionId, error: String(err) });
      }
    }
  }

  sendToClient(ws, { type: 'stopped' });
}

// Tear down the whole session (not just the current turn): deleteSession fires _claudeKill (kills any
// in-flight Claude/OpenCode turn) + _sidecarUnsub, clears the idle timer, and drops the session from the
// manager's maps. The WS stays open so the client can immediately start a fresh session.
async function handleDisconnect(ws: ServerWebSocket<WSData>): Promise<void> {
  const sessionId = wsToSessionMap.get(ws);
  if (sessionId) {
    try {
      sessionManager.deleteSession(sessionId);
      wsToSessionMap.delete(ws as any);
      logger.info('Disconnected chat session', { sessionId });
    } catch (err) {
      logger.error('Failed to disconnect session', { sessionId, error: String(err) });
    }
  }
  sendToClient(ws, { type: 'disconnected' });
}

/**
 * Re-adopt a session whose in-memory record died with the process that held it.
 *
 * The sidecars are PM2 peers, so `pm2 restart officer` does not touch a running turn: the agent keeps
 * generating and keeps committing to chat_session_events. What the restart destroys is purely this
 * process's binding to it — the session record and, critically, the session-scoped subscription that
 * relays the sidecar's events to the browser. Re-creating the record is not enough on its own; without
 * the subscription the client reconnects, receives its replay, and then goes silent for the rest of the
 * turn, which is indistinguishable from the agent having died.
 *
 * Nothing is spawned here. The subscription is a local event-bus filter, so adopting a session that is
 * NOT in fact still live upstream costs a listener that never fires and a record the idle GC collects.
 */
function adoptOrphanedSession(ws: ServerWebSocket<WSData>, sessionId: string, model: string, cwd: string): UserSession {
  const { email, userId } = ws.data;
  const session = sessionManager.getOrCreate(sessionId, email, cwd, model, null);
  session.userId = userId;
  session.piProcess = sessionId as any;

  const onMessage = createMessageHandler(sessionId, model);
  const unsubClaude = sidecar.onClaudeMessage((key, msg, seq) => {
    if (key === sessionId) onMessage(msg, seq);
  });
  const unsubOpenCode = sidecar.onOpenCodeMessage((key, msg, seq) => {
    if (key === sessionId) onMessage(msg, seq);
  });
  const unsub = () => {
    unsubClaude();
    unsubOpenCode();
  };

  // Must be set, and not only for teardown: handleChat treats an absent `_claudeKill` as "first turn of
  // this session" and opens a *second* subscription, which would then deliver every message twice.
  session._claudeKill = () => {
    if (isClaudeModel(model)) sidecar.killClaude(sessionId, userId);
    else sidecar.killOpenCode(sessionId);
    unsub();
  };
  // An adopted session can idle out and be released like any other, and releasing detaches through this
  // field alone. Leaving it unset would drop the record while the listener stayed subscribed — a leak
  // that grows by one every time a browser adopts a session and then goes away.
  session._sidecarUnsub = unsub;

  logger.info('Adopted orphaned chat session after restart', { sessionId, model });
  return session;
}

// Reconnect: re-bind this socket to the (possibly still-live) session and replay every durable event
// queued since the client's cursor — so a brief disconnect never loses turn output or a background
// task:notification. attachWs cancels the pending idle-GC.
async function handleResumeCursor(
  ws: ServerWebSocket<WSData>,
  msg: { sessionId: string; cursor: number; model?: string; cwd?: string; generating?: boolean },
): Promise<void> {
  const { sessionId, cursor } = msg;

  const known = sessionManager.getSession(sessionId);
  const decision = decideResume(known?.model, msg.model);

  if (decision.kind !== 'known') {
    // Adopt on an assumption too: an unbound socket misses the turn entirely, which is worse than a
    // harness guess that only ever costs us the cut-off check below.
    adoptOrphanedSession(ws, sessionId, decision.model, msg.cwd ?? '');
    if (decision.kind === 'assume') {
      logger.warn('resume-cursor for an unknown session with no model; adopting on the default', { sessionId });
    }
  }

  sessionManager.attachWs(sessionId, ws);
  wsToSessionMap.set(ws as any, sessionId);
  try {
    const events = await getChatEventsSince(sessionId, cursor ?? 0);
    for (const { id, event } of events) {
      sendToClient(ws, event as ServerMessage, id);
    }
  } catch (err) {
    logger.error('Failed to replay chat events on resume-cursor', { sessionId, error: String(err) });
  }
  // Only ask when the harness is actually known — never on an assumption. This check writes a DURABLE
  // row, so a wrong answer is permanent: that was B7, where a defaulted `claude-code` made officer ask
  // the Claude sidecar about an OpenCode session, hear "not generating", and write "the agent went
  // away" into a turn that was running perfectly well.
  if (msg.generating && decision.kind !== 'assume') {
    await endTurnIfAgentIsGone([ws], sessionId, decision.model, ws.data.userId);
  }
}

export type ResumeDecision =
  | { kind: 'known'; model: string }
  | { kind: 'adopt'; model: string }
  | { kind: 'assume'; model: string };

/**
 * Which harness a resuming session belongs to, and what that permits.
 *
 * This used to be `msg.model || DEFAULT_MODEL`, and `DEFAULT_MODEL` is `claude-code`. So a resume-cursor
 * that omitted `model` declared every session — OpenCode ones included — to be Claude, with three
 * consequences that all read as something else:
 *
 *  - **Adopted into the wrong harness.** `adoptOrphanedSession` subscribes to the sidecar bus for that
 *    model and pins `session.model` for the rest of its life, so an OpenCode turn's output never
 *    arrived, and stopping it called `killClaude` on a key that sidecar had never held — a dead stop
 *    button, silently.
 *  - **A durable false `cut-off`.** `endTurnIfAgentIsGone` asked the Claude sidecar whether it was
 *    generating, was told `false` because it had never heard of the session, and wrote "the agent went
 *    away" into a turn that was running normally. It survives reload, which is the whole point of
 *    writing it durably, and is therefore unrecoverable from the UI.
 *  - Masked, never fixed, by the client always happening to send `model` next to `sessionId`.
 *
 * Two rules. **The server's own record beats the client's claim** — a session in memory already knows its
 * harness, and letting a socket re-declare it is how the wrong sidecar gets a session in the first place.
 * **A guess is still a guess**: when neither knows, the session is adopted on the default so delivery
 * keeps working, but it is marked `assume` and nothing durable may be written from it.
 *
 * The middle position — refusing to adopt at all when the harness is unknown — was tried and was WRONG,
 * visibly so within the hour. `useChat.ts` sends `model` only `if (modelRef.current)`, so a reconnect
 * without one is ordinary, not exotic; declining to adopt left the socket unbound to a live turn, and
 * the running turn's output went nowhere. On screen: the transcript collapsed to "turn completed
 * without output" and only a refresh — which rebuilds from the durable log — brought it back.
 *
 * So adoption is about DELIVERY and must be generous. Only the durable write needs certainty.
 */
export function decideResume(knownModel: string | undefined, claimedModel: string | undefined): ResumeDecision {
  if (knownModel) return { kind: 'known', model: knownModel };
  if (claimedModel) return { kind: 'adopt', model: claimedModel };
  return { kind: 'assume', model: DEFAULT_MODEL };
}

/**
 * Which background tasks are still outstanding, by replaying the durable log against itself.
 *
 * A task's whole life is two events — `task:started` and, eventually, `task:notification` with a terminal
 * status — so started-minus-notified is the answer, and a Map keyed by taskId keeps the last word on each.
 * There is no third event: a task the agent abandoned without notifying stays here until the log is pruned,
 * which is the honest reading of the record rather than a bug to paper over.
 */
function collectRunningTasks(events: ServerMessage[]): RunningTask[] {
  const running = new Map<string, RunningTask>();
  for (const event of events) {
    if (event.type === 'task:started') {
      running.set(event.taskId, {
        taskId: event.taskId,
        description: event.description,
        taskType: event.taskType,
      });
    } else if (event.type === 'task:notification') {
      running.delete(event.taskId);
    }
  }
  return [...running.values()];
}

/**
 * Re-bind a socket that knows only Claude's transcript uuid.
 *
 * This is the refresh case, and until now it was the hole in an otherwise complete reconnect path. Every
 * piece of the machinery already existed — the session survives a dropped socket, the agent keeps
 * generating into it, `close` only detaches and arms an hour-long idle timer — but the browser came back
 * having forgotten officer's session id, so `resume-cursor` could never fire and the output simply stopped
 * arriving. The uuid in the URL is the one identifier a refresh cannot destroy; the agent's on-disk map
 * turns it back into the key everything else here is written in terms of.
 *
 * Deliberately hands over only the live turn, never the transcript — see `sync:live`.
 */
async function handleAttach(ws: ServerWebSocket<WSData>, msg: { claudeSessionId: string }): Promise<void> {
  const { claudeSessionId } = msg;
  if (!claudeSessionId) return;

  const sessionId = await sidecar.findClaudeSessionKey(claudeSessionId, ws.data.userId);
  if (!sessionId) {
    // No agent, or a transcript it has never run. Nothing is wrong: an ordinary finished conversation
    // opened from history lands here every time. Stay silent and leave the socket as it was — the next
    // `chat` mints a session in the usual way.
    logger.info('Attach found no live session for transcript', { claudeSessionId });
    return;
  }

  // An officer restart takes the in-memory session with it while the agent carries on, so the key can
  // resolve to a session this process has never heard of. Adopting re-subscribes it to the sidecar's bus,
  // which is what makes the rest of the turn arrive.
  // `DEFAULT_MODEL` is right here and wrong in resume-cursor, which is worth being explicit about since
  // it was just removed there: this path reached `sessionId` by asking the CLAUDE sidecar to resolve a
  // `claudeSessionId`, so the harness is not a guess — only Claude could have answered. Attach is
  // Claude-only by construction. If an OpenCode reattach verb ever lands, this stops being safe.
  const existing = sessionManager.getSession(sessionId);
  const session = existing ?? adoptOrphanedSession(ws, sessionId, DEFAULT_MODEL, '');

  sessionManager.attachWs(sessionId, ws);
  wsToSessionMap.set(ws as any, sessionId);

  // The client learns officer's key here, so any *later* drop of this socket goes down the existing
  // cursor-replay path instead of coming back through attach.
  sendToClient(ws, {
    type: 'session:init',
    sessionId,
    model: session.model,
    cwd: session.cwd,
    context: session.meta.context,
    contextId: session.meta.contextId,
  });

  // `isGenerating` is officer's own belief and is only as good as this process's memory of the turn. For
  // an adopted session it is a fresh record's default, so ask the agent — the same question, and for the
  // same reason, as `endTurnIfAgentIsGone`.
  const isGenerating = existing ? session.isGenerating : await sidecar.isClaudeGenerating(sessionId, ws.data.userId);
  session.isGenerating = isGenerating;

  // One read serves both answers: the head of the log is the cursor, and folding the whole log gives the
  // tasks still outstanding. Reading it all is affordable because attach happens once per socket and this
  // server has one user; a `getLastChatEventSeq` would only have saved a second round trip.
  let cursor = 0;
  let runningTasks: RunningTask[] = [];
  try {
    const events = await getChatEventsSince(sessionId, 0);
    cursor = events.at(-1)?.id ?? 0;
    runningTasks = collectRunningTasks(events.map((e) => e.event as ServerMessage));
  } catch (err) {
    logger.error('Failed to read the durable log on attach', { sessionId, error: String(err) });
  }

  sendToClient(ws, {
    type: 'sync:live',
    sessionId,
    isGenerating,
    cursor,
    runningTasks,
    // Whatever the agent had typed but not yet finished as a message. The transcript on disk cannot
    // supply it — the harness writes an assistant message only once it is complete — so this is the one
    // piece of the turn a refresh would otherwise genuinely lose.
    streamingText: session.streamBuffer,
  });

  logger.info('Attached socket to live session by transcript id', {
    sessionId,
    claudeSessionId,
    isGenerating,
    recoveredTasks: runningTasks.length,
  });
}

/**
 * The client came back still believing a turn is running. Check whether it is, and if it isn't, say so.
 *
 * A restart of the agent sidecar takes its persistent sessions with it, and nothing downstream notices:
 * the turn simply stops emitting. The browser's socket is fine, the conversation looks alive, and the
 * spinner runs forever — a refresh doesn't help either, because there is no ending in the transcript to
 * read. This is the one moment we can catch it, so the answer is written durably: a reload after this
 * shows the same explanation rather than a conversation that trails off mid-tool-call.
 *
 * Only the claude harness is asked. OpenCode runs a turn per invocation and has no equivalent question,
 * so its sessions are left alone rather than guessed at.
 */
async function endTurnIfAgentIsGone(
  targets: Iterable<ServerWebSocket<WSData> | null>,
  sessionId: string,
  model: string,
  userId: number,
): Promise<void> {
  if (!isClaudeModel(model)) return;
  if (await sidecar.isClaudeGenerating(sessionId, userId)) return;

  const session = sessionManager.getSession(sessionId);
  if (session) session.isGenerating = false;

  const event: ServerMessage = { type: 'cut-off' };
  try {
    const seq = await appendChatEvent(sessionId, event);
    for (const target of targets) sendToClient(target, event, seq);
  } catch (err) {
    // Still tell every client — an un-replayable explanation beats a spinner that never stops.
    logger.error('Failed to persist cut-off notice', { sessionId, error: String(err) });
    for (const target of targets) sendToClient(target, event);
  }
  logger.info('Ended a turn whose agent had gone', { sessionId });
}

// The other half of the same problem: the agent restarts while the browser sits there with a healthy
// socket, so nothing ever reconnects and nothing ever asks. A fresh agent process means every turn we
// still believe is running belongs to a process that no longer exists. On a fresh officer this loop is
// empty — it has no sessions yet — which is exactly right, because that case is the reconnect's to catch.
sidecar.onClaudeSidecarStarted(() => {
  for (const session of sessionManager.getAllSessions()) {
    if (!session.isGenerating) continue;
    // No owner recorded, no question asked. `isClaudeGenerating` is now scoped to a caller, and there is no
    // safe id to substitute — asking as the owner would let a member's orphaned session be answered with the
    // owner's authority, and asking as nobody is not a thing. Leaving it marked generating is the same
    // outcome as before this loop existed, and it self-corrects on the next reconnect.
    if (session.userId === undefined) continue;
    void endTurnIfAgentIsGone(
      session.sockets as Set<ServerWebSocket<WSData>>,
      session.sessionId,
      session.model,
      session.userId,
    );
  }
});

export const chatWebsocket = {
  open,
  message,
  close,
  drain() {},
};