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

import type { ChatEvent, MessageCost } from '../../api/chat/types';
import type { OpenCodeRunParams } from '../protocol';
import { mapServeEvent } from './serve-events';

// How an OpenCode turn runs. The only way, since 2026-08-10.
//
// It used to be `opencode run --format json`, a subprocess per turn with `stdin: 'ignore'`. Everything
// that path could not do followed from that one closed pipe: no token streaming, no mid-turn injection,
// no queue, and a stop that could only kill the session rather than interrupt it. The serve offers all
// four as primitives, each verified end to end before the subprocess was deleted
// (docs/opencode-fork-decision.md, docs/opencode-serve-migration-plan.md).
//
// There is no fallback engine any more. If this path breaks, the recovery is git, not a config flag —
// a deliberate choice made while nothing depended on OpenCode.
//
// ── One global stream, demultiplexed ──
//
// The serve publishes each turn on two streams (docs/opencode-serve-migration-plan.md). This reads the
// LIVE one, `GET /api/event`, because it is a strict superset of the durable stream's content — same
// `tool.called`, `tool.success`, `step.ended`, `text.ended`, PLUS the deltas — and deltas are the point.
//
// It is GLOBAL: one socket carries every session on the box, so everything here filters on `sessionID`.
// Forgetting that would splice one conversation into another. There is exactly one subscription for the
// process, opened on the first turn and shared, because opening one per turn would multiply the same
// firehose by the number of turns.
//
// What this loses versus the durable stream is `durable.seq`, i.e. replay-after-the-fact. That matters
// for surviving an officer restart mid-turn and is deliberately NOT in Phase B: the sidecar commits
// every event to `chat_session_events` as it arrives (unchanged from the subprocess path), which is the
// same durability guarantee the subprocess had. Reading the durable stream to recover a turn this
// process never saw is its own change.

type ServeConfig = {
  /** The serve's base URL, e.g. http://127.0.0.1:53100 */
  baseUrl: string;
  /** Used when a turn names no cwd — same fallback the subprocess runner applies. */
  fallbackCwd: string;
};

/**
 * What a turn reports to the sidecar it runs in.
 *
 * Lived in `runner.ts` until the subprocess path was deleted (Phase D). `opencode:event` is not a wire
 * event: the sidecar translates each one into a TurnMessage and commits it to `chat_session_events`
 * before officer sees anything, so the durable record does not depend on officer being up.
 * `opencode:session` is the routing fact — which `ses_…` to resume — and goes over the wire live.
 */
export type RunnerMessage =
  | { type: 'opencode:event'; sessionKey: string; event: ChatEvent }
  | { type: 'opencode:session'; sessionKey: string; sessionId: string };

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

type ServeTurn = {
  sessionKey: string;
  openCodeSessionId: string;
  cost: MessageCost;
  done: boolean;
  emit: Emit;
  finish: (event: ChatEvent) => void;
};

/** Live turns, by OpenCode session id — the id the event stream speaks. */
const byOpenCodeId = new Map<string, ServeTurn>();
/** The same turns by officer's key, which is what `kill` and the Live panel use. */
const bySessionKey = new Map<string, ServeTurn>();

// ── The one shared subscription ──

let streamStarted = false;

/**
 * Read `/api/event` forever, routing each event to the turn that owns it.
 *
 * Reconnects on drop with a fixed delay. A serve restart, a network blip or the stream simply ending
 * must not permanently deafen the sidecar — every subsequent turn would hang with no output, which is
 * the worst failure this path has, because it looks exactly like a slow model.
 */
function ensureEventStream(config: ServeConfig): void {
  if (streamStarted) return;
  streamStarted = true;

  void (async () => {
    for (;;) {
      try {
        const res = await fetch(`${config.baseUrl}/api/event`, { headers: { accept: 'text/event-stream' } });
        if (!res.ok || !res.body) throw new Error(`event stream → ${res.status}`);

        const reader = res.body.getReader();
        const decoder = new TextDecoder();
        let buffer = '';

        for (;;) {
          const { done, value } = await reader.read();
          if (done) break;
          buffer += decoder.decode(value, { stream: true });
          const lines = buffer.split('\n');
          buffer = lines.pop() ?? '';
          for (const line of lines) {
            if (!line.startsWith('data:')) continue;
            try {
              handleServeEvent(JSON.parse(line.slice(5).trim()));
            } catch {
              /* a frame we could not parse is not worth killing the stream over */
            }
          }
        }
      } catch (err) {
        console.error('[opencode] event stream dropped, reconnecting:', err instanceof Error ? err.message : err);
      }
      await Bun.sleep(1_000);
    }
  })();
}

