Sybil-2/web/tests/agent-room.test.mjs

import assert from "node:assert/strict";
import test from "node:test";
import {
  AgentRoomClient, AgentReadSourceChangedError, advanceAgentCursor, agentAuthFrame, agentDisplayMessages, agentUnreadCount,
  applyAgentEvent, applyAgentPage, applyAgentReadState, createAgentReadSession, createAgentReadWriter,
  createAgentRoomState, mergeAgentMessages, normalizeAgentCursor, parseAgentRoomEvent, syncAgentHistory,
  applyAgentHistoryPage, agentUnreadCountIsPartial,
} from "../src/lib/agent-room.ts";
import { createAgentOptimisticOutbox } from "../src/lib/agent-optimistic-outbox.ts";

const message = (id, role = "assistant", extra = {}) => ({
  id: `entry:${id}`, createdAt: `2026-10-14T10:00:${String(id % 60).padStart(2, "0")}.000Z`,
  role, content: `message ${id}`, name: role === "assistant" ? "Sybil Agent" : "buzzert",
  metadata: { kind: "message", entryId: id, status: "complete" }, ...extra,
});
const page = (cursor, messages = [], hasMore = false) => ({
  conversationId: "1", messages, cursor, beforeCursor: messages.length ? String(messages[0].metadata.entryId) : cursor, hasMore,
});
const activity = (queuedMessages = []) => ({ state: "idle", isThinking: false, queued: queuedMessages.length, backgroundWorkers: 0, tools: [], queuedMessages });
const sourceA = "4599d348-e053-4937-987d-1392746fb2ce", sourceB = "22222222-2222-4222-8222-222222222222";
const deferred = () => { let resolve; const promise = new Promise(yes => { resolve = yes; }); return { promise, resolve }; };
const sourceRoom = () => applyAgentPage(applyAgentReadState(createAgentRoomState(), { sourceId: sourceA, cursor: "90", updatedAt: null }), page("100", [message(100)]));

test("tool starts and results merge across replay without downgrading completed cards", () => {
  const start = message(2, "tool", { id: "tool:1:call1", metadata: { kind: "tool_call", entryId: 2, status: "initiated", args: { query: "cookie" }, startedAt: "2026-10-14T10:00:00Z" } });
  const done = message(3, "tool", { id: start.id, metadata: { kind: "tool_call", entryId: 3, status: "completed", resultPreview: "found" } });
  const messages = mergeAgentMessages(mergeAgentMessages([], [start]), [done, start]);
  assert.equal(messages.length, 1);
  assert.equal(messages[0].metadata.status, "completed");
  assert.equal(messages[0].metadata.resultPreview, "found");
  assert.deepEqual(messages[0].metadata.args, { query: "cookie" });
  assert.equal(messages[0].metadata.entryId, 3);
  assert.equal(messages[0].createdAt, start.createdAt);
});

test("forward recovery reads every missing page including invisible entries and counts only unread assistant messages", async () => {
  const requests = [];
  const responses = [page("2", [message(1), message(2, "user")], true), page("4", [], true), page("7", [message(5), message(6, "tool"), message(7)], false)];
  let state = createAgentRoomState();
  await syncAgentHistory({ messages: async after => { requests.push(after); return responses.shift(); } }, "0", update => { state = applyAgentPage(state, update); });
  assert.deepEqual(requests, ["0", "2", "4"]);
  assert.equal(state.cursor, "7");
  assert.equal(agentUnreadCount(state), 3);
  state = applyAgentReadState(state, { cursor: "5", updatedAt: null });
  assert.equal(agentUnreadCount(state), 1);
  state = applyAgentReadState(state, { cursor: "1", updatedAt: null });
  assert.equal(state.readCursor, "5", "a stale tab response cannot regress read state");
  assert.equal(agentUnreadCount(state), 1);
});

