Sybil-2/server/tests/agent-push.test.ts

import assert from 'node:assert/strict';
import { createECDH, generateKeyPairSync, randomUUID, verify } from 'node:crypto';
import { EventEmitter, once } from 'node:events';
import { createServer } from 'node:http';
import { execFileSync } from 'node:child_process';
import { mkdtempSync, rmSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { dirname, join, resolve } from 'node:path';
import test from 'node:test';
import { fileURLToPath } from 'node:url';
import Fastify from 'fastify';
import sensible from '@fastify/sensible';
import webpush from 'web-push';
import type { Context, Registration } from '../src/agent-push-contract.js';
import type { PushConfig } from '../src/agent-push-config.js';
import type { SendResult } from '../src/agent-push-transports.js';

const root = resolve(dirname(fileURLToPath(import.meta.url)), '..');
const directory = mkdtempSync(join(tmpdir(), 'sybil-push-test-'));
process.env.DATABASE_URL = `file:${join(directory, 'push.db')}`;
process.env.ADMIN_TOKEN = 'push-test-bearer-0123456789';
process.env.AGENT_PUSH_ENABLED = 'false';
for (const key of ['AGENT_PUSH_APNS_KEY_FILE','AGENT_PUSH_APNS_KEY_ID','AGENT_PUSH_APNS_TEAM_ID','AGENT_PUSH_VAPID_FILE']) delete process.env[key];
execFileSync(process.execPath, [join(root, 'node_modules/prisma/build/index.js'), 'migrate', 'deploy'], { cwd: root, env: process.env, stdio: 'pipe' });
const [{ prisma }, { AgentPushCoordinator, registerPushRoutes }, { createPushSender, isPublicAddress, createAPNsRequest, createWebRequest }, { validatePushEndpoint }, { env }] = await Promise.all([
  import('../src/db.js'), import('../src/agent-push.js'), import('../src/agent-push-transports.js'), import('../src/agent-push-contract.js'), import('../src/env.js'),
]);
const apns: Registration = { transport: 'apns', enabled: true, token: 'a'.repeat(64), environment: 'sandbox', topic: 'net.buzzert.sybil2' };
const config: PushConfig = { enabled: true, adminToken: process.env.ADMIN_TOKEN,
  apns: { key: 'fixture-unused-by-injected-sender', keyId: 'ABCDEFGHIJ', teamId: 'DQQH5H6GBD', topic: apns.topic, environment: 'sandbox' },
  web: { publicKey: 'fixture-unused', privateKey: 'fixture-unused', subject: 'mailto:tests@example.com' } };
function subscription(): Registration {
  const curve = createECDH('prime256v1'); curve.generateKeys();
  return { transport: 'web', enabled: true, subscription: { endpoint: `https://web.push.apple.com/test/${randomUUID()}`, expirationTime: null,
    keys: { p256dh: curve.getPublicKey().toString('base64url'), auth: Buffer.alloc(16, 1).toString('base64url') } } };
}
async function fixture(options: { warmupMs?: number; config?: PushConfig } = {}) {
  let now = Date.now();
  const source = randomUUID();
  let context: Context = { version: 1, sourceId: source, conversationId: '1', cursor: '10', hasMore: false,
    latestUserMessageAt: new Date(now - 30_000).toISOString(), nativeReadCursor: '0', uncertain: false,
    webPresence: { active: false, unknown: false, observedAt: new Date(now).toISOString() }, candidates: [] };
  const sent: { registration: Registration; identity: any; messageText?: string }[] = [], reads: any[] = [];
  let result: SendResult = { outcome: 'accepted' }, failure = false;
  const manager = new AgentPushCoordinator({ db: prisma, config: options.config || config, now: () => now,
    warmupMs: options.warmupMs ?? 0, context: async after => { if (failure) throw new Error('upstream secret must not leak');
      return { ...context, webPresence: { ...context.webPresence, observedAt: new Date(now).toISOString() }, candidates: after === undefined ? [] : context.candidates.filter(c => Number(c.cursor) > Number(after)) }; },
    messageText: async () => 'The Agent finished the requested work.',
    send: async (registration, identity, _expiresAt, _now, messageText) => { sent.push({ registration, identity, messageText }); return result; }, broadcast: value => reads.push(value) });
  await manager.initialize();
  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 }; },
    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 }] }; },
  };
}
test.beforeEach(async () => {
  for (const table of ['AgentPushDelivery','AgentPushConnection','AgentPushPresence','AgentPushInstallation','AgentPushCoordinator','AgentReadState']) await prisma.$executeRawUnsafe(`DELETE FROM "${table}"`);
});
test.after(async () => { await prisma.$disconnect(); rmSync(directory, { recursive: true, force: true }); });