function handleServeEvent(raw: unknown): void {
  const mapped = mapServeEvent(raw as never);
  if (!mapped?.sessionId) return;

  const turn = byOpenCodeId.get(mapped.sessionId);
  if (!turn || turn.done) return; // another session's, or one we have already finished

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

  for (const event of mapped.events) {
    // An error from the harness ends the turn: nothing follows a failed step, and leaving the turn open
    // would strand the UI on a spinner.
    if (event.type === 'error') {
      turn.finish(event);
      return;
    }
    turn.emit({ type: 'opencode:event', sessionKey: turn.sessionKey, event });
  }

  // `tool-calls` means the model paused to run a tool and will continue. Anything else is the end.
  if (mapped.stepFinish && mapped.stepFinish !== 'tool-calls') {
    turn.finish({ type: 'result', cost: turn.cost });
  }
}

// ── HTTP helpers ──

async function serveJson<T>(config: ServeConfig, path: string, init: RequestInit & { cwd: string }): Promise<T | null> {
  const { cwd, ...rest } = init;
  const res = await fetch(`${config.baseUrl}${path}`, {
    ...rest,
    headers: {
      'content-type': 'application/json',
      // The location is per REQUEST on this surface, not a property of the session. A call without it
      // runs against the serve's own directory, which is not where the user's files are.
      'x-opencode-directory': cwd,
      ...(rest.headers ?? {}),
    },
    signal: AbortSignal.timeout(30_000),
  });
  if (!res.ok) throw new Error(`${path} → ${res.status}`);
  const text = await res.text();
  if (!text) return null;
  // This surface wraps everything in `{data: …}`; the legacy one does not. Reading `.id` off the
  // envelope silently yields undefined, which is how an entire afternoon disappeared once.
  const body = JSON.parse(text) as { data?: T } | T;
  return (body as { data?: T }).data ?? (body as T);
}

// ── The turn ──

export async function runOpenCodeTurnOnServe(
  params: OpenCodeRunParams,
  config: ServeConfig,
  emit: Emit,
): Promise<void> {
  const { sessionKey } = params;
  const cwd = params.cwd || config.fallbackCwd;

  ensureEventStream(config);

  // ── A message arriving while a turn is running is an INJECTION, not a new turn ──
  //
  // This is where the subprocess and the serve genuinely part company. `opencode run` had no input
  // channel, so a second message could only supersede: kill the process, start again, lose the turn.
  // The serve takes another prompt into the RUNNING turn, so the right move is to hand it over and keep
  // the existing turn exactly as it is.
  //
  // Keeping the same turn object is the load-bearing part. Retiring it and registering a replacement —
  // which is what this did at first — stops officer routing the events the serve is still producing,
  // while the serve carries on regardless. The output goes nowhere and the turn looks hung.
  //
  // `steer` because the user typed it during the turn and means it now; officer's composer already
  // treats a send-while-generating as "add this to what you are doing". A prompt sent when nothing is
  // running takes `queue`, which is a no-op with an empty queue but never accidentally merges two
  // messages into one turn.
  const live = bySessionKey.get(sessionKey);
  if (live && !live.done) {
    try {
      await serveJson(config, `/api/session/${live.openCodeSessionId}/prompt`, {
        method: 'POST',
        body: JSON.stringify({
          prompt: { text: params.prompt, files: promptFiles(params.images) },
          delivery: 'steer',
        }),
        cwd,
      });
    } catch (err) {
      // The turn itself is unharmed — only the injection failed — so say so and leave it running.
      emit({
        type: 'opencode:event',
        sessionKey,
        event: { type: 'error', message: `OpenCode would not take that mid-turn: ${errText(err)}` },
      });
    }
    return;
  }

  let openCodeSessionId = params.resumeSessionId ?? '';

  try {
    if (!openCodeSessionId) {
      const created = await serveJson<{ id: string }>(config, '/api/session', {
        method: 'POST',
        body: JSON.stringify({ location: { directory: cwd } }),
        cwd,
      });
      if (!created?.id) throw new Error('session create returned no id');
      openCodeSessionId = created.id;
    }

    if (params.model) {
      // `providerID/modelID`, the same string the subprocess passes to `--model`.
      const slash = params.model.indexOf('/');
      const providerID = slash > 0 ? params.model.slice(0, slash) : 'opencode';
      const id = slash > 0 ? params.model.slice(slash + 1) : params.model;
      await serveJson(config, `/api/session/${openCodeSessionId}/model`, {
        method: 'POST',
        body: JSON.stringify({ model: { providerID, id } }),
        cwd,
      });
    }
  } catch (err) {
    emit({
      type: 'opencode:event',
      sessionKey,
      event: { type: 'error', message: `Could not start an OpenCode session: ${errText(err)}` },
    });
    return;
  }

  const turn: ServeTurn = {
    sessionKey,
    openCodeSessionId,
    cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 },
    done: false,
    emit,
    finish: (event) => retire(turn, event),
  };
  byOpenCodeId.set(openCodeSessionId, turn);
  bySessionKey.set(sessionKey, turn);

  // Officer learns which `ses_…` to resume next time — a routing fact, not transcript.
  emit({ type: 'opencode:session', sessionKey, sessionId: openCodeSessionId });

  try {
    await serveJson(config, `/api/session/${openCodeSessionId}/prompt`, {
      method: 'POST',
      // `delivery` is stated explicitly because it DEFAULTS to `"steer"`, which injects into a running
      // turn. For an ordinary send that is the wrong default — two quick messages would merge into one
      // turn instead of running in order. `steer` is Phase C's job, wired to the button that means it.
      body: JSON.stringify({
        prompt: { text: params.prompt, files: promptFiles(params.images) },
        delivery: 'queue',
      }),
      cwd,
    });
  } catch (err) {
    retire(turn, { type: 'error', message: `OpenCode refused the prompt: ${errText(err)}` });
  }
}