test("recent history is one cursorless window and older history uses the returned source boundary", async () => {
  const requests = [];
  const client = new AgentRoomClient({ fetch: async (url, options) => {
    requests.push({ url, signal: options.signal });
    return new Response(JSON.stringify({ ...page("1200", [message(1200)], true), beforeCursor: "1101" }));
  } });
  const controller = new AbortController();
  let state = applyAgentHistoryPage(createAgentRoomState(), await client.history(undefined, controller.signal));
  assert.equal(requests.length, 1, "hasMore must not drain old history");
  assert.equal(requests[0].url, "/api/v1/agent/messages?limit=100");
  assert.equal(requests[0].signal, controller.signal);
  assert.equal(state.historyLoaded, true);
  assert.equal(state.cursor, "1200");
  assert.equal(state.beforeCursor, "1101", "boundary is not derived from visible message 1200");
  assert.equal(state.hasOlder, true);
  await client.history(state.beforeCursor, controller.signal);
  assert.equal(requests[1].url, "/api/v1/agent/messages?before=1101&limit=100");
});

test("delayed older pages preserve newer edits, live cursor, source and read state", () => {
  let state = applyAgentHistoryPage(sourceRoom(), { ...page("100", [message(100)], true), beforeCursor: "90" });
  state = applyAgentPage(state, page("110", [message(100, "assistant", { content: "New streamed text" }), message(110)]));
  const next = applyAgentHistoryPage(state, { ...page("89", [message(80), message(100, "assistant", { content: "Old snapshot" })], true), beforeCursor: "70" }, "90");
  assert.equal(next.messages.find(m => m.id === "entry:100").content, "New streamed text");
  assert.equal(next.cursor, "110");
  assert.equal(next.readCursor, "90");
  assert.equal(next.sourceId, sourceA);
  assert.equal(next.beforeCursor, "70");
  assert.equal(next.hasOlder, true);
  assert.equal(applyAgentHistoryPage(next, page("89", [message(80)], false), "90"), next, "retired boundary cannot replace the next window");
});

test("empty internal pages advance older boundaries without chasing another page or moving read state", () => {
  const initial = applyAgentHistoryPage(createAgentRoomState(), { ...page("1200", [], true), beforeCursor: "1101" });
  assert.equal(initial.historyLoaded, true);
  assert.equal(initial.messages.length, 0);
  const next = applyAgentHistoryPage(initial, { ...page("1100", [], true), beforeCursor: "1001" }, "1101");
  assert.equal(next.beforeCursor, "1001");
  assert.equal(next.cursor, "1200");
  assert.equal(next.readCursor, "0");
  assert.equal(next.hasOlder, true);
  const end = applyAgentHistoryPage(next, { ...page("0"), beforeCursor: "1001" }, "1001");
  assert.equal(end.hasOlder, false);
  assert.equal(end.cursor, "1200");
});

test("malformed or stalled older pages cannot change history or delivery boundaries", () => {
  const state = applyAgentHistoryPage(createAgentRoomState(), { ...page("100", [message(100)], true), beforeCursor: "90" });
  for (const bad of [
    { ...page("89", [], true), beforeCursor: "90" },
    { ...page("89", [], true), beforeCursor: null },
    { ...page("91", [], false), beforeCursor: "80" },
    { ...page("89", [], false), beforeCursor: "91" },
    { ...page("bad", [], false), beforeCursor: "80" },
  ]) assert.throws(() => applyAgentHistoryPage(state, bad, "90"), /cursor|boundary/);
  assert.throws(() => applyAgentHistoryPage(state, { ...page("89"), conversationId: "different" }, "90"), /conversation changed/);
  assert.equal(state.beforeCursor, "90");
  assert.equal(state.cursor, "100");
});

test("unread counts disclose unknown older replies until reading the latest rendered window", () => {
  let state = applyAgentHistoryPage(createAgentRoomState(), { ...page("1200", [message(1199), message(1200, "user")], true), beforeCursor: "1101" });
  assert.equal(agentUnreadCount(state), 1);
  assert.equal(agentUnreadCountIsPartial(state), true);
  state = applyAgentReadState(state, { cursor: "1100", updatedAt: null });
  assert.equal(agentUnreadCountIsPartial(state), false);
  assert.equal(agentUnreadCount(state), 1);
  state = applyAgentReadState(state, { cursor: "1200", updatedAt: null });
  assert.equal(agentUnreadCount(state), 0);
  assert.equal(state.hasOlder, true, "reading latest does not fetch or discard older pages");
});

