From a4029246d5ffa4aa4ec23580652474d776fc678b Mon Sep 17 00:00:00 2001 From: li-xiu-qi Date: Tue, 27 Oct 2026 23:35:17 +0800 Subject: [PATCH] fix(coding-agent): keep subagents alive across compaction --- packages/coding-agent/docs/extensions.md | 14 +- .../coding-agent/src/core/agent-session.ts | 59 ++++- .../coding-agent/src/core/extensions/types.ts | 16 +- .../src/features/step-subagent.ts | 146 +++++++++++- .../src/features/subagent/execute.ts | 2 +- .../src/features/subagent/lane-lifecycle.ts | 117 ++++++++- .../test/step-subagent-events.test.ts | 50 +++- .../307-compact-aborts-subagent.test.ts | 222 ++++++++++++++++++ ...253-manual-compact-during-response.test.ts | 13 +- 9 files changed, 617 insertions(+), 22 deletions(-) create mode 100644 packages/coding-agent/test/suite/regressions/307-compact-aborts-subagent.test.ts diff --git a/packages/coding-agent/docs/extensions.md b/packages/coding-agent/docs/extensions.md index fdfe9f674..2da9bf26e 100644 --- a/packages/coding-agent/docs/extensions.md +++ b/packages/coding-agent/docs/extensions.md @@ -328,7 +328,8 @@ user sends another prompt ◄───────────────── /name or pi.setSessionName() └─► session_info_changed -/compact or auto-compaction +/compact (manual) + ├─► session_detach_foreground (best-effort; hand foreground work to a lane) ├─► session_before_compact (can cancel or customize) ├─► session_compact (success) └─► session_compact_failed (failure or abort) @@ -449,6 +450,17 @@ pi.on("session_before_fork", async (event, ctx) => { After a successful fork or clone, step emits `session_shutdown` for the old extension instance, reloads and rebinds extensions for the new session, then emits `session_start` with `reason: "fork"` and `previousSessionFile`. Do cleanup work in `session_shutdown`, then reestablish any in-memory state in `session_start`. +#### session_detach_foreground + +Fired only for manual compaction, before it waits for the active run to settle. A host whose foreground work can continue in the background (for example a blocking `subagent` call) hands that work to a lane here and returns at once, so compaction does not wait for the work to finish. Detach is best-effort: when no handler claims the work, or the host cannot own a lane, compaction keeps its old behaviour and waits for the run. The return value is ignored. + +```typescript +pi.on("session_detach_foreground", async (event, ctx) => { + // event.reason - "manual" + // Move long-running foreground work into a background lane here. +}); +``` + #### session_before_compact / session_compact / session_compact_failed Fired on compaction. See [compaction.md](compaction.md) for details. diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 54acd866d..b39d5b57a 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -353,6 +353,8 @@ export class AgentSession { // Compaction state private _compactionAbortController: AbortController | undefined = undefined; + /** True while a manual compaction is waiting for the current run to settle. */ + private _manualCompactionPending = false; private _autoCompactionAbortController: AbortController | undefined = undefined; private _overflowRecoveryAttempted = false; private _liveContext?: LiveContextManager; @@ -820,6 +822,7 @@ export class AgentSession { !settings.enabled || !this.settingsManager.getCompactionEnabled() || this._manualClm || + this._manualCompactionPending || this._autoClmAbortController || this._blocksNativeThreshold() || signal?.aborted || @@ -2626,20 +2629,53 @@ export class AgentSession { /** * Manually compact the session context. * - * This is the manual entry point used by `/compact`, RPC, and extensions. It is - * separate from automatic threshold/overflow compaction, which enters through - * `_checkCompaction()` and `_runAutoCompaction()`. After preparation and the - * `session_before_compact` hook, both paths call the lower-level `compact()` - * function imported from `./compaction/index.ts`, unless the hook cancels or - * supplies a custom result. - * - * Aborts the current agent operation first. Manual compaction never retries or - * continues the interrupted agent turn. + * This is the manual entry point used by `/compact`, RPC, and extensions. It + * shares the non-aborting lifecycle with automatic threshold/overflow + * compaction: it never cancels the current run. Before waiting for the run to + * settle it offers foreground work to extensions through + * `session_detach_foreground`; a host that hands its work to a background lane + * lets compaction proceed while that work keeps running. If no host claims the + * work, the request waits as before, which keeps a foreground subagent alive + * until its result can be compacted with the session. * * @param customInstructions Optional instructions for the compaction summary */ async compact(customInstructions?: string): Promise { - await this.abort(); + this._manualCompactionPending = true; + try { + await this._detachForegroundWorkForCompaction(); + await this.waitForIdle(); + return await this._compactNow(customInstructions); + } finally { + this._manualCompactionPending = false; + } + } + + /** + * Ask extensions to hand foreground work to a background lane before manual + * compaction waits for the run. Best-effort by design: a failed or absent + * handler leaves the work in the foreground, and `waitForIdle()` keeps the old + * behaviour of waiting for it to finish. + */ + private async _detachForegroundWorkForCompaction(): Promise { + if (!this._extensionRunner.hasHandlers("session_detach_foreground")) return; + try { + await this._extensionRunner.emit({ type: "session_detach_foreground", reason: "manual" }); + } catch { + // Detaching is an optimisation, not a precondition for compaction. + } + } + + /** + * Run manual compaction once the session is idle. Automatic compaction enters + * through `_checkCompaction()` and `_runAutoCompaction()`; after preparation + * and the `session_before_compact` hook, both paths call the lower-level + * `compact()` function imported from `./compaction/index.ts`, unless the hook + * cancels or supplies a custom result. + * + * @param customInstructions Optional instructions for the compaction summary + */ + private async _compactNow(customInstructions?: string): Promise { this._compactionAbortController = new AbortController(); this._emit({ type: "compaction_start", reason: "manual" }); let fromExtension = false; @@ -2975,6 +3011,9 @@ export class AgentSession { private async _runAutoCompaction(reason: "overflow" | "threshold", willRetry: boolean): Promise { const runSignal = this._agentRunAbortController?.signal; if (runSignal?.aborted) return false; + // A manual compaction request owns the next rewrite. Auto-compaction must not + // win the race and leave the manual request with "Already compacted". + if (this._manualCompactionPending) return false; if (reason === "threshold" && this._blocksNativeThreshold()) return false; const settings = this.settingsManager.getCompactionSettings(); let started = false; diff --git a/packages/coding-agent/src/core/extensions/types.ts b/packages/coding-agent/src/core/extensions/types.ts index 7fc046175..b3136dfec 100644 --- a/packages/coding-agent/src/core/extensions/types.ts +++ b/packages/coding-agent/src/core/extensions/types.ts @@ -737,6 +737,18 @@ export interface SessionTreeEvent { fromExtension?: boolean; } +/** + * Fired before manual compaction waits for the session to settle, so a host can + * hand its foreground work to a background lane instead of blocking compaction + * until that work finishes. Detach is best-effort: compaction still waits when + * no handler claims the work. + */ +export interface SessionDetachForegroundEvent { + type: "session_detach_foreground"; + /** What triggered the detach request. Only manual compaction uses it today. */ + reason: "manual"; +} + export type SessionEvent = | SessionStartEvent | SessionInfoChangedEvent @@ -747,7 +759,8 @@ export type SessionEvent = | SessionCompactFailedEvent | SessionShutdownEvent | SessionBeforeTreeEvent - | SessionTreeEvent; + | SessionTreeEvent + | SessionDetachForegroundEvent; // ============================================================================ // Agent Events @@ -1349,6 +1362,7 @@ export interface ExtensionAPI { on(event: "session_shutdown", handler: ExtensionHandler): void; on(event: "session_before_tree", handler: ExtensionHandler): void; on(event: "session_tree", handler: ExtensionHandler): void; + on(event: "session_detach_foreground", handler: ExtensionHandler): void; on(event: "context", handler: ExtensionHandler): void; on( event: "before_provider_request", diff --git a/packages/coding-agent/src/features/step-subagent.ts b/packages/coding-agent/src/features/step-subagent.ts index 11ad89bc1..2c41d5ca8 100644 --- a/packages/coding-agent/src/features/step-subagent.ts +++ b/packages/coding-agent/src/features/step-subagent.ts @@ -40,6 +40,7 @@ import { createLaneLifecycle, getLiveSubagentSession, MAX_RETAINED_FINISHED_LANES, + type StepSubagentLaneRuntime, } from "./subagent/lane-lifecycle.ts"; import { renderAgentNotification, @@ -818,6 +819,147 @@ export function createStepSubagentExtension(options: StepSubagentExtensionOption clearLaneWidget, }); + // Foreground calls that manual compaction may hand to a background lane. + // Keyed by tool call id so several concurrent foreground calls can each be + // detached independently; the entry is removed as soon as the call returns. + interface ForegroundSubagentRun { + toolCallId: string; + params: SubagentParams; + ctx: ExtensionContext; + controller: AbortController; + /** Child session id chosen before the run starts, so an early detach still knows it. */ + sessionId: string; + details?: StepSubagentDetails; + onLaneUpdate?: (result: AgentToolResult) => void; + detach?: () => BackgroundAgentLane | string | undefined; + } + const foregroundRuns = new Map(); + const subagentListWidgetKey = (toolCallId: string): string => `step-subagent-list:${toolCallId}`; + const clearSubagentListWidget = (ctx: ExtensionContext, toolCallId: string): void => { + if (!ctx.hasUI) return; + try { + ctx.ui.setWidget(subagentListWidgetKey(toolCallId), undefined); + } catch { + // The host may already have torn its UI down; widget cleanup is best-effort. + } + }; + const foregroundDetailsFallback = (params: SubagentParams, ctx: ExtensionContext): StepSubagentDetails => { + const mode: StepSubagentDetails["mode"] = params.tasks ? "parallel" : params.chain ? "chain" : "single"; + return { + mode, + agentScope: params.agentScope ?? "user", + userAgentsDir: path.join(resolved.agentDir, "agents"), + projectAgentsDir: path.join(ctx.cwd, resolved.configDirName, "agents"), + results: [], + }; + }; + + /** + * Run a foreground subagent call so it can be detached mid-flight. The tool + * result races the real run against a detach request: when compaction asks + * for the foreground work, the running promise is adopted by a background + * lane and this call returns at once with the lane id. The child process is + * never respawned or aborted by the detach itself. + */ + const runDetachableForeground = async ( + toolCallId: string, + params: SubagentParams, + signal: AbortSignal | undefined, + onUpdate: AgentToolUpdateCallback | undefined, + ctx: ExtensionContext, + ): Promise> => { + const controller = new AbortController(); + const abortFromTool = (): void => controller.abort(); + if (signal) { + if (signal.aborted) controller.abort(); + else signal.addEventListener("abort", abortFromTool, { once: true }); + } + // Choose the child session id before the run starts. A detach can arrive + // before the first streamed update, and the lane must still reuse this + // child's transcript for later replies. + const sessionId = `${SUBAGENT_SESSION_ID_PREFIX}${randomUUID()}`; + const run: ForegroundSubagentRun = { toolCallId, params, ctx, controller, sessionId }; + foregroundRuns.set(toolCallId, run); + let detached = false; + let settled = false; + let resolveDetached!: (result: AgentToolResult) => void; + const detachedResult = new Promise>((resolve) => { + resolveDetached = resolve; + }); + const captureUpdate = (result: AgentToolResult): void => { + if (!detached) onUpdate?.(result); + if (result.details) run.details = result.details; + run.onLaneUpdate?.(result); + }; + const laneRuntime: StepSubagentLaneRuntime = { sessionId, execution: "blocking" }; + const rawRunPromise = withSubagentListWidget(toolCallId, ctx, captureUpdate, (update) => + executeSubagent(params, controller.signal, update, ctx, resolved, laneRuntime), + ); + // Mark the call settled before the raced promise resolves, so a detach + // request that arrives after the child finished cannot adopt a dead run. + const runPromise = rawRunPromise.then( + (result) => { + settled = true; + return result; + }, + (error) => { + settled = true; + throw error; + }, + ); + run.detach = () => { + if (detached || settled) return undefined; + // Respect the same host/config gates as `run_in_background`: a host that + // cannot own lanes keeps the foreground call and compaction waits for it. + if (backgroundUnavailableReason(ctx, options.backgroundSettings?.())) return undefined; + const lane = lifecycle.adoptForegroundRun({ + params, + ctx, + controller, + details: run.details ?? foregroundDetailsFallback(params, ctx), + promise: runPromise, + sessionId: run.sessionId, + turnTimeoutMs: resolveBackgroundTimeoutMs(undefined, options.backgroundSettings?.()?.timeoutSeconds), + setUpdate: (update) => { + run.onLaneUpdate = update; + }, + }); + if (typeof lane === "string") return lane; + detached = true; + clearSubagentListWidget(ctx, toolCallId); + resolveDetached( + makeToolResult( + lane.details, + [ + `Subagent detached to background lane ${lane.id} because manual compaction was requested.`, + "It keeps running; you will get one when it finishes.", + "Do not wait for it with sleep or polling.", + ].join(" "), + ), + ); + return lane; + }; + try { + return await Promise.race([runPromise, detachedResult]); + } finally { + foregroundRuns.delete(toolCallId); + if (signal) signal.removeEventListener("abort", abortFromTool); + } + }; + + // Manual compaction asks for foreground work before it waits for the run. + // Detach is best-effort: a host that cannot own lanes, or a session already + // at its lane limit, leaves the call foreground and compaction waits for it. + pi.on("session_detach_foreground", async () => { + for (const run of [...foregroundRuns.values()]) { + try { + run.detach?.(); + } catch { + // One failed detach must not stop the others or break compaction. + } + } + }); + // Lanes belong to this session runtime. Quit, signals, /new, /resume, // /fork, and reload all tear the runtime down through session_shutdown; // lanes are stopped rather than carried into the next session. @@ -843,9 +985,7 @@ export function createStepSubagentExtension(options: StepSubagentExtensionOption execute: async (toolCallId, params, signal, onUpdate, ctx) => { const callParams = params as SubagentParams; const foreground = (runParams: SubagentParams) => - withSubagentListWidget(toolCallId, ctx, onUpdate, (update) => - executeSubagent(runParams, signal, update, ctx, resolved), - ); + runDetachableForeground(toolCallId, runParams, signal, onUpdate, ctx); if (!callParams.run_in_background) return foreground(callParams); const settings = options.backgroundSettings?.(); diff --git a/packages/coding-agent/src/features/subagent/execute.ts b/packages/coding-agent/src/features/subagent/execute.ts index 68669e8f3..62557f856 100644 --- a/packages/coding-agent/src/features/subagent/execute.ts +++ b/packages/coding-agent/src/features/subagent/execute.ts @@ -386,7 +386,7 @@ export async function runPreparedSubagent( const records = tasks.map((task) => makeEmptyResult(task.agent, task.task)); let lastFrameAt = 0; let pendingFrame: ReturnType | undefined; - const execution = laneRuntime ? "background" : "blocking"; + const execution = laneRuntime?.execution ?? (laneRuntime ? "background" : "blocking"); const createdAt = tasks.map(() => Date.now()); const createdTasks = new Set(); const reportCreated = (agent: StepAgentConfig, index: number): void => { diff --git a/packages/coding-agent/src/features/subagent/lane-lifecycle.ts b/packages/coding-agent/src/features/subagent/lane-lifecycle.ts index c7e79e4aa..43236f417 100644 --- a/packages/coding-agent/src/features/subagent/lane-lifecycle.ts +++ b/packages/coding-agent/src/features/subagent/lane-lifecycle.ts @@ -69,6 +69,12 @@ export interface StepSubagentLaneRuntime { keepAlive?: boolean; /** Wall-clock limit for each child turn, in ms. */ turnTimeoutMs?: number; + /** + * Telemetry label for this run. Defaults to `background` when a lane runtime is + * present; a detachable foreground call passes `blocking` so it is not + * misreported as a background lane just because it carries a session id. + */ + execution?: "blocking" | "background"; } /** Live children by subagent session id. One parent process owns its children. */ @@ -198,6 +204,13 @@ export interface LaneLifecycle { prepared: PreparedSubagentRun, turnTimeoutMs: number, ) => BackgroundAgentLane | string; + /** + * Adopt an already running foreground call as a background lane. The caller + * stops awaiting `promise` and returns its own tool result; this lane owns the + * promise from then on and sends the final notification when it settles. + * Returns an error string, and adopts nothing, at the running-lane limit. + */ + adoptForegroundRun: (input: ForegroundRunAdoption) => BackgroundAgentLane | string; stopLane: (lane: BackgroundAgentLane) => string; replyToLane: (lane: BackgroundAgentLane, prompt: string, interrupt: boolean) => string; laneStatus: (lane: BackgroundAgentLane) => string; @@ -212,6 +225,30 @@ export interface LaneLifecycle { type LaneTurnSource = { prepared: PreparedSubagentRun } | { params: SubagentParams }; +/** A foreground subagent call handed to the lane lifecycle mid-flight. */ +export interface ForegroundRunAdoption { + params: SubagentParams; + ctx: ExtensionContext; + /** Controller whose signal the running call already passed to its children. */ + controller: AbortController; + /** Last streamed details; seeds the lane so its widget and status are live. */ + details: StepSubagentDetails; + /** The in-flight call. The lane attaches its final handling to this promise. */ + promise: Promise>; + /** Notification level for the adopted lane. Defaults to `final`. */ + subscribe?: BackgroundLaneSubscribeLevel; + /** Child session id when the foreground call already reported one. */ + sessionId?: string; + /** + * Wall-clock limit for later background turns started from this lane, for + * example an `agent_send` reply. The already-running adopted turn keeps its + * original foreground semantics; this only bounds future turns. + */ + turnTimeoutMs?: number; + /** Receives the lane's update callback so later streamed deltas reach the lane. */ + setUpdate?: (update: (result: AgentToolResult) => void) => void; +} + export function createLaneLifecycle(host: LaneLifecycleHost): LaneLifecycle { const { pi, lanes, updateLaneWidget } = host; /** Bumped by shutdown; lanes and callbacks from an earlier epoch are dead. */ @@ -336,6 +373,19 @@ export function createLaneLifecycle(host: LaneLifecycleHost): LaneLifecycle { return host.run(prepared, signal, onUpdate, lane.ctx, laneRuntime); })(); lane.promise = promise; + attachRunPromise(lane, promise, generation); + }; + + /** + * Final handling shared by started and adopted lanes: publish the result, run a + * queued follow-up turn if one exists, otherwise notify and arm idle expiry. + * Stale generations (a reply, stop, or shutdown intervened) are dropped. + */ + const attachRunPromise = ( + lane: BackgroundAgentLane, + promise: Promise>, + generation: number, + ): void => { void promise .then((result) => { if (!isCurrent(lane, generation)) return; @@ -430,6 +480,71 @@ export function createLaneLifecycle(host: LaneLifecycleHost): LaneLifecycle { return lane; }; + /** + * Adopt a foreground call that is already running. The caller returns its own + * "detached" tool result immediately; from here on the lane owns the original + * promise, its streamed updates, and its final notification. Nothing is + * respawned, so the child process keeps the work it was already doing. + */ + const adoptForegroundRun: LaneLifecycle["adoptForegroundRun"] = (input) => { + const running = [...lanes.values()].filter((lane) => lane.status === "running").length; + if (running >= MAX_RUNNING_LANES) { + return `This session already has ${running} running background lanes (limit ${MAX_RUNNING_LANES}); the foreground subagent was left running and compaction will wait for it.`; + } + const id = newLaneId(); + const now = Date.now(); + const startedAt = input.details.startedAt ?? now; + const details: StepSubagentDetails = { + ...input.details, + agentId: id, + status: "running", + startedAt, + updatedAt: now, + }; + const lane: BackgroundAgentLane = { + id, + params: input.params, + ctx: input.ctx, + controller: input.controller, + status: "running", + details, + promise: input.promise, + queuedPrompts: [], + startedAt, + updatedAt: now, + // The adopted run is already generation 1; a later reply bumps it. + runGeneration: 1, + subscribe: input.subscribe ?? (input.params.subscribe === "progress" ? "progress" : "final"), + lastProgressNotifyAt: now, + sessionId: input.sessionId ?? `${SUBAGENT_SESSION_ID_PREFIX}${randomUUID()}`, + // The adopted turn already runs with foreground semantics; later turns + // started from this lane use the supplied background limit. + turnTimeoutMs: input.turnTimeoutMs ?? 0, + }; + laneEpochs.set(lane, epoch); + lanes.set(id, lane); + const generation = lane.runGeneration; + const onUpdate = (partial: AgentToolResult): void => { + if (!isCurrent(lane, generation)) return; + if (partial.details) { + lane.details = { + ...partial.details, + agentId: lane.id, + status: "running", + startedAt: lane.startedAt, + updatedAt: Date.now(), + }; + lane.updatedAt = Date.now(); + } + maybeNotifyLaneProgress(pi, lane); + updateLaneWidget(lane); + }; + input.setUpdate?.(onUpdate); + attachRunPromise(lane, input.promise, generation); + updateLaneWidget(lane); + return lane; + }; + /** Shared "stop" verb: interrupt a lane's run and end its rpc children. A * parallel lane owns one keep-alive child per task (session ids with `-N` * suffixes), so every live child is stopped, not only task 0's. */ @@ -582,5 +697,5 @@ export function createLaneLifecycle(host: LaneLifecycleHost): LaneLifecycle { return running; }; - return { launchLane, stopLane, replyToLane, laneStatus, shutdown }; + return { launchLane, adoptForegroundRun, stopLane, replyToLane, laneStatus, shutdown }; } diff --git a/packages/coding-agent/test/step-subagent-events.test.ts b/packages/coding-agent/test/step-subagent-events.test.ts index 42dd7cfc8..0e8ab591a 100644 --- a/packages/coding-agent/test/step-subagent-events.test.ts +++ b/packages/coding-agent/test/step-subagent-events.test.ts @@ -949,7 +949,7 @@ test("status reports state, usage, child sessions, and the output tail without w runner: async (input) => { input.onUpdate?.({ ...textResult(""), exitCode: -1, messages: [] }); await gate.promise; - return { ...textResult("x".repeat(40_000) + "THE END"), sessionFile: "/sessions/child.jsonl" }; + return { ...textResult(`${"x".repeat(40_000)}THE END`), sessionFile: "/sessions/child.jsonl" }; }, })(api); const lane = laneIdOf(await startLane(tools, { agent: "general", task: "long" })); @@ -996,6 +996,48 @@ test("an oversized final output is saved to disk and the notification keeps its } }); +test("a detached foreground subagent keeps its child session id for later replies", async () => { + let releaseFirst!: (result: StepSubagentRunResult) => void; + const firstDone = new Promise((resolve) => { + releaseFirst = resolve; + }); + const { api, tools, sent, handlers } = createApi(); + const inputs: StepSubagentRunInput[] = []; + createStepSubagentExtension({ + agentDir: "/tmp/step-agent-test", + runner: async (input) => { + inputs.push(input); + return inputs.length === 1 ? firstDone : textResult("second done"); + }, + })(api); + const callPromise = tools + .get("subagent")! + .execute( + "call", + { agent: "general", task: "first" } as never, + undefined, + undefined, + createContext("/workspace"), + ) as Promise>; + await waitFor(() => inputs.length === 1); + + const detach = handlers.get("session_detach_foreground"); + expect(detach).toHaveLength(1); + await detach![0]!({ type: "session_detach_foreground", reason: "manual" }, createContext("/workspace")); + + const detached = await callPromise; + const lane = laneIdOf(detached); + expect(textOf(detached)).toContain(`Subagent detached to background lane ${lane}`); + releaseFirst(textResult("first done")); + await waitFor(() => sent.some((message) => message.details?.event === "background_done")); + + expect(textOf(await agentSend(tools, { lane_id: lane, action: "reply", prompt: "second" }))).toContain( + "Started follow-up", + ); + await waitFor(() => inputs.length === 2); + expect(inputs[1]!.sessionId).toBe(inputs[0]!.sessionId); +}); + test("session shutdown stops this session's lanes silently and leaves another session's lanes running", async () => { const gate = deferred(); const signals: AbortSignal[] = []; @@ -1098,9 +1140,13 @@ test("idle keep-alive children are stopped after the TTL and a reply resumes qui * virtual budget instead, and report what did arrive. */ const settle = async (count: number): Promise => { - for (let i = 0; i < 500; i += 1) { + for (let i = 0; i < 1_000; i += 1) { if (sent.filter((m) => m.details?.event === "background_done").length >= count) return true; await vi.advanceTimersByTimeAsync(10); + // Yield to the real event loop as well: the final notification is + // published from an async continuation, and a loaded runner can lose + // the race if we only advance virtual time. + await new Promise((resolve) => setImmediate(resolve)); } return false; }; diff --git a/packages/coding-agent/test/suite/regressions/307-compact-aborts-subagent.test.ts b/packages/coding-agent/test/suite/regressions/307-compact-aborts-subagent.test.ts new file mode 100644 index 000000000..6afe72095 --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/307-compact-aborts-subagent.test.ts @@ -0,0 +1,222 @@ +import { existsSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { AgentMessage } from "@step-harness/agent-core"; +import { fauxAssistantMessage, fauxToolCall } from "@step-harness/providers"; +import { afterEach, describe, expect, it } from "vitest"; +import { + createStepSubagentExtension, + type StepSubagentRunInput, + type StepSubagentRunResult, +} from "../../../src/features/step-subagent.ts"; +import { createHarness, type Harness } from "../harness.ts"; + +// Regression for https://github.com/stepfun-ai/Step-Code/issues/307 + +function subagentResult(text: string): StepSubagentRunResult { + return { + messages: [ + { + role: "assistant", + content: [{ type: "text", text }], + api: "anthropic-messages", + provider: "step", + model: "step-3.7-flash", + usage: { + input: 1, + output: 1, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 2, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: Date.now(), + }, + ], + stderr: "", + exitCode: 0, + usage: { input: 1, output: 1, cost: 0, contextTokens: 2, turns: 1 }, + startedAt: Date.now(), + updatedAt: Date.now(), + } as unknown as StepSubagentRunResult; +} + +function laneEvents(messages: AgentMessage[]): string[] { + return messages.flatMap((message) => { + if (message.role !== "custom" || message.customType !== "agent-notification") return []; + const event = (message.details as { event?: string } | undefined)?.event; + return event ? [event] : []; + }); +} + +async function waitFor(predicate: () => boolean, timeoutMs = 2_000): Promise { + const start = Date.now(); + while (!predicate()) { + if (Date.now() - start > timeoutMs) throw new Error("timed out waiting for condition"); + await new Promise((resolve) => setTimeout(resolve, 10)); + } +} + +describe("issue #307: manual compaction must not abort a running subagent", () => { + const harnesses: Harness[] = []; + const tempDirs: string[] = []; + + afterEach(() => { + while (harnesses.length > 0) { + harnesses.pop()?.cleanup(); + } + for (const dir of tempDirs.splice(0)) { + if (existsSync(dir)) rmSync(dir, { recursive: true, force: true }); + } + }); + + it("keeps a foreground subagent running when /compact rewrites the parent context", async () => { + const tempDir = mkdtempSync(join(tmpdir(), "step-307-")); + tempDirs.push(tempDir); + let markSubagentStarted: () => void = () => {}; + const subagentStarted = new Promise((resolve) => { + markSubagentStarted = resolve; + }); + let subagentAborted = false; + let releaseSubagent: (result: StepSubagentRunResult) => void = () => {}; + const subagentSettled = new Promise((resolve) => { + releaseSubagent = resolve; + }); + + const harness = await createHarness({ + models: [{ id: "faux-1", contextWindow: 1000, maxTokens: 1000 }], + settings: { + compaction: { contextProjection: "off", enabled: false, reserveTokens: 200, keepRecentTokens: 2 }, + }, + initialActiveToolNames: ["subagent"], + extensionFactories: [ + (pi) => { + createStepSubagentExtension({ + includeBuiltinAgents: true, + agentDir: join(tempDir, "agent"), + runner: (input: StepSubagentRunInput) => { + input.signal?.addEventListener( + "abort", + () => { + subagentAborted = true; + releaseSubagent(subagentResult("Subagent was aborted")); + }, + { once: true }, + ); + markSubagentStarted(); + return subagentSettled; + }, + })(pi); + pi.on("session_before_compact", async (event) => ({ + compaction: { + summary: `${event.reason} summary`, + firstKeptEntryId: event.preparation.firstKeptEntryId, + tokensBefore: event.preparation.tokensBefore, + details: {}, + }, + })); + }, + ], + }); + harnesses.push(harness); + + harness.setResponses([ + fauxAssistantMessage(fauxToolCall("subagent", { agent: "general", task: "long review" }), { + stopReason: "toolUse", + }), + fauxAssistantMessage("after subagent"), + ]); + + const promptPromise = harness.session.prompt("Run a long review subagent."); + await subagentStarted; + + const compactPromise = harness.session.compact(); + await new Promise((resolve) => setTimeout(resolve, 100)); + if (!subagentAborted) releaseSubagent(subagentResult("review done")); + await Promise.all([promptPromise, compactPromise]); + + expect(subagentAborted).toBe(false); + }); + + it("hands a foreground subagent to a background lane instead of waiting for it", async () => { + const tempDir = mkdtempSync(join(tmpdir(), "step-307-detach-")); + tempDirs.push(tempDir); + let markSubagentStarted: () => void = () => {}; + const subagentStarted = new Promise((resolve) => { + markSubagentStarted = resolve; + }); + let subagentAborted = false; + let subagentFinished = false; + let releaseSubagent: (result: StepSubagentRunResult) => void = () => {}; + const subagentSettled = new Promise((resolve) => { + releaseSubagent = (result) => { + subagentFinished = true; + resolve(result); + }; + }); + + const harness = await createHarness({ + models: [{ id: "faux-1", contextWindow: 1000, maxTokens: 1000 }], + settings: { + compaction: { contextProjection: "off", enabled: false, reserveTokens: 200, keepRecentTokens: 2 }, + }, + initialActiveToolNames: ["subagent"], + extensionFactories: [ + (pi) => { + createStepSubagentExtension({ + includeBuiltinAgents: true, + agentDir: join(tempDir, "agent"), + runner: (input: StepSubagentRunInput) => { + input.signal?.addEventListener( + "abort", + () => { + subagentAborted = true; + releaseSubagent(subagentResult("Subagent was aborted")); + }, + { once: true }, + ); + markSubagentStarted(); + return subagentSettled; + }, + })(pi); + pi.on("session_before_compact", async (event) => ({ + compaction: { + summary: `${event.reason} summary`, + firstKeptEntryId: event.preparation.firstKeptEntryId, + tokensBefore: event.preparation.tokensBefore, + details: {}, + }, + })); + }, + ], + }); + harnesses.push(harness); + // The harness runs extensions in print mode by default; lanes are a TUI/RPC + // capability, so bind the mode the detach path is meant for. + await harness.session.bindExtensions({ mode: "tui" }); + + harness.setResponses([ + fauxAssistantMessage(fauxToolCall("subagent", { agent: "general", task: "long review" }), { + stopReason: "toolUse", + }), + fauxAssistantMessage("after subagent"), + fauxAssistantMessage("after notification"), + ]); + + const promptPromise = harness.session.prompt("Run a long review subagent."); + await subagentStarted; + + const compactPromise = harness.session.compact(); + // Compaction must settle while the subagent is still running: the foreground + // call was handed to a lane, so there is nothing left to wait for. + await Promise.all([promptPromise, compactPromise]); + expect(subagentAborted).toBe(false); + expect(subagentFinished).toBe(false); + expect(harness.eventsOfType("compaction_end")).toHaveLength(1); + + releaseSubagent(subagentResult("review done")); + await waitFor(() => laneEvents(harness.session.messages).includes("background_done")); + expect(laneEvents(harness.session.messages)).toContain("background_done"); + }); +}); diff --git a/packages/coding-agent/test/suite/regressions/7253-manual-compact-during-response.test.ts b/packages/coding-agent/test/suite/regressions/7253-manual-compact-during-response.test.ts index 86fc473d0..f40ac77ab 100644 --- a/packages/coding-agent/test/suite/regressions/7253-manual-compact-during-response.test.ts +++ b/packages/coding-agent/test/suite/regressions/7253-manual-compact-during-response.test.ts @@ -23,7 +23,7 @@ describe("issue #7253: manual compaction during an active response", () => { } }); - it("persists the aborted response before running the requested manual compaction", async () => { + it("waits for the active response before running the requested manual compaction", async () => { let markSecondResponseStarted = () => {}; const secondResponseStarted = new Promise((resolve) => { markSecondResponseStarted = resolve; @@ -75,9 +75,16 @@ describe("issue #7253: manual compaction during an active response", () => { (entry) => entry.type === "message" && entry.message.role === "assistant" && entry.message.stopReason === "aborted", ); + const completedResponseIndex = entries.findIndex( + (entry) => + entry.type === "message" && + entry.message.role === "assistant" && + entry.message.content.some((part) => part.type === "text" && part.text.startsWith("second response:")), + ); const compactionIndex = entries.findIndex((entry) => entry.type === "compaction"); - expect(abortedResponseIndex).toBeGreaterThan(-1); - expect(compactionIndex).toBeGreaterThan(abortedResponseIndex); + expect(abortedResponseIndex).toBe(-1); + expect(completedResponseIndex).toBeGreaterThan(-1); + expect(compactionIndex).toBeGreaterThan(completedResponseIndex); expect(entries.filter((entry) => entry.type === "compaction")).toHaveLength(1); }); });