From d24268c7dd292789e9d1d81e705aec791cf30118 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 07:36:05 +0000 Subject: [PATCH] fix(server): Stop interrupts a provider turn that starts while it lands Stop plans from stored provider turns. When it landed while the provider was still accepting the prompt, it saw no turn, marked the run "interrupted before provider start", and never interrupted the provider. The turn kept running and every follow-up or rollback failed with "previous turn is still running". Ingestion now stores an attempt's first root provider turn and rechecks the attempt under the thread command lock that Stop holds. Stop either sees the turn and interrupts it, or has already settled the attempt, and ingestion interrupts the turn itself. Fixes pingdotgg/t3code#25715 AI-assisted; verified with the tests and tools listed above. --- .../RunExecutionService.test.ts | 66 +++++++++++++++++++ .../orchestration-v2/RunExecutionService.ts | 46 ++++++++++++- .../src/orchestration-v2/runtimeLayer.ts | 1 + .../testkit/ProviderReplayHarness.ts | 1 + 4 files changed, 113 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index eec68b1e84a54..0e584bcf0c0a5 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -60,6 +60,7 @@ import * as McpAppModelContext from "../mcpApps/McpAppModelContext.ts"; import * as PreviewAgentActivity from "../preview/PreviewAgentActivity.ts"; import * as RunExecutionService from "./RunExecutionService.ts"; import * as RunFinalizationService from "./RunFinalizationService.ts"; +import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; const mockEventSink = (implementation: Partial) => @@ -93,6 +94,7 @@ const layerRunExecutionTest = RunExecutionService.layer.pipe( layerTurnItemPositionsTest, mockEventSink({}), IdAllocator.layer, + ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ ingestNormalized: () => Effect.succeed([]), }), @@ -920,6 +922,7 @@ it.effect("passes the thread's MCP app context to the provider under a safe key" layerTurnItemPositionsTest, mockEventSink({}), IdAllocator.layer, + ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ ingestNormalized: () => Effect.succeed([]), }), @@ -954,6 +957,7 @@ it.effect("fails the run when its ownership check cannot be read before calling ), }), IdAllocator.layer, + ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ ingestNormalized: () => Effect.succeed([]), }), @@ -1166,6 +1170,7 @@ it.effect("starts the provider when checkpoint baseline capture fails", () => ]).pipe(Effect.as([])), }), IdAllocator.layer, + ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ ingestNormalized: () => Effect.succeed([]), }), @@ -1288,6 +1293,7 @@ it.effect.each(["failure", "interruption", "stale-attempt", "start-guard"] as co }), }), IdAllocator.layer, + ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ ingestNormalized: () => Effect.succeed([]), }), @@ -1468,6 +1474,7 @@ it.effect("ingests new member threads of an owned child after the root turn term writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }), }), IdAllocator.layer, + ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ ingestNormalized: (input) => Effect.gen(function* () { @@ -1723,6 +1730,7 @@ it.effect("ends a nested native subagent and its thread when the root run is int ), }), IdAllocator.layer, + ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ ingestNormalized: () => Effect.succeed([]), }), @@ -2169,6 +2177,7 @@ it.effect( writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }), }), IdAllocator.layer, + ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ ingestNormalized: (input) => Ref.update(ingested, (current) => [...current, input.event]).pipe(Effect.as([])), @@ -2605,6 +2614,7 @@ it.effect("settles native subagents when a workspace handoff completes the root ), }), IdAllocator.layer, + ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ ingestNormalized: () => Effect.succeed([]), }), @@ -3204,6 +3214,60 @@ it.effect("does not interrupt a provider when ingestion fails before the root tu }), ); +it.effect("interrupts a provider turn that starts after Stop settled its attempt", () => + Effect.gen(function* () { + const now = yield* DateTime.now; + const key = "turn-after-stop"; + const ids = backgroundScenarioIds(key); + const stopped = yield* Ref.make(false); + const accepted = yield* Deferred.make(); + const interrupted = yield* Deferred.make(); + yield* captureRootRunTermination({ + key, + initialProviderTurnId: null, + shouldFinalizeRun: () => Ref.get(stopped).pipe(Effect.map((value) => !value)), + shouldStartProviderTurn: () => Ref.get(stopped).pipe(Effect.map((value) => !value)), + // Stop commits while the provider accepts the prompt. + startTurn: () => + Ref.set(stopped, true).pipe(Effect.andThen(Deferred.succeed(accepted, undefined))), + events: () => + Stream.unwrap( + Deferred.await(accepted).pipe( + Effect.as( + Stream.make({ + type: "provider_turn.updated" as const, + driver, + providerTurn: { + id: ids.rootProviderTurnId, + providerThreadId: ids.providerThreadId, + nodeId: ids.rootNodeId, + runAttemptId: ids.attemptId, + nativeTurnRef: null, + ordinal: 1, + status: "running" as const, + startedAt: now, + completedAt: null, + }, + }).pipe( + Stream.concat( + Stream.fromEffect( + Deferred.await(interrupted).pipe( + Effect.as(rootTerminalEvent(ids, "interrupted")), + ), + ), + ), + ), + ), + ), + ), + interruptTurn: (input) => Deferred.succeed(interrupted, input).pipe(Effect.asVoid), + }); + const call = yield* Deferred.await(interrupted); + assert.equal(call.providerTurnId, ids.rootProviderTurnId); + assert.equal(call.providerThread.id, ids.providerThreadId); + }), +); + // The server browser keeps an agent's tab focused for exactly this span. it.effect.each(["completed", "failed"] as const)( "reports the thread executing until its %s run's event consumer ends", @@ -3463,6 +3527,7 @@ function captureRootRunTermination(input: { }), }), IdAllocator.layer, + ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ ingestNormalized: input.ingestNormalized ?? (() => Effect.succeed([])), }), @@ -4001,6 +4066,7 @@ function runBackgroundItemScenario( writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }), }), IdAllocator.layer, + ThreadCommandExecutor.layer, Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({ ingestNormalized: (input) => { const write = Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index e9a7d674ff086..612a130e019aa 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -55,6 +55,7 @@ import { makeProviderFailureTurnItem, } from "@t3tools/provider-core/server/failure"; import * as RunFinalizationService from "./RunFinalizationService.ts"; +import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; import { userFacingDatabaseErrorMessage } from "./UserFacingErrors.ts"; @@ -614,6 +615,7 @@ export const layer: Layer.Layer< | McpAppModelContext.McpAppModelContext | PreviewAgentActivity.PreviewAgentActivity | TurnItemPositionStore.TurnItemPositionStoreV2 + | ThreadCommandExecutor.ThreadCommandExecutor > = Layer.effect( RunExecutionServiceV2, Effect.gen(function* () { @@ -628,6 +630,7 @@ export const layer: Layer.Layer< const turnItemPositions = yield* TurnItemPositionStore.TurnItemPositionStoreV2; const finalizationObserver = yield* RunFinalizationService.RunFinalizationObserver; const turnAnalytics = yield* ProviderUsageAnalytics.ProviderTurnAnalytics; + const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor; const writeFinalRunEvents = Effect.fn("RunExecutionService.writeFinalRunEvents")( function* (input: { @@ -1104,6 +1107,7 @@ export const layer: Layer.Layer< const openRunOwnedSubagents = yield* Ref.make(emptyOpenRunOwnedSubagentProjection()); // The last stored root snapshot's pendingBackgroundWork; adapters that omit it leave it false. let rootSnapshotPendingWork = false; + let rootTurnOwnershipChecked = false; const assistantStreaming = makeAssistantStreamingFilter(responseStreamingMode); // Set when ingestion ends or finalization settles the run's open items, which a later // streaming snapshot must not reopen. @@ -1589,7 +1593,47 @@ export const layer: Layer.Layer< event, DateTime.toEpochMillis(yield* DateTime.now), ); - if (deliveredEvent) { + const startedRootTurn = + event.type === "provider_turn.updated" && + event.providerTurn.runAttemptId === input.attempt.id && + event.providerTurn.providerThreadId === input.providerThread.id && + !isTerminalProviderTurnStatus(event.providerTurn.status) && + !rootTurnOwnershipChecked + ? event.providerTurn + : undefined; + const shouldStartProviderTurn = input.shouldStartProviderTurn; + if (deliveredEvent && startedRootTurn !== undefined && shouldStartProviderTurn) { + rootTurnOwnershipChecked = true; + // Stop plans under this lock from stored turns. Storing the first turn and rechecking + // the attempt under it means Stop either saw the turn or already settled the run. + const [stored, owned] = yield* threadCommands.withLock( + input.run.threadId, + Effect.all([ + ingestDelivered(event, deliveredEvent), + shouldStartProviderTurn().pipe(Effect.orElseSucceed(() => true)), + ]), + ); + storedEventCount = stored; + if (!owned) { + yield* Ref.get(latestProviderThread).pipe( + Effect.flatMap((providerThread) => + input.session.interruptTurn({ + providerThread, + providerTurnId: startedRootTurn.id, + }), + ), + Effect.catchCauseIf( + (cause) => !Cause.hasInterruptsOnly(cause), + (cause) => + Effect.logWarning( + "orchestration V2 failed to interrupt a provider turn that started after Stop", + { runId: input.run.id, cause }, + ), + ), + Effect.forkIn(scope), + ); + } + } else if (deliveredEvent) { storedEventCount = yield* ingestDelivered(event, deliveredEvent); } else if ( event.type === "turn_item.updated" && diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index a19894058c3a5..d83f0603d7f7c 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -160,6 +160,7 @@ const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe( IdAllocator.layer, layerProviderEventIngestorProvided, TurnItemPositionStore.layer, + ThreadCommandExecutor.layer, ), ), ); diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index e3672cfcc1c92..c742f83d37586 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -429,6 +429,7 @@ export function layerWithRegistry( layerProviderEventIngestorProvided, layerServerSettings, layerStores, + ThreadCommandExecutor.layer, ), ), );