test("queued user messages keep stable IDs and disappear if withdrawn or promoted", () => {
  const pending = message(1, "user", { id: "inbound:sybil:request1", metadata: { kind: "message", status: "queued" } });
  let state = applyAgentEvent(createAgentRoomState(), { type: "activity", conversationId: "1", activity: activity([pending]) });
  assert.equal(agentDisplayMessages(state).length, 1);
  assert.equal(state.cursor, "0", "pending rows never advance replay or read cursors");
  assert.equal(agentUnreadCount(state), 0);
  state = applyAgentPage(state, page("4", [message(4, "user", { id: pending.id })]));
  assert.equal(agentDisplayMessages(state).length, 1);
  assert.equal(agentDisplayMessages(state)[0].metadata.status, "complete");
  state = applyAgentEvent(state, { type: "activity", conversationId: "1", activity: activity() });
  assert.equal(agentDisplayMessages(state).length, 1, "committed rows survive queue removal");
  const onlyPending = applyAgentEvent(createAgentRoomState(), { type: "activity", conversationId: "1", activity: activity([pending]) });
  assert.deepEqual(agentDisplayMessages(applyAgentEvent(onlyPending, { type: "activity", conversationId: "1", activity: activity() })), []);
});

test("failed or stalled page delivery preserves the last applied cursor for replay", async () => {
  let state = createAgentRoomState();
  let calls = 0;
  await assert.rejects(syncAgentHistory({ messages: async () => {
    if (++calls === 1) return page("10", [message(10)], true);
    throw new Error("network unavailable");
  } }, "0", next => { state = applyAgentPage(state, next); }), /network unavailable/);
  assert.equal(state.cursor, "10");
  await assert.rejects(syncAgentHistory({ messages: async () => page("10", [], true) }, state.cursor, next => { state = applyAgentPage(state, next); }), /did not advance/);
  assert.equal(state.cursor, "10");
  assert.throws(() => applyAgentPage(state, page("bad", [message(11)])), /cursor/);
  assert.equal(state.messages.length, 1);
  assert.throws(() => applyAgentPage(state, { ...page("11"), conversationId: "different" }), /conversation changed/);
  assert.equal(advanceAgentCursor("9007199254740993", "9007199254740992"), "9007199254740993");
  assert.throws(() => normalizeAgentCursor("-1"), /cursor/);
});

test("read acknowledgments use only the rendered cursor and failed persistence retains unread replies", async () => {
  let state = applyAgentPage(createAgentRoomState(), page("5", [message(5)]));
  const renderedCursor = state.cursor;
  state = applyAgentPage(state, page("6", [message(6)]));
  const saved = [];
  let fail = true;
  const writer = createAgentReadWriter({
    current: () => state.readCursor,
    save: async cursor => { saved.push(cursor); if (fail) throw new Error("database offline"); return { cursor, updatedAt: null }; },
    apply: read => { state = applyAgentReadState(state, read); },
  });
  await assert.rejects(writer.request(renderedCursor), /database offline/);
  assert.equal(state.readCursor, "0");
  assert.equal(agentUnreadCount(state), 2);
  fail = false;
  await writer.flush();
  assert.deepEqual(saved, ["5", "5"]);
  assert.equal(state.readCursor, "5");
  assert.equal(agentUnreadCount(state), 1, "message6 arrived after render and must remain unread");
});

test("read acknowledgments coalesce concurrent seen snapshots and reject unacknowledged cursors", async () => {
  let cursor = "0";
  let release;
  const gate = new Promise(resolve => { release = resolve; });
  const saves = [];
  const writer = createAgentReadWriter({ current: () => cursor, save: async value => {
    saves.push(value); if (value === "3") await gate; return { cursor: value, updatedAt: null };
  }, apply: read => { cursor = read.cursor; } });
  const first = writer.request("3"), second = writer.request("7");
  release(); await Promise.all([first, second]);
  assert.deepEqual(saves, ["3", "7"]);
  assert.equal(cursor, "7");
  const broken = createAgentReadWriter({ current: () => "0", save: async () => ({ cursor: "0", updatedAt: null }), apply: () => assert.fail("must not apply a failed acknowledgment") });
  await assert.rejects(broken.request("2"), /did not acknowledge/);
});

