Sybil-2/web/src/lib/agent-room.ts

import type { Message } from "./api";

export type AgentActivity = {
  state: "idle" | "thinking" | "tool" | "waiting" | "error";
  isThinking: boolean;
  queued: number;
  backgroundWorkers: number;
  tools: Array<{ id: string; name: string; status: string; output?: string }>;
  queuedMessages: Message[];
  wait?: { due: number; reason?: string };
  error?: string;
};
export type AgentMessagesPage = {
  conversationId: string; messages: Message[]; cursor: string | null;
  beforeCursor: string | null; hasMore: boolean;
};
export type AgentDescriptor = {
  id: string; title: string; model: string; activity: AgentActivity;
  websocket: string; protocol: string;
};
export type AgentReadState = { cursor: string; updatedAt: string | null };
export type AgentRoomState = {
  conversationId: string | null; messages: Message[]; cursor: string;
  readCursor: string; activity: AgentActivity | null;
};
export type AgentRoomEvent =
  | ({ type: "messages" } & AgentMessagesPage)
  | { type: "activity"; conversationId: string; activity: AgentActivity }
  | { type: "ready"; conversationId?: string }
  | { type: "pong" | "heartbeat" }
  | { type: "error"; error?: string; message?: string };

const object = (value: unknown): Record<string, unknown> | undefined => value !== null && typeof value === "object" && !Array.isArray(value)
  ? value as Record<string, unknown> : undefined;
const defined = (value: Record<string, unknown>) => Object.fromEntries(Object.entries(value).filter(([, item]) => item !== undefined));
export function normalizeAgentCursor(value: unknown): string {
  if (typeof value !== "string" || !/^\d+$/.test(value)) throw new Error("Invalid agent message cursor.");
  return BigInt(value).toString();
}
export function advanceAgentCursor(current: string, incoming: string): string {
  const a = normalizeAgentCursor(current), b = normalizeAgentCursor(incoming);
  return BigInt(a) >= BigInt(b) ? a : b;
}
function metadata(message: Message) { return object(message.metadata); }
function messageCursor(message: Message): string | null {
  const entryId = metadata(message)?.entryId;
  if (typeof entryId === "number" && Number.isSafeInteger(entryId) && entryId >= 0) return String(entryId);
  return typeof entryId === "string" && /^\d+$/.test(entryId) ? normalizeAgentCursor(entryId) : null;
}
function mergeMessage(previous: Message, incoming: Message): Message {
  const prior = metadata(previous), next = metadata(incoming);
  // A delayed activity snapshot must not demote an already placed user message.
  if (prior?.status === "complete" && next?.status === "queued") return previous;
  if (prior?.kind !== "tool_call" || next?.kind !== "tool_call") return { ...previous, ...incoming };
  const terminal = prior.status === "completed" || prior.status === "failed";
  const staleStart = terminal && next.status === "initiated";
  const merged = staleStart ? { ...defined(next), ...defined(prior) } : { ...defined(prior), ...defined(next) };
  if (!next.startedAt && prior.startedAt) merged.startedAt = prior.startedAt;
  const oldArgs = object(prior.args), newArgs = object(next.args);
  if (oldArgs && (!newArgs || Object.keys(newArgs).length === 0)) merged.args = oldArgs;
  return { ...previous, ...incoming, ...(staleStart ? { content: previous.content, name: previous.name } : {}),
    createdAt: previous.createdAt < incoming.createdAt ? previous.createdAt : incoming.createdAt, metadata: merged };
}

