From f735341050f77c323833579e73cbfba472cb230c Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 19 Aug 2026 01:59:06 +0000 Subject: [PATCH] fix(schedules): converge unprojected pauses and release replayed bookkeeping jobs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01CZwZPDBgxy4mCWWg6zQSjq --- api/server/controllers/agents/request.js | 12 ++++- packages/api/src/schedules/engine.spec.ts | 54 ++++++++++++++++++++++ packages/api/src/schedules/engine.ts | 19 ++++++++ packages/api/src/schedules/erasure.spec.ts | 39 ++++++++++++++++ packages/api/src/schedules/erasure.ts | 33 ++++++++++--- 5 files changed, 150 insertions(+), 7 deletions(-) diff --git a/api/server/controllers/agents/request.js b/api/server/controllers/agents/request.js index 97e6b840ee..3b9c084fda 100644 --- a/api/server/controllers/agents/request.js +++ b/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`, diff --git a/packages/api/src/schedules/engine.spec.ts b/packages/api/src/schedules/engine.spec.ts index 373c00bc41..4e1a582712 100644 --- a/packages/api/src/schedules/engine.spec.ts +++ b/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', () => { diff --git a/packages/api/src/schedules/engine.ts b/packages/api/src/schedules/engine.ts index 0723883ac2..92a884447a 100644 --- a/packages/api/src/schedules/engine.ts +++ b/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}:`, diff --git a/packages/api/src/schedules/erasure.spec.ts b/packages/api/src/schedules/erasure.spec.ts index d6f0d99fb2..064c45df1a 100644 --- a/packages/api/src/schedules/erasure.spec.ts +++ b/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, + 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, diff --git a/packages/api/src/schedules/erasure.ts b/packages/api/src/schedules/erasure.ts index 1f0c357cfe..82ac826db2 100644 --- a/packages/api/src/schedules/erasure.ts +++ b/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 - * evidence — the clustered mirror of the engine reconciler's retained-job branch. + * Converges a run from its OWN identity-matched job — the clustered mirror of the engine + * reconciler's retained-job and pause branches. */ -async function settleFromRetainedJob( +async function settleFromObservedJob( deps: ScheduleErasureDeps, run: IScheduleRun, job: JobState, ): Promise { + // 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 { 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.