function retire(turn: ServeTurn, event: ChatEvent | null): void {
  if (turn.done) return;
  turn.done = true;
  if (byOpenCodeId.get(turn.openCodeSessionId) === turn) byOpenCodeId.delete(turn.openCodeSessionId);
  if (bySessionKey.get(turn.sessionKey) === turn) bySessionKey.delete(turn.sessionKey);
  if (event) turn.emit({ type: 'opencode:event', sessionKey: turn.sessionKey, event });
}

const errText = (err: unknown): string => (err instanceof Error ? err.message : String(err));

/**
 * Images as `prompt.files`, which this surface takes by URI.
 *
 * **`data:` URIs, not `file://`.** Measured, because the choice is not obvious and the wrong one fails
 * at the provider rather than at the API: a `file://` attachment is accepted with a 200 and then dies
 * inside the turn with `Anthropic Messages media must contain valid base64`. A `data:` URI round-trips
 * and the model describes the image.
 *
 * This is strictly better than the subprocess path, which has to spill each image to a temp file for
 * `--file` and delete it afterwards. Here the bytes go in the request and there is nothing to clean up.
 *
 * Silently dropping these is exactly defect B4 — the user sees their image in their own bubble and the
 * model never receives it — so this exists before the serve path is switched on for anyone, not after.
 */
function promptFiles(images: OpenCodeRunParams['images']): { uri: string; name: string }[] | undefined {
  if (!images?.length) return undefined;
  return images.map((image, index) => ({
    uri: `data:${image.mediaType || 'image/png'};base64,${image.data}`,
    name: `attachment-${index + 1}`,
  }));
}

/** The turns running right now, for `opencode:list` and the Live panel. */
export function listRunningServeTurns(): { sessionKey: string }[] {
  return [...bySessionKey.keys()].map((sessionKey) => ({ sessionKey }));
}

/**
 * Stop a turn without destroying its session — the thing the subprocess path cannot do.
 *
 * `POST /interrupt` leaves the conversation intact and resumable, where killing a subprocess ended it.
 */
export async function killServeTurn(sessionKey: string, config: ServeConfig): Promise<void> {
  const turn = bySessionKey.get(sessionKey);
  if (!turn) return;
  try {
    await fetch(`${config.baseUrl}/api/session/${turn.openCodeSessionId}/interrupt`, {
      method: 'POST',
      headers: { 'x-opencode-directory': config.fallbackCwd },
      signal: AbortSignal.timeout(10_000),
    });
  } catch {
    /* interrupt is best-effort; the turn is retired either way so the UI is never stuck */
  }
  retire(turn, { type: 'stopped' });
}

/**
 * Retire every live turn, for shutdown.
 *
 * Note what this does NOT do, and why it is right: the turns keep running inside the serve, which is a
 * separate process and survives us. Officer stops routing them and says so in the transcript. When the
 * subprocess ran turns, shutdown had to kill children or they were orphaned; here the work is somebody
 * else's and killing it would be the wrong call.
 */
export function stopAllServeTurns(message: string): number {
  const turns = [...bySessionKey.values()];
  for (const turn of turns) retire(turn, { type: 'error', message });
  return turns.length;
}