platform/src/servers/api/chat/opencode/client.ts

import type { OpenCodeEvent } from './event-mapper';
import { logger } from '../logger';

// One connection per `opencode serve` base URL: a single shared SSE subscription on `GET /event`
// demultiplexed to per-session listeners, plus the REST calls a turn needs. The SDK-style
// `/session/*` route family is used (feature-complete, incl. DELETE).

type Listener = (event: OpenCodeEvent) => void;

class ServerConnection {
  private listeners = new Map<string, Set<Listener>>();
  private sseRunning = false;

  constructor(private readonly baseUrl: string) {}

  subscribe(sessionId: string, listener: Listener): () => void {
    let set = this.listeners.get(sessionId);
    if (!set) {
      set = new Set();
      this.listeners.set(sessionId, set);
    }
    set.add(listener);
    void this.ensureSse();

    return () => {
      const current = this.listeners.get(sessionId);
      if (!current) return;
      current.delete(listener);
      if (current.size === 0) this.listeners.delete(sessionId);
    };
  }

  private dispatch(event: OpenCodeEvent): void {
    const sessionId = event.properties?.sessionID;
    if (typeof sessionId !== 'string') return;
    const set = this.listeners.get(sessionId);
    if (!set) return;
    for (const listener of set) {
      try {
        listener(event);
      } catch (err) {
        logger.error('opencode SSE listener threw', { error: String(err) });
      }
    }
  }

  /** Keep a single `/event` SSE stream open while any session has listeners; reconnect on drop. */
  private async ensureSse(): Promise<void> {
    if (this.sseRunning) return;
    this.sseRunning = true;

    void (async () => {
      while (this.listeners.size > 0) {
        try {
          const res = await fetch(`${this.baseUrl}/event`, { headers: { accept: 'text/event-stream' } });
          if (!res.body) throw new Error('no SSE body');
          const reader = res.body.getReader();
          const decoder = new TextDecoder();
          let buffer = '';

          while (true) {
            const { value, done } = await reader.read();
            if (done) break;
            buffer += decoder.decode(value, { stream: true });

            // SSE frames are separated by a blank line.
            let sep = buffer.indexOf('\n\n');
            while (sep !== -1) {
              const frame = buffer.slice(0, sep);
              buffer = buffer.slice(sep + 2);
              this.handleFrame(frame);
              sep = buffer.indexOf('\n\n');
            }
          }
        } catch (err) {
          logger.warn('opencode SSE stream error; will reconnect', { baseUrl: this.baseUrl, error: String(err) });
        }
        if (this.listeners.size > 0) await new Promise((r) => setTimeout(r, 500));
      }
      this.sseRunning = false;
    })();
  }

  private handleFrame(frame: string): void {
    const data = frame
      .split('\n')
      .filter((line) => line.startsWith('data:'))
      .map((line) => line.slice(5).trim())
      .join('');
    if (!data) return;
    try {
      this.dispatch(JSON.parse(data) as OpenCodeEvent);
    } catch {
      /* non-JSON keep-alive or partial frame */
    }
  }

  private async postJson<T>(path: string, body: unknown): Promise<T> {
    const res = await fetch(`${this.baseUrl}${path}`, {
      method: 'POST',
      headers: { 'content-type': 'application/json' },
      body: JSON.stringify(body),
    });
    if (!res.ok) throw new Error(`opencode POST ${path} → ${res.status} ${await res.text().catch(() => '')}`);
    return (await res.json()) as T;
  }

  async createSession(metadata?: Record<string, unknown>, title?: string): Promise<string> {
    // OpenCode has no per-session `directory` (a session inherits the server's cwd). We instead tag
    // the session with our own free-form `metadata` (e.g. { officer: { cwd } }) — round-trips on the
    // list + detail endpoints and is never touched by opencode core — to know where it belongs.
    const body: Record<string, unknown> = {};
    if (metadata) body.metadata = metadata;
    if (title) body.title = title;
    const session = await this.postJson<{ id?: string }>('/session', body);
    if (!session.id) throw new Error('opencode POST /session returned no id');
    return session.id;
  }

  async postMessage(
    sessionId: string,
    providerID: string,
    modelID: string,
    text: string,
    system?: string,
  ): Promise<void> {
    // `system` is appended to OpenCode's built-in system prompt (additive, not an override) and is
    // sent to the LLM as a system message — it never appears as a visible chat part.
    const body: Record<string, unknown> = {
      model: { providerID, modelID },
      parts: [{ type: 'text', text }],
    };
    if (system) body.system = system;
    await this.postJson(`/session/${sessionId}/message`, body);
  }

  async abort(sessionId: string): Promise<void> {
    await this.postJson(`/session/${sessionId}/abort`, {}).catch(() => {});
  }

  // ── Session history (REST; OpenCode's SQLite store is the source of truth) ──

  async listSessions(): Promise<OpenCodeSessionInfo[]> {
    const res = await fetch(`${this.baseUrl}/session`);
    if (!res.ok) throw new Error(`opencode GET /session → ${res.status}`);
    return (await res.json()) as OpenCodeSessionInfo[];
  }

  async getMessages(sessionId: string): Promise<OpenCodeStoredMessage[]> {
    const res = await fetch(`${this.baseUrl}/session/${sessionId}/message`);
    if (!res.ok) throw new Error(`opencode GET /session/${sessionId}/message → ${res.status}`);
    return (await res.json()) as OpenCodeStoredMessage[];
  }

  async deleteSession(sessionId: string): Promise<boolean> {
    const res = await fetch(`${this.baseUrl}/session/${sessionId}`, { method: 'DELETE' });
    return res.ok;
  }

  async renameSession(sessionId: string, title: string): Promise<boolean> {
    const res = await fetch(`${this.baseUrl}/session/${sessionId}`, {
      method: 'PATCH',
      headers: { 'content-type': 'application/json' },
      body: JSON.stringify({ title }),
    });
    return res.ok;
  }
}

// Shapes returned by the `/session/*` read endpoints (verified against opencode 1.17.9).
export type OpenCodeSessionInfo = {
  id: string;
  title?: string;
  time?: { created?: number; updated?: number };
  location?: { directory?: string | null };
  metadata?: Record<string, unknown>;
};

// Our own namespaced session metadata (stored under the free-form `metadata.officer` key).
export type OfficerSessionMeta = { cwd?: string };
export const officerMeta = (m?: Record<string, unknown>): OfficerSessionMeta =>
  (m?.officer as OfficerSessionMeta) ?? {};

export type OpenCodeStoredPart = {
  type?: string;
  text?: string;
  tool?: string;
  callID?: string;
  state?: { input?: Record<string, unknown>; output?: string; error?: string };
};

export type OpenCodeStoredMessage = {
  info?: { role?: string; modelID?: string; model?: { modelID?: string; providerID?: string } };
  parts?: OpenCodeStoredPart[];
};

const connections = new Map<string, ServerConnection>();

export function getConnection(baseUrl: string): ServerConnection {
  let conn = connections.get(baseUrl);
  if (!conn) {
    conn = new ServerConnection(baseUrl);
    connections.set(baseUrl, conn);
  }
  return conn;
}

export type { ServerConnection };