From 7d5341a80757cd7c11ef909512e0d5287f74ad73 Mon Sep 17 00:00:00 2001 From: "t3-code[bot]" <269035359+t3-code[bot]@users.noreply.github.com> Date: Wed, 5 Aug 2026 15:59:46 +0000 Subject: [PATCH 1/2] fix(server): interrupt turns after approval cancellation --- .../Layers/ProviderCommandReactor.test.ts | 124 +++++++++++++++++- .../Layers/ProviderCommandReactor.ts | 9 +- 2 files changed, 127 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 97379b94b88..383d8284b61 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -2606,7 +2606,7 @@ describe("ProviderCommandReactor", () => { }); }); - it("reacts to thread.approval.respond by forwarding provider approval response", async () => { + it("forwards approval declines without interrupting the provider turn", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; @@ -2631,20 +2631,136 @@ describe("ProviderCommandReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.approval.respond", - commandId: CommandId.make("cmd-approval-respond"), + commandId: CommandId.make("cmd-approval-decline"), threadId: ThreadId.make("thread-1"), requestId: asApprovalRequestId("approval-request-1"), - decision: "accept", + decision: "decline", createdAt: now, }), ); await waitFor(() => harness.respondToRequest.mock.calls.length === 1); + await harness.drain(); + expect(harness.respondToRequest.mock.calls[0]?.[0]).toEqual({ + threadId: "thread-1", + requestId: "approval-request-1", + decision: "decline", + }); + expect(harness.interruptTurn).not.toHaveBeenCalled(); + }); + + it("responds to approval cancellation before interrupting the provider turn", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + const callOrder: Array<"respond" | "interrupt"> = []; + harness.respondToRequest.mockImplementation(() => + Effect.sync(() => { + callOrder.push("respond"); + }), + ); + harness.interruptTurn.mockImplementation(() => + Effect.sync(() => { + callOrder.push("interrupt"); + }), + ); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-set-for-approval-cancel"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: asTurnId("turn-1"), + lastError: null, + updatedAt: now, + }, + createdAt: now, + }), + ); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.approval.respond", + commandId: CommandId.make("cmd-approval-cancel"), + threadId: ThreadId.make("thread-1"), + requestId: asApprovalRequestId("approval-request-1"), + decision: "cancel", + createdAt: now, + }), + ); + + await waitFor(() => harness.interruptTurn.mock.calls.length === 1); + await harness.drain(); expect(harness.respondToRequest.mock.calls[0]?.[0]).toEqual({ threadId: "thread-1", requestId: "approval-request-1", - decision: "accept", + decision: "cancel", + }); + expect(harness.interruptTurn.mock.calls[0]?.[0]).toEqual({ + threadId: "thread-1", }); + expect(callOrder).toEqual(["respond", "interrupt"]); + }); + + it("does not interrupt when the approval cancellation response fails", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + harness.respondToRequest.mockImplementation(() => + Effect.fail( + new ProviderAdapterRequestError({ + provider: ProviderDriverKind.make("codex"), + method: "session/request_permission", + detail: "Approval response failed", + }), + ), + ); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-set-for-failed-approval-cancel"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: asTurnId("turn-1"), + lastError: null, + updatedAt: now, + }, + createdAt: now, + }), + ); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.approval.respond", + commandId: CommandId.make("cmd-approval-cancel-failed"), + threadId: ThreadId.make("thread-1"), + requestId: asApprovalRequestId("approval-request-1"), + decision: "cancel", + createdAt: now, + }), + ); + + await waitFor(async () => { + const readModel = await harness.readModel(); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); + return ( + thread?.activities.some( + (activity) => activity.kind === "provider.approval.respond.failed", + ) ?? false + ); + }); + await harness.drain(); + + expect(harness.respondToRequest).toHaveBeenCalledTimes(1); + expect(harness.interruptTurn).not.toHaveBeenCalled(); }); it("reacts to thread.user-input.respond by forwarding structured user input answers", async () => { diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index ff639797179..f1afd492ecc 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -1227,13 +1227,14 @@ const make = Effect.gen(function* () { }); } - yield* providerService + const responseSucceeded = yield* providerService .respondToRequest({ threadId: event.payload.threadId, requestId: event.payload.requestId, decision: event.payload.decision, }) .pipe( + Effect.as(true), Effect.catchCause((cause) => appendProviderFailureActivity({ threadId: event.payload.threadId, @@ -1245,9 +1246,13 @@ const make = Effect.gen(function* () { turnId: null, createdAt: event.payload.createdAt, requestId: event.payload.requestId, - }), + }).pipe(Effect.as(false)), ), ); + + if (responseSucceeded && event.payload.decision === "cancel") { + yield* providerService.interruptTurn({ threadId: event.payload.threadId }); + } }); const processUserInputResponseRequested = Effect.fn("processUserInputResponseRequested")( From 5c752eaf8280262815e51afad0c2e48151b06aa3 Mon Sep 17 00:00:00 2001 From: "t3-code[bot]" <269035359+t3-code[bot]@users.noreply.github.com> Date: Wed, 5 Aug 2026 21:23:15 +0000 Subject: [PATCH 2/2] fix(server): report approval cancellation interrupt failures --- .../Layers/ProviderCommandReactor.test.ts | 350 ++++++++++++------ .../Layers/ProviderCommandReactor.ts | 13 +- 2 files changed, 241 insertions(+), 122 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 383d8284b61..a551987e688 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -233,7 +233,7 @@ describe("ProviderCommandReactor", () => { turnId: asTurnId("turn-1"), }), ); - const interruptTurn = vi.fn((_: unknown) => Effect.void); + const interruptTurn = vi.fn(() => Effect.void); const respondToRequest = vi.fn(() => Effect.void); const respondToUserInput = vi.fn(() => Effect.void); const stopSession = vi.fn((input: unknown) => @@ -2606,7 +2606,7 @@ describe("ProviderCommandReactor", () => { }); }); - it("forwards approval declines without interrupting the provider turn", async () => { + it("reacts to thread.approval.respond by forwarding provider approval response", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; @@ -2631,136 +2631,20 @@ describe("ProviderCommandReactor", () => { await Effect.runPromise( harness.engine.dispatch({ type: "thread.approval.respond", - commandId: CommandId.make("cmd-approval-decline"), + commandId: CommandId.make("cmd-approval-respond"), threadId: ThreadId.make("thread-1"), requestId: asApprovalRequestId("approval-request-1"), - decision: "decline", + decision: "accept", createdAt: now, }), ); await waitFor(() => harness.respondToRequest.mock.calls.length === 1); - await harness.drain(); expect(harness.respondToRequest.mock.calls[0]?.[0]).toEqual({ threadId: "thread-1", requestId: "approval-request-1", - decision: "decline", + decision: "accept", }); - expect(harness.interruptTurn).not.toHaveBeenCalled(); - }); - - it("responds to approval cancellation before interrupting the provider turn", async () => { - const harness = await createHarness(); - const now = "2026-01-01T00:00:00.000Z"; - const callOrder: Array<"respond" | "interrupt"> = []; - harness.respondToRequest.mockImplementation(() => - Effect.sync(() => { - callOrder.push("respond"); - }), - ); - harness.interruptTurn.mockImplementation(() => - Effect.sync(() => { - callOrder.push("interrupt"); - }), - ); - - await Effect.runPromise( - harness.engine.dispatch({ - type: "thread.session.set", - commandId: CommandId.make("cmd-session-set-for-approval-cancel"), - threadId: ThreadId.make("thread-1"), - session: { - threadId: ThreadId.make("thread-1"), - status: "running", - providerName: "codex", - runtimeMode: "approval-required", - activeTurnId: asTurnId("turn-1"), - lastError: null, - updatedAt: now, - }, - createdAt: now, - }), - ); - - await Effect.runPromise( - harness.engine.dispatch({ - type: "thread.approval.respond", - commandId: CommandId.make("cmd-approval-cancel"), - threadId: ThreadId.make("thread-1"), - requestId: asApprovalRequestId("approval-request-1"), - decision: "cancel", - createdAt: now, - }), - ); - - await waitFor(() => harness.interruptTurn.mock.calls.length === 1); - await harness.drain(); - expect(harness.respondToRequest.mock.calls[0]?.[0]).toEqual({ - threadId: "thread-1", - requestId: "approval-request-1", - decision: "cancel", - }); - expect(harness.interruptTurn.mock.calls[0]?.[0]).toEqual({ - threadId: "thread-1", - }); - expect(callOrder).toEqual(["respond", "interrupt"]); - }); - - it("does not interrupt when the approval cancellation response fails", async () => { - const harness = await createHarness(); - const now = "2026-01-01T00:00:00.000Z"; - harness.respondToRequest.mockImplementation(() => - Effect.fail( - new ProviderAdapterRequestError({ - provider: ProviderDriverKind.make("codex"), - method: "session/request_permission", - detail: "Approval response failed", - }), - ), - ); - - await Effect.runPromise( - harness.engine.dispatch({ - type: "thread.session.set", - commandId: CommandId.make("cmd-session-set-for-failed-approval-cancel"), - threadId: ThreadId.make("thread-1"), - session: { - threadId: ThreadId.make("thread-1"), - status: "running", - providerName: "codex", - runtimeMode: "approval-required", - activeTurnId: asTurnId("turn-1"), - lastError: null, - updatedAt: now, - }, - createdAt: now, - }), - ); - - await Effect.runPromise( - harness.engine.dispatch({ - type: "thread.approval.respond", - commandId: CommandId.make("cmd-approval-cancel-failed"), - threadId: ThreadId.make("thread-1"), - requestId: asApprovalRequestId("approval-request-1"), - decision: "cancel", - createdAt: now, - }), - ); - - await waitFor(async () => { - const readModel = await harness.readModel(); - const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); - return ( - thread?.activities.some( - (activity) => activity.kind === "provider.approval.respond.failed", - ) ?? false - ); - }); - await harness.drain(); - - expect(harness.respondToRequest).toHaveBeenCalledTimes(1); - expect(harness.interruptTurn).not.toHaveBeenCalled(); }); it("reacts to thread.user-input.respond by forwarding structured user input answers", async () => { @@ -3053,4 +2937,228 @@ describe("ProviderCommandReactor", () => { expect(thread?.session?.providerInstanceId).toBe(ProviderInstanceId.make("codex_work")); expect(thread?.session?.activeTurnId).toBeNull(); }); + + it("forwards approval declines without interrupting the provider turn", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-set-for-approval-decline"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + createdAt: now, + }), + ); + + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.approval.respond", + commandId: CommandId.make("cmd-approval-decline"), + threadId: ThreadId.make("thread-1"), + requestId: asApprovalRequestId("approval-request-1"), + decision: "decline", + createdAt: now, + }), + ); + + await waitFor(() => harness.respondToRequest.mock.calls.length === 1); + await harness.drain(); + expect(harness.respondToRequest.mock.calls[0]?.[0]).toEqual({ + threadId: "thread-1", + requestId: "approval-request-1", + decision: "decline", + }); + expect(harness.interruptTurn).not.toHaveBeenCalled(); + }); + + it("responds to approval cancellation before interrupting the provider turn", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + const callOrder: Array<"respond" | "interrupt"> = []; + harness.respondToRequest.mockImplementation(() => + Effect.sync(() => { + callOrder.push("respond"); + }), + ); + harness.interruptTurn.mockImplementation(() => + Effect.sync(() => { + callOrder.push("interrupt"); + }), + ); + + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-set-for-approval-cancel"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: asTurnId("turn-1"), + lastError: null, + updatedAt: now, + }, + createdAt: now, + }), + ); + + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.approval.respond", + commandId: CommandId.make("cmd-approval-cancel"), + threadId: ThreadId.make("thread-1"), + requestId: asApprovalRequestId("approval-request-1"), + decision: "cancel", + createdAt: now, + }), + ); + + await waitFor(() => harness.interruptTurn.mock.calls.length === 1); + await harness.drain(); + expect(harness.respondToRequest.mock.calls[0]?.[0]).toEqual({ + threadId: "thread-1", + requestId: "approval-request-1", + decision: "cancel", + }); + expect(harness.interruptTurn.mock.calls[0]?.[0]).toEqual({ + threadId: "thread-1", + }); + expect(callOrder).toEqual(["respond", "interrupt"]); + }); + + it("does not interrupt when the approval cancellation response fails", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + harness.respondToRequest.mockImplementation(() => + Effect.fail( + new ProviderAdapterRequestError({ + provider: ProviderDriverKind.make("codex"), + method: "session/request_permission", + detail: "Approval response failed", + }), + ), + ); + + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-set-for-failed-approval-cancel"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: asTurnId("turn-1"), + lastError: null, + updatedAt: now, + }, + createdAt: now, + }), + ); + + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.approval.respond", + commandId: CommandId.make("cmd-approval-cancel-response-failed"), + threadId: ThreadId.make("thread-1"), + requestId: asApprovalRequestId("approval-request-1"), + decision: "cancel", + createdAt: now, + }), + ); + + await waitFor(async () => { + const readModel = await harness.readModel(); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); + return ( + thread?.activities.some( + (activity) => activity.kind === "provider.approval.respond.failed", + ) ?? false + ); + }); + await harness.drain(); + + expect(harness.respondToRequest).toHaveBeenCalledTimes(1); + expect(harness.interruptTurn).not.toHaveBeenCalled(); + }); + + it("records a failed interrupt after a successful approval cancellation", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + harness.interruptTurn.mockImplementation(() => + Effect.fail( + new ProviderAdapterRequestError({ + provider: ProviderDriverKind.make("codex"), + method: "turn/interrupt", + detail: "Turn interrupt failed", + }), + ), + ); + + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-set-for-failed-approval-interrupt"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: asTurnId("turn-1"), + lastError: null, + updatedAt: now, + }, + createdAt: now, + }), + ); + + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.approval.respond", + commandId: CommandId.make("cmd-approval-cancel-interrupt-failed"), + threadId: ThreadId.make("thread-1"), + requestId: asApprovalRequestId("approval-request-1"), + decision: "cancel", + createdAt: now, + }), + ); + + await waitFor(async () => { + const readModel = await harness.readModel(); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); + return ( + thread?.activities.some((activity) => activity.kind === "provider.turn.interrupt.failed") ?? + false + ); + }); + await harness.drain(); + + const readModel = await harness.readModel(); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); + expect(harness.respondToRequest).toHaveBeenCalledTimes(1); + expect(harness.interruptTurn).toHaveBeenCalledTimes(1); + expect( + thread?.activities.find((activity) => activity.kind === "provider.turn.interrupt.failed"), + ).toMatchObject({ + summary: "Provider turn interrupt failed", + turnId: "turn-1", + payload: { + detail: expect.stringContaining("Turn interrupt failed"), + }, + }); + }); }); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index f1afd492ecc..3a09928730c 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -1251,7 +1251,18 @@ const make = Effect.gen(function* () { ); if (responseSucceeded && event.payload.decision === "cancel") { - yield* providerService.interruptTurn({ threadId: event.payload.threadId }); + yield* providerService.interruptTurn({ threadId: event.payload.threadId }).pipe( + Effect.catchCause((cause) => + appendProviderFailureActivity({ + threadId: event.payload.threadId, + kind: "provider.turn.interrupt.failed", + summary: "Provider turn interrupt failed", + detail: Cause.pretty(cause), + turnId: thread.session?.activeTurnId ?? null, + createdAt: event.payload.createdAt, + }), + ), + ); } });