platform/src/servers/sidecar/claude/claude-manager.ts

import { existsSync } from 'node:fs';
import { homedir } from 'node:os';
import { join } from 'node:path';
import { query, type Query } from '@anthropic-ai/claude-agent-sdk';
import type { ChatEvent, PromptImage } from '../../api/chat/types';
import type { ClaudeSpawnParams, ClaudeSpawnStreamingParams, ClaudeCodeResult } from '../protocol';
import { setClaudeSession, clearClaudeSession, getClaudeSession } from './state';
import { createParseState, processMessage } from './stream-parser';

const SEND_TIMEOUT_MS = 30 * 60 * 1000;

// Where the Claude Code CLI lives. This was hardcoded to /usr/local/bin/claude, which dated from the
// bwrap-sandboxed architecture: the jail ro-bound /usr and saw nothing else, so the installer's real
// target (~/.local/bin/claude) had to be symlinked into a path the sandbox could reach. That sandbox
// is gone, and the hardcoded path made the sidecar unrunnable anywhere it does not exist — a stock
// macOS host has no /usr/local/bin at all.
//
// Resolution order mirrors OPENCODE_BIN in the opencode sidecar: an explicit pin, then PATH, then the
// locations Anthropic's installer actually writes to.
function resolveClaudeBin(): string {
  const pinned = process.env.CLAUDE_BIN;
  if (pinned) return pinned;

  const onPath = Bun.which('claude');
  if (onPath) return onPath;

  const candidates = [join(homedir(), '.local', 'bin', 'claude'), '/usr/local/bin/claude', '/opt/homebrew/bin/claude'];
  return candidates.find((candidate) => existsSync(candidate)) ?? 'claude';
}

const CLAUDE_BIN = resolveClaudeBin();
console.log(`[claude] CLI resolved to ${CLAUDE_BIN}`);

// Capture original HOME before user-instance overrides it
const HOST_HOME = process.env.HOME!;
const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data');

// Tear a persistent session down after this long with no new turn (see PersistentSession below).
const IDLE_TIMEOUT_MS = 30 * 60 * 1000;

// MCP config paths, set by user-instance at startup
let mcpHostPath: string | undefined; // path on the host filesystem

export function setMcpConfigPath(hostPath: string): void {
  mcpHostPath = hostPath;
}

// ── Blocking send (one-shot; used by the non-streaming 'claude:spawn' command) ──

type ClaudeCodeOutput = {
  result: string;
  session_id: string;
  cost_usd: number;
  input_tokens: number;
  output_tokens: number;
  is_error: boolean;
};

export async function spawnClaude(params: ClaudeSpawnParams): Promise<ClaudeCodeResult> {
  const { prompt, sessionKey } = params;

  const existingSession = getClaudeSession(sessionKey);

  const claudeArgs = [CLAUDE_BIN, '-p', prompt, '--dangerously-skip-permissions', '--output-format', 'json'];

  const mcpConfig = mcpHostPath;
  if (mcpConfig) claudeArgs.push('--mcp-config', mcpConfig);

  const subModel = params.model?.split('/')[1];
  if (subModel) claudeArgs.push('--model', subModel);

  if (existingSession) {
    claudeArgs.push('--resume', existingSession);
  }

  const spawnCmd = claudeArgs;
  const spawnCwd = params.cwd ?? HOST_HOME;

  const proc = Bun.spawn(spawnCmd, {
    stdin: 'pipe',
    stdout: 'pipe',
    stderr: 'pipe',
    cwd: spawnCwd,
    env: process.env as Record<string, string>,
  });

  const timeout = setTimeout(() => {
    try {
      proc.kill();
    } catch {
      /* already dead */
    }
  }, SEND_TIMEOUT_MS);

  try {
    const [stdout, stderr] = await Promise.all([new Response(proc.stdout).text(), new Response(proc.stderr).text()]);
    const exitCode = await proc.exited;
    clearTimeout(timeout);

    if (exitCode !== 0 && !stdout.trim()) {
      throw new Error(`claude exited with code ${exitCode}: ${stderr.trim().slice(0, 200)}`);
    }

    let output: ClaudeCodeOutput;
    try {
      output = JSON.parse(stdout) as ClaudeCodeOutput;
    } catch {
      return {
        text: stdout.trim() || '(no response)',
        sessionId: sessionKey,
        model: 'claude-code',
        cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 },
      };
    }

    if (output.is_error) {
      throw new Error(output.result || 'Claude Code returned an error');
    }

    if (output.session_id) {
      setClaudeSession(sessionKey, output.session_id);
    }

    return {
      text: output.result || '(no response)',
      sessionId: sessionKey,
      model: 'claude-code',
      cost: {
        inputTokens: output.input_tokens ?? 0,
        outputTokens: output.output_tokens ?? 0,
        totalUSD: output.cost_usd ?? 0,
      },
    };
  } catch (err) {
    clearTimeout(timeout);
    throw err;
  }
}

