From dd7fdb7cb0f781128792082d852513570aab11b4 Mon Sep 17 00:00:00 2001 From: Daniel Armstrong Date: Sat, 14 Nov 2026 18:45:43 +1100 Subject: [PATCH 1/4] fix(server): Stop reaches background work only the provider sees A switch refused over background work tells the user to press Stop, but when that work is visible only to the adapter probe (Claude native subagents, held wakes, Codex settled-turn work) the thread shows no Stop control and thread.stop never reached the provider. Stop now probes the latest run's live session and interrupts it, and the thread error banner offers Stop on that refusal. Fixes #36450 Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 46 +++++++++++++++---- .../ProviderSwitch.integration.test.ts | 29 ++++++++++-- apps/web/src/components/ChatView.tsx | 1 + .../src/components/chat/ChatBannerLane.tsx | 11 +++++ .../src/components/chat/ThreadErrorBanner.tsx | 12 ++++- .../orchestrationV2PendingBackgroundWork.ts | 5 ++ 6 files changed, 90 insertions(+), 14 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 0140b7b4dc071..5137d258063d1 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1684,6 +1684,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return false; }); + // Some background work (Claude native subagents, held wakes) only the adapter probe sees. + // Stop probes it too, or an idle thread cannot stop the work a switch is refused over. + const runHasLiveBackgroundWork = Effect.fn("Orchestrator.runHasLiveBackgroundWork")(function* ( + run: OrchestrationV2Run | undefined, + projection: Pick, + ) { + const providerThread = projection.providerThreads.find( + (candidate) => candidate.id === run?.providerThreadId, + ); + const session = projection.providerSessions.find( + (candidate) => candidate.id === providerThread?.providerSessionId, + ); + if (session === undefined || session.status === "stopped" || session.status === "error") + return false; + return yield* sessionHasBackgroundWork( + session, + providerThread === undefined ? [] : [providerThread], + ); + }); + // Runs in every command transaction; avoid reading thousands of finished leaf nodes. const readCommandProjection = ( threadId: ThreadId, @@ -12852,6 +12872,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio "turnItems", "subagents", "runtimeRequests", + "providerSessions", ], // Thread-wide, like the Waiting strip: a settled thread's background // work can belong to an earlier run than the one Stop targets. @@ -12877,15 +12898,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio run?.providerThreadId === null ? undefined : projection.providerThreads.find((candidate) => candidate.id === run?.providerThreadId); + const isLatestRun = + run?.id === projection.runs.findLast((candidate) => candidate.status !== "queued")?.id; const hasBackgroundWork = - run?.id === projection.runs.findLast((candidate) => candidate.status !== "queued")?.id && - derivePendingBackgroundWork({ + isLatestRun && + (derivePendingBackgroundWork({ latestRun: run, providerThreads: projection.providerThreads, turnItems: projection.turnItems, activeProviderThreadId: projection.thread.activeProviderThreadId, runs: projection.runs, - }).length > 0; + }).length > 0 || + (run !== undefined && + !isActiveRun(run) && + (yield* runHasLiveBackgroundWork(run, projection)))); // A failed start has no provider turn. Background work still belongs // to the provider thread, so Stop reaches its latest accepted turn. const providerTurn = @@ -13464,10 +13490,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio effects: Ref.Ref>, ) { const projection = yield* projectionStore - .getThreadRecords(command.threadId, ["runs", "providerThreads", "turnItems", "subagents"], { - turnItemTypes: ["command_execution", "dynamic_tool", "subagent"], - turnItemStatuses: ["pending", "running", "waiting"], - }) + .getThreadRecords( + command.threadId, + ["runs", "providerThreads", "turnItems", "subagents", "providerSessions"], + { + turnItemTypes: ["command_execution", "dynamic_tool", "subagent"], + turnItemStatuses: ["pending", "running", "waiting"], + }, + ) .pipe( Effect.mapError( (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }), @@ -13494,7 +13524,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio turnItems: projection.turnItems, activeProviderThreadId: projection.thread.activeProviderThreadId, runs: projection.runs, - }).length > 0 + }).length > 0 || (yield* runHasLiveBackgroundWork(latestRun, projection)) ? latestRun : undefined); if (target !== undefined) { diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index 965e5bd44f2dc..9bdf748295a10 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -24,7 +24,10 @@ import { TurnItemId, ProviderDriverKind, } from "@t3tools/contracts"; -import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; +import { + derivePendingBackgroundWork, + isBackgroundWorkRefusal, +} from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; @@ -444,7 +447,10 @@ function makeTestAdapter(input: { } }), steerTurn: () => Effect.void, - interruptTurn: () => Effect.void, + interruptTurn: (interrupt) => + interrupt.requestRuntimeRestart === true && input.backgroundWork !== undefined + ? Ref.set(input.backgroundWork, false) + : Effect.void, respondToRuntimeRequest: () => Effect.void, readThreadSnapshot: () => unimplemented(input.driver, "readThreadSnapshot unused in provider switch test"), @@ -2801,6 +2807,10 @@ describe("orchestration v2 provider switching", () => { "refuses the hand-off while the previous instance still runs background work", { background: true, queued: false, sharedHistory: false }, ], + [ + "lets Stop release an idle session whose projection has no background work", + { background: true, queued: false, sharedHistory: false, stopAfterRefusal: true }, + ], [ "keeps a busy previous session running when a queued hand-off starts", { background: true, queued: true, sharedHistory: false }, @@ -2809,7 +2819,7 @@ describe("orchestration v2 provider switching", () => { "holds a queued account switch while the replaced session still runs background work", { background: true, queued: true, sharedHistory: true }, ], - ] as const)("%s", ([, { background, queued, sharedHistory }]) => + ] as const)("%s", ([, { background, queued, sharedHistory, ...options }]) => Effect.gen(function* () { const cwd = yield* checkpointWorkspace("handoff-releases-session"); const capturedTurns = yield* Ref.make>([]); @@ -2975,7 +2985,7 @@ describe("orchestration v2 provider switching", () => { // pendingBackgroundTasks stays empty while the probe still reports work. assert.deepEqual(afterFirst.providerThreads[0]?.pendingBackgroundTasks ?? [], []); const refused = yield* send(2, secondSelection, ":refused").pipe(Effect.flip); - assert.include(String(refused.cause), "still running background agents or commands"); + assert.isTrue(isBackgroundWorkRefusal(String(refused.cause))); const unchanged = yield* orchestrator.getThreadProjection(handoffThreadId); assert.lengthOf(unchanged.runs, 1); assert.deepEqual(yield* Ref.get(closedSessions), []); @@ -2984,7 +2994,16 @@ describe("orchestration v2 provider switching", () => { unchanged.providerSessions.map((session) => [session.id, session.status]), [[firstSession!.id, "ready"]], ); - yield* Ref.set(backgroundWork, false); + if ("stopAfterRefusal" in options) { + yield* orchestrator.dispatch({ + type: "thread.stop", + commandId: CommandId.make("command:handoff-releases-session:stop"), + threadId: handoffThreadId, + }); + yield* worker.drain(); + } else { + yield* Ref.set(backgroundWork, false); + } } yield* send(2, secondSelection); diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index abc4cbdbf27a8..110e619758a0f 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -3409,6 +3409,7 @@ export default function ChatView(props: ChatViewProps) { threadErrorBannerKey={threadErrorBannerKey} setThreadError={setThreadError} setThreadErrorBannerDismissTick={setThreadErrorBannerDismissTick} + onStop={canOperateThread ? onInterruptStable : undefined} />
void; setThreadErrorBannerDismissTick: Dispatch>; + onStop: (() => void) | undefined; }) { const { providerStatus, @@ -40,6 +41,7 @@ export function ChatBannerLane(props: { threadErrorBannerKey, setThreadError, setThreadErrorBannerDismissTick, + onStop, } = props; return (
@@ -53,6 +55,15 @@ export function ChatBannerLane(props: { error={threadError} errorClass={threadErrorClass} errorCode={threadErrorCode} + onStop={ + onStop === undefined + ? undefined + : () => { + // Not a session dismissal: a repeated refusal must show again. + setThreadError(threadId, null); + onStop(); + } + } onDismiss={() => { setThreadError(threadId, null); dismissThreadErrorBannerForSession(threadErrorBannerKey); diff --git a/apps/web/src/components/chat/ThreadErrorBanner.tsx b/apps/web/src/components/chat/ThreadErrorBanner.tsx index 7c8f4b7fee78c..966425df0e698 100644 --- a/apps/web/src/components/chat/ThreadErrorBanner.tsx +++ b/apps/web/src/components/chat/ThreadErrorBanner.tsx @@ -4,6 +4,7 @@ import { CHATGPT_WORKSPACE_ADMIN_REQUEST, CHATGPT_WORKSPACE_USAGE_LIMIT_CODE, } from "@t3tools/shared/usageLimits"; +import { isBackgroundWorkRefusal } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import { memo } from "react"; import { writeTextToClipboard } from "../../hooks/useCopyToClipboard"; import { Alert, AlertAction, AlertDescription } from "../ui/alert"; @@ -74,15 +75,19 @@ export const ThreadErrorBanner = memo(function ThreadErrorBanner({ onDismiss, errorClass, errorCode, + onStop, }: { error: string | null; errorClass?: OrchestrationV2ProviderFailureClass | null; errorCode?: string | null; onDismiss?: () => void; + onStop?: (() => void) | undefined; }) { if (!error) return null; const workspaceLimit = errorCode === CHATGPT_WORKSPACE_USAGE_LIMIT_CODE; const chatGptLimit = workspaceLimit || errorCode === CHATGPT_USAGE_LIMIT_CODE; + // The refusal says "press Stop", but work only the adapter sees shows no Stop elsewhere. + const offerStop = onStop !== undefined && isBackgroundWorkRefusal(error); const variant = errorClass === "usage_limit" || chatGptLimit ? "warning" : "error"; return (
@@ -108,8 +113,13 @@ export const ThreadErrorBanner = memo(function ThreadErrorBanner({ )} - {(chatGptLimit || onDismiss) && ( + {(chatGptLimit || offerStop || onDismiss) && ( + {offerStop ? ( + + ) : null} {workspaceLimit ? (