test('first source bootstrap never alerts history; later durable candidates debounce, survive restart and deduplicate', async () => {
  const f = await fixture(); assert.equal((await f.manager.config()).sourceId, f.source); assert.equal(f.sent.length, 0);
  f.candidate(); await f.manager.tick(); assert.equal(f.sent.length, 0);
  f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 1);
  await f.manager.tick(); assert.equal(f.sent.length, 1);
  assert.deepEqual(f.sent[0].identity, { version: 1, sourceId: f.source, kind: 'message', messageId: 'outbound:11', cursor: '11' });
  assert.equal(f.sent[0].messageText, 'The Agent finished the requested work.');
  assert.equal((await prisma.agentPushDelivery.findFirst())?.status, 'accepted');
  const restarted = new AgentPushCoordinator({ ...f.manager.options, warmupMs: 0 }); await restarted.tick(); assert.equal(f.sent.length, 1);
});

test('preview resolves only the coalesced eligible root candidate after attention clears and survives a pending restart', async () => {
  const f = await fixture(), tab = randomUUID(), lookups: unknown[] = [];
  f.manager.options.messageText = async candidate => { lookups.push(candidate); return `Public reply ${candidate.messageId}`; };
  f.candidate('11', 'unknown'); await f.manager.tick(); f.advance(2_000); await f.manager.tick();
  assert.equal(lookups.length, 0);
  await f.manager.presence(tab, { sequence: 1, visibility: 'visible', focused: true, interacted: false });
  f.candidate('12', 'worker'); await f.manager.tick(); f.candidate('13', 'timer'); await f.manager.tick(); f.advance(2_000); await f.manager.tick();
  assert.equal(lookups.length, 0, 'focused clients do not cause repeated message fetches');
  await f.manager.retirePresence(tab); f.advance(6_000);
  const restarted = new AgentPushCoordinator({ ...f.manager.options, warmupMs: 0 }); await restarted.tick();
  assert.equal(f.sent.length, 1); assert.equal(f.sent[0].messageText, 'Public reply outbound:13');
  assert.equal((lookups[0] as any).conversationId, '1'); assert.equal((lookups[0] as any).messageId, 'outbound:13');
  assert.ok(!JSON.stringify(await prisma.agentPushDelivery.findMany()).includes('Public reply'), 'receipts contain no preview text');
});

for (const change of ['read', 'focus', 'source', 'revoke']) test(`preview lookup cannot bypass ${change} changes before the send intent`, async () => {
  const f = await fixture(); f.candidate(); await f.manager.tick(); f.advance(2_000);
  f.manager.options.messageText = async () => {
    if (change === 'read') f.context({ nativeReadCursor: '11' });
    if (change === 'source') f.context({ sourceId: randomUUID() });
    if (change === 'focus') await f.manager.presence(randomUUID(), { sequence: 1, visibility: 'visible', focused: true, interacted: false });
    if (change === 'revoke') await f.manager.revoke(f.installationId);
    return 'Public text that must not be sent now';
  };
  await f.manager.tick(); assert.equal(f.sent.length, 0);
  assert.equal((await prisma.agentPushDelivery.findFirstOrThrow()).attempts, 0);
});

test('unavailable public text suppresses without claiming a provider send; the next valid lookup sends actual text', async () => {
  const f = await fixture(); f.candidate(); await f.manager.tick(); f.advance(2_000);
  f.manager.options.messageText = async () => { throw new Error('private upstream detail'); };
  await f.manager.tick(); assert.equal(f.sent.length, 0);
  const pending = await prisma.agentPushDelivery.findFirstOrThrow();
  assert.equal(pending.status, 'pending'); assert.equal(pending.attempts, 0);
  assert.equal(pending.lastSuppressionReason, 'source_or_coordinator_unavailable');
  f.manager.options.messageText = async () => 'Recovered public reply';
  await f.manager.tick(); assert.equal(f.sent[0].messageText, 'Recovered public reply');
});

test('all tabs veto while focused; hidden heartbeats and stale sequences cannot extend short transition grace', async () => {
  const f = await fixture(), tab1 = randomUUID(), tab2 = randomUUID();
  await f.manager.presence(tab1, { sequence: 1, visibility: 'hidden', focused: false, interacted: false });
  await f.manager.presence(tab2, { sequence: 1, visibility: 'visible', focused: true, interacted: false });
  f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0, 'focused but idle still vetoes');
  let receipt = await prisma.agentPushDelivery.findFirstOrThrow();
  assert.equal(receipt.lastSuppressionReason, 'foreground_focused'); assert.equal(receipt.lastSuppressionClientKind, 'web');
  await f.manager.presence(tab2, { sequence: 2, visibility: 'hidden', focused: false, interacted: true });
  f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0, 'recent input vetoes blur');
  await f.manager.presence(tab2, { sequence: 1, visibility: 'visible', focused: true, interacted: false });
  assert.equal((await prisma.agentPushPresence.findUniqueOrThrow({ where: { id: tab2 } })).sequence, 2);
  await f.manager.presence(tab2, { sequence: 3, visibility: 'hidden', focused: false, interacted: true });
  f.advance(4_000); await f.manager.tick(); assert.equal(f.sent.length, 1, 'hidden receipts do not renew attention');
  receipt = await prisma.agentPushDelivery.findFirstOrThrow();
  assert.equal(receipt.lastSuppressionReason, 'inactive_grace'); assert.ok(receipt.lastSuppressedAt);
});

