feat(agent): add opt-in push notifications

2026-10-14 22:03:21 -07:00
parent 80748f764f
commit a1c0264d36
51 changed files with 4815 additions and 121 deletions
@@ -1,8 +1,8 @@
 import assert from "node:assert/strict";
 import test from "node:test";
 import {
-  AgentRoomClient, advanceAgentCursor, agentAuthFrame, agentDisplayMessages, agentUnreadCount,
-  applyAgentEvent, applyAgentPage, applyAgentReadState, createAgentReadWriter,
+  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";
@@ -16,6 +16,9 @@ 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" } });
@@ -113,6 +116,93 @@ test("read acknowledgments coalesce concurrent seen snapshots and reject unackno
  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 => {
@@ -168,6 +258,18 @@ test("client uses Sybil auth and routes while websocket credentials remain in th
  assert.deepEqual(JSON.parse(agentAuthFrame(token, "42")), { type: "auth", token, after: "42" });
 });

+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 = {};