// ── Persistent streaming sessions (Agent SDK) ──
//
// Each sessionKey gets ONE long-lived `query()` driven by a streaming-input queue. The process stays
// alive BETWEEN turns, so background work (Bash run_in_background, Monitor) and its `task_notification`
// return to a live harness instead of being orphaned when the turn ends. A turn = one user message
// pushed onto the input queue; the single consumer loop maps every SDK message (assistant text/tools,
// tool results, turn `result`, and — crucially — post-turn `task_started`/`task_notification`) to a
// ChatEvent and forwards it. Idle policy: torn down after IDLE_TIMEOUT_MS with no new turn, or on an
// explicit kill (user "disconnect"). "Stop" is interrupt() — it ends the turn but keeps the session.

/**
 * `content` is an Anthropic `MessageParam` content: a bare string, or the block array we build when the
 * turn carries images. Blocks are what make an attached screenshot actually reach the model — the string
 * form can only carry the client's `[Attached image: …]` placeholder, which is a label, not the picture.
 */
type ContentBlock =
  | { type: 'text'; text: string }
  | { type: 'image'; source: { type: 'base64'; media_type: string; data: string } };

type SdkUserMessage = {
  type: 'user';
  message: { role: 'user'; content: string | ContentBlock[] };
  parent_tool_use_id: null;
  session_id: string;
};

type PersistentSession = {
  sessionKey: string;
  query: Query;
  pushTurn: (prompt: string, images?: PromptImage[]) => void;
  closeInput: () => void;
  abort: AbortController;
  emit: (event: ChatEvent) => void;
  isGenerating: boolean;
  pendingTasks: Set<string>; // background tasks started but not yet notified; suppress idle-GC while non-empty
  /**
   * The user pressed stop and we are waiting for the turn to fall over. The SDK reports an interrupt as
   * an ordinary failed `result` — `is_error` with no text — which is indistinguishable downstream from
   * the harness actually breaking, and reached the user as "Claude Code returned an error". Only the
   * side that called `interrupt()` knows better, so it says so here.
   */
  interrupted: boolean;
  /**
   * When the `PreCompact` hook fired, so the `compact_boundary` that closes it can carry how long the
   * silence lasted. The harness reports the boundary but not the duration, and the duration is the part
   * that explains the wait.
   */
  compactStartedAt?: number;
  idleTimer?: ReturnType<typeof setTimeout>;
  /**
   * Fires when a turn that claims to be generating has emitted nothing for too long.
   *
   * The idle timer above answers the opposite question — how long a session with NO turn in flight may
   * sit before it is collected — so neither one covers a turn that is wedged. Nothing did: a turn could
   * stop producing events and stay `isGenerating` forever, and every client showed a spinner with no
   * timeout of its own. On 2026-08-08 that ran for seventeen minutes inside a compaction and was
   * indistinguishable, from the phone, from a dead chat.
   */
  stallTimer?: ReturnType<typeof setTimeout>;
};

