313 lines
19 KiB
JavaScript
313 lines
19 KiB
JavaScript
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,
|
|
} 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("initial synchronization reads every 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("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");
|
|
});
|