From 5833a0868e53096d5966a34354bdcf9c741b055d Mon Sep 17 00:00:00 2001 From: James Magahern Date: Wed, 14 Oct 2026 23:06:34 -0700 Subject: [PATCH] fix(agent): bound push suppression to observed browser attention --- docs/api/rest.md | 12 ++- docs/api/streaming-chat.md | 6 +- .../migration.sql | 5 ++ server/prisma/schema.prisma | 5 ++ server/src/agent-push-contract.ts | 4 +- server/src/agent-push.ts | 89 ++++++++++++++----- server/src/agent.ts | 2 +- server/tests/agent-push.test.ts | 79 +++++++++++++++- web/src/lib/agent-push-browser.ts | 2 +- web/src/lib/agent-room.ts | 2 +- web/tests/agent-push.browser.mjs | 19 ++++ web/tests/agent-push.test.mjs | 2 +- web/tests/agent-room.test.mjs | 4 +- 13 files changed, 194 insertions(+), 37 deletions(-) create mode 100644 server/prisma/migrations/20261015010000_bound_agent_push_attention/migration.sql diff --git a/docs/api/rest.md b/docs/api/rest.md index 195cb9a..3394008 100644 --- a/docs/api/rest.md +++ b/docs/api/rest.md @@ -698,7 +698,7 @@ All push/config/presence routes require a configured `ADMIN_TOKEN` and bearer au ### `GET /v1/agent/push/config` ```json -{"version":1,"enabled":false,"apnsAvailable":false,"webPushAvailable":false,"sourceId":null,"quarantined":false,"policy":{"promptGraceMs":5000,"debounceMs":2000,"interactionGraceMs":30000,"presenceHeartbeatMs":15000,"presenceLeaseMs":60000}} +{"version":1,"enabled":false,"apnsAvailable":false,"webPushAvailable":false,"sourceId":null,"quarantined":false,"policy":{"promptGraceMs":5000,"debounceMs":2000,"interactionGraceMs":5000,"inactiveGraceMs":5000,"unknownGraceMs":5000,"presenceHeartbeatMs":15000,"presenceLeaseMs":60000}} ``` `vapidPublicKey` is present only when Web Push is enabled and its credentials are valid. No private keys, tokens, paths or configured account identifiers are returned. Source discovery and browser presence run independently of transport enablement/installation consent. Clients bind local consent to a known source UUID as well as the authenticated backend; null/unknown source cannot opt in. A source change retires prior local consent/installation and requires a new UUID plus explicit user opt-in, rechecking the source after asynchronous permission/subscription/registration work. `sourceId` is the persisted bound runtime UUID or null before discovery. A changed runtime source or regressed durable cursor quarantines delivery and cancels pending alerts without resetting the canonical read cursor. There is deliberately no HTTP read-reset/rebaseline endpoint; an operator must reconcile replacement runtime data with source-aware clients before enabling a new baseline. @@ -735,9 +735,11 @@ A UUID per browser document/tab, independent of notification permission and sele {"sequence":12,"visibility":"visible","focused":true,"interacted":false} ``` -Response `{ "expiresAt":"2026-10-15T03:00:00.000Z" }`. Sequence is a nonnegative signed 32-bit integer; stale/out-of-order heartbeats neither overwrite state nor extend the lease. Heartbeat every 15 seconds and on focus/visibility changes and recent pointer/key/touch input. Times are assigned by the server. Visible+focused is active even without recent interaction; any recent interaction blocks alerts for another 30 seconds. Leases expire after 60 seconds. All authenticated workspace sessions must report, even without notification opt-in; at most 512 unexpired sessions are accepted. Associate the UUID with WS auth as described in the streaming contract. Unknown authenticated web sockets block dispatch until classified or their heartbeat lease expires. +Response `{ "expiresAt":"2026-10-15T03:00:00.000Z" }`. Sequence is a nonnegative signed 32-bit integer; stale/out-of-order heartbeats neither overwrite state nor extend the lease. Report every 15 seconds and on focus, blur, visibility changes, pagehide/pageshow and trusted pointer/key/touch/wheel/input events. Observe `document.visibilityState`, `document.hidden` and `document.hasFocus()`; pagehide explicitly reports hidden/unfocused. Times are assigned by the server. Each authenticated document's explicit lease lasts 60 seconds. Visible **and focused** blocks even while idle. Interaction reported while visible blocks for 5 seconds, including a subsequent blur/hide; hidden reports cannot create new interaction evidence. A live focused-to-hidden/unfocused transition adds at most 5 seconds of grace. Repeated hidden/unfocused heartbeats do not renew that transition grace. All authenticated workspace sessions participate independently, even without a socket or notification opt-in; at most 512 unexpired sessions are accepted. An inactive tab cannot override another active tab. Associate the same UUID and authenticated credential generation with WS auth as described in the streaming contract. -`DELETE /v1/agent/presence/:sessionId` returns 204 and retires focus/visibility best effort; recent-interaction grace is retained until expiry. It does not revoke push consent. Backend restarts impose a full 60-second attention warm-up, with leases preserved across restart. +Unobserved authenticated clients (including a declared web UUID with no matching observation) share **one persisted 5-second introduction window per credential generation**, starting at the first such connection. Socket pings/pongs, reconnects, new unobserved UUIDs and backend restarts do not renew it. With no stable anonymous identity, subsequent raw connections cannot earn a new window: this deliberately favors eventual notification eligibility over indefinite suppression by uninstrumented clients. Fresh explicit document observations always retain their own veto. Expiring an observed lease does not itself create unknown attention. Connection liveness and web attention are separate; neither a socket nor its absence proves foreground state. + +`DELETE /v1/agent/presence/:sessionId` returns 204 and releases focus/visibility best effort with the same short transition/interaction grace. It does not revoke push consent. Backend restarts still impose a separate 60-second safety warm-up, with explicit leases and the spent anonymous introduction window preserved across restart. If a page disappears without a successful hidden/blur receipt, its last explicit focus can block for the remaining 60-second lease. ### Scheduling, read reconciliation and delivery identities @@ -745,6 +747,8 @@ Only complete user-visible root assistant messages enter the durable outbox. Too All web sessions must be inactive at dispatch, including the runtime-native browser. Before each provider request, the backend refreshes runtime read/prompt/presence, merges native read into the canonical shared read row, and transactionally rechecks opt-in/version, owner generation, source freshness, read cursor and all attention leases. Unknown/malformed/unavailable source data suppresses sends. Messages expire after five minutes; provider TTL is at most 60 seconds. The active outbox coalesces per installation; completed receipts are capped at 10,000 and expire after seven days, while revocation tombstones remain durable. There are at most four attempts, only after definite retryable provider rejections, with bounded backoff. A persisted send intent protects crash recovery: interrupted/ambiguous network sends are never automatically replayed. Exactly-once provider/device delivery is not promised; a final focus/read race after dispatch remains possible. +Each outbox row retains nullable `lastSuppressionReason`, `lastSuppressionClientKind` and server `lastSuppressedAt` for operator diagnosis, even after read cancellation or a later send. These fields record one evaluated veto, not a full session census, and do not retroactively explain older rows. Reasons are fixed codes for foreground focus, recent interaction, inactive/unknown grace, runtime active/unknown, prompt grace, startup warm-up or source/coordinator unavailability. Client kinds are `web`, `unknown`, `runtime_web`, `runtime_unknown` or `system`; no token, session/connection/installation identifier, endpoint or transcript is logged. They are internal database diagnostics, not additions to client responses. Runtime unknown also covers its independent restart/clock/source safety checks; it is not proof that a browser was active. + Alert payload identity is top-level `{version:1,sourceId,kind:"message",messageId,cursor}`. APNs adds `aps.alert` with title **Sybil Agent**, body **You have a new message.**, default sound and `thread-id:sybil-agent-root`. Web Push adds those same generic title/body strings, fixed `url:"/?agent=root"` and `tag:"sybil-agent:"+sourceId`. No transcript preview, filename, event source, token or key appears in a notification. Normal APNs alerts do not request critical/time-sensitive entitlements. Monotonic canonical read advances cancel pending alerts and broadcast `read_state` on authenticated WS. New read-aware clients pass the response's exact source/cursor to their notification manager; receiving/tapping a push does not itself mark history read. Native APNs clear payload is `{version:1,sourceId,kind:"read_state",readCursor,aps:{"content-available":1}}`, push type background/priority 5, coalesced to at most one request per minute per installation that has an earlier delivered/ambiguous alert. App foreground/background handlers reconcile delivered OS notifications by source/cursor. Silent pushes are best effort and cannot guarantee removal when an app is terminated or force-quit; foreground read sync remains required. @@ -756,3 +760,5 @@ Monotonic canonical read advances cancel pending alerts and broadcast `read_stat 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. + +The runtime browser uses the same 15-second explicit heartbeat, 60-second lease and 5-second interaction/hidden/blur grace. Unobserved direct runtime sockets, event streams and REST pollers share one persisted 5-second introduction window per runtime source, beginning at their first establishment/observation. Polling and reconnects cannot renew it, including with fresh unobserved UUIDs. Existing source/clock uncertainty and the separate 60-second runtime restart safety interval still suppress; the backend never ignores a returned `unknown:true`. diff --git a/docs/api/streaming-chat.md b/docs/api/streaming-chat.md index 27392f9..496bab6 100644 --- a/docs/api/streaming-chat.md +++ b/docs/api/streaming-chat.md @@ -377,7 +377,9 @@ Attached input admission and its request binding are atomic. Replay verifies the ### Agent attention and shared read events (push v1) -Web auth frames add `{ "clientKind":"web", "presenceSessionId":"per-document-uuid" }` to the existing `type`, `token`, and optional `after`. Report the same UUID to `PUT /v1/agent/presence/:sessionId` across the entire authenticated workspace. Native iOS clients add `clientKind:"native"`; no browser-attention lease is required for a native app. These declarations classify a client under the existing shared principal and do not grant new authority. Old/unclassified authenticated sockets conservatively veto push delivery until closed/expired. A web-classified socket without a live matching presence receipt is also unknown. Transport ping/pong renews connection liveness, never user activity. The proxy classifies itself upstream using the authenticated `X-Sybil-Client: backend-proxy-v1` upgrade header. +Web auth frames always add `clientKind:"web"` to the existing `type`, `token`, and optional `after`. They also add `presenceSessionId:"per-document-uuid"` when secure UUID creation succeeds; omit only that field if unavailable. Report the same UUID to `PUT /v1/agent/presence/:sessionId` across the entire authenticated workspace. Native iOS clients add `clientKind:"native"`; no browser-attention lease is required for a native app. These declarations classify a client under the existing shared principal and do not grant new authority. Explicit document visibility/focus/interaction receipts, bound to that credential generation and session, determine attention independently of sockets. Every fresh visible/focused session suppresses, even while idle; hidden/blur transitions and recent visible interaction have 5-second grace, and unrenewed explicit observations expire after 60 seconds. + +Old/unclassified authenticated sockets and web sockets with no matching observation share one persisted 5-second introduction grace per credential generation. Reconnects, missing/new UUIDs and transport heartbeat cannot extend or restart it. A spent anonymous window does not weaken any genuine document lease; an expired observed lease does not become unknown merely because its socket remains connected. Ping/pong updates only existing connection liveness, preserving the original kind/session and never recreating a removed row as unknown. The proxy classifies itself upstream using the authenticated `X-Sybil-Client: backend-proxy-v1` upgrade header. The backend adds this optional server-to-client frame after a successful monotonic read advance, including a runtime-native browser acknowledgment merged into the shared row: @@ -389,4 +391,4 @@ The backend adds this optional server-to-client frame after a successful monoton Distinguish an unbound room from one already bound to a legacy absent/null source or a source UUID. Bind on the first accepted read state; if history arrives first, pin the legacy state before publishing it. A different source in a later GET, stream frame or PUT acknowledgment, including null-to-UUID or UUID-to-null, must stop stale read writes and require an explicit fresh room load. Preserve the old snapshot and cursors until that reload, then fetch read state before fresh history; never attach a new UUID to an old cursor. Ordinary foreground/network recovery must not reset this stopped room. HTTP 409, an invalid acknowledgment cursor or an acknowledgment below the requested cursor also stops the writer. A transient transport failure can retry only within the same pinned source. None of these client reloads resets server read state. -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; its unknown sessions participate in the same conservative veto. No Web Push is used for invisible read synchronization. +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. diff --git a/server/prisma/migrations/20261015010000_bound_agent_push_attention/migration.sql b/server/prisma/migrations/20261015010000_bound_agent_push_attention/migration.sql new file mode 100644 index 0000000..19ce1d8 --- /dev/null +++ b/server/prisma/migrations/20261015010000_bound_agent_push_attention/migration.sql @@ -0,0 +1,5 @@ +ALTER TABLE "AgentPushPresence" ADD COLUMN "inactiveUntil" DATETIME; +ALTER TABLE "AgentPushCoordinator" ADD COLUMN "unknownSince" DATETIME; +ALTER TABLE "AgentPushDelivery" ADD COLUMN "lastSuppressionReason" TEXT; +ALTER TABLE "AgentPushDelivery" ADD COLUMN "lastSuppressionClientKind" TEXT; +ALTER TABLE "AgentPushDelivery" ADD COLUMN "lastSuppressedAt" DATETIME; diff --git a/server/prisma/schema.prisma b/server/prisma/schema.prisma index 6e1cfb8..1e32ec8 100644 --- a/server/prisma/schema.prisma +++ b/server/prisma/schema.prisma @@ -249,6 +249,7 @@ model AgentPushPresence { visible Boolean focused Boolean lastInteractionAt DateTime? + inactiveUntil DateTime? expiresAt DateTime @@index([ownerEpoch, expiresAt]) } @@ -273,6 +274,7 @@ model AgentPushCoordinator { localPromptAt DateTime? nativeActive Boolean @default(false) nativeUnknown Boolean @default(true) + unknownSince DateTime? uncertain Boolean @default(true) lockOwner String? lockUntil DateTime? @@ -296,6 +298,9 @@ model AgentPushDelivery { expiresAt DateTime updatedAt DateTime @updatedAt outcome String? + lastSuppressionReason String? + lastSuppressionClientKind String? + lastSuppressedAt DateTime? @@index([status, dueAt]) @@index([installationId, status]) @@index([ownerEpoch, sourceId, cursor]) diff --git a/server/src/agent-push-contract.ts b/server/src/agent-push-contract.ts index 7f783d4..e198a7c 100644 --- a/server/src/agent-push-contract.ts +++ b/server/src/agent-push-contract.ts @@ -1,8 +1,8 @@ import { createHash, createPublicKey, ECDH } from 'node:crypto'; import { z } from 'zod'; -export const PUSH_POLICY = Object.freeze({ promptGraceMs: 5_000, debounceMs: 2_000, interactionGraceMs: 30_000, - presenceHeartbeatMs: 15_000, presenceLeaseMs: 60_000 }); +export const PUSH_POLICY = Object.freeze({ promptGraceMs: 5_000, debounceMs: 2_000, interactionGraceMs: 5_000, + inactiveGraceMs: 5_000, unknownGraceMs: 5_000, presenceHeartbeatMs: 15_000, presenceLeaseMs: 60_000 }); export const MAX_AGE_MS = 5 * 60_000; export const MAX_ATTEMPTS = 4; export const ROOT_OWNER = 'root'; diff --git a/server/src/agent-push.ts b/server/src/agent-push.ts index 57151d5..389bf65 100644 --- a/server/src/agent-push.ts +++ b/server/src/agent-push.ts @@ -15,6 +15,7 @@ type Options = { db: PrismaClient; config: PushConfig; context: (after?: string, broadcast?: (state: ReadState) => void; warmupMs?: number }; const date = (ms: number) => new Date(ms); const activeStatuses = ['pending', 'intent']; +type Suppression = { reason: string; clientKind: 'web' | 'unknown' | 'runtime_web' | 'runtime_unknown' | 'system' }; /** Sole coordinator: persisted source checkpoint, installations, presence, claims and outcomes. */ export class AgentPushCoordinator { @@ -23,7 +24,7 @@ export class AgentPushCoordinator { private readonly now: () => number; private readonly warmUntil: number; private started = false; - private untrackedConnections = new Set(); + private untrackedConnections = new Map(); private running = false; private timer?: ReturnType; private stopped = false; @@ -42,7 +43,7 @@ export class AgentPushCoordinator { await tx.agentPushInstallation.updateMany({ where: { enabled: true }, data: { enabled: false } }); await tx.agentPushDelivery.updateMany({ where: { status: { in: activeStatuses } }, data: { status: 'cancelled', outcome: 'owner_changed' } }); await tx.agentPushCoordinator.update({ where: { id: 'root' }, data: { ownerEpoch: this.epoch, contextObservedAt: null, - nativeUnknown: true, uncertain: true, lockOwner: null, lockUntil: null } }); + nativeUnknown: true, unknownSince: null, uncertain: true, lockOwner: null, lockUntil: null } }); } }); this.started = true; @@ -102,22 +103,46 @@ export class AgentPushCoordinator { 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'); - const data = { ownerEpoch: this.epoch, sequence: value.sequence, visible: value.visibility === 'visible', focused: value.focused, - lastInteractionAt: value.interacted ? date(now) : old?.ownerEpoch === this.epoch ? old.lastInteractionAt : null, expiresAt: date(now + PUSH_POLICY.presenceLeaseMs) }; + const previous = old?.ownerEpoch === this.epoch ? old : null; + const visible = value.visibility === 'visible', focused = visible && value.focused; + const data = { ownerEpoch: this.epoch, sequence: value.sequence, visible, focused, + lastInteractionAt: value.interacted && visible ? date(now) : previous?.lastInteractionAt || null, + // Only a transition out of proven focus starts grace. Inactive heartbeats cannot renew it. + inactiveUntil: focused ? null : previous?.visible && previous.focused && previous.expiresAt.getTime() > now ? date(now + PUSH_POLICY.inactiveGraceMs) : previous?.inactiveUntil || null, + expiresAt: date(now + PUSH_POLICY.presenceLeaseMs) }; const row = await tx.agentPushPresence.upsert({ where: { id }, create: { id, ...data }, update: data }); return { expiresAt: row.expiresAt.toISOString() }; }); } async retirePresence(id: string) { - // Keep recent interaction's grace even after a best-effort pagehide release. - await this.options.db.agentPushPresence.updateMany({ where: { id, ownerEpoch: this.epoch }, data: { visible: false, focused: false } }); + await this.options.db.$transaction(async tx => { + const old = await tx.agentPushPresence.findUnique({ where: { id } }); + if (old?.ownerEpoch !== this.epoch) return; + 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 } }); + }); } async connect(id: string, kind?: string, sessionId?: string) { - this.untrackedConnections.add(id); - await this.options.db.agentPushConnection.upsert({ where: { id }, create: { id, ownerEpoch: this.epoch, kind: kind || 'unknown', sessionId, - expiresAt: date(this.now() + PUSH_POLICY.presenceLeaseMs) }, update: { expiresAt: date(this.now() + PUSH_POLICY.presenceLeaseMs) } }); + const now = this.now(); + if (kind !== 'native' && !this.untrackedConnections.has(id)) this.untrackedConnections.set(id, { at: now, kind: kind === 'web' ? 'web' : 'unknown' }); + await this.initialize(); + await this.options.db.$transaction(async tx => { + const connection = await tx.agentPushConnection.upsert({ where: { id }, create: { id, ownerEpoch: this.epoch, kind: kind || 'unknown', sessionId, + expiresAt: date(now + PUSH_POLICY.presenceLeaseMs) }, update: { expiresAt: date(now + PUSH_POLICY.presenceLeaseMs) } }); + const observed = connection.kind === 'web' && connection.sessionId + ? await tx.agentPushPresence.findFirst({ where: { id: connection.sessionId, ownerEpoch: this.epoch } }) : null; + if (connection.kind !== 'native' && !observed) { + // Raw clients have no stable identity. One persisted introduction window per owner + // prevents reconnects (including invented UUIDs) from manufacturing attention. + await tx.agentPushCoordinator.updateMany({ where: { id: 'root', ownerEpoch: this.epoch, unknownSince: null }, data: { unknownSince: date(now) } }); + } + }); this.untrackedConnections.delete(id); } + 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 notePrompt() { if (!this.options.config.adminToken) return; @@ -238,12 +263,20 @@ export class AgentPushCoordinator { if (read) this.options.broadcast?.(read); return context.hasMore; } - private async browserActive(tx: Tx, now: number): Promise { - if (this.untrackedConnections.size) return true; - const presences = await tx.agentPushPresence.findMany({ where: { ownerEpoch: this.epoch, expiresAt: { gt: date(now) } } }); - if (presences.some(v => (v.visible && v.focused) || (v.lastInteractionAt && v.lastInteractionAt.getTime() + PUSH_POLICY.interactionGraceMs > now))) return true; + 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 connections = await tx.agentPushConnection.findMany({ where: { ownerEpoch: this.epoch, expiresAt: { gt: date(now) } } }); - return connections.some(v => v.kind !== 'native' && (v.kind !== 'web' || !presences.some(p => p.id === v.sessionId))); + 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', + clientKind: unobserved[0]?.kind === 'web' ? 'web' : firstUntracked?.kind || 'unknown' }; } private async dispatch(id: string) { const claim = await this.options.db.$transaction(async tx => { @@ -251,6 +284,10 @@ 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), + lastSuppressionReason: reason, lastSuppressionClientKind: clientKind, lastSuppressedAt: date(now) } }); + }; const installation = await tx.agentPushInstallation.findUnique({ where: { id: row.installationId } }); const read = await this.readState(tx); if (!installation?.enabled || installation.ownerEpoch !== this.epoch || installation.version !== row.installationVersion || row.expiresAt.getTime() <= now @@ -258,11 +295,17 @@ export class AgentPushCoordinator { await tx.agentPushDelivery.update({ where: { id }, data: { status: 'cancelled', outcome: 'obsolete' } }); return; } 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) return; - if (row.kind === 'message' && (state.nativeActive || state.nativeUnknown || await this.browserActive(tx, now) - || (state.latestUserMessageAt && state.latestUserMessageAt.getTime() + PUSH_POLICY.promptGraceMs > now) - || (state.localPromptAt && state.localPromptAt.getTime() + PUSH_POLICY.promptGraceMs > now))) { - await tx.agentPushDelivery.update({ where: { id }, data: { dueAt: date(now + PUSH_POLICY.debounceMs) } }); return; + || 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; + } + 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); + 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; + } } if (row.attempts >= MAX_ATTEMPTS) { await tx.agentPushDelivery.update({ where: { id }, data: { status: 'failed', outcome: 'attempt_limit' } }); return; } const registration = parseSafe(InstallationBody, JSON.parse(installation.payload)); @@ -312,7 +355,13 @@ export class AgentPushCoordinator { await this.options.db.agentPushDelivery.deleteMany({ where: { status: { notIn: activeStatuses }, updatedAt: { lt: date(this.now() - 7 * 86_400_000) } } }); } catch { // An unavailable source, DB, malformed contract or clock uncertainty blocks dispatch without leaking payloads. - try { await this.options.db.agentPushCoordinator.updateMany({ where: { id: 'root', lockOwner: this.instanceId }, data: { uncertain: true } }); } catch { /* Retry on next loop. */ } + try { + await this.options.db.$transaction(async tx => { + const held = await tx.agentPushCoordinator.updateMany({ where: { id: 'root', lockOwner: this.instanceId }, data: { uncertain: true } }); + if (held.count) await tx.agentPushDelivery.updateMany({ where: { ownerEpoch: this.epoch, status: 'pending' }, data: { + lastSuppressionReason: 'source_or_coordinator_unavailable', lastSuppressionClientKind: 'system', lastSuppressedAt: date(this.now()) } }); + }); + } 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. */ } this.running = false; diff --git a/server/src/agent.ts b/server/src/agent.ts index 5b94bea..56b8a92 100644 --- a/server/src/agent.ts +++ b/server/src/agent.ts @@ -138,7 +138,7 @@ function attachAgentSocket(app: FastifyInstance, baseUrl: string, apiKey: string }; const authTimeout = setTimeout(() => browser.close(4401, 'Authentication required'), 5000); const heartbeat = setInterval(() => { if (!alive) browser.terminate(); else { alive = false; browser.ping(); } }, 25_000); - browser.on('pong', () => { alive = true; if (authenticated && env.ADMIN_TOKEN) void push.connect(connectionId).catch(() => {}); }); + browser.on('pong', () => { alive = true; if (authenticated && env.ADMIN_TOKEN) void push.heartbeat(connectionId).catch(() => {}); }); const dispose = () => { if (disposed) return; disposed = true; clearTimeout(authTimeout); clearInterval(heartbeat); cleanup.delete(dispose); diff --git a/server/tests/agent-push.test.ts b/server/tests/agent-push.test.ts index 322a4ec..b7c5358 100644 --- a/server/tests/agent-push.test.ts +++ b/server/tests/agent-push.test.ts @@ -70,17 +70,88 @@ test('first source bootstrap never alerts history; later durable candidates debo const restarted = new AgentPushCoordinator({ ...f.manager.options, warmupMs: 0 }); await restarted.tick(); assert.equal(f.sent.length, 1); }); -test('all tabs veto: visible focused idle, recent interaction and unknown connections; leases eventually expire', async () => { - const f = await fixture(), tab1 = randomUUID(), tab2 = randomUUID(), unknown = randomUUID(); +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.connect(unknown); f.advance(31_000); await f.manager.tick(); assert.equal(f.sent.length, 0, 'unknown authenticated socket vetoes'); - f.advance(61_000); await f.manager.tick(); assert.equal(f.sent.length, 1, 'expired stale sessions retire'); + 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 () => { diff --git a/web/src/lib/agent-push-browser.ts b/web/src/lib/agent-push-browser.ts index 17cea08..da98b3e 100644 --- a/web/src/lib/agent-push-browser.ts +++ b/web/src/lib/agent-push-browser.ts @@ -94,7 +94,7 @@ export const disableAgentBrowserPush = () => controller.disable(); export const refreshAgentBrowserPush = () => controller.refresh(); const presence = createAgentPresenceReporter({ - snapshot: () => ({ visibility: document.visibilityState === "visible" ? "visible" : "hidden", focused: document.hasFocus() }), + snapshot: () => ({ visibility: !document.hidden && document.visibilityState === "visible" ? "visible" : "hidden", focused: document.hasFocus() }), send: (value, keepalive) => { const id = getAgentPresenceSessionId(); return id ? api.presence(id, value, keepalive) : Promise.reject(new Error("Attention session unavailable.")); }, }); let stopSession: (() => void) | undefined; diff --git a/web/src/lib/agent-room.ts b/web/src/lib/agent-room.ts index e0c1bdf..88d0ede 100644 --- a/web/src/lib/agent-room.ts +++ b/web/src/lib/agent-room.ts @@ -145,7 +145,7 @@ export function parseAgentRoomEvent(raw: string): AgentRoomEvent | null { } export function agentAuthFrame(token: string | null, cursor: string, presenceSessionId?: string): string { return JSON.stringify({ type: "auth", token, after: normalizeAgentCursor(cursor), - ...(agentPushUuid(presenceSessionId) ? { clientKind: "web", presenceSessionId } : {}) }); + clientKind: "web", ...(agentPushUuid(presenceSessionId) ? { presenceSessionId } : {}) }); } export type AgentSubmission = { id: number; status: "queued" | "placed" | "done" | "unanswered"; messageId: string; attachmentIds?: string[] }; diff --git a/web/tests/agent-push.browser.mjs b/web/tests/agent-push.browser.mjs index 26dc6d5..921e16c 100644 --- a/web/tests/agent-push.browser.mjs +++ b/web/tests/agent-push.browser.mjs @@ -108,6 +108,20 @@ try { assert.equal(sessionPaths.length, 2); const secondPath = sessionPaths.find(path => !path.endsWith(firstSession)); await expectEventually(() => requests.some(request => request.path === secondPath && request.body?.focused === true)); + // Playwright forces every target focused/visible. Supply controlled DOM state + // to exercise the production reporter's blur/visibility wiring deterministically. + const beforeBlur = requests.length; + await page.evaluate(() => { + Object.defineProperty(document, "hasFocus", { configurable: true, value: () => false }); + window.dispatchEvent(new Event("blur")); + }); + await expectEventually(() => requests.slice(beforeBlur).some(request => request.path.endsWith(firstSession) && request.body?.visibility === "visible" && request.body?.focused === false)); + await page.evaluate(() => { + Object.defineProperty(document, "hidden", { configurable: true, value: true }); + document.dispatchEvent(new Event("visibilitychange")); + }); + await expectEventually(() => requests.slice(beforeBlur).some(request => request.path.endsWith(firstSession) && request.body?.visibility === "hidden" && request.body?.focused === false)); + await page.evaluate(() => { delete document.hasFocus; delete document.hidden; window.dispatchEvent(new Event("focus")); }); await second.getByRole("textbox").first().click(); await second.getByRole("textbox").first().pressSequentially("interaction"); await expectEventually(() => requests.some(request => request.path === secondPath && request.body?.interacted === true)); @@ -116,6 +130,11 @@ try { assert.deepEqual(sequences, [...new Set(sequences)].sort((a, b) => a - b)); } await page.bringToFront(); + const beforeHide = requests.length; + await page.evaluate(() => window.dispatchEvent(new Event("pagehide"))); + await expectEventually(() => requests.slice(beforeHide).some(request => request.path.endsWith(firstSession) && request.body?.visibility === "hidden" && request.body?.focused === false)); + await page.evaluate(() => window.dispatchEvent(new Event("pageshow"))); + await expectEventually(() => requests.slice(beforeHide).some(request => request.path.endsWith(firstSession) && request.body?.visibility === "visible" && request.body?.focused === true)); const worker = context.serviceWorkers().find(worker => new URL(worker.url()).pathname === "/sw.js"); assert.ok(worker, "actual service worker activated"); const push = async cursor => worker.evaluate(async ({ cursor, sourceId }) => { diff --git a/web/tests/agent-push.test.mjs b/web/tests/agent-push.test.mjs index fa22920..ace2291 100644 --- a/web/tests/agent-push.test.mjs +++ b/web/tests/agent-push.test.mjs @@ -405,7 +405,7 @@ test("attention is per-tab, ordered, focused while idle, and recent input persis c.interact(); const report = c.report(); failure.reject(new Error("offline")); await assert.rejects(report); await c.report(); const id = randomUUID(), frame = JSON.parse(agentAuthFrame("fixture-token", "2", id)); assert.equal(frame.clientKind, "web"); assert.equal(frame.presenceSessionId, id); - assert.equal(JSON.parse(agentAuthFrame(null, "0", "not-an-id")).clientKind, undefined); + assert.deepEqual(JSON.parse(agentAuthFrame(null, "0", "not-an-id")), { type: "auth", token: null, after: "0", clientKind: "web" }); }); test("read frames validate source/cursor and preserve existing read semantics", () => { diff --git a/web/tests/agent-room.test.mjs b/web/tests/agent-room.test.mjs index 8e389e0..ce166d0 100644 --- a/web/tests/agent-room.test.mjs +++ b/web/tests/agent-room.test.mjs @@ -254,8 +254,8 @@ test("client uses Sybil auth and routes while websocket credentials remain in th const url = client.websocketUrl("https://sybil.example/"); assert.equal(url, "wss://sybil.example/api/v1/agent/ws"); assert.equal(new URL(url).search, ""); - assert.deepEqual(JSON.parse(agentAuthFrame(null, "42")), { type: "auth", token: null, after: "42" }); - assert.deepEqual(JSON.parse(agentAuthFrame(token, "42")), { type: "auth", token, after: "42" }); + assert.deepEqual(JSON.parse(agentAuthFrame(null, "42")), { type: "auth", token: null, after: "42", clientKind: "web" }); + assert.deepEqual(JSON.parse(agentAuthFrame(token, "42")), { type: "auth", token, after: "42", clientKind: "web" }); }); test("client sends the pinned source and exposes409 without requiring a JSON response body", async () => {