platform/src/servers/sidecar/claude/stream-parser.ts

/**
 * Claude Code NDJSON Stream Parser
 *
 * Parses the stream-json output format from Claude Code CLI into ChatEvents.
 * Handles: stream_event (deltas), assistant (text/tool_use), user (tool_result),
 * system init, and result messages.
 */

import type { ChatEvent, MessageCost } from '../../api/chat/types';

type SessionCallback = (sessionId: string) => void;

type StreamParserCallbacks = {
  onEvent: (event: ChatEvent) => void;
  onSessionId: SessionCallback;
};

type ParseState = {
  /**
   * Accumulated deltas, keyed by the `parent_tool_use_id` that produced them ('' for the agent you are
   * talking to). One shared buffer was wrong as soon as a subagent ran: its deltas appended to whatever
   * the main agent had said so far, and the next flush emitted the two of them concatenated as a single
   * message from the main agent.
   */
  textBuffers: Map<string, string>;
  gotResult: boolean;
  /**
   * Tool call ids belonging to questions the user is being asked.
   *
   * Kept so their results can be un-flagged: the sidecar answers `AskUserQuestion` by DENYING it with the
   * user's words (the only hook decision that can carry a result), and the harness stamps every denied tool
   * `is_error`. So a perfectly good answer arrives looking like a failed tool call — red in the transcript,
   * and read by the agent as something that went wrong rather than as the reply to its own question.
   */
  questionCallIds: Set<string>;
};

/** Both harnesses ask the human the same way and name the tool differently. */
const QUESTION_TOOLS = new Set(['question', 'AskUserQuestion']);

/** '' when a message is the main agent's own. Keys the buffer map and stamps every event we emit. */
function parentOf(msg: Record<string, unknown>): string {
  const id = msg.parent_tool_use_id;
  return typeof id === 'string' ? id : '';
}

/** Only set the field when there IS a parent, so main-agent events stay byte-identical on the wire. */
const parented = (parent: string) => (parent ? { parentToolUseId: parent } : {});

function flushTextBuffer(state: ParseState, parent: string, onEvent: (event: ChatEvent) => void): void {
  const buffered = state.textBuffers.get(parent);
  if (buffered) {
    onEvent({ type: 'text', text: buffered, ...parented(parent) });
    state.textBuffers.delete(parent);
  }
}

function handleStreamEvent(msg: Record<string, unknown>, state: ParseState, onEvent: (event: ChatEvent) => void): void {
  const event = msg.event as Record<string, unknown> | undefined;
  if (event?.type === 'content_block_delta') {
    const delta = event.delta as Record<string, unknown> | undefined;
    if (delta?.type === 'text_delta' && typeof delta.text === 'string') {
      const parent = parentOf(msg);
      state.textBuffers.set(parent, (state.textBuffers.get(parent) ?? '') + delta.text);
      onEvent({ type: 'delta', text: delta.text, ...parented(parent) });
    }
  }
}

function handleAssistant(msg: Record<string, unknown>, state: ParseState, onEvent: (event: ChatEvent) => void): void {
  const message = msg.message as Record<string, unknown> | undefined;
  const content = message?.content as Array<Record<string, unknown>> | undefined;
  if (!Array.isArray(content)) return;

  const parent = parentOf(msg);

  for (const block of content) {
    if (block.type === 'text' && typeof block.text === 'string') {
      onEvent({ type: 'text', text: block.text, ...parented(parent) });
      state.textBuffers.delete(parent);
    } else if (block.type === 'tool_use') {
      flushTextBuffer(state, parent, onEvent);
      if (QUESTION_TOOLS.has((block.name as string) ?? '')) state.questionCallIds.add((block.id as string) ?? '');
      onEvent({
        type: 'tool:start',
        toolCallId: (block.id as string) ?? '',
        toolName: (block.name as string) ?? 'unknown',
        toolInput: (block.input as Record<string, unknown>) ?? {},
        ...parented(parent),
      });
    }
  }
}

function handleUser(msg: Record<string, unknown>, state: ParseState, onEvent: (event: ChatEvent) => void): void {
  const message = msg.message as Record<string, unknown> | undefined;
  const content = message?.content as Array<Record<string, unknown>> | undefined;
  if (!Array.isArray(content)) return;

  const parent = parentOf(msg);

  for (const block of content) {
    if (block.type === 'tool_result') {
      let output = '';
      if (typeof block.content === 'string') {
        output = block.content;
      } else if (Array.isArray(block.content)) {
        output = (block.content as Array<Record<string, unknown>>)
          .filter((c) => c.type === 'text')
          .map((c) => c.text as string)
          .join('\n');
      }
      const toolCallId = (block.tool_use_id as string) ?? '';
      // An answered question is not a failed tool call. See `questionCallIds`: the answer channel is a
      // denial, so the flag says only that the tool was prevented from resolving itself — which is the
      // mechanism working, not a fault. A question that timed out or lost its session lands here too, and
      // is equally not an error: its output is a sentence explaining that, which the agent should read as
      // information rather than as something to retry.
      const isError = state.questionCallIds.has(toolCallId) ? false : ((block.is_error as boolean) ?? false);
      onEvent({ type: 'tool:result', toolCallId, output, isError, ...parented(parent) });
    }
  }
}

