fix(02-11): release guardian lock before provider egress

- Commit a final authorization claim under the lifecycle lock\n- Keep provider calls and retry outcomes outside database transactions\n- Retain authorization claims with notification delivery diagnostics

src/__tests__/notification-delivery-identity.test.ts

@@ -1,6 +1,8 @@
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
 const transactionMock = vi.fn();
+const authorizationClaimQueryMock = vi.fn();
+const authorizationUpdateMock = vi.fn();
 
 vi.mock("@/lib/db", () => ({
   prisma: {
@@ -11,6 +13,9 @@ vi.mock("@/lib/db", () => ({
     },
     user: { findUnique: vi.fn() },
     accountGrant: { findMany: vi.fn() },
+    notificationEgressAuthorization: {
+      update: (...args: unknown[]) => authorizationUpdateMock(...args),
+    },
     auditLog: { create: vi.fn() },
     $transaction: (...args: unknown[]) => transactionMock(...args),
   },
@@ -66,6 +71,8 @@ const recipientUserId = "guardian-recipient";
 
 beforeEach(() => {
   vi.resetAllMocks();
+  authorizationClaimQueryMock.mockResolvedValue([{ id: "authorization" }]);
+  authorizationUpdateMock.mockResolvedValue({ id: "authorization" });
   vi.mocked(prisma.user.findUnique).mockResolvedValue({
     managedProfileAt: new Date(),
   } as never);
@@ -85,7 +92,7 @@ beforeEach(() => {
   transactionMock.mockImplementation(
     async (operation: (tx: unknown) => Promise<unknown>) =>
       operation({
-        $queryRaw: vi.fn(),
+        $queryRaw: (...args: unknown[]) => authorizationClaimQueryMock(...args),
         user: {
           findUnique: vi.fn(async ({ where }: { where: { id: string } }) =>
             where.id === recordUserId
@@ -132,12 +139,79 @@ describe("notification delivery identity", () => {
         message: "Record schedule",
       }),
     );
-    // Preflight chooses recipient channels, then the provider call is
-    // re-authorized under the managed-profile lock for the actual egress.
+    // Preflight chooses recipient channels, then a short locked transaction
+    // records the final authorization claim before provider I/O starts.
     expect(transactionMock).toHaveBeenCalledTimes(2);
-    expect(transactionMock).toHaveBeenLastCalledWith(expect.any(Function), {
-      maxWait: 5_000,
-      timeout: 60_000,
+    expect(transactionMock).toHaveBeenLastCalledWith(expect.any(Function));
+    expect(authorizationClaimQueryMock).toHaveBeenCalledTimes(3);
+    expect(authorizationUpdateMock).toHaveBeenCalledWith({
+      where: { id: "authorization" },
+      data: { completedAt: expect.any(Date), outcome: "ok" },
+    });
+  });
+
+  it("does not invoke a provider when final Guardian authorization is revoked", async () => {
+    let transactionCount = 0;
+    transactionMock.mockImplementation(
+      async (operation: (tx: unknown) => Promise<unknown>) => {
+        transactionCount += 1;
+        return operation({
+          $queryRaw: (...args: unknown[]) =>
+            authorizationClaimQueryMock(...args),
+          user: {
+            findUnique: vi.fn(async ({ where }: { where: { id: string } }) =>
+              where.id === recordUserId
+                ? { managedProfileAt: new Date() }
+                : { managedProfileAt: null },
+            ),
+          },
+          accountGrant: {
+            findFirst: vi.fn(async () =>
+              transactionCount === 1 ? { id: "grant" } : null,
+            ),
+          },
+        });
+      },
+    );
+
+    const outcome = await dispatchNotification({
+      eventType: "MEDICATION_REMINDER",
+      userId: recordUserId,
+      recordUserId,
+      recipientUserId,
+      title: "Record content",
+      message: "Record schedule",
+    });
+
+    expect(outcome).toEqual({
+      dispatched: false,
+      channelsAttempted: 0,
+      channelsSucceeded: 0,
+    });
+    expect(authorizationClaimQueryMock).toHaveBeenCalledTimes(2);
+    expect(sendViaWebPushMock).not.toHaveBeenCalled();
+  });
+
+  it("records a provider timeout on the claim without retaining the lifecycle lock", async () => {
+    sendViaWebPushMock.mockRejectedValue(new Error("provider timeout"));
+
+    const outcome = await dispatchNotification({
+      eventType: "MEDICATION_REMINDER",
+      userId: recordUserId,
+      recordUserId,
+      recipientUserId,
+      title: "Record content",
+      message: "Record schedule",
+    });
+
+    expect(outcome).toEqual({
+      dispatched: false,
+      channelsAttempted: 1,
+      channelsSucceeded: 0,
+    });
+    expect(authorizationUpdateMock).toHaveBeenCalledWith({
+      where: { id: "authorization" },
+      data: { completedAt: expect.any(Date), outcome: "sender_threw" },
     });
   });
 

src/lib/jobs/reminder/__tests__/notification-retention.test.ts

@@ -3,6 +3,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
 const h = vi.hoisted(() => ({
   pushDeleteMany: vi.fn(),
   eventDeleteMany: vi.fn(),
+  authorizationDeleteMany: vi.fn(),
   meta: {} as Record<string, unknown>,
 }));
 
@@ -11,6 +12,7 @@ vi.mock("../shared", async (importOriginal) => ({
   getWorkerPrisma: () => ({
     pushAttempt: { deleteMany: h.pushDeleteMany },
     notificationEvent: { deleteMany: h.eventDeleteMany },
+    notificationEgressAuthorization: { deleteMany: h.authorizationDeleteMany },
   }),
 }));
 
@@ -37,10 +39,11 @@ beforeEach(() => {
   for (const key of Object.keys(h.meta)) delete h.meta[key];
   h.pushDeleteMany.mockResolvedValue({ count: 3 });
   h.eventDeleteMany.mockResolvedValue({ count: 5 });
+  h.authorizationDeleteMany.mockResolvedValue({ count: 7 });
 });
 
 describe("notification retention", () => {
-  it("prunes record event anchors with the canonical bounded attempt retention", async () => {
+  it("prunes notification ledgers with the canonical bounded attempt retention", async () => {
     await handlePushAttemptCleanup([]);
 
     expect(h.pushDeleteMany).toHaveBeenCalledWith({
@@ -49,7 +52,11 @@ describe("notification retention", () => {
     expect(h.eventDeleteMany).toHaveBeenCalledWith({
       where: { createdAt: { lt: expect.any(Date) } },
     });
+    expect(h.authorizationDeleteMany).toHaveBeenCalledWith({
+      where: { authorizedAt: { lt: expect.any(Date) } },
+    });
     expect(h.meta.push_attempt_cleanup_deleted).toBe(3);
     expect(h.meta.notification_event_cleanup_deleted).toBe(5);
+    expect(h.meta.notification_egress_authorization_cleanup_deleted).toBe(7);
   });
 });

src/lib/jobs/reminder/cleanup-handlers.ts

@@ -127,9 +127,9 @@ export async function handleMoodReminderCleanup(
  * notification events are claims rather than delivery diagnostics, but after
  * the same horizon their dedup windows have elapsed and they have no reader.
  *
- * The DELETE is unbounded by user. Both ledgers therefore carry a direct
- * `created_at` index, so a `WHERE created_at < cutoff` scan is bounded by the
- * size of the trailing edge rather than the live working set. On a
+ * The DELETE is unbounded by user. The event and attempt ledgers carry direct
+ * `created_at` indexes, and final egress claims carry `authorized_at`, so each
+ * trailing-edge scan is bounded by the size of the live working set. On a
  * one-million-row table with the documented retention
  * window, the daily prune touches ~11k rows (1M / 90d × 1d) and
  * completes in milliseconds.
@@ -153,9 +153,20 @@ export async function handlePushAttemptCleanup(
       const deletedEvents = await p.notificationEvent.deleteMany({
         where: { createdAt: { lt: cutoff } },
       });
-      const deleted = deletedAttempts.count + deletedEvents.count;
+      const deletedAuthorizations =
+        await p.notificationEgressAuthorization.deleteMany({
+          where: { authorizedAt: { lt: cutoff } },
+        });
+      const deleted =
+        deletedAttempts.count +
+        deletedEvents.count +
+        deletedAuthorizations.count;
       evt.addMeta("push_attempt_cleanup_deleted", deletedAttempts.count);
       evt.addMeta("notification_event_cleanup_deleted", deletedEvents.count);
+      evt.addMeta(
+        "notification_egress_authorization_cleanup_deleted",
+        deletedAuthorizations.count,
+      );
       return jobDone({ deleted });
     } catch (err) {
       evt.addWarning(`push-attempt-cleanup failed: ${err}`);

src/lib/notifications/delivery-identity.ts

@@ -1,10 +1,13 @@
+import { randomUUID } from "node:crypto";
+
 import { prisma } from "@/lib/db";
 import {
   activeGuardianWhere,
   withManagedProfileLock,
 } from "@/lib/managed-profiles/lifecycle";
+import type { Prisma } from "@/generated/prisma/client";
 import type {
-  ManagedGuardianFanoutEvent,
+  ChannelType,
   NotificationPayload,
 } from "@/lib/notifications/types";
 
@@ -82,11 +85,11 @@ export async function resolveManagedGuardianRecipientIds(
  * own destination. Legacy payloads remain valid only for ordinary self
  * delivery; a managed record must name an active Guardian explicitly. This is
  * a preflight for recipient channel selection; managed delivery is authorized
- * again under the same lock at each provider egress below.
+ * again under the same lock at each provider egress claim below.
  *
- * Cross-principal authorization linearizes at the locked grant read below.
- * The egress helper keeps the same lock through the actual provider call, so
- * a revocation cannot commit between Guardian authorization and delivery.
+ * Cross-principal authorization preflight uses the locked grant read below.
+ * The final authorization is a separate short transaction that commits a
+ * durable egress claim immediately before the provider operation starts.
  */
 export async function resolveNotificationDeliveryIdentity(
   payload: NotificationPayload,
@@ -124,54 +127,112 @@ export async function resolveNotificationDeliveryIdentity(
 }
 
 /**
- * Execute a single managed Guardian channel send while holding the lifecycle
- * advisory lock. The provider call intentionally remains inside this short
- * transaction: managed Guardian revocation and expiry transitions take the
- * same lock, so neither can commit after this authorization but before egress.
+ * Create the final authorization claim for one managed Guardian channel.
+ *
+ * The claim commits while holding the same lifecycle lock as Guardian
+ * revocation and expiry. Its commit is the linearization point: if a
+ * revocation commits first, this returns null and no provider operation may
+ * start. If the claim commits first, it authorizes one provider operation
+ * that is already in flight from the lifecycle's perspective. The provider
+ * operation runs only after this transaction commits, so slow networks never
+ * retain a database transaction, advisory lock, or pooled connection. This is
+ * the logical send time, not a promise that a later revocation or expiry can
+ * recall an already-authorized provider operation before its first byte.
  */
-export async function withManagedGuardianEgressAuthorization<T>(
+export async function claimManagedGuardianEgressAuthorization(
   delivery: NotificationDeliveryIdentity,
-  send: () => Promise<T>,
-): Promise<T | null> {
-  if (!delivery.managed) return send();
+  channel: ChannelType,
+  eventType: string,
+): Promise<{ id: string } | null> {
+  if (!delivery.managed) return null;
 
   return withManagedGuardianAuthorization(
     delivery.recordUserId,
     delivery.recipientUserId,
-    send,
-    { maxWait: 5_000, timeout: 60_000 },
+    async (tx) => {
+      const id = randomUUID();
+      const claims = await tx.$queryRaw<Array<{ id: string }>>`
+        INSERT INTO "notification_egress_authorizations" (
+          "id",
+          "record_user_id",
+          "recipient_user_id",
+          "channel",
+          "event_type",
+          "authorized_at"
+        )
+        SELECT
+          ${id},
+          ${delivery.recordUserId},
+          ${delivery.recipientUserId},
+          ${channel},
+          ${eventType},
+          clock_timestamp()
+        WHERE EXISTS (
+          SELECT 1
+          FROM "account_grants"
+          WHERE "grantor_id" = ${delivery.recordUserId}
+            AND "grantee_id" = ${delivery.recipientUserId}
+            AND "access" = 'MANAGE'::"account_grant_access"
+            AND "accepted_at" IS NOT NULL
+            AND "revoked_at" IS NULL
+            AND (
+              "expires_at" IS NULL
+              OR "expires_at" > clock_timestamp()
+            )
+        )
+        RETURNING "id"
+      `;
+      return claims[0] ?? null;
+    },
   );
 }
 
+/**
+ * Record a completed provider operation outside the lifecycle transaction.
+ * This is observability only: authorization always rechecks the active grant,
+ * and every retry creates a new claim regardless of a prior outcome.
+ */
+export async function completeManagedGuardianEgressAuthorization(
+  authorizationId: string,
+  outcome: "ok" | "hard_reject" | "transient_failure" | "sender_threw",
+): Promise<void> {
+  try {
+    await prisma.notificationEgressAuthorization.update({
+      where: { id: authorizationId },
+      data: { completedAt: new Date(), outcome },
+    });
+  } catch {
+    // The claim is intentionally durable even if the best-effort completion
+    // marker cannot be written after an otherwise independent provider call.
+  }
+}
+
 async function withManagedGuardianAuthorization<T>(
   recordUserId: string,
   recipientUserId: string,
-  operation: () => Promise<T>,
-  options?: { maxWait: number; timeout: number },
+  operation: (tx: Prisma.TransactionClient) => Promise<T>,
 ): Promise<T | null> {
-  return prisma.$transaction(
-    (tx) =>
-      withManagedProfileLock(tx, recordUserId, async (lockedRecord) => {
-        if (lockedRecord?.managedProfileAt == null) return null;
+  return prisma.$transaction((tx) =>
+    withManagedProfileLock(tx, recordUserId, async (lockedRecord) => {
+      if (lockedRecord?.managedProfileAt == null) return null;
 
-        const recipient = await tx.user.findUnique({
-          where: { id: recipientUserId },
-          select: { managedProfileAt: true },
-        });
-        if (!recipient || recipient.managedProfileAt != null) return null;
+      const recipient = await tx.user.findUnique({
+        where: { id: recipientUserId },
+        select: { managedProfileAt: true },
+      });
+      if (!recipient || recipient.managedProfileAt != null) return null;
 
-        const guardian = await tx.accountGrant.findFirst({
-          where: {
-            grantorId: recordUserId,
-            granteeId: recipientUserId,
-            ...activeGuardianWhere(new Date()),
-          },
-          select: { id: true },
-        });
-        if (!guardian) return null;
+      const guardian = await tx.accountGrant.findFirst({
+        where: {
+          grantorId: recordUserId,
+          granteeId: recipientUserId,
+          ...activeGuardianWhere(new Date()),
+        },
+        select: { id: true },
+      });
+      if (!guardian) return null;
 
-        return operation();
-      }),
-    options,
+      return operation(tx);
+    }),
   );
 }

src/lib/notifications/dispatcher.ts

@@ -31,9 +31,10 @@ import {
 } from "@/lib/notifications/client-managed-apns";
 import { recordPushAttempt } from "@/lib/notifications/senders/push-attempt-record";
 import {
+  claimManagedGuardianEgressAuthorization,
+  completeManagedGuardianEgressAuthorization,
   resolveManagedGuardianRecipientIds,
   resolveNotificationDeliveryIdentity,
-  withManagedGuardianEgressAuthorization,
 } from "@/lib/notifications/delivery-identity";
 
 /**
@@ -300,25 +301,54 @@ export async function dispatchNotification(
       }
 
       try {
-        const outcome = await withManagedGuardianEgressAuthorization(
-          delivery,
-          () =>
-            sendToChannel(
-              channel.type as ChannelType,
-              channel.config,
-              delivery.recipientUserId,
-              recipientPayload,
-            ),
+        const prepared = prepareChannelSend(
+          channel.type as ChannelType,
+          channel.config,
+          delivery.recipientUserId,
+          recipientPayload,
         );
-        if (outcome === null) {
-          getEvent()?.addWarning(
-            "Guardian notification authorization changed before channel egress",
-          );
-          continue;
+        let outcome: SendOutcome;
+        if ("outcome" in prepared) {
+          outcome = prepared.outcome;
+          channelsAttempted += 1;
+        } else {
+          const authorization = delivery.managed
+            ? await claimManagedGuardianEgressAuthorization(
+                delivery,
+                channel.type as ChannelType,
+                recipientPayload.eventType,
+              )
+            : null;
+          if (delivery.managed && authorization === null) {
+            getEvent()?.addWarning(
+              "Guardian notification authorization changed before channel egress",
+            );
+            continue;
+          }
+
+          channelsAttempted += 1;
+          try {
+            outcome = await prepared.send();
+          } catch (err) {
+            if (authorization) {
+              await completeManagedGuardianEgressAuthorization(
+                authorization.id,
+                "sender_threw",
+              );
+            }
+            throw err;
+          }
+          if (authorization) {
+            await completeManagedGuardianEgressAuthorization(
+              authorization.id,
+              outcome.ok
+                ? "ok"
+                : outcome.hardReject
+                  ? "hard_reject"
+                  : "transient_failure",
+            );
+          }
         }
-
-        channelsAttempted += 1;
-
         if (outcome.ok) {
           await recordChannelSuccess({
             id: channel.id,
@@ -387,12 +417,20 @@ export async function dispatchNotification(
   };
 }
 
-async function sendToChannel(
+type PreparedChannelSend =
+  { send: () => Promise<SendOutcome> } | { outcome: SendOutcome };
+
+/**
+ * Decrypt and validate a recipient-owned channel before the final managed
+ * Guardian authorization claim. The returned closure is the only code that
+ * can begin provider I/O, and is invoked immediately after that claim commits.
+ */
+function prepareChannelSend(
   type: ChannelType,
   encryptedConfig: string,
   recipientUserId: string,
   payload: NotificationPayload,
-): Promise<SendOutcome> {
+): PreparedChannelSend {
   let decrypted: string;
   try {
     decrypted = decrypt(encryptedConfig);
@@ -406,9 +444,11 @@ async function sendToChannel(
     // a key-rotation gap every channel would otherwise auto-disable for the
     // wrong reason.
     return {
-      ok: false,
-      hardReject: true,
-      reason: `${type.toLowerCase()}_config_decrypt_failed`,
+      outcome: {
+        ok: false,
+        hardReject: true,
+        reason: `${type.toLowerCase()}_config_decrypt_failed`,
+      },
     };
   }
 
@@ -421,12 +461,14 @@ async function sendToChannel(
         getEvent()?.addWarning("Failed to parse Telegram channel config");
         // Malformed config decodes the same on every retry — permanent.
         return {
-          ok: false,
-          hardReject: true,
-          reason: "telegram_config_parse_failed",
+          outcome: {
+            ok: false,
+            hardReject: true,
+            reason: "telegram_config_parse_failed",
+          },
         };
       }
-      return sendViaTelegram(config, payload);
+      return { send: () => sendViaTelegram(config, payload) };
     }
     case "NTFY": {
       let config: NtfyChannelConfig;
@@ -436,12 +478,14 @@ async function sendToChannel(
         getEvent()?.addWarning("Failed to parse ntfy channel config");
         // Malformed config decodes the same on every retry — permanent.
         return {
-          ok: false,
-          hardReject: true,
-          reason: "ntfy_config_parse_failed",
+          outcome: {
+            ok: false,
+            hardReject: true,
+            reason: "ntfy_config_parse_failed",
+          },
         };
       }
-      return sendViaNtfy(config, payload);
+      return { send: () => sendViaNtfy(config, payload) };
     }
     case "WEBHOOK": {
       let config: WebhookChannelConfig;
@@ -451,12 +495,14 @@ async function sendToChannel(
         getEvent()?.addWarning("Failed to parse webhook channel config");
         // Malformed config decodes the same on every retry — permanent.
         return {
-          ok: false,
-          hardReject: true,
-          reason: "webhook_config_parse_failed",
+          outcome: {
+            ok: false,
+            hardReject: true,
+            reason: "webhook_config_parse_failed",
+          },
         };
       }
-      return sendViaWebhook(config, payload);
+      return { send: () => sendViaWebhook(config, payload) };
     }
     case "EMAIL": {
       let config: EmailChannelConfig;
@@ -466,15 +512,17 @@ async function sendToChannel(
         getEvent()?.addWarning("Failed to parse email channel config");
         // Malformed config decodes the same on every retry — permanent.
         return {
-          ok: false,
-          hardReject: true,
-          reason: "email_config_parse_failed",
+          outcome: {
+            ok: false,
+            hardReject: true,
+            reason: "email_config_parse_failed",
+          },
         };
       }
-      return sendViaEmail(config, payload);
+      return { send: () => sendViaEmail(config, payload) };
     }
     case "WEB_PUSH": {
-      return sendViaWebPush(recipientUserId, payload);
+      return { send: () => sendViaWebPush(recipientUserId, payload) };
     }
     case "APNS": {
       // APNs config lives on the Device row, not in the channel config.
@@ -482,14 +530,16 @@ async function sendToChannel(
       // we just keep the symmetric path so future per-user overrides
       // (e.g. a custom APNs topic) can hang off the channel without
       // changing the dispatcher shape.
-      return sendViaApns(recipientUserId, payload);
+      return { send: () => sendViaApns(recipientUserId, payload) };
     }
     default:
       getEvent()?.addWarning(`Unknown notification channel type: ${type}`);
       return {
-        ok: false,
-        hardReject: false,
-        reason: "unknown_channel_type",
+        outcome: {
+          ok: false,
+          hardReject: false,
+          reason: "unknown_channel_type",
+        },
       };
   }
 }