diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index e1d898925..4f0745631 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,11 @@ ## [Unreleased] +### Fixed + +- Fixed async `bash({ async: true })` jobs finishing silently: completed or failed session-managed background bash jobs now enqueue an `async-job-result` follow-up message into the originating chat session automatically, while explicit completed-job polling, explicit cancellation, and parent aborts acknowledge the job and suppress duplicate or unwanted idle turns. Suppression is checked again at the streaming boundary so a completed job polled while its automatic follow-up is staged does not later deliver a duplicate. Async delivery bookkeeping is now bounded by the existing background-job retention defaults, suppressions stay tied to retained jobs so disposed-session running jobs cannot later fall back into the owner session, 12KB–50KB follow-up outputs persist their full output path before inline preview truncation, just-under-threshold raw outputs remain fully inline instead of being preview-truncated without a `fullOutputPath`, shared-manager lifecycle tracking prevents owner-session disposal from dropping later-session jobs while cleaning stale handlers from disposed fork/subagent sessions, and non-blocking delivery attempts keep one live streaming session from delaying unrelated async job completions. +- Fixed a background bash job leak where a failed async-manager registration (disposed manager/session or a capacity race mid-flight) left a never-executed job permanently reported as `running` to `__atomic_bash_job` polls; the managed job entry is now discarded when registration fails and the tool call error is surfaced unchanged. + ## [0.9.4-alpha.10] - 2026-07-03 ### Fixed diff --git a/packages/coding-agent/docs/tools.md b/packages/coding-agent/docs/tools.md index ea87af7d7..18119170f 100644 --- a/packages/coding-agent/docs/tools.md +++ b/packages/coding-agent/docs/tools.md @@ -28,9 +28,11 @@ Before writing, Atomic verifies the current file against the tagged snapshot. If ## `bash` and `bashInterceptor` -The `bash` tool executes shell commands in the session workspace, with optional PTY or background-job handling. When `pty: true` is requested, local execution uses the bundled Rust-backed PTY session so commands see a real terminal, including headless/tool-only and async job calls; if the native PTY package is unavailable, Atomic degrades to normal pipe execution. Set `PI_NO_PTY=1` or `ATOMIC_NO_PTY=1` to force normal pipe execution. Completed foreground results include oh-my-pi-style `timeoutSeconds`, `requestedTimeoutSeconds`, `wallTimeMs`, and non-zero `exitCode` metadata; background jobs use `details.async: { state, jobId, type: "bash" }`, can be polled with `bash({"command":"__atomic_bash_job "})`, can be cancelled with `bash({"command":"__atomic_bash_job_cancel "})`, and preserve overflow output in a temporary `fullOutputPath` when polling output is truncated. `bashInterceptor.enabled` defaults to `false`; interception is not auto-enabled. +The `bash` tool executes shell commands in the session workspace, with optional PTY or background-job handling. When `pty: true` is requested, local execution uses the bundled Rust-backed PTY session so commands see a real terminal, including headless/tool-only and async job calls; if the native PTY package is unavailable, Atomic degrades to normal pipe execution. Set `PI_NO_PTY=1` or `ATOMIC_NO_PTY=1` to force normal pipe execution. Completed foreground results include oh-my-pi-style `timeoutSeconds`, `requestedTimeoutSeconds`, `wallTimeMs`, and non-zero `exitCode` metadata; background jobs use `details.async: { state, jobId, type: "bash" }`, can be polled with `bash({"command":"__atomic_bash_job "})`, can be cancelled with `bash({"command":"__atomic_bash_job_cancel "})`, and preserve overflow output in a temporary `fullOutputPath` when polling output is truncated. -When explicitly enabled in settings, built-in bash interceptor rules block common shell substitutes for first-class tools (`cat`/`grep`/`find`/in-place `sed`/redirection, etc.) only when the corresponding tool is available. Enabled bash tool calls are also offered to `user_bash` extension handlers before local execution. Atomic checks the original command, the internal-URL-expanded command, configured-prefix forms, `spawnHook`-rewritten commands, and a leading `cd path && command` or `cd path; command`-stripped form only when structured `cwd` was omitted, so interceptors can route commands by effective working directory without overriding explicit `cwd`. The bash schema accepts `cwd`, `env`, `timeout`, `pty`, and `async`; `cwd` and `env` are honored by the local executor, `timeout` defaults to 300s and is clamped to 1..3600s, and normal sessions enable tracked async jobs with bounded retention. +When a session-managed background bash job completes or fails, Atomic sends an `async-job-result` custom follow-up into the conversation automatically (`display: true`, delivered as a follow-up turn). Small results are inlined, and results whose raw output stays below the persistence threshold remain fully inline even if the formatted follow-up header pushes the message over the preview limit; persisted large results include a preview plus the retained `fullOutputPath` (persisted before the normal polling truncation limit so 12KB–50KB outputs remain recoverable). If the model explicitly polls a completed job with `__atomic_bash_job ` before the queued follow-up is delivered, or cancels a job with `__atomic_bash_job_cancel `, Atomic acknowledges the result and suppresses duplicate auto-delivery while keeping the job pollable until normal bounded retention/TTL cleanup. Suppression is tied to the retained job rather than a short timer, so disposed-session jobs cannot later fall back into another session after a long-running command completes. Session delivery attempts are non-blocking across sessions: a live streaming session can defer its own follow-up until the stream boundary without delaying unrelated completed jobs. Session disposal removes that session's pending async delivery handlers; a shared manager remains alive while other live sessions still own active jobs, then cleans up when the last session is disposed. Direct SDK/tool-factory uses only get automatic delivery when they provide an async job manager/delivery handler; otherwise async jobs remain manually pollable. + +When explicitly enabled in settings, built-in bash interceptor rules block common shell substitutes for first-class tools (`cat`/`grep`/`find`/in-place `sed`/redirection, etc.) only when the corresponding tool is available. Enabled bash tool calls are also offered to `user_bash` extension handlers before local execution. Atomic checks the original command, the internal-URL-expanded command, configured-prefix forms, `spawnHook`-rewritten commands, and a leading `cd path && command` or `cd path; command`-stripped form only when structured `cwd` was omitted, so interceptors can route commands by effective working directory without overriding explicit `cwd`. The bash schema accepts `cwd`, `env`, `timeout`, `pty`, and `async`; `cwd` and `env` are honored by the local executor, `timeout` defaults to 300s and is clamped to 1..3600s, and normal sessions enable tracked async jobs with bounded retention. `bashInterceptor.enabled` defaults to `false`; interception is not auto-enabled. ```json { diff --git a/packages/coding-agent/src/core/agent-session-events.ts b/packages/coding-agent/src/core/agent-session-events.ts index 59bd594bd..599f81478 100644 --- a/packages/coding-agent/src/core/agent-session-events.ts +++ b/packages/coding-agent/src/core/agent-session-events.ts @@ -1,3 +1,4 @@ +import { disposeSessionAsyncJobManager } from "./async/session-manager.js"; import type { AgentEvent, AgentMessage } from "@earendil-works/pi-agent-core"; import type { AssistantMessage, Message, TextContent } from "@earendil-works/pi-ai/compat"; import { cleanupSessionResources } from "@earendil-works/pi-ai/compat"; @@ -405,6 +406,7 @@ export function dispose(this: AgentSession): void { this._extensionRunner.invalidate( "This extension ctx is stale after session replacement or reload. Do not use a captured pi or command ctx after ctx.newSession(), ctx.fork(), ctx.switchSession(), or ctx.reload(). For newSession, fork, and switchSession, move post-replacement work into withSession and use the ctx passed to withSession. For reload, do not use the old ctx after await ctx.reload().", ); + disposeSessionAsyncJobManager(this._asyncJobManager, this._asyncJobManagerSessionId); this._disconnectFromAgent(); this._eventListeners = []; cleanupSessionResources(this.sessionId); diff --git a/packages/coding-agent/src/core/agent-session-methods.ts b/packages/coding-agent/src/core/agent-session-methods.ts index a6c8d80a8..981f7ed48 100644 --- a/packages/coding-agent/src/core/agent-session-methods.ts +++ b/packages/coding-agent/src/core/agent-session-methods.ts @@ -22,9 +22,10 @@ import type { BashExecutionMessage, CustomMessage } from "./messages.ts"; import type { ModelRegistry } from "./model-registry.ts"; import type { PromptTemplate } from "./prompt-templates.ts"; import type { ResourceLoader } from "./resource-loader.ts"; -import type { BranchSummaryEntry, SessionManager } from "./session-manager.ts"; +import type { BranchSummaryEntry, SessionManager } from "./session-manager.js"; import type { SettingsManager } from "./settings-manager.ts"; import type { BuildSystemPromptOptions } from "./system-prompt.ts"; +import type { AsyncJobManager } from "./async/job-manager.js"; import type { BashOperations } from "./tools/bash.ts"; import type { AgentSessionEvent, @@ -376,5 +377,7 @@ export interface AgentSessionInternalSurface extends AgentSessionMethodSurface, _baseSystemPromptOptions: BuildSystemPromptOptions; _systemPromptOverride?: string; _lastAssistantMessage: AssistantMessage | undefined; + _asyncJobManager: AsyncJobManager; + _asyncJobManagerSessionId: symbol; } diff --git a/packages/coding-agent/src/core/agent-session-tool-registry.ts b/packages/coding-agent/src/core/agent-session-tool-registry.ts index 09f0475ce..768a64bc6 100644 --- a/packages/coding-agent/src/core/agent-session-tool-registry.ts +++ b/packages/coding-agent/src/core/agent-session-tool-registry.ts @@ -6,6 +6,7 @@ import { createAllToolDefinitions, defaultToolNames } from "./tools/index.ts"; import { createToolDefinitionFromAgentTool } from "./tools/tool-definition-wrapper.ts"; import type { AgentSessionInternalSurface as AgentSession } from "./agent-session-methods.ts"; import type { ToolDefinitionEntry } from "./agent-session-types.ts"; +import { createSessionAsyncDeliveryHandler } from "./async/session-manager.js"; export function _refreshToolRegistry(this: AgentSession, options?: { activeToolNames?: string[]; includeAllExtensionTools?: boolean }): void { const previousRegistryNames = new Set(this._toolRegistry.keys()); @@ -154,6 +155,9 @@ export function _buildRuntime(this: AgentSession, options: { shellPath, interceptorEnabled: () => this.settingsManager.getBashInterceptorEnabled(), availableTools: activeBuiltinTools, + asyncJobManager: this._asyncJobManager, + asyncJobDeliveryHandler: createSessionAsyncDeliveryHandler(this, this._asyncJobManager, this._asyncJobManagerSessionId), + asyncJobSessionId: this._asyncJobManagerSessionId, interceptor: async (context) => { const result = await this._extensionRunner.emitUserBash({ type: "user_bash", diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index e803cbcd8..9139ef575 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -15,9 +15,11 @@ import type { Api, AssistantMessage, Model } from "@earendil-works/pi-ai/compat" import type { BashExecutionMessage, CustomMessage } from "./messages.ts"; import type { ModelRegistry } from "./model-registry.ts"; import type { ResourceLoader } from "./resource-loader.ts"; -import type { SessionManager } from "./session-manager.ts"; +import type { SessionManager } from "./session-manager.js"; import type { SettingsManager } from "./settings-manager.ts"; import type { BuildSystemPromptOptions } from "./system-prompt.ts"; +import type { AsyncJobManager } from "./async/job-manager.js"; +import { createSessionAsyncJobManager } from "./async/session-manager.js"; import { installAgentSessionAccessors } from "./agent-session-accessors.ts"; import { agentSessionAutoCompactionMethods } from "./agent-session-auto-compaction.ts"; import { agentSessionBashMethods } from "./agent-session-bash.ts"; @@ -117,7 +119,8 @@ export class AgentSession { protected _baseSystemPromptOptions!: BuildSystemPromptOptions; protected _systemPromptOverride?: string; protected _lastAssistantMessage: AssistantMessage | undefined = undefined; - + protected _asyncJobManager: AsyncJobManager; + protected _asyncJobManagerSessionId: symbol; constructor(config: AgentSessionConfig) { this.agent = config.agent; this.sessionManager = config.sessionManager; @@ -134,8 +137,10 @@ export class AgentSession { this._baseToolsOverride = config.baseToolsOverride; this._sessionStartEvent = config.sessionStartEvent ?? { type: "session_start", reason: "startup" }; this._orchestrationContext = config.orchestrationContext; - const internals = this as unknown as AgentSessionInternalSurface; + const asyncJobManagerHandle = createSessionAsyncJobManager(internals); + this._asyncJobManager = asyncJobManagerHandle.manager; + this._asyncJobManagerSessionId = asyncJobManagerHandle.sessionId; internals._handleAgentEvent = internals._handleAgentEvent.bind(this); this._unsubscribeAgent = this.agent.subscribe(internals._handleAgentEvent); internals._installAgentToolHooks(); diff --git a/packages/coding-agent/src/core/async/format.ts b/packages/coding-agent/src/core/async/format.ts new file mode 100644 index 000000000..547873d0f --- /dev/null +++ b/packages/coding-agent/src/core/async/format.ts @@ -0,0 +1,52 @@ +import type { ManagedAsyncBashJob, AsyncJobDeliveryMessage } from "./types.js"; + +const INLINE_OUTPUT_LIMIT = 12_000; +const LARGE_OUTPUT_PREVIEW_LIMIT = 4_000; + +function truncateText(text: string, maxChars: number): string { + if (text.length <= maxChars) return text; + return text.slice(0, maxChars).trimEnd(); +} + +function jobOutput(job: ManagedAsyncBashJob): string { + const output = job.output.trimEnd(); + if (output.length > 0) return output; + return job.error ? `(no output)\n\n${job.error}` : "(no output)"; +} + +function statusLine(job: ManagedAsyncBashJob): string | undefined { + if (job.error) return `Error: ${job.error}`; + if (job.exitCode !== undefined && job.exitCode !== null && job.exitCode !== 0) { + return `Command exited with code ${job.exitCode}`; + } + return undefined; +} + +export function formatAsyncResultForFollowUp(job: ManagedAsyncBashJob): AsyncJobDeliveryMessage { + const elapsedMs = (job.endedAt ?? Date.now()) - job.startedAt; + const header = `Async bash job ${job.jobId} ${job.status}: ${job.command}`; + const body = jobOutput(job); + const status = statusLine(job); + const inline = [header, body, status].filter((part): part is string => part !== undefined && part.length > 0).join("\n\n"); + let content = inline; + if (inline.length > INLINE_OUTPUT_LIMIT && job.fullOutputPath) { + const preview = truncateText(body, LARGE_OUTPUT_PREVIEW_LIMIT); + content = [header, preview, `[Output truncated for async follow-up. Full output: ${job.fullOutputPath}]`, status] + .filter((part): part is string => part !== undefined && part.length > 0) + .join("\n\n"); + } + return { + customType: "async-job-result", + content, + display: true, + details: { + jobId: job.jobId, + type: "bash", + status: job.status, + command: job.command, + exitCode: job.exitCode, + fullOutputPath: job.fullOutputPath, + wallTimeMs: elapsedMs, + }, + }; +} diff --git a/packages/coding-agent/src/core/async/job-manager.ts b/packages/coding-agent/src/core/async/job-manager.ts new file mode 100644 index 000000000..4443b8fd4 --- /dev/null +++ b/packages/coding-agent/src/core/async/job-manager.ts @@ -0,0 +1,263 @@ +import { COMPLETED_JOB_TTL_MS, MAX_MANAGED_BASH_JOBS, type ManagedBashJob } from "../tools/bash-async-jobs.js"; +import { formatAsyncResultForFollowUp } from "./format.js"; +import type { AsyncJobDeliveryCallback, AsyncJobDeliveryHandler, AsyncJobDeliveryMessage, ManagedAsyncBashJob } from "./types.js"; + +const DEFAULT_MAX_RUNNING_JOBS = 15; +const DELIVERY_RETRY_BASE_MS = 500; +const DELIVERY_RETRY_MAX_MS = 30_000; +const DELIVERY_RETRY_JITTER_MS = 200; + +interface AsyncJobManagerOptions { + onJobComplete: AsyncJobDeliveryCallback; + maxRunningJobs?: number; + maxRetainedJobs?: number; + completedJobTtlMs?: number; +} + +interface Delivery { + jobId: string; + message: AsyncJobDeliveryMessage; + attempt: number; + nextAttemptAt: number; + promise?: Promise; +} + +interface RegisteredSession { + disposed: boolean; + activeJobIds: Set; +} + +export class AsyncJobManager { + static #instance: AsyncJobManager | undefined; + + static instance(): AsyncJobManager | undefined { + return AsyncJobManager.#instance; + } + + static setInstance(value: AsyncJobManager | undefined): void { + AsyncJobManager.#instance = value; + } + + static resetForTests(): void { + AsyncJobManager.#instance = undefined; + } + + readonly #jobs = new Map(); + readonly #deliveries: Delivery[] = []; + readonly #suppressedDeliveries = new Set(); + readonly #inFlightDeliveries = new Map(); + readonly #deliveryHandlers = new Map(); + readonly #jobSessions = new Map(); + readonly #sessions = new Map(); + readonly #onJobComplete: AsyncJobDeliveryCallback; + readonly #maxRunningJobs: number; + readonly #maxRetainedJobs: number; + readonly #completedJobTtlMs: number; + #timer: NodeJS.Timeout | undefined; + #disposed = false; + #runningDeliveryLoop = false; + + constructor(options: AsyncJobManagerOptions) { + this.#onJobComplete = options.onJobComplete; + this.#maxRunningJobs = Math.max(1, Math.floor(options.maxRunningJobs ?? DEFAULT_MAX_RUNNING_JOBS)); + this.#maxRetainedJobs = Math.max(1, Math.floor(options.maxRetainedJobs ?? MAX_MANAGED_BASH_JOBS)); + this.#completedJobTtlMs = Math.max(0, Math.floor(options.completedJobTtlMs ?? COMPLETED_JOB_TTL_MS)); + } + + get disposed(): boolean { + return this.#disposed; + } + + get atCapacity(): boolean { + let running = 0; + for (const job of this.#jobs.values()) if (job.status === "running") running += 1; + return running >= this.#maxRunningJobs; + } + + + registerSession(): symbol { + if (this.#disposed) throw new Error("Async job manager is disposed"); + const sessionId = Symbol("async-job-session"); + this.#sessions.set(sessionId, { disposed: false, activeJobIds: new Set() }); + return sessionId; + } + + releaseSession(sessionId: symbol): void { + const session = this.#sessions.get(sessionId); + if (!session) return; + session.disposed = true; + this.acknowledgeDeliveries([...session.activeJobIds]); + this.#sessions.delete(sessionId); + if (this.#sessions.size === 0) this.dispose(); + } + + isSessionDisposed(sessionId: symbol): boolean { + return this.#disposed || this.#sessions.get(sessionId)?.disposed !== false; + } + registerBashJob(job: ManagedBashJob, onComplete?: AsyncJobDeliveryHandler, sessionId?: symbol): void { + if (this.#disposed) throw new Error("Async job manager is disposed"); + this.#pruneRetention(); + if (this.atCapacity) throw new Error(`Background job limit reached (${this.#maxRunningJobs}). Wait for running jobs to finish or cancel one.`); + if (sessionId !== undefined && this.isSessionDisposed(sessionId)) throw new Error("Async job session is disposed"); + this.#suppressedDeliveries.delete(job.jobId); + if (onComplete) this.#deliveryHandlers.set(job.jobId, onComplete); + else this.#deliveryHandlers.delete(job.jobId); + if (sessionId !== undefined) { + this.#jobSessions.set(job.jobId, sessionId); + this.#sessions.get(sessionId)?.activeJobIds.add(job.jobId); + } else this.#jobSessions.delete(job.jobId); + this.#jobs.set(job.jobId, job); + this.#pruneRetention(); + } + + completeBashJob(job: ManagedBashJob): void { + if (this.#disposed || this.#suppressedDeliveries.has(job.jobId)) return; + if (job.status === "running") return; + this.#jobs.set(job.jobId, job); + this.#pruneRetention(); + if (this.#jobs.has(job.jobId)) this.#enqueueDelivery(job as ManagedAsyncBashJob); + } + + acknowledgeDeliveries(jobIds: readonly string[]): void { + for (const jobId of jobIds) { + this.#suppressedDeliveries.add(jobId); + for (let index = this.#deliveries.length - 1; index >= 0; index -= 1) { + if (this.#deliveries[index]?.jobId === jobId) this.#deliveries.splice(index, 1); + } + this.#deliveryHandlers.delete(jobId); + this.#completeTrackedJob(jobId); + } + this.#pruneRetention(); + } + + isDeliverySuppressed(jobId: string): boolean { + return this.#suppressedDeliveries.has(jobId); + } + + deliveryState(): { queued: number; delivering: boolean; pendingJobIds: string[] } { + this.#pruneRetention(); + return { + queued: this.#deliveries.length, + delivering: this.#runningDeliveryLoop || this.#inFlightDeliveries.size > 0, + pendingJobIds: [...this.#deliveries.map((delivery) => delivery.jobId), ...this.#inFlightDeliveries.keys()], + }; + } + + retentionState(): { jobs: number; suppressions: number; handlers: number; queued: number; sessions: number } { + this.#pruneRetention(); + return { jobs: this.#jobs.size, suppressions: this.#suppressedDeliveries.size, handlers: this.#deliveryHandlers.size, queued: this.#deliveries.length + this.#inFlightDeliveries.size, sessions: this.#sessions.size }; + } + + dispose(): void { + this.#disposed = true; + if (this.#timer) clearTimeout(this.#timer); + this.#timer = undefined; + this.#deliveries.length = 0; + this.#inFlightDeliveries.clear(); + this.#deliveryHandlers.clear(); + this.#jobSessions.clear(); + this.#sessions.clear(); + this.#jobs.clear(); + this.#suppressedDeliveries.clear(); + } + + #enqueueDelivery(job: ManagedAsyncBashJob): void { + if (this.#inFlightDeliveries.has(job.jobId)) return; + for (let index = this.#deliveries.length - 1; index >= 0; index -= 1) if (this.#deliveries[index]?.jobId === job.jobId) this.#deliveries.splice(index, 1); + this.#deliveries.push({ jobId: job.jobId, message: formatAsyncResultForFollowUp(job), attempt: 0, nextAttemptAt: Date.now() }); + this.#pruneRetention(); + this.#scheduleDeliveryLoop(25); + } + + + #completeTrackedJob(jobId: string): void { + const sessionId = this.#jobSessions.get(jobId); + if (sessionId !== undefined) this.#sessions.get(sessionId)?.activeJobIds.delete(jobId); + this.#jobSessions.delete(jobId); + } + #pruneRetention(now = Date.now()): void { + for (const [jobId, job] of this.#jobs) { + if (job.status !== "running" && job.endedAt !== undefined && now - job.endedAt > this.#completedJobTtlMs) { + this.#jobs.delete(jobId); + this.#completeTrackedJob(jobId); + } + } + for (const jobId of this.#suppressedDeliveries) { + if (!this.#jobs.has(jobId)) this.#suppressedDeliveries.delete(jobId); + } + const oldestTerminalJobs = [...this.#jobs.values()] + .filter((job) => job.status !== "running") + .sort((a, b) => (a.endedAt ?? a.startedAt) - (b.endedAt ?? b.startedAt)); + while (this.#jobs.size > this.#maxRetainedJobs && oldestTerminalJobs.length > 0) { + const job = oldestTerminalJobs.shift(); + if (job) { + this.#jobs.delete(job.jobId); + this.#completeTrackedJob(job.jobId); + } + } + for (let index = this.#deliveries.length - 1; index >= 0; index -= 1) { + const delivery = this.#deliveries[index]; + if (!delivery || !this.#jobs.has(delivery.jobId) || this.#suppressedDeliveries.has(delivery.jobId)) this.#deliveries.splice(index, 1); + } + while (this.#deliveries.length + this.#inFlightDeliveries.size > this.#maxRetainedJobs && this.#deliveries.length > 0) this.#deliveries.shift(); + for (const jobId of this.#deliveryHandlers.keys()) if (!this.#jobs.has(jobId) || this.#suppressedDeliveries.has(jobId)) this.#deliveryHandlers.delete(jobId); + for (const session of this.#sessions.values()) for (const jobId of session.activeJobIds) if (!this.#jobs.has(jobId)) session.activeJobIds.delete(jobId); + } + + #scheduleDeliveryLoop(delayMs: number): void { + if (this.#disposed) return; + if (this.#timer) clearTimeout(this.#timer); + this.#timer = setTimeout(() => { + this.#timer = undefined; + void this.#runDeliveryLoop(); + }, delayMs); + this.#timer.unref?.(); + } + + async #runDeliveryLoop(): Promise { + if (this.#runningDeliveryLoop || this.#disposed) return; + this.#runningDeliveryLoop = true; + try { + while (!this.#disposed) { + const now = Date.now(); + const delivery = this.#deliveries.find((candidate) => candidate.nextAttemptAt <= now); + if (!delivery) break; + this.#deliveries.splice(this.#deliveries.indexOf(delivery), 1); + this.#startDelivery(delivery); + } + } finally { + this.#runningDeliveryLoop = false; + } + this.#scheduleNextDelivery(); + } + + #startDelivery(delivery: Delivery): void { + if (this.#suppressedDeliveries.has(delivery.jobId) || !this.#jobs.has(delivery.jobId) || this.#inFlightDeliveries.has(delivery.jobId)) return; + const handler = this.#deliveryHandlers.get(delivery.jobId) ?? this.#onJobComplete; + this.#inFlightDeliveries.set(delivery.jobId, delivery); + delivery.promise = (async () => handler(delivery.message))() + .then(() => { + if (!this.#suppressedDeliveries.has(delivery.jobId) && this.#jobs.has(delivery.jobId)) { + this.#deliveryHandlers.delete(delivery.jobId); + this.#completeTrackedJob(delivery.jobId); + } + }) + .catch(() => { + if (this.#disposed || this.#suppressedDeliveries.has(delivery.jobId) || !this.#jobs.has(delivery.jobId)) return; + delivery.attempt += 1; + const jitter = Math.floor(Math.random() * DELIVERY_RETRY_JITTER_MS); + delivery.nextAttemptAt = Date.now() + Math.min(DELIVERY_RETRY_MAX_MS, DELIVERY_RETRY_BASE_MS * 2 ** delivery.attempt) + jitter; + this.#deliveries.push(delivery); + }) + .finally(() => { + this.#inFlightDeliveries.delete(delivery.jobId); + this.#pruneRetention(); + this.#scheduleNextDelivery(); + }); + } + + #scheduleNextDelivery(): void { + const next = this.#deliveries.reduce((soonest, delivery) => soonest === undefined ? delivery.nextAttemptAt : Math.min(soonest, delivery.nextAttemptAt), undefined); + if (next !== undefined) this.#scheduleDeliveryLoop(Math.max(0, next - Date.now())); + } +} diff --git a/packages/coding-agent/src/core/async/session-manager.ts b/packages/coding-agent/src/core/async/session-manager.ts new file mode 100644 index 000000000..43316e67b --- /dev/null +++ b/packages/coding-agent/src/core/async/session-manager.ts @@ -0,0 +1,78 @@ +import type { CustomMessage } from "../messages.js"; +import type { SendMessageOptions } from "../extensions/index.js"; +import { AsyncJobManager } from "./job-manager.js"; +import type { AsyncJobDeliveryHandler, AsyncJobDeliveryMessage } from "./types.js"; + +export interface SessionAsyncJobManagerHandle { + manager: AsyncJobManager; + owns: boolean; + sessionId: symbol; +} +interface AsyncDeliverySession { + readonly isStreaming?: boolean; + sendCustomMessage( + message: Pick, "customType" | "content" | "display" | "details">, + options?: SendMessageOptions, + ): Promise; +} + +const STREAMING_DELIVERY_POLL_MS = 10; + +function scheduleBoundaryCheck(callback: () => void): NodeJS.Timeout { + const timer = setTimeout(callback, STREAMING_DELIVERY_POLL_MS); + timer.unref?.(); + return timer; +} + +function waitForStreamingBoundary(session: AsyncDeliverySession, isStale: () => boolean): Promise<"ready" | "stale"> { + return new Promise((resolve) => { + let timer: NodeJS.Timeout | undefined; + const settle = (value: "ready" | "stale") => { + if (timer) clearTimeout(timer); + resolve(value); + }; + const check = () => { + timer = undefined; + if (isStale()) { + settle("stale"); + return; + } + if (session.isStreaming !== true) { + settle("ready"); + return; + } + timer = scheduleBoundaryCheck(check); + }; + check(); + }); +} + +export function createSessionAsyncDeliveryHandler(session: AsyncDeliverySession, manager?: AsyncJobManager, sessionId?: symbol): AsyncJobDeliveryHandler { + return async (message: AsyncJobDeliveryMessage) => { + const isStale = () => + manager?.disposed === true || + manager?.isDeliverySuppressed(message.details.jobId) === true || + (sessionId !== undefined && manager?.isSessionDisposed(sessionId) === true); + if (await waitForStreamingBoundary(session, isStale) === "stale") return; + if (isStale()) return; + await session.sendCustomMessage(message, { deliverAs: "followUp", triggerTurn: true }); + }; +} + +export function createSessionAsyncJobManager(session: AsyncDeliverySession): SessionAsyncJobManagerHandle { + const existing = AsyncJobManager.instance(); + if (existing) return { manager: existing, owns: false, sessionId: existing.registerSession() }; + let manager: AsyncJobManager; + manager = new AsyncJobManager({ + onJobComplete: (message) => createSessionAsyncDeliveryHandler(session, manager)(message), + }); + AsyncJobManager.setInstance(manager); + return { manager, owns: true, sessionId: manager.registerSession() }; +} + +export function disposeSessionAsyncJobManager(manager: AsyncJobManager | undefined, sessionId: symbol | undefined): void { + if (!manager || !sessionId) return; + manager.releaseSession(sessionId); + if (manager.disposed && AsyncJobManager.instance() === manager) AsyncJobManager.setInstance(undefined); +} + diff --git a/packages/coding-agent/src/core/async/types.ts b/packages/coding-agent/src/core/async/types.ts new file mode 100644 index 000000000..c45265a41 --- /dev/null +++ b/packages/coding-agent/src/core/async/types.ts @@ -0,0 +1,28 @@ +import type { ManagedBashJob } from "../tools/bash-async-jobs.js"; + +export type AsyncJobStatus = "running" | "completed" | "failed"; + +export interface AsyncJobDeliveryDetails { + jobId: string; + type: "bash"; + status: AsyncJobStatus; + command: string; + exitCode?: number | null; + fullOutputPath?: string; + wallTimeMs?: number; +} + +export interface AsyncJobDeliveryMessage { + customType: "async-job-result"; + content: string; + display: true; + details: AsyncJobDeliveryDetails; +} + +export type AsyncJobDeliveryCallback = (message: AsyncJobDeliveryMessage) => void | Promise; + +export type AsyncJobDeliveryHandler = (message: AsyncJobDeliveryMessage) => void | Promise; + +export interface ManagedAsyncBashJob extends ManagedBashJob { + status: AsyncJobStatus; +} diff --git a/packages/coding-agent/src/core/tools/bash-async-execution.ts b/packages/coding-agent/src/core/tools/bash-async-execution.ts new file mode 100644 index 000000000..153e15142 --- /dev/null +++ b/packages/coding-agent/src/core/tools/bash-async-execution.ts @@ -0,0 +1,84 @@ +import type { AsyncJobManager } from "../async/job-manager.js"; +import type { AsyncJobDeliveryHandler } from "../async/types.js"; +import { invalidateNativeSearchCache } from "./search-native.js"; +import { createAsyncOutputAppender } from "./bash-async-output.js"; +import { createManagedBashJob, discardManagedBashJob, formatAsyncJobError } from "./bash-async-jobs.js"; +import type { BashOperations, BashToolDetails } from "./bash.js"; + +interface StartAsyncBashCommandOptions { + command: string; + cwd: string; + env: NodeJS.ProcessEnv; + pty?: boolean; + timeoutSeconds: number; + requestedTimeoutSeconds?: number; + signal?: AbortSignal; + operations: BashOperations; + manager?: AsyncJobManager; + deliveryHandler?: AsyncJobDeliveryHandler; + sessionId?: symbol; +} + +export async function startAsyncBashCommand(options: StartAsyncBashCommandOptions): Promise<{ + content: Array<{ type: "text"; text: string }>; + details: BashToolDetails; +}> { + if (options.manager?.atCapacity) throw new Error("Background job limit reached. Wait for running jobs to finish or cancel one."); + const job = createManagedBashJob(options.command, options.cwd, options.timeoutSeconds, options.requestedTimeoutSeconds); + try { + options.manager?.registerBashJob(job, options.deliveryHandler, options.sessionId); + } catch (registerError) { + // The job was inserted into the managed map but never started; drop it so + // it cannot linger as a permanently-"running" zombie entry (see + // discardManagedBashJob). Registration failures (disposed manager/session, + // capacity race) still surface to the caller as tool errors. + discardManagedBashJob(job.jobId); + throw registerError; + } + const appendAsyncOutput = createAsyncOutputAppender(job, { persistAfterBytes: 12_000 }); + const onParentAbort = () => { + options.manager?.acknowledgeDeliveries([job.jobId]); + job.abortController?.abort(); + }; + if (options.signal?.aborted) onParentAbort(); + else options.signal?.addEventListener("abort", onParentAbort, { once: true }); + void (async () => { + let error: Error | string | undefined; + let exitCode: number | null | undefined; + try { + exitCode = (await options.operations.exec(options.command, options.cwd, { + onData: appendAsyncOutput.append, + timeout: options.timeoutSeconds, + env: options.env, + pty: options.pty, + signal: job.abortController?.signal, + })).exitCode; + } catch (execError) { + error = execError instanceof Error ? execError : String(execError); + } + try { + await appendAsyncOutput.close(); + } catch (closeError) { + error ??= closeError instanceof Error ? closeError : String(closeError); + } + if (error !== undefined) { + job.status = "failed"; + job.error = job.abortController?.signal.aborted ? "aborted" : formatAsyncJobError(error); + } else { + job.exitCode = exitCode; + job.status = exitCode && exitCode !== 0 ? "failed" : "completed"; + } + job.endedAt = Date.now(); + invalidateNativeSearchCache(); + options.manager?.completeBashJob(job); + options.signal?.removeEventListener("abort", onParentAbort); + })(); + return { + content: [{ type: "text", text: `Started async bash command ${job.jobId}: ${options.command}\nPoll with bash({ command: "__atomic_bash_job ${job.jobId}" }); cancel with bash({ command: "__atomic_bash_job_cancel ${job.jobId}" })` }], + details: { + async: { jobId: job.jobId, type: "bash", state: "running", command: options.command, status: "running" }, + timeoutSeconds: options.timeoutSeconds, + ...(options.requestedTimeoutSeconds !== undefined ? { requestedTimeoutSeconds: options.requestedTimeoutSeconds } : {}), + }, + }; +} diff --git a/packages/coding-agent/src/core/tools/bash-async-jobs.ts b/packages/coding-agent/src/core/tools/bash-async-jobs.ts index f5ce16ff3..99012c10f 100644 --- a/packages/coding-agent/src/core/tools/bash-async-jobs.ts +++ b/packages/coding-agent/src/core/tools/bash-async-jobs.ts @@ -16,8 +16,8 @@ export interface ManagedBashJob { abortController?: AbortController; } -const MAX_MANAGED_BASH_JOBS = 100; -const COMPLETED_JOB_TTL_MS = 30 * 60 * 1000; +export const MAX_MANAGED_BASH_JOBS = 100; +export const COMPLETED_JOB_TTL_MS = 30 * 60 * 1000; const managedBashJobs = new Map(); export function formatAsyncJobError(error: unknown): string { const message = error instanceof Error ? error.message : String(error); @@ -48,6 +48,23 @@ export function createManagedBashJob(command: string, cwd: string, timeoutSecond return job; } +/** + * Remove a managed job that never actually started executing (for example when + * async-manager registration fails after the map insert). Without this, the + * entry would stay "running" forever: TTL cleanup only evicts settled jobs, so + * the zombie would linger until the max-jobs overflow forcibly removed it. + */ +export function discardManagedBashJob(jobId: string): void { + const job = managedBashJobs.get(jobId); + if (!job) return; + deleteJobOutput(job); + managedBashJobs.delete(jobId); +} + +export function listManagedBashJobIds(): string[] { + return [...managedBashJobs.keys()]; +} + export function getManagedBashJob(jobId: string): ManagedBashJob | undefined { cleanupManagedBashJobs(); const job = managedBashJobs.get(jobId); diff --git a/packages/coding-agent/src/core/tools/bash-async-output.ts b/packages/coding-agent/src/core/tools/bash-async-output.ts index 6a4e3e92b..ad4048dd3 100644 --- a/packages/coding-agent/src/core/tools/bash-async-output.ts +++ b/packages/coding-agent/src/core/tools/bash-async-output.ts @@ -30,7 +30,8 @@ function utf8Prefix(text: string, maxBytes: number): string { return text.slice(0, end); } -export function createAsyncOutputAppender(job: BashAsyncOutputTarget): BashAsyncOutputAppender { +export function createAsyncOutputAppender(job: BashAsyncOutputTarget, options?: { persistAfterBytes?: number }): BashAsyncOutputAppender { + const persistAfterBytes = options?.persistAfterBytes ?? DEFAULT_MAX_BYTES; let outputBytes = 0; let truncated = false; let fullOutputStream: WriteStream | undefined; @@ -50,6 +51,7 @@ export function createAsyncOutputAppender(job: BashAsyncOutputTarget): BashAsync const text = sanitizeDecodedOutput(decoded); if (text.length === 0) return; const bytes = byteLength(text); + if (outputBytes + bytes > persistAfterBytes) ensureFullOutputStream(); if (outputBytes + bytes > DEFAULT_MAX_BYTES) { ensureFullOutputStream(); const remaining = Math.max(0, DEFAULT_MAX_BYTES - outputBytes); diff --git a/packages/coding-agent/src/core/tools/bash.ts b/packages/coding-agent/src/core/tools/bash.ts index fa35d8bcd..360e5cef3 100644 --- a/packages/coding-agent/src/core/tools/bash.ts +++ b/packages/coding-agent/src/core/tools/bash.ts @@ -11,10 +11,12 @@ import { truncateToVisualLines } from "../../modes/interactive/components/visual import { theme } from "../../modes/interactive/theme/theme.ts"; import { waitForChildProcess } from "../../utils/child-process.ts"; import { getShellConfig, getShellEnv, killProcessTree, trackDetachedChildPid, untrackDetachedChildPid } from "../../utils/shell.ts"; +import type { AsyncJobManager } from "../async/job-manager.js"; +import type { AsyncJobDeliveryMessage } from "../async/types.js"; import type { BashResult } from "../bash-executor.ts"; import type { ToolDefinition, ToolRenderResultOptions } from "../extensions/types.ts"; -import { createAsyncOutputAppender } from "./bash-async-output.ts"; -import { abortManagedBashJob, createManagedBashJob, formatAsyncJobError, getManagedBashJob } from "./bash-async-jobs.ts"; +import { startAsyncBashCommand } from "./bash-async-execution.js"; +import { abortManagedBashJob, getManagedBashJob } from "./bash-async-jobs.ts"; import { stripLeadingCdCommand } from "./bash-leading-cd.ts"; import { executeNativePty } from "./bash-pty-native.ts"; import { checkBashInterceptionCandidates, DEFAULT_BASH_INTERCEPTOR_RULES, type BashInterceptorRule } from "./bash-interceptor.ts"; @@ -116,22 +118,25 @@ export interface BashInterceptorResult { } export type BashInterceptor = (context: BashSpawnContext) => Promise | BashInterceptorResult | undefined; export interface BashToolOptions { - /** Custom operations for command execution. Default: local shell */ operations?: BashOperations; - /** Command prefix prepended to every command (for example shell setup commands) */ + /** Prefix prepended to every shell command before execution. */ commandPrefix?: string; - /** Optional explicit shell path from settings */ + /** Override shell executable resolution for local bash operations. */ shellPath?: string; - /** Hook to adjust command, cwd, or env before execution */ + /** Last-mile hook for rewriting the command/cwd/env spawn context. */ spawnHook?: BashSpawnHook; + /** Optional command interceptor used by extensions and parity tests. */ interceptor?: BashInterceptor; interceptorEnabled?: boolean | (() => boolean); availableTools?: string[]; interceptorRules?: BashInterceptorRule[]; + /** Enable background bash jobs and session-managed async result delivery. */ asyncEnabled?: boolean; + asyncJobManager?: AsyncJobManager; + asyncJobDeliveryHandler?: (message: AsyncJobDeliveryMessage) => void | Promise; + asyncJobSessionId?: symbol; } -const BASH_PREVIEW_LINES = 5; -const BASH_UPDATE_THROTTLE_MS = 100; +const BASH_PREVIEW_LINES = 5, BASH_UPDATE_THROTTLE_MS = 100; const DEFAULT_TIMEOUT_SECONDS = 300; const MIN_TIMEOUT_SECONDS = 1; const MAX_TIMEOUT_SECONDS = 3600; @@ -266,6 +271,7 @@ export function createBashToolDefinition( const availableTools = options?.availableTools ?? ["read", "search", "find", "edit", "write"]; const interceptorRules = options?.interceptorRules ?? DEFAULT_BASH_INTERCEPTOR_RULES; const asyncEnabled = options?.asyncEnabled ?? false; + const asyncJobManager = options?.asyncJobManager, asyncJobDeliveryHandler = options?.asyncJobDeliveryHandler, asyncJobSessionId = options?.asyncJobSessionId; return { name: "bash", label: "bash", @@ -285,6 +291,7 @@ export function createBashToolDefinition( if (jobStatusMatch) { const job = getManagedBashJob(jobStatusMatch[1]!); if (!job) throw new Error(`Unknown bash async job: ${jobStatusMatch[1]}`); + if (job.status !== "running") asyncJobManager?.acknowledgeDeliveries([job.jobId]); const text = [`Job ${job.jobId}: ${job.status}`, `Command: ${job.command}`, job.error ? `Error: ${job.error}` : undefined, job.output].filter(Boolean).join("\n"); return { content: [{ type: "text", text }], details: { async: { jobId: job.jobId, type: "bash", state: job.status, command: job.command, status: job.status }, exitCode: job.exitCode, timeoutSeconds: job.timeoutSeconds, ...(job.requestedTimeoutSeconds !== undefined ? { requestedTimeoutSeconds: job.requestedTimeoutSeconds } : {}), ...(job.fullOutputPath ? { fullOutputPath: job.fullOutputPath } : {}), wallTimeMs: (job.endedAt ?? Date.now()) - job.startedAt } }; } @@ -292,6 +299,7 @@ export function createBashToolDefinition( if (jobCancelMatch) { const job = abortManagedBashJob(jobCancelMatch[1]!); if (!job) throw new Error(`Unknown bash async job: ${jobCancelMatch[1]}`); + asyncJobManager?.acknowledgeDeliveries([job.jobId]); return { content: [{ type: "text", text: `Cancellation requested for bash job ${job.jobId}` }], details: { async: { jobId: job.jobId, type: "bash", state: job.status, command: job.command, status: job.status } } }; } const timeout = normalizeTimeoutSeconds(bashCommand.timeout); @@ -325,26 +333,19 @@ export function createBashToolDefinition( const executionContext = primaryInterception ? spawnContext : (strippedCdContext ?? spawnContext); if (bashCommand.async) { if (!asyncEnabled) throw new Error("bash async execution is disabled"); - const job = createManagedBashJob(executionContext.command, executionContext.cwd, timeout, bashCommand.timeout !== undefined && bashCommand.timeout !== timeout ? bashCommand.timeout : undefined); - const appendAsyncOutput = createAsyncOutputAppender(job); - const onParentAbort = () => job.abortController?.abort(); - if (signal?.aborted) onParentAbort(); else signal?.addEventListener("abort", onParentAbort, { once: true }); - void (async () => { - let error: unknown, exitCode: number | null | undefined; - try { - exitCode = (await ops.exec(executionContext.command, executionContext.cwd, { onData: appendAsyncOutput.append, timeout, env: executionContext.env, pty: bashCommand.pty, signal: job.abortController?.signal })).exitCode; - } catch (execError: unknown) { error = execError; } - try { await appendAsyncOutput.close(); } catch (closeError: unknown) { error ??= closeError; } - if (error !== undefined) { - job.status = "failed"; - job.error = job.abortController?.signal.aborted ? "aborted" : formatAsyncJobError(error); - } else { - job.exitCode = exitCode; - job.status = exitCode && exitCode !== 0 ? "failed" : "completed"; - } - job.endedAt = Date.now(); invalidateNativeSearchCache(); signal?.removeEventListener("abort", onParentAbort); - })(); - return { content: [{ type: "text", text: `Started async bash command ${job.jobId}: ${executionContext.command}\nPoll with bash({ command: "__atomic_bash_job ${job.jobId}" }); cancel with bash({ command: "__atomic_bash_job_cancel ${job.jobId}" })` }], details: { async: { jobId: job.jobId, type: "bash", state: "running", command: executionContext.command, status: "running" }, timeoutSeconds: timeout, ...(job.requestedTimeoutSeconds !== undefined ? { requestedTimeoutSeconds: job.requestedTimeoutSeconds } : {}) } }; + return startAsyncBashCommand({ + command: executionContext.command, + cwd: executionContext.cwd, + env: executionContext.env, + pty: bashCommand.pty, + timeoutSeconds: timeout, + requestedTimeoutSeconds: bashCommand.timeout !== undefined && bashCommand.timeout !== timeout ? bashCommand.timeout : undefined, + signal, + operations: ops, + manager: asyncJobManager, + deliveryHandler: asyncJobDeliveryHandler, + sessionId: asyncJobSessionId, + }); } const output = new OutputAccumulator({ tempFilePrefix: `${APP_NAME}-bash` }); let acceptingOutput = true; diff --git a/packages/coding-agent/test/agent-session-async-bash.test.ts b/packages/coding-agent/test/agent-session-async-bash.test.ts new file mode 100644 index 000000000..3979a54d7 --- /dev/null +++ b/packages/coding-agent/test/agent-session-async-bash.test.ts @@ -0,0 +1,160 @@ +import { existsSync, mkdirSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Agent, type AgentMessage } from "@earendil-works/pi-agent-core"; +import { EventStream, getModel, type AssistantMessage, type AssistantMessageEvent, type TextContent } from "@earendil-works/pi-ai/compat"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { AgentSession } from "../src/core/agent-session.ts"; +import { AsyncJobManager } from "../src/core/async/job-manager.ts"; +import { AuthStorage } from "../src/core/auth-storage.ts"; +import { convertToLlm } from "../src/core/messages.ts"; +import { ModelRegistry } from "../src/core/model-registry.ts"; +import { SessionManager } from "../src/core/session-manager.ts"; +import { SettingsManager } from "../src/core/settings-manager.ts"; +import { createTestResourceLoader } from "./utilities.ts"; + +class MockAssistantStream extends EventStream { + constructor() { + super( + (event) => event.type === "done" || event.type === "error", + (event) => { + if (event.type === "done") return event.message; + if (event.type === "error") return event.error; + throw new Error("Unexpected event type"); + }, + ); + } +} + +function createAssistantMessage(text: string): AssistantMessage { + return { + role: "assistant", + content: [{ type: "text", text }], + api: "anthropic-messages", + provider: "anthropic", + 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(), + }; +} + +async function waitFor(predicate: () => boolean | Promise, timeoutMs = 1_500): Promise { + const start = Date.now(); + while (!(await predicate())) { + if (Date.now() - start > timeoutMs) throw new Error("Timed out waiting for condition"); + await new Promise((resolve) => setTimeout(resolve, 10)); + } +} + +function messageText(message: AgentMessage): string { + if (typeof message.content === "string") return message.content; + return message.content + .filter((part): part is TextContent => typeof part === "object" && part !== null && part.type === "text") + .map((part) => part.text) + .join("\n"); +} + +function createSession(tempDir: string, onTurn: (userTexts: string[], stream: MockAssistantStream) => void): AgentSession { + const model = getModel("anthropic", "claude-sonnet-4-5")!; + const agent = new Agent({ + convertToLlm, + getApiKey: () => "test-key", + initialState: { model, systemPrompt: "Test", tools: [] }, + streamFn: (_model, context) => { + const stream = new MockAssistantStream(); + queueMicrotask(() => onTurn(context.messages.filter((message) => message.role === "user").map(messageText), stream)); + return stream; + }, + }); + const authStorage = AuthStorage.create(join(tempDir, "auth.json")); + authStorage.setRuntimeApiKey("anthropic", "test-key"); + return new AgentSession({ + agent, + sessionManager: SessionManager.inMemory(), + settingsManager: SettingsManager.create(tempDir, tempDir), + cwd: tempDir, + modelRegistry: ModelRegistry.create(authStorage, tempDir), + resourceLoader: createTestResourceLoader(), + }); +} + +describe("AgentSession async bash auto-delivery", () => { + let tempDir: string; + let session: AgentSession | undefined; + + beforeEach(() => { + tempDir = join(tmpdir(), `atomic-async-bash-session-${Date.now()}-${Math.random().toString(36).slice(2)}`); + mkdirSync(tempDir, { recursive: true }); + }); + + afterEach(() => { + session?.dispose(); + AsyncJobManager.instance()?.dispose(); + AsyncJobManager.resetForTests(); + if (existsSync(tempDir)) rmSync(tempDir, { recursive: true }); + }); + + it("starts an idle follow-up turn from actual async bash completion", async () => { + const turns: string[][] = []; + session = createSession(tempDir, (userTexts, stream) => { + turns.push(userTexts); + stream.push({ type: "start", partial: createAssistantMessage("") }); + stream.push({ type: "done", reason: "stop", message: createAssistantMessage("done") }); + }); + const bash = session.getToolDefinition("bash"); + expect(bash).toBeDefined(); + await bash?.execute("bash-idle", { command: "printf idle-async", async: true }); + await waitFor(() => turns.some((turn) => turn.some((text) => text.includes("idle-async")))); + }); + + it("queues actual async bash completion as a follow-up while streaming and drains it after the turn", async () => { + let finishFirstTurn: (() => void) | undefined; + const turns: string[][] = []; + session = createSession(tempDir, (userTexts, stream) => { + turns.push(userTexts); + stream.push({ type: "start", partial: createAssistantMessage("") }); + if (userTexts.some((text) => text.includes("streaming-async"))) { + stream.push({ type: "done", reason: "stop", message: createAssistantMessage("follow-up") }); + return; + } + finishFirstTurn = () => stream.push({ type: "done", reason: "stop", message: createAssistantMessage("first") }); + }); + const firstPrompt = session.prompt("First message"); + await waitFor(() => session?.isStreaming === true); + const bash = session.getToolDefinition("bash"); + expect(bash).toBeDefined(); + await bash?.execute("bash-streaming", { command: "printf streaming-async", async: true }); + await new Promise((resolve) => setTimeout(resolve, 80)); + expect(turns.some((turn) => turn.some((text) => text.includes("streaming-async")))).toBe(false); + finishFirstTurn?.(); + await firstPrompt; + await waitFor(() => turns.some((turn) => turn.some((text) => text.includes("streaming-async")))); + }); + + it("suppresses a streaming-staged async result when polling acknowledges it before the follow-up drains", async () => { + let finishFirstTurn: (() => void) | undefined; + const turns: string[][] = []; + session = createSession(tempDir, (userTexts, stream) => { + turns.push(userTexts); + stream.push({ type: "start", partial: createAssistantMessage("") }); + finishFirstTurn = () => stream.push({ type: "done", reason: "stop", message: createAssistantMessage("done") }); + }); + const firstPrompt = session.prompt("First message"); + await waitFor(() => session?.isStreaming === true); + const bash = session.getToolDefinition("bash"); + expect(bash).toBeDefined(); + const started = await bash?.execute("bash-streaming-stale", { command: "printf stale-async", async: true }); + const jobId = started?.details?.async?.jobId; + expect(jobId).toBeDefined(); + await waitFor(() => AsyncJobManager.instance()?.deliveryState().delivering === true); + await waitFor(async () => { + const polled = await bash?.execute("bash-poll-stale", { command: `__atomic_bash_job ${jobId}` }); + return polled?.content.some((item) => item.type === "text" && item.text.includes("stale-async")) === true; + }); + finishFirstTurn?.(); + await firstPrompt; + await new Promise((resolve) => setTimeout(resolve, 120)); + expect(turns.some((turn) => turn.some((text) => text.includes("stale-async")))).toBe(false); + }); +}); diff --git a/packages/coding-agent/test/async-job-manager.test.ts b/packages/coding-agent/test/async-job-manager.test.ts new file mode 100644 index 000000000..cb7703489 --- /dev/null +++ b/packages/coding-agent/test/async-job-manager.test.ts @@ -0,0 +1,456 @@ +import { existsSync, readFileSync } from "node:fs"; +import { afterEach, describe, expect, it } from "vitest"; +import { AsyncJobManager } from "../src/core/async/job-manager.ts"; +import { createSessionAsyncDeliveryHandler, createSessionAsyncJobManager, disposeSessionAsyncJobManager } from "../src/core/async/session-manager.ts"; +import type { AsyncJobDeliveryMessage } from "../src/core/async/types.ts"; +import type { SendMessageOptions } from "../src/core/extensions/index.ts"; +import { listManagedBashJobIds, type ManagedBashJob } from "../src/core/tools/bash-async-jobs.ts"; +import { createBashToolDefinition } from "../src/core/tools/bash.ts"; +async function waitFor(predicate: () => boolean | Promise, timeoutMs = 1_000): Promise { + const start = Date.now(); + while (!(await predicate())) { + if (Date.now() - start > timeoutMs) throw new Error("Timed out waiting for condition"); + await new Promise((resolve) => setTimeout(resolve, 10)); + } +} + +function text(result: { content: Array<{ type: string; text?: string }> }): string { + return result.content.map((item) => item.text ?? "").join("\n"); +} + +function requireJobId(value: string | undefined): string { + if (!value) throw new Error("Expected async job id"); + return value; +} + + +interface CapturedCustomMessage { + message: AsyncJobDeliveryMessage; + options: SendMessageOptions | undefined; +} + +function createCapturedSession(captured: CapturedCustomMessage[], options?: { isStreaming?: boolean }): { + readonly isStreaming?: boolean; + sendCustomMessage: (message: AsyncJobDeliveryMessage, options?: SendMessageOptions) => Promise; +} { + return { + get isStreaming() { return options?.isStreaming; }, + sendCustomMessage: async (message, options) => { + captured.push({ message, options }); + }, + }; +} + +afterEach(() => { + AsyncJobManager.instance()?.dispose(); + AsyncJobManager.resetForTests(); +}); + +describe("AsyncJobManager", () => { + it("delivers bash completions through session sendCustomMessage as a follow-up turn", async () => { + const captured: CapturedCustomMessage[] = []; + const session = createCapturedSession(captured); + const { manager } = createSessionAsyncJobManager(session); + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: manager, + asyncJobDeliveryHandler: createSessionAsyncDeliveryHandler(session), + operations: { exec: async (_command, _cwd, { onData }) => { onData(Buffer.from("session done\n")); return { exitCode: 0 }; } }, + }); + + await bash.execute("bash-session-async", { command: "echo session done", async: true }); + + await waitFor(() => captured.length === 1); + expect(captured[0]?.message.customType).toBe("async-job-result"); + expect(captured[0]?.message.display).toBe(true); + expect(captured[0]?.message.content).toContain("session done"); + expect(captured[0]?.options).toEqual({ deliverAs: "followUp", triggerTurn: true }); + }); + + it("routes shared singleton jobs to the session that started each job", async () => { + const ownerCaptured: CapturedCustomMessage[] = []; + const laterCaptured: CapturedCustomMessage[] = []; + const ownerSession = createCapturedSession(ownerCaptured); + const laterSession = createCapturedSession(laterCaptured); + const { manager, owns } = createSessionAsyncJobManager(ownerSession); + const laterHandle = createSessionAsyncJobManager(laterSession); + expect(owns).toBe(true); + expect(laterHandle.manager).toBe(manager); + expect(laterHandle.owns).toBe(false); + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: laterHandle.manager, + asyncJobDeliveryHandler: createSessionAsyncDeliveryHandler(laterSession), + operations: { exec: async (_command, _cwd, { onData }) => { onData(Buffer.from("later session\n")); return { exitCode: 0 }; } }, + }); + + await bash.execute("bash-later-session", { command: "echo later session", async: true }); + + await waitFor(() => laterCaptured.length === 1); + expect(ownerCaptured).toHaveLength(0); + expect(laterCaptured[0]?.message.content).toContain("later session"); + }); + + + it("keeps the shared manager alive when the owner disposes while a later session has an active job", async () => { + const ownerCaptured: CapturedCustomMessage[] = []; + const laterCaptured: CapturedCustomMessage[] = []; + const ownerSession = createCapturedSession(ownerCaptured); + const laterSession = createCapturedSession(laterCaptured); + const ownerHandle = createSessionAsyncJobManager(ownerSession); + const laterHandle = createSessionAsyncJobManager(laterSession); + let releaseExec: (() => void) | undefined; + const execFinished = new Promise((resolve) => { releaseExec = resolve; }); + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: laterHandle.manager, + asyncJobDeliveryHandler: createSessionAsyncDeliveryHandler(laterSession, laterHandle.manager, laterHandle.sessionId), + asyncJobSessionId: laterHandle.sessionId, + operations: { exec: async (_command, _cwd, { onData }) => { await execFinished; onData(Buffer.from("later active\n")); return { exitCode: 0 }; } }, + }); + + await bash.execute("bash-later-active", { command: "echo later active", async: true }); + disposeSessionAsyncJobManager(ownerHandle.manager, ownerHandle.sessionId); + expect(laterHandle.manager.disposed).toBe(false); + expect(AsyncJobManager.instance()).toBe(laterHandle.manager); + releaseExec?.(); + + await waitFor(() => laterCaptured.length === 1); + expect(ownerCaptured).toHaveLength(0); + expect(laterCaptured[0]?.message.content).toContain("later active"); + disposeSessionAsyncJobManager(laterHandle.manager, laterHandle.sessionId); + expect(laterHandle.manager.disposed).toBe(true); + expect(AsyncJobManager.instance()).toBeUndefined(); + }); + + it("releases disposed non-owner streaming handlers without blocking other sessions", async () => { + const staleCaptured: CapturedCustomMessage[] = []; + const liveCaptured: CapturedCustomMessage[] = []; + const ownerSession = createCapturedSession([]); + const staleSession = createCapturedSession(staleCaptured, { isStreaming: true }); + const liveSession = createCapturedSession(liveCaptured); + const ownerHandle = createSessionAsyncJobManager(ownerSession); + const staleHandle = createSessionAsyncJobManager(staleSession); + const liveHandle = createSessionAsyncJobManager(liveSession); + const makeBash = (textValue: string, handle: typeof staleHandle, session: typeof staleSession) => createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: handle.manager, + asyncJobDeliveryHandler: createSessionAsyncDeliveryHandler(session, handle.manager, handle.sessionId), + asyncJobSessionId: handle.sessionId, + operations: { exec: async (_command, _cwd, { onData }) => { onData(Buffer.from(`${textValue}\n`)); return { exitCode: 0 }; } }, + }); + + await makeBash("stale", staleHandle, staleSession).execute("bash-stale", { command: "echo stale", async: true }); + await waitFor(() => staleHandle.manager.deliveryState().delivering === true || staleHandle.manager.deliveryState().queued === 0); + disposeSessionAsyncJobManager(staleHandle.manager, staleHandle.sessionId); + await waitFor(() => staleHandle.manager.deliveryState().delivering === false); + await makeBash("live", liveHandle, liveSession).execute("bash-live", { command: "echo live", async: true }); + + await waitFor(() => liveCaptured.length === 1); + expect(staleCaptured).toHaveLength(0); + expect(liveCaptured[0]?.message.content).toContain("live"); + expect(ownerHandle.manager.disposed).toBe(false); + disposeSessionAsyncJobManager(ownerHandle.manager, ownerHandle.sessionId); + disposeSessionAsyncJobManager(liveHandle.manager, liveHandle.sessionId); + expect(ownerHandle.manager.disposed).toBe(true); + }); + + it("does not let one live streaming session block unrelated async delivery", async () => { + const streamingCaptured: CapturedCustomMessage[] = []; + const liveCaptured: CapturedCustomMessage[] = []; + const ownerSession = createCapturedSession([]); + const streamingSession = createCapturedSession(streamingCaptured, { isStreaming: true }); + const liveSession = createCapturedSession(liveCaptured); + const ownerHandle = createSessionAsyncJobManager(ownerSession); + const streamingHandle = createSessionAsyncJobManager(streamingSession); + const liveHandle = createSessionAsyncJobManager(liveSession); + const makeBash = (textValue: string, handle: typeof streamingHandle, session: typeof streamingSession) => createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: handle.manager, + asyncJobDeliveryHandler: createSessionAsyncDeliveryHandler(session, handle.manager, handle.sessionId), + asyncJobSessionId: handle.sessionId, + operations: { exec: async (_command, _cwd, { onData }) => { onData(Buffer.from(`${textValue}\n`)); return { exitCode: 0 }; } }, + }); + + await makeBash("streaming blocks", streamingHandle, streamingSession).execute("bash-streaming-blocker", { command: "echo streaming blocks", async: true }); + await waitFor(() => streamingHandle.manager.deliveryState().delivering === true); + await makeBash("live proceeds", liveHandle, liveSession).execute("bash-live-proceeds", { command: "echo live proceeds", async: true }); + + await waitFor(() => liveCaptured.length === 1); + expect(streamingCaptured).toHaveLength(0); + expect(liveCaptured[0]?.message.content).toContain("live proceeds"); + disposeSessionAsyncJobManager(streamingHandle.manager, streamingHandle.sessionId); + disposeSessionAsyncJobManager(liveHandle.manager, liveHandle.sessionId); + disposeSessionAsyncJobManager(ownerHandle.manager, ownerHandle.sessionId); + }); + + it("keeps disposed running job suppression until completion so fallback delivery cannot occur", async () => { + const delivered: AsyncJobDeliveryMessage[] = []; + const manager = new AsyncJobManager({ onJobComplete: (message) => { delivered.push(message); }, completedJobTtlMs: 20 }); + const ownerSessionId = manager.registerSession(); + const disposedSessionId = manager.registerSession(); + const disposedSession = createCapturedSession([]); + let releaseExec: (() => void) | undefined; + const execFinished = new Promise((resolve) => { releaseExec = resolve; }); + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: manager, + asyncJobDeliveryHandler: createSessionAsyncDeliveryHandler(disposedSession, manager, disposedSessionId), + asyncJobSessionId: disposedSessionId, + operations: { exec: async (_command, _cwd, { onData }) => { await execFinished; onData(Buffer.from("disposed complete\n")); return { exitCode: 0 }; } }, + }); + + const started = await bash.execute("bash-disposed-running", { command: "echo disposed complete", async: true }); + const jobId = requireJobId(started.details?.async?.jobId); + manager.releaseSession(disposedSessionId); + await new Promise((resolve) => setTimeout(resolve, 60)); + expect(manager.isDeliverySuppressed(jobId)).toBe(true); + releaseExec?.(); + await waitFor(async () => text(await bash.execute("bash-disposed-poll", { command: `__atomic_bash_job ${jobId}` })).includes("disposed complete")); + await new Promise((resolve) => setTimeout(resolve, 80)); + + expect(delivered).toHaveLength(0); + expect(manager.isDeliverySuppressed(jobId)).toBe(true); + manager.releaseSession(ownerSessionId); + }); + + it("delivers completed bash jobs as async-job-result custom messages", async () => { + const delivered: AsyncJobDeliveryMessage[] = []; + const manager = new AsyncJobManager({ onJobComplete: (message) => { delivered.push(message); } }); + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: manager, + operations: { exec: async (_command, _cwd, { onData }) => { onData(Buffer.from("done\n")); return { exitCode: 0 }; } }, + }); + const started = await bash.execute("bash-async", { command: "echo done", async: true }); + expect(started.details?.async?.status).toBe("running"); + await waitFor(() => delivered.length === 1); + expect(delivered[0]?.customType).toBe("async-job-result"); + expect(delivered[0]?.display).toBe(true); + expect(delivered[0]?.content).toContain("Async bash job"); + expect(delivered[0]?.content).toContain("done"); + expect(delivered[0]?.details.status).toBe("completed"); + manager.dispose(); + }); + + it("acknowledges completed jobs when polled before queued delivery runs", async () => { + const delivered: AsyncJobDeliveryMessage[] = []; + const manager = new AsyncJobManager({ onJobComplete: (message) => { delivered.push(message); } }); + let releaseExec: (() => void) | undefined; + const execFinished = new Promise((resolve) => { releaseExec = resolve; }); + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: manager, + operations: { exec: async (_command, _cwd, { onData }) => { await execFinished; onData(Buffer.from("polled\n")); return { exitCode: 0 }; } }, + }); + const started = await bash.execute("bash-async", { command: "echo polled", async: true }); + const jobId = requireJobId(started.details?.async?.jobId); + releaseExec?.(); + await waitFor(() => manager.deliveryState().queued === 1); + const polled = await bash.execute("bash-poll", { command: `__atomic_bash_job ${jobId}` }); + expect(text(polled)).toContain("polled"); + expect(manager.isDeliverySuppressed(jobId)).toBe(true); + await new Promise((resolve) => setTimeout(resolve, 60)); + expect(delivered).toHaveLength(0); + manager.dispose(); + }); + + it("drops a streaming-staged delivery when the job is acknowledged before the session boundary", async () => { + const captured: CapturedCustomMessage[] = []; + let streaming = true; + const session = createCapturedSession(captured, { get isStreaming() { return streaming; } }); + const manager = new AsyncJobManager({ onJobComplete: createSessionAsyncDeliveryHandler(session) }); + const handler = createSessionAsyncDeliveryHandler(session, manager); + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: manager, + asyncJobDeliveryHandler: handler, + operations: { exec: async (_command, _cwd, { onData }) => { onData(Buffer.from("queued then polled\n")); return { exitCode: 0 }; } }, + }); + const started = await bash.execute("bash-stream-stale", { command: "echo queued then polled", async: true }); + const jobId = requireJobId(started.details?.async?.jobId); + await waitFor(async () => text(await bash.execute("bash-poll", { command: `__atomic_bash_job ${jobId}` })).includes("queued then polled")); + streaming = false; + await new Promise((resolve) => setTimeout(resolve, 60)); + expect(captured).toHaveLength(0); + expect(manager.isDeliverySuppressed(jobId)).toBe(true); + manager.dispose(); + }); + + it("settles delayed streaming delivery without sending after manager disposal", async () => { + const captured: CapturedCustomMessage[] = []; + const session = createCapturedSession(captured, { isStreaming: true }); + const manager = new AsyncJobManager({ onJobComplete: () => undefined }); + const handler = createSessionAsyncDeliveryHandler(session, manager); + const message: AsyncJobDeliveryMessage = { + customType: "async-job-result", + content: "Async bash job job-disposed completed: echo disposed\n\ndisposed", + display: true, + details: { jobId: "job-disposed", type: "bash", status: "completed", command: "echo disposed", exitCode: 0 }, + }; + let settled = false; + const delivery = Promise.resolve(handler(message)).then(() => { settled = true; }); + + manager.dispose(); + await Promise.race([ + delivery, + new Promise((_resolve, reject) => setTimeout(() => reject(new Error("delivery wait did not settle after dispose")), 100)), + ]); + + expect(settled).toBe(true); + expect(manager.disposed).toBe(true); + expect(captured).toHaveLength(0); + }); + it("retries delivery failures", async () => { + const delivered: AsyncJobDeliveryMessage[] = []; + let attempts = 0; + const manager = new AsyncJobManager({ + onJobComplete: (message) => { + attempts += 1; + if (attempts === 1) throw new Error("temporary failure"); + delivered.push(message); + }, + }); + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: manager, + operations: { exec: async (_command, _cwd, { onData }) => { onData(Buffer.from("retry\n")); return { exitCode: 0 }; } }, + }); + await bash.execute("bash-async", { command: "echo retry", async: true }); + await waitFor(() => delivered.length === 1, 1_500); + expect(attempts).toBe(2); + manager.dispose(); + }); + + it("enforces the running job bound", async () => { + const manager = new AsyncJobManager({ onJobComplete: () => undefined, maxRunningJobs: 1 }); + let releaseExec: (() => void) | undefined; + const execFinished = new Promise((resolve) => { releaseExec = resolve; }); + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: manager, + operations: { exec: async () => { await execFinished; return { exitCode: 0 }; } }, + }); + await bash.execute("bash-async-1", { command: "sleep", async: true }); + await expect(bash.execute("bash-async-2", { command: "sleep", async: true })).rejects.toThrow(/Background job limit reached/); + releaseExec?.(); + manager.dispose(); + }); + + it("discards the managed job when async registration fails on a disposed manager/session", async () => { + const captured: CapturedCustomMessage[] = []; + const session = createCapturedSession(captured); + const handle = createSessionAsyncJobManager(session); + let execStarted = false; + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: handle.manager, + asyncJobDeliveryHandler: createSessionAsyncDeliveryHandler(session, handle.manager, handle.sessionId), + asyncJobSessionId: handle.sessionId, + operations: { exec: async () => { execStarted = true; return { exitCode: 0 }; } }, + }); + disposeSessionAsyncJobManager(handle.manager, handle.sessionId); + const before = listManagedBashJobIds(); + await expect(bash.execute("bash-zombie", { command: "echo zombie", async: true })).rejects.toThrow(/disposed/); + const leaked = listManagedBashJobIds().filter((jobId) => !before.includes(jobId)); + expect(leaked).toEqual([]); + expect(execStarted).toBe(false); + }); + + it("suppresses auto-delivery after explicit async bash cancellation while preserving pollability", async () => { + const delivered: AsyncJobDeliveryMessage[] = []; + const manager = new AsyncJobManager({ onJobComplete: (message) => { delivered.push(message); } }); + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: manager, + operations: { exec: async (_command, _cwd, { signal }) => new Promise<{ exitCode: number | null }>((_resolve, reject) => { + signal?.addEventListener("abort", () => reject(new Error("aborted")), { once: true }); + }) }, + }); + const started = await bash.execute("bash-cancel-start", { command: "sleep", async: true }); + const jobId = requireJobId(started.details?.async?.jobId); + await bash.execute("bash-cancel", { command: `__atomic_bash_job_cancel ${jobId}` }); + await waitFor(async () => text(await bash.execute("bash-poll", { command: `__atomic_bash_job ${jobId}` })).includes("failed")); + await new Promise((resolve) => setTimeout(resolve, 60)); + expect(delivered).toHaveLength(0); + expect(manager.isDeliverySuppressed(jobId)).toBe(true); + manager.dispose(); + }); + + it("suppresses auto-delivery when the parent tool signal aborts an async bash job", async () => { + const delivered: AsyncJobDeliveryMessage[] = []; + const manager = new AsyncJobManager({ onJobComplete: (message) => { delivered.push(message); } }); + const controller = new AbortController(); + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: manager, + operations: { exec: async (_command, _cwd, { signal }) => new Promise<{ exitCode: number | null }>((_resolve, reject) => { + signal?.addEventListener("abort", () => reject(new Error("aborted")), { once: true }); + }) }, + }); + const started = await bash.execute("bash-parent-abort", { command: "sleep", async: true }, controller.signal); + const jobId = requireJobId(started.details?.async?.jobId); + controller.abort(); + await waitFor(async () => text(await bash.execute("bash-poll", { command: `__atomic_bash_job ${jobId}` })).includes("failed")); + await new Promise((resolve) => setTimeout(resolve, 60)); + expect(delivered).toHaveLength(0); + expect(manager.isDeliverySuppressed(jobId)).toBe(true); + manager.dispose(); + }); + + it("bounds retained jobs, suppressions, handlers, and queued deliveries", () => { + const manager = new AsyncJobManager({ onJobComplete: () => undefined, maxRetainedJobs: 2, completedJobTtlMs: 60_000 }); + const now = Date.now(); + for (let index = 0; index < 4; index += 1) { + const job: ManagedBashJob = { jobId: `job-${index}`, command: "echo", cwd: process.cwd(), status: "completed", output: `${index}`, startedAt: now + index, endedAt: now + index + 1 }; + manager.registerBashJob(job, () => undefined); + manager.completeBashJob(job); + manager.acknowledgeDeliveries([job.jobId]); + } + expect(manager.retentionState()).toEqual({ jobs: 2, suppressions: 2, handlers: 0, queued: 0, sessions: 0 }); + manager.dispose(); + }); + + it("keeps just-under-threshold async follow-up output inline when formatted text exceeds preview limit", async () => { + const delivered: AsyncJobDeliveryMessage[] = []; + const manager = new AsyncJobManager({ onJobComplete: (message) => { delivered.push(message); } }); + const output = "y".repeat(11_980); + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: manager, + operations: { exec: async (_command, _cwd, { onData }) => { onData(Buffer.from(output)); return { exitCode: 0 }; } }, + }); + + await bash.execute("bash-boundary", { command: "boundary", async: true }); + await waitFor(() => delivered.length === 1); + expect(delivered[0]?.details.fullOutputPath).toBeUndefined(); + expect(delivered[0]?.content).toContain(output); + expect(delivered[0]?.content).not.toContain("Output truncated for async follow-up"); + expect(delivered[0]?.content.length).toBeGreaterThan(12_000); + manager.dispose(); + }); + + it("persists full output for 12KB-50KB async follow-up truncation without changing poll output", async () => { + const delivered: AsyncJobDeliveryMessage[] = []; + const manager = new AsyncJobManager({ onJobComplete: (message) => { delivered.push(message); } }); + const output = "x".repeat(20_000); + const bash = createBashToolDefinition(process.cwd(), { + asyncEnabled: true, + asyncJobManager: manager, + operations: { exec: async (_command, _cwd, { onData }) => { onData(Buffer.from(output)); return { exitCode: 0 }; } }, + }); + const started = await bash.execute("bash-large", { command: "large", async: true }); + const jobId = requireJobId(started.details?.async?.jobId); + await waitFor(() => delivered.length === 1); + const path = delivered[0]?.details.fullOutputPath; + expect(path).toBeDefined(); + expect(path && existsSync(path)).toBe(true); + expect(path ? readFileSync(path, "utf8") : "").toHaveLength(output.length); + expect(delivered[0]?.content).toContain("Output truncated for async follow-up"); + expect(delivered[0]?.content).toContain(`Full output: ${path}`); + expect(delivered[0]?.content.length).toBeLessThan(output.length); + const polled = await bash.execute("bash-large-poll", { command: `__atomic_bash_job ${jobId}` }); + expect(text(polled)).toContain(output); + manager.dispose(); + }); +});