fix(schedules): converge unprojected pauses and release replayed bookkeeping jobs

Two of the four findings I flagged as pre-existing to my closeout commits. Both are
in code this PR introduces, so merging would have shipped them; both are the same
capacity/evidence-leak class the last four rounds have been closing. The remaining
two (slotless rows escaping the per-user cap, and the non-rotating `started`
reconciliation bucket) are genuinely latent and deliberately left alone.

An UNPROJECTED PAUSE held a global capacity slot forever. The pause projection is
what moves a run row off `started`; `recordScheduleOutcome` already retries it, but
the request controller discarded the result, so three failed attempts were dropped
silently. The armed engine's reconciler replays that state, but the clustered sweep
did not: a paused job is not terminal, so the retained-job path returned without
settling, and the dead-delivery path never inspects an identity-matched job at all.
The row stayed `started` with no cutoff that would ever clear it.

The call site now surfaces the failure, and the sweep converges it, mirroring the
reconciler's pause branch: project `requires_action` (which frees the slot) but do
NOT release the job's evidence — unlike a terminal job it is still live, awaiting an
approval. The resume hand-off fence still defers, so re-projecting cannot release a
slot a continuation just claimed.

The BOOKKEEPING REPLAY pass leaked its retained job. A run reaches that pass only
because its owner crashed before bookkeeping — which is also before it could release
the job it retained for exactly this recovery. The active-run pass clears its own;
this one never did, and a preserved job is deliberately kept WITHOUT `completedAt`
so the store's finished-job sweep cannot reap it early, so nothing else ever would.
It now clears after `finalizeBookkeeping` succeeds — identity-guarded, a no-op when
no job is retained, and skipped entirely when the replay itself failed, since the
retained job is the only surviving evidence in that case.

Tests: pause projected and the slot freed while the live job's evidence is kept,
paused job deferred during a resume hand-off, retained job released once replayed
bookkeeping is durable, and retained job kept when the replay fails.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq

api/server/controllers/agents/request.js

@@ -1628,7 +1628,17 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
                 `[ResumableAgentController] Pause persistence barrier changed before release: ${streamId}`,
               );
             }
