From b2f9e6b4e2210fcc6ed54444cccf19015667f3f6 Mon Sep 17 00:00:00 2001 From: T3 Code PR Stack <41898282+github-actions[bot]@users.noreply.github.com> Date: Sun, 2 Aug 2026 10:56:04 +0200 Subject: [PATCH] fix(server): bound provider control commands --- .../Layers/ProviderCommandReactor.test.ts | 75 ++++++++++++++++++- .../Layers/ProviderCommandReactor.ts | 70 +++++++++++------ 2 files changed, 122 insertions(+), 23 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index e1019634fc99..b36f6dbca243 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -167,6 +167,7 @@ describe("ProviderCommandReactor", () => { readonly startSessionEffect?: ( session: ProviderSession, ) => Effect.Effect; + readonly interruptTurnEffect?: ProviderServiceShape["interruptTurn"]; }) { const now = "2026-01-01T00:00:00.000Z"; const baseDir = @@ -252,7 +253,7 @@ describe("ProviderCommandReactor", () => { turnId: asTurnId("turn-1"), }), ); - const interruptTurn = vi.fn((_: unknown) => Effect.void); + const interruptTurn = vi.fn(input?.interruptTurnEffect ?? ((_: unknown) => Effect.void)); const respondToRequest = vi.fn(() => Effect.void); const respondToUserInput = vi.fn(() => Effect.void); const stopSession = vi.fn((input: unknown) => @@ -2490,6 +2491,78 @@ describe("ProviderCommandReactor", () => { }); }); + it("bounds a hung provider interrupt so later thread starts still run", async () => { + const harness = await createHarness({ + interruptTurnEffect: () => Effect.never, + }); + const now = "2026-01-01T00:00:00.000Z"; + const secondThreadId = ThreadId.make("thread-2"); + const modelSelection: ModelSelection = { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }; + + await harness.dispatch({ + type: "thread.create", + commandId: CommandId.make("cmd-create-thread-after-hung-interrupt"), + threadId: secondThreadId, + projectId: asProjectId("project-1"), + title: "Independent thread", + modelSelection, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt: now, + }); + await harness.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-running-session-before-hung-interrupt"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "codex", + runtimeMode: "approval-required", + activeTurnId: asTurnId("turn-hung-interrupt"), + lastError: null, + updatedAt: now, + }, + createdAt: now, + }); + await harness.dispatch({ + type: "thread.turn.interrupt", + commandId: CommandId.make("cmd-hung-interrupt"), + threadId: ThreadId.make("thread-1"), + turnId: asTurnId("turn-hung-interrupt"), + createdAt: now, + }); + await harness.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-start-after-hung-interrupt"), + threadId: secondThreadId, + message: { + messageId: asMessageId("message-after-hung-interrupt"), + role: "user", + text: "this thread must still start", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt: now, + }); + + await waitFor(async () => { + const readModel = await harness.readModel(); + const interrupted = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); + return interrupted?.session?.status === "ready"; + }); + await waitFor(() => harness.sendTurn.mock.calls.length === 1, 8_000); + expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ + threadId: secondThreadId, + }); + }); + it("starts a fresh session when only projected session state exists", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 88d38d886644..e2626345d121 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -65,6 +65,8 @@ import { } from "../../serverSettings.ts"; import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts"; import { GitWorkflowService } from "../../git/GitWorkflowService.ts"; + +const PROVIDER_CONTROL_TIMEOUT = Duration.seconds(5); const isProviderAdapterRequestError = Schema.is(ProviderAdapterRequestError); const isProviderDriverKind = Schema.is(ProviderDriverKind); @@ -1180,30 +1182,38 @@ const make = Effect.gen(function* () { } // Orchestration turn ids are not provider turn ids, so interrupt by session. - // Clearing the projection here is authoritative: a persisted session may - // say "running" even though its provider process disappeared yesterday. - // Provider cancellation remains best-effort so Abort always restores a - // usable composer without mistaking a quiet, live process for a dead one. - yield* providerService + // Clear the projection before touching the provider. This state transition + // is authoritative and must not depend on a cooperative protocol peer. + yield* setThreadSession({ + threadId: event.payload.threadId, + session: { + ...thread.session, + status: "ready", + activeTurnId: null, + updatedAt: event.payload.createdAt, + }, + createdAt: event.payload.createdAt, + }); + + // Provider cancellation is best-effort and bounded. Some protocol peers + // never answer cancellation; an interruptible timeout releases this + // thread's command lane even in that case. + const interruptResult = yield* providerService .interruptTurn({ threadId: event.payload.threadId }) .pipe( + Effect.interruptible, + Effect.timeoutOption(PROVIDER_CONTROL_TIMEOUT), Effect.catchCause((cause) => - Effect.logWarning("provider turn interrupt failed", { cause }), + Effect.logWarning("provider turn interrupt failed", { + threadId: event.payload.threadId, + cause: Cause.pretty(cause), + }).pipe(Effect.as(Option.some(undefined))), ), ); - - const latestThread = yield* resolveThread(event.payload.threadId); - const session = latestThread?.session; - if (session && session.status !== "stopped") { - yield* setThreadSession({ + if (Option.isNone(interruptResult)) { + yield* Effect.logWarning("provider turn interrupt timed out", { threadId: event.payload.threadId, - session: { - ...session, - status: "ready", - activeTurnId: null, - updatedAt: event.payload.createdAt, - }, - createdAt: event.payload.createdAt, + timeout: Duration.format(PROVIDER_CONTROL_TIMEOUT), }); } }); @@ -1344,10 +1354,7 @@ const make = Effect.gen(function* () { } const now = event.payload.createdAt; - if (context.session && context.session.status !== "stopped") { - yield* providerService.stopSession({ threadId: context.threadId }); - } - + const shouldStopProvider = context.session && context.session.status !== "stopped"; yield* setThreadSession({ threadId: context.threadId, session: { @@ -1364,6 +1371,25 @@ const make = Effect.gen(function* () { }, createdAt: now, }); + + if (shouldStopProvider) { + const stopResult = yield* providerService.stopSession({ threadId: context.threadId }).pipe( + Effect.interruptible, + Effect.timeoutOption(PROVIDER_CONTROL_TIMEOUT), + Effect.catchCause((cause) => + Effect.logWarning("provider session stop failed", { + threadId: context.threadId, + cause: Cause.pretty(cause), + }).pipe(Effect.as(Option.some(undefined))), + ), + ); + if (Option.isNone(stopResult)) { + yield* Effect.logWarning("provider session stop timed out", { + threadId: context.threadId, + timeout: Duration.format(PROVIDER_CONTROL_TIMEOUT), + }); + } + } }); const setRecoveryFailureState = Effect.fn("setRecoveryFailureState")(function* (input: {