/*
 How long a generating turn may say nothing before we call it stalled.

 Generous on purpose, because the legitimate silences here are long: compaction narrates nothing for
 as long as it takes (minutes on a large conversation), and a single deep tool call can be quiet for a
 while too. This is a backstop against turns that will never speak again, not a latency budget — too
 tight and it would kill work that was about to succeed, which is worse than the hang it prevents.
*/
const STALL_TIMEOUT_MS = 10 * 60 * 1000;
/** Compaction gets longer still: it is the known-slowest silent phase, and the one that stalled. */
const COMPACT_STALL_TIMEOUT_MS = 20 * 60 * 1000;

const sessions = new Map<string, PersistentSession>();

/** A hand-rolled async iterable we can push turns onto and close on teardown. */
function makeInputQueue() {
  const buf: SdkUserMessage[] = [];
  let wake: (() => void) | null = null;
  let closed = false;
  async function* gen(): AsyncGenerator<SdkUserMessage> {
    while (true) {
      if (buf.length) {
        yield buf.shift()!;
        continue;
      }
      if (closed) return;
      await new Promise<void>((r) => {
        wake = r;
      });
    }
  }
  return {
    gen: gen(),
    push(m: SdkUserMessage) {
      buf.push(m);
      wake?.();
      wake = null;
    },
    close() {
      closed = true;
      wake?.();
      wake = null;
    },
  };
}

/**
 * (Re)arm the stall watchdog. Called on every emitted event, so any sign of life pushes it back.
 *
 * On expiry it ends the turn the same way a real failure would — `isGenerating` off, idle re-armed,
 * and an `error` the client can render — rather than tearing the session down. The agent process is
 * left alive deliberately: it may still be working, and the next turn resumes it. What this guarantees
 * is that the CLIENT is told, which is the part that was missing.
 */
function armStall(session: PersistentSession): void {
  if (session.stallTimer) clearTimeout(session.stallTimer);
  if (!session.isGenerating) {
    session.stallTimer = undefined;
    return;
  }
  const compacting = session.compactStartedAt !== undefined;
  session.stallTimer = setTimeout(
    () => {
      if (!session.isGenerating) return;
      const waited = Math.round((compacting ? COMPACT_STALL_TIMEOUT_MS : STALL_TIMEOUT_MS) / 60000);
      console.error(
        `[claude:stall:${session.sessionKey}] no events for ${waited}m${compacting ? ' (compacting)' : ''} — ending the turn`,
      );
      session.isGenerating = false;
      session.compactStartedAt = undefined;
      session.interrupted = false;
      if (session.pendingTasks.size === 0) armIdle(session);
      session.emit({
        type: 'error',
        message: compacting
          ? `Compaction has produced nothing for ${waited} minutes and has been given up on. Your conversation is safe — send again to continue.`
          : `The agent has produced nothing for ${waited} minutes and has been given up on. Your conversation is safe — send again to continue.`,
      });
    },
    compacting ? COMPACT_STALL_TIMEOUT_MS : STALL_TIMEOUT_MS,
  );
}

function armIdle(session: PersistentSession): void {
  if (session.idleTimer) clearTimeout(session.idleTimer);
  session.idleTimer = setTimeout(() => {
    // Heartbeat: never GC a session that's mid-turn or still has background tasks running — a long
    // silent run_in_background job would otherwise be killed along with its pending task_notification.
    if (session.isGenerating || session.pendingTasks.size > 0) {
      armIdle(session);
      return;
    }
    killClaudeSession(session.sessionKey);
  }, IDLE_TIMEOUT_MS);
}