-            await settleScheduledRun({ status: 'requires_action' });
+            // The pause projection is what moves the run row off `started` and frees its
+            // GLOBAL capacity slot. recordScheduleOutcome already retried it; a `false`
+            // here means every attempt failed, leaving the row `started` while the job
+            // sits `requires_action`. Surface it — the armed engine's reconciler replays
+            // this state, and the clustered sweep now converges it too, but a silent drop
+            // gave neither a reason to look.
+            if (!(await settleScheduledRun({ status: 'requires_action' }))) {
+              logger.error(
+                `[ResumableAgentController] Failed to project the scheduled pause for ${streamId}; run stays active until reconciliation replays it`,
+              );
+            }
           } else {
             logger.debug(
               `[ResumableAgentController] Skipping stale pause persistence — ${streamId} no longer owns its barrier`,

packages/api/src/schedules/engine.spec.ts

@@ -401,6 +401,60 @@ describe('bookkeeping replay rotation', () => {
     // row's counters never land.
     expect(methods.markRunsReconciled).toHaveBeenCalledWith([poison]);
   });
+
+  /**
+   * A terminal run reaches this pass only because its OWNER crashed before bookkeeping —
+   * which is also before it could release the job it retained for exactly this recovery.
+   * A preserved job is kept WITHOUT `completedAt` so the store's finished-job sweep cannot
+   * reap it early, so if this pass does not clear it, nothing ever does.
+   */
+  it('releases the retained job once replayed bookkeeping is durable', async () => {
+    const methods = makeMethods(makeClaimedSchedule());
+    methods.claimDueSchedule.mockResolvedValue(null);
+    const unbookkept = {
+      _id: 'row-2',
+      scheduleId: 'sched-crashed',
+      scheduledFor: new Date(0),
+      user: 'u1',
+      status: 'success',
+      conversationId: 'convo-crashed',
+    };
+    (methods.getUnbookkeptRuns as jest.Mock).mockResolvedValue([unbookkept]);
+    const clearReconciledJob = jest.fn(async () => undefined);
+
+    await reconcileOnce(makeDeps(methods, { clearReconciledJob }));
+
+    expect(methods.finalizeBookkeeping).toHaveBeenCalledWith(
+      expect.objectContaining({ scheduleId: 'sched-crashed' }),
+    );
+    expect(clearReconciledJob).toHaveBeenCalledWith('convo-crashed', {
+      scheduleId: 'sched-crashed',
+      scheduledFor: new Date(0),
+    });
+  });
+
+  /** The retained job is the only surviving evidence when bookkeeping fails, so it must
+   *  outlive a failed replay rather than be cleared alongside it. */
+  it('keeps the retained job when the bookkeeping replay itself fails', async () => {
+    const methods = makeMethods(makeClaimedSchedule());
+    methods.claimDueSchedule.mockResolvedValue(null);
+    (methods.getUnbookkeptRuns as jest.Mock).mockResolvedValue([
+      {
+        _id: 'row-3',
+        scheduleId: 'sched-broken-bk',
+        scheduledFor: new Date(0),
+        user: 'u1',
+        status: 'success',
+        conversationId: 'convo-broken',
+      },
+    ]);
+    (methods.finalizeBookkeeping as jest.Mock).mockRejectedValue(new Error('mongo down'));
+    const clearReconciledJob = jest.fn(async () => undefined);
+
+    await reconcileOnce(makeDeps(methods, { clearReconciledJob }));
+
+    expect(clearReconciledJob).not.toHaveBeenCalled();
+  });
 });
 
 describe('reconciliation abort fence', () => {

packages/api/src/schedules/engine.ts

@@ -300,6 +300,25 @@ export function startScheduleEngine(deps: ScheduleEngineDeps): ScheduleEngine {
               autoDisableAfterFailures: runLimits.autoDisableAfterFailures,
               balanceSkipDisableThreshold: BALANCE_SKIP_DISABLE_THRESHOLD,
             });
+            // The run is terminal but its OWNER crashed before bookkeeping, which is also
+            // before it could release the job it retained for exactly this recovery. The
+            // active-run pass clears its own; this one never did, and a preserved job is
+            // kept WITHOUT `completedAt` precisely so the store's finished-job sweep
+            // cannot reap it early — so nothing else would ever have. Identity-guarded,
+            // and a no-op when no retained job is there.
+            if (run.conversationId) {
+              await deps
+                .clearReconciledJob(run.conversationId, {
+                  scheduleId: run.scheduleId,
+                  scheduledFor: run.scheduledFor,
+                })
+                .catch((clearError) =>
+                  logger.warn(
+                    `[schedules] failed to clear retained job after bookkeeping replay for ${run.scheduleId}:`,
+                    clearError,
+                  ),
+                );
+            }
           } catch (rowError) {
             logger.error(
               `[schedules] bookkeeping replay failed for run ${run.scheduleId}:`,

packages/api/src/schedules/erasure.spec.ts

@@ -262,6 +262,45 @@ describe('topology-safe dead-delivery convergence', () => {
     expect(clearReconciledJob).not.toHaveBeenCalled();
   });
 
+  /**
+   * A pause whose `requires_action` projection failed every retry leaves the row `started`
+   * while the job is durably paused. The armed reconciler replays that; with no engine the
+   * row held its GLOBAL capacity slot forever, since a paused job is not terminal and the
+   * dead-delivery path never inspects an identity-matched job.
+   */
+  it('projects a pause the owner never managed to record, freeing the slot', async () => {
+    const { methods, clearReconciledJob } = await convergeOnce({
+      job: {
+        status: 'requires_action',
+        scheduleId: 'schedule-1',
+        scheduledFor: '2026-08-17T12:00:00.000Z',
+      } as unknown as JobState,
+    });
+    expect(methods.recordRunOutcome).toHaveBeenCalledWith(
+      expect.objectContaining({
+        scheduleId: 'schedule-1',
+        status: 'requires_action',
+        conversationId: 'conversation-1',
+      }),
+    );
+    // The job is still LIVE awaiting approval — its evidence must not be released.
+    expect(clearReconciledJob).not.toHaveBeenCalled();
+  });
+
+  /** A resume claims capacity before flipping the job back to `running`; re-projecting the
+   *  pause during that hand-off would release the slot the continuation just took. */
+  it('defers a paused job while a resume hand-off is in flight', async () => {
+    const { methods } = await convergeOnce({
+      run: { resumeClaimedAt: new Date() } as Partial<IScheduleRun>,
+      job: {
+        status: 'requires_action',
+        scheduleId: 'schedule-1',
+        scheduledFor: '2026-08-17T12:00:00.000Z',
+      } as unknown as JobState,
+    });
+    expect(methods.recordRunOutcome).not.toHaveBeenCalled();
+  });
+
   it('never settles a terminal job while its abort is still in flight', async () => {
     const { methods, clearReconciledJob } = await convergeOnce({
       run: { abortRequestedAt: new Date() } as Partial<IScheduleRun>,

packages/api/src/schedules/erasure.ts

@@ -130,17 +130,35 @@ async function settleAbandonedRuns(deps: ScheduleErasureDeps, scheduleId: string
 }
 
 /**
- * Settles a run from its OWN identity-matched terminal job, then releases the retained
+ * Converges a run from its OWN identity-matched job — the clustered mirror of the engine
- * evidence — the clustered mirror of the engine reconciler's retained-job branch.
+ * reconciler's retained-job and pause branches.
  */
-async function settleFromRetainedJob(
+async function settleFromObservedJob(
   deps: ScheduleErasureDeps,
   run: IScheduleRun,
   job: JobState,
 ): Promise<void> {
+  // A PAUSE the owner never managed to project. `recordRunOutcome('requires_action')`
+  // moves the row off `started`, which is what frees its global capacity slot; the job
+  // itself stays live awaiting approval, so its evidence is NOT released here. Without
+  // this the row held a slot forever wherever no engine is armed, since a paused job is
+  // not terminal and the dead-delivery path never looks at an identity-matched job.
+  if (job.status === 'requires_action') {
+    if (run.status !== 'started') {
+      return;
+    }
+    await deps.methods.recordRunOutcome({
+      scheduleId: run.scheduleId,
+      scheduledFor: run.scheduledFor,
+      status: 'requires_action',
+      conversationId: run.conversationId,
+      autoDisableAfterFailures: Number.MAX_SAFE_INTEGER,
+    });
+    return;
+  }
   const terminal = TERMINAL_JOB_OUTCOMES[job.status];
   if (terminal == null) {
-    // `running`/`requires_action`: the owner is alive and owns the settlement.
+    // `running`: the owner is alive and owns the settlement.
     return;
   }
   // An `aborted` job flips the moment abortJob wins its status CAS — BEFORE its owner
@@ -177,7 +195,7 @@ async function settleFromRetainedJob(
  * EVIDENCE ONLY.
  *
  * `fireSchedule` reserves the run and its global capacity slot before the delivery reaches
- * the chat route, and the clustered entrypoint arms no engine — so two states would
+ * the chat route, and the clustered entrypoint arms no engine — so three states would
  * otherwise hold that slot indefinitely:
  *
  * - A RETAINED TERMINAL JOB. The generation finished, but its owner's Mongo outcome write
@@ -185,6 +203,9 @@ async function settleFromRetainedJob(
  *   job as the only surviving evidence. The armed engine's reconciler replays exactly
  *   this; with no engine nothing did, so the run stayed `started` and its retained
  *   generation evidence lived until store expiry.
+ * - AN UNPROJECTED PAUSE. The generation paused for approval, but the owner's
+ *   `requires_action` projection failed every retry, so the row never left `started` even
+ *   though the pause itself is durable in the job.
  * - A DEAD DELIVERY. A pre-generation rejection (interactive limiter, PII, moderation)
  *   dead-letters the delivery while the run stays `started` with no job at all.
  *
@@ -223,7 +244,7 @@ async function settleStrandedRuns(deps: ScheduleErasureDeps): Promise<void> {
         continue;
       }
       if (jobMatchesRun(job.state, run)) {
-        await settleFromRetainedJob(deps, run, job.state as JobState);
+        await settleFromObservedJob(deps, run, job.state as JobState);
         continue;
       }
       // No job of THIS occurrence's identity, so the durable delivery is the authority.