test('blur and visible recent interaction have short grace; fresh focus renews a live lease', 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(2_000);
  await f.manager.presence(tab, { sequence: 2, visibility: 'visible', focused: false, interacted: true });
  await f.manager.tick(); assert.equal(f.sent.length, 0);
  assert.equal((await prisma.agentPushDelivery.findFirstOrThrow()).lastSuppressionReason, 'recent_interaction');
  f.advance(3_000); await f.manager.presence(tab, { sequence: 3, visibility: 'visible', focused: false, interacted: false });
  f.advance(3_000); await f.manager.tick(); assert.equal(f.sent.length, 1);
  await f.manager.presence(tab, { sequence: 4, visibility: 'visible', focused: true, interacted: false });
  f.candidate('12'); await f.manager.tick(); f.advance(45_000);
  await f.manager.presence(tab, { sequence: 5, visibility: 'visible', focused: true, interacted: false });
  f.advance(20_000); await f.manager.tick(); assert.equal(f.sent.length, 1, 'fresh foreground heartbeat extends attention');
  f.advance(41_000); await f.manager.tick(); assert.equal(f.sent.length, 2, 'stale focus expires');
});

test('anonymous introduction grace survives heartbeat, new UUID reconnects and coordinator restart without renewing', async () => {
  const f = await fixture(), connection = randomUUID();
  await f.manager.connect(connection); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick();
  assert.equal(f.sent.length, 0);
  const first = await prisma.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } });
  assert.equal((await prisma.agentPushDelivery.findFirstOrThrow()).lastSuppressionReason, 'unknown_client_grace');
  await f.manager.heartbeat(connection); await f.manager.disconnect(connection);
  await f.manager.connect(randomUUID(), 'web', randomUUID()); f.advance(4_000); await f.manager.tick(); assert.equal(f.sent.length, 1);
  f.candidate('12'); await f.manager.tick(); f.advance(2_000);
  const restarted = new AgentPushCoordinator({ ...f.manager.options, warmupMs: 0 });
  await restarted.connect(randomUUID()); await restarted.tick(); assert.equal(f.sent.length, 2);
  assert.deepEqual((await prisma.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } })).unknownSince, first.unknownSince);
  // Spending the anonymous budget never disables an independent authenticated browser lease.
  await f.manager.presence(randomUUID(), { sequence: 1, visibility: 'visible', focused: true, interacted: false });
  f.candidate('13'); await f.manager.tick(); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 2);
});

test('expired observed web lease is inactive with a live socket; transport heartbeat preserves attribution and never recreates it', async () => {
  const f = await fixture(), tab = randomUUID(), connection = randomUUID(), native = randomUUID();
  await f.manager.presence(tab, { sequence: 1, visibility: 'visible', focused: true, interacted: false });
  await f.manager.connect(connection, 'web', tab); await f.manager.connect(native, 'native');
  f.candidate(); await f.manager.tick(); f.advance(61_000);
  await f.manager.heartbeat(connection); await f.manager.heartbeat(native); await f.manager.tick(); assert.equal(f.sent.length, 1);
  assert.equal((await prisma.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } })).unknownSince, null);
  const row = await prisma.agentPushConnection.findUniqueOrThrow({ where: { id: connection } });
  assert.equal(row.kind, 'web'); assert.equal(row.sessionId, tab);
  assert.equal((await prisma.agentPushConnection.findUniqueOrThrow({ where: { id: native } })).kind, 'native');
  await f.manager.disconnect(native); await f.manager.heartbeat(native);
  assert.equal(await prisma.agentPushConnection.findUnique({ where: { id: native } }), null);
  await f.manager.presence(tab, { sequence: 2, visibility: 'hidden', focused: false, interacted: false });
  assert.equal((await prisma.agentPushPresence.findUniqueOrThrow({ where: { id: tab } })).inactiveUntil, null, 'stale focus cannot start fresh grace');
});

