From dba634ec4accc40984117fcb5c4d9717ca92f4ff Mon Sep 17 00:00:00 2001 From: Alex Chambers Date: Wed, 21 Oct 2026 14:25:20 +0200 Subject: [PATCH 1/5] fix(server): messages from other threads no longer reset a pending question When a Claude turn was waiting on AskUserQuestion, a t3_thread_send in auto mode steered into it. Claude aborted the question and asked it again, while T3 kept the old card pending, so the user's answers went nowhere and they had to retype them. Auto sends now queue behind a turn that is waiting on a live question or approval. When Claude still abandons a request, such as on an explicit Steer, the adapter marks it cancelled so the dead card closes. Fixes #15517 Co-Authored-By: Claude Opus 5.5 --- .../src/mcp/toolkits/orchestrator/tools.ts | 2 +- .../Adapters/ClaudeAdapterV2.test.ts | 111 ++++++++++++++ .../Adapters/ClaudeAdapterV2.ts | 42 ++++++ .../ThreadManagementService.test.ts | 138 ++++++++++++++++++ .../ThreadManagementService.ts | 30 +++- .../orchestrator-mcp-server.md | 2 +- 6 files changed, 321 insertions(+), 4 deletions(-) diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index 9c24fb62d8e93..2409d8b92df9b 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -234,7 +234,7 @@ const ThreadUpdateTool = Tool.make("t3_thread_update", { const ThreadSendTool = Tool.make("t3_thread_send", { description: - "Send a message to any T3 thread in this environment. The target cannot have broader permission modes than the caller. Use delegate_task with the full review context and a new clientRequestId for independent review rounds. For a follow-up that needs the same child's conversation, send to its childThreadId. After the original task completes, a requested follow-up creates a fresh task in that child thread and returns taskId when it starts a turn or restarts a turn without its own task; its completion notifies the parent, and the original result stays unchanged. A steer that joins a running turn returns that turn's task, if it has one. mode='auto' starts an idle thread, steers a fully active turn, or queues behind a turn that is not yet steerable. Use queue for a separate follow-up turn, steer for an in-flight update, or restart to interrupt-and-restart the active turn. clientRequestId makes retries idempotent.", + "Send a message to any T3 thread in this environment. The target cannot have broader permission modes than the caller. Use delegate_task with the full review context and a new clientRequestId for independent review rounds. For a follow-up that needs the same child's conversation, send to its childThreadId. After the original task completes, a requested follow-up creates a fresh task in that child thread and returns taskId when it starts a turn or restarts a turn without its own task; its completion notifies the parent, and the original result stays unchanged. A steer that joins a running turn returns that turn's task, if it has one. mode='auto' starts an idle thread, steers a fully active turn, or queues behind a turn that is not yet steerable or is waiting on the user's answer. Use queue for a separate follow-up turn, steer for an in-flight update, or restart to interrupt-and-restart the active turn. clientRequestId makes retries idempotent.", parameters: OrchestratorMcpThreadSendInput, success: OrchestratorMcpThreadSendResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index aeb128bf6febf..3ce0b2a0bb2af 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -1331,6 +1331,117 @@ describe("ClaudeAdapterV2 resume compaction", () => { ); }); +describe("ClaudeAdapterV2 abandoned questions", () => { + it.effect("closes a question Claude stopped waiting on", () => + Effect.scoped( + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const attachmentsDir = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-claude-abandoned-question-", + }); + let openedOptions: ClaudeAdapterV2.ClaudeAgentSdkQueryOptions | undefined; + const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({ + instanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID, + settings: DEFAULT_CLAUDE_SETTINGS, + environment: {}, + attachmentsDir, + fileSystem, + path: yield* Path.Path, + crypto: yield* Crypto.Crypto, + idAllocator, + queryRunner: { + allocateSessionId: Effect.succeed("native-thread-claude-abandoned-question"), + open: (input) => + Effect.sync(() => { + openedOptions = input.options; + return { + messages: Stream.never, + offer: () => Effect.void, + setModel: () => Effect.void, + applyFlagSettings: () => Effect.void, + setPermissionMode: () => Effect.void, + backgroundTasks: () => Effect.succeed(false), + stopTask: () => Effect.void, + interrupt: Effect.void, + close: Effect.void, + }; + }), + forkSession: () => Effect.die("unused"), + subagentLaunchToolUseId: () => Effect.succeed(null), + assertComplete: Effect.void, + }, + }); + const threadId = ThreadId.make("thread-claude-abandoned-question"); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make("provider-session-claude-abandoned-question"), + modelSelection: CLAUDE_TEST_MODEL_SELECTION, + runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection: CLAUDE_TEST_MODEL_SELECTION, + runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY, + }); + yield* runtime.startTurn( + makeClaudeTestTurnInput({ + threadId, + providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-claude-abandoned-question"), + text: "ask me", + attachments: [], + }), + ); + const canUseTool = openedOptions?.canUseTool; + assert.isFunction(canUseTool); + const controller = new AbortController(); + const requestUpdates = yield* runtime.events.pipe( + Stream.filter((event) => event.type === "runtime_request.updated"), + Stream.tap(() => Effect.sync(() => controller.abort())), + Stream.take(2), + Stream.runCollect, + Effect.forkScoped, + ); + const result = yield* Effect.promise(() => + canUseTool!( + "AskUserQuestion", + { + questions: [ + { + header: "Color", + question: "Which color?", + options: [ + { label: "Red", description: "Pick red." }, + { label: "Blue", description: "Pick blue." }, + ], + multiSelect: false, + }, + ], + }, + { + signal: controller.signal, + toolUseID: "tool-abandoned-question", + requestId: "request-abandoned-question", + }, + ), + ).pipe(Effect.forkScoped); + + assert.equal((yield* Fiber.join(result))?.behavior, "deny"); + const statuses = Array.from(yield* Fiber.join(requestUpdates), (event) => + event.type === "runtime_request.updated" ? event.runtimeRequest.status : null, + ); + assert.deepEqual(statuses, ["pending", "cancelled"]); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); +}); + describe("ClaudeAdapterV2 attachments", () => { it.effect.each([ { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 91e926cc85e91..043062a0561aa 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -6441,6 +6441,42 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return { node, request, turnItem }; }); + // A steer offered with priority "now" aborts a pending permission + // callback, and Claude asks again as a new request. Left pending, the + // old card takes answers Claude no longer waits for (#15517). + const cancelAbandonedRequest = Effect.fnUntraced(function* (artifacts: { + readonly node: OrchestrationV2ExecutionNode; + readonly request: OrchestrationV2RuntimeRequest; + readonly turnItem: OrchestrationV2TurnItem; + }) { + const now = yield* DateTime.now; + yield* Effect.all( + [ + emitProviderEvent({ + type: "runtime_request.updated", + driver: CLAUDE_PROVIDER, + runtimeRequest: { ...artifacts.request, status: "cancelled", resolvedAt: now }, + }), + emitProviderEvent({ + type: "node.updated", + driver: CLAUDE_PROVIDER, + node: { ...artifacts.node, status: "cancelled", completedAt: now }, + }), + emitProviderEvent({ + type: "turn_item.updated", + driver: CLAUDE_PROVIDER, + turnItem: { + ...artifacts.turnItem, + status: "cancelled", + completedAt: now, + updatedAt: now, + }, + }), + ], + { concurrency: 1 }, + ); + }); + const ensureReasoningBlock = Effect.fnUntraced(function* ( context: ActiveClaudeTurnContext, itemId: string, @@ -8851,6 +8887,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }), ), ); + if (callbackOptions.signal.aborted && !(yield* Deferred.isDone(answers))) { + yield* cancelAbandonedRequest(artifacts); + } return callbackOptions.signal.aborted ? ({ behavior: "deny", @@ -8963,6 +9002,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }), ), ); + if (callbackOptions.signal.aborted && !(yield* Deferred.isDone(decision))) { + yield* cancelAbandonedRequest(artifacts); + } return permissionResultFromDecision({ toolName, diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 8d1f613180a3a..cd3dbfd252ce3 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -1,15 +1,21 @@ import { expect, it } from "@effect/vitest"; import { CommandId, + EventId, MessageId, type OrchestrationV2Run, type OrchestrationV2StoredEvent, type OrchestrationV2ThreadProjection, ProjectId, + ProviderDriverKind, ProviderInstanceId, + ProviderSessionId, + ProviderTurnId, RunId, + RuntimeRequestId, ThreadId, } from "@t3tools/contracts"; +import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as Deferred from "effect/Deferred"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -20,10 +26,16 @@ import * as Stream from "effect/Stream"; import * as ServerSettings from "../serverSettings.ts"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as CommandReceiptStore from "./CommandReceiptStore.ts"; +import * as EffectOutbox from "./EffectOutbox.ts"; +import * as EventStore from "./EventStore.ts"; import * as Orchestrator from "./Orchestrator.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; +import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; const layerReceiptsTest = CommandReceiptStore.layer.pipe( Layer.provide(SqlitePersistence.layerMemory), @@ -337,3 +349,129 @@ it.effect.each([ expect(dispatched).toEqual(settles ? ["thread.settle"] : []); }), ); + +it.effect("an auto send queues behind a turn waiting on the user's answer", () => { + const projectId = ProjectId.make("project:thread-management:auto-send"); + const threadId = ThreadId.make("thread:thread-management:auto-send"); + const instanceId = ProviderInstanceId.make("codex"); + const layerDatabase = SqlitePersistence.layerMemory; + const adapter = { + instanceId, + driver: ProviderDriverKind.make("codex"), + getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }), + openSession: () => Effect.die("Runs here never reach a provider"), + } as ProviderAdapter.ProviderAdapterV2["Service"]; + const layerTest = ThreadManagementService.layer.pipe( + Layer.provideMerge( + Layer.mergeAll( + layerDatabase, + ProjectionStore.layer.pipe(Layer.provide(layerDatabase)), + EventStore.layer.pipe(Layer.provide(layerDatabase)), + EffectOutbox.layer.pipe(Layer.provide(layerDatabase)), + ProviderReplayHarness.layerWithRegistry( + { name: "auto-send" }, + ProviderAdapterRegistry.layerFromAdapters([adapter]), + { databaseLayer: layerDatabase, runEffectWorker: false }, + ), + ), + ), + Layer.provideMerge(ServerSettings.layerTest()), + Layer.provideMerge(CommandReceiptStore.layer.pipe(Layer.provide(layerDatabase))), + ); + + return Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threads = yield* ThreadManagementService.ThreadManagementService; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("command:auto-send:create"), + threadId, + projectId, + title: "Auto send", + modelSelection: { instanceId, model: "gpt-5.1-codex" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:auto-send:first"), + threadId, + messageId: MessageId.make("message:auto-send:first"), + text: "Ask me a question", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const now = yield* DateTime.now; + const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; + const providerTurnId = ProviderTurnId.make("provider-turn:auto-send"); + yield* projections.apply({ + id: EventId.make("event:auto-send:running"), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status: "running", startedAt: now }, + }); + yield* projections.apply({ + id: EventId.make("event:auto-send:turn"), + type: "provider-turn.updated", + threadId, + occurredAt: now, + payload: { + id: providerTurnId, + providerThreadId: run.providerThreadId!, + nodeId: run.rootNodeId!, + runAttemptId: run.activeAttemptId!, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + yield* projections.apply({ + id: EventId.make("event:auto-send:question"), + type: "runtime-request.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { + id: RuntimeRequestId.make("request:auto-send"), + nodeId: run.rootNodeId!, + providerTurnId, + nativeRequestRef: null, + kind: "user_input", + status: "pending", + responseCapability: { + type: "live", + providerSessionId: ProviderSessionId.make("provider-session:auto-send"), + }, + createdAt: now, + resolvedAt: null, + }, + }); + + const sent = yield* threads.sendToThread({ + projectId, + commandId: CommandId.make("command:auto-send:other-thread"), + threadId, + messageId: MessageId.make("message:auto-send:other-thread"), + text: "A message from another thread", + attachments: [], + mode: "auto", + createdBy: "agent", + creationSource: "mcp", + }); + + expect(sent.delivery).toBe("queued"); + expect(sent.run.status).toBe("queued"); + }).pipe(Effect.provide(layerTest)); +}); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 543b3e2f89fe1..b36d39a2623af 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -504,6 +504,26 @@ function latestSteerableRun( .toSorted((left, right) => right.ordinal - left.ordinal)[0]; } +// Steering into a turn blocked on the user's answer can abort the question, +// and the provider asks it again with the user's typed answers gone (#15517). +function runAwaitsUser( + projection: Pick, + run: OrchestrationV2Run, +): boolean { + const turnIds = new Set( + projection.providerTurns + .filter((turn) => turn.runAttemptId === run.activeAttemptId) + .map((turn) => turn.id), + ); + return projection.runtimeRequests.some( + (request) => + request.status === "pending" && + request.responseCapability.type === "live" && + request.providerTurnId !== null && + turnIds.has(request.providerTurnId), + ); +} + const SETTLE_AFTER_RUN_WAIT_MS = 24 * 60 * 60 * 1_000; const make = Effect.gen(function* () { @@ -751,7 +771,11 @@ const make = Effect.gen(function* () { const sendToThread: ThreadManagementServiceShape["sendToThread"] = Effect.fn( "ThreadManagementService.sendToThread", )(function* (input) { - const target = yield* getProjectThreadRecords(input, ["runs", "providerTurns"]); + const target = yield* getProjectThreadRecords(input, [ + "runs", + "providerTurns", + "runtimeRequests", + ]); if (target.thread.archivedAt !== null) { return yield* new ThreadManagementThreadArchivedError({ threadId: input.threadId, @@ -774,7 +798,9 @@ const make = Effect.gen(function* () { targetRunId: steerableRun.id, }; } else if (input.mode === "auto" && steerableRun !== undefined) { - dispatchMode = { type: "steer_active", targetRunId: steerableRun.id }; + dispatchMode = runAwaitsUser(target, steerableRun) + ? { type: "queue_after_active" } + : { type: "steer_active", targetRunId: steerableRun.id }; } else { dispatchMode = { type: input.mode === "queue" ? "queue_after_active" : "start_immediately", diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 44aa2492a8282..c473d447e3759 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -413,7 +413,7 @@ detail also exposes an in-flight title regeneration. Sends a message to any ordinary or delegated thread in the environment: - `auto` starts an idle thread, steers a fully active turn, or queues behind a - turn that is not yet steerable; + turn that is not yet steerable or is waiting on the user's answer; - `queue` creates a separate follow-up run after active work; - `steer` requires a steerable active provider turn; and - `restart` requires an active provider turn and uses the orchestrator's From 5ac202ae34df2607bced074176920c762a5203c1 Mon Sep 17 00:00:00 2001 From: Alex Chambers Date: Mon, 26 Oct 2026 09:36:18 +0100 Subject: [PATCH 2/5] fix(server): resolve merge with main and satisfy new lint rules Drop duplicate imports left by the auto-merge, remove issue numbers from comments, and drive the abandoned-question test's abort through Effect.abortSignal instead of a hand-built AbortController. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts | 7 ++++--- .../src/orchestration-v2/Adapters/ClaudeAdapterV2.ts | 2 +- .../src/orchestration-v2/ThreadManagementService.test.ts | 3 --- .../server/src/orchestration-v2/ThreadManagementService.ts | 2 +- 4 files changed, 6 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 9898372236686..0c88b79bf072b 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -1470,10 +1470,11 @@ describe("ClaudeAdapterV2 abandoned questions", () => { ); const canUseTool = openedOptions?.canUseTool; assert.isFunction(canUseTool); - const controller = new AbortController(); + const abortScope = yield* Scope.make(); + const signal = yield* Effect.abortSignal.pipe(Scope.provide(abortScope)); const requestUpdates = yield* runtime.events.pipe( Stream.filter((event) => event.type === "runtime_request.updated"), - Stream.tap(() => Effect.sync(() => controller.abort())), + Stream.tap(() => Scope.close(abortScope, Exit.void)), Stream.take(2), Stream.runCollect, Effect.forkScoped, @@ -1495,7 +1496,7 @@ describe("ClaudeAdapterV2 abandoned questions", () => { ], }, { - signal: controller.signal, + signal, toolUseID: "tool-abandoned-question", requestId: "request-abandoned-question", }, diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 26cb015e6a904..9315f92e55a60 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -6914,7 +6914,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( // A steer offered with priority "now" aborts a pending permission // callback, and Claude asks again as a new request. Left pending, the - // old card takes answers Claude no longer waits for (#15517). + // old card takes answers Claude no longer waits for. const cancelAbandonedRequest = Effect.fnUntraced(function* (artifacts: { readonly node: OrchestrationV2ExecutionNode; readonly request: OrchestrationV2RuntimeRequest; diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 328def0b8c149..9273c1d9997e3 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -11,15 +11,12 @@ import { type OrchestrationV2StoredEvent, type OrchestrationV2ThreadProjection, ProjectId, - ProviderDriverKind, ProviderInstanceId, - ProviderSessionId, ProviderTurnId, RunId, RuntimeRequestId, ThreadId, } from "@t3tools/contracts"; -import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as Deferred from "effect/Deferred"; import * as DateTime from "effect/DateTime"; import * as Option from "effect/Option"; diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 96dbc53fc61fa..cb7b142bea012 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -527,7 +527,7 @@ function latestSteerableRun( } // Steering into a turn blocked on the user's answer can abort the question, -// and the provider asks it again with the user's typed answers gone (#15517). +// and the provider asks it again with the user's typed answers gone. function runAwaitsUser( projection: Pick, run: OrchestrationV2Run, From 8f8b437aa4853f7f21cabe9b4d95f6c9a440082f Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 1 Nov 2026 04:52:05 +0000 Subject: [PATCH 3/5] fix(server): preserve answered requests when callbacks abort --- .../Adapters/ClaudeAdapterV2.test.ts | 34 +++++- apps/server/src/orchestration-v2/EventSink.ts | 62 ++++++---- .../ProviderEventIngestor.test.ts | 109 ++++++++++++------ 3 files changed, 143 insertions(+), 62 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index cb06ba8205fa2..71fadc818f777 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -1719,9 +1719,20 @@ describe("ClaudeAdapterV2 abandoned questions", () => { const abortScope = yield* Scope.make(); const signal = yield* Effect.abortSignal.pipe(Scope.provide(abortScope)); const requestUpdates = yield* runtime.events.pipe( - Stream.filter((event) => event.type === "runtime_request.updated"), - Stream.tap(() => Scope.close(abortScope, Exit.void)), - Stream.take(2), + Stream.filter( + (event) => + event.type === "runtime_request.updated" || + event.type === "node.updated" || + event.type === "turn_item.updated", + ), + Stream.tap((event) => + event.type === "runtime_request.updated" + ? Scope.close(abortScope, Exit.void) + : Effect.void, + ), + Stream.takeUntil( + (event) => event.type === "turn_item.updated" && event.turnItem.status === "cancelled", + ), Stream.runCollect, Effect.forkScoped, ); @@ -1750,10 +1761,21 @@ describe("ClaudeAdapterV2 abandoned questions", () => { ).pipe(Effect.forkScoped); assert.equal((yield* Fiber.join(result))?.behavior, "deny"); - const statuses = Array.from(yield* Fiber.join(requestUpdates), (event) => - event.type === "runtime_request.updated" ? event.runtimeRequest.status : null, - ); + const updates = Array.from(yield* Fiber.join(requestUpdates)); + const statuses = updates + .filter((event) => event.type === "runtime_request.updated") + .map((event) => + event.type === "runtime_request.updated" ? event.runtimeRequest.status : null, + ); assert.deepEqual(statuses, ["pending", "cancelled"]); + assert.equal( + updates.findLast((event) => event.type === "node.updated")?.node.status, + "cancelled", + ); + assert.equal( + updates.findLast((event) => event.type === "turn_item.updated")?.turnItem.status, + "cancelled", + ); }).pipe( Effect.provide( Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index b210d3c9840ae..a896b6c68e38a 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -291,11 +291,7 @@ const layerBase: Layer.Layer< const staleRequests = new Set(); const staleNodes = new Set(); for (const event of events) { - if ( - event.type !== "runtime-request.updated" || - event.payload.kind !== "user_input" || - event.payload.status !== "cancelled" - ) + if (event.type !== "runtime-request.updated" || event.payload.status !== "cancelled") continue; const current = yield* projectionStore.getRuntimeRequest( event.threadId, @@ -303,7 +299,7 @@ const layerBase: Layer.Layer< ); if ( current?.status !== "pending" || - current.kind !== "user_input" || + current.kind !== event.payload.kind || current.providerTurnId !== event.payload.providerTurnId || current.responseCapability.type === "message" ) { @@ -311,22 +307,44 @@ const layerBase: Layer.Layer< staleNodes.add(event.payload.nodeId); } } - return events.filter((event) => { - switch (event.type) { - case "runtime-request.updated": - return event.payload.status !== "cancelled" || !staleRequests.has(event.payload.id); - case "node.updated": - return event.payload.status !== "cancelled" || !staleNodes.has(event.payload.id); - case "turn-item.updated": - return ( - event.payload.type !== "user_input_request" || - event.payload.status !== "cancelled" || - !staleRequests.has(event.payload.requestId) - ); - default: - return true; - } - }); + return yield* Effect.filter( + events, + Effect.fnUntraced(function* (event) { + const requestId = + event.type === "node.updated" && + (event.payload.kind === "user_input_request" || + event.payload.kind === "approval_request") + ? event.payload.runtimeRequestId + : event.type === "turn-item.updated" && + (event.payload.type === "user_input_request" || + event.payload.type === "approval_request") + ? event.payload.requestId + : null; + if ( + requestId !== null && + (event.type === "node.updated" || event.type === "turn-item.updated") && + event.payload.status === "cancelled" + ) { + const current = yield* projectionStore.getRuntimeRequest(event.threadId, requestId); + if (current?.status === "resolved") return false; + } + switch (event.type) { + case "runtime-request.updated": + return event.payload.status !== "cancelled" || !staleRequests.has(event.payload.id); + case "node.updated": + return event.payload.status !== "cancelled" || !staleNodes.has(event.payload.id); + case "turn-item.updated": + return ( + (event.payload.type !== "user_input_request" && + event.payload.type !== "approval_request") || + event.payload.status !== "cancelled" || + !staleRequests.has(event.payload.requestId) + ); + default: + return true; + } + }), + ); }); const normalizeEvents = (events: ReadonlyArray) => { diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index 7f1006f1c5ec5..6ec36983df601 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -1309,9 +1309,13 @@ layer("ProviderEventIngestorV2", (it) => { }), ); - it.effect( - "preserves an answer committed after terminal normalization reads a pending question", - () => + it.effect.each([ + { kind: "user_input", source: "terminal" }, + { kind: "user_input", source: "callback" }, + { kind: "command", source: "callback" }, + ] as const)( + "preserves a $kind response committed before a $source cancellation", + ({ kind, source }) => Effect.scoped( Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; @@ -1339,7 +1343,7 @@ layer("ProviderEventIngestorV2", (it) => { nodeId, providerTurnId, nativeRequestRef: null, - kind: "user_input", + kind, status: "pending", responseCapability: { type: "live", providerSessionId }, createdAt: now, @@ -1351,7 +1355,7 @@ layer("ProviderEventIngestorV2", (it) => { runId: null, parentNodeId: null, rootNodeId: nodeId, - kind: "user_input_request", + kind: kind === "command" ? "approval_request" : "user_input_request", status: "waiting", countsForRun: false, providerThreadId, @@ -1377,9 +1381,9 @@ layer("ProviderEventIngestorV2", (it) => { startedAt: now, completedAt: null, updatedAt: now, - type: "user_input_request", - requestId: request.id, - questions: [], + ...(kind === "command" + ? { type: "approval_request", requestId: request.id, requestKind: "command" } + : { type: "user_input_request", requestId: request.id, questions: [] }), }; const seedEvents: Array = [threadEvent]; for (const payload of [ @@ -1408,35 +1412,71 @@ layer("ProviderEventIngestorV2", (it) => { Effect.provide(Layer.fresh(ProviderEventIngestor.layer)), Effect.provideService(EventSink.EventSinkV2, gatedSink), ); - const terminal = yield* ingestor - .ingestNormalized({ - providerSessionId, - providerInstanceId: modelSelection.instanceId, - threadId, - event: { - type: "provider_turn.updated", - driver: CODEX_DRIVER, - providerTurn: { - id: providerTurnId, - providerThreadId, - nodeId, - runAttemptId: null, - nativeTurnRef: null, - ordinal: 1, - status: "completed", - startedAt: now, - completedAt: now, - }, - }, - }) - .pipe(Effect.forkScoped); + const terminal = yield* ( + source === "terminal" + ? ingestor.ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId, + event: { + type: "provider_turn.updated", + driver: CODEX_DRIVER, + providerTurn: { + id: providerTurnId, + providerThreadId, + nodeId, + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status: "completed", + startedAt: now, + completedAt: now, + }, + }, + }) + : Effect.forEach( + [ + { + type: "runtime_request.updated" as const, + driver: CODEX_DRIVER, + runtimeRequest: { ...request, status: "cancelled" as const, resolvedAt: now }, + }, + { + type: "node.updated" as const, + driver: CODEX_DRIVER, + node: { ...node, status: "cancelled" as const, completedAt: now }, + }, + { + type: "turn_item.updated" as const, + driver: CODEX_DRIVER, + turnItem: { + ...item, + status: "cancelled" as const, + completedAt: now, + updatedAt: now, + }, + }, + ], + (event) => + ingestor.ingestNormalized({ + providerSessionId, + providerInstanceId: modelSelection.instanceId, + threadId, + event, + }), + { concurrency: 1 }, + ).pipe(Effect.map((batches) => batches.flat())) + ).pipe(Effect.forkScoped); yield* Deferred.await(normalized); - const answers = { decision: "Use the existing workspace" }; + const response = + kind === "command" + ? { decision: "accept" as const } + : { answers: { decision: "Use the existing workspace" } }; const responseEvents: Array = []; for (const payload of [ { type: "runtime-request.updated" as const, - payload: { ...request, status: "resolved" as const, answers, resolvedAt: now }, + payload: { ...request, status: "resolved" as const, ...response, resolvedAt: now }, }, { type: "node.updated" as const, @@ -1458,12 +1498,13 @@ layer("ProviderEventIngestorV2", (it) => { const committedTerminal = yield* Fiber.join(terminal); assert.deepEqual( committedTerminal.map((stored) => stored.event.type), - ["provider-turn.updated"], + source === "terminal" ? ["provider-turn.updated"] : [], ); const projection = yield* projections.getThreadProjection(threadId); const answered = projection.runtimeRequests.find((entry) => entry.id === request.id)!; assert.equal(answered.status, "resolved"); - assert.deepEqual(answered.answers, answers); + assert.deepEqual(answered.answers, "answers" in response ? response.answers : undefined); + assert.equal(answered.decision, "decision" in response ? response.decision : undefined); assert.equal(projection.nodes.find((entry) => entry.id === nodeId)!.status, "completed"); assert.equal( projection.turnItems.find((entry) => entry.id === item.id)!.status, From 9c2ca68cbd142b826c73adf513c189ad2014ec9d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 21:26:15 +0000 Subject: [PATCH 4/5] test(server): drop Claude abandoned-question test main already covers Main's withdrawn-request table test exercises AskUserQuestion cancellation. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 134 ------------------ 1 file changed, 134 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index a1049f03b0526..043f3b8a756d9 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -2096,140 +2096,6 @@ describe("ClaudeAdapterV2 resume compaction", () => { ); }); -describe("ClaudeAdapterV2 abandoned questions", () => { - it.effect("closes a question Claude stopped waiting on", () => - Effect.scoped( - Effect.gen(function* () { - const fileSystem = yield* FileSystem.FileSystem; - const idAllocator = yield* IdAllocator.IdAllocatorV2; - const attachmentsDir = yield* fileSystem.makeTempDirectoryScoped({ - prefix: "t3-claude-abandoned-question-", - }); - let openedOptions: ClaudeAdapterV2.ClaudeAgentSdkQueryOptions | undefined; - const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({ - instanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID, - settings: DEFAULT_CLAUDE_SETTINGS, - environment: {}, - attachmentsDir, - fileSystem, - path: yield* Path.Path, - crypto: yield* Crypto.Crypto, - idAllocator, - queryRunner: { - allocateSessionId: Effect.succeed("native-thread-claude-abandoned-question"), - open: (input) => - Effect.sync(() => { - openedOptions = input.options; - return { - messages: Stream.never, - offer: () => Effect.void, - setModel: () => Effect.void, - applyFlagSettings: () => Effect.void, - setPermissionMode: () => Effect.void, - backgroundTasks: () => Effect.succeed(false), - stopTask: () => Effect.void, - interrupt: Effect.void, - close: Effect.void, - }; - }), - forkSession: () => Effect.die("unused"), - subagentLaunchToolUseId: () => Effect.succeed(null), - assertComplete: Effect.void, - }, - }); - const threadId = ThreadId.make("thread-claude-abandoned-question"); - const runtime = yield* adapter.openSession({ - threadId, - providerSessionId: ProviderSessionId.make("provider-session-claude-abandoned-question"), - modelSelection: CLAUDE_TEST_MODEL_SELECTION, - runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY, - }); - const providerThread = yield* runtime.ensureThread({ - threadId, - modelSelection: CLAUDE_TEST_MODEL_SELECTION, - runtimePolicy: CLAUDE_TEST_RUNTIME_POLICY, - }); - yield* runtime.startTurn( - makeClaudeTestTurnInput({ - threadId, - providerThread, - now: yield* DateTime.now, - attemptId: RunAttemptId.make("attempt-claude-abandoned-question"), - text: "ask me", - attachments: [], - }), - ); - const canUseTool = openedOptions?.canUseTool; - assert.isFunction(canUseTool); - const abortScope = yield* Scope.make(); - const signal = yield* Effect.abortSignal.pipe(Scope.provide(abortScope)); - const requestUpdates = yield* runtime.events.pipe( - Stream.filter( - (event) => - event.type === "runtime_request.updated" || - event.type === "node.updated" || - event.type === "turn_item.updated", - ), - Stream.tap((event) => - event.type === "runtime_request.updated" - ? Scope.close(abortScope, Exit.void) - : Effect.void, - ), - Stream.takeUntil( - (event) => event.type === "turn_item.updated" && event.turnItem.status === "cancelled", - ), - Stream.runCollect, - Effect.forkScoped, - ); - const result = yield* Effect.promise(() => - canUseTool!( - "AskUserQuestion", - { - questions: [ - { - header: "Color", - question: "Which color?", - options: [ - { label: "Red", description: "Pick red." }, - { label: "Blue", description: "Pick blue." }, - ], - multiSelect: false, - }, - ], - }, - { - signal, - toolUseID: "tool-abandoned-question", - requestId: "request-abandoned-question", - }, - ), - ).pipe(Effect.forkScoped); - - assert.equal((yield* Fiber.join(result))?.behavior, "deny"); - const updates = Array.from(yield* Fiber.join(requestUpdates)); - const statuses = updates - .filter((event) => event.type === "runtime_request.updated") - .map((event) => - event.type === "runtime_request.updated" ? event.runtimeRequest.status : null, - ); - assert.deepEqual(statuses, ["pending", "cancelled"]); - assert.equal( - updates.findLast((event) => event.type === "node.updated")?.node.status, - "cancelled", - ); - assert.equal( - updates.findLast((event) => event.type === "turn_item.updated")?.turnItem.status, - "cancelled", - ); - }).pipe( - Effect.provide( - Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), - ), - ), - ), - ); -}); - describe("ClaudeAdapterV2 attachments", () => { it.effect.each([ ...[ From 50a74324d7e6ad58e3d1fb3b655a3e21166591dc Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 21:31:12 +0000 Subject: [PATCH 5/5] fix(server): decide queue for auto sends before the orchestrator steers them Main resolves an auto delivery intent in the orchestrator, which ignored the queue chosen for a turn waiting on the user. Send those as a plain queue, check every active run, and give the test a steerable provider session so it fails without the fix. Co-Authored-By: Claude Opus 5.5 --- .../ThreadManagementService.test.ts | 36 ++++++++++++++++--- .../ThreadManagementService.ts | 8 ++--- 2 files changed, 35 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index ec3e673109db7..2bfd5e3ff2c14 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -276,6 +276,7 @@ it.effect.each([ }, ] : [], + runtimeRequests: [], messages: [], turnItems: [], } as unknown as OrchestrationV2ThreadProjection), @@ -902,8 +903,35 @@ it.effect("an auto send queues behind a turn waiting on the user's answer", () = creationSource: "web", }); const now = yield* DateTime.now; - const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; + const initial = yield* orchestrator.getThreadProjection(threadId); + const run = initial.runs[0]!; const providerTurnId = ProviderTurnId.make("provider-turn:auto-send"); + const providerSessionId = ProviderSessionId.make("provider-session:auto-send"); + yield* projections.apply({ + id: EventId.make("event:auto-send:session"), + type: "provider-session.attached", + threadId, + occurredAt: now, + payload: { + id: providerSessionId, + driver: adapter.driver, + providerInstanceId: instanceId, + status: "ready", + cwd: "/repo", + model: "gpt-5.1-codex", + capabilities: CodexProviderCapabilitiesV2, + createdAt: now, + updatedAt: now, + lastError: null, + }, + }); + yield* projections.apply({ + id: EventId.make("event:auto-send:provider-thread"), + type: "provider-thread.updated", + threadId, + occurredAt: now, + payload: { ...initial.providerThreads[0]!, providerSessionId }, + }); yield* projections.apply({ id: EventId.make("event:auto-send:running"), type: "run.updated", @@ -942,10 +970,7 @@ it.effect("an auto send queues behind a turn waiting on the user's answer", () = nativeRequestRef: null, kind: "user_input", status: "pending", - responseCapability: { - type: "live", - providerSessionId: ProviderSessionId.make("provider-session:auto-send"), - }, + responseCapability: { type: "live", providerSessionId }, createdAt: now, resolvedAt: null, }, @@ -967,6 +992,7 @@ it.effect("an auto send queues behind a turn waiting on the user's answer", () = expect(sent.run.status).toBe("queued"); }).pipe(Effect.provide(layerTest)); }); + it.effect.each([ { name: "kills the thread's own agent and ends its turn", diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 3b50a9b203b2f..3fb202c9bb36a 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -604,6 +604,7 @@ function runAwaitsUser( projection: Pick, run: OrchestrationV2Run, ): boolean { + if (!isActiveRun(run)) return false; const turnIds = new Set( projection.providerTurns .filter((turn) => turn.runAttemptId === run.activeAttemptId) @@ -899,7 +900,8 @@ const make = Effect.gen(function* () { const steerableRun = latestSteerableRun(target); // A fresh send never joins a running turn, so it needs none to steer or restart. const mode = - input.freshContext && (input.mode === "steer" || input.mode === "restart") + (input.freshContext && (input.mode === "steer" || input.mode === "restart")) || + (input.mode === "auto" && target.runs.some((run) => runAwaitsUser(target, run))) ? "queue" : input.mode; let dispatchMode: Extract< @@ -918,9 +920,7 @@ const make = Effect.gen(function* () { targetRunId: steerableRun.id, }; } else if (mode === "auto" && steerableRun !== undefined) { - dispatchMode = runAwaitsUser(target, steerableRun) - ? { type: "queue_after_active" } - : { type: "steer_active", targetRunId: steerableRun.id }; + dispatchMode = { type: "steer_active", targetRunId: steerableRun.id }; } else { dispatchMode = { type: mode === "queue" ? "queue_after_active" : "start_immediately",