fix: Codex round — owner-edit rollback, resume self-overlap, retention count, evidence on failure

Address all 5 findings from Codex review of 409bae4c0:

- Roll back reservations superseded by owner edits (P2): holdsClaim keyed on the
  claim token skipped the rollback on an owner EDIT (which rotates the token but
  keeps the lease), leaving a ghost 'started' row that consumed capacity/overlap
  until the orphan sweep. Replaced with holdsLease keyed on the lease HOLDER
  (leaseBy) — a takeover changes leaseBy (leave the row), an owner edit keeps it
  (delete this worker's own unposted reservation).

- Keep paused runs resumable when pause bookkeeping fails (P2): reserveScheduledResume
  now uses hasOtherActiveRun (excludes the run's own occurrence), so a paused run whose
  own row is still 'started' (its pause bookkeeping failed transiently) isn't mistaken
  for an overlap with itself and rejected.

- Preserve evidence when early-abort bookkeeping fails (P2): the createJob->liveness
  early abort now preserves the job for reconcile when recordScheduleOutcome's write
  fails, instead of deleting the only evidence while the run stays 'started'.

- Verify attachment retention updated every file (P2): the markFilesUsed wiring now
  checks updateFilesUsage cleared EVERY requested file and throws otherwise, so a
  file deleted/expired mid-request fails the create/update rather than silently
  persisting a schedule whose first fire drops attachments.

- Surface abort failures during account-deletion quiesce (P2): abortScheduledJob /
  abortActiveRun report whether the abort was delivered; quiesceUserSchedules logs a
  prominent per-run warning when a peer-worker generation couldn't be confirmed
  aborted (the unshared-topology limitation, which needs USE_REDIS_STREAMS).

api/server/controllers/agents/request.js

@@ -408,13 +408,20 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
         logger.info(
           `[AgentController] Scheduled fire aborted before start; schedule ${scheduleId} no longer active`,
         );
-        await recordScheduleOutcome({
+        const outcomeRecorded = await recordScheduleOutcome({
           scheduleId,
           scheduledFor,
           status: 'interrupted',
           conversationId: streamId,
         });
-        await GenerationJobManager.completeJob(streamId).catch(() => undefined);
+        // If the outcome write failed (transient Mongo across its retries), preserve
+        // the job so the reconciler can finalize this run instead of deleting the
+        // only evidence while the run row is still `started` (which would leave a
+        // deleted schedule draining until the orphan cutoff). Mirrors the other
+        // scheduled terminal paths.
+        await GenerationJobManager.completeJob(streamId, undefined, {
+          preserveForReconcile: !outcomeRecorded,
+        }).catch(() => undefined);
         return res.json({ streamId, conversationId, status: 'aborted' });
       }
     }

api/server/routes/schedules.js

@@ -43,11 +43,22 @@ const handlers = createSchedulesHandlers({
     return (files ?? []).map((file) => file.file_id);
   },
   markFilesUsed: async (fileIds, userId) => {
-    await methods.updateFilesUsage(
+    // Verify EVERY requested file was actually marked used (TTL cleared). A file can
+    // be deleted/expire between the ownership check and here; updateFilesUsage then
+    // returns fewer docs without throwing, and a silent success would persist a
+    // schedule whose attachments the first fire drops. Throw so retainFiles retries /
+    // fails, rather than committing a schedule with unretained files.
+    const requested = new Set(fileIds).size;
+    // updateFilesUsage dedupes and returns only the docs it actually updated.
+    const updated = await methods.updateFilesUsage(
       fileIds.map((file_id) => ({ file_id })),
       undefined,
       { user: userId },
     );
+    const cleared = Array.isArray(updated) ? updated.length : 0;
+    if (cleared !== requested) {
+      throw new Error(`attachment retention incomplete: ${cleared}/${requested} files marked used`);
+    }
   },
   fireNow: fireScheduleNow,
   // Quiesce-then-erase delete: stops new claims, aborts in-flight loopback runs,

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

@@ -24,6 +24,7 @@ function makeSchedule(overrides: Partial<FireableSchedule> = {}): FireableSchedu
     target: 'new',
     enabled: true,
     claimToken: 'ct-1',
+    leaseBy: 'inst-1',
     runCount: 0,
     failureCount: 0,
     balanceSkipCount: 0,
@@ -88,7 +89,7 @@ function makeMethods() {
       },
     ),
     revalidateClaim: jest.fn(async () => true),
-    holdsClaim: jest.fn(async () => true),
+    holdsLease: jest.fn(async () => true),
     deleteScheduleRun: jest.fn(async (id: string, when: Date, _status?: string) => {
       runs.delete(key(id, when));
     }),
@@ -268,7 +269,7 @@ describe('fireSchedule', () => {
       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);
+    (methods.holdsLease as jest.Mock).mockResolvedValue(false);
     mockFetch(async () => okResponse());
     const result = await fireSchedule(makeDeps(methods), makeSchedule(), LIMITS, dueAt());
     expect(result.skipped).toBe('capacity');

packages/api/src/schedules/fire.ts

@@ -154,13 +154,15 @@ 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.
+  // Rolls back a reserved `started` run row — but ONLY if we still OWN the lease.
+  // Fenced on the lease HOLDER (`leaseBy`), not the claim token: a lease takeover by
+  // another worker changes `leaseBy` (that worker may have advanced past this
+  // occurrence, so deleting the row would erase the only evidence — leave it for the
+  // reconciler); an owner edit only rotates the token and keeps `leaseBy`, so this
+  // worker still owns the lease and must delete its own unposted reservation (else a
+  // ghost `started` row consumes capacity/overlap until the orphan sweep).
   const rollbackReservation = async () => {
-    if (claimToken != null && (await methods.holdsClaim(schedule.id, claimToken))) {
+    if (schedule.leaseBy != null && (await methods.holdsLease(schedule.id, schedule.leaseBy))) {
       await methods.deleteScheduleRun(schedule.id, scheduledFor, 'started');
     }
   };

packages/api/src/schedules/service.ts

@@ -275,21 +275,24 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
     abortScheduledJob: async (conversationId, identity, options) => {
       const store = GenerationJobManager.getJobStore();
       if (store == null) {
-        return;
+        return false;
       }
       const job = await store.getJob(conversationId);
-      // Identity guard: never abort a replacement turn that reused the
-      // conversationId, and never re-terminalize (clobber) an already-settled
-      // job — its evidence must survive for the reconciler.
+      // A null/identity-mismatched job is NOT reachable from this replica: it may be
+      // a live generation on a peer worker's private in-memory store (unshared
+      // topology). Report false so the caller knows the abort was NOT delivered.
       if (job == null || !jobMatchesIdentity(job, identity)) {
-        return;
+        return false;
       }
+      // Already terminal — nothing to abort, and its evidence must survive: treat as
+      // delivered (the run is no longer generating).
       if (job.status !== 'running' && job.status !== 'requires_action') {
-        return;
+        return true;
       }
       await GenerationJobManager.abortJob(conversationId, {
         preserveForReconcile: options?.preserve ?? true,
       });
+      return true;
     },
     clearReconciledJob: async (conversationId, identity) => {
       const store = GenerationJobManager.getJobStore();
@@ -473,9 +476,10 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
     if (schedule == null) {
       return 'gone';
     }
-    // A paused run is `requires_action`, so another `started` run for the schedule is
-    // a DIFFERENT active occurrence — resuming over it would break per-schedule overlap.
-    if (await methods.hasActiveRun(scheduleId)) {
+    // Overlap = a DIFFERENT occurrence is currently `started`. Exclude this run's own
+    // row: if its pause bookkeeping failed transiently the row can still be `started`,
+    // and treating that as an overlap would wrongly reject resuming THIS same occurrence.
+    if (await methods.hasOtherActiveRun(scheduleId, new Date(scheduledFor))) {
       return 'overlap';
     }
     // Read-only capacity gate BEFORE promoting, so we never mutate a row a concurrent
@@ -499,21 +503,26 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
     return 'ok';
   }
 
-  /** Aborts an active run's loopback job (identity-guarded). */
+  /** Aborts an active run's loopback job (identity-guarded). Returns whether the
+   * abort was delivered (false when the job wasn't reachable — e.g. a peer worker's
+   * private store, or a transient error). */
   async function abortActiveRun(
     run: { scheduleId: string; scheduledFor: Date; conversationId?: string },
     preserve: boolean,
-  ): Promise<void> {
+  ): Promise<boolean> {
     if (!run.conversationId) {
-      return;
+      return false;
     }
-    await engineDeps
+    return engineDeps
       .abortScheduledJob(
         run.conversationId,
         { scheduleId: run.scheduleId, scheduledFor: run.scheduledFor },
         { preserve },
       )
-      .catch((err) => logger.warn('[schedules] failed to abort run job on quiesce:', err));
+      .catch((err) => {
+        logger.warn('[schedules] failed to abort run job on quiesce:', err);
+        return false;
+      });
   }
 
   /**
@@ -548,11 +557,26 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
   async function quiesceUserSchedules(userId: string): Promise<void> {
     await methods.disableUserSchedulesForDeletion(userId);
     const active = await methods.getActiveRunsForUser(userId);
+    const unconfirmed: string[] = [];
     for (const run of active) {
       // Do NOT preserve for reconcile: account deletion hard-deletes these run
       // rows, so no reconcile pass will ever finalize/clear a retained job — a
       // preserved job would leak in the store. Let the abort settle it directly.
-      await abortActiveRun(run, false);
+      const aborted = await abortActiveRun(run, false);
+      if (!aborted && run.conversationId) {
+        unconfirmed.push(run.conversationId);
+      }
+    }
+    // Surface any abort that could not be confirmed: in a clustered deployment
+    // without a shared stream store the run's generation may live on a peer worker
+    // and keep persisting messages for the now-deleted account. This is the unshared
+    // topology's known limitation (see the init warning); make it visible per run.
+    if (unconfirmed.length > 0) {
+      logger.warn(
+        `[schedules] account-deletion quiesce could not confirm abort for ${unconfirmed.length} ` +
+          `in-flight scheduled run(s) [${unconfirmed.join(', ')}] — a peer worker's generation may ` +
+          'still persist data. Guaranteed quiescing requires a shared stream store (USE_REDIS_STREAMS).',
+      );
     }
   }
 

packages/api/src/schedules/types.ts

@@ -76,7 +76,7 @@ export interface ScheduleEngineDeps {
     conversationId: string,
     identity: JobIdentity,
     options?: { preserve?: boolean },
-  ) => Promise<void>;
+  ) => Promise<boolean>;
   /**
    * Whether every engine replica can observe the SAME jobs (Redis-backed, or a
    * single process). When false — e.g. clustered workers each with a private

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

@@ -863,6 +863,39 @@ describe('claim-token fencing (stale worker cannot mutate an edited/deleted sche
   });
 });
 
+describe('holdsLease (owner-edit vs lease-takeover discriminator)', () => {
+  it('stays true across an owner edit (token rotates, leaseBy kept) and false on takeover', async () => {
+    const schedule = await methods.createSchedule(scheduleData());
+    const claim = await methods.claimDueSchedule({ instanceId: 'engine-1', leaseMs: 60_000 });
+    expect(claim?.leaseBy).toBe('engine-1');
+    // Owner edit rotates the claim token but does NOT touch leaseBy -> this worker
+    // still owns the lease, so a rollback of its own reservation is safe.
+    await methods.updateScheduleById(schedule.id, schedule.user, { name: 'edited' });
+    expect(await methods.holdsLease(schedule.id, 'engine-1')).toBe(true);
+    // A different worker re-claiming (after expiry) changes leaseBy -> takeover.
+    await methods.releaseLease(schedule.id);
+    await methods.claimDueSchedule({ instanceId: 'engine-2', leaseMs: 60_000 });
+    expect(await methods.holdsLease(schedule.id, 'engine-1')).toBe(false);
+  });
+});
+
+describe('hasOtherActiveRun (excludes the run itself)', () => {
+  it('excludes the checked occurrence and detects a different active one', async () => {
+    const schedule = await methods.createSchedule(scheduleData());
+    const paused = new Date('2026-07-20T12:00:00Z');
+    const active = new Date('2026-07-21T12:00:00Z');
+    // A paused occurrence coexists with a different, active (started) occurrence.
+    await ScheduleRun.create(
+      runData(schedule, { scheduledFor: paused, status: 'requires_action' }),
+    );
+    await methods.reserveStartedRun(runData(schedule, { scheduledFor: active }));
+    // Resuming the paused occurrence: the OTHER started occurrence is an overlap.
+    expect(await methods.hasOtherActiveRun(schedule.id, paused)).toBe(true);
+    // Checking the active occurrence itself: its own started row is excluded.
+    expect(await methods.hasOtherActiveRun(schedule.id, active)).toBe(false);
+  });
+});
+
 describe('deletion quiescing (soft-delete, drain, erase)', () => {
   it('markScheduleDeleting hides + un-claims; erase waits for active runs to drain', async () => {
     const schedule = await methods.createScheduleWithSlot(scheduleData(), 10);

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

@@ -96,7 +96,8 @@ export type ScheduleMethods = {
   ) => 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>;
+  holdsLease: (id: string, leaseBy: string) => Promise<boolean>;
+  hasOtherActiveRun: (scheduleId: string, scheduledFor: Date) => Promise<boolean>;
   advanceSchedule: (
     id: string,
     nextRunAt: Date | null,
@@ -384,18 +385,19 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
   }
 
   /**
-   * 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.
+   * Whether this worker STILL owns the lease it took (same `leaseBy` holder, lease
+   * unexpired). Fences a rollback delete of a reserved run row against a lease
+   * TAKEOVER while NOT skipping it on an owner edit: a takeover changes `leaseBy`
+   * (another worker re-claimed and may have advanced past the occurrence — deleting
+   * would erase the only evidence, so leave it for the reconciler), whereas an owner
+   * edit only rotates `claimToken` and leaves `leaseBy` intact, so the worker still
+   * owns the lease and should delete its own unposted reservation.
    */
-  async function holdsClaim(id: string, claimToken: string): Promise<boolean> {
+  async function holdsLease(id: string, leaseBy: string): Promise<boolean> {
     const row = await Schedule()
       .findOne({
         id,
-        claimToken,
+        leaseBy,
         $expr: { $gt: [{ $ifNull: ['$leaseUntil', new Date(0)] }, '$$NOW'] },
       })
       .select('_id')
@@ -403,6 +405,20 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
     return row != null;
   }
 
+  /**
+   * Whether a DIFFERENT occurrence of the schedule is currently `started`. Used by
+   * the HITL resume overlap check so a paused run whose own row is still `started`
+   * (e.g. its pause bookkeeping failed transiently) is not mistaken for an overlap
+   * with itself — only a truly concurrent occurrence blocks the resume.
+   */
+  async function hasOtherActiveRun(scheduleId: string, scheduledFor: Date): Promise<boolean> {
+    const row = await ScheduleRun()
+      .findOne({ scheduleId, status: 'started', scheduledFor: { $ne: scheduledFor } })
+      .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
@@ -892,7 +908,8 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
     acquireManualRunLease,
     releaseLease,
     revalidateClaim,
-    holdsClaim,
+    holdsLease,
+    hasOtherActiveRun,
     advanceSchedule,
     disableSchedule,
     insertScheduleRun,