test('presence classification cannot borrow a session from a different credential generation', async () => {
  const f = await fixture(), tab = randomUUID();
  await f.manager.presence(tab, { sequence: 1, visibility: 'hidden', focused: false, interacted: false });
  await prisma.agentPushPresence.update({ where: { id: tab }, data: { ownerEpoch: 'another-generation' } });
  await f.manager.connect(randomUUID(), 'web', tab); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick();
  assert.equal(f.sent.length, 0); assert.equal((await prisma.agentPushDelivery.findFirstOrThrow()).lastSuppressionReason, 'unknown_client_grace');
});

test('read at final context refresh cancels without a send and preserves the prior suppression evidence', async () => {
  const f = await fixture(); await f.manager.connect(randomUUID()); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick();
  const prior = await prisma.agentPushDelivery.findFirstOrThrow(), original = f.manager.options.context; let calls = 0;
  f.manager.options.context = async (...args) => { if (++calls === 2) await f.manager.advanceRead('11', f.source); return original(...args); };
  f.advance(4_000); await f.manager.tick();
  const receipt = await prisma.agentPushDelivery.findFirstOrThrow();
  assert.equal(f.sent.length, 0); assert.equal(receipt.attempts, 0); assert.equal(receipt.outcome, 'read');
  assert.equal(receipt.lastSuppressionReason, prior.lastSuppressionReason); assert.deepEqual(receipt.lastSuppressedAt, prior.lastSuppressedAt);
});

test('instrumented hidden web socket is inactive; missing lease and native-runtime unknown veto', async () => {
  const f = await fixture(), tab = randomUUID(), connection = randomUUID();
  await f.manager.connect(connection, 'web', tab); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
  await f.manager.presence(tab, { sequence: 0, visibility: 'hidden', focused: false, interacted: false });
  f.context({ webPresence: { active: false, unknown: true, observedAt: '' } }); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
  f.context({ webPresence: { active: false, unknown: false, observedAt: '' } }); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 1);
});

test('prompt grace resets on latest accepted prompt, root bursts coalesce, unknown provenance never sends', async () => {
  const f = await fixture(); f.context({ latestUserMessageAt: new Date(f.now).toISOString() });
  f.candidate(); await f.manager.tick(); f.advance(3_000); f.candidate('12', 'timer'); await f.manager.tick();
  f.context({ latestUserMessageAt: new Date(f.now).toISOString() }); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
  f.advance(3_000); await f.manager.tick(); assert.equal(f.sent.length, 1); assert.equal(f.sent[0].identity.cursor, '12');
  f.candidate('13', 'unknown'); await f.manager.tick(); f.advance(10_000); await f.manager.tick(); assert.equal(f.sent.length, 1);
});

test('canonical read and native read cancel pending sends and broadcast source-bound state', async () => {
  const f = await fixture(); f.candidate(); await f.manager.tick(); await f.manager.advanceRead('11', f.source); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
  f.candidate('12'); await f.manager.tick(); f.context({ nativeReadCursor: '12' }); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
  assert.equal((await f.manager.readState()).cursor, '12'); assert.equal(f.reads.at(-1).sourceId, f.source);
  assert.equal((await f.manager.advanceRead('1')).cursor, '12');
  await assert.rejects(f.manager.advanceRead('13', randomUUID()), { statusCode: 409 });
});

test('read clears target APNs only, coalesce and rate-limit; Web Push is never invisible', async () => {
  const f = await fixture(); await f.manager.register(randomUUID(), subscription()); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 2);
  await f.manager.advanceRead('11', f.source); await f.manager.tick();
  const clears = f.sent.filter(v => v.identity.kind === 'read_state'); assert.equal(clears.length, 1); assert.equal(clears[0].registration.transport, 'apns');
  await f.manager.advanceRead('12', f.source); await f.manager.tick(); assert.equal(f.sent.length, 3);
  f.advance(60_000); await f.manager.tick(); assert.equal(f.sent.filter(v => v.identity.kind === 'read_state').length, 2);
});

test('unavailable source, uncertainty, startup warm-up and source replacement suppress safely', async () => {
  const f = await fixture({ warmupMs: 60_000 }); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
  f.advance(60_000); f.failure(true); await f.manager.tick(); assert.equal(f.sent.length, 0);
  f.failure(false); f.context({ uncertain: true }); await f.manager.tick(); assert.equal(f.sent.length, 0);
  f.context({ uncertain: false, sourceId: randomUUID(), latestUserMessageAt: null }); await f.manager.tick(); assert.equal(f.sent.length, 0);
  assert.equal((await f.manager.config()).quarantined, true); assert.equal((await f.manager.config()).sourceId, f.source);
  assert.equal((await f.manager.readState()).sourceId, f.source);
});

