Reconcile agent notifications from events and pending deadlines

server/tests/agent-push-events.test.ts

@@ -0,0 +1,24 @@
+import assert from 'node:assert/strict';
+import { once } from 'node:events';
+import { createServer } from 'node:http';
+import test from 'node:test';
+import { WebSocketServer } from 'ws';
+import { subscribeNotificationEvents } from '../src/agent-push-events.js';
+
+test('notification observer authenticates, reconnects and stops without leaking missed invalidations', async t => {
+  const server = createServer(); const sockets = new WebSocketServer({ server });
+  server.listen(0, '127.0.0.1'); await once(server, 'listening');
+  let connections = 0, wakes = 0;
+  sockets.on('connection', (socket, request) => {
+    connections++;
+    assert.equal(request.url, '/api/notification-events');
+    assert.equal(request.headers.authorization, 'Bearer fixture-key');
+    assert.equal(request.headers['x-sybil-client'], 'backend-observer-v1');
+    socket.send(JSON.stringify({ type: 'notification_changed', version: 1 }));
+  });
+  const stop = subscribeNotificationEvents({ baseUrl: `http://127.0.0.1:${(server.address() as any).port}`, apiKey: 'fixture-key' }, () => { wakes++; }, { retryMs: 20 });
+  t.after(async () => { stop(); for (const socket of sockets.clients) socket.terminate(); sockets.close(); await new Promise<void>(resolve => server.close(() => resolve())); });
+  const waitFor = async (count: number) => { const deadline = Date.now() + 3000; while (wakes < count) { assert.ok(Date.now() < deadline); await new Promise(resolve => setTimeout(resolve, 10)); } };
+  await waitFor(1); for (const socket of sockets.clients) socket.terminate(); await waitFor(2);
+  assert.equal(connections, 2); stop(); await new Promise(resolve => setTimeout(resolve, 80)); assert.equal(connections, 2);
+});

server/tests/agent-push.test.ts

@@ -51,7 +51,7 @@ async function fixture(options: { warmupMs?: number; config?: PushConfig } = {})
   const installationId = randomUUID(); await manager.register(installationId, apns); await manager.tick();
   return { manager, installationId, source, sent, reads, get now() { return now; }, advance(ms: number) { now += ms; },
     result(value: SendResult) { result = value; }, failure(value: boolean) { failure = value; },
-    context(value: Partial<Context>) { context = { ...context, ...value }; },
+    context(value: Partial<Context>) { context = { ...context, ...value }; manager.wake(); },
     candidate(cursor = '11', trigger: Context['candidates'][number]['trigger'] = 'user') { context = { ...context, cursor,
       candidates: [...context.candidates, { cursor, messageId: `outbound:${cursor}`, createdAt: new Date(now).toISOString(), trigger }] }; },
   };
@@ -439,3 +439,40 @@ test('endpoint rebind also retires disabled registrations from a prior credentia
   await assert.rejects(current.register(f.installationId, apns), { statusCode: 409 });
   await current.revoke(newer); assert.equal(await prisma.agentPushInstallation.count({ where: { enabled: true } }), 0);
 });
+
+
+async function eventually(predicate: () => Promise<boolean> | boolean) {
+  const deadline = Date.now() + 5000;
+  while (!await predicate()) { if (Date.now() > deadline) assert.fail('event-driven reconciliation timed out'); await new Promise(resolve => setTimeout(resolve, 10)); }
+}
+test('event scheduler stays idle, coalesces wakes, dispatches at the deadline and retains events during a snapshot', async t => {
+  const f = await fixture();
+  let wake!: () => void, observations = 0, unsubscribed = false, wakeDuringFetch = false;
+  const original = f.manager.options.context;
+  const manager = new AgentPushCoordinator({ ...f.manager.options, now: Date.now, subscribe: callback => { wake = callback; return () => { unsubscribed = true; }; },
+    context: async (...args) => { observations++; const result = await original(...args);
+      if (wakeDuringFetch) { wakeDuringFetch = false; wake(); }
+      return { ...result, webPresence: { ...result.webPresence, observedAt: new Date().toISOString() } }; } });
+  t.after(() => manager.stop()); manager.start();
+  await eventually(() => observations === 1); await new Promise(resolve => setTimeout(resolve, 2300));
+  assert.equal(observations, 1, 'no two-second source polling while idle');
+  f.candidate(); wake(); wake(); wake();
+  await eventually(async () => Boolean(await prisma.agentPushDelivery.findFirst({ where: { status: 'pending' } })));
+  await eventually(() => f.sent.length === 1);
+  const settled = observations;
+  await new Promise(resolve => setTimeout(resolve, 300)); assert.equal(observations, settled, 'no repeated polling after the outbox drains');
+  wakeDuringFetch = true; wake(); await eventually(() => observations >= settled + 2);
+  await manager.stop(); assert.equal(unsubscribed, true);
+});
+
+test('presence wake releases long suppression promptly but cannot shorten a provider retry', async () => {
+  const f = await fixture(), tab = randomUUID();
+  await f.manager.presence(tab, { sequence: 1, visibility: 'visible', focused: true, interacted: false });
+  f.candidate(); await f.manager.tick(); f.advance(2000); await f.manager.tick();
+  assert.equal((await prisma.agentPushDelivery.findFirstOrThrow()).dueAt.getTime(), f.now + 58000, 'focus waits for the lease, not another two-second tick');
+  await f.manager.presence(tab, { sequence: 2, visibility: 'hidden', focused: false, interacted: false });
+  await f.manager.tick(); assert.equal((await prisma.agentPushDelivery.findFirstOrThrow()).dueAt.getTime(), f.now + 5000);
+  f.advance(5000); f.result({ outcome: 'retry', retryAfterMs: 10000 }); await f.manager.tick(); assert.equal(f.sent.length, 1);
+  f.advance(100); f.manager.wake(); await f.manager.tick(); assert.equal(f.sent.length, 1, 'event does not bypass provider backoff');
+  f.advance(9900); f.result({ outcome: 'accepted' }); await f.manager.tick(); assert.equal(f.sent.length, 2);
+});

server/tests/agent.test.ts

@@ -48,6 +48,7 @@ const upstream = createServer(async (req, res) => {
 const sourceSockets = new WebSocketServer({ noServer: true });
 upstream.on('upgrade', (req, socket, head) => {
   assert.equal(req.headers.authorization, `Bearer ${privateKey}`);
+  if (req.url === '/api/notification-events') { socket.end('HTTP/1.1 404 Not Found\r\nConnection: close\r\n\r\n'); return; }
   assert.equal(req.url, '/api/ws');
   connections++;
   sourceSockets.handleUpgrade(req, socket, head, ws => {