From 12ae3f989c0b0f0c1b79441131f4d6cb9d068225 Mon Sep 17 00:00:00 2001 From: Pixel Perfect Date: Sat, 3 Oct 2026 13:26:26 -0700 Subject: [PATCH 1/6] fix(server): continue work after agent model changes --- .../src/mcp/toolkits/thread/handlers.ts | 10 +- apps/server/src/mcp/toolkits/thread/tools.ts | 2 +- .../SelectionRestart.integration.test.ts | 203 ++++++++++++++++++ .../ThreadManagementService.ts | 53 +++++ .../src/relay/AgentAwarenessRelay.test.ts | 1 + 5 files changed, 263 insertions(+), 6 deletions(-) diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index f54065c0824e..7656d04c57e5 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -8,7 +8,6 @@ import { type OrchestrationV2Command, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; -import { modelSelectionCommandType } from "@t3tools/shared/model"; import * as McpToolAccess from "../../McpToolAccess.ts"; import { @@ -188,15 +187,16 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { threads, projection: { thread }, } = yield* readThread(input.threadId); - const type = modelSelectionCommandType(thread.providerInstanceId, input.modelSelection); const result = yield* threads - .dispatch({ - type, + .configureModelSelection({ + projectId: thread.projectId, threadId: thread.id, commandId: yield* newCommandId(), modelSelection: input.modelSelection, + createdBy: "agent", + creationSource: "mcp", }) - .pipe(Effect.mapError(dispatchFailure)); + .pipe(Effect.mapError(unavailable)); return { sequence: result.sequence }; }), ), diff --git a/apps/server/src/mcp/toolkits/thread/tools.ts b/apps/server/src/mcp/toolkits/thread/tools.ts index b5fb7d6be3a6..bbda68dcbd8e 100644 --- a/apps/server/src/mcp/toolkits/thread/tools.ts +++ b/apps/server/src/mcp/toolkits/thread/tools.ts @@ -192,7 +192,7 @@ const ThreadConfigurationTool = Tool.make("t3_thread_configuration", { const ThreadConfigureTool = Tool.make("t3_thread_configure", { ...commandTool, description: - "Set a thread's provider, model and options with the existing selection command. Omit threadId for this thread. This does not change permission modes. Use orchestrator_capabilities to choose a selection.", + "Set a thread's provider, model and options. Omit threadId for this thread. A changed selection restarts a running turn and automatically continues its unfinished user request; an interrupted calling turn may not receive a tool result. An already-active selection is a no-op. Providers that cannot restart reject the change without interrupting the turn. This does not change permission modes. Use orchestrator_capabilities to choose a selection.", parameters: Schema.Struct({ threadId: Schema.optional(ThreadId), modelSelection: ModelSelection, diff --git a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts index 958fb89cc8e4..a98bd859a7f9 100644 --- a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts +++ b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts @@ -32,6 +32,7 @@ import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as ThreadManagementService from "./ThreadManagementService.ts"; const driver = ProviderDriverKind.make("codex"); const providerInstanceId = ProviderInstanceId.make("codex-restart-test"); @@ -388,6 +389,208 @@ function makeCompletingHandoffAdapter( }; } +it.live.each([ + "saved-model", + "options", + "provider", + "unchanged", + "idle", + "unsupported", + "preparing", +] as const)("configures the calling thread without losing work: %s", (scenario) => + Effect.scoped( + Effect.gen(function* () { + const name = `configure-continuation-${scenario}`; + const cwd = yield* checkpointWorkspace(name); + const threadId = ThreadId.make(`thread:${name}`); + const projectId = ProjectId.make(`project:${name}`); + const otherInstanceId = ProviderInstanceId.make("codex-configure-target"); + const selection = { + ...initialSelection, + options: [ + { id: "reasoningEffort", value: "low" }, + { id: "fastMode", value: true }, + ], + } satisfies ModelSelection; + const targetSelection: ModelSelection = + scenario === "unchanged" + ? { ...selection, options: selection.options.toReversed() } + : scenario === "options" + ? { ...selection, options: [{ id: "reasoningEffort", value: "xhigh" }] } + : { + ...replacementSelection, + instanceId: scenario === "provider" ? otherInstanceId : providerInstanceId, + }; + const state = yield* Ref.make({ + activeTurn: null, + opened: [], + started: [], + closedSessionCount: 0, + failedReplacementOpen: true, + }); + const starts: Array = []; + const capabilities = { + ...pooledCapabilities, + turns: { + ...pooledCapabilities.turns, + supportsSteeringByInterruptRestart: scenario !== "unsupported", + }, + }; + const adapters = [providerInstanceId, otherInstanceId].map((instanceId) => { + const base = makeRestartAdapter(state, capabilities, instanceId); + return { + ...base, + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" } as const), + openSession: (input) => + base.openSession(input).pipe( + Effect.map((session) => ({ + ...session, + startTurn: (input) => + Effect.gen(function* () { + starts.push(input); + yield* session.startTurn(input); + }), + })), + ), + } satisfies ProviderAdapter.ProviderAdapterV2["Service"]; + }); + const registry = Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistryV2, { + get: (instanceId) => + Effect.succeed(adapters.find((adapter) => adapter.instanceId === instanceId)!), + list: () => Effect.succeed([providerInstanceId, otherInstanceId]), + getMetadata: () => + Effect.succeed({ + driver, + continuationKey: "codex:home:/shared", + enabled: true, + capabilities, + }), + }); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threads = yield* ThreadManagementService.ThreadManagementService; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`${name}:create`), + threadId, + projectId, + title: name, + modelSelection: selection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + if (scenario !== "idle") { + const afterSequence = yield* orchestrator.getThreadEventSequence(threadId); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`${name}:first`), + threadId, + messageId: MessageId.make(`${name}:first`), + text: "Switch to the requested model and finish the poem.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + if (scenario !== "preparing") { + yield* worker.drain(); + yield* orchestrator.streamStoredEventsFrom({ threadId, afterSequence }).pipe( + Stream.filter( + ({ event }) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + ); + } + } + if (scenario === "saved-model") { + yield* orchestrator.dispatch({ + type: "thread.model-selection.set", + commandId: CommandId.make(`${name}:save-next-model`), + threadId, + modelSelection: targetSelection, + }); + yield* worker.drain(); + } + const before = yield* orchestrator.getThreadProjection(threadId); + const sequence = yield* orchestrator.getThreadEventSequence(threadId); + const configure = threads.configureModelSelection({ + projectId, + threadId, + commandId: CommandId.make(`${name}:configure`), + modelSelection: targetSelection, + createdBy: "agent", + creationSource: "mcp", + }); + if (scenario === "unsupported" || scenario === "preparing") { + const error = yield* Effect.flip(configure); + assert.equal( + error._tag, + scenario === "preparing" + ? "ThreadManagementNoSteerableRunError" + : "OrchestratorDispatchError", + ); + assert.deepEqual(yield* orchestrator.getThreadProjection(threadId), before); + assert.equal(yield* orchestrator.getThreadEventSequence(threadId), sequence); + return; + } + if (scenario === "unchanged" || scenario === "idle") { + const result = yield* configure; + const after = yield* orchestrator.getThreadProjection(threadId); + if (scenario === "unchanged") { + assert.equal(result.sequence, sequence); + assert.deepEqual(after, before); + } else { + assert.deepEqual(after.thread.modelSelection, targetSelection); + assert.isEmpty(after.runs); + assert.isEmpty(after.messages); + } + return; + } + yield* configure; + yield* worker.drain(); + yield* orchestrator.streamStoredEventsFrom({ threadId, afterSequence: sequence }).pipe( + Stream.filter(({ event }) => + scenario === "options" + ? event.type === "provider-turn.updated" && event.payload.status === "running" + : event.type === "run.updated" && event.payload.status === "completed", + ), + Stream.take(1), + Stream.runDrain, + ); + yield* worker.drain(); + const after = yield* orchestrator.getThreadProjection(threadId); + assert.lengthOf(starts, 2); + assert.deepEqual(starts[1]?.modelSelection, targetSelection); + assert.match(starts[1]!.message.text, /continue|finish/i); + assert.equal(after.runs[0]?.id, before.runs[0]?.id); + assert.deepEqual(after.runs[0]?.modelSelection, targetSelection); + assert.deepEqual(after.thread.modelSelection, targetSelection); + assert.lengthOf(after.attempts, 2); + assert.equal(after.attempts[0]?.status, "superseded"); + assert.equal(after.messages[1]?.createdBy, "agent"); + assert.equal(after.messages[1]?.creationSource, "mcp"); + }).pipe( + Effect.provide( + ThreadManagementService.layer.pipe( + Layer.provideMerge( + ProviderReplayHarness.layerWithRegistry({ name }, registry, { + runEffectWorker: scenario !== "preparing", + }), + ), + ), + ), + ); + }), + ), +); + it.live("restarts selection as a new attempt and retries after old-session cleanup", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 3da3ef12438d..ba367ab5218a 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -29,6 +29,7 @@ import { ThreadId, type TurnItemId, } from "@t3tools/contracts"; +import { modelSelectionCommandType, modelSelectionsEqual } from "@t3tools/shared/model"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; @@ -334,6 +335,14 @@ export interface ThreadManagementServiceShape { readonly sendToThread: ( input: ThreadManagementSendInput, ) => Effect.Effect; + readonly configureModelSelection: ( + input: ThreadManagementProvenance & { + readonly projectId: ProjectId; + readonly threadId: ThreadId; + readonly commandId: CommandId; + readonly modelSelection: ModelSelection; + }, + ) => Effect.Effect<{ readonly sequence: number }, ThreadManagementFailure>; readonly waitForThread: ( input: ThreadManagementWaitInput, ) => Effect.Effect; @@ -585,6 +594,49 @@ const make = Effect.gen(function* () { ), ); + const configureModelSelection = Effect.fn( + "orchestrationV2.threadManagement.configureModelSelection", + )(function* (input: Parameters[0]) { + const target = yield* getProjectThreadRecords(input, ["runs", "providerTurns"]); + const activeRun = latestActiveRun(target); + if ( + modelSelectionsEqual( + activeRun?.modelSelection ?? target.thread.modelSelection, + input.modelSelection, + ) + ) { + return { sequence: yield* orchestrator.getThreadEventSequence(input.threadId) }; + } + if (activeRun === undefined) { + return yield* orchestrator.dispatch({ + type: modelSelectionCommandType(target.thread.providerInstanceId, input.modelSelection), + commandId: input.commandId, + threadId: input.threadId, + modelSelection: input.modelSelection, + }); + } + if (latestSteerableRun(target)?.id !== activeRun.id) { + return yield* new ThreadManagementNoSteerableRunError({ + threadId: input.threadId, + mode: "restart", + }); + } + // Persist the replacement input with the selection before interrupting the + // calling turn, which may never receive its configure tool result. + return yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: input.commandId, + threadId: input.threadId, + messageId: MessageId.make(`${input.commandId}:model-continuation`), + text: "The requested model change has been applied. Continue the unfinished user request from where the previous turn stopped. Do not repeat the model change if the requested selection is already active.", + attachments: [], + modelSelection: input.modelSelection, + dispatchMode: { type: "restart_active", targetRunId: activeRun.id }, + createdBy: input.createdBy, + creationSource: input.creationSource, + }); + }); + const sendToThread: ThreadManagementServiceShape["sendToThread"] = (input) => Effect.gen(function* () { const target = yield* getProjectThreadRecords(input, ["runs", "providerTurns"]); @@ -914,6 +966,7 @@ const make = Effect.gen(function* () { getThreadShell: orchestrator.getThreadShell, listProjectThreads, sendToThread, + configureModelSelection, waitForThread, settleAfterRun, settleThread, diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index f10af26391c6..1d993e3c998f 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -208,6 +208,7 @@ const makeTestRelay = Effect.fnUntraced(function* ( getProjectThread: unused, listProjectThreads: unused, sendToThread: unused, + configureModelSelection: unused, waitForThread: unused, settleAfterRun: unused, settleThread: unused, From dd8f8fb8d450438fb9d9c2e5750732856464b75d Mon Sep 17 00:00:00 2001 From: Pixel Perfect Date: Sun, 11 Oct 2026 10:42:23 -0700 Subject: [PATCH 2/6] fix(server): preserve model configuration rejection details --- apps/server/src/mcp/toolkits/core.test.ts | 91 +++++++++++++++++++ .../src/mcp/toolkits/thread/handlers.ts | 17 +++- 2 files changed, 107 insertions(+), 1 deletion(-) diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index a938ba7fd999..fc176c9b00b4 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -255,6 +255,97 @@ it("bounds public command rejections and redacts internal dispatch causes", () = }); }); +it.effect.each([ + { + name: "reports a turn that cannot be restarted", + error: new ThreadManagement.ThreadManagementNoSteerableRunError({ + threadId, + mode: "restart", + }), + code: "thread_not_sendable", + message: `Thread ${threadId} has no running turn that can be restarted.`, + }, + { + name: "preserves a public dispatch rejection", + error: new OrchestratorDispatchError({ + commandId: CommandId.make("mcp-core-configure"), + commandType: "session.model.set", + cause: "The target provider instance is unavailable.", + }), + code: "orchestration_error", + message: "The target provider instance is unavailable.", + }, + { + name: "preserves a public command rejection", + error: new OrchestratorCommandRejectedError({ + commandId: CommandId.make("mcp-core-configure"), + commandType: "session.model.set", + cause: "The model cannot be changed while the thread is archived.", + }), + code: "orchestration_error", + message: "The model cannot be changed while the thread is archived.", + }, + { + name: "redacts a wrapped dispatch cause", + error: new OrchestratorDispatchError({ + commandId: CommandId.make("mcp-core-configure"), + commandType: "session.model.set", + cause: new Error("private-storage-path"), + }), + code: "orchestration_error", + message: "The operation could not be completed.", + }, + { + name: "redacts a service storage failure", + error: new ThreadManagement.ThreadManagementProjectionLoadError({ + projectId: McpToolAccessTestkit.liveThreadShell(threadId).projectId, + threadId, + cause: "private-storage-path", + }), + code: "orchestration_error", + message: "The operation could not be completed.", + }, +])("thread configuration $name", ({ error, code, message }) => + Effect.gen(function* () { + const server = yield* McpServer.McpServer; + const result = yield* server + .callTool({ + name: "t3_thread_configure", + arguments: { modelSelection: { instanceId: "codex", model: "gpt-5" } }, + }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, scope), + Effect.provideService(McpSchema.McpServerClient, client), + ); + expect(declaredFailure(result)).toEqual({ + _tag: "OrchestratorMcpFailure", + code, + message, + }); + expect(result.structuredContent).toBeUndefined(); + expect(JSON.stringify(result)).not.toContain("private-storage-path"); + }).pipe( + Effect.provide( + layerThreadToolkit.pipe( + Layer.provideMerge(McpServer.McpServer.layer), + Layer.provide(NodeCrypto.layer), + Layer.provide( + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: (id) => Effect.succeed(McpToolAccessTestkit.liveThreadShell(id)), + getProjectThreadRecords: ({ threadId }) => + Effect.succeed( + McpToolAccessTestkit.idleThreadProjection( + McpToolAccessTestkit.liveThreadShell(threadId), + ), + ), + configureModelSelection: () => Effect.fail(error), + }), + ), + ), + ), + ), +); + it.effect("returns an HTML render reference that Codex and Claude tool rows both carry", () => Effect.gen(function* () { const server = yield* McpServer.McpServer; diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index 7656d04c57e5..4bfbb67a8f61 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -196,7 +196,22 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { createdBy: "agent", creationSource: "mcp", }) - .pipe(Effect.mapError(unavailable)); + .pipe( + Effect.mapError((error) => { + switch (error._tag) { + case "ThreadManagementNoSteerableRunError": + return new OrchestratorMcpFailure({ + code: "thread_not_sendable", + message: error.message, + }); + case "OrchestratorDispatchError": + case "OrchestratorCommandRejectedError": + return dispatchFailure(error); + default: + return unavailable(); + } + }), + ); return { sequence: result.sequence }; }), ), From 1b564bc7237af29d42e0993486d9e35061c4258e Mon Sep 17 00:00:00 2001 From: Pixel Perfect Date: Sun, 11 Oct 2026 11:04:44 -0700 Subject: [PATCH 3/6] fix(server): reject model changes in archived threads --- apps/server/src/mcp/toolkits/core.test.ts | 6 +++ .../src/mcp/toolkits/thread/handlers.ts | 1 + .../SelectionRestart.integration.test.ts | 40 +++++++++++++++---- .../ThreadManagementService.ts | 5 +++ 4 files changed, 45 insertions(+), 7 deletions(-) diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 2ddf5feac3d1..21075ac9db69 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -263,6 +263,12 @@ it("bounds public command rejections and redacts internal dispatch causes", () = }); it.effect.each([ + { + name: "reports an archived thread", + error: new ThreadManagement.ThreadManagementThreadArchivedError({ threadId }), + code: "thread_not_sendable", + message: `Thread ${threadId} is archived and cannot receive messages.`, + }, { name: "reports a disabled provider instance", error: new ThreadManagement.ThreadManagementProviderInstanceDisabledError({ diff --git a/apps/server/src/mcp/toolkits/thread/handlers.ts b/apps/server/src/mcp/toolkits/thread/handlers.ts index 796a058c0b0a..15cae2a10048 100644 --- a/apps/server/src/mcp/toolkits/thread/handlers.ts +++ b/apps/server/src/mcp/toolkits/thread/handlers.ts @@ -199,6 +199,7 @@ export const layer = McpToolAccess.toLayer(ThreadToolkit, { .pipe( Effect.mapError((error) => { switch (error._tag) { + case "ThreadManagementThreadArchivedError": case "ThreadManagementNoSteerableRunError": return new OrchestratorMcpFailure({ code: "thread_not_sendable", diff --git a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts index 5431225a15ed..a58f2e566607 100644 --- a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts +++ b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts @@ -401,6 +401,7 @@ it.live.each([ "unsupported", "preparing", "disabled", + "archived", ] as const)("configures the calling thread without losing work: %s", (scenario) => Effect.scoped( Effect.gen(function* () { @@ -526,8 +527,26 @@ it.live.each([ }); yield* worker.drain(); } + if (scenario === "archived") { + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make(`${name}:archive`), + threadId, + }); + } const before = yield* orchestrator.getThreadProjection(threadId); const sequence = yield* orchestrator.getThreadEventSequence(threadId); + if (scenario === "archived") { + assert.isNotNull(before.thread.archivedAt); + const run = before.runs[0]!; + assert.equal(run.status, "running"); + assert.isTrue( + before.providerTurns.some( + (turn) => turn.runAttemptId === run.activeAttemptId && turn.status === "running", + ), + ); + assert.lengthOf(starts, 1); + } const configure = threads.configureModelSelection({ projectId, threadId, @@ -536,15 +555,22 @@ it.live.each([ createdBy: "agent", creationSource: "mcp", }); - if (scenario === "unsupported" || scenario === "preparing" || scenario === "disabled") { + if ( + scenario === "unsupported" || + scenario === "preparing" || + scenario === "disabled" || + scenario === "archived" + ) { const error = yield* Effect.flip(configure); assert.equal( error._tag, - scenario === "disabled" - ? "ThreadManagementProviderInstanceDisabledError" - : scenario === "preparing" - ? "ThreadManagementNoSteerableRunError" - : "OrchestratorDispatchError", + scenario === "archived" + ? "ThreadManagementThreadArchivedError" + : scenario === "disabled" + ? "ThreadManagementProviderInstanceDisabledError" + : scenario === "preparing" + ? "ThreadManagementNoSteerableRunError" + : "OrchestratorDispatchError", ); if (scenario === "disabled") { assert.deepEqual(before.runs[0]?.modelSelection, selection); @@ -617,7 +643,7 @@ it.live.each([ CommandReceiptStore.layer.pipe(Layer.provide(layerDatabase)), ProviderReplayHarness.layerWithRegistry({ name }, registry, { databaseLayer: layerDatabase, - runEffectWorker: scenario !== "preparing", + runEffectWorker: scenario !== "preparing" && scenario !== "archived", }), ), ), diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index e31da99a351d..7a09d668cfe0 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -743,6 +743,11 @@ const make = Effect.gen(function* () { "orchestrationV2.threadManagement.configureModelSelection", )(function* (input: Parameters[0]) { const target = yield* getProjectThreadRecords(input, ["runs", "providerTurns"]); + if (target.thread.archivedAt !== null) { + return yield* new ThreadManagementThreadArchivedError({ + threadId: input.threadId, + }); + } const activeRun = latestActiveRun(target); if ( modelSelectionsEqual( From ff3739bb45940a7396253581b479db8e56de0994 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 19:58:16 +0000 Subject: [PATCH 4/6] test(server): reproduce issue 15300 agent model switch interrupting the run Co-Authored-By: Claude Opus 5.5 --- .../SelectionRestart.integration.test.ts | 182 ++++++++++++++++++ 1 file changed, 182 insertions(+) diff --git a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts index 03cdf5e85a64..cba6da7301ee 100644 --- a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts +++ b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts @@ -1,6 +1,7 @@ import { assert, it } from "@effect/vitest"; import { CommandId, + EnvironmentId, EventId, MessageId, type ModelSelection, @@ -32,6 +33,16 @@ import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import { McpSchema, McpServer } from "effect/ai"; +import * as McpHttpServer from "../mcp/McpHttpServer.ts"; +import * as McpInvocationContext from "../mcp/McpInvocationContext.ts"; +import * as CommandReceiptStore from "./CommandReceiptStore.ts"; +import * as ThreadManagementService from "./ThreadManagementService.ts"; +import * as ThreadSearch from "./ThreadSearch.ts"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; +import * as ServerSettings from "../serverSettings.ts"; const driver = ProviderDriverKind.make("codex"); const providerInstanceId = ProviderInstanceId.make("codex-restart-test"); @@ -279,6 +290,177 @@ function makeRestartAdapter( }; } +// An agent that switches its own thread's model through MCP is the running +// turn the switch interrupts, so the configure call must leave behind a +// continuation of the unfinished request (#15300). +it.live("t3_thread_configure from the running turn continues the user request", () => + Effect.scoped( + Effect.gen(function* () { + const name = "configure-continuation"; + const cwd = yield* checkpointWorkspace(name); + const threadId = ThreadId.make(`thread:${name}`); + const otherInstanceId = ProviderInstanceId.make("codex-configure-target"); + const targetSelection = { + instanceId: otherInstanceId, + model: replacementSelection.model, + options: [{ id: "reasoningEffort", value: "low" }], + } satisfies ModelSelection; + const state = yield* Ref.make({ + activeTurn: null, + opened: [], + started: [], + closedSessionCount: 0, + failedReplacementOpen: true, + }); + const adapters = [providerInstanceId, otherInstanceId].map((instanceId) => + makeRestartAdapter(state, pooledCapabilities, instanceId), + ); + const registry = Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistryV2, { + get: (instanceId) => + Effect.succeed(adapters.find((adapter) => adapter.instanceId === instanceId)!), + list: () => Effect.succeed([providerInstanceId, otherInstanceId]), + getMetadata: () => + Effect.succeed({ + driver, + continuationKey: "codex:home:/shared", + enabled: true, + capabilities: pooledCapabilities, + }), + }); + const layerDatabase = SqlitePersistence.layerMemory; + const layerOrchestration = ThreadManagementService.layer.pipe( + Layer.provideMerge( + Layer.mergeAll( + ServerSettings.layerTest({ + providerInstances: { + [providerInstanceId]: { driver, enabled: true }, + [otherInstanceId]: { driver, enabled: true }, + }, + }), + CommandReceiptStore.layer.pipe(Layer.provide(layerDatabase)), + ProviderReplayHarness.layerWithRegistry({ name }, registry, { + databaseLayer: layerDatabase, + }), + ), + ), + ); + + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const server = yield* McpServer.McpServer; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`${name}:create`), + threadId, + projectId: ProjectId.make(`project:${name}`), + title: name, + modelSelection: initialSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const firstSequence = yield* orchestrator.getThreadEventSequence(threadId); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`${name}:first`), + threadId, + messageId: MessageId.make(`${name}:first`), + text: "Switch to the requested model, then write a four-line poem.", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* orchestrator.streamStoredEventsFrom({ threadId, afterSequence: firstSequence }).pipe( + Stream.filter( + ({ event }) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + ); + const before = yield* orchestrator.getThreadProjection(threadId); + const sequence = yield* orchestrator.getThreadEventSequence(threadId); + + const result = yield* server + .callTool({ + name: "t3_thread_configure", + arguments: { modelSelection: targetSelection }, + }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, { + environmentId: EnvironmentId.make(`environment:${name}`), + requestNamespace: `provider-session:${name}`, + thread: { + threadId, + providerSessionId: `provider-session:${name}`, + providerInstanceId, + }, + client: undefined, + issuedAt: 0, + capabilities: new Set(["orchestration"] as const), + }), + Effect.provideService( + McpSchema.McpServerClient, + McpSchema.McpServerClient.of({ + clientId: 1, + protocolVersion: "2025-06-18", + clientCapabilities: {}, + clientInfo: { name, version: "1" }, + initializePayload: { + protocolVersion: "2025-06-18", + capabilities: {}, + clientInfo: { name, version: "1" }, + }, + getClient: Effect.die("unused"), + }), + ), + ); + assert.notEqual(result.isError, true, JSON.stringify(result.content)); + yield* worker.drain(); + yield* orchestrator.streamStoredEventsFrom({ threadId, afterSequence: sequence }).pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + event.payload.id === before.runs[0]?.id && + ["completed", "interrupted", "failed", "cancelled"].includes(event.payload.status), + ), + Stream.take(1), + Stream.runDrain, + ); + yield* worker.drain(); + + const after = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual(after.thread.modelSelection, targetSelection); + assert.deepEqual( + after.runs.map((run) => [run.id, run.status]), + [[before.runs[0]!.id, "completed"]], + ); + assert.deepEqual( + (yield* Ref.get(state)).started.map((turn) => turn.model), + [initialSelection.model, targetSelection.model], + ); + assert.deepEqual(after.runs[0]?.modelSelection, targetSelection); + }).pipe( + Effect.provide( + McpHttpServer.layerThreadToolkit.pipe( + Layer.provideMerge(McpServer.McpServer.layer), + Layer.provide(NodeCrypto.layer), + Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})), + Layer.provide(Layer.mock(ScheduledTaskService.ScheduledTaskService)({})), + Layer.provideMerge(layerOrchestration), + ), + ), + ); + }), + ), +); + it.live("retries a failed selection restart session open before completing", () => Effect.scoped( Effect.gen(function* () { From de9ca3fd6aea401c100f745c2f5f1be376404308 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 20:35:35 +0000 Subject: [PATCH 5/6] test(server): reproduce selection restart accepted after archive (#15300) Co-Authored-By: Claude Opus 5.5 --- .../SelectionRestart.integration.test.ts | 114 ++++++++++++++++++ 1 file changed, 114 insertions(+) diff --git a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts index cba6da7301ee..96f7b1577994 100644 --- a/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts +++ b/apps/server/src/orchestration-v2/SelectionRestart.integration.test.ts @@ -18,11 +18,13 @@ import { ThreadId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Result from "effect/Result"; import * as Stream from "effect/Stream"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; @@ -92,6 +94,7 @@ function makeRestartAdapter( state: Ref.Ref, sessionCapabilities: OrchestrationV2ProviderCapabilities = pooledCapabilities, providerInstanceId = initialSelection.instanceId, + interruptGate?: Deferred.Deferred, ): ProviderAdapter.ProviderAdapterV2["Service"] { return { instanceId: providerInstanceId, @@ -265,6 +268,9 @@ function makeRestartAdapter( steerTurn: () => Effect.void, interruptTurn: () => Effect.gen(function* () { + if (interruptGate !== undefined) { + yield* Deferred.await(interruptGate); + } const active = (yield* Ref.get(state)).activeTurn; if (active !== null) { const updatedAt = yield* DateTime.now; @@ -461,6 +467,114 @@ it.live("t3_thread_configure from the running turn continues the user request", ), ); +// The configure precheck runs outside the thread lock, so an archive can +// commit between it and the restart dispatch. Archive leaves the run running +// until its detach effect interrupts the turn; that restart must be rejected. +it.live("rejects a selection restart dispatched after the thread is archived", () => + Effect.scoped( + Effect.gen(function* () { + const name = "selection-restart-archived"; + const cwd = yield* checkpointWorkspace(name); + const threadId = ThreadId.make(`thread:${name}`); + const state = yield* Ref.make({ + activeTurn: null, + opened: [], + started: [], + closedSessionCount: 0, + failedReplacementOpen: true, + }); + // Holds the archive detach before it can interrupt the running turn. + const interruptGate = yield* Deferred.make(); + const layerRegistry = ProviderAdapterRegistry.layerSingle( + makeRestartAdapter(state, pooledCapabilities, providerInstanceId, interruptGate), + ); + + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + yield* Effect.addFinalizer(() => Deferred.succeed(interruptGate, undefined)); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`${name}:create`), + threadId, + projectId: ProjectId.make(`project:${name}`), + title: name, + modelSelection: initialSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + const firstSequence = yield* orchestrator.getThreadEventSequence(threadId); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`${name}:first`), + threadId, + messageId: MessageId.make(`${name}:first`), + text: "first", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* orchestrator.streamStoredEventsFrom({ threadId, afterSequence: firstSequence }).pipe( + Stream.filter( + ({ event }) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ), + Stream.take(1), + Stream.runDrain, + ); + const activeRunId = (yield* orchestrator.getThreadProjection(threadId)).runs[0]?.id; + if (activeRunId === undefined) { + return yield* Effect.die("active archived restart test run is missing"); + } + + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make(`${name}:archive`), + threadId, + }); + const archived = yield* orchestrator.getThreadProjection(threadId); + assert.isNotNull(archived.thread.archivedAt); + assert.deepEqual( + archived.runs.map((run) => [run.id, run.status]), + [[activeRunId, "running"]], + ); + + const result = yield* orchestrator + .dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`${name}:configure`), + threadId, + messageId: MessageId.make(`${name}:configure:model-continuation`), + text: "continue", + attachments: [], + modelSelection: replacementSelection, + dispatchMode: { type: "restart_active", targetRunId: activeRunId }, + createdBy: "agent", + creationSource: "mcp", + }) + .pipe(Effect.result); + assert.equal( + Result.isFailure(result) ? result.failure._tag : "accepted", + "OrchestratorDispatchError", + ); + + const after = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + after.messages.map((message) => message.id), + archived.messages.map((message) => message.id), + ); + assert.lengthOf(after.attempts, archived.attempts.length); + }).pipe(Effect.provide(ProviderReplayHarness.layerWithRegistry({ name }, layerRegistry))); + }), + ), +); + it.live("retries a failed selection restart session open before completing", () => Effect.scoped( Effect.gen(function* () { From 1bf57c2c5edbfd533c29247422a80a45deb896ac Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 20:41:48 +0000 Subject: [PATCH 6/6] fix(server): reject steers and restarts into an archived thread (#15300) Archive commits before its asynchronous detach stops the running turn, so a restart validated against the unarchived thread could still write a continuation and new attempt. dispatchSteerIntoRun now rejects archived or deleted threads under the thread lock before any event or restart. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/Orchestrator.ts | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 8c8620088fa9..140fc66abefb 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -4108,6 +4108,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly delegatedCompletion?: OrchestrationV2ConversationMessage["delegatedCompletion"]; readonly forceRestart: boolean; }) { + // Archive commits before its asynchronous detach stops the run, so a steer + // or restart validated against the unarchived thread can still land here. + const thread = input.projection.thread; + if (thread.archivedAt !== null || thread.deletedAt !== null) { + return yield* new OrchestratorDispatchError({ + commandId: input.command.commandId, + commandType: input.command.type, + cause: `Thread ${input.command.threadId} is not active.`, + }); + } const targetRun = input.projection.runs.find((candidate) => candidate.id === input.targetRunId); if (targetRun === undefined) { return yield* new OrchestratorDispatchError({