Skip to content

Commit efc1751

Browse files
juliusmarmingejuliusmarminge
committed
fix(server): serialize recovery cancellation with thread commands
1 parent 5c2b13f commit efc1751

4 files changed

Lines changed: 45 additions & 22 deletions

File tree

‎apps/server/src/orchestration-v2/RunExecutionService.test.ts‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@ const layerTurnItemPositionsTest = Layer.mock(TurnItemPositionStore.TurnItemPosi
6464
const layerRunExecutionTest = RunExecutionService.layer.pipe(
6565
Layer.provide(
6666
Layer.mergeAll(
67+
ThreadCommandExecutor.layer,
6768
McpAppModelContext.layerEmpty,
6869
Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
6970
layerTurnItemPositionsTest,
@@ -748,6 +749,7 @@ it.effect("passes the thread's MCP app context to the provider under a safe key"
748749
RunExecutionService.layer.pipe(
749750
Layer.provide(
750751
Layer.mergeAll(
752+
ThreadCommandExecutor.layer,
751753
Layer.mock(McpAppModelContext.McpAppModelContext)({
752754
forThread: () =>
753755
Effect.succeed([
@@ -788,6 +790,7 @@ it.effect("fails the run when its ownership check cannot be read before calling
788790
const layerTest = RunExecutionService.layer.pipe(
789791
Layer.provide(
790792
Layer.mergeAll(
793+
ThreadCommandExecutor.layer,
791794
McpAppModelContext.layerEmpty,
792795
Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
793796
layerTurnItemPositionsTest,
@@ -1060,6 +1063,7 @@ it.effect("starts the provider when checkpoint baseline capture fails", () =>
10601063
const layerTest = RunExecutionService.layer.pipe(
10611064
Layer.provide(
10621065
Layer.mergeAll(
1066+
ThreadCommandExecutor.layer,
10631067
McpAppModelContext.layerEmpty,
10641068
Layer.mock(CheckpointService.CheckpointServiceV2)({
10651069
captureBaseline: () =>
@@ -1179,6 +1183,7 @@ it.effect.each(["failure", "interruption", "stale-attempt", "start-guard"] as co
11791183
const layerTest = RunExecutionService.layer.pipe(
11801184
Layer.provide(
11811185
Layer.mergeAll(
1186+
ThreadCommandExecutor.layer,
11821187
McpAppModelContext.layerEmpty,
11831188
Layer.mock(CheckpointService.CheckpointServiceV2)({
11841189
captureBaseline: () =>
@@ -1349,6 +1354,7 @@ it.effect("ingests new member threads of an owned child after the root turn term
13491354
const layerTest = RunExecutionService.layer.pipe(
13501355
Layer.provide(
13511356
Layer.mergeAll(
1357+
ThreadCommandExecutor.layer,
13521358
McpAppModelContext.layerEmpty,
13531359
Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
13541360
layerTurnItemPositionsTest,
@@ -1579,6 +1585,7 @@ it.effect("ends a nested native subagent and its thread when the root run is int
15791585
const layerTest = RunExecutionService.layer.pipe(
15801586
Layer.provide(
15811587
Layer.mergeAll(
1588+
ThreadCommandExecutor.layer,
15821589
McpAppModelContext.layerEmpty,
15831590
Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
15841591
layerTurnItemPositionsTest,
@@ -1935,6 +1942,7 @@ it.effect(
19351942
const layerTest = RunExecutionService.layer.pipe(
19361943
Layer.provide(
19371944
Layer.mergeAll(
1945+
ThreadCommandExecutor.layer,
19381946
McpAppModelContext.layerEmpty,
19391947
Layer.mock(CheckpointService.CheckpointServiceV2)({
19401948
captureBaseline: () => Effect.void,
@@ -2151,6 +2159,7 @@ it.effect("drops late root provider-thread writes from a superseded attempt", ()
21512159
const layerTest = RunExecutionService.layer.pipe(
21522160
Layer.provide(
21532161
Layer.mergeAll(
2162+
ThreadCommandExecutor.layer,
21542163
McpAppModelContext.layerEmpty,
21552164
Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
21562165
layerTurnItemPositionsTest,
@@ -2351,6 +2360,7 @@ it.effect(
23512360
const layerTest = RunExecutionService.layer.pipe(
23522361
Layer.provide(
23532362
Layer.mergeAll(
2363+
ThreadCommandExecutor.layer,
23542364
McpAppModelContext.layerEmpty,
23552365
Layer.mock(CheckpointService.CheckpointServiceV2)({
23562366
captureBaseline: () => Effect.void,
@@ -2521,6 +2531,7 @@ it.effect(
25212531
const layerTest = RunExecutionService.layer.pipe(
25222532
Layer.provide(
25232533
Layer.mergeAll(
2534+
ThreadCommandExecutor.layer,
25242535
McpAppModelContext.layerEmpty,
25252536
Layer.mock(CheckpointService.CheckpointServiceV2)({
25262537
captureBaseline: () => Effect.void,
@@ -2684,6 +2695,7 @@ it.effect(
26842695
const layerTest = RunExecutionService.layer.pipe(
26852696
Layer.provide(
26862697
Layer.mergeAll(
2698+
ThreadCommandExecutor.layer,
26872699
McpAppModelContext.layerEmpty,
26882700
Layer.mock(CheckpointService.CheckpointServiceV2)({
26892701
captureBaseline: () => Effect.void,
@@ -3125,6 +3137,7 @@ it.effect(
31253137
const layerTest = RunExecutionService.layer.pipe(
31263138
Layer.provide(
31273139
Layer.mergeAll(
3140+
ThreadCommandExecutor.layer,
31283141
McpAppModelContext.layerEmpty,
31293142
Layer.mock(CheckpointService.CheckpointServiceV2)({
31303143
captureBaseline: () => Effect.void,
@@ -4812,6 +4825,7 @@ function runBackgroundItemScenario(
48124825
const layerTest = RunExecutionService.layer.pipe(
48134826
Layer.provide(
48144827
Layer.mergeAll(
4828+
ThreadCommandExecutor.layer,
48154829
McpAppModelContext.layerEmpty,
48164830
Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
48174831
layerTurnItemPositionsTest,

‎apps/server/src/orchestration-v2/RunExecutionService.ts‎

Lines changed: 29 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@ import {
5353
} from "@t3tools/provider-core/server/failure";
5454
import * as RunFinalizationService from "./RunFinalizationService.ts";
5555
import * as TurnItemPositionStore from "./TurnItemPositionStore.ts";
56+
import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
5657

5758
export interface ProviderEventRoutingState {
5859
readonly ownedThreadIds: ReadonlySet<ThreadId>;
@@ -573,6 +574,7 @@ export const layer: Layer.Layer<
573574
| ServerSettings.ServerSettingsService
574575
| McpAppModelContext.McpAppModelContext
575576
| TurnItemPositionStore.TurnItemPositionStoreV2
577+
| ThreadCommandExecutor.ThreadCommandExecutor
576578
> = Layer.effect(
577579
RunExecutionServiceV2,
578580
Effect.gen(function* () {
@@ -583,6 +585,7 @@ export const layer: Layer.Layer<
583585
const serverSettings = yield* ServerSettings.ServerSettingsService;
584586
const mcpAppModelContext = yield* McpAppModelContext.McpAppModelContext;
585587
const turnItemPositions = yield* TurnItemPositionStore.TurnItemPositionStoreV2;
588+
const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor;
586589
const finalizationObserver = yield* RunFinalizationService.RunFinalizationObserver;
587590

588591
const writeFinalRunEvents = (input: {
@@ -1231,28 +1234,32 @@ export const layer: Layer.Layer<
12311234
let threadDisposition: ProviderTerminalEvent["threadDisposition"] = "reusable";
12321235
if (yield* Deferred.isDone(publicationFailed)) {
12331236
const stopped = yield* Effect.exit(
1234-
Effect.gen(function* () {
1235-
// The start guard checks both the original attempt and running status.
1236-
const stillOwned = yield* (
1237-
input.shouldStartProviderTurn ??
1238-
input.shouldFinalizeRun ??
1239-
(() => Effect.succeed(true))
1240-
)();
1241-
if (!stillOwned) {
1242-
return "stale" as const;
1243-
}
1244-
if (rootProviderTurnId === null) {
1245-
return "unknown" as const;
1246-
}
1247-
yield* input.session
1248-
.interruptTurn({
1249-
providerThread,
1250-
providerTurnId: rootProviderTurnId,
1251-
requestRuntimeRestart: true,
1252-
})
1253-
.pipe(Effect.timeout("10 seconds"));
1254-
return "stopped" as const;
1255-
}),
1237+
threadCommands.withLock(
1238+
input.run.threadId,
1239+
Effect.gen(function* () {
1240+
// Adapters can stop thread-wide work before matching the turn ID.
1241+
// Keep Stop and queue advancement behind this ownership check and stop.
1242+
const stillOwned = yield* (
1243+
input.shouldStartProviderTurn ??
1244+
input.shouldFinalizeRun ??
1245+
(() => Effect.succeed(true))
1246+
)();
1247+
if (!stillOwned) {
1248+
return "stale" as const;
1249+
}
1250+
if (rootProviderTurnId === null) {
1251+
return "unknown" as const;
1252+
}
1253+
yield* input.session
1254+
.interruptTurn({
1255+
providerThread,
1256+
providerTurnId: rootProviderTurnId,
1257+
requestRuntimeRestart: true,
1258+
})
1259+
.pipe(Effect.timeout("30 seconds"));
1260+
return "stopped" as const;
1261+
}),
1262+
),
12561263
);
12571264
if (Exit.isFailure(stopped)) {
12581265
if (Cause.hasInterruptsOnly(stopped.cause)) {

‎apps/server/src/orchestration-v2/runtimeLayer.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -150,6 +150,7 @@ const layerProviderAuthServiceProvided = ProviderAuthService.layer.pipe(
150150
const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe(
151151
Layer.provide(
152152
Layer.mergeAll(
153+
ThreadCommandExecutor.layer,
153154
McpAppModelContext.layer,
154155
layerCheckpointServiceProvided,
155156
layerEventSinkProvided,

‎apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -379,6 +379,7 @@ export function layerWithRegistry<Error>(
379379
const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe(
380380
Layer.provide(
381381
Layer.mergeAll(
382+
ThreadCommandExecutor.layer,
382383
McpAppModelContext.layerEmpty,
383384
layerCheckpointServiceProvided,
384385
layerEventSinkProvided,

0 commit comments

Comments
 (0)