test("source replacement through GET or WS quiesces queued old-history reads without relabeling their cursor", async () => {
  for (const via of ["GET", "WS"]) {
    let room = sourceRoom(); const saved = [], invalidated = [];
    const session = createAgentReadSession({ initial: room,
      save: async (cursor, sourceId) => { saved.push({ cursor, sourceId }); throw new Error("temporarily offline"); },
      apply: read => { room = applyAgentReadState(room, read); }, invalidated: error => invalidated.push(error),
    });
    await assert.rejects(session.request("100", sourceA), /temporarily offline/);
    const replacement = { cursor: "0", sourceId: sourceB, updatedAt: null };
    const incoming = via === "WS" ? parseAgentRoomEvent(JSON.stringify({ type: "read_state", ...replacement })) : replacement;
    assert.throws(() => session.receive(incoming), AgentReadSourceChangedError);
    for (let i = 0; i < 3; i++) await session.flush();
    await assert.rejects(session.request("100", sourceA), AgentReadSourceChangedError);
    await assert.rejects(session.request("5", sourceB), AgentReadSourceChangedError);
    assert.deepEqual(saved, [{ cursor: "100", sourceId: sourceA }], `${via}: no write to B and no retry loop`);
    assert.equal(room.sourceId, sourceA); assert.equal(room.readCursor, "90"); assert.equal(room.cursor, "100");
    assert.equal(session.sourceId(), sourceA); assert.equal(invalidated.length, 1);
  }
});

test("source replacement cancels an in-flight old read and prevents a coalesced larger write", async () => {
  let room = sourceRoom(); const gate = deferred(), saved = [];
  const session = createAgentReadSession({ initial: room,
    save: async (cursor, sourceId) => { saved.push({ cursor, sourceId }); return gate.promise; },
    apply: read => { room = applyAgentReadState(room, read); },
  });
  const first = session.request("100", sourceA), second = session.request("110", sourceA);
  assert.throws(() => session.receive({ cursor: "0", sourceId: sourceB, updatedAt: null }), AgentReadSourceChangedError);
  gate.resolve({ cursor: "100", sourceId: sourceA, updatedAt: null });
  await Promise.all([first, second]); await session.flush();
  assert.deepEqual(saved, [{ cursor: "100", sourceId: sourceA }]);
  assert.equal(room.sourceId, sourceA); assert.equal(room.readCursor, "90");
});

test("mismatched acknowledgment and server409 stop retries without applying a read to new history", async () => {
  for (const mode of ["ack", "409"]) {
    let room = sourceRoom(), saves = 0, invalidated = 0;
    const session = createAgentReadSession({ initial: room,
      save: async () => { saves++; if (mode === "409") throw Object.assign(new Error("cursor not valid for this source"), { status: 409 });
        return { cursor: "100", sourceId: sourceB, updatedAt: null }; },
      apply: read => { room = applyAgentReadState(room, read); }, invalidated: () => invalidated++,
    });
    await assert.rejects(session.request("100", sourceA), AgentReadSourceChangedError);
    await session.flush(); await session.flush();
    await assert.rejects(session.request("101", sourceA), AgentReadSourceChangedError);
    assert.equal(saves, 1); assert.equal(invalidated, 1); assert.equal(room.readCursor, "90"); assert.equal(room.sourceId, sourceA);
  }
});

test("read writer stops after an ignored acknowledgment instead of repeating the same PUT forever", async () => {
  let saves = 0;
  const writer = createAgentReadWriter({ current: () => "0", save: async cursor => { saves++; return { cursor, updatedAt: null }; }, apply: () => {} });
  await assert.rejects(writer.request("2"), /did not acknowledge/);
  await writer.flush(); await writer.request("3"); assert.equal(saves, 1);
});