test('known transient rejections retry boundedly; interrupted intent and ambiguous I/O never replay', async () => {
  const f = await fixture(); f.result({ outcome: 'retry', retryAfterMs: 1_000 }); f.candidate(); await f.manager.tick(); f.advance(2_000);
  for (let i = 0; i < 6; i++) { await f.manager.tick(); f.advance(2_000); }
  assert.equal(f.sent.length, 4); assert.equal((await prisma.agentPushDelivery.findFirst())?.status, 'failed');
  f.candidate('12'); f.result({ outcome: 'ambiguous' }); await f.manager.tick(); f.advance(2_000); await f.manager.tick();
  f.advance(60_000); await f.manager.tick(); assert.equal(f.sent.length, 5);
  f.candidate('13'); await f.manager.tick(); await prisma.agentPushDelivery.updateMany({ where: { status: 'pending' }, data: { status: 'intent', updatedAt: new Date(f.now - 40_000) } });
  await f.manager.tick(); assert.equal(f.sent.length, 5); assert.equal(await prisma.agentPushDelivery.count({ where: { status: 'ambiguous' } }), 2);
});

test('revoke and rotation prevent pending sends; permanent rejection cannot invalidate a newer endpoint version', async () => {
  const f = await fixture(); const same = await f.manager.register(f.installationId, apns); assert.equal(same.version, 1);
  f.candidate(); await f.manager.tick(); await f.manager.revoke(f.installationId); f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
  const newId = randomUUID(); await f.manager.register(newId, apns); f.candidate('12'); await f.manager.tick(); f.advance(2_000);
  f.manager.options.send = async () => { await f.manager.register(newId, { ...apns, token: 'b'.repeat(64) }); return { outcome: 'invalid' }; };
  await f.manager.tick(); const latest = await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: newId } }); assert.equal(latest.enabled, true); assert.equal(latest.version, 2);
});

test('credential generation change disables opt-ins; two coordinators do not send one delivery twice', async () => {
  const f = await fixture(); f.candidate(); await f.manager.tick(); f.advance(2_000);
  const second = new AgentPushCoordinator({ ...f.manager.options, warmupMs: 0 }); await second.initialize();
  await Promise.all([f.manager.tick(), second.tick()]); assert.equal(f.sent.length, 1);
  const rotated = new AgentPushCoordinator({ ...f.manager.options, config: { ...config, adminToken: 'rotated-fixture-token' }, warmupMs: 0 }); await rotated.initialize();
  assert.equal((await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: f.installationId } })).enabled, false);
});

test('authenticated routes reject open mode, wrong owner/body, malformed endpoints and secrets are never reflected', async () => {
  const app = Fastify({ logger: false }); await app.register(sensible);
  await registerPushRoutes(app, prisma, { baseUrl: 'http://unused.invalid' }, () => {}, { config, context: async () => { throw new Error(); }, send: async () => { throw new Error('must never send'); } });
  const headers = { authorization: `Bearer ${process.env.ADMIN_TOKEN}` }, url = `/v1/agent/push/installations/${randomUUID()}`;
  assert.equal((await app.inject({ url: '/v1/agent/push/config' })).statusCode, 401);
  assert.equal((await app.inject({ url: '/v1/agent/push/config', headers: { authorization: 'Bearer wrong' } })).statusCode, 403);
  const valid = await app.inject({ method: 'PUT', url, headers, payload: apns }); assert.equal(valid.statusCode, 200); assert.ok(!valid.body.includes(apns.token));
  assert.equal((await app.inject({ method: 'PUT', url, headers, payload: { ...apns, ownerId: 'someone-else' } })).statusCode, 400);
  const invalid = subscription(); if (invalid.transport === 'web') invalid.subscription.endpoint = 'https://127.0.0.1/private-secret';
  const response = await app.inject({ method: 'PUT', url, headers, payload: invalid }); assert.equal(response.statusCode, 400); assert.ok(!response.body.includes('private-secret'));
  const saved = env.ADMIN_TOKEN; env.ADMIN_TOKEN = undefined;
  try { assert.equal((await app.inject({ url: '/v1/agent/push/config', headers })).statusCode, 503); } finally { env.ADMIN_TOKEN = saved; }
  assert.equal((await app.inject({ method: 'DELETE', url, headers })).statusCode, 204);
  await app.close();
});

test('provider endpoint validation rejects SSRF targets and private/reserved DNS addresses', () => {
  for (const value of ['http://fcm.googleapis.com/x','https://user:pass@fcm.googleapis.com/x','https://fcm.googleapis.com.evil.example/x', 'https://fcm.googleapis.com:8443/x', 'https://[::1]/x','https://web.push.apple.com/x#fragment']) assert.throws(() => validatePushEndpoint(value));
  for (const address of ['127.0.0.1','10.0.0.1','169.254.1.1','192.168.1.1','172.16.0.1','::1','::ffff:127.0.0.1','fd00::1','fe80::1','2001:db8::1']) assert.equal(isPublicAddress(address), false, address);
  assert.equal(isPublicAddress('8.8.8.8'), true); assert.equal(isPublicAddress('2606:4700::1111'), true);
});

