🧽 fix: Codex round 9 — spend snapshot arms on nonempty drains too

The round-6 scoping fix only cleared the pre-drain snapshot when the
drain came back EMPTY. On a nonempty drain the `finally` cleared just
the drained ids, so a stale arm — typically a cancel whose
cross-replica clear was lost — survived the boundary. It would then
immediately seal the continuation meant to answer the steer that had
just been injected, and land on an empty boundary as
`preempt_incomplete`: the interrupt appears to work, and the answer to
it is truncated.

A boundary that runs has spent its seal, so everything armed at
snapshot time is spent whether or not it came back from the drain. The
`finally` now clears the union of the snapshot and the drained ids.
Arms that land AFTER the snapshot are still spared — their queue items
are live and uninjected, which is the property round 6 added.

Also fixes an api-workspace CI failure of mine: `resume.spec.js` stubs
`GenerationJobManager` wholesale, and the round-3/4 resume work added
two calls (`updateMetadata`, `rearmQueuedPreempts`) the stub did not
define, so 34 specs threw. Stub extended.

Tests: +2 (a nonempty drain clears a stale snapshot arm; a nonempty
drain spares an arm that landed mid-drain). Counterfactually verified —
the stale-arm spec fails against the unfixed drain. 132 packages/api
specs, 60 resume specs green.

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

@@ -50,6 +50,10 @@ const mockGenerationJobManager = {
   getJobStore: jest.fn(() => mockJobStore),
   getResumeState: jest.fn(),
   setContentParts: jest.fn(),
+  /** Resume moves ownership: the new owner records its own seal capability
+   *  and rebuilds armed interrupts from the durable queue. */
+  updateMetadata: jest.fn().mockResolvedValue(undefined),
+  rearmQueuedPreempts: jest.fn().mockResolvedValue(0),
   emitChunk: jest.fn(),
   emitDone: jest.fn(),
   emitError: jest.fn(),

packages/api/src/agents/steering/__tests__/runtime.spec.ts

@@ -409,6 +409,64 @@ describe('createSteerPreemptBoundaryHook', () => {
     expect(GenerationJobManager.isPreemptRequested(streamId)).toBe(false);
   });
 
+  /**
+   * A cancel whose cross-replica clear was lost leaves a stale arm. If the
+   * next boundary drains a DIFFERENT steer, clearing only the drained id
+   * would leave the stale one level-triggered — it would immediately seal
+   * the continuation meant to answer the steer just injected, landing on an
+   * empty boundary as `preempt_incomplete`.
+   */
+  it('a nonempty drain also clears stale arms held since the snapshot', async () => {
+    const streamId = `preempt-stale-snapshot-${Date.now()}`;
+    const job = await GenerationJobManager.createJob(streamId, 'user-1');
+
+    /** Stale: armed, but its steer never reaches the queue (cancelled). */
+    await GenerationJobManager.requestPreempt(streamId, 'steer-cancelled', job.createdAt);
+    /** Live: queued and armed, and this is what the boundary will drain. */
+    await GenerationJobManager.steering.enqueue(streamId, {
+      ...buildSteer('steer-live', 'interrupt me'),
+      preempt: true,
+    });
+    await GenerationJobManager.requestPreempt(streamId, 'steer-live', job.createdAt);
+    expect(GenerationJobManager.isPreemptRequested(streamId)).toBe(true);
+
+    const hook = createSteerPreemptBoundaryHook({
+      streamId,
+      jobCreatedAt: job.createdAt,
+      applySteer: jest.fn(),
+    });
+    const output: SteerDrainOutput = await hook(boundaryInput(), abortSignal);
+
+    expect(output.injectedMessages).toHaveLength(1);
+    /** Both the drained id and the stale snapshot id are spent. */
+    expect(GenerationJobManager.isPreemptRequested(streamId)).toBe(false);
+  });
+
+  /** An arm that lands AFTER the snapshot is backed by a live queue item. */
+  it('a nonempty drain spares an arm that landed after the snapshot', async () => {
+    const streamId = `preempt-post-snapshot-${Date.now()}`;
+    const job = await GenerationJobManager.createJob(streamId, 'user-1');
+    await GenerationJobManager.steering.enqueue(streamId, {
+      ...buildSteer('steer-first', 'first'),
+      preempt: true,
+    });
+    await GenerationJobManager.requestPreempt(streamId, 'steer-first', job.createdAt);
+
+    const hook = createSteerPreemptBoundaryHook({
+      streamId,
+      jobCreatedAt: job.createdAt,
+      applySteer: async () => {
+        /** Arrives mid-drain, after the snapshot was taken. */
+        await GenerationJobManager.requestPreempt(streamId, 'steer-later', job.createdAt);
+      },
+    });
+    await hook(boundaryInput(), abortSignal);
+
+    expect(GenerationJobManager.getArmedPreemptIds(streamId, job.createdAt)).toEqual([
+      'steer-later',
+    ]);
+  });
+
   it('clears the request even when applySteer throws mid-drain', async () => {
     const streamId = `preempt-clears-on-error-${Date.now()}`;
     const job = await GenerationJobManager.createJob(streamId, 'user-1');

packages/api/src/agents/steering/runtime.ts

@@ -139,14 +139,28 @@ async function drainAndBuildInjections(opts: SteerDrainHookOptions): Promise<Inj
   } catch (error) {
     logger.error(`[steering] Drain interrupted for ${streamId}; injecting applied items:`, error);
   } finally {
-    /** Not awaited: this runs on the OWNER, where the local disarm is
-     *  synchronous and already effective — the publish only informs other
-     *  replicas, and blocking a boundary drain on it would delay injection. */
-    void GenerationJobManager.noteSteersRemoved(
-      streamId,
-      steers.map((item) => item.steerId),
-      jobCreatedAt,
-    );
+    /**
+     * Everything armed at snapshot time PLUS everything just drained. The
+     * boundary has spent its seal, so a snapshotted id that did NOT come back
+     * from the drain is stale — its steer left the queue by another route
+     * (typically a cancel whose cross-replica clear was lost). Clearing only
+     * the drained ids would leave that one level-triggered, sealing the very
+     * continuation meant to answer the steer we just injected and landing on
+     * an empty boundary as `preempt_incomplete`.
+     *
+     * Ids armed AFTER the snapshot are deliberately excluded: their queue
+     * items are live and uninjected, and disarming them would strand an
+     * interrupt the client was already told about.
+     *
+     * Not awaited: this runs on the OWNER, where the local disarm is
+     * synchronous and already effective — the publish only informs other
+     * replicas, and blocking a boundary drain on it would delay injection.
+     */
+    const spent = new Set(armedBeforeDrain);
+    for (const item of steers) {
+      spent.add(item.steerId);
+    }
+    void GenerationJobManager.noteSteersRemoved(streamId, [...spent], jobCreatedAt);
   }
   return injectedMessages;
 }