diff --git a/packages/agent/src/agent-loop.ts b/packages/agent/src/agent-loop.ts index a4a5a70f2..36ba3587e 100644 --- a/packages/agent/src/agent-loop.ts +++ b/packages/agent/src/agent-loop.ts @@ -647,15 +647,18 @@ async function runLoopBody( stream.push({ type: "turn_end", message, toolResults }); - pendingMessages = steeringMessagesFromExecution ?? ((await config.getSteeringMessages?.()) || []); + const steering = steeringMessagesFromExecution ?? ((await config.getSteeringMessages?.()) || []); + const asides = (await config.getAsideMessages?.()) || []; + pendingMessages = asides.length > 0 ? [...steering, ...asides] : steering; } - // Agent would stop here. Check for follow-up messages. + // Agent would stop here. Drain non-interrupting asides + follow-up messages. await config.onBeforeYield?.(); + const asideMessages = (await config.getAsideMessages?.()) || []; const followUpMessages = (await config.getFollowUpMessages?.()) || []; - if (followUpMessages.length > 0) { - // Set as pending so inner loop processes them - pendingMessages = followUpMessages; + if (asideMessages.length > 0 || followUpMessages.length > 0) { + // Set as pending so the inner loop processes them before stopping. + pendingMessages = [...asideMessages, ...followUpMessages]; continue; } diff --git a/packages/agent/src/agent.ts b/packages/agent/src/agent.ts index 99496de45..2dbe08de0 100644 --- a/packages/agent/src/agent.ts +++ b/packages/agent/src/agent.ts @@ -319,6 +319,7 @@ export class Agent { #onAssistantMessageEvent?: (message: AssistantMessage, event: AssistantMessageEvent) => void; #onHarmonyLeak?: (event: HarmonyAuditEvent) => void | Promise; #onBeforeYield?: () => Promise | void; + #asideMessageProvider?: () => AgentMessage[] | Promise; #telemetry?: AgentLoopConfig["telemetry"]; #appendOnlyContext?: AppendOnlyContextManager; @@ -629,6 +630,15 @@ export class Agent { this.#onBeforeYield = fn; } + /** + * Provide a source of non-interrupting "aside" messages (e.g. background-job + * completions, late LSP diagnostics) drained at each step boundary. Never + * aborts in-flight tools. See `AgentLoopConfig.getAsideMessages`. + */ + setAsideMessageProvider(fn: (() => AgentMessage[] | Promise) | undefined): void { + this.#asideMessageProvider = fn; + } + emitExternalEvent(event: AgentEvent) { switch (event.type) { case "message_start": @@ -999,6 +1009,7 @@ export class Agent { return this.#dequeueSteeringMessages(); }, getFollowUpMessages: async () => this.#dequeueFollowUpMessages(), + getAsideMessages: async () => (await this.#asideMessageProvider?.()) ?? [], onBeforeYield: () => this.#onBeforeYield?.(), telemetry: this.#telemetry, }; diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index 246c8ed60..de32f298d 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -132,6 +132,17 @@ export interface AgentLoopConfig extends SimpleStreamOptions { * continues with another turn. */ getFollowUpMessages?: () => Promise; + /** + * Returns non-interrupting "aside" messages to inject at a step boundary. + * + * Polled after each tool batch (before the next LLM call) AND at the yield + * check. Unlike steering, these NEVER abort in-flight tools — they are passive + * notifications (e.g. background-job completions, late LSP diagnostics) that + * should reach the model between requests without waiting for the agent to + * fully stop. Returned messages are appended to the context with normal + * message events and keep the loop running so the model can react. + */ + getAsideMessages?: () => Promise; /** * Hook fired right before the loop would exit. * diff --git a/packages/coding-agent/src/session/agent-session.ts b/packages/coding-agent/src/session/agent-session.ts index 927c13cc6..51f083502 100644 --- a/packages/coding-agent/src/session/agent-session.ts +++ b/packages/coding-agent/src/session/agent-session.ts @@ -1174,7 +1174,6 @@ export class AgentSession { this.agent.setRawSseEventInterceptor(this.#onSseEvent); this.yieldQueue = new YieldQueue({ isStreaming: () => this.isStreaming, - injectStreaming: message => this.agent.followUp(message), injectIdle: async messages => { const first = messages[0]; if (!first) return; @@ -1189,7 +1188,10 @@ export class AgentSession { ); }, }); - this.agent.setOnBeforeYield(() => this.yieldQueue.flush("streaming")); + // Background-job completions / late diagnostics are pulled into the run at + // each step boundary as non-interrupting asides (see Agent.getAsideMessages), + // so they reach the model between requests without waiting for a yield. + this.agent.setAsideMessageProvider(() => this.yieldQueue.drainMessages()); this.#convertToLlm = config.convertToLlm ?? convertToLlm; this.#rebuildSystemPrompt = config.rebuildSystemPrompt; this.#getMcpServerInstructions = config.getMcpServerInstructions; @@ -3040,7 +3042,7 @@ export class AgentSession { this.#isDisposed = true; this.#pendingBackgroundExchanges = []; this.yieldQueue.clear(); - this.agent.setOnBeforeYield(undefined); + this.agent.setAsideMessageProvider(undefined); this.#evalExecutionDisposing = true; try { if (this.#extensionRunner?.hasHandlers("session_shutdown")) { diff --git a/packages/coding-agent/src/session/yield-queue.ts b/packages/coding-agent/src/session/yield-queue.ts index a329531af..bb92a8e5e 100644 --- a/packages/coding-agent/src/session/yield-queue.ts +++ b/packages/coding-agent/src/session/yield-queue.ts @@ -10,7 +10,7 @@ export interface YieldDispatcher

{ export interface YieldQueueOptions { isStreaming: () => boolean; - injectStreaming(msg: AgentMessage): void; + injectStreaming?(msg: AgentMessage): void; injectIdle(messages: AgentMessage[]): Promise; scheduleIdleFlush(run: () => Promise): void; } @@ -85,7 +85,7 @@ export class YieldQueue { if (!message) continue; if (mode === "streaming") { try { - this.#options.injectStreaming(message); + this.#options.injectStreaming?.(message); } catch (error) { logger.warn("Yield queue streaming dispatch failed", { kind, error: formatError(error) }); } @@ -102,6 +102,24 @@ export class YieldQueue { } } + /** + * Build and remove all queued messages, applying each dispatcher's staleness + * filter. No injection side effects — used for pull-based delivery at agent + * step boundaries (see `Agent.setAsideMessageProvider`), so background-job + * completions and late diagnostics reach the model between requests without + * the agent having to stop. + */ + drainMessages(): AgentMessage[] { + const messages: AgentMessage[] = []; + for (const [kind, dispatcher] of this.#dispatchers) { + const entries = this.#drain(kind); + if (entries.length === 0) continue; + const message = this.#build(kind, dispatcher, entries); + if (message) messages.push(message); + } + return messages; + } + clear(): void { this.#entries.clear(); this.#idleFlushPending = false;