From d4e6196fb91bd31c45600b1e6a12cb984086a7bf Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sat, 10 Oct 2026 23:09:40 -0700 Subject: [PATCH] fix: verify response waits before provider dispatch Retain the final admission receipt when preparation has started but the provider callback has not run. Conflicting execution evidence still requires recovery. Co-Authored-By: Paperclip --- .../satisfied-connection-intents.test.ts | 20 +++++++++++++++---- .../modules/run-dispatch/adapters/postgres.ts | 5 +++++ .../src/services/pre-dispatch-review-wait.ts | 7 +++++-- 3 files changed, 26 insertions(+), 6 deletions(-) diff --git a/server/src/__tests__/satisfied-connection-intents.test.ts b/server/src/__tests__/satisfied-connection-intents.test.ts index a2c95450d20..437fc91d3f6 100644 --- a/server/src/__tests__/satisfied-connection-intents.test.ts +++ b/server/src/__tests__/satisfied-connection-intents.test.ts @@ -191,19 +191,31 @@ describePostgres("already available connection requests", () => { details: expect.objectContaining({ previousConnectionId: f.connection.id, connectionId: replacement.id }) })])); }); - it("rearms an unstarted wait without manufacturing a workspace repair hold", async () => { + it.each(["queued", "running"] as const)("rearms an unstarted %s wait without manufacturing a workspace repair hold", async (status) => { const f = await seed(); const run = await retry(f); await satisfiedConnectionIntentService(db).sweepPending(); const key = `connection-intent:${f.intent.id}:expired`; const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation", status: "claimed", idempotencyKey: key, runId: run.id }).returning(); await db.update(connectionIntentDeliveries).set({ deliveredAt: new Date() }); - await db.update(heartbeatRuns).set({ status: "queued", wakeupRequestId: wake!.id, resultJson: null, + await db.update(heartbeatRuns).set({ status, startedAt: status === "running" ? new Date() : null, + runtimeModeResolvedAt: status === "running" ? new Date() : null, + controllerBootId: status === "running" ? randomUUID() : null, + executionStage: status === "running" ? "preparing" : null, wakeupRequestId: wake!.id, resultJson: null, contextSnapshot: { issueId: f.issueId, interactionId: f.intent.id, connectionIntentResolution: "existing_connection", interactionKind: "connection_intent", interactionStatus: "expired", mutation: "interaction", wakeReason: "issue_commented", source: "connection_intent.resolved" } }).where(eq(heartbeatRuns.id, run.id)); await db.update(connectionGrants).set({ status: "revoked" }); - expect(await createPostgresRunDispatchAdapter(db).cancelStaleQueuedRun({ companyId: f.companyId, runId: run.id, - expectedStatus: "queued", now: new Date() })).toMatchObject({ outcome: "cancelled", errorCode: "issue_waiting_for_response" }); + const dispatch = createPostgresRunDispatchAdapter(db); + if (status === "queued") { + expect(await dispatch.cancelStaleQueuedRun({ companyId: f.companyId, runId: run.id, + expectedStatus: "queued", now: new Date() })).toMatchObject({ outcome: "cancelled", errorCode: "issue_waiting_for_response" }); + } else { + const invoke = vi.fn(async () => null); + expect(await dispatch.dispatchResolvedInteractionIfCurrent({ companyId: f.companyId, runId: run.id, + expectedStatus: "running", now: new Date(), dispatch: invoke })).toMatchObject({ dispatched: false, + cancellation: { outcome: "cancelled", errorCode: "issue_waiting_for_response" } }); + expect(invoke).not.toHaveBeenCalled(); + } const cancelled = (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]; expect(await legacyExecutionNeedsReconciliationWithEvidence(db, cancelled)).toBe(false); expect((await db.select().from(connectionIntentDeliveries))[0].deliveredAt).toBeNull(); diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 3a8828f8498..947e7d33863 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -882,6 +882,7 @@ export function createPostgresRunDispatchAdapter( decision: Extract, { stale: true }>, expectedStatus: "queued" | "running", now: Date, + providerDispatchNotStarted = false, ): Promise { const [row] = await tx .update(heartbeatRuns) @@ -893,6 +894,9 @@ export function createPostgresRunDispatchAdapter( resultJson: { ...parseObject(run.resultJson), stopReason: decision.errorCode, + ...(providerDispatchNotStarted && decision.errorCode === "issue_waiting_for_response" + ? { preDispatchResponseWait: { version: 1, providerWorkStarted: false } } + : {}), ...(decision.errorCode === "execution_reconciliation_required" ? { executionWait: decision.details } : {}), @@ -1146,6 +1150,7 @@ export function createPostgresRunDispatchAdapter( decision, input.expectedStatus, input.now, + true, // This locked gate has not handed off to the provider callback. ); return { dispatched: false as const, cancellation }; } diff --git a/server/src/services/pre-dispatch-review-wait.ts b/server/src/services/pre-dispatch-review-wait.ts index 2f5413b8f6f..7448d5f8e55 100644 --- a/server/src/services/pre-dispatch-review-wait.ts +++ b/server/src/services/pre-dispatch-review-wait.ts @@ -14,6 +14,9 @@ type DispatchFields = "startedAt" | "runtimeModeResolvedAt" | "processPid" | "pr export function isPreDispatchReviewWait( run: Pick & Partial>, ): boolean { + const finalResponseWait = run.errorCode === "issue_waiting_for_response" && + (run.resultJson?.preDispatchResponseWait as { version?: number; providerWorkStarted?: boolean } | undefined)?.version === 1 && + (run.resultJson?.preDispatchResponseWait as { providerWorkStarted?: boolean } | undefined)?.providerWorkStarted === false; return run.runtimeMode === "legacy" && run.status === "cancelled" && (run.errorCode === "issue_continuation_waiting_on_review" || (run.errorCode === "issue_waiting_for_response" && @@ -26,10 +29,10 @@ export function isPreDispatchReviewWait( run.resultJson?.stopReason === run.errorCode && run.resultJson?.timeoutSource === "stale_queued_run_gate" && run.resultJson?.workspaceRestoreFailure !== "restore_unsafe_archive" && - run.startedAt === null && run.runtimeModeResolvedAt === null && + (finalResponseWait || (run.startedAt === null && run.runtimeModeResolvedAt === null)) && run.processPid === null && run.processGroupId === null && run.processStartedAt === null && run.nativeIssueId === null && run.nativeSessionId === null && run.sessionIdAfter === null && - run.controllerBootId === null && run.controllerLeaseExpiresAt === null && run.executionStage === null; + (finalResponseWait || (run.controllerBootId === null && run.controllerLeaseExpiresAt === null && run.executionStage === null)); } /** The row receipt is a candidate, not permission to discard conflicting