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

import type { ChatEvent, TurnMessage } from '../../api/chat/types';
import { officerModelForApiModel } from '../../api/chat/model-aliases';

// Translation from a harness's ChatEvent stream to the browser-facing turn messages, moved here from the
// main server (it was `createEventHandler` in `chat/websocket.ts`, once per harness). Both the agent and
// the opencode sidecar use this one copy. It lives with the process that produces the
// stream because it is stateful: `delta` events accumulate into a buffer that has to be flushed as one
// `assistant:text` at the next boundary (a tool call, or the end of the turn). A consumer downstream of
// a socket that can drop cannot hold that state correctly.
//
// Pure apart from the buffer, and the buffer is the whole of it — which is what makes it testable.

export type TurnOutput = {
  msg: TurnMessage;
  /** false only for `assistant:delta`: superseded by the `assistant:text` that follows, never persisted. */
  durable: boolean;
};

export type TurnStream = {
  push: (event: ChatEvent) => TurnOutput[];
};

export function createTurnStream(sessionId: string): TurnStream {
  // One buffer per speaker, keyed by the Task tool call that owns it ('' for the main agent). A subagent
  // runs concurrently with the agent that spawned it, so a single buffer interleaved their sentences and
  // flushed the result as one message attributed to whoever happened to hit the next boundary.
  const buffers = new Map<string, string>();

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

  // Emit whatever deltas have accumulated as one complete message. Called at every boundary where that
  // speaker stops talking, so the transcript holds text blocks rather than a thousand fragments.
  function flush(parent: string): TurnOutput[] {
    const text = buffers.get(parent);
    if (!text) return [];
    buffers.delete(parent);
    return [{ msg: { type: 'assistant:text', text, ...parented(parent) }, durable: true }];
  }

  /** A turn's end invalidates every buffer, the subagents' included. */
  function flushAll(): TurnOutput[] {
    return [...buffers.keys()].flatMap(flush);
  }

  function push(event: ChatEvent): TurnOutput[] {
    const parent = ('parentToolUseId' in event ? event.parentToolUseId : undefined) ?? '';

    switch (event.type) {
      case 'session':
        // Not durable: the id names the log rather than belonging in it, and a client replaying the log
        // had to know the id to ask for it. Persisting it would put a second copy of the answer inside
        // the question.
        return [{ msg: { type: 'session:claude', claudeSessionId: event.claudeSessionId }, durable: false }];

      case 'delta':
        buffers.set(parent, (buffers.get(parent) ?? '') + event.text);
        return [{ msg: { type: 'assistant:delta', text: event.text, ...parented(parent) }, durable: false }];

      case 'text': {
        // An explicit full text block wins over the accumulated deltas that produced it.
        const text = event.text || buffers.get(parent) || '';
        buffers.delete(parent);
        return text ? [{ msg: { type: 'assistant:text', text, ...parented(parent) }, durable: true }] : [];
      }

      case 'tool:start':
        return [
          ...flush(parent),
          {
            msg: {
              type: 'tool:start',
              toolCallId: event.toolCallId,
              toolName: event.toolName,
              toolInput: event.toolInput,
              ...parented(parent),
            },
            durable: true,
          },
        ];

      case 'tool:result':
        return [
          {
            msg: {
              type: 'tool:result',
              toolCallId: event.toolCallId,
              output: event.output,
              isError: event.isError,
              ...parented(parent),
            },
            durable: true,
          },
        ];

      case 'result':
        return [
          ...flushAll(),
          {
            msg: { type: 'result', sessionId, cost: event.cost, claudeSessionId: event.claudeSessionId },
            durable: true,
          },
        ];

      case 'error':
        return [{ msg: { type: 'error', message: event.message }, durable: true }];

      case 'stopped':
        return [{ msg: { type: 'stopped' }, durable: true }];

      case 'task:started':
        return [
          {
            msg: {
              type: 'task:started',
              taskId: event.taskId,
              description: event.description,
              taskType: event.taskType,
            },
            durable: true,
          },
        ];

      case 'task:notification':
        return [
          {
            msg: { type: 'task:notification', taskId: event.taskId, status: event.status, summary: event.summary },
            durable: true,
          },
        ];

      case 'compact:start':
        // Flushed first: whatever the agent had said before compaction began is finished text, and the
        // notice belongs after it rather than in the middle of a paragraph it interrupted.
        return [...flushAll(), { msg: { type: 'compact:start', trigger: event.trigger }, durable: true }];

      case 'model:switched':
        // Flushed for the same reason, and it matters more here: the marker's whole job is to say where in
        // the conversation the tier changed, so it must land after the last thing the old model said.
        return [
          ...flushAll(),
          {
            msg: {
              type: 'model:switched',
              from: event.from,
              to: event.to,
              ok: event.ok,
              ...(event.error ? { error: event.error } : {}),
            },
            durable: true,
          },
        ];

      case 'model:resolved': {
        // Not durable and not flushed: it says nothing about WHERE in the conversation anything happened,
        // only what the harness is running. `session:claude` is the same shape for the same reason.
        //
        // A null mapping is dropped rather than forwarded raw. `<synthetic>`, mythos and every OpenCode id
        // resolve to null, and offering the picker a model this machine may not reach is worse than
        // leaving it as it was.
        const officerId = officerModelForApiModel(event.apiModel);
        if (!officerId) return [];
        return [{ msg: { type: 'model:resolved', model: officerId }, durable: false }];
      }

      case 'plan:mode':
        // Flushed like `model:switched`, and for the same reason: the marker's job is to say WHERE in the
        // conversation the rules changed, which is worthless if it lands mid-paragraph.
        return [
          ...flushAll(),
          {
            msg: {
              type: 'plan:mode',
              enabled: event.enabled,
              ok: event.ok,
              ...(event.error ? { error: event.error } : {}),
            },
            durable: true,
          },
        ];

      case 'plan:proposed':
        // Flushed because the agent's closing words before a plan are its introduction to it, and a card
        // spliced into an unfinished sentence reads as an interruption rather than a conclusion.
        return [
          ...flushAll(),
          { msg: { type: 'plan:proposed', requestId: event.requestId, plan: event.plan }, durable: true },
        ];

      case 'plan:decided':
        return [
          {
            msg: {
              type: 'plan:decided',
              requestId: event.requestId,
              approved: event.approved,
              ...(event.feedback ? { feedback: event.feedback } : {}),
            },
            durable: true,
          },
        ];

      case 'compact:done':
        return [
          {
            msg: {
              type: 'compact:done',
              trigger: event.trigger,
              preTokens: event.preTokens,
              durationMs: event.durationMs,
            },
            durable: true,
          },
        ];
    }
  }

  return { push };
}