fix: round five: pause awaits title billing, delete re-drive, re-enable attachment recheck, replay firedAt

- HITL pause records only after the full persistence barrier (disconnect
  partial, user-message save, AND the immediate-mode title billing), so a
  deletion drain that cascades off the recorded pause cannot race title
  usage writes back into a deleted account.
- The list-path retry re-drives the WHOLE deletion (abort, settle, erase)
  via the service instead of only probing eraseScheduleIfDrained, so a row
  stranded mid-drain with an active run gets its abort re-driven too.
- Re-enabling a schedule with stored attachments revalidates ownership of
  the effective file list and renews the bounded upload hold, instead of
  silently firing without files whose hold expired while disabled.
- Bookkeeping replays (finalizeBookkeeping + recordSkippedRun same-skip
  retry) project the run row's ORIGINAL firedAt instead of the replay time.

api/server/controllers/agents/request.js

@@ -977,13 +977,14 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
         // in-flight user-message / conversation save, then tear down WITHOUT saving a
         // partial response, emitting a terminal event, or completing the job.
         if (client?.pendingApproval) {
-          // A disconnect-partial save launched while this segment streamed must land
-          // before the pause is recorded: the run's pause hand-off (and any deletion
-          // drain that settles a paused run) treats the recorded pause as "all of this
-          // segment's writes are durable". Never-rejecting by construction.
-          if (partialSavePromise) {
-            await partialSavePromise;
-          }
+          // Every write launched during this segment must land before the pause is
+          // recorded: a deletion drain treats a recorded pause as settleable and can
+          // cascade immediately after. That covers the disconnect-partial save, the
+          // background user-message save, AND the immediate-mode title — a title is
+          // billed work (balance upsert + transaction insert), and its aborted task
+          // unwinding through usage persistence after the cascade would recreate rows
+          // for a deleted account. All never-rejecting by construction.
+          await awaitPendingPersistence();
           if (response?.databasePromise) {
             try {
               await response.databasePromise;

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

@@ -538,12 +538,13 @@ describe('create idempotency', () => {
   });
 });
 
-describe('deferred erase retry', () => {
-  it('re-drives the erase of the caller’s soft-deleted schedules on list', async () => {
-    const deps = makeCreateDeps();
+describe('deferred deletion retry', () => {
+  it('re-drives the full deletion of the caller’s soft-deleted schedules on list', async () => {
+    const deps = makeCreateDeps({
+      deleteSchedule: jest.fn(async () => 'deleted'),
+    } as Partial<SchedulesHandlersDeps>);
     (deps.methods.getSchedulesByUser as jest.Mock) = jest.fn(async () => []);
     (deps.methods.getDeletingScheduleIds as jest.Mock) = jest.fn(async () => ['stranded-1']);
-    (deps.methods.eraseScheduleIfDrained as jest.Mock) = jest.fn(async () => true);
     const { res } = makeRes();
 
     await createSchedulesHandlers(deps).listSchedules(
@@ -553,7 +554,28 @@ describe('deferred erase retry', () => {
     // Fire-and-forget, so let the microtask chain settle before asserting.
     await new Promise((resolve) => setImmediate(resolve));
 
-    expect(deps.methods.eraseScheduleIfDrained).toHaveBeenCalledWith('stranded-1');
+    // The service delete (abort + settle + erase), not a bare erase probe: a schedule
+    // stranded mid-drain with a still-active run needs the abort re-driven too.
+    expect(deps.deleteSchedule).toHaveBeenCalledWith('stranded-1', 'user-1');
+  });
+
+  it('keeps listing even when a stranded deletion re-drive rejects', async () => {
+    const deps = makeCreateDeps({
+      deleteSchedule: jest.fn(async () => {
+        throw new Error('still draining');
+      }),
+    } as Partial<SchedulesHandlersDeps>);
+    (deps.methods.getSchedulesByUser as jest.Mock) = jest.fn(async () => []);
+    (deps.methods.getDeletingScheduleIds as jest.Mock) = jest.fn(async () => ['stranded-1']);
+    const { res, captured } = makeRes();
+
+    await createSchedulesHandlers(deps).listSchedules(
+      { user: { id: 'user-1' } } as unknown as ServerRequest,
+      res,
+    );
+    await new Promise((resolve) => setImmediate(resolve));
+
+    expect(captured.body).toEqual(expect.objectContaining({ schedules: [] }));
   });
 });
 
@@ -645,6 +667,85 @@ describe('updateSchedule refuses field-less payloads', () => {
   });
 });
 
+describe('updateSchedule re-enable attachment revalidation', () => {
+  const disabledWithFiles = () =>
+    ({
+      id: 'sched-1',
+      enabled: false,
+      agent_id: 'agent-1',
+      cadence: { type: 'daily', hour: 9, minute: 0 },
+      timezone: 'UTC',
+      nextRunAt: new Date('2026-07-31T09:00:00Z'),
+      configRevision: 3,
+      file_ids: ['file-a', 'file-b'],
+    }) as unknown as ISchedule;
+
+  const makeReEnableReq = () =>
+    ({
+      params: { id: 'sched-1' },
+      body: { enabled: true },
+      user: { id: 'user-1', tenantId: 't1', role: 'USER' },
+    }) as unknown as ServerRequest;
+
+  it('refuses re-enabling when a stored attachment is no longer owned', async () => {
+    const deps = makeCreateDeps({
+      isUserDeleting: jest.fn(async () => false),
+      filterOwnedFileIds: jest.fn(async () => ['file-a']),
+    } as Partial<SchedulesHandlersDeps>);
+    (deps.methods.getScheduleById as jest.Mock).mockResolvedValue(disabledWithFiles());
+    const { res, captured } = makeRes();
+
+    await createSchedulesHandlers(deps).updateSchedule(makeReEnableReq(), res);
+
+    // The bounded upload hold only renews while the schedule fires, so a long-
+    // disabled schedule can have lost its uploads; silently firing without them
+    // is worse than telling the owner to replace the attachments.
+    expect(captured.status).toBe(400);
+    expect(deps.methods.updateScheduleById).not.toHaveBeenCalled();
+  });
+
+  it('renews the retention hold on the stored attachments before committing', async () => {
+    const markFilesUsed = jest.fn(async () => undefined);
+    const deps = makeCreateDeps({
+      isUserDeleting: jest.fn(async () => false),
+      markFilesUsed,
+    } as Partial<SchedulesHandlersDeps>);
+    (deps.methods.getScheduleById as jest.Mock).mockResolvedValue(disabledWithFiles());
+    const { res, captured } = makeRes();
+
+    await createSchedulesHandlers(deps).updateSchedule(makeReEnableReq(), res);
+
+    expect(markFilesUsed).toHaveBeenCalledWith(['file-a', 'file-b'], 'user-1');
+    expect(captured.status ?? 200).toBe(200);
+    expect(deps.methods.updateScheduleById).toHaveBeenCalled();
+  });
+
+  it('skips the stored-attachment recheck when the edit replaces file_ids', async () => {
+    const filterOwnedFileIds = jest.fn(async (ids: string[]) => ids);
+    const deps = makeCreateDeps({
+      isUserDeleting: jest.fn(async () => false),
+      filterOwnedFileIds,
+    } as Partial<SchedulesHandlersDeps>);
+    (deps.methods.getScheduleById as jest.Mock).mockResolvedValue(disabledWithFiles());
+    const { res, captured } = makeRes();
+
+    await createSchedulesHandlers(deps).updateSchedule(
+      {
+        params: { id: 'sched-1' },
+        body: { enabled: true, file_ids: ['file-new'] },
+        user: { id: 'user-1', tenantId: 't1', role: 'USER' },
+      } as unknown as ServerRequest,
+      res,
+    );
+
+    // Supplied file_ids are validated by validatePayload; the stored list is
+    // about to be overwritten, so rechecking it would refuse a valid replacement.
+    expect(captured.status ?? 200).toBe(200);
+    expect(filterOwnedFileIds).toHaveBeenCalledTimes(1);
+    expect(filterOwnedFileIds).toHaveBeenCalledWith(['file-new'], 'user-1');
+  });
+});
+
 describe('late-create compensation with a live manual run', () => {
   it('treats a draining rollback as compensated (the teardown owns it)', async () => {
     const deps = makeCreateDeps();

packages/api/src/schedules/handlers.ts

@@ -279,27 +279,29 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH
   }
 
   /**
-   * Re-drives the deferred erase of the caller's soft-deleted schedules, off the
+   * Re-drives the FULL deletion of the caller's soft-deleted schedules, off the
    * response path.
    *
-   * A `deleting` row is erased by whichever actor observes it drained: the delete
-   * request itself, or the terminal outcome write (erase-on-settle). Both are
-   * best-effort single attempts, and the reconciler that would otherwise retry does not
-   * exist in the clustered entrypoint — so one transient failure, or a lease that
-   * outlived the delete, strands a hidden row holding the user's prompt indefinitely.
-   * The row is hidden from the list, so the owner cannot even retry it themselves.
+   * A `deleting` row is settled and erased by the delete request itself, or by the
+   * terminal outcome write (erase-on-settle). Both are best-effort, and the
+   * reconciler that would otherwise retry does not exist in the clustered
+   * entrypoint. Meanwhile the row is HIDDEN from the owner's list, so once a delete
+   * answers 503-unconfirmed there is no UI/API-list handle left to retry the drain
+   * with — the "please retry" the response asks for has nothing to click.
    *
-   * A read the owner performs anyway is the cheapest place to retry: bounded, scoped to
-   * their own rows, and a no-op when nothing is deleting (the erase re-checks drained-ness
-   * itself, so this can never race a live run).
+   * A read the owner performs anyway is therefore the re-driver: bounded, scoped to
+   * their own rows, a no-op when nothing is deleting. Re-driving the WHOLE delete
+   * (abort, settle, erase) rather than only the erase is what un-strands a row whose
+   * active run never settled; every step is idempotent and evidence-guarded, so
+   * repeated polls race harmlessly.
    */
-  function retryDeferredErases(userId: string): void {
+  function retryDeferredDeletions(userId: string): void {
     void deps.methods
       .getDeletingScheduleIds(userId, DEFERRED_ERASE_RETRY_LIMIT)
       .then((ids) =>
-        Promise.all(ids.map((id) => deps.methods.eraseScheduleIfDrained(id).catch(() => false))),
+        Promise.all(ids.map((id) => deps.deleteSchedule(id, userId).catch(() => 'unconfirmed'))),
       )
-      .catch((err) => logger.warn('[schedules] deferred erase retry failed', err));
+      .catch((err) => logger.warn('[schedules] deferred deletion retry failed', err));
   }
 
   async function listSchedules(req: ServerRequest, res: Response): Promise<void> {
@@ -307,7 +309,7 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH
       deps.methods.getSchedulesByUser(requestUser(req).id),
       deps.getLimits(requestUser(req)),
     ]);
-    retryDeferredErases(requestUser(req).id);
+    retryDeferredDeletions(requestUser(req).id);
     res.json({
       schedules: schedules.map(toWireSchedule),
       limits: { maxPerUser: limits.maxPerUser },
@@ -653,6 +655,26 @@ export function createSchedulesHandlers(deps: SchedulesHandlersDeps): SchedulesH
       res.status(500).json({ error: 'Failed to retain schedule attachments' });
       return;
     }
+    // Re-enabling with STORED attachments: the bounded upload hold only renews while
+    // the schedule fires, so a schedule that sat disabled past the hold can have lost
+    // its uploads. Validate the effective list and renew its hold now, mirroring the
+    // stored-agent recheck above — otherwise the re-enable succeeds and the next run
+    // silently fires without the missing files instead of telling the user to
+    // replace them.
+    if (reEnabled && parsed.data.file_ids == null && existing.file_ids?.length) {
+      const stillOwned = await deps.filterOwnedFileIds(existing.file_ids, user.id);
+      if (stillOwned.length !== existing.file_ids.length) {
+        res.status(400).json({
+          error:
+            'One or more attached files are no longer available. Replace the attachments before re-enabling.',
+        });
+        return;
+      }
+      if (!(await retainFiles(existing.file_ids, user.id))) {
+        res.status(500).json({ error: 'Failed to retain schedule attachments' });
+        return;
+      }
+    }
     // FENCED on the revision this edit was computed from. `nextRunAt` above is derived
     // from (cadence, timezone) resolved against the row read at the top of this handler,
     // so two overlapping edits — one changing cadence, one changing timezone — would

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

@@ -748,6 +748,28 @@ describe('recordRunOutcome idempotency + crash-retry (bookkeeping)', () => {
     // No longer surfaced as needing bookkeeping.
     expect(await methods.getUnbookkeptRuns(new Date(Date.now() + 1000), 100)).toHaveLength(0);
   });
+
+  it('replayed bookkeeping projects the ORIGINAL firedAt, not the replay time', async () => {
+    const schedule = await methods.createSchedule(scheduleData());
+    const originalFiredAt = new Date('2026-07-20T12:00:03Z');
+    await methods.insertScheduleRun(runData(schedule, { scheduledFor, firedAt: originalFiredAt }));
+    await ScheduleRun.updateOne(
+      { scheduleId: schedule.id, scheduledFor },
+      { $set: { status: 'success', bookkept: false } },
+    );
+
+    // The reconciler replays minutes (or several retries) later — lastRun must
+    // still show when the run actually fired, not when recovery caught up.
+    await methods.finalizeBookkeeping({
+      scheduleId: schedule.id,
+      scheduledFor,
+      status: 'success',
+      autoDisableAfterFailures: 3,
+    });
+
+    const updated = await getSchedule(schedule.id);
+    expect(updated.lastRun?.firedAt?.toISOString()).toBe(originalFiredAt.toISOString());
+  });
 });
 
 describe('bookkeeping policy is crash-retryable / streak-correct', () => {

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

@@ -1201,15 +1201,18 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
     // re-running bookkeeping, so nothing else would ever repair a half-applied skip.
     const inserted = await insertScheduleRun({ ...data, firedAt, bookkept: false });
     let rowRevision = inserted?.configRevision;
+    let rowFiredAt = firedAt;
     if (inserted == null) {
       const existing = await ScheduleRun()
         .findOne({ scheduleId: data.scheduleId, scheduledFor: data.scheduledFor })
-        .select('status configRevision')
-        .lean<Pick<IScheduleRun, 'status' | 'configRevision'>>();
+        .select('status configRevision firedAt')
+        .lean<Pick<IScheduleRun, 'status' | 'configRevision' | 'firedAt'>>();
       if (existing == null || existing.status !== data.status) {
         return;
       }
       rowRevision = existing.configRevision;
+      // A retry re-projects the ORIGINAL occurrence, not the retry moment.
+      rowFiredAt = existing.firedAt ?? firedAt;
     }
     // Same single seam as recordRunOutcome: the config fence is DERIVED from the row,
     // never passed by the caller. A skip decided under an older owner config must not
@@ -1219,7 +1222,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
       await applyBalanceSkipBookkeeping({
         scheduleId: data.scheduleId,
         scheduledFor: data.scheduledFor,
-        firedAt,
+        firedAt: rowFiredAt,
         rowRevision,
         balanceSkipDisableThreshold,
       });
@@ -1227,7 +1230,7 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
       await applyOverlapSkipBookkeeping({
         scheduleId: data.scheduleId,
         scheduledFor: data.scheduledFor,
-        firedAt,
+        firedAt: rowFiredAt,
         rowRevision,
       });
     }
@@ -1319,15 +1322,19 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
   async function finalizeBookkeeping(params: RecordRunOutcomeParams): Promise<void> {
     // Same single seam as recordRunOutcome: derive the config fence from the row, so the
     // crash-retry path cannot apply bookkeeping the inline path would have refused.
+    // The row's ORIGINAL firedAt is reused too: this replay runs minutes (or many
+    // retries) after the fact, and stamping the replay time onto lastRun would keep
+    // walking the card's timestamp forward every time the recovery path fires.
     const run = await ScheduleRun()
       .findOne({ scheduleId: params.scheduleId, scheduledFor: params.scheduledFor })
-      .select('configRevision')
-      .lean<Pick<IScheduleRun, 'configRevision'>>();
+      .select('configRevision firedAt')
+      .lean<Pick<IScheduleRun, 'configRevision' | 'firedAt'>>();
+    const firedAt = run?.firedAt ?? new Date();
     if (params.status === 'skipped_balance') {
       await applyBalanceSkipBookkeeping({
         scheduleId: params.scheduleId,
         scheduledFor: params.scheduledFor,
-        firedAt: new Date(),
+        firedAt,
         conversationId: params.conversationId,
         rowRevision: run?.configRevision,
         balanceSkipDisableThreshold: params.balanceSkipDisableThreshold,
@@ -1336,13 +1343,13 @@ export function createScheduleMethods(mongoose: typeof import('mongoose')): Sche
       await applyOverlapSkipBookkeeping({
         scheduleId: params.scheduleId,
         scheduledFor: params.scheduledFor,
-        firedAt: new Date(),
+        firedAt,
         rowRevision: run?.configRevision,
       });
     } else {
       await applyTerminalBookkeeping({
         ...params,
-        firedAt: new Date(),
+        firedAt,
         expectConfigRevision: run?.configRevision,
       });
     }