fix: generation-fence scheduled aborts

Codex review 4778439593, P1. abortScheduledJob checks a job's scheduled identity and
then awaits before calling abortJob, which took only a streamId. An interactive turn
replacing the job at that conversationId in the window received the abort signal, the
terminal event and the cleanup meant for the scheduled run.

abortJob now accepts `expectedCreatedAt` and enforces it against a FRESH read before any
side effect — abort signal, local controller, terminal write — so a caller's stale
decision cannot reach a replacement. A refused abort reports success:false, and
abortScheduledJob now returns that, matching how it already reports an unreachable job:
the caller learns the abort was NOT delivered instead of assuming it landed, so the
account-deletion drain treats the run as unconfirmed rather than proceeding.

abortJob's own terminal delete is now fenced too, derived from the job it read rather
than from the caller. Everything in the abort operates on that snapshot, so the cleanup
should not be able to remove a hash some later turn owns — the same
derive-the-fence-from-the-row-you-observed rule already used for run bookkeeping.

This is the last member of the generation-fence family opened by the deleteJob guard,
the create-path fold and the stamp epoch. What it does NOT close: the abort signal is a
pub/sub emit keyed by streamId, so a replacement landing between the fresh read and the
emit is still theoretically reachable. That window is now microseconds rather than
several awaits, and closing it fully needs a fence inside the transport, which is not a
v1 change.

The mismatch case fails against the pre-fix manager; the matching-fence and no-fence
cases pass on both sides, the latter confirming existing callers are unaffected.

packages/api/src/schedules/service.ts

@@ -354,10 +354,17 @@ export function createSchedulesService(deps: SchedulesServiceDeps): SchedulesSer
         }
         return true;
       }
