server/src/agent-push.ts

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;
}