test("legacy source-absent sessions stay compatible but cannot adopt a source after rendering old history", async () => {
  let room = applyAgentPage(createAgentRoomState(), page("5", [message(5)])); const saved = [];
  const session = createAgentReadSession({ initial: room,
    save: async (cursor, sourceId) => { saved.push({ cursor, sourceId }); return { cursor, updatedAt: null }; },
    apply: read => { room = applyAgentReadState(room, read); },
  });
  session.receive({ cursor: "0", updatedAt: null });
  assert.equal(room.sourceId, null);
  await session.request("5", room.sourceId);
  assert.deepEqual(saved, [{ cursor: "5", sourceId: null }]); assert.equal(room.readCursor, "5");
  assert.throws(() => session.receive({ cursor: "0", sourceId: sourceB, updatedAt: null }), AgentReadSourceChangedError);
  await session.flush(); assert.equal(saved.length, 1);
  assert.throws(() => applyAgentReadState(applyAgentPage(createAgentRoomState(), page("5", [message(5)])), { cursor: "0", sourceId: sourceA, updatedAt: null }), AgentReadSourceChangedError);
  assert.throws(() => applyAgentReadState(sourceRoom(), { cursor: "0", updatedAt: null }), AgentReadSourceChangedError);
});

test("rendered snapshot source must match the pinned session and new sessions cannot read before binding", async () => {
  for (const initial of [createAgentRoomState(), sourceRoom()]) {
    const session = createAgentReadSession({ initial, save: async () => assert.fail("must not send unbound or relabeled reads"), apply: () => {} });
    await assert.rejects(session.request("100", sourceB), AgentReadSourceChangedError); await session.flush();
  }
  let room = createAgentRoomState(); const saved = [];
  const session = createAgentReadSession({ initial: room,
    save: async (cursor, sourceId) => { saved.push({ cursor, sourceId }); return { cursor, sourceId, updatedAt: null }; },
    apply: read => { room = applyAgentReadState(room, read); },
  });
  session.receive({ cursor: "0", sourceId: sourceA, updatedAt: null });
  room = applyAgentPage(room, page("5", [message(5)])); await session.request(room.cursor, room.sourceId);
  assert.deepEqual(saved, [{ cursor: "5", sourceId: sourceA }]); assert.equal(room.readCursor, "5");
});

test("failed persistence after acknowledgment retains the original receipt key across reload", async () => {
  let stored = [], writes = 0;
  const storage = { read: () => structuredClone(stored), write: rows => {
    if (++writes === 3) throw new Error("storage unavailable after acknowledgment");
    stored = structuredClone(rows);
  } };
  const requests = [];
  const open = () => createAgentOptimisticOutbox({
    storage, changed: () => {}, newId: () => "stable-request",
    submit: async (_content, requestId) => {
      requests.push(requestId);
      return { status: "placed", messageId: `inbound:sybil:${requestId}` };
    },
  });
  const original = open();
  await original.send("retry me");
  const accepted = original.display([])[0];
  assert.equal(accepted.metadata.status, "complete", "a storage error cannot undo server acceptance");
  assert.equal(stored[0].status, "sending", "the pre-admission receipt remains durable if its update fails");
  const restored = open();
  assert.deepEqual(requests, ["stable-request"], "reload never retries automatically");
  const uncertain = restored.display([])[0];
  assert.equal(uncertain.metadata.status, "unconfirmed");
  assert.equal(uncertain.id, accepted.id);
  assert.equal(uncertain.createdAt, accepted.createdAt);
  await restored.retry(uncertain.id);
  assert.deepEqual(requests, ["stable-request", "stable-request"], "manual retry cannot create a second admission");
  restored.reconcile({ messages: [message(1, "user", { id: accepted.id, content: "retry me" })] });
  assert.deepEqual(stored, [], "canonical echo finishes local cleanup");
});

