fix: Codex round 16 — preserve live-expired scheduled approvals, DB-time claims

1. ApprovalLifecycle.expire: for a scheduled fire, omit completedAt so the aborted
   job is preserved (cross-store signal — Redis keeps the running TTL via
   transitionStatus's preserveTerminal, the in-memory finished-job sweep keys on
   completedAt), letting reconcile settle the requires_action run in ~2 min instead
   of the 25-h abandoned-pause cutoff. Covers the runtime expiry path that
   round-15's store-cleanup fix missed, for both stores.
2. claimDueSchedule: compare nextRunAt/leaseUntil against MongoDB $$NOW (via $expr
   and an aggregation-pipeline $set for the lease) instead of each worker's process
   clock, so replica clock skew can't claim future occurrences early or mis-time a
   lease. Drops the now-unused process-clock override param.

packages/api/src/stream/ApprovalLifecycle.ts

@@ -112,13 +112,22 @@ export class ApprovalLifecycle {
    * `expectedActionId` for the same stale-decision protection as `resolve`.
    */
   async expire(streamId: string, expectedActionId?: string): Promise<boolean> {
+    // A scheduled fire's expired approval must be RETAINED so the schedules
+    // reconciler can read `jobStatus === 'aborted'` and settle its `requires_action`
+    // run within ~2 min. Omitting completedAt is the cross-store preserve signal
+    // (Redis keeps it on the running TTL; the in-memory finished-job sweep keys on
+    // completedAt), so it isn't reaped before reconciliation — unlike a normal
+    // expired approval, which sets completedAt so terminal-cleanup can reclaim it.
+    const job = await this.store.getJob(streamId).catch(() => null);
+    const preserveForSchedule = job?.scheduleId != null;
     const ok = await this.store.transitionStatus(streamId, {
       from: 'requires_action',
       to: 'aborted',
       clear: ['pendingAction', 'pendingActionId'],
-      // completedAt lets the stores' terminal-cleanup reclaim the job; without
-      // it an expired approval lingers in the in-memory map indefinitely.
-      patch: { error: 'Approval expired before a decision was made', completedAt: Date.now() },
+      patch: {
+        error: 'Approval expired before a decision was made',
+        ...(preserveForSchedule ? {} : { completedAt: Date.now() }),
+      },
       expectActionId: expectedActionId,
     });
     if (ok) {

packages/data-schemas/src/methods/schedule.ts

@@ -21,7 +21,6 @@ const COUNTED_FOR_WINDOW = 64;
 export interface ClaimDueScheduleParams {
   instanceId: string;
   leaseMs: number;
-  now?: Date;
 }
 
 export interface RecordRunOutcomeParams {
@@ -172,20 +171,32 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
    * due schedule regardless of replica count, with or without Redis.
    */
   async function claimDueSchedule(params: ClaimDueScheduleParams): Promise<ISchedule | null> {
-    const now = params.now ?? new Date();
+    // Compare due-ness and lease expiry against MongoDB's own clock (`$$NOW`), not
+    // each worker's process clock: all replicas race on the persisted nextRunAt /
+    // leaseUntil, so a skewed worker must not claim future occurrences early or set
+    // a mis-timed lease. `nextRunAt` existence is gated by the plain filter (a bare
+    // $expr $lte would match a missing field as null); a missing leaseUntil is
+    // treated as epoch so it's always claimable.
     return Schedule()
       .findOneAndUpdate(
         {
           enabled: true,
-          nextRunAt: { $lte: now },
-          $or: [{ leaseUntil: { $exists: false } }, { leaseUntil: { $lt: now } }],
-        },
-        {
-          $set: {
-            leaseUntil: new Date(now.getTime() + params.leaseMs),
-            leaseBy: params.instanceId,
+          nextRunAt: { $exists: true, $ne: null },
+          $expr: {
+            $and: [
+              { $lte: ['$nextRunAt', '$$NOW'] },
+              { $lt: [{ $ifNull: ['$leaseUntil', new Date(0)] }, '$$NOW'] },
+            ],
           },
         },
+        [
+          {
+            $set: {
+              leaseUntil: { $add: ['$$NOW', params.leaseMs] },
+              leaseBy: params.instanceId,
+            },
+          },
+        ],
         { new: true, sort: { nextRunAt: 1 } },
       )
       .lean<ISchedule>();