fix: Codex round — lease-takeover rollback, manual snapshot, resume overlap, retain-first create, metadata-before-live-check

Address all 5 findings from Codex review of 7270ea593:

- Don't delete reserved runs after a lease takeover (P2): if a fire stalls until
  its lease expires and another worker re-claims the occurrence and advances past
  it, deleting the reserved 'started' row would erase the only evidence. Added
  holdsClaim (same token + unexpired lease) and gate both rollback deletes
  (capacity + superseded) on it — a loser leaves the row for the reconciler.

- Re-read the schedule after taking the manual lease (P2): acquireManualRunLease
  now returns the FRESH post-image row, so run-now fires the current snapshot (with
  the new claim token) rather than a stale pre-edit prompt/agent that a merged token
  would let revalidateClaim wave through.

- Return overlap when the resume promotion loses (P2): reserveScheduledResume now
  returns 'overlap' (not 'ok') when promoteRunToStarted loses to a newer occurrence,
  so the resume defers with the approval claimable instead of running a second
  concurrent occurrence with the row still requires_action.

- Retain attachments before creating schedules (P2): create now fail-fasts on an
  obvious over-limit and retains files BEFORE createScheduleWithSlot, so a persisted
  claimable schedule never references TTL-expiring uploads (matches the update path).

- Fence deletes after schedule metadata is attached (P2): request.js writes the job's
  scheduleId/scheduledFor BEFORE the isScheduleLive re-check, so a concurrent quiesce
  can identity-match and abort this job; the re-check then catches a delete that
  landed in the tiny createJob -> metadata window.

api/server/controllers/agents/request.js

@@ -393,39 +393,37 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
     const job = await GenerationJobManager.createJob(streamId, userId, conversationId);
     const jobCreatedAt = job.createdAt; // Capture creation time to detect job replacement
 
-    // Re-fence a scheduled fire against a delete/quiesce that landed in the
-    // claim -> POST window: the reservation row existed but this job did not yet,
-    // so the deletion's identity-guarded abort could not have seen it. If the
-    // schedule is no longer live, terminalize the run and tear the job down BEFORE
-    // any messages are persisted, so a delete / account-deletion cascade can't be
-    // defeated by data this request would otherwise write. Scoped to scheduled
-    // fires (scheduleId is null for interactive chat), so this never affects it.
-    if (scheduleId && !(await isScheduleLive(scheduleId))) {
-      logger.info(
-        `[AgentController] Scheduled fire aborted before start; schedule ${scheduleId} no longer active`,
-      );
-      await recordScheduleOutcome({
-        scheduleId,
-        scheduledFor,
-        status: 'interrupted',
-        conversationId: streamId,
-      });
-      await GenerationJobManager.completeJob(streamId).catch(() => undefined);
-      return res.json({ streamId, conversationId, status: 'aborted' });
+    // Re-fence a scheduled fire against a delete/quiesce that landed in the claim ->
+    // POST window (the reservation row existed but this job did not yet, so the
+    // deletion's identity-guarded abort could not have seen it). Attach the schedule
+    // identity to the job FIRST — so a concurrent quiesce can now identity-match and
+    // abort THIS job — then re-check liveness: if a delete landed in the tiny
+    // createJob -> metadata window (before its abort could match), terminalize the
+    // run and tear the job down BEFORE any messages are persisted. Scoped to
+    // scheduled fires (scheduleId is null for interactive chat), so this never
+    // affects it.
+    if (scheduleId) {
+      await GenerationJobManager.updateMetadata(streamId, { scheduleId, scheduledFor });
+      if (!(await isScheduleLive(scheduleId))) {
+        logger.info(
+          `[AgentController] Scheduled fire aborted before start; schedule ${scheduleId} no longer active`,
+        );
+        await recordScheduleOutcome({
+          scheduleId,
+          scheduledFor,
+          status: 'interrupted',
+          conversationId: streamId,
+        });
+        await GenerationJobManager.completeJob(streamId).catch(() => undefined);
+        return res.json({ streamId, conversationId, status: 'aborted' });
+      }
     }
 
     req._resumableStreamId = streamId;
     getMCPRequestContext(req, undefined, { cleanupOnResponse: false });
 
-    // For scheduled fires, persist the schedule identifiers into the job metadata
-    // BEFORE acknowledging the fire below. The loopback caller treats the JSON
-    // `started` response as the run being underway; if these were written only
-    // after and the process crashed in that window, the generation would run with
-    // no scheduleId and no terminal hook could ever record the ScheduleRun outcome
-    // (reconciliation would then treat a successful run as an orphan).
-    if (scheduleId) {
-      await GenerationJobManager.updateMetadata(streamId, { scheduleId, scheduledFor });
-    }
+    // The schedule identifiers were already persisted into the job metadata above
+    // (before the liveness re-check); the bulk updateMetadata below re-affirms them.
 
     // Send JSON response IMMEDIATELY so client can connect to SSE stream
     // This is critical: tool loading (MCP OAuth) may emit events that the client needs to receive

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

@@ -88,6 +88,7 @@ function makeMethods() {
       },
     ),
     revalidateClaim: jest.fn(async () => true),