function createSession(params: ClaudeSpawnStreamingParams, onEvent: (event: ChatEvent) => void): PersistentSession {
  const { sessionKey } = params;
  const input = makeInputQueue();
  const abort = new AbortController();

  const session: PersistentSession = {
    sessionKey,
    query: undefined as unknown as Query,
    pushTurn: () => {},
    closeInput: () => input.close(),
    abort,
    emit: onEvent,
    isGenerating: false,
    pendingTasks: new Set<string>(),
    interrupted: false,
  };

  // Strip the nested-session guard vars so the SDK can spawn `claude` (mirrors the old spawn env clean).
  const { CLAUDECODE: _c, CLAUDE_CODE_ENTRYPOINT: _e, CLAUDE_CODE_SSE_PORT: _s, ...cleanEnv } = process.env;

  const resumeId = getClaudeSession(sessionKey) ?? params.resumeSessionId;
  const subModel = params.model?.split('/')[1];

  const q = query({
    prompt: input.gen as AsyncIterable<SdkUserMessage>,
    options: {
      cwd: params.cwd ?? HOST_HOME,
      permissionMode: 'bypassPermissions',
      allowDangerouslySkipPermissions: true,
      includePartialMessages: true,
      // The only warning that compaction is about to happen. Everything else the harness does narrates
      // itself through the message stream; compaction goes silent for as long as it takes — 2.5 minutes
      // in the worst case on disk here — and the stream resumes with no explanation of the gap. The hook
      // returns immediately and never throws: it is a notification, and it must not be able to stall or
      // fail the compaction it is announcing.
      hooks: {
        PreCompact: [
          {
            hooks: [
              async (input) => {
                session.compactStartedAt = Date.now();
                const trigger = 'trigger' in input && input.trigger === 'manual' ? 'manual' : 'auto';
                session.emit({ type: 'compact:start', trigger });
                // Re-arm on the compaction budget: this hook fires as the long silence BEGINS, so the
                // deadline the turn is holding was sized for ordinary work and is about to be wrong.
                armStall(session);
                return { continue: true };
              },
            ],
          },
        ],
      },
      abortController: abort,
      pathToClaudeCodeExecutable: CLAUDE_BIN,
      settingSources: ['user', 'project', 'local'],
      env: cleanEnv as Record<string, string>,
      stderr: (d: string) => {
        if (d.trim()) console.error(`[claude:stream:${sessionKey}] ${d.slice(0, 300)}`);
      },
      ...(subModel ? { model: subModel } : {}),
      ...(resumeId ? { resume: resumeId } : {}),
      ...(mcpHostPath ? { extraArgs: { 'mcp-config': mcpHostPath } } : {}),
    },
  });

  session.query = q;
  session.pushTurn = (prompt: string, images?: PromptImage[]) => {
    if (session.idleTimer) clearTimeout(session.idleTimer);
    session.isGenerating = true;
    // A turn that dies before its FIRST event is the case the emit-path arming cannot reach.
    armStall(session);
    // Images first, then the text: the model reads what it is looking at before what to do about it.
    const content: string | ContentBlock[] = images?.length
      ? [
          ...images.map(
            (img): ContentBlock => ({
              type: 'image',
              source: { type: 'base64', media_type: img.mediaType, data: img.data },
            }),
          ),
          { type: 'text', text: prompt },
        ]
      : prompt;
    input.push({
      type: 'user',
      message: { role: 'user', content },
      parent_tool_use_id: null,
      session_id: sessionKey,
    });
  };

  sessions.set(sessionKey, session);

  // Single consumer loop for the session's whole life. Turn-end (`result`) and errors flip isGenerating
  // and (re)arm the idle timer; the session process stays alive so later task_notifications still flow.
  void (async () => {
    const state = createParseState();
    const emit = (raw: ChatEvent) => {
      // A turn we interrupted ends in a failed `result`. That is the stop landing, not a fault.
      const event: ChatEvent =
        raw.type === 'error' && session.interrupted
          ? { type: 'stopped' }
          : // The boundary knows what it dropped; only this side knows how long it took, because the start
            // came from a hook rather than from the stream.
            raw.type === 'compact:done' && session.compactStartedAt
            ? { ...raw, durationMs: Date.now() - session.compactStartedAt }
            : raw;
      if (event.type === 'compact:done') session.compactStartedAt = undefined;
      if (event.type === 'task:started') {
        // Work is running — hold off idle-GC until it finishes.
        session.pendingTasks.add(event.taskId);
        if (session.idleTimer) {
          clearTimeout(session.idleTimer);
          session.idleTimer = undefined;
        }
      } else if (event.type === 'task:notification') {
        session.pendingTasks.delete(event.taskId);
        if (!session.isGenerating && session.pendingTasks.size === 0) armIdle(session);
      } else if (event.type === 'result' || event.type === 'error' || event.type === 'stopped') {
        session.isGenerating = false;
        // Whatever ended the turn, the interrupt is spent — a later genuine error must not wear it.
        session.interrupted = false;
        if (session.pendingTasks.size === 0) armIdle(session);
      }
      // Any event at all is a sign of life, so push the stall deadline back. Placed after the branch
      // above so a terminal event disarms rather than re-arms it.
      armStall(session);
      session.emit(event);
    };
    try {
      for await (const msg of q as AsyncGenerator<Record<string, unknown>>) {
        processMessage(msg, state, {
          onEvent: emit,
          onSessionId: (id: string) => setClaudeSession(sessionKey, id),
        });
      }
    } catch (err) {
      if (!abort.signal.aborted) {
        session.emit({ type: 'error', message: err instanceof Error ? err.message : String(err) });
      }
    } finally {
      if (session.idleTimer) clearTimeout(session.idleTimer);
      if (session.stallTimer) clearTimeout(session.stallTimer);
      sessions.delete(sessionKey);
    }
  })();

  return session;
}

