From 38a0197964707036471e2ae6a9c121df82448731 Mon Sep 17 00:00:00 2001 From: Garrett Fincke Date: Fri, 6 Nov 2026 12:07:29 -0500 Subject: [PATCH 1/3] fix: compaction no longer ends background agents when a message skips it Co-Authored-By: Claude --- .../src/orchestration-v2/Orchestrator.ts | 16 ++- .../SteeringCompletion.integration.test.ts | 128 +++++++++++++++++- apps/web/src/components/ChatView.logic.ts | 11 +- apps/web/src/components/ChatView.tsx | 3 + .../src/state/contextWindow.test.ts | 19 +++ .../client-runtime/src/state/contextWindow.ts | 15 ++ 6 files changed, 187 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index bcec0a13cf20..d60ba9b8ed06 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -9837,8 +9837,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio occurredAt: now, payload: { ...queuedRun, queuePosition: 1 }, }); + // Stopping can end the provider process and its background work, so the message waits instead. + const targetProviderThread = projection.providerThreads.find( + (thread) => thread.id === target.providerThreadId, + ); + const targetSession = projection.providerSessions.find( + (session) => session.id === targetProviderThread?.providerSessionId, + ); + const holdsBackgroundWork = + (targetProviderThread?.pendingBackgroundTasks?.length ?? 0) > 0 || + (targetProviderThread !== undefined && + targetSession !== undefined && + (yield* sessionHasBackgroundWork(targetSession, [targetProviderThread]))); const stops = - isLiveRun(target) && !(yield* stopReachedRun(command, command.threadId, target.id)); + isLiveRun(target) && + !holdsBackgroundWork && + !(yield* stopReachedRun(command, command.threadId, target.id)); return yield* stopRunForQueuedCommand({ command, events, diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 2b08c0b8345b..e52a874adecf 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -491,6 +491,7 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* const started: ProviderAdapter.ProviderAdapterV2TurnInput[] = []; const steered: string[] = []; const interrupted: ProviderTurnId[] = []; + const sessionBackgroundWork = { running: false }; const capabilities = { ...CodexProviderCapabilitiesV2, turns: { @@ -572,6 +573,8 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* updatedAt: now, }), resumeThread: ({ providerThread }) => Effect.succeed(providerThread), + hasPendingBackgroundWork: Effect.sync(() => sessionBackgroundWork.running), + hasPendingBackgroundWorkForThread: () => Effect.sync(() => sessionBackgroundWork.running), startTurn, compactThread: startTurn, steerTurn: (turn) => @@ -648,7 +651,16 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* yield* Fiber.join(running); return threadId; }); - return { cwd, events, started, steered, interrupted, layer, startFirstTurn }; + return { + cwd, + events, + started, + steered, + interrupted, + sessionBackgroundWork, + layer, + startFirstTurn, + }; }); it.effect("steers a delegated task question notice into the running parent turn", () => @@ -1385,6 +1397,120 @@ it.effect("marks the model switch on a promoted message that restarts the turn", ), ); +// Claude lists background commands on the provider thread; only its session knows background agents. +it.effect.each([ + { background: "none", stopsCompaction: true }, + { background: "command", stopsCompaction: false }, + { background: "agent", stopsCompaction: false }, +] as const)( + "steering into a compaction stops it: $stopsCompaction with background work: $background", + ({ background, stopsCompaction }) => + Effect.scoped( + Effect.gen(function* () { + const { events, started, sessionBackgroundWork, layer, startFirstTurn } = + yield* nextTurnSelectionHarness(`steer-compaction-${background}`); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const threadId = yield* startFirstTurn; + const first = started[0]!; + const send = (id: string, text: string) => + orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(id), + threadId, + messageId: MessageId.make(`message:${id}`), + text, + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "queue_after_active" }, + createdBy: "user", + creationSource: "web", + }); + yield* send("compact", "/compact"); + + const firstProjection = yield* orchestrator.getThreadProjection(threadId); + const turn = firstProjection.providerTurns[0]!; + const settled = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === first.runId && + event.payload.status === "waiting", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...turn, status: "completed", completedAt: yield* DateTime.now }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: turn.providerThreadId, + providerTurnId: turn.id, + runOrdinal: first.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(settled); + yield* worker.drain(); + yield* orchestrator.resumeQueuedRuns; + yield* worker.drain(); + const compaction = started[1]!; + assert.equal(compaction.message.text, "/compact"); + + sessionBackgroundWork.running = background === "agent"; + if (background === "command") { + const providerThread = (yield* orchestrator.getThreadProjection(threadId)) + .providerThreads[0]!; + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("event:background-roster"), + type: "provider-thread.updated", + threadId, + occurredAt: yield* DateTime.now, + payload: { + ...providerThread, + pendingBackgroundTasks: [ + { taskId: "gate", description: "Wait for the gate", kind: "command" }, + ], + }, + }, + ], + }); + } + yield* send("queued", "status"); + const queuedRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === MessageId.make("message:queued"), + )!; + yield* orchestrator.dispatch({ + type: "queued-message.promote-to-steer", + commandId: CommandId.make("promote"), + threadId, + queuedRunId: queuedRun.id, + targetRunId: compaction.runId, + }); + yield* worker.drain(); + + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + projection.runs.find((run) => run.id === compaction.runId)?.status, + stopsCompaction ? "interrupted" : "running", + ); + const queued = projection.runs.find((run) => run.id === queuedRun.id); + assert.equal(queued?.status, "queued"); + assert.equal(queued?.queuePosition, 1); + }).pipe(Effect.provide(layer)); + }), + ), +); + it.effect("keeps a queued message queued when promoted into a run that Stop reached", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index d4b54a7e5a84..8bdec13e6d7f 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -1559,8 +1559,11 @@ export function presentThreadShellFromProjection( }; } -/** Background work the provider still runs after the latest run ended. */ -export function deriveThreadPendingBackgroundTasks(projection: ThreadStructure | null) { +/** Background work the provider still runs after the latest run ended, plus watched pull requests. */ +export function deriveThreadPendingBackgroundTasks( + projection: ThreadStructure | null, + options?: { readonly includePullRequestWatches?: boolean }, +) { if (projection === null) return []; const sessionError = projection.providerSessions.findLast( @@ -1576,7 +1579,9 @@ export function deriveThreadPendingBackgroundTasks(projection: ThreadStructure | turnItems: projection.turnItems, activeProviderThreadId: projection.thread.activeProviderThreadId, runs: projection.runs, - pullRequests: projection.thread.pullRequests, + ...(options?.includePullRequestWatches === false + ? {} + : { pullRequests: projection.thread.pullRequests }), }), ]; } diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 76ad543630fb..b683759fbacf 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -2501,6 +2501,9 @@ export default function ChatView(props: ChatViewProps) { !compactDisabled && !hasHeldQueuedRuns && multipleModelSelections === null && + // Background agents keep the session live and report into its next turn. + deriveThreadPendingBackgroundTasks(serverProjection, { includePullRequestWatches: false }) + .length === 0 && shouldOfferResumeCompaction({ provider: selectedProvider, usedTokens: activeContextWindow.usedTokens, diff --git a/packages/client-runtime/src/state/contextWindow.test.ts b/packages/client-runtime/src/state/contextWindow.test.ts index 3f1a4d3d4199..a35fa409d922 100644 --- a/packages/client-runtime/src/state/contextWindow.test.ts +++ b/packages/client-runtime/src/state/contextWindow.test.ts @@ -43,6 +43,8 @@ function sendInput( readonly promptOutsideWindow?: boolean; readonly text?: string; readonly threadIdle?: boolean; + readonly backgroundTask?: boolean; + readonly backgroundAgent?: boolean; } = {}, ) { const source = overrides.source ?? claude; @@ -67,6 +69,9 @@ function sendInput( providerInstanceId: source, contextUsage: null, updatedAt: DateTime.makeUnsafe(lastTurnAt), + ...(overrides.backgroundTask === true && id === "provider-thread-current" + ? { pendingBackgroundTasks: [{ taskId: "gate", kind: "command" }] } + : {}), }); const visibleTurnItems = overrides.promptOutsideWindow === true @@ -123,6 +128,18 @@ function sendInput( ? [{ id: RunId.make("queued-run"), status: "queued", queueHeld: true, startedAt: null }] : []), ], + turnItems: + overrides.backgroundAgent === true + ? [ + { + id: "item-agent", + type: "subagent", + status: "running", + runId: RunId.make("run-1"), + prompt: "Identify the block wall", + }, + ] + : [], runtimeRequests: [ ...(overrides.dismissed === true ? [ @@ -254,6 +271,8 @@ describe("shouldCompactBeforeSend", () => { { name: "while the queue is held", heldQueue: true, expected: false }, { name: "while a request waits for an answer", pendingRequest: true, expected: false }, { name: "while the thread is working", threadIdle: false, expected: false }, + { name: "while a background command runs", backgroundTask: true, expected: false }, + { name: "while a background agent runs", backgroundAgent: true, expected: false }, { name: "with no conversation but /compact", onlyCompactMessages: true, expected: false }, { name: "for a typed /compact", text: "/compact keep the plan", expected: false }, ] as const)("$name", ({ name: _name, expected, ...overrides }) => { diff --git a/packages/client-runtime/src/state/contextWindow.ts b/packages/client-runtime/src/state/contextWindow.ts index 66c48807465a..c8ff69bb7e0a 100644 --- a/packages/client-runtime/src/state/contextWindow.ts +++ b/packages/client-runtime/src/state/contextWindow.ts @@ -13,6 +13,7 @@ import { COMPACTION_SUGGESTION_MIN_TOKENS, isClaudeResumeCompactionQuestion, } from "@t3tools/shared/claudeCompaction"; +import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; import * as DateTime from "effect/DateTime"; import { resolvePendingHandoff } from "../handoff.ts"; @@ -247,6 +248,7 @@ export function shouldCompactBeforeSend(input: { | "providerTurns" | "runtimeRequests" | "runs" + | "turnItems" | "visibleTurnItems" >; readonly visibleTurnItems: ReadonlyArray<{ readonly item: ThreadStructureTurnItem }>; @@ -296,6 +298,19 @@ export function shouldCompactBeforeSend(input: { const providerThread = projection.providerThreads.find( (thread) => thread.id === activeProviderThreadId, ); + // Background agents keep the session live and report into its next turn. + if ( + derivePendingBackgroundWork({ + latestRun: projection.runs.findLast( + (run) => run.status !== "queued" && run.status !== "rolled_back", + ), + providerThreads: projection.providerThreads, + turnItems: projection.turnItems, + activeProviderThreadId, + runs: projection.runs, + }).length > 0 + ) + return false; const liveUsage = latestProviderTurnTokenUsage( projection.providerTurns.filter((turn) => turn.providerThreadId === activeProviderThreadId), input.selection.instanceId === projection.thread.modelSelection.instanceId ? projection : null, From 4642a649e73c4c805e5991015d27dfa46b8e4996 Mon Sep 17 00:00:00 2001 From: Garrett Fincke Date: Fri, 6 Nov 2026 13:25:28 -0500 Subject: [PATCH 2/3] fix: web and mobile share one background-work check before compacting Co-Authored-By: Claude --- apps/web/src/components/ChatView.logic.ts | 11 ++---- apps/web/src/components/ChatView.tsx | 4 +-- .../src/state/contextWindow.test.ts | 27 ++++++++++++++ .../client-runtime/src/state/contextWindow.ts | 36 ++++++++++++------- 4 files changed, 56 insertions(+), 22 deletions(-) diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index 8bdec13e6d7f..d4b54a7e5a84 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -1559,11 +1559,8 @@ export function presentThreadShellFromProjection( }; } -/** Background work the provider still runs after the latest run ended, plus watched pull requests. */ -export function deriveThreadPendingBackgroundTasks( - projection: ThreadStructure | null, - options?: { readonly includePullRequestWatches?: boolean }, -) { +/** Background work the provider still runs after the latest run ended. */ +export function deriveThreadPendingBackgroundTasks(projection: ThreadStructure | null) { if (projection === null) return []; const sessionError = projection.providerSessions.findLast( @@ -1579,9 +1576,7 @@ export function deriveThreadPendingBackgroundTasks( turnItems: projection.turnItems, activeProviderThreadId: projection.thread.activeProviderThreadId, runs: projection.runs, - ...(options?.includePullRequestWatches === false - ? {} - : { pullRequests: projection.thread.pullRequests }), + pullRequests: projection.thread.pullRequests, }), ]; } diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index b683759fbacf..804dcc6ec4fe 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -94,6 +94,7 @@ import { threadSupportsProviderHandoff, } from "@t3tools/client-runtime/state/thread-workflows"; import { resolvePendingHandoff } from "@t3tools/client-runtime/handoff"; +import { hasPendingProviderBackgroundWork } from "@t3tools/client-runtime/state/context-window"; import { HandoffRow } from "./chat/V2LifecycleRow"; import { codexFeedbackMessage, @@ -2502,8 +2503,7 @@ export default function ChatView(props: ChatViewProps) { !hasHeldQueuedRuns && multipleModelSelections === null && // Background agents keep the session live and report into its next turn. - deriveThreadPendingBackgroundTasks(serverProjection, { includePullRequestWatches: false }) - .length === 0 && + (serverProjection === null || !hasPendingProviderBackgroundWork(serverProjection)) && shouldOfferResumeCompaction({ provider: selectedProvider, usedTokens: activeContextWindow.usedTokens, diff --git a/packages/client-runtime/src/state/contextWindow.test.ts b/packages/client-runtime/src/state/contextWindow.test.ts index a35fa409d922..dce4afc87e96 100644 --- a/packages/client-runtime/src/state/contextWindow.test.ts +++ b/packages/client-runtime/src/state/contextWindow.test.ts @@ -45,6 +45,7 @@ function sendInput( readonly threadIdle?: boolean; readonly backgroundTask?: boolean; readonly backgroundAgent?: boolean; + readonly laterRun?: "queued" | "rolled_back"; } = {}, ) { const source = overrides.source ?? claude; @@ -127,6 +128,19 @@ function sendInput( ...(overrides.heldQueue === true ? [{ id: RunId.make("queued-run"), status: "queued", queueHeld: true, startedAt: null }] : []), + ...(overrides.laterRun === undefined + ? [] + : [ + { + id: RunId.make("run-2"), + ordinal: 2, + providerInstanceId: source, + modelSelection: sourceSelection, + status: overrides.laterRun, + startedAt: overrides.laterRun === "queued" ? null : "2026-10-30T14:21:00.000Z", + contextHandoffId: null, + }, + ]), ], turnItems: overrides.backgroundAgent === true @@ -273,6 +287,19 @@ describe("shouldCompactBeforeSend", () => { { name: "while the thread is working", threadIdle: false, expected: false }, { name: "while a background command runs", backgroundTask: true, expected: false }, { name: "while a background agent runs", backgroundAgent: true, expected: false }, + { + name: "while a background agent runs behind a queued message", + backgroundAgent: true, + laterRun: "queued", + expected: false, + }, + // Like web and idle compaction, a rolled-back latest run holds no background work. + { + name: "after a rollback that left a background command listed", + backgroundTask: true, + laterRun: "rolled_back", + expected: true, + }, { name: "with no conversation but /compact", onlyCompactMessages: true, expected: false }, { name: "for a typed /compact", text: "/compact keep the plan", expected: false }, ] as const)("$name", ({ name: _name, expected, ...overrides }) => { diff --git a/packages/client-runtime/src/state/contextWindow.ts b/packages/client-runtime/src/state/contextWindow.ts index c8ff69bb7e0a..b21cd4195655 100644 --- a/packages/client-runtime/src/state/contextWindow.ts +++ b/packages/client-runtime/src/state/contextWindow.ts @@ -234,6 +234,29 @@ export function shouldOfferResumeCompaction(input: { ); } +/** + * Whether the provider still runs background agents or commands for the thread's latest run. + * Pull request watches do not count, and a queued run has not taken over the session yet. + */ +export function hasPendingProviderBackgroundWork( + projection: Pick, +): boolean { + const latestRun = projection.runs.reduce( + (latest, run) => + run.status === "queued" || (latest !== null && latest.ordinal > run.ordinal) ? latest : run, + null, + ); + return ( + derivePendingBackgroundWork({ + latestRun, + providerThreads: projection.providerThreads, + turnItems: projection.turnItems, + activeProviderThreadId: projection.thread.activeProviderThreadId, + runs: projection.runs, + }).length > 0 + ); +} + /** * Whether a plain send into an idle thread compacts it first, as web's composer does under * "When you return". Web reads the same state for its "Compact and send" button. A send that @@ -299,18 +322,7 @@ export function shouldCompactBeforeSend(input: { (thread) => thread.id === activeProviderThreadId, ); // Background agents keep the session live and report into its next turn. - if ( - derivePendingBackgroundWork({ - latestRun: projection.runs.findLast( - (run) => run.status !== "queued" && run.status !== "rolled_back", - ), - providerThreads: projection.providerThreads, - turnItems: projection.turnItems, - activeProviderThreadId, - runs: projection.runs, - }).length > 0 - ) - return false; + if (hasPendingProviderBackgroundWork(projection)) return false; const liveUsage = latestProviderTurnTokenUsage( projection.providerTurns.filter((turn) => turn.providerThreadId === activeProviderThreadId), input.selection.instanceId === projection.thread.modelSelection.instanceId ? projection : null, From 1cab0a0eb8fcaf0ef6da25d3f1ede790573f4ff1 Mon Sep 17 00:00:00 2001 From: Garrett Fincke Date: Fri, 13 Nov 2026 23:36:34 -0500 Subject: [PATCH 3/3] test(server): the compaction steering test waits for recorded events It promoted the message before the compaction's turn was recorded as running, and read the queued run before the orchestrator had started it. Both now wait for the event, and the row with no background work asserts the message is sent next. Co-Authored-By: Claude --- .../SteeringCompletion.integration.test.ts | 40 +++++++++++++++++-- 1 file changed, 37 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index d426705892f4..e0783a6902fd 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -2469,8 +2469,20 @@ it.effect.each([ }); yield* Fiber.join(settled); yield* worker.drain(); + const compacting = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && + event.payload.status === "running" && + event.payload.runAttemptId !== first.attemptId, + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); yield* orchestrator.resumeQueuedRuns; yield* worker.drain(); + yield* Fiber.join(compacting); const compaction = started[1]!; assert.equal(compaction.message.text, "/compact"); @@ -2499,6 +2511,17 @@ it.effect.each([ const queuedRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( (run) => run.userMessageId === MessageId.make("message:queued"), )!; + const leftQueue = yield* orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === queuedRun.id && + event.payload.status !== "queued", + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); yield* orchestrator.dispatch({ type: "queued-message.promote-to-steer", commandId: CommandId.make("promote"), @@ -2507,15 +2530,26 @@ it.effect.each([ targetRunId: compaction.runId, }); yield* worker.drain(); + // A stopped compaction hands the queue to the message once its terminal is ingested. + if (stopsCompaction) { + yield* Fiber.join(leftQueue); + yield* worker.drain(); + } const projection = yield* orchestrator.getThreadProjection(threadId); assert.equal( projection.runs.find((run) => run.id === compaction.runId)?.status, stopsCompaction ? "interrupted" : "running", ); - const queued = projection.runs.find((run) => run.id === queuedRun.id); - assert.equal(queued?.status, "queued"); - assert.equal(queued?.queuePosition, 1); + assert.deepEqual( + started.map((turn) => turn.message.text), + stopsCompaction ? ["first", "/compact", "status"] : ["first", "/compact"], + ); + if (!stopsCompaction) { + const queued = projection.runs.find((run) => run.id === queuedRun.id); + assert.equal(queued?.status, "queued"); + assert.equal(queued?.queuePosition, 1); + } }).pipe(Effect.provide(layer)); }), ),