+    holdsClaim: jest.fn(async () => true),
     deleteScheduleRun: jest.fn(async (id: string, when: Date, _status?: string) => {
       runs.delete(key(id, when));
     }),
@@ -260,6 +261,23 @@ describe('fireSchedule', () => {
     expect(global.fetch).toHaveBeenCalledTimes(1);
   });
 
+  it('preserves the reserved run for reconcile when the lease was taken over', async () => {
+    const { methods, runs } = makeMethods();
+    // At capacity, so this fire will roll back its reservation.
+    for (let i = 0; i < 5; i++) {
+      runs.set(`other-${i}:x`, { status: 'started' });
+    }
+    // Simulate a lease takeover: this worker no longer holds the claim.
+    (methods.holdsClaim as jest.Mock).mockResolvedValue(false);
+    mockFetch(async () => okResponse());
+    const result = await fireSchedule(makeDeps(methods), makeSchedule(), LIMITS, dueAt());
+    expect(result.skipped).toBe('capacity');
+    // The reserved row must NOT be deleted (another worker owns the occurrence now);
+    // it's left for the reconciler so the occurrence stays reconcilable.
+    expect(methods.deleteScheduleRun).not.toHaveBeenCalled();
+    expect([...runs.entries()].some(([k]) => k.startsWith('sched-1:'))).toBe(true);
+  });
+
   it('skips overlap when a prior run is still active', async () => {
     const { methods, runs, calls } = makeMethods();
     runs.set('sched-1:prior', { status: 'started' });

packages/api/src/schedules/fire.ts

@@ -154,6 +154,17 @@ export async function fireSchedule(
       // concurrent owner edit or a lease-expiry re-claim isn't clobbered.
       () => methods.advanceSchedule(schedule.id, nextRunAt, scheduledFor, claimToken);
 
+  // Rolls back a reserved `started` run row — but ONLY if we still hold the claim.
+  // If our lease expired and another worker re-claimed this occurrence (rotating the
+  // token) and advanced past it, deleting the row would erase the only evidence for
+  // an occurrence that is then neither fired nor reconcilable, so leave it for the
+  // reconciler's orphan sweep instead.
+  const rollbackReservation = async () => {
+    if (claimToken != null && (await methods.holdsClaim(schedule.id, claimToken))) {
+      await methods.deleteScheduleRun(schedule.id, scheduledFor, 'started');
+    }
+  };
+
   if (nextRunAt == null) {
     await methods.disableSchedule(schedule.id, 'invalid_schedule', claimToken);
     await advance();
@@ -278,7 +289,7 @@ export async function fireSchedule(
     // request-scoped check). Roll back (status-fenced) if over.
     const active = await deps.countActiveRunsGlobal();
     if (active > ownerLimits.fireConcurrency) {
-      await methods.deleteScheduleRun(schedule.id, scheduledFor, 'started');
+      await rollbackReservation();
       // Automatic claims keep the claim's lease as a backoff so the nextRunAt-sorted
       // claimer doesn't immediately re-pick this row and starve others; nextRunAt is
       // untouched, so the occurrence retries once the lease expires. A manual run-now
@@ -302,7 +313,7 @@ export async function fireSchedule(
       claimToken != null &&
       !(await methods.revalidateClaim(schedule.id, claimToken, !options?.manual))
     ) {
-      await methods.deleteScheduleRun(schedule.id, scheduledFor, 'started');
+      await rollbackReservation();
       await advance();
       return { fired: false, skipped: 'superseded' as const };
     }

packages/api/src/schedules/handlers.ts

@@ -148,6 +148,22 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH
     if (!(await validatePayload(req, res, parsed.data, limits))) {
       return;
     }
+    // Fail fast on an obvious over-limit BEFORE retaining attachments, so the common
+    // case never clears an upload TTL it then can't use. The {user, slot} partial
+    // unique index below is the atomic arbiter for the concurrent-create race.
+    if ((await deps.methods.countSchedulesByUser(user.id)) >= limits.maxPerUser) {
+      res.status(400).json({
+        error: `Schedule limit reached (${limits.maxPerUser}). Delete a schedule to add another.`,
+      });
+      return;
+    }
+    // Retain attachments BEFORE creating, so a persisted (claimable) schedule never
+    // references uploads still eligible for TTL expiry — there is no create-then-
+    // retain window where a crash or a failed rollback leaves the two inconsistent.
+    if (parsed.data.file_ids?.length && !(await retainFiles(parsed.data.file_ids, user.id))) {
+      res.status(500).json({ error: 'Failed to retain schedule attachments' });
+      return;
+    }
     const id = `sched_${randomUUID()}`;
     const nextRunAt = parsed.data.enabled
       ? computeNextRunAt({
@@ -158,7 +174,9 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH
       : undefined;
     // Atomic cap: createScheduleWithSlot claims a free per-user slot via the
     // {user, slot} partial unique index, so concurrent creates can never exceed
-    // maxPerUser (no check-then-insert window). 'limit' means all slots are taken.
+    // maxPerUser. 'limit' means a concurrent racer took the last slot after the
+    // pre-check above; the just-retained files are then unreferenced (a rare, minor
+    // leak of the user's own uploads) — acceptable vs. a partial/expiring commit.
     const created = await deps.methods.createScheduleWithSlot(
       {
         ...parsed.data,
@@ -175,13 +193,6 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH
       });
       return;
     }
-    // Retain attachments; on total failure roll the schedule back so a persisted
-    // schedule never outlives its attachments (which the upload TTL would reap).
-    if (parsed.data.file_ids?.length && !(await retainFiles(parsed.data.file_ids, user.id))) {
-      await deps.methods.deleteScheduleById(id, user.id).catch(() => undefined);
-      res.status(500).json({ error: 'Failed to retain schedule attachments' });
-      return;
-    }
     logger.info(`[schedules] created ${id} for user ${user.id}`);
     res.status(201).json(toWireSchedule(created));
   }

packages/api/src/schedules/service.ts

@@ -370,21 +370,24 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
     schedule: FireableSchedule,
     limits: ScheduleLimits,
   ): Promise<FireResult | null> {
-    const claimToken = await methods.acquireManualRunLease(
+    const leased = await methods.acquireManualRunLease(
       schedule.id,
       schedule.user,
       MANUAL_RUN_LEASE_MS,
     );
-    if (claimToken == null) {
+    if (leased == null) {
       return null;
     }
+    const claimToken = leased.claimToken;
     try {
-      // Carry the fresh claim token so the manual fire's lease release is fenced.
-      return await fireSchedule(engineDeps, { ...schedule, claimToken }, limits, new Date(), {
-        manual: true,
-      });
+      // Fire the FRESH leased row (post-image with the new claim token), not the
+      // snapshot the route read before the lease — an edit that committed in the
+      // window in between is reflected, so a stale prompt/agent is never dispatched.
+      return await fireSchedule(engineDeps, leased, limits, new Date(), { manual: true });
     } catch (err) {
-      await methods.releaseLease(schedule.id, claimToken).catch(() => undefined);
+      if (claimToken != null) {
+        await methods.releaseLease(schedule.id, claimToken).catch(() => undefined);
+      }
       throw err;
     }
   }
@@ -483,15 +486,15 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
     if (active >= limits.fireConcurrency) {
       return 'capacity';
     }
-    // Reserve the single active slot. Best-effort: 'overlap' (a different occurrence
-    // won the slot since the check above) leaves the row paused and the resume runs
-    // undercounted until it settles; 'missing' means a concurrent same-pause resume
-    // already promoted it. Never rolled back.
+    // Reserve the single active slot. If a different occurrence won the slot since
+    // the read-only check above, promoteRunToStarted returns 'overlap' — defer the
+    // resume (approval stays claimable) rather than running a second concurrent
+    // occurrence with the paused row still `requires_action` (which overlap/capacity
+    // accounting would miss). 'missing' means a concurrent same-pause resume already
+    // promoted it — proceed. Never rolled back.
     const promoted = await methods.promoteRunToStarted(scheduleId, new Date(scheduledFor));
     if (promoted === 'overlap') {
-      logger.warn(
-        `[schedules] resumed run could not reserve the active slot (overlap): ${scheduleId}`,
-      );
+      return 'overlap';
     }
     return 'ok';
   }

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

@@ -808,6 +808,16 @@ describe('acquireManualRunLease / releaseLease', () => {
     expect(reacquired).toBeTruthy();
   });
 
+  it('returns the FRESH row (reflecting an edit) with a new claim token', async () => {
+    const schedule = await methods.createSchedule(scheduleData({ name: 'original' }));
+    // An edit commits after a caller read the schedule but before the lease is taken.
+    await methods.updateScheduleById(schedule.id, schedule.user, { name: 'edited' });
+    const leased = await methods.acquireManualRunLease(schedule.id, schedule.user, 60_000);
+    // The lease returns the post-image, so a manual fire uses the edited snapshot.
+    expect(leased?.name).toBe('edited');
+    expect(leased?.claimToken).toBeTruthy();
+  });
+
   it('blocks against a held engine lease and rejects a non-owner', async () => {
     const schedule = await methods.createSchedule(scheduleData());
     const claimed = await methods.claimDueSchedule({ instanceId: 'engine-1', leaseMs: 60_000 });

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

@@ -93,9 +93,10 @@ export type ScheduleMethods = {
     id: string,
     userId: string | Types.ObjectId,
     leaseMs: number,
-  ) => Promise<string | null>;
+  ) => Promise<ISchedule | null>;
   releaseLease: (id: string, expectedClaimToken?: string) => Promise<void>;
   revalidateClaim: (id: string, claimToken: string, requireEnabled?: boolean) => Promise<boolean>;
+  holdsClaim: (id: string, claimToken: string) => Promise<boolean>;
   advanceSchedule: (
     id: string,
     nextRunAt: Date | null,
@@ -314,19 +315,21 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
   /**
    * Takes the schedule's lease for a manual run-now, serializing concurrent
    * `POST /:id/run` requests (and blocking against an engine claim) so a
-   * double-click can't start two runs. Owner-scoped. Returns the fresh
-   * `claimToken` (to carry through the fire), or null if already leased.
+   * double-click can't start two runs. Owner-scoped. Returns the FRESH schedule row
+   * (post-image, with the new claim token) so the caller fires the current snapshot
+   * — an edit that committed after the route read the schedule but before this lease
+   * is reflected here, not the stale pre-edit prompt/agent. Null if already leased.
    */
   async function acquireManualRunLease(
     id: string,
     userId: string | Types.ObjectId,
     leaseMs: number,
-  ): Promise<string | null> {
+  ): Promise<ISchedule | null> {
     // Compare/expire the lease against Mongo's `$$NOW` (same CAS shape as
     // claimDueSchedule), not this worker's clock: a skewed replica must not read a
     // Mongo-written automatic-fire lease as expired early and start a second run.
     const claimToken = randomUUID();
-    const row = await Schedule()
+    return Schedule()
       .findOneAndUpdate(
         {
           id,
@@ -338,7 +341,6 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
         { new: true },
       )
       .lean<ISchedule>();
-    return row != null ? claimToken : null;
   }
 
   /**
@@ -381,6 +383,26 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
     return row != null;
   }
 
+  /**
+   * Whether this worker STILL holds the lease it claimed (same claim token, lease
+   * unexpired) — regardless of enabled/deleting state. Used to fence a rollback
+   * delete of a reserved run row against a lease TAKEOVER: if the lease expired and
+   * another worker re-claimed the occurrence (rotating the token) and advanced past
+   * it, deleting the reserved row would erase the only evidence, so the loser must
+   * leave the row for the reconciler instead.
+   */
+  async function holdsClaim(id: string, claimToken: string): Promise<boolean> {
+    const row = await Schedule()
+      .findOne({
+        id,
+        claimToken,
+        $expr: { $gt: [{ $ifNull: ['$leaseUntil', new Date(0)] }, '$$NOW'] },
+      })
+      .select('_id')
+      .lean();
+    return row != null;
+  }
+
   /**
    * Advances past a fired (or skipped) occurrence and releases the lease. When
    * `expectedNextRunAt` is given, the update is predicated on the schedule still
@@ -870,6 +892,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
     acquireManualRunLease,
     releaseLease,
     revalidateClaim,
+    holdsClaim,
     advanceSchedule,
     disableSchedule,
     insertScheduleRun,