[codex] Preserve APNs delivery failure context

infra/relay/src/agentActivity/ApnsDeliveries.test.ts

@@ -7,7 +7,9 @@ import { describe, expect, it } from "@effect/vitest";
 import * as NodeCrypto from "node:crypto";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
+import * as Logger from "effect/Logger";
 import * as Redacted from "effect/Redacted";
+import * as References from "effect/References";
 import {
   FetchHttpClient,
   HttpClient,
@@ -629,6 +631,17 @@ describe("ApnsDeliveries", () => {
 
   it.effect("processes signed jobs through APNs and records attempts", () => {
     const attempts: Array<DeliveryAttempts.DeliveryAttemptInput> = [];
+    const transportErrors: Array<ApnsDeliveries.ApnsDeliveryTransportError> = [];
+    const logger = Logger.make(({ fiber }) => {
+      const annotation = fiber.getRef(References.CurrentLogAnnotations).error;
+      if (!Redacted.isRedacted(annotation)) {
+        return;
+      }
+      const error = Redacted.value(annotation);
+      if (ApnsDeliveries.isApnsDeliveryTransportError(error)) {
+        transportErrors.push(error);
+      }
+    });
     const payload = makeApnsDeliveryJobPayload({
       kind: "live_activity_update",
       userId: target.user_id,
@@ -657,7 +670,29 @@ describe("ApnsDeliveries", () => {
           token: "activity-token",
         },
       ]);
-    }).pipe(Effect.provide(makeLayer({ attempts })));
+      expect(transportErrors).toHaveLength(1);
+      const error = transportErrors[0]!;
+      expect(error).toMatchObject({
+        deviceId: target.device_id,
+        kind: "live_activity_update",
+        sourceJobId: "job-1",
+        apnsErrorTag: "ApnsJwtSigningError",
+        requestStage: null,
+      });
+      expect(error.cause).toBeInstanceOf(ApnsClient.ApnsJwtSigningError);
+      expect(error.cause).toMatchObject({
+        teamId: "team-id",
+        keyId: "key-id",
+      });
+      expect((error.cause as ApnsClient.ApnsJwtSigningError).cause).toBeDefined();
+    }).pipe(
+      Effect.provide(
+        Layer.mergeAll(
+          makeLayer({ attempts }),
+          Logger.layer([logger], { mergeWithExisting: false }),
+        ),
+      ),
+    );
   });
 
   it.effect("processes signed push notification jobs through APNs and records attempts", () => {

infra/relay/src/agentActivity/ApnsDeliveries.ts

@@ -7,12 +7,14 @@ import type {
 import {
   RelayAgentActivityAggregateState as RelayAgentActivityAggregateStateSchema,
   RelayAgentAwarenessPreferences as RelayAgentAwarenessPreferencesSchema,
+  RelayDeliveryKind as RelayDeliveryKindSchema,
 } from "@t3tools/contracts/relay";
 import * as Context from "effect/Context";
 import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
+import * as Redacted from "effect/Redacted";
 import * as Schema from "effect/Schema";
 
 import {
@@ -87,6 +89,28 @@ export class ApnsDeliveryJobClaimInFlight extends Schema.TaggedErrorClass<ApnsDe
   }
 }
 
+export class ApnsDeliveryTransportError extends Schema.TaggedErrorClass<ApnsDeliveryTransportError>()(
+  "ApnsDeliveryTransportError",
+  {
+    deviceId: Schema.String,
+    kind: RelayDeliveryKindSchema,
+    sourceJobId: Schema.NullOr(Schema.String),
+    apnsErrorTag: Schema.Literals([
+      "ApnsJwtEncodingError",
+      "ApnsJwtSigningError",
+      "ApnsHttpRequestError",
+    ]),
+    requestStage: Schema.NullOr(Schema.Literals(["send", "read-response"])),
+    cause: Schema.Defect(),
+  },
+) {
+  override get message(): string {
+    return `APNs ${this.kind} delivery failed for device ${this.deviceId}.`;
+  }
+}
+
+export const isApnsDeliveryTransportError = Schema.is(ApnsDeliveryTransportError);
+
 const decodeRelayAgentActivityAggregateStateJson = Schema.decodeUnknownOption(
   Schema.fromJsonString(RelayAgentActivityAggregateStateSchema),
 );
@@ -302,6 +326,42 @@ function deliveryAttemptOutcome(result: Apns.ApnsDeliveryResult) {
   };
 }
 
+const recoverApnsDeliveryTransportError = (
+  input: {
+    readonly deviceId: string;
+    readonly kind: RelayDeliveryKind;
+    readonly sourceJobId: string | null;
+  },
+  cause: Apns.ApnsError,
+): Effect.Effect<Apns.ApnsDeliveryResult> => {
+  const error = new ApnsDeliveryTransportError({
+    deviceId: input.deviceId,
+    kind: input.kind,
+    sourceJobId: input.sourceJobId,
+    apnsErrorTag: cause._tag,
+    requestStage: cause._tag === "ApnsHttpRequestError" ? cause.stage : null,
+    cause,
+  });
+  return Effect.logError(error.message).pipe(
+    Effect.annotateLogs({
+      error: Redacted.make(error, { label: error._tag }),
+      "error.type": error._tag,
+      "error.apns_error_tag": error.apnsErrorTag,
+      ...(error.requestStage === null ? {} : { "error.request_stage": error.requestStage }),
+      ...(error.stack === undefined ? {} : { "error.stack": error.stack }),
+      "relay.mobile.device_id": error.deviceId,
+      "relay.delivery.kind": error.kind,
+      ...(error.sourceJobId === null ? {} : { "relay.delivery.job_id": error.sourceJobId }),
+    }),
+    Effect.as({
+      ok: false,
+      status: 0,
+      reason: cause.message,
+      apnsId: null,
+    }),
+  );
+};
+
 interface LiveActivityDeliveryTarget {
   readonly user_id: string;
   readonly device_id: string;
@@ -440,6 +500,15 @@ export const make = Effect.gen(function* () {
       { ...input, aggregate } as SendLiveActivityDeliveryInput,
       now,
     );
+    const recoverTransportError = (cause: Apns.ApnsError) =>
+      recoverApnsDeliveryTransportError(
+        {
+          deviceId: input.target.device_id,
+          kind: input.kind,
+          sourceJobId: input.sourceJobId ?? null,
+        },
+        cause,
+      );
     if (input.sourceJobId) {
       const claim = yield* attempts.claimSourceJob({
         userId: input.target.user_id,
@@ -476,14 +545,11 @@ export const make = Effect.gen(function* () {
         issuedAtUnixSeconds: epochSeconds,
       })
       .pipe(
-        Effect.catch((error) =>
-          Effect.succeed({
-            ok: false,
-            status: 0,
-            reason: error.message,
-            apnsId: null,
-          }),
-        ),
+        Effect.catchTags({
+          ApnsJwtEncodingError: recoverTransportError,
+          ApnsJwtSigningError: recoverTransportError,
+          ApnsHttpRequestError: recoverTransportError,
+        }),
       );
     if (result.ok) {
       yield* liveActivities.markDelivery({
@@ -551,6 +617,15 @@ export const make = Effect.gen(function* () {
       token: input.token,
       notification,
     });
+    const recoverTransportError = (cause: Apns.ApnsError) =>
+      recoverApnsDeliveryTransportError(
+        {
+          deviceId: input.target.device_id,
+          kind: "push_notification",
+          sourceJobId: input.sourceJobId ?? null,
+        },
+        cause,
+      );
     if (input.sourceJobId) {
       const claim = yield* attempts.claimSourceJob({
         userId: input.target.user_id,
@@ -593,14 +668,11 @@ export const make = Effect.gen(function* () {
         issuedAtUnixSeconds: epochSeconds,
       })
       .pipe(
-        Effect.catch((error) =>
-          Effect.succeed({
-            ok: false,
-            status: 0,
-            reason: error.message,
-            apnsId: null,
-          }),
-        ),
+        Effect.catchTags({
+          ApnsJwtEncodingError: recoverTransportError,
+          ApnsJwtSigningError: recoverTransportError,
+          ApnsHttpRequestError: recoverTransportError,
+        }),
       );
     if (isPermanentApnsTokenFailure(result)) {
       yield* liveActivities.invalidateDeliveryToken({