From ecd60e3cd4313643dede3cb2fcf13b4e2e9a2a75 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebastian=20M=C3=BCller?= Date: Fri, 11 Sep 2026 18:47:54 +0200 Subject: [PATCH] fix: withdraw background command notices the model already read (#2159) * feat(coding-agent): keep the system prompt static across refinements and deliver harness state in-context Fixes RES-1318. Applying a refinement no longer rebuilds and swaps agent.state.systemPrompt, so the provider prefix cache survives every refinement. Instead: - _applyRefine appends one durable refinement_notice custom message per applied refinement at the apply seam, labeled [auto-refinement] / [user-refinement] / [self-refinement] by source, with the trigger line and applied edits in the digest's own notation (rollbacks print as rollbacks). Zero-applied-edit refinements emit nothing and the notice never starts a turn. convertToLlm passes the new type through as a user message while the refinement_outcome audit message stays filtered and untouched. - The harness digest injection is removed from the system prompt entirely (buildSystemPrompt loses the harnessState option). The digest now appears at cold context boundaries only, reusing formatHarnessStateForPrompt byte-identically: fresh sessions get a digest custom message as the first context message; compaction attaches a digest snapshot to the compaction entry mechanically (never through the summarizer) so the post-compaction head message renders memories-first before the summary on both the initial and update-merge paths; resumes append a fresh digest at the tail only when the newest digest in live context no longer matches disk state (identity dedupe), which is also the migration path for pre-change sessions. The new compaction entry field and custom message types ride the existing schema-tolerant JSON entry stream (optional field, generic custom messages): backward-compatible, no daemon protocol or schema revision change. * test(coding-agent): adapt message-sequence pins to the session-start harness digest Every session now opens with a harness_digest custom message, so tests that pinned exact message sequences, first-message identity, provider-context text lists, or entry counts see one extra leading custom message. Adds a conversationMessages helper to the suite harness (filters the digest) and updates the affected assertions; provider-context probes now select the last user message instead of the first. * fix(coding-agent): harden harness digest recency and keep digests out of the summarizer Review fixes on #2098: - _latestContextHarnessDigest picked the last digest by array position, but retained pre-compaction messages are presented after the compaction head while being chronologically older, so an old retained digest could defeat resume dedupe. Recency is now the greatest message timestamp among all in-context digest carriers. - The session-start harness digest entry was fed to the first compaction's summarizer as an ordinary custom message. prepareCompaction now excludes harness_digest entries from summarizer input; the digest is only attached mechanically at the new compaction head. * fix(coding-agent): label non-serialized refine.run notices as self-refinement Review fix on #2098: _consumePendingRequestedRefine forwarded agent-callable refine.run requests through refine() without a source, so the notice carried the [user-refinement] label in non-serialized sessions. refine() now accepts an explicit source override and the pending-request consumer passes "self". * test(coding-agent): pin the self source on the pending refine.run consumer * fix(coding-agent): treat the digest as boundary-injected mechanical context everywhere Review round on #2098 (three findings): - Untouched sessions stay empty: the fresh-session digest is now injected lazily at the first committed turn instead of at construction, so abandoned-draft cleanup and raw message-count emptiness checks (daemon-session-list lifecycle/activity, isEmptyDraftContent, branch seedability, hasExistingSession) keep seeing an empty session. Non-empty contexts (resume) still inject at construction with identity dedupe. - Tree navigation is a cold boundary: navigateTree rebuilds the context, so it now runs the same refresh-iff-stale digest check as resume (fork/clone/switch construct a new session and were already covered by the constructor path). - Headless terminal selection skips harness_digest like other non-output customs, so resume-and-print returns the saved final assistant output. Sweep of other message-kind-sensitive consumers: branch summarization now excludes harness_digest from summarizer input (same rule as compaction); estimateContextTokens and side-question seeding intentionally keep the digest (it is live context); _repliedToParentSinceTask and getUserMessagesForForking are kind-filtered already. * chore(coding-agent): trim narrative comments and redundant test asserts on the digest work Comments cut to one-line invariant guards; duplicate digest-notation and content asserts removed where a kept unit pin already covers the same rendering (labels and notation stay pinned in refinement-outcome-message.test, source plumbing stays pinned per path). No behavioral change. * fix(coding-agent): scope the first-turn digest to the turn that delivers it Review fix on #2098 (Bugbot thread 2919fcf3): the first-turn digest was persisted directly at commit, before agent.prompt ran, so cancelling or clearing that first turn stripped the turn's captured messages but left the digest in state and in the session file - resurfacing the emptiness-check class through the cancel path. The digest now rides the turn's next-turn delivery records: it persists at message_end like the rest of the turn, first-turn cancellation strips and restores it with the turn, and the lazy injection re-delivers on the next committed turn. * test(coding-agent): event-order pins include the digest custom pair on the first turn * fix(coding-agent): deliver the first-turn digest on skip-policy turns and re-arm on cancel Two follow-up review findings on the lazy-injection shape: - Custom-triggered turns use the skip next-turn-context policy, so a fresh session whose first turn was sendCustomMessage(triggerTurn) ran without the digest. The pending digest now attaches on every committed first turn; skip policy still withholds ordinary pending next-turn messages. - A cancelled first turn whose digest record was already durable dropped the digest from live context without restoring it (restorable records skip durable ones). Lazy injection now owns digest delivery uniformly: cancelled turns never restore the digest message and instead re-arm the pending flag, so the next committed turn re-delivers a fresh one. * fix(coding-agent): never park the harness digest as pending next-turn context Review fix on #2098 (Bugbot skip-turn-digest-lost-on-failure): a failed commit parked the undelivered digest in _pendingNextTurnMessages, which skip-policy custom-trigger turns never drain, and a parked copy plus a re-armed injection could double-deliver. Parking now filters the digest out and re-arms _harnessDigestPending instead, keeping lazy injection the single owner of digest delivery. * feat(coding-agent): unify synthetic user-channel messages under one bracket grammar Every machine-injected user-channel message now opens with one grammar: `[(: )(
)]`, a blank line, then the payload. - agent_message: `[agent-message from :]`; message id and endpoint ids live in details only. Detection keys on customType; the old header parser stays for persisted legacy transcripts (accept-both, emit-new). - heartbeat_prompt: `[heartbeat: run#]` instead of the raw prompt masquerading as the user. - async_bash_completion: `[bash-done pid: exit:]`; the standing BashHandle inspection hint is gone (it duplicated the system prompt). - rlm child notices: `[child-exited: no-reply|cancelled child:]` and `[child-failed child:]`. - goal_context: `[goal: continuation|budget-limit|objective-updated]` replaces the XML wrapper. - ipython_state_restored: `[python-state-restored]` label (customType keeps its string); goal prose says "Python REPL" instead of "ipython". - compactionSummary/branchSummary/harness digest prefixes gain `[compaction-summary]`/`[branch-summary]`/`[harness-digest]` first lines. clearQueuedAgentMessages now matches queued actions by the delivered custom message's customType, with the legacy text parser as fallback. Fixes RES-1321 * fix(coding-agent): sanitize names interpolated into bracket headers One owner (sanitizeMessageHeaderValue) strips brackets, newlines, commas, and ":" from header-interpolated names: an RLM child name can no longer close the [child-failed]/[child-exited] header early, and an unlabeled sender named "parent:root" can no longer render as a relationship label. Addresses the Bugbot/Macroscope review findings on PR 2115. * fix(coding-agent): keep the heartbeat header on one line for whitespace-bearing schedules parseAgentCronSchedule's every/in regexes accept embedded newlines, so the schedule shown in the [heartbeat: ...] header goes through the shared header sanitizer; the exact expression stays in details.schedule. * fix(coding-agent): bring the remaining built-in injectors into the bracket grammar Review sweep of every built-in customType against convertToLlm found three model-visible kinds still emitting unmarked content: - autonomous_status -> `[autonomous-status: on|off]` + usage payload - autonomous continuations -> `[autonomous-continuation]` + prompt, and `[autonomous-continuation: gate-failed]` owned by buildAutonomousGateFailureContinuation (covers the headless injector too) - prime-agent.update_complete -> `[update-complete]` + notice - ipython_state (kernel persisted through compaction) -> `[python-state]` All other custom types are either grammar-compliant or filtered out of LLM context by convertToLlm / entry-only. A table-driven pin now walks every constructor-backed synthetic kind and asserts a grammar-conforming first line. * test(coding-agent): adapt daemon-mode and model-extension pins to the bracket grammar The full CI matrix (first run since the PR retargeted main) caught two expectations outside test/suite still pinning pre-grammar text: the retained- subagent send now asserts the exact `[agent-message from ...]` prompt, and the model_select injected-prompt pin carries the `[heartbeat: ...]` header. * feat(runtime): tell the host when a background command result is read A background command usually exits before the model looks at its handle, so the completion notice is queued while the model is already reading the result in the next cell. Track that read on the handle: reads before the notice is dispatched cancel it outright, and reads after the host accepted it send one bash.consumed request so the host can withdraw the queued notice. Only reads that reach the model count: the running command's buffer, and awaits by consumers that outlive their cell, are not result reads. * feat(coding-agent): withdraw background command notices the model already read A queued completion notice for a command whose result the kernel already read is noise: it re-announces work the model just inspected and starts an extra turn for it. Handle bash.consumed by dropping the matching notice while it is still queued; delivered notices are no longer clearable, so nothing is retracted after the model saw them, and an unread completion is delivered as before. Notices are matched on pid and command because pids are reused across handles. The queue and transcript label becomes "Background command finished", which describes what happened instead of naming the transport. * fix(runtime): only count result reads that a live cell receives The first version marked a finished handle as read on any output/tail/poll call, so a detached watcher polling the handle between turns withdrew the completion notice. An idle session then never learned the command finished: the notice was its only wake-up. Decide it from one rule instead, in one place: a read counts only while a cell is executing, because the value and anything the reader prints ride that cell's turn, and an awaiting reader must also be one that live cell waits for, since a discarded or detached wrapper task awaits a value nobody sees. The live cell comes from the kernel's own active-cell state rather than the cell contextvar, which detached tasks keep long after their cell ended. * fix(coding-agent): withdraw one notice per result read A read belongs to one handle, so it may retire one notice. Cancelling every action that matches the pid and command meant a pid reused for an identical command took a second, unread notice down with the first. Withdraw the earliest match only, which is the notice of the older handle the read belongs to. The RLM docs still taught the old rule (await in the creating cell to suppress) and the old "Shell message" name, so they now describe the withdrawal and which reads count. * fix: drop function duplicated by the main merge * test: trim redundant pins and slim comments --- .../.changes/res-1341-consumed-bash-notice.md | 1 + packages/coding-agent/docs/rlm.md | 4 +- .../coding-agent/src/core/agent-session.ts | 35 +++++++++ packages/coding-agent/src/core/messages.ts | 2 +- packages/coding-agent/src/core/prompts/rlm.ts | 2 +- packages/coding-agent/src/core/rlm-runtime.ts | 22 ++++++ .../2053-background-bash-passivation.test.ts | 34 ++++++++ .../2068-shell-message-steering.test.ts | 58 +++++++++++++- prime-agent-runtime/src/rlm/bash.py | 78 ++++++++++++++++++- prime-agent-runtime/src/rlm/repl.py | 8 ++ prime-agent-runtime/test/test_repl.py | 72 +++++++++++++++-- 11 files changed, 302 insertions(+), 14 deletions(-) create mode 100644 packages/coding-agent/.changes/res-1341-consumed-bash-notice.md diff --git a/packages/coding-agent/.changes/res-1341-consumed-bash-notice.md b/packages/coding-agent/.changes/res-1341-consumed-bash-notice.md new file mode 100644 index 000000000..1a770080b --- /dev/null +++ b/packages/coding-agent/.changes/res-1341-consumed-bash-notice.md @@ -0,0 +1 @@ +- Background command completion notices are now withdrawn when the agent already read the result: reading a finished `bash()` handle from a live cell (awaiting it, or calling `poll()`, `output()`, or `tail()`) drops a notice that is still queued and stops one that was not sent yet. Reads that no cell receives, such as a detached watcher polling between turns, still leave the notice in place, so an unread completion wakes an idle session exactly as before, and a delivered notice is never retracted. The queue and transcript label for these messages is now "Background command finished" instead of "Shell message received". diff --git a/packages/coding-agent/docs/rlm.md b/packages/coding-agent/docs/rlm.md index 12f440594..205ec990f 100644 --- a/packages/coding-agent/docs/rlm.md +++ b/packages/coding-agent/docs/rlm.md @@ -55,9 +55,9 @@ checks = bash("npm test") checks.pid ``` -When an unawaited handle's process group finishes, Prime Agent sends a Shell message with its PID and foreground exit code. A busy agent receives it as steering at the next safe turn boundary, without interrupting a running tool. An idle agent resumes to handle it. `await handle` and `handle.poll()` still return the foreground result before shell background jobs finish. The kernel stays resident until the process group is reaped, including for handles awaited in their creating cell. The message asks the agent to inspect the saved handle with `poll()`, `output()`, or `tail()` and continue the task. `await bash(...)` stays synchronous from the agent's perspective and does not send a second Shell message. +When an unawaited handle's process group finishes, Prime Agent sends a "Background command finished" notice with its PID and foreground exit code. A busy agent receives it as steering at the next safe turn boundary, without interrupting a running tool. An idle agent resumes to handle it. `await handle` and `handle.poll()` still return the foreground result before shell background jobs finish. The kernel stays resident until the process group is reaped, including for handles awaited in their creating cell. The message asks the agent to inspect the saved handle with `poll()`, `output()`, or `tail()` and continue the task. `await bash(...)` stays synchronous from the agent's perspective and does not send a second notice. -Awaiting the original handle in its creating cell suppresses the Shell message, even after the command finishes. If an `asyncio.as_completed` wrapper task finishes before the cell starts consuming its result, that cached-result read does not mark the handle as awaited. A Shell message can still arrive. Await the original handle in the creating cell to suppress it. +Reading the finished result withdraws the notice. Any read from a live cell counts -- `await handle`, `handle.poll()`, `handle.output()`, `handle.tail()` -- so a notice that is still queued is dropped and a notice that was not sent yet is never sent. Reads that no cell receives do not count: a detached watcher polling the handle between turns leaves the notice in place, because that notice is the only wake-up an idle session gets, and an `asyncio.as_completed` or `create_task` wrapper that resolves with no live cell waiting for it does not count either. A notice the agent already received is never retracted. Each `bash()` call is its own process, while Python state, `os.chdir(...)`, and `os.environ[...]` changes persist in the kernel and apply to later `bash()` calls. Prime Agent extensions may intentionally add custom tools, but the built-in RLM design does not require a separate model tool for every capability. diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 431c10837..c8411ee20 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -237,6 +237,7 @@ import type { ResourceExtensionPaths, ResourceLoader } from "./resource-loader.j import { type CreateRlmSubagentRuntimeOptions, createAsyncBashCompletionHostHandler, + createAsyncBashConsumedHostHandler, createDefaultRlmSubagentSessionName, createRlmCreateSessionHostHandler, createRlmDeleteSubagentHostHandler, @@ -5024,6 +5025,37 @@ export class AgentSession { for (const id of ids) this._durableRlmTerminalNoticeActionIds.delete(id); } + /** + * The kernel read the command's result before the notice reached the model, so + * the notice has nothing left to report: drop it while it is still queued. + * Delivered notices are no longer clearable, which makes this a no-op. + */ + private _withdrawAsyncBashCompletionNotice(details: { pid: number; command: string }): void { + // One read withdraws one notice: pid reuse can queue an identical key twice, + // and the read belongs to the older handle, which is the earlier notice. + const notice = this._actionStore + .clearableActions() + .find((action) => this._isAsyncBashCompletionActionFor(action, details)); + if (!notice) return; + this._cancelSessionActions( + (action) => action === notice, + new Error("Background command completion notice withdrawn: the kernel read the result first."), + ); + this._emitQueueUpdate(); + } + + private _isAsyncBashCompletionActionFor( + action: QueuedSessionAction, + details: { pid: number; command: string }, + ): boolean { + if (action.payload.kind !== "turn") return false; + const message = primaryDeliveryRecord(action).message; + if (message.role !== "custom" || message.customType !== ASYNC_BASH_COMPLETION_CUSTOM_TYPE) return false; + // pids are reused across handles, so the command has to match too. + const completion = message.details as AsyncBashCompletionDetails | undefined; + return completion?.pid === details.pid && completion.command === details.command; + } + private async _promptInjectedMessage( text: string, message: CustomMessage, @@ -9709,6 +9741,9 @@ export class AgentSession { } } }), + "bash.consumed": createAsyncBashConsumedHostHandler((details) => { + this._withdrawAsyncBashCompletionNotice(details); + }), "rlm.find_models": createRlmFindModelsHostHandler((query, limit) => this.findRlmModels(query, limit)), "rlm.list_subagents": createRlmListSubagentsHostHandler(() => this.listRlmSubagents()), "rlm.delete_subagent": createRlmDeleteSubagentHostHandler((target) => this.deleteRlmSubagent(target)), diff --git a/packages/coding-agent/src/core/messages.ts b/packages/coding-agent/src/core/messages.ts index fbb687285..09f3c5cea 100644 --- a/packages/coding-agent/src/core/messages.ts +++ b/packages/coding-agent/src/core/messages.ts @@ -47,7 +47,7 @@ export const HARNESS_DIGEST_CUSTOM_TYPE = "harness_digest"; export const RLM_CHILD_FAILURE_CUSTOM_TYPE = "rlm_child_failure"; export const RLM_CHILD_TERMINAL_NOTICE_CUSTOM_TYPE = "rlm_child_terminal_notice"; export const ASYNC_BASH_COMPLETION_CUSTOM_TYPE = "async_bash_completion"; -export const ASYNC_BASH_COMPLETION_PREVIEW_LABEL = "Shell message received"; +export const ASYNC_BASH_COMPLETION_PREVIEW_LABEL = "Background command finished"; /** * Names and other metadata interpolated into a `[ ...]` header line must not diff --git a/packages/coding-agent/src/core/prompts/rlm.ts b/packages/coding-agent/src/core/prompts/rlm.ts index f20c7566d..c8b932cc8 100644 --- a/packages/coding-agent/src/core/prompts/rlm.ts +++ b/packages/coding-agent/src/core/prompts/rlm.ts @@ -12,7 +12,7 @@ export interface RlmPromptOptions { } const LONG_RUNNING_WORK_PROMPT = [ - "For slow or independently completing work, use a nonblocking control loop: start the work, record its handle or output location, then end your turn. A `bash()` handle left running beyond its creating cell sends a completion follow-up; when it arrives, inspect the saved handle and continue.", + "For slow or independently completing work, use a nonblocking control loop: start the work, record its handle or output location, then end your turn. A `bash()` handle left running beyond its creating cell sends a completion follow-up; when it arrives, inspect the saved handle and continue. Reading a finished handle's result first cancels that follow-up.", "When delegation is available and useful, assign independent substantive tasks to separate workers. Start independent workers without waiting for each one sequentially, and let them run in parallel.", "Do not keep the turn open by polling with `time.sleep()` or shell `sleep`, and do not replace polling with a long blocking `await`. Await only the short operation needed to start work or inspect a result that is already available; otherwise end the turn.", ].join("\n"); diff --git a/packages/coding-agent/src/core/rlm-runtime.ts b/packages/coding-agent/src/core/rlm-runtime.ts index b9f655be8..fdef11f9c 100644 --- a/packages/coding-agent/src/core/rlm-runtime.ts +++ b/packages/coding-agent/src/core/rlm-runtime.ts @@ -73,6 +73,13 @@ interface AsyncBashCompletionRequest { } type AsyncBashCompletionHandler = (request: AsyncBashCompletionRequest) => void | Promise; + +interface AsyncBashConsumedRequest { + pid: number; + command: string; +} + +type AsyncBashConsumedHandler = (request: AsyncBashConsumedRequest) => void | Promise; export type RlmListSubagentsHandler = () => RlmListSubagentsResult | Promise; export type RlmDeleteSubagentHandler = (target: string) => Promise; export type RlmFindModelsHandler = (query: string, limit: number) => RlmFindModelsResult | Promise; @@ -232,6 +239,21 @@ export function createAsyncBashCompletionHostHandler(handler: AsyncBashCompletio }; } +/** The kernel read a finished command's result, so its completion notice is stale. */ +export function createAsyncBashConsumedHostHandler(handler: AsyncBashConsumedHandler): HostRequestHandler { + return async (payload) => { + const { pid, command } = payload; + if (typeof pid !== "number" || !Number.isInteger(pid) || pid <= 0) { + throw new Error("bash.consumed pid must be a positive integer"); + } + if (typeof command !== "string" || !command) { + throw new Error("bash.consumed command must be a non-empty string"); + } + await handler({ pid, command }); + return {}; + }; +} + /** Search a bounded authenticated model catalog without adding it to the system prompt. */ export function createRlmFindModelsHostHandler(handler: RlmFindModelsHandler): HostRequestHandler { return async (payload) => { diff --git a/packages/coding-agent/test/suite/regressions/2053-background-bash-passivation.test.ts b/packages/coding-agent/test/suite/regressions/2053-background-bash-passivation.test.ts index 3577a046b..4da5a7745 100644 --- a/packages/coding-agent/test/suite/regressions/2053-background-bash-passivation.test.ts +++ b/packages/coding-agent/test/suite/regressions/2053-background-bash-passivation.test.ts @@ -198,6 +198,40 @@ describeRuntime("#2053 background kernel bash residency", () => { } }); + it("withdraws the queued completion notice when a later cell reads the result", async () => { + const notice = createDeferred(); + const release = createDeferred(); + try { + const { session, kernel } = await start(async () => { + notice.resolve(); + await release.promise; + }); + harness!.setResponses([ + async () => { + // The notice queues while the model is busy in the cell that reads the result. + release.resolve(); + await vi.waitFor(() => expect(session.getSteeringMessages()).toHaveLength(1)); + const read = await kernel.execute("handle.output()"); + expect(read.result).toContain("read-later"); + await vi.waitFor(() => expect(session.getSteeringMessages()).toEqual([])); + return fauxAssistantMessage("Read the background result directly."); + }, + ]); + await kernel.execute("from rlm import bash\nhandle = bash('printf read-later')"); + await notice.promise; + await session.prompt("Keep working."); + await session.waitForIdle(); + expect( + session.messages.filter( + (message) => message.role === "custom" && message.customType === ASYNC_BASH_COMPLETION_CUSTOM_TYPE, + ), + ).toEqual([]); + expect(harness!.eventsOfType("agent_start")).toHaveLength(1); + } finally { + release.resolve(); + } + }); + it("reports a terminal readiness failure without retrying or retaining completed work", async () => { const beforeCompletion = vi.fn(async () => {}); const { session, kernel } = await start(beforeCompletion, false); diff --git a/packages/coding-agent/test/suite/regressions/2068-shell-message-steering.test.ts b/packages/coding-agent/test/suite/regressions/2068-shell-message-steering.test.ts index 04c07e297..012e91157 100644 --- a/packages/coding-agent/test/suite/regressions/2068-shell-message-steering.test.ts +++ b/packages/coding-agent/test/suite/regressions/2068-shell-message-steering.test.ts @@ -24,6 +24,13 @@ function completeShell(harness: Harness) { return (harness.session as unknown as KernelSession)._createKernelHostHandlers()["bash.completed"]!(completion); } +function readShellResult(harness: Harness, command = completion.command) { + return (harness.session as unknown as KernelSession)._createKernelHostHandlers()["bash.consumed"]!({ + pid: completion.pid, + command, + }); +} + function shellMessages(harness: Harness) { return harness.session.messages.filter( (message) => message.role === "custom" && message.customType === ASYNC_BASH_COMPLETION_CUSTOM_TYPE, @@ -81,7 +88,7 @@ describe("#2068 shell message steering", () => { const queued = harness.session.getSessionActionRecoverySnapshot().actions; expect(queued).toContainEqual(expect.objectContaining({ delivery: "next_turn_boundary" })); const preview = harness.session.getSteeringMessagePreviews()[0]!; - expect(preview).toBe("Shell message received: pid 42, exit 0"); + expect(preview).toBe("Background command finished: pid 42, exit 0"); expect(formatQueuedMessagePreview(preview, "Steering")).toBe(preview); } finally { release.resolve(); @@ -114,11 +121,58 @@ describe("#2068 shell message steering", () => { await harness.session.waitForIdle(); expect(requests).toBe(1); expect(shellMessages(harness)).toHaveLength(1); + // A result read after delivery has nothing to withdraw. + await readShellResult(harness); + expect(shellMessages(harness)).toHaveLength(1); expect(isAgentSessionMessage(shellMessages(harness)[0]!)).toBe(false); expect(harness.session.getSteeringMessages()).toEqual([]); expect(harness.session.getFollowUpMessages()).toEqual([]); }); + it("withdraws a queued shell notice once the kernel reads the result", async () => { + const started = createDeferred(); + const release = createDeferred(); + const tool: AgentTool = { + name: "wait", + label: "Wait", + description: "Hold the current tool until released", + parameters: Type.Object({}), + execute: async () => { + started.resolve(); + await release.promise; + return { content: [{ type: "text", text: "released" }], details: {} }; + }, + }; + const harness = await createHarness({ tools: [tool] }); + harnesses.push(harness); + harness.setResponses([ + fauxAssistantMessage(fauxToolCall("wait", {}), { stopReason: "toolUse" }), + fauxAssistantMessage("Finished the original work."), + ]); + const original = harness.session.prompt("Continue working."); + try { + await started.promise; + await completeShell(harness); + expect(harness.session.getSteeringMessages()).toHaveLength(1); + // pids are reused across handles, so another command must not withdraw this notice. + await readShellResult(harness, "other command"); + expect(harness.session.getSteeringMessages()).toHaveLength(1); + // pid reuse can queue an identical key twice; one read withdraws one notice. + await completeShell(harness); + expect(harness.session.getSteeringMessages()).toHaveLength(2); + await readShellResult(harness); + expect(harness.session.getSteeringMessages()).toHaveLength(1); + await readShellResult(harness); + expect(harness.session.getSteeringMessages()).toEqual([]); + } finally { + release.resolve(); + await original; + } + await harness.session.waitForIdle(); + expect(shellMessages(harness)).toEqual([]); + expect(harness.eventsOfType("agent_start")).toHaveLength(1); + }); + it("renders shell-specific queue and transcript labels without generic delivery prefixes", () => { const message = createAsyncBashCompletionMessage(completion); const preview = `${ASYNC_BASH_COMPLETION_PREVIEW_LABEL}: pid 42, exit 0`; @@ -131,7 +185,7 @@ describe("#2068 shell message steering", () => { .render(120) .join("\n") .replace(/\u001b\[[0-9;]*m/g, ""); - expect(render()).toContain("◆ Shell message received · pid 42 · exit 0"); + expect(render()).toContain("◆ Background command finished · pid 42 · exit 0"); expect(render()).not.toMatch(/Follow-up:|Steering:|Agent message received/); component.setExpanded(true); expect(render()).toContain("[bash-done pid:42 exit:0]"); diff --git a/prime-agent-runtime/src/rlm/bash.py b/prime-agent-runtime/src/rlm/bash.py index 43cbcfb69..cf325725b 100644 --- a/prime-agent-runtime/src/rlm/bash.py +++ b/prime-agent-runtime/src/rlm/bash.py @@ -158,6 +158,18 @@ def _creating_cell_waits_for( return _completion_reaches(awaiter, targets) +def _live_cell_owner() -> asyncio.Task[Any] | None: + """Body task of the cell executing right now, ignoring detached context copies.""" + try: + from . import repl + + if repl.is_active(): + return repl.active_cell_task() + except (ImportError, RuntimeError): + pass + return None + + @dataclass(frozen=True) class BashResult: exit_code: int @@ -242,6 +254,8 @@ class BashHandle: self._result: BashResult | None = None self._callbacks: list[Callable[[], None]] = [] self._reap_callback: Callable[[], None] | None = None + self._result_consumed = False + self._consumed_notice: Callable[[], None] | None = None self._callback_lock = threading.Lock() # Serializes kill/reap so a pid fallback can never outlive the process handle. self._kill_lock = threading.Lock() @@ -364,14 +378,17 @@ class BashHandle: def output(self) -> str: self._released = True + self._note_result_consumed() return self._buffer.text() def tail(self, n: int = 50) -> str: self._released = True + self._note_result_consumed() return "\n".join(self._buffer.text().splitlines()[-n:]) def poll(self) -> BashResult | None: self._released = True + self._note_result_consumed() return self._result if self._done.is_set() else None def kill(self, sig: int = signal.SIGTERM, grace: float = 5.0) -> None: @@ -633,6 +650,26 @@ class BashHandle: return callback() + def _note_result_consumed(self, awaiter: asyncio.Task[Any] | None = None) -> None: + """Record a result read that reaches the model: only reads during a live + cell count (a detached reader between turns must keep the notice — it is + the idle session's only wake-up), and an awaiting reader must be one the + live cell waits for.""" + if not self._done.is_set(): + return + owner = _live_cell_owner() + if owner is None: + return + if awaiter is not None and not _creating_cell_waits_for(owner, awaiter): + return + with self._callback_lock: + if self._result_consumed: + return + self._result_consumed = True + notice, self._consumed_notice = self._consumed_notice, None + if notice is not None: + notice() + def _schedule_background_completion_notice(self) -> None: cell_finished = self._creating_cell_finished if cell_finished is None: @@ -667,7 +704,7 @@ class BashHandle: # The cell may do other work before awaiting this handle. Do not classify # it as detached until that whole cell has crossed its completion barrier. await cell_finished.wait() - if self._awaited_by_creating_cell or not repl.is_active(): + if self._awaited_by_creating_cell or self._result_consumed or not repl.is_active(): return command = self.command if len(command) > _COMPLETION_NOTICE_COMMAND_CAP: @@ -680,7 +717,10 @@ class BashHandle: "exitCode": result.exit_code, } ) - if not isinstance(reply, dict) or reply.get("status") != "ok": + if isinstance(reply, dict) and reply.get("status") == "ok": + # Notice accepted by the host; later reads must ask it to withdraw. + self._arm_consumed_notice(command) + else: sys.stderr.write( f"Background bash completion follow-up for pid {self._pid} was not accepted. " "Inspect the saved handle with poll(), output(), or tail().\n" @@ -693,6 +733,38 @@ class BashHandle: # Reap and deliver (or report rejection) before releasing kernel residency. repl.emit({"application/vnd.prime-agent.bash-activity+json": {**activity, "active": False}}) + def _arm_consumed_notice(self, command: str) -> None: + # Armed only post-acceptance: the withdrawal can never overtake its notice. + loop = asyncio.get_running_loop() + + def dispatch() -> None: + def start() -> None: + task = loop.create_task(self._notify_result_consumed(command)) + task.add_done_callback(_consume_notice_task) + + try: + loop.call_soon_threadsafe(start) + except RuntimeError: + pass # notifying loop already closed + + with self._callback_lock: + if not self._result_consumed: + self._consumed_notice = dispatch + return + dispatch() + + async def _notify_result_consumed(self, command: str) -> None: + from . import repl + + if not repl.is_active(): + return + try: + await repl.host_request( + {"type": "bash.consumed", "pid": self._pid, "command": command} + ) + except (OSError, RuntimeError): + return # bridge closed at teardown; old hosts error-reply — both fine + async def _wait_reaped(self) -> None: loop = asyncio.get_running_loop() future: asyncio.Future[None] = loop.create_future() @@ -857,6 +929,8 @@ class BashHandle: or _creating_cell_waits_for(self._creating_cell_task, current_task) ): self._awaited_by_creating_cell = True + if completed: + self._note_result_consumed(current_task) def __repr__(self) -> str: state = f"exit_code={self._result.exit_code}" if self._result else "running" diff --git a/prime-agent-runtime/src/rlm/repl.py b/prime-agent-runtime/src/rlm/repl.py index d3324be77..413b2a83a 100644 --- a/prime-agent-runtime/src/rlm/repl.py +++ b/prime-agent-runtime/src/rlm/repl.py @@ -116,6 +116,14 @@ def current_cell_completion_context() -> tuple[asyncio.Event, asyncio.Task[Any] return execution.finished, execution.owner +def active_cell_task() -> asyncio.Task[Any] | None: + """The cell body task executing right now, or None between cells (global + state, not the cell contextvar — detached tasks keep stale context copies).""" + with _interrupt_lock: + task = _active["task"] + return task if isinstance(task, asyncio.Task) and not task.done() else None + + async def host_request(data: dict[str, Any]) -> dict[str, Any]: """Send one typed request to the host and await its raw reply dict.""" if _loop is None: diff --git a/prime-agent-runtime/test/test_repl.py b/prime-agent-runtime/test/test_repl.py index 6a2b5daf5..27b3b45fd 100644 --- a/prime-agent-runtime/test/test_repl.py +++ b/prime-agent-runtime/test/test_repl.py @@ -756,6 +756,72 @@ class ReplTest(unittest.TestCase): inspected = self.repl.execute("bash-inspect", "handle.poll().output") self.assertIn("detached", one(inspected, "result")["text"]) + def test_result_read_after_notice_requests_one_withdrawal(self): + reads = ( + ("poll", "handle.poll().output"), + ("await", "(await handle).output"), + ) + for label, read_code in reads: + with self.subTest(label=label): + command = f"sleep 0.05; printf withdrawn-{label}" + started = self.repl.execute( + f"withdraw-{label}", + f"from rlm import bash\nhandle = bash({command!r})\nhandle.pid", + ) + pid = int(one(started, "result")["text"]) + notice = wait_for_host_request(self.repl, started) + self.assertEqual(notice["data"]["type"], "bash.completed") + reply_ok(self.repl, notice) + read = self.repl.execute(f"withdraw-{label}-read", read_code) + self.assertIn(f"withdrawn-{label}", one(read, "result")["text"]) + request = wait_for_host_request(self.repl, read) + self.assertEqual( + request["data"], {"type": "bash.consumed", "pid": pid, "command": command} + ) + # Old host: error reply is absorbed and the withdrawal never repeats. + self.repl.send( + {"type": "host_reply", "id": request["id"], "data": {"status": "error", "error": "unknown"}} + ) + if label == "poll": + again = self.repl.execute("withdraw-again", "handle.output()\nhandle.tail(1)") + self.assertIsNone(one(again, "host_request")) + + def test_detached_read_between_turns_keeps_bash_completion(self): + # A watcher reading the handle with no cell running reaches nobody: the + # notice is the only wake-up an idle session gets, so it must survive. + code = "\n".join( + [ + "from rlm import bash", + "import asyncio", + "handle = bash('sleep 0.2; printf detached-read')", + "async def watch():", + " await handle._wait()", + " globals()['seen'] = handle.output()", + "asyncio.create_task(watch())", + "handle.pid", + ] + ) + started = self.repl.execute("detached-read", code) + pid = int(one(started, "result")["text"]) + request = wait_for_host_request(self.repl, started) + self.assertEqual(request["data"]["type"], "bash.completed") + self.assertEqual(request["data"]["pid"], pid) + reply_ok(self.repl, request) + probe = self.repl.execute("detached-read-probe", "await asyncio.sleep(0.05)\nseen") + self.assertIsNone(one(probe, "host_request")) + self.assertEqual(one(probe, "result")["text"], "'detached-read'") + + def test_result_read_in_creating_cell_suppresses_bash_completion(self): + events = self.repl.execute( + "read-in-cell", + "from rlm import bash\nimport asyncio\nhandle = bash('printf read-in-cell')\n" + "await asyncio.wait_for(handle._wait(), 5)\nhandle.output()", + ) + self.assertEqual(one(events, "result")["text"], "'read-in-cell'") + probe = self.repl.execute("read-in-cell-probe", "await asyncio.sleep(0.05)") + self.assertIsNone(one(events, "host_request")) + self.assertIsNone(one(probe, "host_request")) + def test_wrapper_awaits_in_creating_cell_suppress_bash_completion(self): def task_group(label: str, await_expression: str) -> str: return "\n".join( @@ -840,7 +906,6 @@ class ReplTest(unittest.TestCase): "consumer = asyncio.create_task(consume())\n" "await asyncio.sleep(0.05)" ), - "later-cell": "iterator = asyncio.as_completed([handle])", } for label, snippet in snippets.items(): with self.subTest(label=label): @@ -849,11 +914,6 @@ class ReplTest(unittest.TestCase): f"as-completed-{label}", f"from rlm import bash\nimport asyncio\nhandle = bash({command!r})\n{snippet}", ) - if label == "later-cell": - events += self.repl.execute( - "as-completed-later-await", - "for completed in iterator:\n await completed", - ) events += self.repl.execute( f"as-completed-{label}-probe", "await asyncio.sleep(0.05)" ) -- knotmirror 0.0.1