-      await GenerationJobManager.abortJob(conversationId, {
+      // Carry the stamp of the generation whose identity was just checked. The check
+      // above and the abort's side effects are separated by awaits, so without this an
+      // interactive turn replacing the job at this conversationId in that window would
+      // receive the abort signal, terminal event and cleanup meant for the scheduled
+      // run. A refused abort reports false, matching the unreachable-job case: the
+      // caller learns the abort was not delivered rather than assuming it landed.
+      const aborted = await GenerationJobManager.abortJob(conversationId, {
         preserveForReconcile: options?.preserve ?? true,
+        expectedCreatedAt: job.createdAt,
       });
-      return true;
+      return aborted.success;
     },
     clearReconciledJob: async (conversationId, identity) => {
       const store = GenerationJobManager.getJobStore();

packages/api/src/stream/GenerationJobManager.ts

@@ -866,6 +866,13 @@ class GenerationJobManagerClass {
        * failed can still observe the abort. The reconciler deletes it afterward.
        */
       preserveForReconcile?: boolean;
+      /**
+       * Generation fence. A streamId is reused across generations, so a caller that
+       * decided to abort based on an EARLIER observation (the scheduler checks a job's
+       * scheduled identity, then awaits) must carry the stamp it saw: without it an
+       * interactive turn replacing the job in that window is aborted instead.
+       */
+      expectedCreatedAt?: number;
     },
   ): Promise<AbortResult> {
     const jobData = await this.jobStore.getJob(streamId);
@@ -884,6 +891,22 @@ class GenerationJobManagerClass {
       };
     }
 
+    // Checked against a FRESH read and before any side effect (abort signal, local
+    // controller, terminal write), so a caller's stale decision cannot reach a
+    // replacement generation.
+    if (options?.expectedCreatedAt != null && jobData.createdAt !== options.expectedCreatedAt) {
+      logger.debug(`[GenerationJobManager] Abort skipped (generation mismatch): ${streamId}`);
+      recordGenerationJob(this.storeLabel, 'abort_failed');
+      return {
+        text: '',
+        content: [],
+        jobData: null,
+        success: false,
+        finalEvent: null,
+        collectedUsage: [],
+      };
+    }
+
     // Emit abort signal for cross-replica support (Redis mode)
     // This ensures the generating replica receives the abort signal
     if (this.eventTransport.emitAbort) {
@@ -992,7 +1015,10 @@ class GenerationJobManagerClass {
     if (this._cleanupOnComplete && !options?.preserveForReconcile) {
       this.runtimeState.delete(streamId);
       // Don't cleanup eventTransport here - let the abort event fully transmit first.
-      await this.jobStore.deleteJob(streamId);
+      // Fenced on the generation this abort actually OBSERVED, derived here rather than
+      // taken from the caller: everything above operated on `jobData`, so a turn that
+      // replaced the job while the abort was unwinding must not have its hash deleted.
+      await this.jobStore.deleteJob(streamId, jobData.createdAt);
     } else if (options?.preserveForReconcile) {
       // Retain WITHOUT completedAt so the finished-job sweep can't reap it before
       // the schedules reconciler observes the abort; the reconciler deletes it.

packages/api/src/stream/__tests__/abortGenerationFence.spec.ts

@@ -0,0 +1,71 @@
+/**
+ * abortJob must act on the generation its CALLER observed, not on whatever occupies the
+ * streamId by the time the abort runs. The scheduler checks a job's scheduled identity
+ * and then awaits before aborting, so an interactive turn reusing the conversation in
+ * that window would otherwise receive the abort signal, terminal event and cleanup.
+ */
+
+/** Suppress winston Console transport output (survives jest.resetModules) */
+jest.spyOn(console, 'log').mockImplementation();
+
+async function makeManager() {
+  const { GenerationJobManagerClass } = await import('../GenerationJobManager');
+  const { InMemoryJobStore } = await import('../implementations/InMemoryJobStore');
+  const { InMemoryEventTransport } = await import('../implementations/InMemoryEventTransport');
+  const manager = new GenerationJobManagerClass();
+  manager.configure({
+    jobStore: new InMemoryJobStore({ ttlAfterComplete: 0 }),
+    eventTransport: new InMemoryEventTransport(),
+    isRedis: false,
+  });
+  manager.initialize();
+  return manager;
+}
+
+describe('abortJob generation fence', () => {
+  beforeEach(() => {
+    jest.resetModules();
+  });
+
+  it('refuses to abort when the observed generation was replaced', async () => {
+    const manager = await makeManager();
+    const doomed = await manager.createJob('conv-1', 'user-1', 'conv-1');
+    const observedCreatedAt = (await manager.getJobStore().getJob('conv-1'))?.createdAt;
+    // An interactive turn reuses the conversation before the abort lands.
+    await manager.createJob('conv-1', 'user-1', 'conv-1');
+    const replacement = await manager.getJobStore().getJob('conv-1');
+
+    const result = await manager.abortJob('conv-1', { expectedCreatedAt: observedCreatedAt });
+
+    // Reported as NOT delivered, so the scheduler treats the run as unconfirmed rather
+    // than assuming it stopped.
+    expect(result.success).toBe(false);
+    // The replacement keeps running: not signalled, not torn down.
+    expect((await manager.getJobStore().getJob('conv-1'))?.createdAt).toBe(replacement?.createdAt);
+    expect(await manager.hasJob('conv-1')).toBe(true);
+    // The replacement's controller is a different one; the doomed generation's own
+    // controller is irrelevant here — what matters is the live turn was untouched.
+    expect(doomed.abortController.signal.aborted).toBe(false);
+  });
+
+  it('aborts normally when the fence matches the live generation', async () => {
+    const manager = await makeManager();
+    const job = await manager.createJob('conv-1', 'user-1', 'conv-1');
+    const observedCreatedAt = (await manager.getJobStore().getJob('conv-1'))?.createdAt;
+
+    const result = await manager.abortJob('conv-1', { expectedCreatedAt: observedCreatedAt });
+
+    expect(result.success).toBe(true);
+    expect(job.abortController.signal.aborted).toBe(true);
+  });
+
+  it('aborts when no fence is supplied, preserving existing callers', async () => {
+    const manager = await makeManager();
+    const job = await manager.createJob('conv-1', 'user-1', 'conv-1');
+
+    const result = await manager.abortJob('conv-1');
+
+    expect(result.success).toBe(true);
+    expect(job.abortController.signal.aborted).toBe(true);
+  });
+});