test('real APNs ES256 request construction uses message text, explicit environment and background clear headers through injected HTTP2 boundary', async () => {
  const key = generateKeyPairSync('ec', { namedCurve: 'prime256v1' });
  const requests: any[] = []; const now = Date.now(), sourceId = randomUUID();
  const sender = createPushSender({ ...config, apns: { ...config.apns!, key: key.privateKey.export({ format: 'pem', type: 'pkcs8' }).toString() } },
    { apns: async request => { requests.push(request); return { status: 200 }; } });
  const identity = { version: 1 as const, sourceId, kind: 'message' as const, messageId: 'outbound:12', cursor: '12' };
  assert.equal((await sender(apns, identity, now + 300_000, now, 'Your **report** is ready.')).outcome, 'accepted');
  assert.equal(requests[0].endpoint, `https://api.sandbox.push.apple.com/3/device/${apns.token}`);
  assert.equal(requests[0].headers['apns-push-type'], 'alert'); assert.equal(requests[0].headers['apns-topic'], apns.topic);
  const [header, claims, signature] = requests[0].headers.authorization.slice(7).split('.');
  assert.equal(JSON.parse(Buffer.from(header, 'base64url').toString()).kid, 'ABCDEFGHIJ');
  assert.equal(JSON.parse(Buffer.from(claims, 'base64url').toString()).iss, 'DQQH5H6GBD');
  assert.equal(verify('sha256', Buffer.from(`${header}.${claims}`), { key: key.publicKey, dsaEncoding: 'ieee-p1363' }, Buffer.from(signature, 'base64url')), true);
  assert.deepEqual(JSON.parse(requests[0].body).aps.alert, { title: 'Sybil Agent', body: 'Your report is ready.' });
  await sender(apns, { version: 1, sourceId, kind: 'read_state', readCursor: '12' }, now + 60_000, now);
  assert.equal(requests[1].headers['apns-push-type'], 'background'); assert.equal(requests[1].headers['apns-priority'], '5');
  assert.deepEqual(JSON.parse(requests[1].body).aps, { 'content-available': 1 });
});

test('standard encrypted aes128gcm + VAPID Web Push uses library request construction and never sends read clears', async () => {
  const vapid = webpush.generateVAPIDKeys(), registrations = subscription(), requests: any[] = [];
  const sender = createPushSender({ ...config, web: { ...vapid, subject: 'mailto:tests@example.com' } }, { web: async request => { requests.push(request); return { status: 201 }; } });
  const now = Date.now(), sourceId = randomUUID();
  assert.equal((await sender(registrations, { version: 1, sourceId, kind: 'message', messageId: 'outbound:1', cursor: '1' }, now + 300_000, now)).outcome, 'accepted');
  assert.equal(requests.length, 1); assert.equal(requests[0].headers['Content-Encoding'], 'aes128gcm'); assert.equal(requests[0].headers.TTL, 60);
  assert.match(requests[0].headers.Authorization, /^vapid t=/); assert.ok(Buffer.isBuffer(requests[0].body)); assert.ok(!requests[0].body.includes('You have a new message.'));
  assert.equal((await sender(registrations, { version: 1, sourceId, kind: 'read_state', readCursor: '1' }, now + 60_000, now)).outcome, 'failed'); assert.equal(requests.length, 1);
  const gone = createPushSender({ ...config, web: { ...vapid, subject: 'mailto:tests@example.com' } }, { web: async () => ({ status: 410 }) });
  assert.equal((await gone(registrations, { version: 1, sourceId, kind: 'message', messageId: 'outbound:2', cursor: '2' }, now + 60_000, now)).outcome, 'invalid');
});


test('production-scoped APNs configuration refuses sandbox registration and cannot send it', async () => {
  const production = { ...config, apns: { ...config.apns!, environment: 'production' as const } };
  const manager = new AgentPushCoordinator({ db: prisma, config: production, context: async () => { throw new Error(); }, messageText: async () => { throw new Error(); }, send: async () => { throw new Error('must not send'); } });
  await assert.rejects(manager.register(randomUUID(), apns), { statusCode: 400 });
  let requests = 0;
  const sender = createPushSender(production, { apns: async () => { requests++; throw new Error('must not send'); } });
  const now = Date.now();
  assert.equal((await sender(apns, { version: 1, sourceId: randomUUID(), kind: 'message', messageId: 'outbound:1', cursor: '1' }, now + 60_000, now)).outcome, 'failed');
  assert.equal(requests, 0);
});

