441 lines
35 KiB
TypeScript
441 lines
35 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
|
import type { Prisma, PrismaClient } from '@prisma/client';
|
|
import type { FastifyInstance, FastifyRequest } from 'fastify';
|
|
import { requireAdmin } from './auth.js';
|
|
import { env } from './env.js';
|
|
import type { PushConfig } from './agent-push-config.js';
|
|
import { loadPushConfig } from './agent-push-config.js';
|
|
import { Cursor, InstallationBody, InstallationId, MAX_AGE_MS, MAX_ATTEMPTS, NotificationContext, PresenceBody, PUSH_POLICY,
|
|
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';
|
|
|
|
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<Context>; messageText: MessageText; send: PushSend; now?: () => number;
|
|
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 {
|
|
readonly instanceId = randomUUID();
|
|
readonly epoch: string;
|
|
private readonly now: () => number;
|
|
private readonly warmUntil: number;
|
|
private started = false;
|
|
private untrackedConnections = new Map<string, { at: number; kind: 'web' | 'unknown' }>();
|
|
private running = false;
|
|
private timer?: ReturnType<typeof setInterval>;
|
|
private stopped = false;
|
|
private observedRoot?: { sourceId: string; conversationId: string };
|
|
constructor(readonly options: Options) {
|
|
this.epoch = ownerEpoch(options.config.adminToken || 'disabled');
|
|
this.now = options.now || Date.now;
|
|
this.warmUntil = this.now() + (options.warmupMs ?? PUSH_POLICY.presenceLeaseMs);
|
|
}
|
|
async initialize() {
|
|
if (this.started) return;
|
|
await this.options.db.$transaction(async tx => {
|
|
const current = await tx.agentPushCoordinator.findUnique({ where: { id: 'root' } });
|
|
if (!current) await tx.agentPushCoordinator.create({ data: { id: 'root', ownerEpoch: this.epoch } });
|
|
else if (current.ownerEpoch !== this.epoch) {
|
|
// Credentials rotate ownership generation; old endpoint opt-ins do not cross it.
|
|
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, unknownSince: null, uncertain: true, lockOwner: null, lockUntil: null } });
|
|
}
|
|
});
|
|
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();
|
|
}
|
|
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' } });
|
|
const config = this.options.config;
|
|
return { version: 1, enabled: config.enabled, apnsAvailable: Boolean(config.enabled && config.apns),
|
|
webPushAvailable: Boolean(config.enabled && config.web), ...(config.enabled && config.web ? { vapidPublicKey: config.web.publicKey } : {}),
|
|
sourceId: state?.sourceId || null, quarantined: state?.quarantined || false, policy: PUSH_POLICY };
|
|
}
|
|
async register(id: string, value: Registration) {
|
|
await this.initialize();
|
|
const config = this.options.config;
|
|
if (!config.enabled) throw fail(503, 'Agent push is disabled');
|
|
if (value.transport === 'apns') {
|
|
if (!config.apns || config.apns.topic !== value.topic || config.apns.environment !== value.environment) throw fail(400, 'APNs topic or environment unavailable');
|
|
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 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 };
|
|
if (await tx.agentPushInstallation.count({ where: { ownerEpoch: this.epoch, enabled: true } }) >= 64 && !old?.enabled) throw fail(429, 'Too many push installations');
|
|
// A rotated/recreated local installation cannot retain another active delivery to the same endpoint.
|
|
const duplicates = await tx.agentPushInstallation.findMany({ where: { fingerprint, revokedAt: null, id: { not: id } }, select: { id: true } });
|
|
if (duplicates.length) {
|
|
await tx.agentPushInstallation.updateMany({ where: { id: { in: duplicates.map(v => v.id) } }, data: { enabled: false, revokedAt: date(this.now()), payload: '', fingerprint: '' } });
|
|
await tx.agentPushDelivery.updateMany({ where: { installationId: { in: duplicates.map(v => v.id) }, status: { in: activeStatuses } }, data: { status: 'cancelled', outcome: 'endpoint_rebound' } });
|
|
}
|
|
await tx.agentPushDelivery.updateMany({ where: { installationId: id, status: { in: activeStatuses } }, data: { status: 'cancelled', outcome: 'registration_changed' } });
|
|
const record = await tx.agentPushInstallation.upsert({ where: { id }, create: { id, ownerEpoch: this.epoch, transport: value.transport, payload, fingerprint, enabled: true },
|
|
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 };
|
|
});
|
|
}
|
|
async revoke(id: string) {
|
|
await this.options.db.$transaction(async tx => {
|
|
const old = await tx.agentPushInstallation.findUnique({ where: { id } });
|
|
if (!old?.revokedAt) await tx.agentPushInstallation.upsert({ where: { id },
|
|
create: { id, ownerEpoch: this.epoch, transport: 'revoked', payload: '', fingerprint: '', enabled: false, revokedAt: date(this.now()) },
|
|
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' } });
|
|
});
|
|
}
|
|
async presence(id: string, value: Presence) {
|
|
const now = this.now();
|
|
return 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');
|
|
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) {
|
|
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) {
|
|
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;
|
|
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()) } });
|
|
}
|
|
async readState(tx: Tx | PrismaClient = this.options.db): Promise<ReadState> {
|
|
const row = await tx.agentReadState.findUnique({ where: { id: 'root' } });
|
|
return { cursor: row?.cursor || '0', updatedAt: row?.updatedAt.toISOString() || null, sourceId: row?.sourceId || null };
|
|
}
|
|
private async mergeRead(tx: Tx, cursor: string, sourceId: string | null): Promise<ReadState | undefined> {
|
|
const old = await tx.agentReadState.findUnique({ where: { id: 'root' } });
|
|
if (old?.sourceId && sourceId && old.sourceId !== sourceId) return;
|
|
if (Number(cursor) <= Number(old?.cursor || 0)) return;
|
|
await tx.$executeRaw`INSERT INTO "AgentReadState" ("id", "cursor", "updatedAt", "sourceId") VALUES ('root', ${cursor}, CURRENT_TIMESTAMP, ${sourceId})
|
|
ON CONFLICT("id") DO UPDATE SET "cursor" = excluded."cursor", "updatedAt" = CURRENT_TIMESTAMP,
|
|
"sourceId" = COALESCE("AgentReadState"."sourceId", excluded."sourceId")
|
|
WHERE CAST(excluded."cursor" AS INTEGER) > CAST("AgentReadState"."cursor" AS INTEGER)`;
|
|
if (sourceId) {
|
|
await tx.$executeRaw`UPDATE "AgentPushDelivery" SET "status" = 'cancelled', "outcome" = 'read', "updatedAt" = CURRENT_TIMESTAMP
|
|
WHERE "ownerEpoch" = ${this.epoch} AND "sourceId" = ${sourceId} AND "kind" = 'message' AND "status" = 'pending' AND CAST("cursor" AS INTEGER) <= ${Number(cursor)}`;
|
|
if (this.options.config.enabled && this.options.config.apns) {
|
|
const installations = await tx.agentPushInstallation.findMany({ where: { ownerEpoch: this.epoch, enabled: true, transport: 'apns' } });
|
|
for (const installation of installations) {
|
|
// Do not send background clears to installations that never received an alert.
|
|
const delivered = await tx.agentPushDelivery.findFirst({ where: { installationId: installation.id, sourceId, kind: 'message', status: { in: ['accepted', 'intent', 'ambiguous'] } } });
|
|
if (!delivered) continue;
|
|
await tx.agentPushDelivery.updateMany({ where: { installationId: installation.id, kind: 'read_state', status: 'pending' }, data: { status: 'cancelled', outcome: 'superseded' } });
|
|
const id = digest(`clear:${installation.id}:${installation.version}:${sourceId}:${cursor}`);
|
|
const now = this.now();
|
|
await tx.agentPushDelivery.upsert({ where: { id }, update: {}, create: { id, ownerEpoch: this.epoch, installationId: installation.id,
|
|
installationVersion: installation.version, sourceId, cursor, messageId: `read:${cursor}`, kind: 'read_state', trigger: 'read',
|
|
dueAt: date(Math.max(now, (installation.lastClearAt?.getTime() || 0) + 60_000)), expiresAt: date(now + MAX_AGE_MS) } });
|
|
}
|
|
}
|
|
}
|
|
return this.readState(tx);
|
|
}
|
|
async advanceRead(cursor: string, suppliedSource?: string): Promise<ReadState> {
|
|
await this.initialize();
|
|
const result = await this.options.db.$transaction(async tx => {
|
|
const coordinator = await tx.agentPushCoordinator.findUnique({ where: { id: 'root' } });
|
|
if (suppliedSource && coordinator?.sourceId !== suppliedSource) throw fail(409, 'Agent source changed; refresh before acknowledging');
|
|
// Quarantine never relabels old read positions as a different runtime.
|
|
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);
|
|
return result.state;
|
|
}
|
|
private async acquire(): Promise<boolean> {
|
|
const now = this.now();
|
|
const result = await this.options.db.agentPushCoordinator.updateMany({ where: { id: 'root', ownerEpoch: this.epoch,
|
|
OR: [{ lockOwner: this.instanceId }, { lockUntil: null }, { lockUntil: { lte: date(now) } }] },
|
|
data: { lockOwner: this.instanceId, lockUntil: date(now + 30_000) } });
|
|
return result.count === 1;
|
|
}
|
|
private async observe(): Promise<boolean> {
|
|
const state = await this.options.db.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } });
|
|
const context = parseSafe(NotificationContext, await this.options.context(state.sourceId ? state.cursor : undefined, state.sourceId || undefined));
|
|
const now = this.now(), observedAt = Date.parse(context.webPresence.observedAt);
|
|
if ((!state.sourceId || (state.sourceId === context.sourceId && Number(context.cursor) >= Number(state.cursor))) && (observedAt > now + 5_000 || observedAt < now - 10_000 || (state.contextObservedAt && observedAt < state.contextObservedAt.getTime())
|
|
|| (context.latestUserMessageAt && Date.parse(context.latestUserMessageAt) > now + 5_000)
|
|
|| (state.latestUserMessageAt && (!context.latestUserMessageAt || Date.parse(context.latestUserMessageAt) < state.latestUserMessageAt.getTime())))) throw new Error('Uncertain runtime clock');
|
|
const read = await this.options.db.$transaction(async tx => {
|
|
const current = await tx.agentPushCoordinator.findUniqueOrThrow({ where: { id: 'root' } });
|
|
if (current.lockOwner !== this.instanceId || current.ownerEpoch !== this.epoch) throw new Error('Lost coordinator lease');
|
|
if ((current.sourceId && current.sourceId !== context.sourceId) || (current.sourceId && Number(context.cursor) < Number(current.cursor))) {
|
|
await tx.agentPushCoordinator.update({ where: { id: 'root' }, data: { quarantined: true, uncertain: true } });
|
|
await tx.agentPushDelivery.updateMany({ where: { status: 'pending' }, data: { status: 'cancelled', outcome: 'source_changed' } });
|
|
return undefined;
|
|
}
|
|
if (current.quarantined) return undefined;
|
|
const initial = !current.sourceId;
|
|
const previousCursor = Number(current.cursor);
|
|
let candidateCursor = previousCursor;
|
|
for (const candidate of context.candidates) {
|
|
if (Number(candidate.cursor) < candidateCursor || Number(candidate.cursor) > Number(context.cursor)) throw new Error('Invalid source order');
|
|
candidateCursor = Number(candidate.cursor);
|
|
}
|
|
await tx.agentPushCoordinator.update({ where: { id: 'root' }, data: { sourceId: context.sourceId, cursor: context.cursor,
|
|
latestUserMessageAt: context.latestUserMessageAt ? date(Date.parse(context.latestUserMessageAt)) : null,
|
|
nativeActive: context.webPresence.active, nativeUnknown: context.webPresence.unknown, contextObservedAt: date(observedAt), uncertain: context.uncertain,
|
|
lockUntil: date(now + 30_000) } });
|
|
await tx.agentReadState.upsert({ where: { id: 'root' }, create: { id: 'root', cursor: '0', sourceId: context.sourceId }, update: {} });
|
|
await tx.agentReadState.updateMany({ where: { id: 'root', sourceId: null }, data: { sourceId: context.sourceId } });
|
|
const changedRead = await this.mergeRead(tx, context.nativeReadCursor, context.sourceId);
|
|
const readState = await this.readState(tx);
|
|
if (!initial && this.options.config.enabled && !context.uncertain) {
|
|
const installations = await tx.agentPushInstallation.findMany({ where: { ownerEpoch: this.epoch, enabled: true } });
|
|
for (const candidate of context.candidates) {
|
|
const createdAt = Date.parse(candidate.createdAt);
|
|
if (Number(candidate.cursor) <= previousCursor || Number(candidate.cursor) <= Number(readState.cursor) || candidate.trigger === 'unknown'
|
|
|| createdAt > now + 5_000 || createdAt + MAX_AGE_MS <= now || (candidate.trigger === 'user' && !context.latestUserMessageAt)) continue;
|
|
for (const installation of installations) {
|
|
const id = digest(`${this.epoch}:${installation.id}:${installation.version}:${context.sourceId}:${candidate.messageId}`);
|
|
if (await tx.agentPushDelivery.findUnique({ where: { id } })) continue;
|
|
const older = await tx.agentPushDelivery.findFirst({ where: { installationId: installation.id, sourceId: context.sourceId, kind: 'message', status: 'pending' }, orderBy: { createdAt: 'asc' } });
|
|
await tx.agentPushDelivery.updateMany({ where: { installationId: installation.id, sourceId: context.sourceId, kind: 'message', status: 'pending' }, data: { status: 'cancelled', outcome: 'coalesced' } });
|
|
const burstStart = older?.createdAt.getTime() || now;
|
|
await tx.agentPushDelivery.create({ data: { id, ownerEpoch: this.epoch, installationId: installation.id, installationVersion: installation.version,
|
|
sourceId: context.sourceId, messageId: candidate.messageId, cursor: candidate.cursor, kind: 'message', trigger: candidate.trigger,
|
|
priority: Math.max(older?.priority || 0, ['timer', 'external_event'].includes(candidate.trigger) ||
|
|
(context.latestUserMessageAt && createdAt - Date.parse(context.latestUserMessageAt) >= 30_000) ? 2 : 1),
|
|
createdAt: date(burstStart), dueAt: date(Math.max(Math.min(now + PUSH_POLICY.debounceMs, burstStart + 15_000),
|
|
context.latestUserMessageAt ? Date.parse(context.latestUserMessageAt) + PUSH_POLICY.promptGraceMs : 0)), expiresAt: date(createdAt + MAX_AGE_MS) } });
|
|
}
|
|
}
|
|
}
|
|
// Keep the durable active outbox plus at most 10,000 completed receipts. The source checkpoint,
|
|
// not receipt retention, prevents historical replay after cleanup.
|
|
await tx.$executeRaw`DELETE FROM "AgentPushDelivery" WHERE "id" IN (
|
|
SELECT "id" FROM "AgentPushDelivery" WHERE "status" NOT IN ('pending','intent') ORDER BY "updatedAt" ASC
|
|
LIMIT MAX(0, (SELECT COUNT(*) FROM "AgentPushDelivery" WHERE "status" NOT IN ('pending','intent')) - 10000))`;
|
|
return changedRead;
|
|
});
|
|
if (read) this.options.broadcast?.(read);
|
|
this.observedRoot = { sourceId: context.sourceId, conversationId: context.conversationId };
|
|
return context.hasMore;
|
|
}
|
|
private async browserSuppression(tx: Tx, now: number): Promise<Suppression | undefined> {
|
|
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) } } });
|
|
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 claim(id: string, commit: boolean) {
|
|
return this.options.db.$transaction(async tx => {
|
|
const now = this.now();
|
|
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
|
|
|| row.ownerEpoch !== this.epoch || row.sourceId !== state.sourceId || state.quarantined || (row.kind === 'message' && Number(row.cursor) <= Number(read.cursor))) {
|
|
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) {
|
|
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));
|
|
if (registration.transport === 'web' && registration.subscription.expirationTime != null && registration.subscription.expirationTime <= now) {
|
|
await tx.agentPushInstallation.update({ where: { id: installation.id }, data: { enabled: false } });
|
|
await tx.agentPushDelivery.update({ where: { id }, data: { status: 'cancelled', outcome: 'expired_subscription' } }); return;
|
|
}
|
|
const identity: Identity = row.kind === 'read_state' ? { version: 1, sourceId: row.sourceId, kind: 'read_state', readCursor: read.cursor }
|
|
: { version: 1, sourceId: row.sourceId, kind: 'message', messageId: row.messageId, cursor: row.cursor };
|
|
if (!commit) return { row, registration, identity, now };
|
|
// Durable crossing intent is committed BEFORE invoking an external adapter. A crash here is ambiguous, never replayed.
|
|
await tx.agentPushDelivery.update({ where: { id }, data: { status: 'intent', attempts: { increment: 1 }, outcome: null } });
|
|
await tx.agentPushCoordinator.update({ where: { id: 'root' }, data: { lockUntil: date(now + 30_000) } });
|
|
if (row.kind === 'read_state') await tx.agentPushInstallation.update({ where: { id: installation.id }, data: { lastClearAt: date(now) } });
|
|
return { row, registration, identity, now };
|
|
});
|
|
}
|
|
private async dispatch(id: string) {
|
|
// Only resolve text for an otherwise dispatchable candidate, outside the DB transaction.
|
|
// Recheck source/read/presence and every claim condition after the bounded public lookup.
|
|
const ready = await this.claim(id, false);
|
|
if (!ready) return;
|
|
let messageText: string | undefined;
|
|
if (ready.identity.kind === 'message') {
|
|
if (this.observedRoot?.sourceId !== ready.identity.sourceId) throw new Error('Public Agent source unavailable');
|
|
messageText = await this.options.messageText({ ...ready.identity, conversationId: this.observedRoot.conversationId });
|
|
if (await this.observe()) return;
|
|
}
|
|
const claim = await this.claim(id, true);
|
|
if (!claim) return;
|
|
let result;
|
|
try { result = await this.options.send(claim.registration, claim.identity, claim.row.expiresAt.getTime(), claim.now, messageText); }
|
|
catch { result = { outcome: 'ambiguous' as const }; }
|
|
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))) } : {}) } });
|
|
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;
|
|
try {
|
|
await this.initialize(); if (!await this.acquire()) 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.
|
|
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;
|
|
// 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) } } });
|
|
} catch {
|
|
// An unavailable source, DB, malformed contract or clock uncertainty blocks dispatch without leaking payloads.
|
|
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;
|
|
}
|
|
}
|
|
}
|
|
|
|
export async function registerPushRoutes(app: FastifyInstance, db: PrismaClient, upstream: { baseUrl: string; apiKey?: string }, broadcast: (state: ReadState) => void,
|
|
overrides?: Partial<Options>): Promise<AgentPushCoordinator> {
|
|
const config = overrides?.config || loadPushConfig();
|
|
const coordinator = new AgentPushCoordinator({ db, config, broadcast,
|
|
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.
|
|
// Share one deadline across the incremental request and its single high-water probe.
|
|
const signal = AbortSignal.timeout(10_000);
|
|
const request = async (cursor?: string) => {
|
|
const url = new URL('/api/notification-context', upstream.baseUrl);
|
|
if (cursor !== undefined) url.searchParams.set('after', cursor); url.searchParams.set('limit', '200');
|
|
return fetch(url, { headers: { Authorization: `Bearer ${upstream.apiKey}`, 'X-Sybil-Client': 'backend-proxy-v1' }, signal, redirect: 'error' });
|
|
};
|
|
let response = await request(after);
|
|
const probe = response.status === 409 && after !== undefined;
|
|
if (probe) { await response.body?.cancel(); response = await request(); }
|
|
if (!response.ok) { await response.body?.cancel(); throw new Error('Runtime unavailable'); }
|
|
const reader = response.body?.getReader(); if (!reader) throw new Error('Invalid runtime response');
|
|
const chunks: Uint8Array[] = []; let length = 0;
|
|
try { for (;;) { const { done, value } = await reader.read(); if (done) break; length += value.length;
|
|
if (length > 512 * 1024) throw new Error('Invalid runtime response'); chunks.push(value); }
|
|
} finally { await reader.cancel(); }
|
|
const context = parseSafe(NotificationContext, JSON.parse(Buffer.concat(chunks).toString()));
|
|
// This probe may only establish quarantine. Never use it to skip a backlog, advance
|
|
// the checkpoint, merge new reads, or replay historical candidates after a generic409.
|
|
if (probe && (!sourceId || context.hasMore || context.candidates.length > 0
|
|
|| (context.sourceId === sourceId && Number(context.cursor) >= Number(after)))) throw new Error('Runtime cursor unavailable');
|
|
return context;
|
|
}, messageText: createMessageText(upstream), send: createPushSender(config), ...overrides });
|
|
const auth = (req: FastifyRequest) => { if (!env.ADMIN_TOKEN) throw fail(503, 'Agent push requires authenticated access'); requireAdmin(req); };
|
|
const safe = <T>(fn: () => Promise<T>): Promise<T> => fn().catch(error => { if (error?.statusCode) throw error; throw fail(503, 'Agent push is unavailable'); });
|
|
const routeOptions = { bodyLimit: 16 * 1024 };
|
|
app.get('/v1/agent/push/config', async (req, reply) => { reply.header('Cache-Control', 'no-store'); auth(req); return safe(() => coordinator.config()); });
|
|
app.put<{ Params: { installationId: string } }>('/v1/agent/push/installations/:installationId', routeOptions, async (req, reply) => {
|
|
reply.header('Cache-Control', 'no-store'); auth(req);
|
|
const id = parseSafe(InstallationId, req.params.installationId), value = parseSafe(InstallationBody, req.body);
|
|
return safe(() => coordinator.register(id, value));
|
|
});
|
|
app.delete<{ Params: { installationId: string } }>('/v1/agent/push/installations/:installationId', async (req, reply) => {
|
|
reply.header('Cache-Control', 'no-store'); auth(req); await safe(() => coordinator.revoke(parseSafe(InstallationId, req.params.installationId))); return reply.code(204).send();
|
|
});
|
|
app.put<{ Params: { sessionId: string } }>('/v1/agent/presence/:sessionId', routeOptions, async (req, reply) => {
|
|
reply.header('Cache-Control', 'no-store'); auth(req); return safe(() => coordinator.presence(parseSafe(InstallationId, req.params.sessionId), parseSafe(PresenceBody, req.body)));
|
|
});
|
|
app.delete<{ Params: { sessionId: string } }>('/v1/agent/presence/:sessionId', async (req, reply) => {
|
|
reply.header('Cache-Control', 'no-store'); auth(req); await safe(() => coordinator.retirePresence(parseSafe(InstallationId, req.params.sessionId))); return reply.code(204).send();
|
|
});
|
|
app.addHook('onReady', async () => { if (config.adminToken && upstream.apiKey) coordinator.start(); });
|
|
app.addHook('onClose', async () => coordinator.stop());
|
|
return coordinator;
|
|
}
|