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, ), ), );