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)" )