From 81fc0297c5c316631bb71bcf524f92cab3a2b3d9 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Wed, 29 Jul 2026 21:08:13 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=A7=AD=20fix:=20Codex=20round=207=20?= =?UTF-8?q?=E2=80=94=20settle=20the=20acknowledgement=20semantics=20(Optio?= =?UTF-8?q?n=20A)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 7's second finding is the incoherence I flagged on the PR: the route persisted `preempt: true` on the durable queue item while returning `preempt: false` when delivery could not be confirmed. Those two then disagreed, and `rearmQueuedPreempts` trusts the DURABLE one — so a resumed owner would honour an interrupt the client had explicitly been told degraded to ordinary steering. Rather than patch the disagreement, this settles the meaning: `preempt` in the 202 means "queued as an interrupt request", NOT "a seal is guaranteed". It mirrors `SteerQueueItem.preempt` exactly, so the response, the durable record, and the resume-time re-arm can never disagree. The gates that ARE knowable stay — the owner's recorded capability and a successful enqueue. Everything past that degrades to the documented fallback of injecting at the next tool boundary. A route cannot synchronously know whether another replica will seal: proving it needs a correlated request/response over pub-sub, and even that only proves the owner heard, not that it is still streaming when the arm lands. Four rounds of tightening this boolean each surfaced a narrower case; the sequence does not converge, so the invariant is now "the flag describes the durable decision" and an unconfirmed arm logs a warning instead of rewriting the answer. Also from this round: a failed disarm publish is retried once and its outcome reported. `handleSteerCancel` keeps `removed: true` — the steer really did leave the queue, and saying otherwise would make the client re-show a chip for a steer that can never arrive — and adds `disarmed: false` so the residual risk is visible rather than swallowed. Damage stays bounded regardless: the empty-boundary self-clear disarms the generation after a single seal. Tests: +1 pinning the response/durable-flag invariant. 159 packages/api specs green. --- .../agents/steering/__tests__/request.spec.ts | 22 +++++++ packages/api/src/agents/steering/request.ts | 58 ++++++++++++++----- .../api/src/stream/GenerationJobManager.ts | 46 +++++++++------ 3 files changed, 95 insertions(+), 31 deletions(-) diff --git a/packages/api/src/agents/steering/__tests__/request.spec.ts b/packages/api/src/agents/steering/__tests__/request.spec.ts index c075ec172d..970c0f4f4d 100644 --- a/packages/api/src/agents/steering/__tests__/request.spec.ts +++ b/packages/api/src/agents/steering/__tests__/request.spec.ts @@ -564,6 +564,28 @@ describe('preempt flag on the steer request', () => { * Cancel is live UI. Without this the request stays armed after its steer * is gone and seals an unrelated stretch of generation. */ + /** + * Option A semantics: the 202's `preempt` mirrors the DURABLE queue flag, + * so the response and `SteerQueueItem.preempt` can never disagree — a + * resumed owner re-arming from the queue then honours exactly what the + * client was told. + */ + it('the 202 flag and the durable queue item always agree', async () => { + const streamId = 'preempt-flag-agrees'; + await createCapableJob(streamId); + + const result = await handleSteerRequest(user, { + conversationId: streamId, + text: 'interrupt me', + preempt: true, + }); + const queued = (await GenerationJobManager.steering.peek(streamId))[0]; + + expect(result.body.preempt).toBe(true); + expect(queued.preempt).toBe(true); + expect(result.body.preempt).toBe(queued.preempt === true); + }); + it('cancelling a preempt steer disarms the request', async () => { const streamId = 'preempt-req-cancel'; await createCapableJob(streamId); diff --git a/packages/api/src/agents/steering/request.ts b/packages/api/src/agents/steering/request.ts index 01d2cfcedb..929db8010c 100644 --- a/packages/api/src/agents/steering/request.ts +++ b/packages/api/src/agents/steering/request.ts @@ -266,14 +266,32 @@ export async function handleSteerRequest( * Strictly AFTER a successful enqueue: an armed request whose steer never * made the durable queue could seal a generation with nothing to inject. * - * The 202 reports what was actually ARMED, not what was asked for: a - * cross-replica publish that failed or reached nobody leaves the owner - * without a poll, so the steer still injects at the next tool boundary but - * must not be labelled as interrupting. + * `preempt` in the 202 means "queued as an interrupt request", NOT "a seal + * is guaranteed" — it mirrors the durable `item.preempt` exactly. A route + * cannot synchronously know whether another replica will seal: proving that + * needs a correlated request/response over pub-sub, and even then the owner + * may finish before the arm lands. Reporting delivery instead made the + * response disagree with the durable flag, which `rearmQueuedPreempts` + * trusts on resume — the two must agree or a resumed owner honours an + * interrupt the client was told had degraded. + * + * The gates that ARE knowable stay: the owner's recorded capability and a + * successful enqueue. Everything past that degrades to the documented + * fallback of injecting at the next tool boundary. */ - const preemptArmed = preemptCapable - ? await GenerationJobManager.requestPreempt(streamId, item.steerId, owner.createdAt) - : false; + if (preemptCapable) { + const armed = await GenerationJobManager.requestPreempt( + streamId, + item.steerId, + owner.createdAt, + ); + if (!armed) { + logger.warn( + `[handleSteerRequest] Preempt arm not confirmed for ${streamId} steer=${item.steerId}; ` + + 'the steer remains queued and will inject at the next boundary', + ); + } + } /** Fire-and-forget: the persisted steer part references these uploads, so * mark them used (parity with `updateFilesUsage` on normal sends) or the @@ -297,7 +315,7 @@ export async function handleSteerRequest( steerId: item.steerId, position: depth, conversationId, - preempt: preemptArmed, + preempt: preemptCapable, }, }; } @@ -341,11 +359,23 @@ export async function handleSteerCancel( /** A cancelled steer must also disarm any preempt request it carried — * cancel is live UI, and a request left armed would seal an unrelated * stretch of generation, drain nothing, and end the run mid-sentence. */ - if (removed) { - /** Awaited: a dropped disarm leaves the owner armed for a steer that no - * longer exists, which costs one sealed-and-empty boundary — a visibly - * truncated answer, not just a stale label. */ - await GenerationJobManager.noteSteersRemoved(streamId, [body.steerId], job.createdAt); + if (!removed) { + return { status: 200, body: { removed } }; } - return { status: 200, body: { removed } }; + /** + * Awaited: a dropped disarm leaves the owner armed for a steer that no + * longer exists, costing one sealed-and-empty boundary — a visibly + * truncated answer, not just a stale label. + * + * `removed` stays true regardless: the steer really did leave the queue, + * and reporting otherwise would make the client re-show a chip for a steer + * that can never arrive. `disarmed: false` surfaces the residual risk + * without lying about the removal. + */ + const disarmed = await GenerationJobManager.noteSteersRemoved( + streamId, + [body.steerId], + job.createdAt, + ); + return { status: 200, body: { removed, ...(disarmed === false && { disarmed: false }) } }; } diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index 237415e731..e38c7668bf 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -3288,9 +3288,9 @@ class GenerationJobManagerClass { * generating replica; without a fence identity the publish is skipped and * the empty-boundary path's self-clear bounds the damage to one seal. */ - noteSteersRemoved(streamId: string, steerIds: string[], jobCreatedAt?: number): Promise { + noteSteersRemoved(streamId: string, steerIds: string[], jobCreatedAt?: number): Promise { if (steerIds.length === 0) { - return Promise.resolve(); + return Promise.resolve(true); } const runtime = this.runtimeState.get(streamId); if (runtime != null && (jobCreatedAt == null || runtime.createdAt === jobCreatedAt)) { @@ -3298,26 +3298,38 @@ class GenerationJobManagerClass { } const createdAt = jobCreatedAt ?? runtime?.createdAt; if (createdAt == null || this.eventTransport.emitPreempt == null) { - return Promise.resolve(); + /** Nothing to publish: the local disarm above is the whole mechanism. */ + return Promise.resolve(true); } /** - * Awaitable so a CANCEL can surface a failed disarm. A dropped clear is + * Awaitable so a CANCEL can react to a failed disarm. A dropped clear is * worse than a dropped arm: the owner keeps a level-triggered request for * a steer that no longer exists, seals its next chunk, drains nothing, - * and truncates an unrelated answer. The empty-boundary self-clear bounds - * that to a single seal, but the truncation still happened. + * and truncates an unrelated answer. Retried once — a transient publish + * error is the common case and the retry is cheap — then reported to the + * caller so it is not silently swallowed. + * + * Damage is bounded even if both attempts fail: the empty-boundary + * self-clear disarms the generation after that single seal. */ - return Promise.resolve( - this.eventTransport.emitPreempt(streamId, { op: 'clear', createdAt, steerIds }), - ).then( - () => undefined, - (error: unknown) => { - logger.error( - `[GenerationJobManager] Failed to publish preempt clear for ${streamId}; ` + - 'the owner may seal once before its empty boundary self-clears:', - error, - ); - }, + const publish = (): Promise => + Promise.resolve( + this.eventTransport.emitPreempt?.(streamId, { op: 'clear', createdAt, steerIds }), + ); + return publish().then( + () => true, + () => + publish().then( + () => true, + (error: unknown) => { + logger.error( + `[GenerationJobManager] Failed to publish preempt clear for ${streamId} after retry; ` + + 'the owner may seal once before its empty boundary self-clears:', + error, + ); + return false; + }, + ), ); }