rlm,ssh: unify handle and result text accessor and add accountable stdin and copy verification
50 unmodified lines51525354555556575859338 unmodified linesExpand all39839940040140240240340440540640740840740941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345419 unmodified lines4744754764774364784794804814823 unmodified lines48648748848944844945045149049149249349449549649715 unmodified lines51351451551647551751851952052148052252352452552648552752852953053149053253353453549453653753849753954054154254354450354554654750654854955055155251 unmodified lines6046056066075666086096105696116126136146159 unmodified lines62562662762858762963063163263363463563663759659759859963863964064164264364464525 unmodified lines6716726736746336756766776786799 unmodified lines68969069169265169369469569669775 unmodified lines77377477577673577777877978078120 unmodified lines80280380480576480680780880981079 unmodified lines890891892893852894895896897898289 unmodified linesExpand all11881189119011911150119211931194119511961155119711981199120012014 unmodified lines12061207120812091168121012111212121312147 unmodified lines122212231224122511841185118612261227122812291230123112322 unmodified lines12351236123712381197123912401241124212434 unmodified lines12481249125012511210125212531254125512561 unmodified line1258125912601261122012621263126412651266614 unmodified linesExpand all188118821883188418431885188618871888188957 unmodified lines19471948194919501909195119521953195419555 unmodified lines50 unmodified linesclass BashTest(unittest.IsolatedAsyncioTestCase): async def test_await_returns_result(self): result = await bash("echo hi") self.assertEqual(result.exit_code, 0) self.assertIn("hi", result.output) self.assertIn("hi", result.text()) self.assertGreaterEqual(result.duration, 0)
handle = bash("echo again") awaited = await handle338 unmodified linesExpand all
async def test_bash_stdin_str_input(self): res = await bash("cat", input="hello stdin string") self.assertEqual(res.exit_code, 0) self.assertEqual(res.output, "hello stdin string") self.assertEqual(res.text(), "hello stdin string") self.assertEqual(res.stdin_delivered, len("hello stdin string")) self.assertEqual(res.stdin_expected, len("hello stdin string"))
async def test_bash_stdin_bytes_input(self): res = await bash("cat", input=b"hello stdin bytes") self.assertEqual(res.exit_code, 0) self.assertEqual(res.output, "hello stdin bytes") self.assertEqual(res.text(), "hello stdin bytes") self.assertEqual(res.stdin_delivered, len(b"hello stdin bytes")) self.assertEqual(res.stdin_expected, len(b"hello stdin bytes"))
async def test_uniform_text_accessor_across_handle_stream_and_result(self): handle = bash("echo 'uniform accessor test'") self.assertIsInstance(handle.text(), str) self.assertIsInstance(handle.stdout.text(), str) result = await handle self.assertEqual(result.text().strip(), "uniform accessor test") self.assertEqual(str(handle).strip(), "uniform accessor test") self.assertEqual(str(result).strip(), "uniform accessor test")
# Ambiguity removed: .output must NOT exist on result or handle or stream with self.assertRaises(AttributeError): _ = result.output with self.assertRaises(AttributeError): _ = handle.output with self.assertRaises(AttributeError): _ = handle.stdout.output
async def test_stdin_delivery_accountability_short_fails_loudly(self): payload = "0123456789abcdefghij" # head -c 5 consumes only 5 bytes of the 20-byte payload res = await bash("head -c 5", input=payload) self.assertNotEqual(res.exit_code, 0) self.assertEqual(res.stdin_delivered, 5) self.assertEqual(res.stdin_expected, 20) self.assertIn("stdin delivery incomplete: delivered 5 of 20 bytes", res.stderr) self.assertIn("stdin delivery incomplete: delivered 5 of 20 bytes", res.text())
async def test_stdin_delivery_accountability_zero_consumption_fails_loudly(self): payload = "unread-payload-data" # Command exits 0 without consuming any stdin res = await bash("python3 -c 'import sys; sys.exit(0)'", input=payload) self.assertNotEqual(res.exit_code, 0) self.assertEqual(res.stdin_delivered, 0) self.assertEqual(res.stdin_expected, len(payload)) expected_msg = f"stdin delivery incomplete: delivered 0 of {len(payload)} bytes" self.assertIn(expected_msg, res.stderr) self.assertIn(expected_msg, res.text())
def test_bash_invalid_input_type_raises(self): with self.assertRaises(TypeError): bash("cat", input=12345)
19 unmodified lines self.assertIsNone(bash_module._check_command_guards(command))
async def test_bash_result_repr_is_bounded(self): res = await bash("echo short") self.assertIn(r"output='short\n'", repr(res)) self.assertIn(r"text='short\n'", repr(res)) self.assertIn("<BashResult", repr(res))
async def test_absurdly_long_output_spills_to_file(self): # Override threshold to test spilling with moderate output3 unmodified lines self.assertTrue(os.path.exists(res.output_file)) self.assertEqual(res.file_path, res.output_file) self.assertIsInstance(res.path, Path) self.assertEqual(str(res.path), res.output_file) self.assertTrue(res.output.startswith("1\n")) self.assertIn("200\n", res.output) self.assertIn("output truncated", res.output) self.assertIn("Full output saved to", res.output) self.assertTrue(res.text().startswith("1\n")) self.assertIn("200\n", res.text()) self.assertIn("output truncated", res.text()) self.assertIn("Full output saved to", res.text()) # Both output and stdout are truncated consistently self.assertIn("output truncated", res.stdout) # Full output available via full_output() full = res.full_output()15 unmodified lines async def test_bash_result_stdout_stderr_separation(self): res = await bash("echo foo; echo bar >&2") self.assertEqual(res.stdout, "foo\n") self.assertEqual(res.stderr, "bar\n") self.assertEqual(res.output, "foo\nbar\n") self.assertEqual(res.text(), "foo\nbar\n")
async def test_bash_input_from_handle_combined(self): src = bash("echo line1; echo line2 >&2") dst = await bash("cat", input=src) self.assertEqual(dst.output, "line1\nline2\n") self.assertEqual(dst.text(), "line1\nline2\n")
async def test_bash_input_from_handle_stdout(self): src = bash("echo only-stdout; echo skip-stderr >&2") dst = await bash("cat", input=src.stdout) self.assertEqual(dst.output, "only-stdout\n") self.assertEqual(dst.text(), "only-stdout\n")
async def test_bash_input_from_handle_stderr(self): src = bash("echo skip-stdout; echo only-stderr >&2") dst = await bash("cat", input=src.stderr) self.assertEqual(dst.output, "only-stderr\n") self.assertEqual(dst.text(), "only-stderr\n")
async def test_bash_stream_pipe_method(self): res = await bash("echo lowercase").stdout.pipe("tr a-z A-Z") self.assertEqual(res.output, "LOWERCASE\n") self.assertEqual(res.text(), "LOWERCASE\n")
res_handle = await bash("echo combined-pipe").pipe("tr a-z A-Z") self.assertEqual(res_handle.output, "COMBINED-PIPE\n") self.assertEqual(res_handle.text(), "COMBINED-PIPE\n")
async def test_bash_input_from_already_completed_handle(self): src = bash("echo line1; echo line2 >&2") await src dst1 = await bash("cat", input=src) self.assertEqual(dst1.output, "line1\nline2\n") self.assertEqual(dst1.text(), "line1\nline2\n")
dst2 = await bash("cat", input=src.stdout) self.assertEqual(dst2.output, "line1\n") self.assertEqual(dst2.text(), "line1\n")
def test_git_subcommand_mentions_native_function(self): with self.assertRaises(ValueError) as ctx: bash("git grep 'pattern'")51 unmodified lines if os.path.exists("/bin/dash"): with mock.patch.dict(os.environ, {"PRIME_AGENT_BASH_SHELL": "/bin/dash"}): result = await bash("echo ok") self.assertEqual(result.exit_code, 0) self.assertIn("ok", result.output) self.assertIn("ok", result.text()) result = await bash("echo ok-default") self.assertEqual(result.exit_code, 0) self.assertIn("ok-default", result.output) self.assertIn("ok-default", result.text())
async def test_backgrounded_tail_and_kill(self): handle = bash("echo start; sleep 30") self.assertIsNone(handle.poll())9 unmodified lines
async def test_kill_escalates_to_sigkill(self): handle = bash("trap '' TERM; echo up; sleep 30") for _ in range(100): if "up" in handle.output(): if "up" in handle.text(): break await asyncio.sleep(0.05) handle.kill(grace=0.2) result = await asyncio.wait_for(handle, timeout=10) self.assertEqual(result.exit_code, -9)
async def test_buffer_cap_keeps_head_and_tail(self): result = await bash("seq 1 400000") self.assertLessEqual(len(result.output), 2 * 1024 * 1024 + 256) self.assertTrue(result.output.startswith("1\n")) self.assertIn("400000", result.output) self.assertIn("bytes dropped", result.output) self.assertLessEqual(len(result.text()), 2 * 1024 * 1024 + 256) self.assertTrue(result.text().startswith("1\n")) self.assertIn("400000", result.text()) self.assertIn("bytes dropped", result.text())
def test_child_env_is_non_interactive(self): """Agent shells have no usable stdin: interactive prompts (git commit opening $EDITOR, credential asks, pagers) can only hang. _child_env()25 unmodified lines 'echo "$GIT_EDITOR|$GIT_SEQUENCE_EDITOR|$GIT_TERMINAL_PROMPTS|$GIT_ASKPASS|$SSH_ASKPASS_REQUIRE"' ) result = await handle self.assertEqual(result.exit_code, 0) self.assertIn("true|true|0|true|never", result.output) self.assertIn("true|true|0|true|never", result.text())
async def test_env_prefix_and_journal(self): with tempfile.TemporaryDirectory() as tmp: journal = os.path.join(tmp, "journal.jsonl")9 unmodified lines result = await handle # The inactive record lands slightly after finalize, once the group exits. records = await _poll_journal(journal, count=2) self.assertEqual(result.exit_code, 0) lines = result.output.splitlines() lines = result.text().splitlines() self.assertEqual(lines[0], "prefixed") self.assertIn("1 dumb", lines[1])
self.assertEqual([r["active"] for r in records], [True, False])75 unmodified lines ): handle = bash("echo fg; sleep 30 &") result = await asyncio.wait_for(handle, timeout=5) self.assertEqual(result.exit_code, 0) self.assertIn("fg", result.output) self.assertIn("fg", result.text()) # The shell stays alive as group leader, anchoring its background job. os.killpg(handle.pid, 0) records = await _poll_journal(journal, count=1) self.assertTrue(records[-1]["active"])20 unmodified lines ): handle = bash("sh -c 'trap \"\" TERM; echo ready; sleep 30' &") await asyncio.wait_for(handle, timeout=5) for _ in range(100): if "ready" in handle.output(): if "ready" in handle.text(): break await asyncio.sleep(0.05) handle.kill(signal.SIGTERM) records = await _poll_journal(journal, count=2, timeout=10)79 unmodified lines handle = bash("echo hi; sleep 30 & true") self.addCleanup(handle.kill, signal.SIGKILL) result = await asyncio.wait_for(handle, timeout=5) self.assertEqual(result.exit_code, 0) self.assertIn("hi", result.output) self.assertIn("hi", result.text())
async def test_awaits_do_not_hold_executor_threads(self): loop = asyncio.get_running_loop() executor = ThreadPoolExecutor(max_workers=1)289 unmodified linesExpand all
with mock.patch.object(bash_module._BoundedBuffer, "write", delayed_write): result = await asyncio.wait_for(bash("printf delayed-output-complete"), timeout=5) self.assertEqual(result.exit_code, 0) self.assertEqual(result.output, "delayed-output-complete") self.assertEqual(result.text(), "delayed-output-complete")
async def test_completion_marker_split_across_reads_is_removed(self): with mock.patch.object(bash_module, "_READ_CHUNK", 7): result = await asyncio.wait_for(bash("printf exact-pre-fence-output"), timeout=5) self.assertEqual(result.output, "exact-pre-fence-output") self.assertEqual(result.text(), "exact-pre-fence-output")
async def test_slow_pump_does_not_lose_foreground_output(self): # Finalization waits for the pump to parse the ordered sentinel. original_pump = bash_module.BashHandle._pump4 unmodified lines
with mock.patch.object(bash_module.BashHandle, "_pump", slow_pump): result = await asyncio.wait_for(bash("printf slow-pump-x"), timeout=5) self.assertEqual(result.exit_code, 0) self.assertIn("slow-pump-x", result.output) self.assertIn("slow-pump-x", result.text())
async def test_sentinel_like_output_and_echoed_wrapper_do_not_truncate(self): token = "0123456789abcdef" * 4 raw_lookalike = "\x1eprime-agent-complete:not-this-invocation\x1f"7 unmodified lines result = await asyncio.wait_for(bash(command), timeout=5) actual_marker = ( bash_module._COMPLETION_PREFIX + token.encode() + bash_module._COMPLETION_SUFFIX ) self.assertIn(raw_lookalike, result.output) self.assertIn("after-sentinel-lookalike", result.output) self.assertNotIn(actual_marker.decode(), result.output) self.assertIn(raw_lookalike, result.text()) self.assertIn("after-sentinel-lookalike", result.text()) self.assertNotIn(actual_marker.decode(), result.text())
async def test_user_alias_cannot_replace_completion_emitter(self): if os.path.basename(bash_module._shell()) != "bash": self.skipTest("bash alias expansion semantics")2 unmodified lines ) try: result = await asyncio.wait_for(handle, timeout=5) self.assertEqual(result.exit_code, 0) self.assertNotIn("alias-expanded", result.output) self.assertNotIn("alias-expanded", result.text()) finally: handle.kill(signal.SIGKILL) await _poll_group_dead(handle.pid)
4 unmodified lines handle = bash("command() { printf function-expanded; }; sleep 30 &") try: result = await asyncio.wait_for(handle, timeout=5) self.assertEqual(result.exit_code, 0) self.assertNotIn("function-expanded", result.output) self.assertNotIn("function-expanded", result.text()) finally: handle.kill(signal.SIGKILL) await _poll_group_dead(handle.pid)
1 unmodified line result = await asyncio.wait_for( bash("printf output-before-shell-kill; kill -KILL $$"), timeout=5 ) self.assertEqual(result.exit_code, -signal.SIGKILL) self.assertIn("output-before-shell-kill", result.output) self.assertIn("output-before-shell-kill", result.text())
async def test_relative_bash_shell_override_rejected(self): with mock.patch.dict(os.environ, {"PRIME_AGENT_BASH_SHELL": "bash"}): with self.assertRaises(ValueError):614 unmodified linesExpand all os.environ.pop("PRIME_AGENT_INTERNAL_ORPHAN_PROCESS_JOURNAL", None) with mock.patch.object(bash_module, "_process_start_id", return_value=None): result = await bash("echo ok") self.assertEqual(result.exit_code, 0) self.assertIn("ok", result.output) self.assertIn("ok", result.text())
async def _poll_group_dead(pgid: int, timeout: float = 5.0) -> None: deadline = asyncio.get_running_loop().time() + timeout57 unmodified lines async def test_timeout_does_not_touch_a_finished_command(self): result = await bash("echo hi", timeout=5.0) self.assertFalse(result.timed_out) self.assertEqual(result.exit_code, 0) self.assertIn("hi", result.output) self.assertIn("hi", result.text())
def test_timeout_validation(self): for bad in (0, -1, float("inf"), float("nan")): with self.assertRaises(ValueError):5 unmodified lines101 unmodified linesExpand all1021031041051061071061071081091101111121131141151161161171181191201211221231231241251261271281291301311311321331341357 unmodified lines101 unmodified linesExpand all
h = self.handle.call(print_and_return) res = await h self.assertEqual(res, 123) self.assertIn("step 1: started", h.output) self.assertIn("step 2: finished", h.output) self.assertIn("step 1: started", h.text()) self.assertIn("step 2: finished", h.text()) self.assertIn("step 1: started", h.text()) self.assertIn("step 2", h.tail(1))
async def test_first_class_remote_bash_handle(self): h = self.handle.bash("echo 'first class remote bash'") self.assertIsInstance(h, RemoteHandle) res = await h self.assertEqual(res.exit_code, 0) self.assertIn("first class remote bash", res.output) self.assertIn("first class remote bash", res.text())
async def test_first_class_remote_bash_handle_with_input(self): h = self.handle.bash("cat", input="remote stdin string") self.assertIsInstance(h, RemoteHandle) res = await h self.assertEqual(res.exit_code, 0) self.assertEqual(res.output, "remote stdin string") self.assertEqual(res.text(), "remote stdin string")
async def test_first_class_remote_bash_handle_with_bash_handle_input(self): from rlm.bash import bash as local_bash local_h = await local_bash("echo 'piped from local handle'") h = self.handle.bash("cat", input=local_h) res = await h self.assertEqual(res.exit_code, 0) self.assertIn("piped from local handle", res.output) self.assertIn("piped from local handle", res.text())
async def test_async_context_manager(self): async with self.handle as rem: await rem.imports("sys")7 unmodified lines706 unmodified linesExpand all70770870971071171171271371471537 unmodified lines75375475575675775775875976076176276376276376476576676715 unmodified lines7837847857867877877887897907914 unmodified lines79679779879980080080180280380410 unmodified lines8158168178188198198208218228237 unmodified lines831832833834835835836837838839840841842843840841842843844845845846847848849850851852852853854855856857858859860860861862863863864865866867868869870871871872873874875876877878879880880881882883884124 unmodified linesExpand all10091010101110121013101310141015101610171634 unmodified linesExpand all706 unmodified linesExpand all self.assertEqual(one(follow, "done")["status"], "ok")
def test_bash_integration(self): events = self.repl.execute( "sh1", "from rlm import bash\nresult = await bash('echo repl-bash')\nresult.output.strip()" "sh1", "from rlm import bash\nresult = await bash('echo repl-bash')\nresult.text().strip()" ) self.assertEqual(one(events, "result")["text"], "'repl-bash'")
events = self.repl.execute(37 unmodified lines "exitCode": 0, }, ) reply_ok(self.repl, request) inspected = self.repl.execute("bash-inspect", "handle.poll().output") inspected = self.repl.execute("bash-inspect", "handle.poll().text()") 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"), ("poll", "handle.poll().text()"), ("await", "(await handle).text()"), ) for label, read_code in reads: with self.subTest(label=label): command = f"sleep 0.05; printf withdrawn-{label}"15 unmodified lines 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)") again = self.repl.execute("withdraw-again", "handle.text()\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: the4 unmodified lines "import asyncio", "handle = bash('sleep 0.2; printf detached-read')", "async def watch():", " await handle._wait()", " globals()['seen'] = handle.output()", " globals()['seen'] = handle.text()", "asyncio.create_task(watch())", "handle.pid", ] )10 unmodified lines 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()", "await asyncio.wait_for(handle._wait(), 5)\nhandle.text()", ) 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"))7 unmodified lines "async def consume():", f" return {await_expression}", "async with asyncio.TaskGroup() as group:", " task = group.create_task(consume())", "task.result().output", "task.result().text()", ] )
snippets = { "gather": "(await asyncio.gather(bash('printf gather')))[0].output", "wait-for": "(await asyncio.wait_for(bash('printf wait-for'), 1)).output", "shield": "(await asyncio.shield(bash('printf shield'))).output", "nested": "(await asyncio.shield(asyncio.gather(bash('printf nested'))))[0].output", "gather": "(await asyncio.gather(bash('printf gather')))[0].text()", "wait-for": "(await asyncio.wait_for(bash('printf wait-for'), 1)).text()", "shield": "(await asyncio.shield(bash('printf shield'))).text()", "nested": "(await asyncio.shield(asyncio.gather(bash('printf nested'))))[0].text()", "as-completed": ( "[(await completed).output for completed in " "[(await completed).text() for completed in " "asyncio.as_completed([bash('printf as-completed')])]" ), "as-completed-delayed-consumer": ( "handle = bash('sleep 0.15; printf as-completed-delayed-consumer')\n" "completed = next(iter(asyncio.as_completed([handle])))\n" "await asyncio.sleep(0)\n" "(await completed).output" "(await completed).text()" ), "as-completed-finished-handle": ( "handle = bash('printf as-completed-finished-handle')\n" "task = asyncio.ensure_future(handle)\n" "await asyncio.to_thread(handle._done.wait)\n" "await asyncio.sleep(0.01)\n" "assert task.done()\n" "(await handle).output" "(await handle).text()" ), "as-completed-nested": ( "[(await completed)[0].output for completed in " "[(await completed)[0].text() for completed in " "asyncio.as_completed([asyncio.shield(asyncio.gather(bash('printf as-completed-nested')))])]" ), "wait": "\n".join( [ "handle = bash('printf wait')", "task = asyncio.ensure_future(handle)", "await asyncio.wait({task})", "task.result().output", "task.result().text()", ] ), "task-group": task_group("task-group", "await handle"), "task-group-gather": task_group("task-group-gather", "(await asyncio.gather(handle))[0]"), "task-group-shield": task_group("task-group-shield", "await asyncio.shield(handle)"), } if sys.version_info >= (3, 13): snippets["as-completed-async"] = ( "[completed.result().output async for completed in " "[completed.result().text() async for completed in " "asyncio.as_completed([bash('printf as-completed-async')])]" ) for label, snippet in snippets.items(): with self.subTest(label=label):124 unmodified linesExpand all if any(event.get("data", {}).get(mime, {}).get("active") is False for event in events): break events.append(self.repl.read_event()) self.assertEqual(len(requests), 1 if released else 0) checked = self.repl.execute("cancel-result", "handle.poll().output") checked = self.repl.execute("cancel-result", "handle.poll().text()") self.assertEqual(one(checked, "result")["text"], "'finished'" if released else "''")
def test_interrupt_during_bash_notifier_construction_rolls_back_activity(self): code = "\n".join(1634 unmodified linesExpand all123- Unified `.text()` as the single text accessor across `BashHandle`, `BashStream`, and `BashResult`, removing ambiguous `.output` attributes and methods.- Added accountable stdin delivery reporting `stdin_delivered` and failing with a non-zero exit code on short or unconsumed writes.- Added verified SSH upload and download verification with file size and SHA-256 integrity checks.43 unmodified lines444546474848495051521 unmodified line54555657585859606061626364134 unmodified linesExpand all43 unmodified linesRun a project's normal commands through its own environment with `bash()`:
```pythonresult = await bash("npm run check")print(result.output)print(result.text())```
For a long command, keep the live handle and let the turn end instead of blocking:
1 unmodified linechecks = bash("npm test")checks.pid```
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.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()`, `text()`, or `tail()` and continue the task. `await bash(...)` stays synchronous from the agent's perspective and does not send a second notice.
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.Reading the finished result withdraws the notice. Any read from a live cell counts -- `await handle`, `handle.poll()`, `handle.text()`, `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.
### 2. Subagents are native RLM calls134 unmodified linesExpand all227 unmodified linesExpand all22822923023123223223323423523664 unmodified lines227 unmodified linesExpand all const pollingLoop = `sb_handle = bash("pnpm --dir web storybook --ci --port 6006")for i in range(15): await asyncio.sleep(2) poll = await bash("curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:6006/") if poll.output == "200": if poll.text() == "200": break`; expect(previewPythonCode(pollingLoop)).toEqual({ language: "bash", text: "pnpm storybook --ci --port 6006 (web)",64 unmodified lines