platform/src/servers/sidecar/opencode/runner.ts

import { existsSync } from 'node:fs';
import type { Subprocess } from 'bun';
import type { ChatEvent, MessageCost } from '../../api/chat/types';
import type { OpenCodeRunParams } from '../protocol';

// Drives one chat turn by spawning `opencode run … --format json` and mapping its newline-delimited
// JSON events to the shared ChatEvent contract. This is the reliable path: `--dir <cwd>` hard-anchors
// every tool to the chat's working directory, and `run` (unlike serve + POST /message) reports tool
// completion faithfully and exits when the turn is done — no wedged "running" tools. `--dangerously-
// skip-permissions` auto-approves so tools never block on an approval we can't answer over this channel.
//
// A watchdog guards against a genuinely hung child (an interactive prompt, a `sleep`, a network stall):
// an inactivity timer (reset on every stdout chunk) and an absolute per-turn ceiling both kill the
// process and emit a clean `error`, so the UI never sits at "Working…" forever.

const INACTIVITY_MS = 120_000; // no stdout for this long → assume wedged, kill
const HARD_CAP_MS = 10 * 60_000; // absolute per-turn ceiling

export type RunnerConfig = {
  bin: string; // absolute path to the opencode binary
  fallbackCwd: string; // used when params.cwd is missing/nonexistent
};

// What a turn reports to the sidecar it runs in. `opencode:event` is deliberately not a wire event any
// more: the sidecar translates each one into a TurnMessage and commits it before officer sees anything,
// so the durable record does not depend on officer being up (see index.ts).
export type RunnerMessage =
  | { type: 'opencode:event'; sessionKey: string; event: ChatEvent }
  | { type: 'opencode:session'; sessionKey: string; sessionId: string };

type Emit = (msg: RunnerMessage) => void;

type RunHandle = {
  proc: Subprocess;
  killedByUser: boolean;
  /**
   * Set when a newer turn has taken this sessionKey over.
   *
   * A killed process dies asynchronously, so a replaced turn's `proc.exited` fires LONG after its
   * replacement is already running and registered under the same key. Without this flag that late
   * handler ran the full completion path against the wrong turn: it emitted `OpenCode exited with code
   * 143` — which the sidecar commits to `chat_session_events`, so a false failure became permanent
   * history — and then deleted its replacement from `running`, which blinded the Live panel, made the
   * stop button a no-op, and orphaned a process nothing could reach.
   */
  superseded: boolean;
  /**
   * End this turn from outside the closure that owns it, with a reason.
   *
   * `killOpenCodeTurn` can kill a process and let `proc.exited` do the rest, because it has time.
   * Shutdown does not: the sidecar is about to call `process.exit`, so nothing asynchronous will ever
   * run again and a turn killed that way would simply stop mid-sentence, leaving a transcript that
   * trails off. Settling synchronously is what puts the explanation in the log before we go.
   */
  finish: (event: ChatEvent) => void;
};

// One turn per sessionKey; a new turn supersedes any stale process for that key.
const running = new Map<string, RunHandle>();

// Shape of `opencode run --format json` events.
//
// Verified live against opencode 1.18.11 (this Mac) and 1.17.9 (alpha) — MEASURED on 2026-08-10, having
// previously been recorded the other way round here: the "this server" in the original note meant alpha,
// and the comment was copied to a machine where it was false. Nothing enforces a version anyway; the
// binary is whatever is installed, and the two machines in this project already differ.
//
// `runner.test.ts` pins the mapping so a shape change fails a test rather than a turn; if it starts
// failing, re-read the NDJSON from the installed binary before editing the test.
type RunPart = {
  type?: string;
  text?: string;
  tool?: string;
  callID?: string;
  state?: { status?: string; input?: unknown; output?: unknown; error?: unknown };
  tokens?: { input?: number; output?: number };
  cost?: number;
};
type RunEvent = { type?: string; sessionID?: string; part?: RunPart };

