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";
import { agentMessageTime, compareAgentMessages, normalizeAgentMessage, UNKNOWN_AGENT_TIME } from "./agent-messages.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; assistantImages?: { version: 1 } };
};
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;
  historyLoaded: boolean; beforeCursor: string | null; hasOlder: boolean;
  /** 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;
  const priorTime = agentMessageTime(previous), nextTime = agentMessageTime(incoming);
  if (prior?.kind !== "tool_call" || next?.kind !== "tool_call") return { ...previous, ...incoming,
    createdAt: nextTime ?? priorTime ?? UNKNOWN_AGENT_TIME };
  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: priorTime && nextTime ? (priorTime < nextTime ? priorTime : nextTime) : priorTime ?? nextTime ?? UNKNOWN_AGENT_TIME, 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(normalizeAgentMessage).map(message => [message.id, message]));
  for (const message of incoming.map(normalizeAgentMessage)) {
    const prior = messages.get(message.id);
    messages.set(message.id, prior ? mergeMessage(prior, message) : message);
  }
  return [...messages.values()].sort(compareAgentMessages);
}

function parseAgentPage(value: unknown): AgentMessagesPage {
  const page = object(value);
  if (!page || typeof page.conversationId !== "string" || !Array.isArray(page.messages) || typeof page.hasMore !== "boolean") throw new Error("Invalid agent messages frame.");
  if (page.cursor !== null) normalizeAgentCursor(page.cursor);
  if (page.beforeCursor != null) normalizeAgentCursor(page.beforeCursor);
  return { ...page, beforeCursor: page.beforeCursor ?? null, messages: page.messages.map(normalizeAgentMessage) } as AgentMessagesPage;
}
function parseAgentActivity(value: unknown): AgentActivity {
  const activity = object(value);
  if (!activity || (activity.tools !== undefined && !Array.isArray(activity.tools))
    || (activity.queuedMessages !== undefined && !Array.isArray(activity.queuedMessages))) throw new Error("Invalid agent activity frame.");
  return { ...activity, tools: activity.tools ?? [],
    queuedMessages: ((activity.queuedMessages ?? []) as unknown[]).map(normalizeAgentMessage) } as AgentActivity;
}
export const createAgentRoomState = (): AgentRoomState => ({
  conversationId: null, messages: [], cursor: "0", readCursor: "0", activity: null,
  historyLoaded: false, beforeCursor: null, hasOlder: false,
});
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 {
  page = parseAgentPage(page);
  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 };
}
/** Recent/older windows have their own boundary; live replay never moves it. */
export function applyAgentHistoryPage(state: AgentRoomState, page: AgentMessagesPage, before?: string): AgentRoomState {
  // A retired or repeated older request cannot overwrite a newer window.
  if (before !== undefined && before !== state.beforeCursor) return state;
  const boundary = page.beforeCursor === null ? null : normalizeAgentCursor(page.beforeCursor);
  const cursor = page.cursor === null ? null : normalizeAgentCursor(page.cursor);
  if (page.hasMore && (boundary === null || boundary === "0"
    || (before !== undefined && BigInt(boundary) >= BigInt(before)))) throw new Error("Agent older history did not advance its cursor.");
  if (before !== undefined && ((cursor !== null && BigInt(cursor) >= BigInt(before))
    || (boundary !== null && BigInt(boundary) > BigInt(before)))) throw new Error("Agent older history returned an invalid boundary.");
  const next = applyAgentPage(state, page);
  return { ...next, historyLoaded: true, beforeCursor: boundary, hasOlder: page.hasMore,
    // Older snapshots must not replace newer socket edits or advance delivery/read state.
    ...(before === undefined ? {} : { cursor: state.cursor, messages: mergeAgentMessages(page.messages, state.messages) }),
  };
}
export function agentUnreadCountIsPartial(state: AgentRoomState): boolean {
  return state.hasOlder && state.beforeCursor !== null && BigInt(state.readCursor) + 1n < BigInt(state.beforeCursor);
}
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: parseAgentActivity(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") {
    return { ...parseAgentPage(event), type: "messages" };
  }
  if (event.type === "activity") {
    if (typeof event.conversationId !== "string") throw new Error("Invalid agent activity frame.");
    return { ...event, activity: parseAgentActivity(event.activity) } as AgentRoomEvent;
  }
  return event as AgentRoomEvent;
}
export function agentAuthFrame(token: string | null, cursor: string, presenceSessionId?: string): string {
  return JSON.stringify({ type: "auth", token, after: normalizeAgentCursor(cursor),
    clientKind: "web", ...(agentPushUuid(presenceSessionId) ? { 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;
  private activeDownloads = 0;
  private readonly downloadWaiters = new Set<() => void>();
  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;
  }
  async descriptor(signal?: AbortSignal): Promise<AgentDescriptor> {
    const value = object(await this.request("", { signal }));
    if (!value || typeof value.id !== "string") throw new Error("Invalid agent descriptor.");
    return { ...value, activity: parseAgentActivity(value.activity) } as AgentDescriptor;
  }
  async messages(after: string, signal?: AbortSignal): Promise<AgentMessagesPage> {
    return parseAgentPage(await this.request(`/messages?${new URLSearchParams({ after: normalizeAgentCursor(after), limit: "200" })}`, { signal }));
  }
  async history(before?: string, signal?: AbortSignal): Promise<AgentMessagesPage> {
    return parseAgentPage(await this.request(`/messages?${new URLSearchParams({ ...(before === undefined ? {} : { before: normalizeAgentCursor(before) }), limit: "100" })}`, { 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();
    signal?.throwIfAborted();
    while (this.activeDownloads >= 2) {
      await new Promise<void>((resolve, reject) => {
        const cleanup = () => { this.downloadWaiters.delete(wake); signal?.removeEventListener("abort", abort); };
        const wake = () => { cleanup(); resolve(); };
        const abort = () => { cleanup(); reject(signal?.reason || new Error("Download canceled.")); };
        this.downloadWaiters.add(wake); signal?.addEventListener("abort", abort, { once: true });
        if (signal?.aborted) abort();
      });
      signal?.throwIfAborted();
    }
    this.activeDownloads++;
    try {
      if (token !== this.token()) throw new Error("Authentication changed. Retry the download.");
      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) { await response.body?.cancel(); throw new Error("Attachment download failed. Retry the download."); }
      const reader = response.body?.getReader();
      if (!reader) throw new Error("Attachment download was incomplete.");
      const bytes = new Uint8Array(file.sizeBytes); let size = 0;
      try {
        for (;;) {
          const { done, value } = await reader.read(); if (done) break;
          signal?.throwIfAborted();
          if (size + value.length > bytes.length) throw new Error("Attachment download exceeds its size limit.");
          bytes.set(value, size); size += value.length;
        }
      } finally { await reader.cancel().catch(() => {}); reader.releaseLock(); }
      if (size !== file.sizeBytes) throw new Error("Attachment download was incomplete.");
      const sha256 = [...new Uint8Array(await crypto.subtle.digest("SHA-256", bytes))].map(n => n.toString(16).padStart(2, "0")).join("");
      if (sha256 !== file.sha256) throw new Error("Attachment checksum did not match.");
      if (file.kind === "image" && response.headers.get("content-type")?.split(";")[0].trim() !== file.mimeType)
        throw new Error("Attachment image type did not match.");
      signal?.throwIfAborted();
      if (token !== this.token()) throw new Error("Authentication changed. Retry the download.");
      return new Blob([bytes], { type: response.headers.get("content-type") || "application/octet-stream" });
    } finally {
      this.activeDownloads--;
      this.downloadWaiters.values().next().value?.();
    }
  }
  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,
  };
}