diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 61929131e9d7..e3d6d73c1e9b 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -11335,8 +11335,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 11b36aeb148e..e3ed3179d139 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -522,6 +522,7 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* const steered: string[] = []; const steeredSelections: Array = []; const interrupted: ProviderTurnId[] = []; + const sessionBackgroundWork = { running: false }; const capabilities = { ...CodexProviderCapabilitiesV2, turns: { @@ -604,6 +605,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) => @@ -697,7 +700,17 @@ const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* yield* (yield* ThreadCommandExecutor.ThreadCommandExecutor).withLock(threadId, Effect.void); return threadId; }); - return { cwd, events, started, steered, steeredSelections, interrupted, layer, startFirstTurn }; + return { + cwd, + events, + started, + steered, + steeredSelections, + interrupted, + sessionBackgroundWork, + layer, + startFirstTurn, + }; }); it.effect.each([ @@ -2463,6 +2476,154 @@ 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(); + 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"); + + 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"), + )!; + 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"), + threadId, + queuedRunId: queuedRun.id, + 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", + ); + 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)); + }), + ), +); + it.effect("keeps a queued message queued when promoted into a run that Stop reached", () => Effect.gen(function* () { const { started, steered, layer, startFirstTurn } = diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 5db96b73fc6b..0e3b39ff0fde 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -102,6 +102,7 @@ import { } from "@t3tools/client-runtime/state/thread-workflows"; import { compactionBlocksSteering } from "@t3tools/client-runtime/state/composer-dispatch"; import { resolvePendingHandoff } from "@t3tools/client-runtime/handoff"; +import { hasPendingProviderBackgroundWork } from "@t3tools/client-runtime/state/context-window"; import { HandoffRow } from "./chat/V2LifecycleRow"; import { codexFeedbackMessage, @@ -2534,6 +2535,8 @@ export default function ChatView(props: ChatViewProps) { !compactDisabled && !hasHeldQueuedRuns && multipleModelSelections === null && + // Background agents keep the session live and report into its next turn. + (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 994d3d62a206..f2decceb2077 100644 --- a/packages/client-runtime/src/state/contextWindow.test.ts +++ b/packages/client-runtime/src/state/contextWindow.test.ts @@ -43,6 +43,9 @@ function sendInput( readonly promptOutsideWindow?: boolean; readonly text?: string; readonly threadIdle?: boolean; + readonly backgroundTask?: boolean; + readonly backgroundAgent?: boolean; + readonly laterRun?: "queued" | "rolled_back"; } = {}, ) { const source = overrides.source ?? claude; @@ -67,6 +70,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 @@ -122,7 +128,32 @@ 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 + ? [ + { + id: "item-agent", + type: "subagent", + status: "running", + runId: RunId.make("run-1"), + prompt: "Identify the block wall", + }, + ] + : [], runtimeRequests: [ ...(overrides.dismissed === true ? [ @@ -254,6 +285,21 @@ describe("compactBeforeSendTokens", () => { { 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: "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 a8f7e0f88c35..5ee8569f95c3 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"; @@ -245,6 +246,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 + ); +} + /** * The tokens a plain send into an idle thread compacts first, as web's composer does under * "When you return", or null when it sends as is. Web reads the same state for its "Compact and @@ -260,6 +284,7 @@ export function compactBeforeSendTokens(input: { | "providerTurns" | "runtimeRequests" | "runs" + | "turnItems" | "visibleTurnItems" >; readonly visibleTurnItems: ReadonlyArray<{ readonly item: ThreadStructureTurnItem }>; @@ -309,6 +334,8 @@ export function compactBeforeSendTokens(input: { const providerThread = projection.providerThreads.find( (thread) => thread.id === activeProviderThreadId, ); + // Background agents keep the session live and report into its next turn. + if (hasPendingProviderBackgroundWork(projection)) return null; const liveUsage = latestProviderTurnTokenUsage( projection.providerTurns.filter((turn) => turn.providerThreadId === activeProviderThreadId), input.selection.instanceId === projection.thread.modelSelection.instanceId ? projection : null,