export function runOpenCodeTurn(params: OpenCodeRunParams, config: RunnerConfig, emit: Emit): void {
  const { sessionKey } = params;

  // Supersede any lingering turn for this session. Mark it BEFORE killing: the flag is what tells its
  // own exit handler that this death was intentional and belongs to nobody.
  const stale = running.get(sessionKey);
  if (stale) {
    stale.superseded = true;
    try {
      stale.proc.kill();
    } catch {
      /* already gone */
    }
    running.delete(sessionKey);
  }

  const args = ['run', '--format', 'json', '--dangerously-skip-permissions'];
  if (params.cwd) args.push('--dir', params.cwd);
  if (params.model) args.push('--model', params.model);
  if (params.resumeSessionId) args.push('--session', params.resumeSessionId);
  args.push(params.prompt);

  const cwd = params.cwd && existsSync(params.cwd) ? params.cwd : config.fallbackCwd;

  const proc = Bun.spawn([config.bin, ...args], {
    cwd,
    stdin: 'ignore', // == /dev/null: `run` hangs waiting on stdin otherwise
    stdout: 'pipe',
    stderr: 'pipe',
  });

  // `finish` is a placeholder for the few synchronous lines until the real one below exists — it closes
  // over `handle`, so the two cannot both be defined first. Nothing can call it in between.
  const handle: RunHandle = { proc, killedByUser: false, superseded: false, finish: () => {} };
  running.set(sessionKey, handle);

  let done = false;
  let reportedSession = false;
  let cost: MessageCost = { inputTokens: 0, outputTokens: 0, totalUSD: 0 };
  let stderrTail = '';

  const emitEvent = (event: ChatEvent) => emit({ type: 'opencode:event', sessionKey, event });

  let inactivityTimer: ReturnType<typeof setTimeout> | undefined;

  /**
   * Retire this turn: stop its watchdogs, release its slot, and optionally say why it ended.
   *
   * The delete is identity-checked because `sessionKey` is not this turn's to own once it has been
   * superseded — the map may already hold a live replacement under that key, and deleting by name alone
   * removed it. `null` retires silently, which is what a superseded turn needs: it must still clear its
   * timers (an armed 10-minute `hardTimer` would otherwise fire an error at whichever turn holds the key
   * by then, reproducing the same cross-talk on a delay) while emitting nothing at all.
   */
  const settle = (event: ChatEvent | null) => {
    if (done) return;
    done = true;
    clearTimeout(hardTimer);
    if (inactivityTimer) clearTimeout(inactivityTimer);
    if (running.get(sessionKey) === handle) running.delete(sessionKey);
    if (event) emitEvent(event);
  };

  const finish = (event: ChatEvent) => settle(event);
  handle.finish = finish;

  // ── Watchdogs ──
  const hardTimer = setTimeout(() => {
    try {
      proc.kill();
    } catch {
      /* already gone */
    }
    finish({ type: 'error', message: `OpenCode turn exceeded ${HARD_CAP_MS / 1000}s and was stopped` });
  }, HARD_CAP_MS);

  const bumpInactivity = () => {
    if (done) return;
    if (inactivityTimer) clearTimeout(inactivityTimer);
    inactivityTimer = setTimeout(() => {
      try {
        proc.kill();
      } catch {
        /* already gone */
      }
      finish({
        type: 'error',
        message: `OpenCode turn stalled (no output for ${INACTIVITY_MS / 1000}s) and was stopped`,
      });
    }, INACTIVITY_MS);
  };
  bumpInactivity();

  // ── Capture a tail of stderr for error reporting ──
  void (async () => {
    const dec = new TextDecoder();
    try {
      for await (const chunk of proc.stderr as unknown as AsyncIterable<Uint8Array>) {
        stderrTail = (stderrTail + dec.decode(chunk, { stream: true })).slice(-2000);
      }
    } catch {
      /* stream closed */
    }
  })();

  // ── Parse stdout: newline-delimited JSON events ──
  void (async () => {
    const dec = new TextDecoder();
    let buf = '';
    try {
      for await (const chunk of proc.stdout as unknown as AsyncIterable<Uint8Array>) {
        bumpInactivity();
        buf += dec.decode(chunk, { stream: true });
        let nl: number;
        while ((nl = buf.indexOf('\n')) >= 0) {
          const line = buf.slice(0, nl).trim();
          buf = buf.slice(nl + 1);
          if (line) handleLine(line);
        }
      }
      const last = buf.trim();
      if (last) handleLine(last);
    } catch {
      /* stream closed / process killed */
    }
  })();

  function handleLine(line: string): void {
    // A retired turn says nothing more. Stdout is drained asynchronously, so a killed process can still
    // have buffered lines in flight — and for a superseded turn those would be emitted under a
    // sessionKey that now belongs to its replacement, interleaving one turn's output into another's.
    if (done) return;

    const mapped = mapRunLine(line);
    if (!mapped) return;

    // Report the OpenCode session id once, so the API can resume it (`--session`) next turn.
    if (!reportedSession && mapped.sessionId) {
      reportedSession = true;
      emit({ type: 'opencode:session', sessionKey, sessionId: mapped.sessionId });
    }

    cost = {
      inputTokens: cost.inputTokens + mapped.costDelta.inputTokens,
      outputTokens: cost.outputTokens + mapped.costDelta.outputTokens,
      totalUSD: cost.totalUSD + mapped.costDelta.totalUSD,
    };

    for (const event of mapped.events) emitEvent(event);
  }

  // ── Completion: process exit is the authoritative turn-done signal ──
  void proc.exited.then((code) => {
    if (done) return;
    // Replaced on purpose: not a result, not an error, and not this turn's session any more.
    if (handle.superseded) {
      settle(null);
      return;
    }
    if (handle.killedByUser) {
      finish({ type: 'stopped' });
      return;
    }
    if (code === 0) {
      finish({ type: 'result', cost });
      return;
    }
    const tail = stderrTail.trim();
    finish({
      type: 'error',
      message: tail ? `OpenCode exited (${code}): ${tail.slice(-500)}` : `OpenCode exited with code ${code}`,
    });
  });
}

