From 35a6c537adc43103ecfe3622d23adf68da4681d1 Mon Sep 17 00:00:00 2001 From: James Magahern Date: Thu, 15 Oct 2026 01:07:00 -0700 Subject: [PATCH] Reconcile agent notifications from events and pending deadlines --- docs/api/rest.md | 2 +- docs/api/streaming-chat.md | 3 + server/src/agent-push-contract.ts | 2 +- server/src/agent-push-events.ts | 35 +++++++ server/src/agent-push.ts | 127 +++++++++++++++++++------ server/tests/agent-push-events.test.ts | 24 +++++ server/tests/agent-push.test.ts | 39 +++++++- server/tests/agent.test.ts | 1 + 8 files changed, 200 insertions(+), 33 deletions(-) create mode 100644 server/src/agent-push-events.ts create mode 100644 server/tests/agent-push-events.test.ts diff --git a/docs/api/rest.md b/docs/api/rest.md index 3c71742..aebcf4f 100644 --- a/docs/api/rest.md +++ b/docs/api/rest.md @@ -847,7 +847,7 @@ The web service worker accepts bounded nonempty preview strings, preserving the Shipping is disabled by default: `AGENT_PUSH_ENABLED=false`. Native transport requires `AGENT_PUSH_APNS_KEY_FILE` (APNs-enabled ES256 `.p8`, regular secret file 0400/0600), `AGENT_PUSH_APNS_KEY_ID`, `AGENT_PUSH_APNS_TEAM_ID`, optional `AGENT_PUSH_APNS_TOPIC` (default `net.buzzert.sybil2`) and `AGENT_PUSH_APNS_ENVIRONMENT` (`production` by default, or `sandbox`; exactly one credential environment per deployment). Web Push requires `AGENT_PUSH_VAPID_FILE`, a private regular JSON file 0400/0600 containing `publicKey`, `privateKey` (base64url P-256) and `subject` (mailto/HTTPS contact). Invalid/missing/unsafe files leave only that adapter unavailable; files and parse errors are never exposed. This implementation neither creates production keys nor changes Apple portal capabilities. Correct native team/topic is `DQQH5H6GBD` / `net.buzzert.sybil2`; parent portal work, push entitlements/provisioning and real device registration are separate prerequisites. An environment-scoped production key cannot send sandbox notifications: sandbox registrations are rejected when production is configured. Concurrent sandbox/production credentials require a future explicit per-environment credential map; do not reuse one scoped key across environments. -Backend independently polls token-authenticated runtime `GET /api/notification-context?after=&limit=200` and sends `X-Sybil-Client: backend-proxy-v1` on private proxy REST/WS traffic. The marker classifies the proxy for the native attention census and never replaces runtime authentication. The runtime source is version 1 with `{sourceId,conversationId,cursor,hasMore,latestUserMessageAt,nativeReadCursor,uncertain,webPresence:{active,unknown,observedAt},candidates:[{messageId,cursor,createdAt,trigger}]}`; trigger is `user|external_event|timer|worker|unknown`. No transcript text is included. `uncertain` and `webPresence.unknown` fail closed. Candidate pages preserve all candidates belonging to each consumed cursor (max 200 candidates/page), advancing across non-candidates; omitted `after` is high-water bootstrap with no old candidates. The source identity survives normal runtime restarts. If an incremental context request returns 409 because its cursor exceeds a replaced/restored runtime high-water mark, the backend makes exactly one authenticated no-after probe under the same 10-second deadline. Only a candidate-free, nonpaginated snapshot proving a different source or lower cursor is accepted, solely to quarantine and cancel old pending alerts. It never resets reads/checkpoints, rebaselines, or enumerates historical candidates. Failed/invalid probes and same-source nonregressions remain unavailable and suppress delivery. Runtime native presence/read is an acknowledgment input to the existing canonical backend read state, not a second independently authoritative read store. +Backend subscribes to the token-authenticated private runtime WebSocket `GET /api/notification-events`. Its initial and coalesced `{type:"notification_changed",version:1}` frames trigger cursor-based reconciliation through `GET /api/notification-context?after=&limit=200`; reconnects recover missed changes from the saved durable cursor. The socket uses bearer header authentication, rejects query strings, carries no content, and uses ping/pong without database access. There is no fixed two-second source poll. Local presence/read/prompt changes also wake reconciliation. One-shot timers cover debounce, grace and lease expiry, provider retry and abandoned send intents; idle maintenance runs hourly. Source failures retry with bounded backoff (1–30 seconds). The coordinator still fetches fresh context and checks every eligibility condition before each provider boundary. The backend sends `X-Sybil-Client: backend-proxy-v1` on private proxy REST/WS traffic and `backend-observer-v1` on the notification observer. The marker classifies the proxy for the native attention census and never replaces runtime authentication. The runtime source is version 1 with `{sourceId,conversationId,cursor,hasMore,latestUserMessageAt,nativeReadCursor,uncertain,nextChangeAt?,webPresence:{active,unknown,observedAt},candidates:[{messageId,cursor,createdAt,trigger}]}`; trigger is `user|external_event|timer|worker|unknown`. `nextChangeAt` is an optional nullable ISO deadline for the next runtime attention/uncertainty expiry; pending work schedules a refresh then and earlier events reschedule it. No transcript text is included. `uncertain` and `webPresence.unknown` fail closed. Candidate pages preserve all candidates belonging to each consumed cursor (max 200 candidates/page), advancing across non-candidates; omitted `after` is high-water bootstrap with no old candidates. The source identity survives normal runtime restarts. If an incremental context request returns 409 because its cursor exceeds a replaced/restored runtime high-water mark, the backend makes exactly one authenticated no-after probe under the same 10-second deadline. Only a candidate-free, nonpaginated snapshot proving a different source or lower cursor is accepted, solely to quarantine and cancel old pending alerts. It never resets reads/checkpoints, rebaselines, or enumerates historical candidates. Failed/invalid probes and same-source nonregressions remain unavailable and suppress delivery. Runtime native presence/read is an acknowledgment input to the existing canonical backend read state, not a second independently authoritative read store. For an otherwise dispatchable message only, the backend resolves its text through the existing authenticated root public endpoint: `GET /api/conversations/root/messages?after=&limit=1&includeTools=false`. The URL/path is fixed to the configured runtime, redirects are rejected, the response is limited to 512 KiB and the request to 10 seconds. The page must match the observed root conversation and exact entry cursor; exactly one message must match the candidate ID, assistant role, `metadata.kind:"message"`, `status:"complete"` and entry ID. Only that message's string `content` is used. The runtime's existing root-candidate/provenance rules remain authoritative; a `worker` trigger qualifies only a later public root reply, never the worker's own output. Source/read/presence and all delivery conditions are rechecked after lookup and before committing send intent. Preview text is not persisted in backend outbox/receipt rows. No runtime change or database migration is required. diff --git a/docs/api/streaming-chat.md b/docs/api/streaming-chat.md index b432c54..9fd68c1 100644 --- a/docs/api/streaming-chat.md +++ b/docs/api/streaming-chat.md @@ -428,3 +428,6 @@ Distinguish an unbound room from one already bound to a legacy absent/null sourc Older clients may ignore unknown server frame types. `read_state` and presence are not client subscribe messages and are never forwarded to runtime as generic frames. Push consent, revoke, presence, retry/expiry and notification payload contracts live in [rest.md](./rest.md#agent-push-notifications-v1). The runtime built-in browser has its own authenticated presence/read inputs and bounded anonymous grace per runtime source. The separate 60-second restart safety intervals and source/clock uncertainty checks remain in effect. No Web Push is used for invisible read synchronization. Enabled notifications include a bounded plain-text preview of the eligible completed root public assistant message by default, including on the lock screen when OS settings allow it. The backend resolves only that candidate's exact public `content`, never activity/reasoning/tool/worker fields or linked resources. APNs uses a native string `aps.alert.body`; Web Push uses `body`. This does not change message IDs, stream frames, read acknowledgment or clearing semantics. See the REST contract for Markdown flattening, payload budgets and web worker update ordering. + + +The backend notification coordinator has its own private, token-authenticated runtime WebSocket at `/api/notification-events`. It receives content-free `{type:"notification_changed",version:1}` invalidations, including an initial frame on every connection, and reconciles durable notification context from its saved cursor. This replaces the fixed two-second backend poll; it does not change public Agent message/read-state frames or require a client release. Native presence/read changes and local backend presence/read/prompt changes wake the coordinator. Pending notification deadlines use one-shot timers (including the runtime's additive nullable `nextChangeAt` field); idle maintenance runs hourly. Reconnection, source failure backoff, and fresh eligibility checks before delivery preserve cursor recovery and suppression behavior. WebSocket ping/pong performs no database operations. diff --git a/server/src/agent-push-contract.ts b/server/src/agent-push-contract.ts index e198a7c..97f02f7 100644 --- a/server/src/agent-push-contract.ts +++ b/server/src/agent-push-contract.ts @@ -22,7 +22,7 @@ export const InstallationBody = z.discriminatedUnion('transport', [ ]); export type Registration = z.infer; export type Subscription = z.infer; -export const NotificationContext = z.object({ version: z.literal(1), sourceId: z.string().uuid(), conversationId: z.string().min(1).max(128), +export const NotificationContext = z.object({ version: z.literal(1), nextChangeAt: z.string().datetime().nullable().optional(), sourceId: z.string().uuid(), conversationId: z.string().min(1).max(128), cursor: Cursor, hasMore: z.boolean(), latestUserMessageAt: z.string().datetime().nullable(), nativeReadCursor: Cursor, uncertain: z.boolean(), webPresence: z.object({ active: z.boolean(), unknown: z.boolean(), observedAt: z.string().datetime() }), candidates: z.array(z.object({ messageId: z.string().min(1).max(200), cursor: Cursor, diff --git a/server/src/agent-push-events.ts b/server/src/agent-push-events.ts new file mode 100644 index 0000000..f5cf6fe --- /dev/null +++ b/server/src/agent-push-events.ts @@ -0,0 +1,35 @@ +import WebSocket from 'ws'; + +/** Events are hints; every connection starts durable cursor reconciliation. */ +export function subscribeNotificationEvents(upstream: { baseUrl: string; apiKey?: string }, changed: () => void, + options: { retryMs?: number; heartbeatMs?: number } = {}) { + let stopped = false, attempts = 0, socket: WebSocket | undefined; + let retry: ReturnType | undefined, heartbeat: ReturnType | undefined; + const connect = () => { + if (stopped || !upstream.apiKey) return; + const url = new URL('/api/notification-events', upstream.baseUrl); + url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'; + socket = new WebSocket(url, { headers: { Authorization: `Bearer ${upstream.apiKey}`, 'X-Sybil-Client': 'backend-observer-v1' }, + handshakeTimeout: 10_000, maxPayload: 1024, followRedirects: false }); + const current = socket; + const alive = () => { clearTimeout(heartbeat); heartbeat = setTimeout(() => current.terminate(), options.heartbeatMs ?? 60_000); heartbeat.unref(); }; + current.on('open', alive); + current.on('ping', alive); + current.on('message', raw => { + try { + const event = JSON.parse(raw.toString()); + if (event.type !== 'notification_changed' || event.version !== 1) { current.close(1002); return; } + attempts = 0; alive(); changed(); + } catch { current.close(1002); } + }); + current.on('error', () => { /* Reconnect without logging headers, URLs, or response bodies. */ }); + current.on('close', () => { + clearTimeout(heartbeat); + if (stopped) return; + retry = setTimeout(connect, Math.min(30_000, (options.retryMs ?? 1_000) * 2 ** Math.min(attempts++, 5))); + retry.unref(); + }); + }; + connect(); + return () => { stopped = true; clearTimeout(retry); clearTimeout(heartbeat); socket?.terminate(); }; +} diff --git a/server/src/agent-push.ts b/server/src/agent-push.ts index 76ac635..620bf34 100644 --- a/server/src/agent-push.ts +++ b/server/src/agent-push.ts @@ -9,14 +9,15 @@ import { Cursor, InstallationBody, InstallationId, MAX_AGE_MS, MAX_ATTEMPTS, Not digest, fail, ownerEpoch, parseSafe, validateSubscription, type Context, type Identity, type Presence, type Registration } from './agent-push-contract.js'; import { createPushSender, type PushSend } from './agent-push-transports.js'; import { createMessageText, type MessageText } from './agent-push-preview.js'; +import { subscribeNotificationEvents } from './agent-push-events.js'; type Tx = Prisma.TransactionClient; export type ReadState = { cursor: string; updatedAt: string | null; sourceId: string | null }; type Options = { db: PrismaClient; config: PushConfig; context: (after?: string, sourceId?: string) => Promise; messageText: MessageText; send: PushSend; now?: () => number; - broadcast?: (state: ReadState) => void; warmupMs?: number }; + broadcast?: (state: ReadState) => void; warmupMs?: number; subscribe?: (changed: () => void) => (() => void) }; const date = (ms: number) => new Date(ms); const activeStatuses = ['pending', 'intent']; -type Suppression = { reason: string; clientKind: 'web' | 'unknown' | 'runtime_web' | 'runtime_unknown' | 'system' }; +type Suppression = { until?: number; reason: string; clientKind: 'web' | 'unknown' | 'runtime_web' | 'runtime_unknown' | 'system' }; /** Sole coordinator: persisted source checkpoint, installations, presence, claims and outcomes. */ export class AgentPushCoordinator { @@ -27,7 +28,15 @@ export class AgentPushCoordinator { private started = false; private untrackedConnections = new Map(); private running = false; - private timer?: ReturnType; + private timer?: ReturnType; + private timerAt = Infinity; + private enabled = false; + private dirty = false; + private reconsider = true; + private unsubscribe?: () => void; + private nextMaintenance = 0; + private nextRuntimeChange?: number; + private failures = 0; private stopped = false; private observedRoot?: { sourceId: string; conversationId: string }; constructor(readonly options: Options) { @@ -51,11 +60,37 @@ export class AgentPushCoordinator { this.started = true; } start() { - if (this.timer || !this.options.config.adminToken) return; - this.timer = setInterval(() => { void this.tick(); }, 2_000); this.timer.unref(); - void this.tick(); + if (this.enabled || !this.options.config.adminToken) return; + this.enabled = true; + this.unsubscribe = this.options.subscribe?.(() => this.wake()); + this.wake(); + } + /** Coalesce events without dropping one that arrives during reconciliation. */ + wake() { + this.reconsider = true; + if (!this.enabled || this.stopped) return; + this.dirty = true; + if (!this.running) this.schedule(this.now() + 100); + } + private schedule(at: number) { + if (!this.enabled || this.stopped || at >= this.timerAt) return; + clearTimeout(this.timer); + this.timerAt = at; + this.timer = setTimeout(() => { this.timer = undefined; this.timerAt = Infinity; void this.tick(); }, Math.max(0, at - this.now())); + this.timer.unref(); + } + private async schedulePending() { + const [pending, intent] = await Promise.all([ + this.options.db.agentPushDelivery.findFirst({ where: { status: 'pending' }, orderBy: { dueAt: 'asc' }, select: { dueAt: true } }), + this.options.db.agentPushDelivery.findFirst({ where: { status: 'intent' }, orderBy: { updatedAt: 'asc' }, select: { updatedAt: true } }), + ]); + this.schedule(Math.max(this.now() + 100, Math.min(this.nextMaintenance, + pending?.dueAt.getTime() ?? Infinity, intent ? intent.updatedAt.getTime() + 30_001 : Infinity))); + } + async stop() { + this.stopped = true; this.unsubscribe?.(); clearTimeout(this.timer); + while (this.running) await new Promise(r => setTimeout(r, 20)); } - async stop() { this.stopped = true; if (this.timer) clearInterval(this.timer); while (this.running) await new Promise(r => setTimeout(r, 20)); } async config() { await this.initialize(); const state = await this.options.db.agentPushCoordinator.findUnique({ where: { id: 'root' } }); @@ -73,7 +108,7 @@ export class AgentPushCoordinator { value = { ...value, token: value.token.toLowerCase() }; } else { if (!config.web) throw fail(503, 'Web Push is unavailable'); validateSubscription(value.subscription, this.now()); } const payload = JSON.stringify(value), fingerprint = digest(value.transport === 'apns' ? `${value.environment}:${value.topic}:${value.token}` : value.subscription.endpoint); - return this.options.db.$transaction(async tx => { + const result = await this.options.db.$transaction(async tx => { const old = await tx.agentPushInstallation.findUnique({ where: { id } }); if (old?.revokedAt) throw fail(409, 'Installation was retired; explicit opt-in requires a new installation ID'); if (old?.enabled && old.ownerEpoch === this.epoch && old.payload === payload) return { installationId: id, transport: old.transport, enabled: true, version: old.version }; @@ -89,6 +124,8 @@ export class AgentPushCoordinator { update: { ownerEpoch: this.epoch, transport: value.transport, payload, fingerprint, enabled: true, version: { increment: 1 } } }); return { installationId: id, transport: record.transport, enabled: record.enabled, version: record.version }; }); + this.wake(); + return result; } async revoke(id: string) { await this.options.db.$transaction(async tx => { @@ -98,10 +135,11 @@ export class AgentPushCoordinator { update: { enabled: false, revokedAt: date(this.now()), version: { increment: 1 }, payload: '', fingerprint: '' } }); await tx.agentPushDelivery.updateMany({ where: { installationId: id, status: { in: activeStatuses } }, data: { status: 'cancelled', outcome: 'revoked' } }); }); + this.wake(); } async presence(id: string, value: Presence) { const now = this.now(); - return this.options.db.$transaction(async tx => { + const result = await this.options.db.$transaction(async tx => { const old = await tx.agentPushPresence.findUnique({ where: { id } }); if (old?.ownerEpoch === this.epoch && value.sequence <= old.sequence) return { expiresAt: old.expiresAt.toISOString() }; if (!old && await tx.agentPushPresence.count({ where: { ownerEpoch: this.epoch, expiresAt: { gt: date(now) } } }) >= 512) throw fail(429, 'Too many presence sessions'); @@ -115,6 +153,8 @@ export class AgentPushCoordinator { const row = await tx.agentPushPresence.upsert({ where: { id }, create: { id, ...data }, update: data }); return { expiresAt: row.expiresAt.toISOString() }; }); + this.wake(); + return result; } async retirePresence(id: string) { await this.options.db.$transaction(async tx => { @@ -123,6 +163,7 @@ export class AgentPushCoordinator { await tx.agentPushPresence.update({ where: { id }, data: { visible: false, focused: false, inactiveUntil: old.visible && old.focused && old.expiresAt.getTime() > this.now() ? date(this.now() + PUSH_POLICY.inactiveGraceMs) : old.inactiveUntil } }); }); + this.wake(); } async connect(id: string, kind?: string, sessionId?: string) { const now = this.now(); @@ -140,17 +181,19 @@ export class AgentPushCoordinator { } }); this.untrackedConnections.delete(id); + this.wake(); } async heartbeat(id: string) { // Transport liveness never creates a client, changes its attribution or proves attention. await this.options.db.agentPushConnection.updateMany({ where: { id, ownerEpoch: this.epoch }, data: { expiresAt: date(this.now() + PUSH_POLICY.presenceLeaseMs) } }); } - async disconnect(id: string) { this.untrackedConnections.delete(id); await this.options.db.agentPushConnection.deleteMany({ where: { id, ownerEpoch: this.epoch } }); } + async disconnect(id: string) { this.untrackedConnections.delete(id); await this.options.db.agentPushConnection.deleteMany({ where: { id, ownerEpoch: this.epoch } }); this.wake(); } async notePrompt() { if (!this.options.config.adminToken) return; await this.initialize(); // Covers submissions arriving after the last runtime snapshot but before the send claim. await this.options.db.agentPushCoordinator.updateMany({ where: { id: 'root', ownerEpoch: this.epoch }, data: { localPromptAt: date(this.now()) } }); + this.wake(); } async readState(tx: Tx | PrismaClient = this.options.db): Promise { const row = await tx.agentReadState.findUnique({ where: { id: 'root' } }); @@ -193,7 +236,7 @@ export class AgentPushCoordinator { const changed = await this.mergeRead(tx, cursor, coordinator?.sourceId || null); return { state: changed || await this.readState(tx), changed: Boolean(changed) }; }); - if (result.changed) this.options.broadcast?.(result.state); + if (result.changed) { this.options.broadcast?.(result.state); this.wake(); } return result.state; } private async acquire(): Promise { @@ -263,22 +306,26 @@ export class AgentPushCoordinator { return changedRead; }); if (read) this.options.broadcast?.(read); + this.nextRuntimeChange = context.nextChangeAt ? Date.parse(context.nextChangeAt) : undefined; this.observedRoot = { sourceId: context.sourceId, conversationId: context.conversationId }; return context.hasMore; } private async browserSuppression(tx: Tx, now: number): Promise { const observations = await tx.agentPushPresence.findMany({ where: { ownerEpoch: this.epoch } }); const presences = observations.filter(v => v.expiresAt.getTime() > now); - if (presences.some(v => v.visible && v.focused)) return { reason: 'foreground_focused', clientKind: 'web' }; - if (presences.some(v => v.lastInteractionAt && v.lastInteractionAt.getTime() + PUSH_POLICY.interactionGraceMs > now)) return { reason: 'recent_interaction', clientKind: 'web' }; - if (presences.some(v => v.inactiveUntil && v.inactiveUntil.getTime() > now)) return { reason: 'inactive_grace', clientKind: 'web' }; + const focused = presences.filter(v => v.visible && v.focused); + if (focused.length) return { reason: 'foreground_focused', clientKind: 'web', until: Math.min(...focused.map(v => v.expiresAt.getTime())) }; + const interactions = presences.map(v => v.lastInteractionAt ? Math.min(v.expiresAt.getTime(), v.lastInteractionAt.getTime() + PUSH_POLICY.interactionGraceMs) : 0).filter(at => at > now); + if (interactions.length) return { reason: 'recent_interaction', clientKind: 'web', until: Math.min(...interactions) }; + const inactive = presences.map(v => Math.min(v.expiresAt.getTime(), v.inactiveUntil?.getTime() || 0)).filter(at => at > now); + if (inactive.length) return { reason: 'inactive_grace', clientKind: 'web', until: Math.min(...inactive) }; const connections = await tx.agentPushConnection.findMany({ where: { ownerEpoch: this.epoch, expiresAt: { gt: date(now) } } }); const unobserved = connections.filter(v => v.kind !== 'native' && (v.kind !== 'web' || !observations.some(p => p.id === v.sessionId))); if (!unobserved.length && !this.untrackedConnections.size) return; const firstUntracked = [...this.untrackedConnections.values()].sort((a, b) => a.at - b.at)[0]; await tx.agentPushCoordinator.updateMany({ where: { id: 'root', ownerEpoch: this.epoch, unknownSince: null }, data: { unknownSince: date(firstUntracked?.at ?? now) } }); const state = await tx.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } }); - if (state.unknownSince!.getTime() + PUSH_POLICY.unknownGraceMs > now) return { reason: 'unknown_client_grace', + if (state.unknownSince!.getTime() + PUSH_POLICY.unknownGraceMs > now) return { reason: 'unknown_client_grace', until: state.unknownSince!.getTime() + PUSH_POLICY.unknownGraceMs, clientKind: unobserved[0]?.kind === 'web' ? 'web' : firstUntracked?.kind || 'unknown' }; } private async claim(id: string, commit: boolean) { @@ -287,8 +334,8 @@ export class AgentPushCoordinator { const state = await tx.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } }); const row = await tx.agentPushDelivery.findUnique({ where: { id } }); if (!row || row.status !== 'pending' || row.dueAt.getTime() > now) return; - const suppress = async ({ reason, clientKind }: Suppression) => { - await tx.agentPushDelivery.update({ where: { id }, data: { dueAt: date(now + PUSH_POLICY.debounceMs), + const suppress = async ({ reason, clientKind, until }: Suppression) => { + await tx.agentPushDelivery.update({ where: { id }, data: { dueAt: date(Math.min(row.expiresAt.getTime(), Math.max(now + 100, until ?? now + 30_000))), lastSuppressionReason: reason, lastSuppressionClientKind: clientKind, lastSuppressedAt: date(now) } }); }; const installation = await tx.agentPushInstallation.findUnique({ where: { id: row.installationId } }); @@ -299,15 +346,15 @@ export class AgentPushCoordinator { } if (!this.options.config.enabled || state.lockOwner !== this.instanceId || !state.lockUntil || state.lockUntil.getTime() <= now || state.uncertain || !state.contextObservedAt || state.contextObservedAt.getTime() < now - 10_000 || now < this.warmUntil) { - await suppress({ reason: now < this.warmUntil ? 'startup_warmup' : 'source_or_coordinator_unavailable', clientKind: 'system' }); return; + await suppress({ reason: now < this.warmUntil ? 'startup_warmup' : 'source_or_coordinator_unavailable', clientKind: 'system', until: now < this.warmUntil ? this.warmUntil : this.nextRuntimeChange }); return; } if (row.kind === 'message') { - const attention: Suppression | undefined = state.nativeActive ? { reason: 'runtime_active', clientKind: 'runtime_web' } - : state.nativeUnknown ? { reason: 'runtime_unknown', clientKind: 'runtime_unknown' } : await this.browserSuppression(tx, now); + const attention: Suppression | undefined = state.nativeActive ? { reason: 'runtime_active', clientKind: 'runtime_web', until: this.nextRuntimeChange } + : state.nativeUnknown ? { reason: 'runtime_unknown', clientKind: 'runtime_unknown', until: this.nextRuntimeChange } : await this.browserSuppression(tx, now); if (attention) { await suppress(attention); return; } if ((state.latestUserMessageAt && state.latestUserMessageAt.getTime() + PUSH_POLICY.promptGraceMs > now) || (state.localPromptAt && state.localPromptAt.getTime() + PUSH_POLICY.promptGraceMs > now)) { - await suppress({ reason: 'prompt_grace', clientKind: 'system' }); return; + await suppress({ reason: 'prompt_grace', clientKind: 'system', until: Math.max(state.latestUserMessageAt?.getTime() || 0, state.localPromptAt?.getTime() || 0) + PUSH_POLICY.promptGraceMs }); return; } } if (row.attempts >= MAX_ATTEMPTS) { await tx.agentPushDelivery.update({ where: { id }, data: { status: 'failed', outcome: 'attempt_limit' } }); return; } @@ -345,32 +392,45 @@ export class AgentPushCoordinator { await this.options.db.$transaction(async tx => { const pending = result.outcome === 'retry' && claim.row.attempts + 1 < MAX_ATTEMPTS; await tx.agentPushDelivery.updateMany({ where: { id, status: 'intent' }, data: { status: pending ? 'pending' : result.outcome === 'retry' ? 'failed' : result.outcome, - outcome: result.outcome, ...(pending ? { dueAt: date(this.now() + (result.retryAfterMs ?? Math.min(60_000, 2_000 * 2 ** claim.row.attempts))) } : {}) } }); + outcome: result.outcome, ...(pending ? { lastSuppressionReason: null, dueAt: date(this.now() + (result.retryAfterMs ?? Math.min(60_000, 2_000 * 2 ** claim.row.attempts))) } : {}) } }); if (result.outcome === 'invalid') await tx.agentPushInstallation.updateMany({ where: { id: claim.row.installationId, version: claim.row.installationVersion, ownerEpoch: this.epoch }, data: { enabled: false } }); }); } async tick() { if (this.running || this.stopped || !this.options.config.adminToken) return; this.running = true; + this.dirty = false; + const reconsider = this.reconsider; this.reconsider = false; + let retryAt: number | undefined; + let acquired = false; try { - await this.initialize(); if (!await this.acquire()) return; + await this.initialize(); + acquired = await this.acquire(); + if (!acquired) { retryAt = this.now() + 30_000; return; } // Only the lease holder recovers expired intents. Never resend a potentially accepted provider request. await this.options.db.agentPushDelivery.updateMany({ where: { status: 'intent', updatedAt: { lt: date(this.now() - 30_000) } }, data: { status: 'ambiguous', outcome: 'interrupted_send' } }); let more = true; - for (let i = 0; i < 10 && more; i++) { if (!await this.acquire()) return; more = await this.observe(); } - if (more) return; // Current native read/prompt/presence must be caught up before dispatch. + for (let i = 0; i < 10 && more; i++) { if (!await this.acquire()) { retryAt = this.now() + 30_000; return; } more = await this.observe(); } + if (more) { retryAt = this.now() + 100; return; } // Catch up before dispatch. + if (reconsider) await this.options.db.agentPushDelivery.updateMany({ where: { status: 'pending', lastSuppressionReason: { not: null }, dueAt: { gt: date(this.now()) } }, data: { dueAt: date(this.now()) } }); const pending = await this.options.db.agentPushDelivery.findMany({ where: { status: 'pending', dueAt: { lte: date(this.now()) } }, orderBy: [{ priority: 'desc' }, { dueAt: 'asc' }], take: 20, select: { id: true } }); for (const row of pending) { - if (this.stopped || !await this.acquire()) break; + if (this.stopped) break; + if (!await this.acquire()) { retryAt = this.now() + 30_000; break; } // Fetch fresh runtime native presence/read/latest prompt before EACH provider boundary. if (await this.observe()) break; await this.dispatch(row.id); } await this.options.db.agentPushDelivery.updateMany({ where: { status: 'pending', expiresAt: { lte: date(this.now()) } }, data: { status: 'cancelled', outcome: 'expired' } }); - await this.options.db.agentPushPresence.deleteMany({ where: { expiresAt: { lt: date(this.now() - 86_400_000) } } }); - await this.options.db.agentPushConnection.deleteMany({ where: { expiresAt: { lt: date(this.now()) } } }); - await this.options.db.agentPushDelivery.deleteMany({ where: { status: { notIn: activeStatuses }, updatedAt: { lt: date(this.now() - 7 * 86_400_000) } } }); + if (this.now() >= this.nextMaintenance) { + await this.options.db.agentPushPresence.deleteMany({ where: { expiresAt: { lt: date(this.now() - 86_400_000) } } }); + await this.options.db.agentPushConnection.deleteMany({ where: { expiresAt: { lt: date(this.now()) } } }); + await this.options.db.agentPushDelivery.deleteMany({ where: { status: { notIn: activeStatuses }, updatedAt: { lt: date(this.now() - 7 * 86_400_000) } } }); + this.nextMaintenance = this.now() + 3_600_000; + } + this.failures = 0; } catch { + retryAt = this.now() + Math.min(30_000, 1_000 * 2 ** Math.min(this.failures++, 5)); // An unavailable source, DB, malformed contract or clock uncertainty blocks dispatch without leaking payloads. try { await this.options.db.$transaction(async tx => { @@ -380,8 +440,14 @@ export class AgentPushCoordinator { }); } catch { /* Retry on next loop. */ } } finally { - try { await this.options.db.agentPushCoordinator.updateMany({ where: { id: 'root', lockOwner: this.instanceId }, data: { lockOwner: null, lockUntil: null } }); } catch { /* Lease expiry recovers. */ } + try { if (acquired) await this.options.db.agentPushCoordinator.updateMany({ where: { id: 'root', lockOwner: this.instanceId }, data: { lockOwner: null, lockUntil: null } }); } catch { /* Lease expiry recovers. */ } + if (retryAt) this.reconsider ||= reconsider; this.running = false; + if (this.enabled && !this.stopped) { + if (retryAt) this.schedule(retryAt); + else if (this.dirty) this.schedule(this.now() + 100); + else try { await this.schedulePending(); } catch { this.schedule(this.now() + 30_000); } + } } } } @@ -390,6 +456,7 @@ export async function registerPushRoutes(app: FastifyInstance, db: PrismaClient, overrides?: Partial): Promise { const config = overrides?.config || loadPushConfig(); const coordinator = new AgentPushCoordinator({ db, config, broadcast, + subscribe: changed => subscribeNotificationEvents(upstream, changed), context: async (after, sourceId) => { if (!upstream.apiKey) throw new Error('Runtime unavailable'); // A replaced/restored runtime can reject the old cursor before returning its identity. diff --git a/server/tests/agent-push-events.test.ts b/server/tests/agent-push-events.test.ts new file mode 100644 index 0000000..5623ed7 --- /dev/null +++ b/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(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); +}); diff --git a/server/tests/agent-push.test.ts b/server/tests/agent-push.test.ts index 29cc9da..fd9bfa6 100644 --- a/server/tests/agent-push.test.ts +++ b/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, ...value }; }, + context(value: Partial) { 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) { + 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); +}); diff --git a/server/tests/agent.test.ts b/server/tests/agent.test.ts index 8f674b4..b5d0012 100644 --- a/server/tests/agent.test.ts +++ b/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 => {