feat(agent): preview public replies in push notifications

server/src/agent-push.ts

@@ -8,10 +8,11 @@ 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>; send: PushSend; now?: () => number;
+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'];
@@ -28,6 +29,7 @@ export class AgentPushCoordinator {
   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;
@@ -261,6 +263,7 @@ export class AgentPushCoordinator {
       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> {
@@ -278,8 +281,8 @@ export class AgentPushCoordinator {
     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 => {
+  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 } });
@@ -313,17 +316,31 @@ export class AgentPushCoordinator {
         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) } });
-      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 };
       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); }
+    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;
@@ -398,7 +415,7 @@ export async function registerPushRoutes(app: FastifyInstance, db: PrismaClient,
       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;
-    }, send: createPushSender(config), ...overrides });
+    }, 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 };