test('raw mocked HTTP2 handles request throws, goaway and close; raw HTTPS has absolute timeout even without socket inactivity', async () => {
  const input = { endpoint: 'https://api.push.apple.com/3/device/fixture', headers: {}, body: Buffer.from('{}') };
  for (const event of ['connect', 'goaway', 'close', 'never']) {
    const session = new EventEmitter() as any; let destroyed = 0;
    session.destroy = () => { destroyed++; }; session.request = () => { throw new Error('fixture request error'); };
    const transport = createAPNsRequest((() => { if (event !== 'never') queueMicrotask(() => session.emit(event)); return session; }) as any, 5);
    await assert.rejects(transport(input), /APNs transport unavailable/); assert.equal(destroyed, 1);
  }
  const req = new EventEmitter() as any; let destroyed = false; req.destroy = () => { destroyed = true; }; req.end = () => {};
  const web = createWebRequest((() => req) as any, 5);
  await assert.rejects(web({ ...input, endpoint: 'https://web.push.apple.com/fixture' }), /Web Push transport unavailable/); assert.equal(destroyed, true);
  const throwing = createWebRequest((() => { throw new Error('fixture constructor error'); }) as any, 5);
  await assert.rejects(throwing({ ...input, endpoint: 'https://web.push.apple.com/fixture' }), /Web Push transport unavailable/);
});


test('DELETE tombstones unknown and known UUIDs across restart; late PUT and rebound endpoints cannot re-enable retired IDs', async () => {
  const f = await fixture(), unknown = randomUUID();
  await f.manager.revoke(unknown);
  await assert.rejects(f.manager.register(unknown, apns), { statusCode: 409 });
  await f.manager.revoke(f.installationId); await f.manager.revoke(f.installationId);
  const restarted = new AgentPushCoordinator({ ...f.manager.options, warmupMs: 0 }); await restarted.initialize();
  await assert.rejects(restarted.register(f.installationId, apns), { statusCode: 409 });
  const first = randomUUID(), second = randomUUID();
  await restarted.register(first, apns); await restarted.register(second, apns);
  await assert.rejects(restarted.register(first, apns), { statusCode: 409 });
  assert.equal((await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: first } })).payload, '');
  assert.equal((await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: second } })).enabled, true);
});


test('concurrent backend prompt admission starts durable grace even before runtime snapshot catches up', async () => {
  const f = await fixture(); f.candidate(); await f.manager.tick(); f.advance(2_000);
  await f.manager.notePrompt(); await f.manager.tick(); assert.equal(f.sent.length, 0);
  f.advance(4_000); await f.manager.tick(); assert.equal(f.sent.length, 0);
  f.advance(2_000); await f.manager.tick(); assert.equal(f.sent.length, 1);
});