/**
 * The turns this process is running right now.
 *
 * The OpenCode analog of `claude-manager.listSessions`, and deliberately thinner. Claude holds a warm
 * session that outlives a turn, so it can report one that is merely open; OpenCode spawns a subprocess
 * per turn and has nothing between them. So a session appears here only while it is generating — which
 * is exactly the state the Live panel exists to show, and the state that was invisible for OpenCode.
 *
 * No `pendingTasks`: `opencode run` has no background-task concept, so reporting 0 would suggest a
 * capability that does not exist rather than an empty one.
 */
export function listRunningOpenCodeTurns(): { sessionKey: string }[] {
  return Array.from(running.keys()).map((sessionKey) => ({ sessionKey }));
}

/**
 * Kill every turn this process is running, because the process itself is going away.
 *
 * A turn is a child of this sidecar only in the bookkeeping sense: `opencode run` is spawned, not
 * supervised, so `pm2 restart officer-opencode` used to leave every in-flight turn ALIVE — reparented,
 * still spending tokens, and still writing files as the agent, while the only reader of its stdout had
 * exited. The turn's output went nowhere and the transcript simply stopped mid-tool-call, which is
 * indistinguishable from the agent hanging.
 *
 * Both halves matter. Killing the children stops the invisible work; settling them synchronously writes
 * a reason into the transcript, so a reload after a restart explains itself instead of trailing off.
 * Returns how many were stopped, so the caller can skip the flush wait when there were none.
 */
export function stopAllOpenCodeTurns(message: string): number {
  const handles = [...running.values()];
  for (const handle of handles) {
    // Suppress the exit handler's own error: this death is accounted for, and `finish` below is the
    // account. Without it a late `proc.exited` would be a second, less accurate ending.
    handle.killedByUser = true;
    try {
      handle.proc.kill();
    } catch {
      /* already gone */
    }
    handle.finish({ type: 'error', message });
  }
  return handles.length;
}

export function killOpenCodeTurn(sessionKey: string): void {
  const handle = running.get(sessionKey);
  if (!handle) return;
  handle.killedByUser = true;
  try {
    handle.proc.kill();
  } catch {
    /* already gone */
  }
  // proc.exited fires → finish({ type: 'stopped' }).
}

// ── The NDJSON → ChatEvent mapping, as a pure function ──
//
// Extracted from `handleLine` so it can be tested without spawning a binary. This is the piece most
// likely to break against a new OpenCode release — the event shape is not a stable contract and this
// project already runs two different versions across two machines — and it was the only untested part
// of the path. `runner.test.ts` pins it.
//
// Pure by construction: it takes a line and returns what should happen, holding no state. The caller
// owns the two stateful concerns, because they span lines rather than belonging to one: emitting the
// session id exactly once, and accumulating cost across steps.

export type MappedRunLine = {
  /** Present on any event that names a session; the caller emits it only the first time. */
  sessionId?: string;
  /** Events to forward, in order. A tool part yields `tool:start` then `tool:result`. */
  events: ChatEvent[];
  /** Per-step tokens/cost to add to the turn total. Zeroes for every non-`step_finish` line. */
  costDelta: { inputTokens: number; outputTokens: number; totalUSD: number };
};

const NO_COST = { inputTokens: 0, outputTokens: 0, totalUSD: 0 };

/** `null` for a line that is not JSON at all — `opencode` interleaves plain log lines with the stream. */
export function mapRunLine(line: string): MappedRunLine | null {
  let evt: RunEvent;
  try {
    evt = JSON.parse(line) as RunEvent;
  } catch {
    return null; // non-JSON log line
  }

  const sessionId = evt.sessionID;
  const events: ChatEvent[] = [];

  switch (evt.type) {
    case 'text': {
      // `run` emits complete text parts (not token deltas) — forward each as final text.
      const text = evt.part?.text;
      if (typeof text === 'string' && text.length > 0) events.push({ type: 'text', text });
      return { sessionId, events, costDelta: NO_COST };
    }
    case 'tool_use': {
      // In `run`, the tool part arrives already resolved (status + output). Emit start then result.
      const part = evt.part;
      if (!part || part.type !== 'tool' || !part.callID) return { sessionId, events, costDelta: NO_COST };
      const st = part.state ?? {};
      events.push({
        type: 'tool:start',
        toolCallId: part.callID,
        toolName: part.tool ?? 'tool',
        toolInput: (st.input as Record<string, unknown>) ?? {},
      });
      const isError = st.status === 'error';
      events.push({
        type: 'tool:result',
        toolCallId: part.callID,
        output: String((isError ? st.error : st.output) ?? ''),
        isError,
      });
      return { sessionId, events, costDelta: NO_COST };
    }
    case 'step_finish': {
      const t = evt.part?.tokens;
      return {
        sessionId,
        events,
        costDelta: {
          inputTokens: t?.input ?? 0,
          outputTokens: t?.output ?? 0,
          totalUSD: typeof evt.part?.cost === 'number' ? evt.part.cost : 0,
        },
      };
    }
    default:
      return { sessionId, events, costDelta: NO_COST }; // step_start etc. — nothing to forward
  }
}