fix: Codex round 11 — pause recording, resume/abort preserve, Redis preserve TTL

- request.js: record a scheduled fire's HITL pause as requires_action before
  returning, so overlap/capacity (which key on 'started') stop counting it as
  active immediately instead of after the reconcile min-age delay.
- request.js: the late-abort finalization records the interrupted outcome first
  and only completeJob's when that write succeeded — leaving a preserved aborted
  job intact rather than overwriting it as an error job (which reconcile would
  count toward failure auto-disable).
- resume.js: gate the resume-failure completeJob on the outcome write, preserving
  the job for reconcile when it failed.
- RedisJobStore: a terminal updateJob without completedAt is the preserve signal,
  so retain the job hash on the longer running TTL (not the 300s completed TTL) —
  giving reconcile a window instead of a 5-minute race; it deletes the job after.

api/server/controllers/agents/__tests__/resume.spec.js

@@ -1104,7 +1104,9 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => {
       await flush();
 
       expect(mockGenerationJobManager.emitError).toHaveBeenCalledWith(CONVO_ID, 'boom');
-      expect(mockGenerationJobManager.completeJob).toHaveBeenCalledWith(CONVO_ID, 'boom');
+      expect(mockGenerationJobManager.completeJob).toHaveBeenCalledWith(CONVO_ID, 'boom', {
+        preserveForReconcile: false,
+      });
       expect(mockDeleteAgentCheckpoint).toHaveBeenCalledWith(CONVO_ID, { type: 'mongo' });
       expect(mockDecrementPendingRequest).toHaveBeenCalledWith(USER_ID);
       expect(mockSaveMessage).not.toHaveBeenCalled();

api/server/controllers/agents/request.js

@@ -793,6 +793,18 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
           if (client) {
             disposeClient(client);
           }
+          // Record a scheduled fire's pause as `requires_action` NOW (not on the
+          // next reconcile sweep): overlap/capacity checks key on `started`, so a
+          // run left `started` while paused would wrongly block a run-now or the
+          // next occurrence. A failed write just falls back to reconcile marking it.
+          if (scheduleId) {
+            await recordScheduleOutcome({
+              scheduleId,
+              scheduledFor,
+              status: 'requires_action',
+              conversationId: streamId,
+            });
+          }
           logger.debug(
             `[ResumableAgentController] Turn paused for approval; awaiting resume: ${streamId}`,
           );
@@ -987,17 +999,24 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
           });
 
           await GenerationJobManager.emitDone(streamId, finalEvent);
-          GenerationJobManager.completeJob(streamId, 'Request aborted');
+          // Record the abort BEFORE completeJob so the run doesn't linger as
+          // `started` (blocking run-now/overlap until the 30-minute orphan cutoff).
+          let abortOutcomeRecorded = true;
           if (scheduleId) {
-            // Record the abort so the run doesn't linger as `started` (which would
-            // block run-now/overlap until the 30-minute orphan cutoff).
-            await recordScheduleOutcome({
+            abortOutcomeRecorded = await recordScheduleOutcome({
               scheduleId,
               scheduledFor,
               status: 'interrupted',
               conversationId: conversation?.conversationId,
             });
           }
+          // Only finalize/clean the job when the outcome is recorded. When it isn't
+          // (Mongo down), leave any reconcile evidence the abort route preserved
+          // (an `aborted` job) intact — overwriting it here as an `error` job would
+          // make reconcile count a user stop toward failure auto-disable.
+          if (abortOutcomeRecorded) {
+            GenerationJobManager.completeJob(streamId, 'Request aborted');
+          }
           await finishResumableRequest(req, userId);
         }
 

api/server/controllers/agents/resume.js

@@ -753,9 +753,12 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
       }
       // Record the schedule error now: for a scheduled fire that paused and
       // then failed on resume, this is the terminal point, and the job is about
-      // to be deleted (the reconciler wouldn't see the error).
+      // to be deleted (the reconciler wouldn't see the error). If the write fails
+      // (Mongo down across its retries), preserve the completed job so reconcile
+      // records the failure instead of the run lingering to the abandonment sweep.
+      let scheduleOutcomeRecorded = true;
       if (job.metadata?.scheduleId) {
-        await recordScheduleOutcome({
+        scheduleOutcomeRecorded = await recordScheduleOutcome({
           scheduleId: job.metadata.scheduleId,
           scheduledFor: job.metadata.scheduledFor,
           status: 'error',
@@ -764,7 +767,9 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
         });
       }
       try {
-        await GenerationJobManager.completeJob(streamId, err?.message ?? 'Resume failed');
+        await GenerationJobManager.completeJob(streamId, err?.message ?? 'Resume failed', {
+          preserveForReconcile: Boolean(job.metadata?.scheduleId) && !scheduleOutcomeRecorded,
+        });
       } catch (completeErr) {
         logger.error('[ResumeAgentController] Failed to finalize failed resume', completeErr);
         // Last resort: force a terminal state so the job isn't orphaned in `running`.

packages/api/src/stream/implementations/RedisJobStore.ts

@@ -541,7 +541,10 @@ export class RedisJobStore implements IJobStore {
     }
 
     if (updates.status && ['complete', 'error', 'aborted'].includes(updates.status)) {
-      await this.applyTerminalContentCleanup(streamId);
+      // A terminal write WITHOUT completedAt is the preserveForReconcile signal
+      // (see GenerationJobManager.completeJob/abortJob) — retain the job hash on
+      // the longer running TTL so reconcile can still observe it.
+      await this.applyTerminalContentCleanup(streamId, updates.completedAt == null);
     }
   }
 
@@ -553,14 +556,19 @@ export class RedisJobStore implements IJobStore {
    * the configured after-complete TTLs. Without sharing this, an expired
    * approval left Redis stream contents around for the full running TTL.
    */
-  private async applyTerminalContentCleanup(streamId: string): Promise<void> {
+  private async applyTerminalContentCleanup(streamId: string, preserve = false): Promise<void> {
     const key = KEYS.job(streamId);
+    // A preserved terminal job (kept WITHOUT completedAt for reconciliation) must
+    // outlive the short completed TTL, or the retained evidence expires before the
+    // scheduler's reconcile window; use the longer running TTL. The reconciler
+    // deletes it after finalizing, so this is a bounded upper limit, not a leak.
+    const jobKeyTtl = preserve ? this.ttl.running : this.ttl.completed;
     // Proactively remove from user's job set (requires reading userId from the job hash)
     const job = await this.getJob(streamId);
     const userJobsKey = job?.userId ? KEYS.userJobs(job.userId, job.tenantId) : null;
 
     if (this.isCluster) {
-      await this.redis.expire(key, this.ttl.completed);
+      await this.redis.expire(key, jobKeyTtl);
       await this.redis.srem(KEYS.runningJobs, streamId);
       await this.redis.srem(KEYS.requiresActionJobs, streamId);
       await this.redis.del(KEYS.steers(streamId));
@@ -582,7 +590,7 @@ export class RedisJobStore implements IJobStore {
       }
     } else {
       const pipeline = this.redis.pipeline();
-      pipeline.expire(key, this.ttl.completed);
+      pipeline.expire(key, jobKeyTtl);
       pipeline.srem(KEYS.runningJobs, streamId);
       pipeline.srem(KEYS.requiresActionJobs, streamId);
       pipeline.del(KEYS.steers(streamId));