async function httpContextFixture(f: Awaited<ReturnType<typeof fixture>>, snapshot: unknown, statuses = { incremental: 409, bootstrap: 200 }) {
  const requests: (string | null)[] = [], sent: unknown[] = [];
  const source = createServer((req, res) => {
    assert.equal(req.headers.authorization, 'Bearer fixture-runtime-secret');
    assert.equal(req.headers['x-sybil-client'], 'backend-proxy-v1');
    const url = new URL(req.url!, 'http://fixture.invalid');
    assert.equal(url.pathname, '/api/notification-context'); assert.equal(url.searchParams.get('limit'), '200');
    const after = url.searchParams.get('after'); requests.push(after);
    res.statusCode = after === null ? statuses.bootstrap : statuses.incremental;
    res.setHeader('Content-Type', 'application/json');
    res.end(JSON.stringify(after === null && statuses.bootstrap === 200 ? snapshot : { error: 'cursor unavailable' }));
  });
  source.listen(0, '127.0.0.1'); await once(source, 'listening');
  const app = Fastify({ logger: false }); await app.register(sensible);
  const manager = await registerPushRoutes(app, prisma, { baseUrl: `http://127.0.0.1:${(source.address() as { port: number }).port}`, apiKey: 'fixture-runtime-secret' }, () => {}, {
    config, warmupMs: 0, now: () => f.now,
    send: async (_registration, identity) => { sent.push(identity); return { outcome: 'accepted' }; },
  });
  return { manager, requests, sent, close: async () => { await manager.stop(); await app.close(); source.closeAllConnections(); await new Promise<void>(resolve => source.close(() => resolve())); } };
}
function highWaterSnapshot(f: Awaited<ReturnType<typeof fixture>>, overrides: Partial<Context> = {}): Context {
  return { version: 1, sourceId: randomUUID(), conversationId: '1', cursor: '3', hasMore: false, latestUserMessageAt: null,
    nativeReadCursor: '0', uncertain: false, webPresence: { active: false, unknown: false, observedAt: new Date(f.now).toISOString() }, candidates: [], ...overrides };
}
for (const identityChanged of [true, false]) test(`actual HTTP409 probes once and quarantines ${identityChanged ? 'replacement source' : 'same-source cursor regression'} without rebasing reads/checkpoint`, async () => {
  const f = await fixture(); await f.manager.advanceRead('4', f.source); f.candidate(); await f.manager.tick(); f.advance(2_000);
  const oldRead = await f.manager.readState();
  const http = await httpContextFixture(f, highWaterSnapshot(f, identityChanged ? {} : { sourceId: f.source }));
  try {
    await http.manager.tick(); assert.deepEqual(http.requests, ['11', null]); assert.equal(http.sent.length, 0);
    const state = await prisma.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } });
    assert.equal(state.quarantined, true); assert.equal(state.sourceId, f.source); assert.equal(state.cursor, '11');
    assert.deepEqual(await http.manager.readState(), oldRead);
    const delivery = await prisma.agentPushDelivery.findFirstOrThrow(); assert.equal(delivery.status, 'cancelled'); assert.equal(delivery.outcome, 'source_changed');
    assert.equal(await prisma.agentPushDelivery.count(), 1, 'probe never enumerates/replays historical candidates');
  } finally { await http.close(); }
});
for (const failure of ['rejected', 'conflict', 'malformed', 'candidates', 'paginated', 'same-source-equal', 'same-source-higher', 'non-conflict']) test(`HTTP cursor recovery stays fail closed for ${failure}`, async () => {
  const f = await fixture(); f.candidate(); await f.manager.tick(); f.advance(2_000); const oldRead = await f.manager.readState();
  const snapshot = failure === 'malformed' ? { invalid: true } : highWaterSnapshot(f, {
    ...(failure === 'candidates' ? { candidates: [{ messageId: 'outbound:3', cursor: '3', createdAt: new Date(f.now).toISOString(), trigger: 'timer' as const }] } : {}),
    ...(failure === 'paginated' ? { hasMore: true } : {}),
    ...(failure.startsWith('same-source') ? { sourceId: f.source, cursor: failure === 'same-source-higher' ? '12' : '11', nativeReadCursor: '11' } : {}),
  });
  const http = await httpContextFixture(f, snapshot, { incremental: failure === 'non-conflict' ? 503 : 409,
    bootstrap: failure === 'rejected' ? 503 : failure === 'conflict' ? 409 : 200 });
  try {
    await http.manager.tick(); assert.deepEqual(http.requests, failure === 'non-conflict' ? ['11'] : ['11', null]); assert.equal(http.sent.length, 0);
    const state = await prisma.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } });
    assert.equal(state.uncertain, true); assert.equal(state.quarantined, false); assert.equal(state.sourceId, f.source); assert.equal(state.cursor, '11');
    assert.deepEqual(await http.manager.readState(), oldRead); assert.equal((await prisma.agentPushDelivery.findFirstOrThrow()).status, 'pending');
  } finally { await http.close(); }
});

test('endpoint rebind retires a provider-disabled UUID, rejects its delayed PUT, and leaves no live endpoint after new owner revoke', async () => {
  const f = await fixture(); f.result({ outcome: 'invalid' }); f.candidate(); await f.manager.tick(); f.advance(2_000); await f.manager.tick();
  const invalid = await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: f.installationId } });
  assert.equal(invalid.enabled, false); assert.equal(invalid.revokedAt, null); assert.notEqual(invalid.fingerprint, '');
  const newer = randomUUID(); await f.manager.register(newer, apns);
  assert.notEqual((await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: f.installationId } })).revokedAt, null);
  await assert.rejects(f.manager.register(f.installationId, apns), { statusCode: 409 });
  await f.manager.revoke(newer); assert.equal(await prisma.agentPushInstallation.count({ where: { enabled: true } }), 0);
});

test('endpoint rebind also retires disabled registrations from a prior credential generation', async () => {
  const f = await fixture();
  const current = new AgentPushCoordinator({ ...f.manager.options, config: { ...config, adminToken: 'new-fixture-bearer-generation' }, warmupMs: 0 });
  await current.initialize(); assert.equal((await prisma.agentPushInstallation.findUniqueOrThrow({ where: { id: f.installationId } })).enabled, false);
  const newer = randomUUID(); await current.register(newer, apns);
  await assert.rejects(f.manager.register(f.installationId, apns), { statusCode: 409 });
  await assert.rejects(current.register(f.installationId, apns), { statusCode: 409 });
  await current.revoke(newer); assert.equal(await prisma.agentPushInstallation.count({ where: { enabled: true } }), 0);
});