/** Start a turn: create the persistent session if needed, then push the prompt as a user message. */
export async function spawnClaudeStreaming(
  params: ClaudeSpawnStreamingParams,
  onEvent: (event: ChatEvent) => void,
): Promise<void> {
  let session = sessions.get(params.sessionKey);
  if (session) {
    session.emit = onEvent; // adopt the latest emitter (equivalent across turns; keeps events flowing)
  } else {
    session = createSession(params, onEvent);
  }
  session.pushTurn(params.prompt, params.images);
}

/** Interrupt the current turn but KEEP the session alive (the "stop" button). */
export async function interruptClaudeSession(sessionKey: string): Promise<boolean> {
  const session = sessions.get(sessionKey);
  if (!session) return false;
  // Set before the await: the failed `result` can arrive while interrupt() is still resolving, and the
  // consumer loop reads this flag to tell a stop from a fault.
  session.interrupted = true;
  try {
    await session.query.interrupt();
    session.isGenerating = false;
    if (session.stallTimer) clearTimeout(session.stallTimer);
    session.stallTimer = undefined;
    return true;
  } catch {
    session.interrupted = false;
    return false;
  }
}

/** Fully tear the session down (the "disconnect" action / idle GC): abort the query + close input. */
export function killClaudeSession(sessionKey: string): boolean {
  const session = sessions.get(sessionKey);
  if (!session) return false;
  if (session.idleTimer) clearTimeout(session.idleTimer);
  if (session.stallTimer) clearTimeout(session.stallTimer);
  try {
    session.abort.abort();
  } catch {
    /* already aborted */
  }
  try {
    session.closeInput();
  } catch {
    /* noop */
  }
  sessions.delete(sessionKey);
  return true;
}

export function clearSession(sessionKey: string): void {
  clearClaudeSession(sessionKey);
}

export function getActiveSessionKeys(): string[] {
  return Array.from(sessions.keys());
}

/**
 * Is a turn actually in flight for this session, right now, in this process?
 *
 * This is the ground truth a reconnecting browser has no way to work out for itself. Officer's own view
 * dies with `pm2 restart officer` while the turn keeps running here, so "officer doesn't remember" means
 * nothing — and if *this* process was the one that restarted, the session is simply absent and the turn
 * it was running is gone, however alive the client still believes it to be.
 */
export function isSessionGenerating(sessionKey: string): boolean {
  return sessions.get(sessionKey)?.isGenerating ?? false;
}