/** Replays and separate tool-start/result pages update stable rows instead of duplicating them. */
export function mergeAgentMessages(previous: readonly Message[], incoming: readonly Message[]): Message[] {
  const messages = new Map(previous.map(message => [message.id, message]));
  for (const message of incoming) {
    const prior = messages.get(message.id);
    messages.set(message.id, prior ? mergeMessage(prior, message) : message);
  }
  return [...messages.values()].sort((a, b) => {
    const byTime = a.createdAt.localeCompare(b.createdAt);
    if (byTime) return byTime;
    const left = messageCursor(a), right = messageCursor(b);
    if (left !== null && right !== null && left !== right) return BigInt(left) < BigInt(right) ? -1 : 1;
    return a.id.localeCompare(b.id);
  });
}
export const createAgentRoomState = (): AgentRoomState => ({
  conversationId: null, messages: [], cursor: "0", readCursor: "0", activity: null,
});
export function applyAgentPage(state: AgentRoomState, page: AgentMessagesPage): AgentRoomState {
  if (state.conversationId !== null && state.conversationId !== page.conversationId) throw new Error("Agent conversation changed; reload the page.");
  // Validate before changing state. Failed delivery/decoding cannot advance replay position.
  const cursor = page.cursor === null ? state.cursor : advanceAgentCursor(state.cursor, page.cursor);
  return { ...state, conversationId: page.conversationId, messages: mergeAgentMessages(state.messages, page.messages), cursor };
}
export function applyAgentReadState(state: AgentRoomState, read: AgentReadState): AgentRoomState {
  return { ...state, readCursor: advanceAgentCursor(state.readCursor, read.cursor) };
}
export function applyAgentEvent(state: AgentRoomState, event: AgentRoomEvent): AgentRoomState {
  if (event.type === "messages") return applyAgentPage(state, event);
  if (event.type !== "activity") return state;
  if (state.conversationId !== null && state.conversationId !== event.conversationId) throw new Error("Agent activity belongs to a different conversation.");
  return { ...state, conversationId: event.conversationId, activity: event.activity };
}
export function agentUnreadCount(state: AgentRoomState): number {
  return state.messages.filter(message => {
    if (message.role !== "assistant" || metadata(message)?.kind !== "message") return false;
    const cursor = messageCursor(message);
    return cursor !== null && BigInt(cursor) > BigInt(state.readCursor);
  }).length;
}
export function agentDisplayMessages(state: AgentRoomState): Message[] {
  const committed = new Set(state.messages.map(message => message.id));
  return mergeAgentMessages(state.messages, (state.activity?.queuedMessages || []).filter(message => !committed.has(message.id)));
}
export function parseAgentRoomEvent(raw: string): AgentRoomEvent | null {
  const event = object(JSON.parse(raw));
  if (!event || typeof event.type !== "string") throw new Error("Invalid agent room frame.");
  if (!["messages", "activity", "ready", "pong", "heartbeat", "error"].includes(event.type)) return null;
  if (event.type === "messages") {
    if (typeof event.conversationId !== "string" || !Array.isArray(event.messages) || typeof event.hasMore !== "boolean") throw new Error("Invalid agent messages frame.");
    if (event.cursor !== null) normalizeAgentCursor(event.cursor);
    for (const item of event.messages) {
      const message = object(item);
      if (!message || typeof message.id !== "string" || typeof message.content !== "string" || typeof message.createdAt !== "string"
        || !["system", "user", "assistant", "tool"].includes(String(message.role))) throw new Error("Invalid agent message.");
    }
  }
  if (event.type === "activity" && (typeof event.conversationId !== "string" || !object(event.activity))) throw new Error("Invalid agent activity frame.");
  return event as AgentRoomEvent;
}
export function agentAuthFrame(token: string | null, cursor: string): string {
  return JSON.stringify({ type: "auth", token, after: normalizeAgentCursor(cursor) });
}