test("client uses Sybil auth and routes while websocket credentials remain in the first frame", async () => {
  const calls = [];
  let token = "first-session-token";
  const client = new AgentRoomClient({ baseUrl: "https://sybil.example/api/", token: () => token, fetch: async (url, init) => {
    calls.push({ url, init }); return new Response(JSON.stringify(init.method === "POST"
      ? { id: 1, status: "queued", messageId: "inbound:sybil:request-1" } : { cursor: "0", updatedAt: null }), { status: 200 });
  } });
  await client.messages("0");
  token = "second-session-token";
  await client.markRead("42");
  await client.submit("hi", "request-1");
  assert.equal(calls[0].url, "https://sybil.example/api/v1/agent/messages?after=0&limit=200");
  assert.equal(calls[0].init.headers.get("Authorization"), "Bearer first-session-token");
  assert.equal(calls[1].init.headers.get("Authorization"), "Bearer second-session-token");
  assert.equal(calls[1].init.method, "PUT");
  assert.deepEqual(JSON.parse(calls[1].init.body), { cursor: "42" });
  assert.deepEqual(JSON.parse(calls[2].init.body), { content: "hi", requestId: "request-1" });
  const url = client.websocketUrl("https://sybil.example/");
  assert.equal(url, "wss://sybil.example/api/v1/agent/ws");
  assert.equal(new URL(url).search, "");
  assert.deepEqual(JSON.parse(agentAuthFrame(null, "42")), { type: "auth", token: null, after: "42", clientKind: "web" });
  assert.deepEqual(JSON.parse(agentAuthFrame(token, "42")), { type: "auth", token, after: "42", clientKind: "web" });
});

test("client sends the pinned source and exposes409 without requiring a JSON response body", async () => {
  for (const json of [true, false]) {
    let body;
    const client = new AgentRoomClient({ fetch: async (_url, init) => {
      body = JSON.parse(init.body);
      return json ? Response.json({ message: "Read source changed" }, { status: 409 }) : new Response(null, { status: 409 });
    } });
    await assert.rejects(client.markRead("100", undefined, sourceA), error => error.status === 409);
    assert.deepEqual(body, { cursor: "100", sourceId: sourceA });
  }
});

test("malformed successful receipts keep optimistic messages retryable under the original request ID", async () => {
  const requests = [];
  let response = {};
  const client = new AgentRoomClient({ fetch: async (_url, init) => {
    requests.push(JSON.parse(init.body));
    return new Response(JSON.stringify(response), { status: 202 });
  } });
  const box = createAgentOptimisticOutbox({ submit: (content, requestId) => client.submit(content, requestId),
    storage: { read: () => [], write: () => {} }, changed: () => {}, newId: () => "retry-id",
  });
  await assert.rejects(box.send("Keep this request"), /invalid message receipt/);
  const row = box.display([])[0];
  assert.equal(row.metadata.status, "failed");
  assert.equal(box.snapshot()[0].seenOnServer, false);
  const valid = { id: 1, status: "queued", messageId: row.id };
  for (const malformed of [null, { error: "Invalid Sybil Agent response" }, { ...valid, id: 0 }, { ...valid, id: 1.5 },
    { ...valid, id: Number.MAX_SAFE_INTEGER + 1 }, { ...valid, status: "unknown" }, { ...valid, status: ["queued"] },
    { ...valid, messageId: "inbound:sybil:other" }]) {
    response = malformed;
    await assert.rejects(box.retry(row.id), /invalid message receipt/);
    assert.equal(box.display([])[0].metadata.status, "failed");
  }
  response = valid;
  await box.retry(row.id);
  assert.equal(box.display([])[0].metadata.status, "queued");
  assert.ok(requests.every(request => request.requestId === "retry-id"), "uncertain responses cannot mint a duplicate admission");
  for (const status of ["placed", "done", "unanswered"]) {
    response = { ...valid, status };
    assert.equal((await client.submit("Keep this request", "retry-id")).status, status);
  }
});

test("protocol ignores unknown frames and rejects malformed message cursors", () => {
  assert.equal(parseAgentRoomEvent('{"type":"future-frame"}'), null);
  assert.deepEqual(parseAgentRoomEvent('{"type":"pong"}'), { type: "pong" });
  assert.throws(() => parseAgentRoomEvent(JSON.stringify({ type: "messages", ...page("oops") })), /cursor/);
  const state = applyAgentEvent(createAgentRoomState(), { type: "activity", conversationId: "1", activity: { ...activity(), state: "thinking", isThinking: true } });
  assert.equal(state.activity.isThinking, true);
  assert.deepEqual(state.messages, [], "a thinking indicator never becomes a chat message");
});