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

import type { Message } from "./api";
import { parseAgentAttachment, type AgentAttachment, type AgentAttachmentCapability } from "./agent-attachments.ts";
import { agentPushCursor, agentPushSourceId, agentPushUuid } from "./agent-push.ts";

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; capabilities?: { attachments?: AgentAttachmentCapability };
};
export type AgentReadState = { cursor: string; updatedAt: string | null; sourceId?: string | null };
export type AgentRoomState = {
  conversationId: string | null; messages: Message[]; cursor: string;
  readCursor: string; activity: AgentActivity | null;
  /** undefined before the first read snapshot; null pins a legacy source-less session. */
  sourceId?: string | null;
};
export type AgentRoomEvent =
  | ({ type: "messages" } & AgentMessagesPage)
  | { type: "activity"; conversationId: string; activity: AgentActivity }
  | { type: "ready"; conversationId?: string }
  | ({ type: "read_state" } & AgentReadState)
  | { 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 class AgentReadSourceChangedError extends Error {
  constructor() { super("Agent history changed. Reload the page before marking messages read."); this.name = "AgentReadSourceChangedError"; }
}
function readSource(state: Pick<AgentRoomState, "sourceId" | "cursor" | "readCursor" | "messages">, read: AgentReadState): string | null {
  if (read.sourceId != null && !agentPushSourceId(read.sourceId)) throw new AgentReadSourceChangedError();
  const incoming = read.sourceId ?? null;
  if (state.sourceId !== undefined && state.sourceId !== incoming) throw new AgentReadSourceChangedError();
  // A source cannot be attached retroactively to already rendered unknown history.
  if (state.sourceId === undefined && incoming !== null && (state.cursor !== "0" || state.readCursor !== "0" || state.messages.length > 0)) throw new AgentReadSourceChangedError();
  return incoming;
}
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 {
  const sourceId = readSource(state, read);
  return { ...state, sourceId, 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 === "read_state") return applyAgentReadState(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", "read_state"].includes(event.type)) return null;
  if (event.type === "read_state" && (!agentPushCursor(event.cursor)
    || (event.updatedAt !== null && typeof event.updatedAt !== "string")
    || (event.sourceId !== undefined && event.sourceId !== null && !agentPushSourceId(event.sourceId)))) throw new Error("Invalid Agent read frame.");
  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, presenceSessionId?: string): string {
  return JSON.stringify({ type: "auth", token, after: normalizeAgentCursor(cursor),
    ...(agentPushUuid(presenceSessionId) ? { clientKind: "web", presenceSessionId } : {}) });
}

export type AgentSubmission = { id: number; status: "queued" | "placed" | "done" | "unanswered"; messageId: string; attachmentIds?: 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 && !(init.body instanceof FormData)) 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 Object.assign(new Error(`Agent API returned ${response.status} without a JSON response.`), { status: response.status }); }
    if (!response.ok) {
      const error = object(value);
      throw Object.assign(new Error(typeof error?.message === "string" ? error.message : typeof error?.error === "string" ? error.error : `Agent API returned ${response.status}.`), { status: 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, sourceId?: string | null): Promise<AgentReadState> {
    return this.request("/read-state", { method: "PUT", body: JSON.stringify({ cursor: normalizeAgentCursor(cursor), ...(agentPushSourceId(sourceId) ? { sourceId } : {}) }), signal });
  }
  async submit(content: string, requestId: string, attachments: AgentAttachment[] = []): Promise<AgentSubmission> {
    const receipt = object(await this.request<unknown>("/messages", { method: "POST", body: JSON.stringify({ content, requestId, ...(attachments.length ? { attachments: attachments.map(({ id }) => ({ id })) } : {}) }) }));
    // 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}`
      || (attachments.length && JSON.stringify(receipt.attachmentIds) !== JSON.stringify(attachments.map(file => file.id)))) throw new Error("Agent API returned an invalid message receipt. Retry to confirm delivery.");
    return receipt as AgentSubmission;
  }
  async upload(file: File, uploadId: string, signal?: AbortSignal): Promise<AgentAttachment> {
    const body = new FormData(); body.append("uploadId", uploadId); body.append("file", file, file.name);
    const value = await this.request<{ attachment: unknown }>("/attachments", { method: "POST", body, signal });
    return parseAgentAttachment(value.attachment);
  }
  async download(attachment: AgentAttachment, signal?: AbortSignal): Promise<Blob> {
    const file = parseAgentAttachment(attachment), headers = new Headers(), token = this.token();
    if (token) headers.set("Authorization", `Bearer ${token}`);
    const response = await this.fetcher(`${this.baseUrl}/v1/agent/attachments/${encodeURIComponent(file.id)}/content`, { headers, signal, cache: "no-store", redirect: "error" });
    if (!response.ok) { const error = await response.json().catch(() => null); throw new Error(error?.message || "Attachment download failed."); }
    const blob = await response.blob();
    if (blob.size !== file.sizeBytes || blob.size > 8 * 1024 * 1024) throw new Error("Attachment download was incomplete.");
    return blob;
  }
  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 stopped = false;
  let running: Promise<void> | undefined;
  const cancel = () => { stopped = true; desired = "0"; };
  const flush = (): Promise<void> => {
    if (stopped) return Promise.resolve();
    if (running) return running;
    running = (async () => {
      while (!stopped && BigInt(desired) > BigInt(options.current())) {
        const captured = desired;
        const saved = await options.save(captured);
        if (stopped) return;
        try {
          if (BigInt(normalizeAgentCursor(saved.cursor)) < BigInt(captured)) throw new Error("Agent read state did not acknowledge the seen messages.");
          options.apply(saved);
          // A rejected/source-mismatched acknowledgment must never become a PUT loop.
          if (BigInt(options.current()) < BigInt(captured)) throw new Error("Agent read state did not acknowledge the seen messages.");
        } catch (error) { cancel(); throw error; }
      }
    })().finally(() => { running = undefined; });
    return running;
  };
  return { request(cursor: string) { if (!stopped) desired = advanceAgentCursor(desired, cursor); return flush(); }, flush, cancel };
}

/** Pin reads to the source that owns this rendered room, including queued writes. */
export function createAgentReadSession(options: {
  initial: AgentRoomState;
  save: (cursor: string, sourceId: string | null) => Promise<AgentReadState>;
  apply: (read: AgentReadState) => void;
  invalidated?: (error: AgentReadSourceChangedError) => void;
}) {
  let binding = options.initial.sourceId, acknowledged = options.initial.readCursor;
  let failure: AgentReadSourceChangedError | undefined;
  const invalidate = () => {
    if (!failure) { failure = new AgentReadSourceChangedError(); writer.cancel(); options.invalidated?.(failure); }
    return failure;
  };
  const receive = (read: AgentReadState) => {
    if (failure) throw failure;
    try {
      const nextSource = readSource({ ...options.initial, sourceId: binding, readCursor: acknowledged }, read);
      const nextCursor = advanceAgentCursor(acknowledged, read.cursor);
      binding = nextSource; acknowledged = nextCursor;
      options.apply({ ...read, sourceId: binding });
    } catch (error) { if (error instanceof AgentReadSourceChangedError) throw invalidate(); throw error; }
  };
  const writer = createAgentReadWriter({
    current: () => acknowledged,
    save: async cursor => {
      if (failure || binding === undefined) throw invalidate();
      const source = binding;
      try { return await options.save(cursor, source); }
      catch (error) {
        if (error instanceof AgentReadSourceChangedError || object(error)?.status === 409) throw invalidate();
        throw error;
      }
    },
    apply: receive,
  });
  return {
    receive,
    request(cursor: string, renderedSourceId: string | null | undefined) {
      if (failure) return Promise.reject(failure);
      if (binding === undefined || renderedSourceId !== binding) return Promise.reject(invalidate());
      return writer.request(cursor);
    },
    flush: writer.flush,
    cancel: writer.cancel,
    sourceId: () => binding,
    readCursor: () => acknowledged,
  };
}