export type AgentSubmission = { id: number; status: "queued" | "placed" | "done" | "unanswered"; messageId: string };
export type AgentRoomClientOptions = { baseUrl?: string; token?: () => string | null; fetch?: typeof globalThis.fetch };
export class AgentRoomClient {
  private readonly baseUrl: string;
  private readonly fetcher: typeof globalThis.fetch;
  private readonly token: () => string | null;
  constructor(options: AgentRoomClientOptions = {}) {
    this.baseUrl = (options.baseUrl || "/api").replace(/\/+$/, "");
    this.fetcher = (options.fetch || globalThis.fetch).bind(globalThis);
    this.token = options.token || (() => null);
  }
  private async request<T>(path: string, init: RequestInit = {}): Promise<T> {
    const headers = new Headers(init.headers);
    const token = this.token();
    if (token) headers.set("Authorization", `Bearer ${token}`);
    if (init.body !== undefined) headers.set("Content-Type", "application/json");
    const response = await this.fetcher(`${this.baseUrl}/v1/agent${path}`, { ...init, headers });
    let value: unknown;
    try { value = await response.json(); }
    catch { throw new Error(`Agent API returned ${response.status} without a JSON response.`); }
    if (!response.ok) {
      const error = object(value);
      throw new Error(typeof error?.message === "string" ? error.message : typeof error?.error === "string" ? error.error : `Agent API returned ${response.status}.`);
    }
    return value as T;
  }
  descriptor(signal?: AbortSignal): Promise<AgentDescriptor> { return this.request("", { signal }); }
  messages(after: string, signal?: AbortSignal): Promise<AgentMessagesPage> {
    return this.request(`/messages?${new URLSearchParams({ after: normalizeAgentCursor(after), limit: "200" })}`, { signal });
  }
  readState(signal?: AbortSignal): Promise<AgentReadState> { return this.request("/read-state", { signal }); }
  markRead(cursor: string, signal?: AbortSignal): Promise<AgentReadState> {
    return this.request("/read-state", { method: "PUT", body: JSON.stringify({ cursor: normalizeAgentCursor(cursor) }), signal });
  }
  async submit(content: string, requestId: string): Promise<AgentSubmission> {
    const receipt = object(await this.request<unknown>("/messages", { method: "POST", body: JSON.stringify({ content, requestId }) }));
    // An HTTP success alone does not acknowledge delivery. Keep the original
    // outbox receipt key retryable if a proxy or upstream returns malformed data.
    if (!receipt || typeof receipt.id !== "number" || !Number.isSafeInteger(receipt.id) || receipt.id <= 0
      || typeof receipt.status !== "string" || !["queued", "placed", "done", "unanswered"].includes(receipt.status)
      || receipt.messageId !== `inbound:sybil:${requestId}`) throw new Error("Agent API returned an invalid message receipt. Retry to confirm delivery.");
    return receipt as AgentSubmission;
  }
  websocketUrl(pageUrl: string): string {
    const url = new URL(`${this.baseUrl}/v1/agent/ws`, pageUrl);
    url.protocol = url.protocol === "https:" ? "wss:" : "ws:";
    return url.toString();
  }
}

/** Read every newer page, including pages containing only invisible internal entries. */
export async function syncAgentHistory(client: Pick<AgentRoomClient, "messages">, from: string,
  apply: (page: AgentMessagesPage) => void, signal?: AbortSignal): Promise<void> {
  let cursor = normalizeAgentCursor(from);
  while (!signal?.aborted) {
    const page = await client.messages(cursor, signal);
    if (signal?.aborted) return;
    const next = normalizeAgentCursor(page.cursor ?? cursor);
    if (BigInt(next) < BigInt(cursor) || (page.hasMore && next === cursor)) throw new Error("Agent history did not advance its cursor.");
    apply(page);
    cursor = next;
    if (!page.hasMore) return;
  }
}

// Retained only to migrate receipt IDs from the former admission sender.
export const AGENT_ADMISSION_STORAGE_KEY = "sybil.agent.pending-admissions.v1";

/** Only successfully persisted, explicitly observed cursors become read state. */
export function createAgentReadWriter(options: {
  current: () => string; save: (cursor: string) => Promise<AgentReadState>; apply: (read: AgentReadState) => void;
}) {
  let desired = "0";
  let running: Promise<void> | undefined;
  const flush = (): Promise<void> => {
    if (running) return running;
    running = (async () => {
      while (BigInt(desired) > BigInt(options.current())) {
        const captured = desired;
        const saved = await options.save(captured);
        if (BigInt(normalizeAgentCursor(saved.cursor)) < BigInt(captured)) throw new Error("Agent read state did not acknowledge the seen messages.");
        options.apply(saved);
      }
    })().finally(() => { running = undefined; });
    return running;
  };
  return { request(cursor: string) { desired = advanceAgentCursor(desired, cursor); return flush(); }, flush };
}