function handleResult(msg: Record<string, unknown>, state: ParseState, callbacks: StreamParserCallbacks): void {
  state.gotResult = true;

  const isError = (msg.is_error as boolean) ?? false;
  const resultText = (msg.result as string) ?? '';

  // A `result` ends the turn, so every buffer is stale — including a subagent's, if it died mid-sentence.
  function flushAll(): void {
    for (const parent of [...state.textBuffers.keys()]) flushTextBuffer(state, parent, callbacks.onEvent);
  }

  if (isError) {
    flushAll();
    callbacks.onEvent({ type: 'error', message: resultText || 'Claude Code returned an error' });
    return;
  }

  flushAll();

  const usage = msg.usage as Record<string, number> | undefined;
  const cost: MessageCost = {
    inputTokens: usage?.input_tokens ?? 0,
    outputTokens: usage?.output_tokens ?? 0,
    totalUSD: (msg.total_cost_usd as number) ?? 0,
  };

  const sessionId = msg.session_id as string | undefined;
  if (sessionId) {
    callbacks.onSessionId(sessionId);
  }

  // Carried on the event as well as into sidecar-local state: the callback only feeds `--resume` here,
  // so without this the platform never learns which transcript file a session it started ended up in.
  callbacks.onEvent({ type: 'result', cost, claudeSessionId: sessionId });
}

function handleSystem(msg: Record<string, unknown>, callbacks: StreamParserCallbacks): void {
  const subtype = msg.subtype as string | undefined;
  if (subtype === 'init') {
    // The one place the harness says which model it is actually running. Officer often does not know —
    // it passes no `--model` when the client named no tier — so without this the picker had nothing to
    // show but the bare harness string. See `model:resolved` in api/chat/types.ts.
    const initModel = msg.model;
    if (typeof initModel === 'string' && initModel) {
      callbacks.onEvent({ type: 'model:resolved', apiModel: initModel });
    }

    const sessionId = msg.session_id as string | undefined;
    if (sessionId) {
      callbacks.onSessionId(sessionId);
      // Also out to the browser, and at the *start* of the turn. The same id used to travel only on
      // `result`, so a chat had no address until its first turn had finished — refresh before that and
      // there was nothing to reconnect by, which is exactly when a long turn is worth reconnecting to.
      callbacks.onEvent({ type: 'session', claudeSessionId: sessionId });
    }
  } else if (subtype === 'task_started') {
    callbacks.onEvent({
      type: 'task:started',
      taskId: (msg.task_id as string) ?? '',
      description: (msg.description as string) ?? '',
      taskType: msg.task_type as string | undefined,
    });
  } else if (subtype === 'task_notification') {
    callbacks.onEvent({
      type: 'task:notification',
      taskId: (msg.task_id as string) ?? '',
      status: (msg.status as 'completed' | 'failed' | 'stopped') ?? 'completed',
      summary: (msg.summary as string) ?? '',
    });
  } else if (subtype === 'compact_boundary') {
    // The harness rewrote its own context. This lands *after* the work — the matching start comes from
    // the `PreCompact` hook — and is the only place the size of what was dropped is reported.
    const meta = (msg.compact_metadata ?? {}) as Record<string, unknown>;
    callbacks.onEvent({
      type: 'compact:done',
      trigger: meta.trigger === 'manual' ? 'manual' : 'auto',
      preTokens: typeof meta.pre_tokens === 'number' ? meta.pre_tokens : 0,
    });
  }
}

/**
 * Map a single already-parsed message object (a Claude Code stream-json message, or the equivalent
 * Agent SDK message — same shapes) to ChatEvents. Returns false if the message type was ignored.
 */
export function processMessage(
  msg: Record<string, unknown>,
  state: ParseState,
  callbacks: StreamParserCallbacks,
): boolean {
  const type = msg.type as string;

  switch (type) {
    case 'stream_event':
      handleStreamEvent(msg, state, callbacks.onEvent);
      break;
    case 'assistant':
      handleAssistant(msg, state, callbacks.onEvent);
      break;
    case 'user':
      handleUser(msg, state, callbacks.onEvent);
      break;
    case 'system':
      handleSystem(msg, callbacks);
      break;
    case 'result':
      handleResult(msg, state, callbacks);
      break;
    default:
      return false;
  }

  return true;
}

/**
 * Process a single NDJSON line from Claude Code's stream output.
 * Returns false if the line was skipped (empty or malformed), true otherwise.
 */
export function processLine(line: string, state: ParseState, callbacks: StreamParserCallbacks): boolean {
  if (!line.trim()) return false;

  let msg: Record<string, unknown>;
  try {
    msg = JSON.parse(line) as Record<string, unknown>;
  } catch {
    return false;
  }

  return processMessage(msg, state, callbacks);
}

/**
 * Create a fresh parse state for a new stream.
 */
export function createParseState(): ParseState {
  return { textBuffers: new Map(), gotResult: false, questionCallIds: new Set() };
}

/**
 * Read an NDJSON stream and process each line.
 * Returns the final parse state.
 */
export async function parseStream(
  stdout: ReadableStream<Uint8Array>,
  callbacks: StreamParserCallbacks,
): Promise<ParseState> {
  const reader = stdout.getReader();
  const decoder = new TextDecoder();
  const state = createParseState();
  let buffer = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const chunk = decoder.decode(value, { stream: true });
    buffer += chunk;
    const lines = buffer.split('\n');
    buffer = lines.pop()!;
    for (const line of lines) {
      processLine(line, state, callbacks);
    }
  }

  if (buffer.trim()) {
    processLine(buffer, state, callbacks);
  }

  return state;
}

export type { ParseState, StreamParserCallbacks };