From 21a7725a0948bed2c07ba72de36177f4f03e6921 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Korm=C3=A1kur?= Date: Mon, 20 Jul 2026 15:00:09 +0000 Subject: [PATCH 1/2] feat: lift subagent async/auto-background limits via owner-routed delivery and quiescence Three-piece architecture so subagents inherit async.enabled and bash.autoBackground.enabled instead of having both force-disabled: - Owner-routed delivery: AsyncJobManager gains registerDeliverySink / waitForOwnerJobs; every AgentSession registers a sink for its own agent id, so background job results inject into the owning agent's run. Owned deliveries with no live sink dead-letter (result retained on the job row) instead of misrouting into the first top-level session. - Quiescence barrier: a subagent's final yield with owner jobs still running/undelivered is a scheduling pause, not completion. The run driver notifies the model once (hub wait/cancel), settles owner work, and folds results in as async-result follow-ups; teardown cancels and awaits surviving jobs before isolation worktree capture/cleanup. - Steering soft channel: queued steering no longer hard-aborts non-interruptible tools; it aborts interruptible waits and raises a cooperative ToolCallContext.steeringSignal. The mid-batch watch runs for every batch, and auto-backgroundable bash backgrounds itself on steer so incoming messages inject promptly with no work lost. --- packages/agent/CHANGELOG.md | 4 + packages/agent/src/agent-loop.ts | 53 +++++---- packages/agent/src/types.ts | 9 ++ packages/agent/test/run-summary.test.ts | 5 +- packages/coding-agent/CHANGELOG.md | 13 +++ .../coding-agent/src/async/job-manager.ts | 92 ++++++++++++++- .../prompts/system/subagent-async-pending.md | 6 + packages/coding-agent/src/sdk.ts | 100 +--------------- .../coding-agent/src/session/agent-session.ts | 88 ++++++++++++++ .../src/session/async-job-delivery.ts | 67 +++++++++++ packages/coding-agent/src/task/executor.ts | 71 +++++++++++- packages/coding-agent/src/tools/bash.ts | 45 ++++++-- .../test/agent-session-async-delivery.test.ts | 96 +++++++++++++++ .../test/agent-session-concurrent.test.ts | 48 ++++---- ...t-session-todo-reminder-async-jobs.test.ts | 5 +- .../test/async-job-manager.test.ts | 109 ++++++++++++++---- packages/coding-agent/test/tools.test.ts | 53 +++++++++ 17 files changed, 689 insertions(+), 175 deletions(-) create mode 100644 packages/coding-agent/src/prompts/system/subagent-async-pending.md create mode 100644 packages/coding-agent/src/session/async-job-delivery.ts create mode 100644 packages/coding-agent/test/agent-session-async-delivery.test.ts diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index 5add5603b..19c7a786f 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Changed + +- Queued steering no longer hard-aborts non-interruptible tools (e.g. `bash`): it aborts interruptible waits only and raises a cooperative steering signal (`ToolCallContext.steeringSignal`) that long-running tools may observe to finish early or background themselves. The mid-batch steering/IRC watch now runs for every tool batch instead of only batches containing an interruptible tool. + ## [17.0.5] - 2026-07-18 ### Added diff --git a/packages/agent/src/agent-loop.ts b/packages/agent/src/agent-loop.ts index c07c4af06..5925b44ff 100644 --- a/packages/agent/src/agent-loop.ts +++ b/packages/agent/src/agent-loop.ts @@ -1804,13 +1804,18 @@ async function executeToolCalls( const shouldInterruptImmediately = interruptMode !== "wait"; const steeringAbortController = new AbortController(); const ircAbortController = new AbortController(); - // Interruptible tools observe steering + external + IRC aborts; every other - // tool only sees steering + external, so an IRC-only interrupt never kills a - // partially side-effecting foreground tool (e.g. `bash`) running alongside a - // pure wait (e.g. `job` poll). - const nonInterruptibleSignal: AbortSignal = signal - ? AbortSignal.any([signal, steeringAbortController.signal]) - : steeringAbortController.signal; + // Cooperative channel: aborted when queued steering (or an interrupting + // peer IRC) is detected mid-batch. Tools receive it via tool context + // (`ctx.steeringSignal`) and MAY react — e.g. an auto-backgroundable bash + // backgrounds itself so the message injects promptly — but it never kills + // anything; ignoring it is always safe. + const steeringSoftController = new AbortController(); + // Interruptible tools (pure waits: hub wait, vibe) observe steering + + // external + IRC aborts. Every other tool sees ONLY the external signal: + // neither queued steering nor a peer IRC ever hard-kills a partially + // side-effecting foreground tool (e.g. `bash`) — those get the cooperative + // steeringSignal above, and the message injects at the next boundary. + const nonInterruptibleSignal: AbortSignal = signal ?? new AbortController().signal; const interruptibleSignal: AbortSignal = signal ? AbortSignal.any([signal, steeringAbortController.signal, ircAbortController.signal]) : AbortSignal.any([steeringAbortController.signal, ircAbortController.signal]); @@ -1876,13 +1881,16 @@ async function executeToolCalls( } } if (steeringQueued) { - // Queued steering upgrades an in-flight IRC interrupt: it aborts the - // shared signal so foreground tools stop as they do for a user Esc. + // Queued steering hard-aborts only interruptible waits and raises the + // cooperative soft signal for everything else: the boundary dequeue + // below injects the message as soon as running tools finish (or + // background themselves), and not-yet-started tools are skipped. // Idempotent — a second steer poll after the abort is a no-op. if (!steeringAbortController.signal.aborted) { interruptState.triggered = true; interruptState.source = steeringSource ?? "unknown"; steeringAbortController.abort(); + steeringSoftController.abort(); } return; } @@ -1890,11 +1898,13 @@ async function executeToolCalls( // must not re-abort, and (unlike steering above) never re-consume a queue. if (interruptState.triggered) return; if (hasIrcInterrupts && (await hasIrcInterrupts())) { - // Peer IRC only aborts interruptible waits: a foreground bash / write - // mid-execution keeps running so we never leave partial side effects. + // Peer IRC hard-aborts interruptible waits only; foreground tools keep + // running (no partial side effects) but get the cooperative soft + // signal so backgroundable work can step aside for the peer message. interruptState.triggered = true; interruptState.source = "irc"; ircAbortController.abort(); + steeringSoftController.abort(); } }; @@ -2077,12 +2087,17 @@ async function executeToolCalls( : effectiveArgs; record.args = executionArgs; + // The cooperative steering signal rides the loop-owned + // ToolCallContext (surfacing as `ctx.toolCall.steeringSignal`): + // AgentToolContext itself is app-built via declaration merging, so + // the loop cannot construct or extend one structurally. const toolContext = getToolContext ? getToolContext({ batchId, index, total: toolCalls.length, toolCalls: toolCallInfos, + steeringSignal: steeringSoftController.signal, }) : undefined; const rawResult = await tool.execute( @@ -2224,16 +2239,14 @@ async function executeToolCalls( } } - // While an interruptible tool call is in flight (e.g. a `hub` wait blocking - // on external work), queued steering or interrupting IRC would otherwise - // wait out the tool's own window. Poll only non-consuming queues and abort - // the shared tool signal so the boundary dequeue below injects the message - // promptly. Gated on immediate-interrupt mode + an interruptible call; - // checkSteering is idempotent (no-op once triggered). + // While tool calls are in flight, queued steering or interrupting IRC would + // otherwise wait out the tools' own window. Poll only non-consuming queues: + // detection hard-aborts interruptible waits, soft-signals cooperative tools + // (auto-background bash), and skips not-yet-started tools, so the boundary + // dequeue below injects the message promptly. Gated on immediate-interrupt + // mode; checkSteering is idempotent (no-op once triggered). const watchSteeringWhileRunning = - shouldInterruptImmediately && - (hasSteeringMessages !== undefined || hasIrcInterrupts !== undefined) && - records.some(record => record.interruptible); + shouldInterruptImmediately && (hasSteeringMessages !== undefined || hasIrcInterrupts !== undefined); const steeringWatchTimer = watchSteeringWhileRunning ? setInterval(() => void checkSteering(), STEERING_INTERRUPT_POLL_MS) : undefined; diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index 9fe6078ab..f72faafc2 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -458,6 +458,15 @@ export interface ToolCallContext { index: number; total: number; toolCalls: Array<{ id: string; name: string }>; + /** + * Cooperative steering signal: aborted when a queued user/steering message + * (or an interrupting peer IRC) is detected while this tool batch runs. + * Unlike the hard abort signal it NEVER kills the tool — long-running + * tools MAY observe it (via `ctx.toolCall.steeringSignal`) to finish early + * or background themselves so the message injects promptly; ignoring it is + * always safe (the message injects at the next batch boundary). + */ + steeringSignal?: AbortSignal; } /** A single tool-call content block emitted by an assistant message. */ diff --git a/packages/agent/test/run-summary.test.ts b/packages/agent/test/run-summary.test.ts index 23c85b093..fa6d6028c 100644 --- a/packages/agent/test/run-summary.test.ts +++ b/packages/agent/test/run-summary.test.ts @@ -558,7 +558,10 @@ describe("skipped tools without spans", () => { description: "slow", parameters: z.object({ value: z.string().optional() }), intent: "omit", - // concurrency: shared (default) — both run in parallel; we abort via steering. + // concurrency: shared (default) — both run in parallel. Interruptible: + // queued steering hard-aborts only interruptible waits; non-interruptible + // tools now run to completion and the steer injects at the boundary. + interruptible: true, execute: async (_id, _args, signal) => { await new Promise((resolve, reject) => { if (!signal) { diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 6f32dc885..11daeea31 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,19 @@ ## [Unreleased] +### Added + +- Added owner-routed async job delivery: every session (including subagents) registers its own delivery sink, so background bash/task results are injected into the owning agent's run instead of the first top-level session; deliveries whose owner is gone are dead-lettered with the result retained on the job row. +- Added `AsyncJobManager.registerDeliverySink` and `AsyncJobManager.waitForOwnerJobs` (with an `excludeSuppressed` filter for quiescence checks). +- Added background-on-steer for auto-backgrounded bash: an incoming user/peer message backgrounds the running command (instead of waiting it out or killing it) so the message is handled promptly. + +### Changed + +- Subagents now inherit `async.enabled` and `bash.autoBackground.enabled` from the parent instead of having both force-disabled. Subagent runs complete only after their own background jobs settle (results are folded into the run as async-result follow-ups, with a one-time notice offering `hub` wait/cancel), and teardown cancels and awaits surviving jobs before isolation worktree capture and cleanup. + +### Fixed + +- Fixed MCP tools repeatedly unmounting and remounting mid-session when server names have overlapping sanitized prefixes (e.g. `atlassian` alongside an imported `atlassian:atlassian`), and stale tools remaining registered after disconnecting a server with special characters in its name. ## [17.0.5] - 2026-07-18 ### Added diff --git a/packages/coding-agent/src/async/job-manager.ts b/packages/coding-agent/src/async/job-manager.ts index 0362379e7..5b2b9abff 100644 --- a/packages/coding-agent/src/async/job-manager.ts +++ b/packages/coding-agent/src/async/job-manager.ts @@ -60,8 +60,19 @@ export interface AsyncJob { queued?: boolean; } +/** Delivery callback for a settled job's result text. */ +export type AsyncJobDeliverySink = (jobId: string, text: string, job?: AsyncJob) => void | Promise; + export interface AsyncJobManagerOptions { - onJobComplete: (jobId: string, text: string, job?: AsyncJob) => void | Promise; + /** + * Delivery sink for UNOWNED completions (jobs registered without an + * `ownerId`). Owned deliveries route exclusively through + * {@link AsyncJobManager.registerDeliverySink}; when the owner has no live + * sink they are dead-lettered (dropped with a warning; the job row keeps + * the result text until retention eviction) — never routed here, which + * would leak one agent's result into another session. + */ + onJobComplete?: AsyncJobDeliverySink; maxRunningJobs?: number; retentionMs?: number; } @@ -128,6 +139,7 @@ export class AsyncJobManager { readonly #watchedJobs = new Set(); readonly #evictionTimers = new Map(); readonly #pollEscalation = new Map(); + readonly #deliverySinks = new Map(); readonly #onJobComplete: AsyncJobManagerOptions["onJobComplete"]; readonly #maxRunningJobs: number; readonly #retentionMs: number; @@ -409,6 +421,55 @@ export class AsyncJobManager { await Promise.all(Array.from(this.#jobs.values()).map(job => job.promise)); } + /** + * Route completions for jobs owned by `ownerId` to `sink`. Sessions register + * their own sink at construction and unregister on dispose. Owned deliveries + * with no live sink are dead-lettered — `onJobComplete` serves only unowned + * deliveries. + * + * Last registration wins for an owner id; the returned unregister clears the + * mapping only while it still points at `sink`, so a revived session's fresh + * registration survives its parked predecessor's late cleanup. + */ + registerDeliverySink(ownerId: string, sink: AsyncJobDeliverySink): () => void { + this.#deliverySinks.set(ownerId, sink); + return () => { + if (this.#deliverySinks.get(ownerId) === sink) this.#deliverySinks.delete(ownerId); + }; + } + + /** + * Wait until every job owned by `ownerId` has settled — its run promise + * resolved, which for cancelled jobs means the underlying process actually + * exited. Jobs registered while waiting (e.g. by a follow-up turn) are + * awaited too. Returns false when `timeoutMs` elapses first. + * + * `excludeSuppressed` skips jobs whose delivery is suppressed (acknowledged + * or `hub`-watched): those can never re-wake a run, so quiescence barriers + * pass it to share one contract with the pending-async-wake predicate. + * Teardown reaps omit it — worktree safety concerns every owner process. + */ + async waitForOwnerJobs( + ownerId: string, + options?: { timeoutMs?: number; excludeSuppressed?: boolean }, + ): Promise { + const deadline = + options?.timeoutMs === undefined ? Number.POSITIVE_INFINITY : Date.now() + Math.max(0, options.timeoutMs); + const awaited = new Set(); + for (;;) { + const pending = this.#filterJobs(this.#jobs.values(), { ownerId }).filter( + job => !awaited.has(job.id) && (options?.excludeSuppressed !== true || !this.isDeliverySuppressed(job.id)), + ); + if (pending.length === 0) return true; + for (const job of pending) awaited.add(job.id); + const settled = await this.#waitForDeliveryPromise( + Promise.all(pending.map(job => job.promise)).then(() => {}), + deadline, + ); + if (!settled) return false; + } + } + async #waitForAllUntil(deadline: number): Promise { const promises = Array.from(this.#jobs.values()).map(job => job.promise); if (promises.length === 0) return true; @@ -489,6 +550,7 @@ export class AsyncJobManager { this.#suppressedDeliveries.clear(); this.#watchedJobs.clear(); this.#pollEscalation.clear(); + this.#deliverySinks.clear(); return jobsSettled && drained; } @@ -655,11 +717,37 @@ export class AsyncJobManager { } } + /** + * Resolve the sink for one delivery attempt: owned deliveries route ONLY to + * their owner's registered sink (a missing sink dead-letters — never the + * default, which would misroute a dead owner's result into another + * session); unowned deliveries use the constructor default. Resolved per + * attempt so a sink registered between retries (e.g. a revived session) + * picks up the retry. + */ + #resolveDeliverySink(ownerId: string | undefined): AsyncJobDeliverySink | undefined { + if (ownerId !== undefined) return this.#deliverySinks.get(ownerId); + return this.#onJobComplete; + } + #deliverDelivery(delivery: AsyncJobDelivery): Promise { + const sink = this.#resolveDeliverySink(delivery.ownerId); + if (!sink) { + // Dead-letter: owned delivery with no live sink (session disposed or + // parked), or unowned delivery with no default sink. Drop it — the + // job row keeps its result/error text until retention eviction, so + // the outcome stays inspectable via job queries and agent:// reads. + logger.warn("Async job delivery dead-lettered: no delivery sink", { + jobId: delivery.jobId, + ownerId: delivery.ownerId, + }); + delivery.promise = Promise.resolve(); + return delivery.promise; + } const promise = (async () => { this.#inFlightDeliveries.push(delivery); try { - await this.#onJobComplete(delivery.jobId, delivery.text, this.#jobs.get(delivery.jobId)); + await sink(delivery.jobId, delivery.text, this.#jobs.get(delivery.jobId)); } catch (error) { delivery.attempt += 1; delivery.lastError = error instanceof Error ? error.message : String(error); diff --git a/packages/coding-agent/src/prompts/system/subagent-async-pending.md b/packages/coding-agent/src/prompts/system/subagent-async-pending.md new file mode 100644 index 000000000..89bf568cb --- /dev/null +++ b/packages/coding-agent/src/prompts/system/subagent-async-pending.md @@ -0,0 +1,6 @@ +Your yield was recorded, but {{count}} background job{{#if multiple}}s{{/if}} you own {{#if multiple}}are{{else}}is{{/if}} still running: {{jobs}}. + +This run completes only after these jobs settle; their results arrive as follow-up messages. Decide now: +- Need the results? Wait for them (`hub` op:"wait"), then submit a fresh `yield` that incorporates them. +- Job no longer needed? Cancel it (`hub` op:"cancel", ids:[...]) and re-yield. +- Otherwise say nothing further; your current yield stands once the jobs finish. diff --git a/packages/coding-agent/src/sdk.ts b/packages/coding-agent/src/sdk.ts index af123be59..40968c997 100644 --- a/packages/coding-agent/src/sdk.ts +++ b/packages/coding-agent/src/sdk.ts @@ -32,7 +32,7 @@ import { formatActiveRepoWatchdogPrompt, formatAdvisorContextPrompt, } from "./advisor"; -import { type AsyncJob, AsyncJobManager } from "./async"; +import { AsyncJobManager } from "./async"; import { AutoLearnController, buildAutoLearnInstructions } from "./autolearn/controller"; import { createAutoresearchExtension } from "./autoresearch"; import { loadCapability } from "./capability"; @@ -106,7 +106,6 @@ import { import { MCP_CONNECTION_STATUS_EVENT_CHANNEL, type McpConnectionStatusEvent } from "./mcp/startup-events"; import { createSessionMemoryRuntimeContext, resolveMemoryBackend } from "./memory-backend"; import type { MnemopiSessionState } from "./mnemopi/state"; -import asyncResultTemplate from "./prompts/tools/async-result.md" with { type: "text" }; import lateDiagnosticTemplate from "./prompts/tools/lsp-late-diagnostic.md" with { type: "text" }; import { AgentLifecycleManager } from "./registry/agent-lifecycle"; import { AgentRegistry, MAIN_AGENT_ID } from "./registry/agent-registry"; @@ -201,60 +200,11 @@ import { EventBus } from "./utils/event-bus"; import { buildNamedToolChoice } from "./utils/tool-choice"; import { buildWorkspaceTree, type WorkspaceTree } from "./workspace-tree"; -type AsyncResultEntry = { - jobId: string; - result: string; - job: AsyncJob | undefined; - durationMs: number | undefined; -}; - -type AsyncResultJobDetails = { - jobId: string; - type?: "bash" | "task"; - label?: string; - durationMs?: number; -}; - -type AsyncResultDetails = { - jobs: AsyncResultJobDetails[]; -}; - type McpNotificationEntry = { serverName: string; uri: string; }; -function buildAsyncResultBatchMessage(entries: AsyncResultEntry[]): CustomMessage | null { - if (entries.length === 0) return null; - const jobs = entries.map(entry => ({ - jobId: entry.jobId, - result: entry.result, - type: entry.job?.type, - label: entry.job?.label, - durationMs: entry.durationMs, - })); - const details: AsyncResultDetails = { - jobs: jobs.map(job => ({ - jobId: job.jobId, - type: job.type, - label: job.label, - durationMs: job.durationMs, - })), - }; - return { - role: "custom", - customType: "async-result", - content: prompt.render(asyncResultTemplate, { - multiple: jobs.length > 1, - jobs, - }), - display: true, - attribution: "agent", - details, - timestamp: Date.now(), - }; -} - type LateDiagnosticsDetails = { files: Array<{ path: string; summary: string; errored: boolean; messages: string[] }>; }; @@ -1562,28 +1512,6 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {} const restrictToolNames = options.restrictToolNames === true; const enableLsp = !restrictToolNames && (options.enableLsp ?? true); const asyncMaxJobs = Math.min(100, Math.max(1, settings.get("async.maxJobs") ?? 100)); - const ASYNC_INLINE_RESULT_MAX_CHARS = 12_000; - const ASYNC_PREVIEW_MAX_CHARS = 4_000; - const formatAsyncResultForFollowUp = async (result: string): Promise => { - if (result.length <= ASYNC_INLINE_RESULT_MAX_CHARS) { - return result; - } - - const preview = `${result.slice(0, ASYNC_PREVIEW_MAX_CHARS)}\n\n[Output truncated. Showing first ${ASYNC_PREVIEW_MAX_CHARS.toLocaleString()} characters.]`; - try { - const { path: artifactPath, id: artifactId } = await sessionManager.allocateArtifactPath("async"); - if (artifactPath && artifactId) { - await Bun.write(artifactPath, result); - return `${preview}\nFull output: artifact://${artifactId}`; - } - } catch (error) { - logger.warn("Failed to persist async follow-up artifact", { - error: error instanceof Error ? error.message : String(error), - }); - } - - return preview; - }; // Only the first top-level session in a process owns an AsyncJobManager. // Subagents inherit the parent's manager via `AsyncJobManager.instance()` // (set below), and any additional top-level session spun up in-process @@ -1592,24 +1520,12 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {} // owning session's manager and break the `task`/`bash` async paths // (issue #1923). The `instance()` guard means later sessions also skip // constructing an orphaned manager that nothing would ever route to. + // Delivery is owner-routed: every AgentSession registers its own sink + // (see session/async-job-delivery.ts), so the manager takes no default + // onJobComplete here. const asyncJobManager = !options.parentTaskPrefix && !AsyncJobManager.instance() - ? new AsyncJobManager({ - maxRunningJobs: asyncMaxJobs, - onJobComplete: async (jobId, result, job) => { - if (!session || asyncJobManager!.isDeliverySuppressed(jobId)) return; - const formattedResult = await formatAsyncResultForFollowUp(result); - if (asyncJobManager!.isDeliverySuppressed(jobId)) return; - - const durationMs = job ? Math.max(0, Date.now() - job.startTime) : undefined; - session.yieldQueue.enqueue("async-result", { - jobId, - result: formattedResult, - job, - durationMs, - }); - }, - }) + ? new AsyncJobManager({ maxRunningJobs: asyncMaxJobs }) : undefined; const scopedAsyncJobManager = asyncJobManager ?? (options.parentTaskPrefix ? AsyncJobManager.instance() : undefined); @@ -2988,12 +2904,6 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {} titleSystemPrompt: options.titleSystemPrompt, }); hasSession = true; - if (asyncJobManager) { - session.yieldQueue.register("async-result", { - isStale: entry => asyncJobManager.isDeliverySuppressed(entry.jobId), - build: buildAsyncResultBatchMessage, - }); - } session.yieldQueue.register("mcp-notification", { build: buildMcpNotificationBatchMessage, }); diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index e39ca813d..07e90e67e 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -348,6 +348,12 @@ import { formatLocalCalendarDate } from "../utils/local-date"; import { generateSessionTitle } from "../utils/title-generator"; import { buildNamedToolChoice, isToolChoiceActive } from "../utils/tool-choice"; import type { VibeModeState } from "../vibe/state"; +import { + ASYNC_INLINE_RESULT_MAX_CHARS, + ASYNC_PREVIEW_MAX_CHARS, + type AsyncResultEntry, + buildAsyncResultBatchMessage, +} from "./async-job-delivery"; import type { AuthStorage } from "./auth-storage"; import type { ClientBridge, ClientBridgePermissionOption, ClientBridgePermissionOutcome } from "./client-bridge"; import { @@ -2010,6 +2016,8 @@ export class AgentSession { * undefined to avoid reading the primary's jobs. */ readonly #asyncJobManager: AsyncJobManager | undefined; + /** Clears this session's owner delivery sink registration; set when a manager + agent id exist. */ + #unregisterAsyncDeliverySink: (() => void) | undefined; #pendingPythonMessages: PythonExecutionMessage[] = []; #activeEvalExecutions = new Set>(); #evalExecutionDisposing = false; @@ -2838,6 +2846,21 @@ export class AgentSession { this.#providerSessionId = config.providerSessionId; this.#inheritedProviderPromptCacheKey = config.providerPromptCacheKeySource === "fork" ? this.agent.promptCacheKey : undefined; + // Owner-routed async delivery: completions for jobs this agent owns are + // injected into THIS session's run as async-result follow-ups. Without a + // registered sink the manager dead-letters owned deliveries, so this + // registration is what makes background jobs usable — for the main + // session and for subagents inheriting the process manager alike. + if (this.#asyncJobManager && this.#agentId) { + const manager = this.#asyncJobManager; + this.#unregisterAsyncDeliverySink = manager.registerDeliverySink(this.#agentId, (jobId, text, job) => + this.#deliverAsyncJobResult(manager, jobId, text, job), + ); + this.yieldQueue.register("async-result", { + isStale: entry => manager.isDeliverySuppressed(entry.jobId), + build: buildAsyncResultBatchMessage, + }); + } this.agent.setAssistantMessageEventInterceptor((message, assistantMessageEvent) => { const event: AgentEvent = { type: "message_update", @@ -4064,6 +4087,67 @@ export class AgentSession { ); } + /** + * Public view of the pending-async-wake state for run drivers: true while + * owner-scoped async work can still re-wake this session's run (a running + * background job with an unsuppressed delivery, or a queued / in-flight + * delivery). The task executor's quiescence barrier polls this to + * distinguish a scheduling pause from terminal completion. + */ + hasPendingAsyncWork(): boolean { + return this.#hasPendingAsyncWake(); + } + + /** + * Settle one generation of owner-scoped async work: wait for running owner + * jobs to finish, deliver their queued results (which enqueue async-result + * follow-ups on this session's yield queue), and wait for the injected + * follow-up turn(s) to go idle. Callers loop while + * {@link hasPendingAsyncWork} still holds — a follow-up turn may start new + * jobs. + */ + async settleAsyncWork(): Promise { + const manager = this.#asyncJobManager; + if (!manager || !this.#agentId) return; + await manager.waitForOwnerJobs(this.#agentId, { excludeSuppressed: true }); + await manager.drainDeliveries({ filter: { ownerId: this.#agentId } }); + await this.waitForIdle(); + } + + /** + * Delivery sink for async jobs owned by this agent: format the result + * (spilling oversized output to an artifact) and enqueue it as an + * async-result follow-up on the yield queue. The queue's idle flush starts + * the follow-up turn when the session is between turns. + */ + async #deliverAsyncJobResult(manager: AsyncJobManager, jobId: string, text: string, job?: AsyncJob): Promise { + if (this.#isDisposed) return; + if (manager.isDeliverySuppressed(jobId)) return; + const formatted = await this.#formatAsyncResultForFollowUp(text); + if (manager.isDeliverySuppressed(jobId)) return; + const durationMs = job ? Math.max(0, Date.now() - job.startTime) : undefined; + this.yieldQueue.enqueue("async-result", { jobId, result: formatted, job, durationMs }); + } + + async #formatAsyncResultForFollowUp(result: string): Promise { + if (result.length <= ASYNC_INLINE_RESULT_MAX_CHARS) { + return result; + } + const preview = `${result.slice(0, ASYNC_PREVIEW_MAX_CHARS)}\n\n[Output truncated. Showing first ${ASYNC_PREVIEW_MAX_CHARS.toLocaleString()} characters.]`; + try { + const { path: artifactPath, id: artifactId } = await this.sessionManager.allocateArtifactPath("async"); + if (artifactPath && artifactId) { + await Bun.write(artifactPath, result); + return `${preview}\nFull output: artifact://${artifactId}`; + } + } catch (error) { + logger.warn("Failed to persist async follow-up artifact", { + error: error instanceof Error ? error.message : String(error), + }); + } + return preview; + } + // ========================================================================= // Event Subscription // ========================================================================= @@ -6782,6 +6866,10 @@ export class AgentSession { } async #disposeOwnedAsyncJobs(): Promise { + // Unregister before cancelling: a job completing during teardown must + // dead-letter rather than enqueue a follow-up into a disposing session. + this.#unregisterAsyncDeliverySink?.(); + this.#unregisterAsyncDeliverySink = undefined; this.#cancelOwnAsyncJobs(); const manager = this.#ownedAsyncJobManager; if (!manager) return; diff --git a/packages/coding-agent/src/session/async-job-delivery.ts b/packages/coding-agent/src/session/async-job-delivery.ts new file mode 100644 index 000000000..5565bbf28 --- /dev/null +++ b/packages/coding-agent/src/session/async-job-delivery.ts @@ -0,0 +1,67 @@ +/** + * Owner-routed async job delivery: formatting and batch-message assembly for + * `async-result` follow-ups. + * + * Each {@link AgentSession} registers a delivery sink for its own agent id + * (`AsyncJobManager.registerDeliverySink`) and enqueues formatted entries on + * its yield queue; the queue's idle flush injects them as a follow-up turn. + * This replaces the old single hardwired `onJobComplete` closure that routed + * every completion — regardless of owner — into the first top-level session. + */ +import { prompt } from "@oh-my-pi/pi-utils"; +import type { AsyncJob } from "../async"; +import asyncResultTemplate from "../prompts/tools/async-result.md" with { type: "text" }; +import type { CustomMessage } from "./messages"; + +/** Result payloads longer than this spill to an artifact with an inline preview. */ +export const ASYNC_INLINE_RESULT_MAX_CHARS = 12_000; +export const ASYNC_PREVIEW_MAX_CHARS = 4_000; + +export interface AsyncResultEntry { + jobId: string; + result: string; + job: AsyncJob | undefined; + durationMs: number | undefined; +} + +type AsyncResultJobDetails = { + jobId: string; + type?: "bash" | "task"; + label?: string; + durationMs?: number; +}; + +export type AsyncResultDetails = { + jobs: AsyncResultJobDetails[]; +}; + +export function buildAsyncResultBatchMessage(entries: AsyncResultEntry[]): CustomMessage | null { + if (entries.length === 0) return null; + const jobs = entries.map(entry => ({ + jobId: entry.jobId, + result: entry.result, + type: entry.job?.type, + label: entry.job?.label, + durationMs: entry.durationMs, + })); + const details: AsyncResultDetails = { + jobs: jobs.map(job => ({ + jobId: job.jobId, + type: job.type, + label: job.label, + durationMs: job.durationMs, + })), + }; + return { + role: "custom", + customType: "async-result", + content: prompt.render(asyncResultTemplate, { + multiple: jobs.length > 1, + jobs, + }), + display: true, + attribution: "agent", + details, + timestamp: Date.now(), + }; +} diff --git a/packages/coding-agent/src/task/executor.ts b/packages/coding-agent/src/task/executor.ts index c5cd02694..6a54bb9bc 100644 --- a/packages/coding-agent/src/task/executor.ts +++ b/packages/coding-agent/src/task/executor.ts @@ -9,6 +9,7 @@ import type { AgentEvent, AgentIdentity, AgentTelemetryConfig } from "@oh-my-pi/ import { recordHandoff, resolveTelemetry } from "@oh-my-pi/pi-agent-core"; import type { Api, Model, ServiceTierByFamily, Usage } from "@oh-my-pi/pi-ai"; import { logger, popLoopPhase, prompt, pushLoopPhase, untilAborted } from "@oh-my-pi/pi-utils"; +import { AsyncJobManager } from "../async"; import type { Rule } from "../capability/rule"; import { ModelRegistry } from "../config/model-registry"; import { @@ -33,6 +34,7 @@ import type { LocalProtocolOptions } from "../internal-urls"; import { callTool } from "../mcp/client"; import type { MCPManager } from "../mcp/manager"; import type { MnemopiSessionState } from "../mnemopi/state"; +import subagentAsyncPendingTemplate from "../prompts/system/subagent-async-pending.md" with { type: "text" }; import subagentSystemPromptTemplate from "../prompts/system/subagent-system-prompt.md" with { type: "text" }; import submitReminderTemplate from "../prompts/system/subagent-yield-reminder.md" with { type: "text" }; import { AgentLifecycleManager } from "../registry/agent-lifecycle"; @@ -815,8 +817,11 @@ export function createSubagentSettings( snapshot["tier.google"] = subagentTiers.google ?? "none"; return Settings.isolated({ ...snapshot, - "async.enabled": false, - "bash.autoBackground.enabled": false, + // Async jobs and bash auto-backgrounding are inherited from the parent: + // background jobs are owner-routed to the subagent's own session, and + // the run driver's quiescence barrier + teardown reap guarantee no + // owner job outlives the run, so worktree capture/cleanup stays + // race-free (previously both were force-disabled here). // Subagents run headless — there is no UI to confirm prompts against, so // the parent task approval is the authorization boundary. Use yolo mode @@ -1717,6 +1722,54 @@ async function driveSessionToYield( } } + // Quiescence barrier (structured concurrency): a final yield with owner + // background jobs still running or undelivered is a scheduling pause, + // not run completion. Wait for the jobs to settle, let their + // async-result follow-up turns run to idle, and only then classify the + // terminal state — the isolation runner captures and destroys the + // worktree right after this run resolves, so no owner job that could + // still re-wake the session may outlive it. Suppressed (acknowledged / + // hub-watched) jobs never re-wake the run and are reaped at teardown. + // + // Before blocking on running jobs, tell the model ONCE what it is + // waiting on so it can `hub` wait/cancel instead of sitting silent + // until the jobs (or the runtime limit) expire. Guarded to a single + // notice per run — a model that yields again with jobs still running + // is then waited out passively. Runs that never yielded (ladder + // exhausted / terminal model error) skip the barrier entirely — more + // injected turns just multiply the failure noise; the teardown reap + // still cancels and awaits their jobs before worktree capture. + let asyncPendingNoticeSent = false; + while (monitor.yieldCalled() && !abortSignal.aborted && session.hasPendingAsyncWork()) { + if (!asyncPendingNoticeSent) { + asyncPendingNoticeSent = true; + const running = session.getAsyncJobSnapshot()?.running ?? []; + if (running.length > 0) { + const jobs = running.map(job => `${job.id}${job.label ? ` (${job.label})` : ""}`).join(", "); + const notice = prompt.render(subagentAsyncPendingTemplate, { + count: running.length, + multiple: running.length > 1, + jobs, + }); + try { + await awaitAbortable(session.prompt(notice, { attribution: "agent", synthetic: true })); + await awaitAbortable(session.waitForIdle()); + } catch (err) { + if (abortSignal.aborted || err instanceof ToolAbortError) throw err; + // A failed notice turn must not kill the run — fall through + // to the passive settle below. + logger.warn("Subagent async-pending notice failed", { + error: err instanceof Error ? err.message : String(err), + }); + } + // Re-evaluate: the notice turn may have cancelled, watched, or + // absorbed the jobs. + continue; + } + } + await awaitAbortable(session.settleAsyncWork()); + } + if (monitor.yieldCalled()) { await session.waitForIdle(); } else { @@ -2768,6 +2821,20 @@ export async function runSubprocess(options: ExecutorOptions): Promise { + steeringSignal?: AbortSignal, + ): Promise { if (signal?.aborted) { return { kind: "aborted" }; } + if (steeringSignal?.aborted) { + return { kind: "steer" }; + } - const waiters: Array> = [ - job.completion, - Bun.sleep(thresholdMs).then(() => ({ kind: "running" as const })), - ]; + const waiters: Array< + Promise + > = [job.completion, Bun.sleep(thresholdMs).then(() => ({ kind: "running" as const }))]; - if (!signal) { + if (!signal && !steeringSignal) { return await Promise.race(waiters); } const { promise: abortedPromise, resolve: resolveAborted } = Promise.withResolvers<{ kind: "aborted" }>(); const onAbort = () => resolveAborted({ kind: "aborted" }); - signal.addEventListener("abort", onAbort, { once: true }); - waiters.push(abortedPromise); + const { promise: steerPromise, resolve: resolveSteer } = Promise.withResolvers<{ kind: "steer" }>(); + const onSteer = () => resolveSteer({ kind: "steer" }); + if (signal) { + signal.addEventListener("abort", onAbort, { once: true }); + waiters.push(abortedPromise); + } + if (steeringSignal) { + steeringSignal.addEventListener("abort", onSteer, { once: true }); + waiters.push(steerPromise); + } try { return await Promise.race(waiters); } finally { - signal.removeEventListener("abort", onAbort); + signal?.removeEventListener("abort", onAbort); + steeringSignal?.removeEventListener("abort", onSteer); } } @@ -903,7 +915,12 @@ export class BashTool implements AgentTool { + let session: AgentSession; + let tempDir: string; + const authStorages: AuthStorage[] = []; + + beforeEach(() => { + tempDir = path.join(os.tmpdir(), `pi-async-delivery-test-${Snowflake.next()}`); + fs.mkdirSync(tempDir, { recursive: true }); + }); + + afterEach(async () => { + if (session) { + await session.dispose(); + } + for (const authStorage of authStorages.splice(0)) { + authStorage.close(); + } + if (tempDir && fs.existsSync(tempDir)) { + removeSyncWithRetries(tempDir); + } + AsyncJobManager.resetForTests(); + }); + + it("injects an owned completion as a follow-up turn and reaches quiescence", async () => { + const model = getBundledModel("anthropic", "claude-sonnet-4-5")!; + const mock = createMockModel({ handler: () => ({ content: ["Done"] }) }); + const agent = new Agent({ + getApiKey: () => "test-key", + initialState: { model, systemPrompt: ["Test"], tools: [] }, + convertToLlm, + streamFn: mock.stream, + }); + const authStorage = await AuthStorage.create(path.join(tempDir, "auth.db")); + authStorages.push(authStorage); + authStorage.setRuntimeApiKey("anthropic", "test-key"); + const manager = new AsyncJobManager({}); + AsyncJobManager.setInstance(manager); + + session = new AgentSession({ + agent, + sessionManager: SessionManager.inMemory(), + settings: Settings.isolated(), + modelRegistry: new ModelRegistry(authStorage), + agentId: "SubAgent", + asyncJobManager: manager, + }); + + const gate = Promise.withResolvers(); + manager.register("bash", "gated job", () => gate.promise, { id: "sub-job", ownerId: "SubAgent" }); + + // A running owned job holds the session out of quiescence. + expect(session.hasPendingAsyncWork()).toBe(true); + + gate.resolve("job finished: ALL GREEN"); + await session.settleAsyncWork(); + + // The completion routed to THIS session (not a global default sink) and + // ran as a follow-up turn whose context carries the job result. + expect(session.hasPendingAsyncWork()).toBe(false); + const sawResult = mock.calls.some(call => + call.context.messages.some(message => { + if (typeof message.content === "string") { + return message.content.includes("ALL GREEN"); + } + return ( + Array.isArray(message.content) && + message.content.some(content => content.type === "text" && content.text.includes("ALL GREEN")) + ); + }), + ); + expect(sawResult).toBe(true); + }); +}); diff --git a/packages/coding-agent/test/agent-session-concurrent.test.ts b/packages/coding-agent/test/agent-session-concurrent.test.ts index ef10ebae3..b3a9044d8 100644 --- a/packages/coding-agent/test/agent-session-concurrent.test.ts +++ b/packages/coding-agent/test/agent-session-concurrent.test.ts @@ -917,19 +917,6 @@ describe("AgentSession concurrent prompt guard", () => { const asyncJobManager = new AsyncJobManager({ maxRunningJobs: 2, retentionMs: 1_000, - onJobComplete: async () => { - deliveryStarted = true; - await deliveryGate.promise; - await session.sendCustomMessage( - { - customType: "async-result", - content: "Background result", - display: true, - attribution: "agent", - }, - { deliverAs: "followUp", triggerTurn: true }, - ); - }, }); AsyncJobManager.setInstance(asyncJobManager); @@ -945,6 +932,21 @@ describe("AgentSession concurrent prompt guard", () => { capabilities: {}, deferAgentInitiatedTurns: true, }); + // Override the session's self-registered sink: the test gates delivery + // and reproduces the ACP follow-up injection explicitly. + asyncJobManager.registerDeliverySink(ownerId, async () => { + deliveryStarted = true; + await deliveryGate.promise; + await session.sendCustomMessage( + { + customType: "async-result", + content: "Background result", + display: true, + attribution: "agent", + }, + { deliverAs: "followUp", triggerTurn: true }, + ); + }); await session.prompt("First message"); expect(session.isStreaming).toBe(false); @@ -994,13 +996,6 @@ describe("AgentSession concurrent prompt guard", () => { const asyncJobManager = new AsyncJobManager({ maxRunningJobs: 3, retentionMs: 1_000, - onJobComplete: async jobId => { - started.add(jobId); - if (jobId === "job-a") { - await deliveryGate.promise; - } - delivered.push(jobId); - }, }); AsyncJobManager.setInstance(asyncJobManager); @@ -1030,6 +1025,19 @@ describe("AgentSession concurrent prompt guard", () => { agentId: "acp-session-a", ownedAsyncJobManager: asyncJobManager, }); + // Override both sessions' self-registered sinks so the test controls + // delivery timing and records routing order. + asyncJobManager.registerDeliverySink("acp-session-a", async jobId => { + started.add(jobId); + if (jobId === "job-a") { + await deliveryGate.promise; + } + delivered.push(jobId); + }); + asyncJobManager.registerDeliverySink("acp-session-b", async jobId => { + started.add(jobId); + delivered.push(jobId); + }); try { asyncJobManager.register("bash", "A", async () => "A", { id: "job-a", ownerId: "acp-session-a" }); diff --git a/packages/coding-agent/test/agent-session-todo-reminder-async-jobs.test.ts b/packages/coding-agent/test/agent-session-todo-reminder-async-jobs.test.ts index 2e8f93c9a..b0b167ec4 100644 --- a/packages/coding-agent/test/agent-session-todo-reminder-async-jobs.test.ts +++ b/packages/coding-agent/test/agent-session-todo-reminder-async-jobs.test.ts @@ -113,7 +113,7 @@ describe("AgentSession todo reminder async-job deferral", () => { authStorage.setRuntimeApiKey("anthropic", "test-key"); modelRegistry = new ModelRegistry(authStorage); sessionManager = SessionManager.create(tempDir.path(), tempDir.path()); - manager = new AsyncJobManager({ onJobComplete: async () => {} }); + manager = new AsyncJobManager({}); gates = []; extensionRunner = { emit: vi.fn().mockResolvedValue(undefined), @@ -148,6 +148,9 @@ describe("AgentSession todo reminder async-job deferral", () => { asyncJobManager: manager, extensionRunner, }); + // Override the session's self-registered sink with a no-op: these tests + // exercise the async-wake deferral gates, not result injection. + manager.registerDeliverySink("Main", () => {}); reminderAttempts = []; ({ promise: firstReminderPromise, resolve: resolveFirstReminder } = Promise.withResolvers()); diff --git a/packages/coding-agent/test/async-job-manager.test.ts b/packages/coding-agent/test/async-job-manager.test.ts index fd0cae496..303141ab5 100644 --- a/packages/coding-agent/test/async-job-manager.test.ts +++ b/packages/coding-agent/test/async-job-manager.test.ts @@ -303,16 +303,13 @@ describe("AsyncJobManager", () => { releaseMainDelivery = resolve; }); const subagentCompletions: Array<{ jobId: string; text: string }> = []; - const manager = new AsyncJobManager({ - retentionMs: 0, - onJobComplete: async (jobId, text) => { - if (jobId === mainJobId) { - notifyMainDeliveryStarted(); - await mainDeliveryReleased; - return; - } - subagentCompletions.push({ jobId, text }); - }, + const manager = new AsyncJobManager({ retentionMs: 0 }); + manager.registerDeliverySink("0-Main", async () => { + notifyMainDeliveryStarted(); + await mainDeliveryReleased; + }); + manager.registerDeliverySink("3-AuthLoader", (jobId, text) => { + subagentCompletions.push({ jobId, text }); }); mainJobId = manager.register("task", "main job", async () => "main result", { ownerId: "0-Main" }); @@ -355,19 +352,15 @@ describe("AsyncJobManager", () => { releaseTargetDelivery = resolve; }); const completions: string[] = []; - const manager = new AsyncJobManager({ - onJobComplete: async jobId => { - if (jobId === mainJobId) { - notifyMainDeliveryStarted(); - await mainDeliveryReleased; - return; - } - if (jobId === targetJobId) { - notifyTargetDeliveryStarted(); - await targetDeliveryReleased; - completions.push(jobId); - } - }, + const manager = new AsyncJobManager({}); + manager.registerDeliverySink("0-Main", async () => { + notifyMainDeliveryStarted(); + await mainDeliveryReleased; + }); + manager.registerDeliverySink("3-AuthLoader", async jobId => { + notifyTargetDeliveryStarted(); + await targetDeliveryReleased; + completions.push(jobId); }); mainJobId = manager.register("task", "main job", async () => "main result", { ownerId: "0-Main" }); @@ -437,6 +430,76 @@ describe("AsyncJobManager", () => { await manager.waitForAll(); expect(manager.getJob(parentJobId)?.status).toBe("cancelled"); }); + + test("routes owned deliveries to the owner's registered sink only", async () => { + const mainDeliveries: string[] = []; + const defaultDeliveries: string[] = []; + const manager = new AsyncJobManager({ + onJobComplete: async jobId => { + defaultDeliveries.push(jobId); + }, + }); + manager.registerDeliverySink("Main", jobId => { + mainDeliveries.push(jobId); + }); + + manager.register("bash", "owned", async () => "ok", { id: "owned-1", ownerId: "Main" }); + manager.register("bash", "unowned", async () => "ok", { id: "unowned-1" }); + await manager.waitForAll(); + await manager.drainDeliveries({ timeoutMs: 500 }); + + expect(mainDeliveries).toEqual(["owned-1"]); + expect(defaultDeliveries).toEqual(["unowned-1"]); + }); + + test("dead-letters an owned delivery when its owner has no live sink", async () => { + const defaultDeliveries: string[] = []; + const manager = new AsyncJobManager({ + onJobComplete: async jobId => { + defaultDeliveries.push(jobId); + }, + }); + const unregister = manager.registerDeliverySink("Sub", () => {}); + unregister(); + + manager.register("bash", "orphan", async () => "orphan result", { id: "orphan-1", ownerId: "Sub" }); + await manager.waitForAll(); + const drained = await manager.drainDeliveries({ timeoutMs: 500 }); + + // Dead-letter drops the delivery (drain settles) without misrouting it + // into the default sink; the outcome stays readable on the job row. + expect(drained).toBe(true); + expect(defaultDeliveries).toEqual([]); + expect(manager.getJob("orphan-1")?.resultText).toBe("orphan result"); + }); + + test("waitForOwnerJobs settles cancelled jobs and skips suppressed ones on request", async () => { + const manager = new AsyncJobManager({}); + manager.register( + "bash", + "hung", + async ({ signal }) => { + await new Promise(resolve => { + if (signal.aborted) return resolve(); + signal.addEventListener("abort", () => resolve(), { once: true }); + }); + return "stopped"; + }, + { id: "hung-1", ownerId: "Sub" }, + ); + + // Quiescence-barrier contract: a watched (suppressed) job can never + // re-wake a run, so the filtered wait treats it as settled. + manager.watchJobs(["hung-1"]); + await expect(manager.waitForOwnerJobs("Sub", { excludeSuppressed: true })).resolves.toBe(true); + + // Teardown-reap contract: the unfiltered wait blocks until the + // cancelled job's body actually finishes. + const reap = manager.waitForOwnerJobs("Sub", { timeoutMs: 1_000 }); + manager.cancelAll({ ownerId: "Sub" }); + await expect(reap).resolves.toBe(true); + expect(manager.getJob("hung-1")?.status).toBe("cancelled"); + }); }); describe("AsyncJobManager smart poll-wait escalation", () => { diff --git a/packages/coding-agent/test/tools.test.ts b/packages/coding-agent/test/tools.test.ts index e6f48e0d7..be9c873df 100644 --- a/packages/coding-agent/test/tools.test.ts +++ b/packages/coding-agent/test/tools.test.ts @@ -1468,6 +1468,59 @@ function b() { await asyncJobManager.dispose(); }); + it("backgrounds a running command when the steering signal fires mid-wait", async () => { + const asyncJobManager = new AsyncJobManager({}); + const autoBackgroundBashTool = wrapToolWithMetaNotice( + new BashTool( + createTestToolSession( + testDir, + Settings.isolated({ + "bash.autoBackground.enabled": true, + // High threshold: only the steering signal can background this. + "bash.autoBackground.thresholdMs": 60_000, + }), + { + getSessionId: () => "test-session", + asyncJobManager, + }, + ), + ), + ); + + const steering = new AbortController(); + steering.abort(); + const result = await autoBackgroundBashTool.execute( + "test-call-steer-background", + { command: "printf 'start\\n'; sleep 0.05; printf 'done\\n'" }, + undefined, + undefined, + { + ...createTestToolContext([]), + toolCall: { + batchId: "batch-1", + index: 0, + total: 1, + toolCalls: [{ id: "test-call-steer-background", name: "bash" }], + steeringSignal: steering.signal, + }, + }, + ); + + // The steer backgrounds the command instead of killing it: the call + // returns a running job and the command finishes on its own. + expect(result.details?.async?.state).toBe("running"); + expect(getTextOutput(result)).toContain("Backgrounded early to handle an incoming message"); + const jobId = result.details?.async?.jobId; + if (!jobId) { + throw new Error("expected a steer-backgrounded job id"); + } + const job = asyncJobManager.getJob(jobId); + expect(job?.status).toBe("running"); + await job?.promise; + expect(asyncJobManager.getJob(jobId)?.status).toBe("completed"); + await asyncJobManager.dispose(); + }); + it("should background instead of timing out when auto-background wait exceeds the effective timeout", async () => { const deliveries: Array<{ jobId: string; text: string }> = []; const asyncJobManager = new AsyncJobManager({ From 57580aa4136d7cf4aebb13533496ee01f5b6e01f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Korm=C3=A1kur?= Date: Mon, 20 Jul 2026 21:51:22 +0000 Subject: [PATCH 2/2] fix(coding-agent): require a fresh yield after async-result deliveries in the quiescence barrier MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The barrier in driveSessionToYield was unreachable for terminal yields: the yield tool's shouldTerminate fired requestAbort("terminate"), so abortSignal was always aborted before the barrier's loop condition ran, and a run with pending owner jobs completed immediately with whatever the pre-async yield said (Codex review on #6119). - Split "stop the free-running turn after yield" from "terminate the run": a terminal yield with pending owner async work now parks the run with a recoverable session abort (budget-stop precedent) via requestYieldTurnStop; only a quiescent yield terminates. - An async-result follow-up injected after a recorded yield un-latches it (transcript-ordered, in the run monitor) and re-runs the reminder ladder, so the run only completes on a yield that postdates every delivered result — including results injected during the notice turn. - A run that never refreshes a superseded yield fails (exit 1) with an explicit reason; the stale payload ships only as failed-run salvage through the existing failed-after-yield finalize path. - Rewrote subagent-async-pending.md: the "your current yield stands" option contradicted the enforced contract. - Regression tests: parked yield -> injected result -> fresh yield wins; refusal -> stale payload fails; no-async fast path unchanged. --- packages/coding-agent/CHANGELOG.md | 2 +- .../prompts/system/subagent-async-pending.md | 4 +- .../src/session/async-job-delivery.ts | 9 +- packages/coding-agent/src/task/executor.ts | 257 +++++++++++++----- .../test/async-job-manager.test.ts | 3 +- .../task/executor-async-quiescence.test.ts | 253 +++++++++++++++++ 6 files changed, 456 insertions(+), 72 deletions(-) create mode 100644 packages/coding-agent/test/task/executor-async-quiescence.test.ts diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 11daeea31..e90581e2e 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -10,7 +10,7 @@ ### Changed -- Subagents now inherit `async.enabled` and `bash.autoBackground.enabled` from the parent instead of having both force-disabled. Subagent runs complete only after their own background jobs settle (results are folded into the run as async-result follow-ups, with a one-time notice offering `hub` wait/cancel), and teardown cancels and awaits surviving jobs before isolation worktree capture and cleanup. +- Subagents now inherit `async.enabled` and `bash.autoBackground.enabled` from the parent instead of having both force-disabled. Subagent runs complete only after their own background jobs settle and the agent submits a `yield` that postdates every delivered result: a terminal yield with jobs still pending parks the run (recoverable turn stop) instead of completing it, async results are folded in as follow-up turns (with a one-time notice offering `hub` wait/cancel), a result delivered after a yield supersedes that yield and re-runs the yield reminder ladder, and a run that never refreshes a superseded yield fails with the stale payload preserved as salvage. Teardown cancels and awaits surviving jobs before isolation worktree capture and cleanup. ### Fixed diff --git a/packages/coding-agent/src/prompts/system/subagent-async-pending.md b/packages/coding-agent/src/prompts/system/subagent-async-pending.md index 89bf568cb..daee84c25 100644 --- a/packages/coding-agent/src/prompts/system/subagent-async-pending.md +++ b/packages/coding-agent/src/prompts/system/subagent-async-pending.md @@ -1,6 +1,6 @@ Your yield was recorded, but {{count}} background job{{#if multiple}}s{{/if}} you own {{#if multiple}}are{{else}}is{{/if}} still running: {{jobs}}. -This run completes only after these jobs settle; their results arrive as follow-up messages. Decide now: +This run completes only after these jobs settle AND you submit a fresh `yield` that accounts for their results. Job results arrive as follow-up messages; a result that arrives after your yield supersedes it — your current yield will NOT be accepted as the final report. Decide now: - Need the results? Wait for them (`hub` op:"wait"), then submit a fresh `yield` that incorporates them. - Job no longer needed? Cancel it (`hub` op:"cancel", ids:[...]) and re-yield. -- Otherwise say nothing further; your current yield stands once the jobs finish. +- Otherwise stand by; when each result arrives, submit a fresh `yield` (repeat your report unchanged if the result does not affect it). diff --git a/packages/coding-agent/src/session/async-job-delivery.ts b/packages/coding-agent/src/session/async-job-delivery.ts index 5565bbf28..061d99dc6 100644 --- a/packages/coding-agent/src/session/async-job-delivery.ts +++ b/packages/coding-agent/src/session/async-job-delivery.ts @@ -13,6 +13,13 @@ import type { AsyncJob } from "../async"; import asyncResultTemplate from "../prompts/tools/async-result.md" with { type: "text" }; import type { CustomMessage } from "./messages"; +/** + * `customType` of the injected async-result follow-up message. The task + * executor's run monitor matches on it to invalidate a previously recorded + * yield: a result injected after the yield supersedes that yield's payload. + */ +export const ASYNC_RESULT_MESSAGE_TYPE = "async-result"; + /** Result payloads longer than this spill to an artifact with an inline preview. */ export const ASYNC_INLINE_RESULT_MAX_CHARS = 12_000; export const ASYNC_PREVIEW_MAX_CHARS = 4_000; @@ -54,7 +61,7 @@ export function buildAsyncResultBatchMessage(entries: AsyncResultEntry[]): Custo }; return { role: "custom", - customType: "async-result", + customType: ASYNC_RESULT_MESSAGE_TYPE, content: prompt.render(asyncResultTemplate, { multiple: jobs.length > 1, jobs, diff --git a/packages/coding-agent/src/task/executor.ts b/packages/coding-agent/src/task/executor.ts index 6a54bb9bc..864558a19 100644 --- a/packages/coding-agent/src/task/executor.ts +++ b/packages/coding-agent/src/task/executor.ts @@ -5,7 +5,7 @@ */ import path from "node:path"; -import type { AgentEvent, AgentIdentity, AgentTelemetryConfig } from "@oh-my-pi/pi-agent-core"; +import type { AgentEvent, AgentIdentity, AgentMessage, AgentTelemetryConfig } from "@oh-my-pi/pi-agent-core"; import { recordHandoff, resolveTelemetry } from "@oh-my-pi/pi-agent-core"; import type { Api, Model, ServiceTierByFamily, Usage } from "@oh-my-pi/pi-ai"; import { logger, popLoopPhase, prompt, pushLoopPhase, untilAborted } from "@oh-my-pi/pi-utils"; @@ -42,6 +42,7 @@ import { AgentRegistry } from "../registry/agent-registry"; import { type CreateAgentSessionOptions, createAgentSession, discoverAuthStorage } from "../sdk"; import type { AgentSession, AgentSessionEvent, Prewalk } from "../session/agent-session"; import type { ArtifactManager } from "../session/artifacts"; +import { ASYNC_RESULT_MESSAGE_TYPE } from "../session/async-job-delivery"; import type { AuthStorage } from "../session/auth-storage"; import { SKILL_PROMPT_MESSAGE_TYPE, USER_INTERRUPT_LABEL } from "../session/messages"; import { SessionManager } from "../session/session-manager"; @@ -877,6 +878,20 @@ interface SubagentRunMonitor { budgetStopRequested(): boolean; /** Resolves when the budget-stop session abort has settled (immediately when no stop fired). */ waitForBudgetStop(): Promise; + /** + * True when a recorded yield was invalidated by a later async-result + * injection and no fresh yield has landed since: the yield payload + * predates background job outcomes the model was shown. + */ + yieldInvalidatedByAsync(): boolean; + /** + * True once a terminal yield with pending owner async work stopped the + * free-running turn (recoverable, like a budget stop) instead of + * terminating the run. Cleared when {@link waitForYieldTurnStop} settles. + */ + yieldTurnStopRequested(): boolean; + /** Resolves when the yield turn-stop session abort has settled (immediately when none fired). */ + waitForYieldTurnStop(): Promise; /** The abort kind for this run, when an abort was requested. */ abortKind(): AbortReason | undefined; /** True when the abort carries a precise external reason (signal / wall-clock / budget). */ @@ -903,6 +918,15 @@ interface SubagentRunMonitor { finish(): void; } +/** + * True when `message` is the session-injected async-result follow-up + * ({@link ASYNC_RESULT_MESSAGE_TYPE}): the transcript-ordered signal that a + * background job outcome landed after whatever the model said before it. + */ +function isAsyncResultInjection(message: AgentMessage | undefined): boolean { + return message?.role === "custom" && message.customType === ASYNC_RESULT_MESSAGE_TYPE; +} + function createSubagentRunMonitor(args: RunMonitorArgs): SubagentRunMonitor { const { index, @@ -954,6 +978,9 @@ function createSubagentRunMonitor(args: RunMonitorArgs): SubagentRunMonitor { let activeSession: AgentSession | null = null; let yieldCalled = false; let yieldCallPending = false; + let yieldInvalidatedByAsync = false; + let yieldTurnStopRequested = false; + let yieldTurnStopPromise: Promise | null = null; // Accumulate usage incrementally from message_end events (no memory for streaming events) const accumulatedUsage: Usage = { @@ -1026,6 +1053,30 @@ function createSubagentRunMonitor(args: RunMonitorArgs): SubagentRunMonitor { : Promise.resolve(); }; + // Yield turn-stop: a terminal yield recorded while owner async work is + // still pending is a scheduling pause, not run completion. Stop the + // free-running turn exactly like a budget stop (session abort, monitor + // signal untouched) so driveSessionToYield's quiescence barrier can settle + // the jobs, fold their results in, and demand a fresh yield. Terminating + // here instead would abort the run signal and make the barrier + // unreachable, completing the run with a payload that predates the job + // outcomes. + const requestYieldTurnStop = () => { + if (yieldTurnStopRequested || abortSent || resolved) return; + yieldTurnStopRequested = true; + const session = activeSession; + yieldTurnStopPromise = session + ? session.abort().catch(error => { + logger.debug("Subagent yield turn-stop abort failed", { + error: error instanceof Error ? error.message : String(error), + }); + }) + : Promise.resolve(); + }; + + /** Owner async work that can still re-wake the run (quiescence barrier predicate). */ + const sessionHasPendingAsyncWork = (): boolean => activeSession?.hasPendingAsyncWork?.() ?? false; + // Handle abort signal if (signal) { signal.addEventListener( @@ -1229,6 +1280,7 @@ function createSubagentRunMonitor(args: RunMonitorArgs): SubagentRunMonitor { if (toolName === "yield") { yieldCalled = true; yieldCallPending = false; + yieldInvalidatedByAsync = false; } }; @@ -1242,6 +1294,16 @@ function createSubagentRunMonitor(args: RunMonitorArgs): SubagentRunMonitor { if (event.message?.role === "assistant") { resetRecentOutput(); } + // An async-result follow-up injected after a recorded yield + // supersedes that yield: its payload predates the job outcome the + // model is now being shown. Un-latch so the quiescence barrier's + // reminder ladder demands a fresh yield. Guarded on the run signal: + // once the run is completing, late injections must not destabilize + // the settled classification. + if (yieldCalled && !abortSignal.aborted && isAsyncResultInjection(event.message)) { + yieldCalled = false; + yieldInvalidatedByAsync = true; + } break; case "tool_execution_start": { @@ -1325,7 +1387,14 @@ function createSubagentRunMonitor(args: RunMonitorArgs): SubagentRunMonitor { isError: event.isError, }) ) { - requestAbort("terminate"); + if (event.toolName === "yield" && sessionHasPendingAsyncWork()) { + // Terminal yield with owner jobs still pending: park the + // run behind the quiescence barrier instead of completing + // it (see requestYieldTurnStop). + requestYieldTurnStop(); + } else { + requestAbort("terminate"); + } } } flushProgress = true; @@ -1567,6 +1636,25 @@ function createSubagentRunMonitor(args: RunMonitorArgs): SubagentRunMonitor { abortReason === "signal" || runtimeLimitExceeded || budgetLimitExceeded || budgetStopRequested, budgetStopRequested: () => budgetStopRequested, waitForBudgetStop: () => budgetStopAbortPromise ?? Promise.resolve(), + yieldInvalidatedByAsync: () => yieldInvalidatedByAsync, + yieldTurnStopRequested: () => yieldTurnStopRequested, + waitForYieldTurnStop: async () => { + const pending = yieldTurnStopPromise; + if (!pending) { + yieldTurnStopRequested = false; + return; + } + try { + await pending; + } finally { + // Clear only after the abort settled so the idempotence gate in + // requestYieldTurnStop stays closed while it is in flight. + if (yieldTurnStopPromise === pending) { + yieldTurnStopPromise = null; + yieldTurnStopRequested = false; + } + } + }, // A soft stop that never escalated still identifies as a budget abort so // the lifecycle can park the agent as resumable instead of killing it. abortKind: () => abortReason ?? (budgetStopRequested ? "budget" : undefined), @@ -1664,83 +1752,106 @@ async function driveSessionToYield( await awaitAbortable(session.prompt(task, { attribution: "agent" })); await awaitAbortable(session.waitForIdle()); } catch (err) { - // A budget stop cancels the free-running turn by aborting the - // session, which can surface here as a rejected prompt. Swallow it - // and drive the forced final yield below; real caller/timeout - // aborts (monitor signal) and genuine failures keep the old path. - if (!monitor.budgetStopRequested() || abortSignal.aborted) throw err; + // A budget stop or a yield turn-stop (terminal yield parked behind + // the async quiescence barrier) cancels the free-running turn by + // aborting the session, which can surface here as a rejected + // prompt. Swallow it and drive the barrier/forced final yield + // below; real caller/timeout aborts (monitor signal) and genuine + // failures keep the old path. + const recoverableStop = monitor.budgetStopRequested() || monitor.yieldTurnStopRequested(); + if (!recoverableStop || abortSignal.aborted) throw err; } const reminderToolChoice = buildNamedToolChoice("yield", session.model); - let retryCount = 0; - while (!monitor.yieldCalled() && retryCount < MAX_YIELD_RETRIES && !abortSignal.aborted) { - // A budget stop collapses the reminder ladder to a single forced - // final yield: wait for the stop's session abort to settle, then - // prompt once with the wrap-up reminder + named tool choice. - const budgetStop = monitor.budgetStopRequested(); - if (budgetStop) { - retryCount = MAX_YIELD_RETRIES - 1; - await monitor.waitForBudgetStop(); - if (monitor.yieldCalled() || abortSignal.aborted) break; - } - // Skip reminders when the model returned a terminal error (e.g. - // rate-limit cap hit, auth failure). Re-prompting would just - // hit the same wall, multiplying the failure noise without - // any chance of producing a yield. - const lastBeforeReminder = session.getLastAssistantMessage(); - if (lastBeforeReminder?.stopReason === "error") break; - try { - retryCount++; - const reminder = prompt.render(submitReminderTemplate, { - retryCount, - maxRetries: MAX_YIELD_RETRIES, - budgetStop, - }); - - const isFinalRetry = retryCount >= MAX_YIELD_RETRIES; - await awaitAbortable( - session.prompt(reminder, { - attribution: "agent", - synthetic: true, - ...(isFinalRetry && reminderToolChoice ? { toolChoice: reminderToolChoice } : {}), - }), - ); - await awaitAbortable(session.waitForIdle()); - } catch (err) { - if (abortSignal.aborted || err instanceof ToolAbortError) { - // Benign control-flow exit — user cancel (^C) or compaction aborting - // pending operations both surface here as ToolAbortError. The outer - // catch and finally already mark the run aborted; logging at ERROR - // would spam operator dashboards with non-failures. - logger.debug("Subagent prompt aborted"); - } else { - logger.error("Subagent prompt failed", { - error: err instanceof Error ? err.message : String(err), + const runYieldLadder = async (): Promise => { + let retryCount = 0; + while (!monitor.yieldCalled() && retryCount < MAX_YIELD_RETRIES && !abortSignal.aborted) { + // A budget stop collapses the reminder ladder to a single forced + // final yield: wait for the stop's session abort to settle, then + // prompt once with the wrap-up reminder + named tool choice. + const budgetStop = monitor.budgetStopRequested(); + if (budgetStop) { + retryCount = MAX_YIELD_RETRIES - 1; + await monitor.waitForBudgetStop(); + if (monitor.yieldCalled() || abortSignal.aborted) break; + } + // Skip reminders when the model returned a terminal error (e.g. + // rate-limit cap hit, auth failure). Re-prompting would just + // hit the same wall, multiplying the failure noise without + // any chance of producing a yield. + const lastBeforeReminder = session.getLastAssistantMessage(); + if (lastBeforeReminder?.stopReason === "error") break; + try { + retryCount++; + const reminder = prompt.render(submitReminderTemplate, { + retryCount, + maxRetries: MAX_YIELD_RETRIES, + budgetStop, }); + + const isFinalRetry = retryCount >= MAX_YIELD_RETRIES; + await awaitAbortable( + session.prompt(reminder, { + attribution: "agent", + synthetic: true, + ...(isFinalRetry && reminderToolChoice ? { toolChoice: reminderToolChoice } : {}), + }), + ); + await awaitAbortable(session.waitForIdle()); + } catch (err) { + if (abortSignal.aborted || err instanceof ToolAbortError) { + // Benign control-flow exit — user cancel (^C) or compaction aborting + // pending operations both surface here as ToolAbortError. The outer + // catch and finally already mark the run aborted; logging at ERROR + // would spam operator dashboards with non-failures. + logger.debug("Subagent prompt aborted"); + } else { + logger.error("Subagent prompt failed", { + error: err instanceof Error ? err.message : String(err), + }); + } } } - } + }; - // Quiescence barrier (structured concurrency): a final yield with owner - // background jobs still running or undelivered is a scheduling pause, - // not run completion. Wait for the jobs to settle, let their - // async-result follow-up turns run to idle, and only then classify the - // terminal state — the isolation runner captures and destroys the - // worktree right after this run resolves, so no owner job that could - // still re-wake the session may outlive it. Suppressed (acknowledged / - // hub-watched) jobs never re-wake the run and are reaped at teardown. + // Yield ladder + quiescence barrier (structured concurrency), one + // loop: each iteration first demands a yield — initially, and again + // whenever an async-result delivery un-latched the previous one + // (including during the notice turn) — then either completes on + // quiescence or settles one generation of owner async work. + // + // A final yield with owner background jobs still running or + // undelivered is a scheduling pause, not run completion — the monitor + // parks such a yield with a recoverable turn-stop instead of + // terminating the run. Jobs are settled and their results folded into + // the run as async-result follow-up turns; each delivered result + // supersedes the yield it postdates, so the reminder ladder re-runs + // to demand a fresh yield that accounts for it. Only a yield with no + // pending owner work left is terminal — the isolation runner captures + // and destroys the worktree right after this run resolves, so no + // owner job that could still re-wake the session may outlive it. + // Suppressed (acknowledged / hub-watched) jobs never re-wake the run + // and are reaped at teardown. // // Before blocking on running jobs, tell the model ONCE what it is // waiting on so it can `hub` wait/cancel instead of sitting silent - // until the jobs (or the runtime limit) expire. Guarded to a single - // notice per run — a model that yields again with jobs still running - // is then waited out passively. Runs that never yielded (ladder - // exhausted / terminal model error) skip the barrier entirely — more + // until the jobs (or the runtime limit) expire. Runs that never yield + // (ladder exhausted / terminal model error) skip the barrier — more // injected turns just multiply the failure noise; the teardown reap // still cancels and awaits their jobs before worktree capture. let asyncPendingNoticeSent = false; - while (monitor.yieldCalled() && !abortSignal.aborted && session.hasPendingAsyncWork()) { + while (!abortSignal.aborted) { + if (!monitor.yieldCalled()) { + await runYieldLadder(); + // Ladder exhausted / terminal model error: classified below + // (missing yield, or stale yield when one was invalidated). + if (!monitor.yieldCalled()) break; + } + // Let the parked yield's turn-stop session abort settle before + // prompting again (mirrors waitForBudgetStop). + await awaitAbortable(monitor.waitForYieldTurnStop()); + if (!session.hasPendingAsyncWork()) break; if (!asyncPendingNoticeSent) { asyncPendingNoticeSent = true; const running = session.getAsyncJobSnapshot()?.running ?? []; @@ -1763,11 +1874,13 @@ async function driveSessionToYield( }); } // Re-evaluate: the notice turn may have cancelled, watched, or - // absorbed the jobs. + // absorbed the jobs — or already re-yielded. continue; } } await awaitAbortable(session.settleAsyncWork()); + // Results delivered during the settle invalidated the recorded + // yield: the next iteration's ladder demands a fresh one. } if (monitor.yieldCalled()) { @@ -1806,6 +1919,18 @@ async function driveSessionToYield( abortReasonText ??= monitor.resolveAbortReasonText(); exitCode = 1; } + + // A recorded yield that async-result deliveries superseded and the + // model never refreshed is stale: fail the run instead of letting the + // parent act on a payload that predates the background job outcomes + // the model was shown. The stale payload still ships through + // finalizeSubprocessOutput's failed-after-yield path (exit 1 + stderr, + // output preserved as salvage). + if (monitor.yieldInvalidatedByAsync() && !abortSignal.aborted) { + exitCode = 1; + error ??= + "Background job results arrived after the subagent's last yield; it did not submit a refreshed yield covering them."; + } } catch (err) { if (abortSignal.aborted && monitor.yieldCalled() && !monitor.runtimeLimitExceeded()) { exitCode = 0; diff --git a/packages/coding-agent/test/async-job-manager.test.ts b/packages/coding-agent/test/async-job-manager.test.ts index 303141ab5..1a336456a 100644 --- a/packages/coding-agent/test/async-job-manager.test.ts +++ b/packages/coding-agent/test/async-job-manager.test.ts @@ -333,7 +333,6 @@ describe("AsyncJobManager", () => { }); test("scoped delivery drain times out while a matching delivery callback is in flight", async () => { - let mainJobId = ""; let targetJobId = ""; let releaseMainDelivery = (): void => {}; let notifyMainDeliveryStarted = (): void => {}; @@ -363,7 +362,7 @@ describe("AsyncJobManager", () => { completions.push(jobId); }); - mainJobId = manager.register("task", "main job", async () => "main result", { ownerId: "0-Main" }); + manager.register("task", "main job", async () => "main result", { ownerId: "0-Main" }); targetJobId = manager.register("task", "subagent job", async () => "subagent result", { ownerId: "3-AuthLoader", }); diff --git a/packages/coding-agent/test/task/executor-async-quiescence.test.ts b/packages/coding-agent/test/task/executor-async-quiescence.test.ts new file mode 100644 index 000000000..25c3b3d90 --- /dev/null +++ b/packages/coding-agent/test/task/executor-async-quiescence.test.ts @@ -0,0 +1,253 @@ +/** + * Quiescence barrier fresh-yield contract (PR #6119 review): a terminal + * `yield` recorded while owner background jobs are still pending parks the + * run instead of terminating it, and an async-result delivered after that + * yield supersedes it — the run only completes on a yield that postdates + * every delivered result. A model that never refreshes its yield must fail + * the run rather than surface the stale payload as a clean success. + */ +import { afterEach, describe, expect, it, vi } from "bun:test"; +import type { AssistantMessage } from "@oh-my-pi/pi-ai"; +import type { LoadExtensionsResult } from "@oh-my-pi/pi-coding-agent/extensibility/extensions/types"; +import type { CreateAgentSessionResult } from "@oh-my-pi/pi-coding-agent/sdk"; +import * as sdkModule from "@oh-my-pi/pi-coding-agent/sdk"; +import type { AgentSession, AgentSessionEvent } from "@oh-my-pi/pi-coding-agent/session/agent-session"; +import { runSubprocess } from "@oh-my-pi/pi-coding-agent/task/executor"; +import type { AgentDefinition } from "@oh-my-pi/pi-coding-agent/task/types"; +import { EventBus } from "@oh-my-pi/pi-coding-agent/utils/event-bus"; + +const baseAgent: AgentDefinition = { name: "task", description: "test", systemPrompt: "test", source: "bundled" }; + +function assistantStopMessage(text: string): AssistantMessage { + return { + role: "assistant", + content: [{ type: "text", text }], + api: "openai-responses", + provider: "openai", + model: "mock", + usage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }, + stopReason: "stop", + timestamp: Date.now(), + }; +} + +interface AsyncQuiescenceHarness { + session: AgentSession; + prompts: string[]; + abortCalls: () => number; + settleCalls: () => number; + emitTerminalYield: (data: unknown) => void; + finishJob: () => void; +} + +/** + * Mock session with the owner-async surface the barrier drives: + * `hasPendingAsyncWork` / `getAsyncJobSnapshot` / `settleAsyncWork`. The job + * "finishes" during the first settle, which injects the async-result + * follow-up (custom message_start) and a plain assistant reaction WITHOUT a + * fresh yield — exactly the review's stale-yield scenario. + */ +function createAsyncSession( + onPrompt: (params: { text: string; promptIndex: number; harness: AsyncQuiescenceHarness }) => void, +): AsyncQuiescenceHarness { + const listeners: Array<(event: AgentSessionEvent) => void> = []; + const state = { messages: [] as AssistantMessage[] }; + const prompts: string[] = []; + let abortCount = 0; + let settleCount = 0; + let pendingAsync = true; + let runningJobs: Array<{ id: string; label?: string }> = [{ id: "job-1", label: "background build" }]; + let toolCallSeq = 0; + + const emit = (event: AgentSessionEvent) => { + for (const listener of [...listeners]) listener(event); + }; + + const emitTerminalYield = (data: unknown) => { + toolCallSeq += 1; + emit({ + type: "tool_execution_end", + toolCallId: `yield-${toolCallSeq}`, + toolName: "yield", + result: { + content: [{ type: "text", text: "Result submitted." }], + details: { status: "success", data }, + }, + } as AgentSessionEvent); + }; + + const finishJob = () => { + pendingAsync = false; + runningJobs = []; + // Owner job completed: the session injects the async-result follow-up + // turn. The model reacts with text only — no fresh yield. + emit({ + type: "message_start", + message: { + role: "custom", + customType: "async-result", + content: "Background job job-1 has completed.\nexit 1: build FAILED", + display: true, + attribution: "agent", + timestamp: Date.now(), + }, + } as AgentSessionEvent); + const reaction = assistantStopMessage("The background build failed after I yielded."); + state.messages.push(reaction); + emit({ type: "message_end", message: reaction } as AgentSessionEvent); + }; + + const harness: AsyncQuiescenceHarness = { + session: undefined as unknown as AgentSession, + prompts, + abortCalls: () => abortCount, + settleCalls: () => settleCount, + emitTerminalYield, + finishJob, + }; + + const session = { + state, + agent: { state: { systemPrompt: ["test"] } }, + model: undefined, + extensionRunner: undefined, + sessionManager: { appendSessionInit: () => {} }, + getActiveToolNames: () => ["read", "yield"], + getEnabledToolNames: () => ["read", "yield"], + setActiveToolsByName: async (_toolNames: string[]) => {}, + subscribe: (listener: (event: AgentSessionEvent) => void) => { + listeners.push(listener); + return () => { + const index = listeners.indexOf(listener); + if (index >= 0) listeners.splice(index, 1); + }; + }, + prompt: async (text: string) => { + prompts.push(text); + onPrompt({ text, promptIndex: prompts.length, harness }); + }, + waitForIdle: async () => {}, + getLastAssistantMessage: () => state.messages[state.messages.length - 1], + hasPendingAsyncWork: () => pendingAsync, + getAsyncJobSnapshot: () => ({ running: runningJobs, recent: [] }), + settleAsyncWork: async () => { + settleCount += 1; + harness.finishJob(); + }, + abort: async () => { + abortCount += 1; + }, + dispose: async () => {}, + }; + harness.session = session as unknown as AgentSession; + return harness; +} + +function mockCreateAgentSession(session: AgentSession) { + return vi.spyOn(sdkModule, "createAgentSession").mockResolvedValue({ + session, + extensionsResult: {} as unknown as LoadExtensionsResult, + setToolUIContext: () => {}, + eventBus: new EventBus(), + } as CreateAgentSessionResult); +} + +describe("runSubprocess async quiescence fresh-yield contract", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it("parks a pending yield, injects the result, and completes on the fresh yield", async () => { + const harness = createAsyncSession(({ promptIndex, harness: h }) => { + if (promptIndex === 1) { + // Terminal yield while the background job is still running. + h.emitTerminalYield({ report: "STALE: build passing (job still running)" }); + return; + } + if (promptIndex === 2) { + // Async-pending notice: the model stands by. The job then + // finishes during the barrier's settle. + return; + } + // Reminder ladder after the async-result invalidated the yield: + // submit the fresh yield that accounts for the job outcome. + h.emitTerminalYield({ report: "FRESH: build failed, see job-1" }); + }); + mockCreateAgentSession(harness.session); + + const result = await runSubprocess({ + cwd: "/tmp", + agent: baseAgent, + task: "do the work", + index: 0, + id: "quiescence-fresh-yield", + }); + + // Run did not terminate on the parked yield: the barrier noticed, the + // job settled, and the ladder demanded exactly one more prompt. + expect(harness.prompts).toHaveLength(3); + expect(harness.prompts[1]).toContain("yield was recorded"); + expect(harness.settleCalls()).toBe(1); + // The parked yield stopped the turn without killing the run. + expect(harness.abortCalls()).toBeGreaterThanOrEqual(1); + // The fresh yield — not the stale one — is the result of record. + expect(result.exitCode).toBe(0); + expect(result.output).toContain("FRESH: build failed"); + expect(result.output).not.toContain("STALE"); + }); + + it("fails the run when the model never refreshes the superseded yield", async () => { + const harness = createAsyncSession(({ promptIndex, harness: h }) => { + if (promptIndex === 1) { + h.emitTerminalYield({ report: "STALE: build passing (job still running)" }); + } + // Notice and every reminder: the model never yields again. + }); + mockCreateAgentSession(harness.session); + + const result = await runSubprocess({ + cwd: "/tmp", + agent: baseAgent, + task: "do the work", + index: 0, + id: "quiescence-stale-refusal", + }); + + // task + notice + full reminder ladder (3). + expect(harness.prompts).toHaveLength(5); + // Stale payload must not read as success; it ships only as failed-run + // salvage with an explicit reason. + expect(result.exitCode).toBe(1); + expect(result.error).toContain("refreshed yield"); + expect(result.output).toContain("STALE: build passing"); + }); + + it("terminates immediately on yield when no owner async work is pending", async () => { + const harness = createAsyncSession(({ promptIndex, harness: h }) => { + if (promptIndex === 1) { + h.finishJob(); + h.emitTerminalYield({ report: "done" }); + } + }); + mockCreateAgentSession(harness.session); + + const result = await runSubprocess({ + cwd: "/tmp", + agent: baseAgent, + task: "do the work", + index: 0, + id: "quiescence-no-async", + }); + + expect(harness.prompts).toHaveLength(1); + expect(result.exitCode).toBe(0); + expect(result.output).toContain("done"); + }); +});