from __future__ import annotations import asyncio import importlib import unittest from pathlib import Path from unittest.mock import AsyncMock, patch rlm_module = importlib.import_module("rlm") class TestSubagentRegistry(unittest.TestCase): def test_lists_parent_scoped_subagents_from_host(self) -> None: payload = { "subagents": [ { "rlm_child_id": "sub-a1b2c3d4", "active_session_id": "child-active-1", "session_id": "child-session-1", "session_name": "worker-1", "session_dir": "/tmp/parent/sub-a1b2c3d4", "status": "completed", }, { "rlm_child_id": "sub-e5f6a7b8", "active_session_id": None, "session_id": "child-session-2", "session_name": "worker-2", "session_dir": "/tmp/parent/sub-e5f6a7b8", "status": "running", }, ] } host_request = AsyncMock(return_value=payload) with patch.object(rlm_module, "host_request", host_request): subagents = asyncio.run(rlm_module.rlm.list_subagents()) self.assertEqual(len(subagents), 2) first, second = subagents self.assertEqual(first.rlm_child_id, "sub-a1b2c3d4") self.assertEqual(first.active_session_id, "child-active-1") self.assertEqual(first.session_id, "child-session-1") self.assertEqual(first.session_name, "worker-1") self.assertEqual(first.session_dir, Path("/tmp/parent/sub-a1b2c3d4")) self.assertEqual(first.status, "completed") self.assertEqual(second.rlm_child_id, "sub-e5f6a7b8") self.assertIsNone(second.active_session_id) self.assertEqual(second.session_id, "child-session-2") self.assertEqual(second.session_name, "worker-2") self.assertEqual(second.session_dir, Path("/tmp/parent/sub-e5f6a7b8")) self.assertEqual(second.status, "running") host_request.assert_awaited_once_with("rlm.list_subagents") def test_lists_failed_subagents_from_host(self) -> None: payload = { "subagents": [ { "rlm_child_id": "sub-deadbeef", "active_session_id": None, "session_id": "child-session-failed", "session_name": "crashed-worker", "session_dir": "/tmp/parent/sub-deadbeef", "status": "error", } ] } host_request = AsyncMock(return_value=payload) with patch.object(rlm_module, "host_request", host_request): (failed,) = asyncio.run(rlm_module.list_subagents()) self.assertEqual(failed.status, "error") def test_forwards_model_and_thinking_to_host(self) -> None: host_request = AsyncMock( return_value={ "rlm_child_id": "sub-a1b2c3d4", "name": "api-reviewer", "session_dir": "/tmp/parent/sub-a1b2c3d4", "model": "anthropic/claude-opus-5", } ) with patch.object(rlm_module, "host_request", host_request): result = asyncio.run( rlm_module.rlm.spawn( "check the API", name="api-reviewer", model="anthropic/claude-opus-5", thinking="high", host="remote-box", ) ) expected_prompt = rlm_module.compose_child_brief("check the API", name="api-reviewer") host_request.assert_awaited_once_with( "rlm.run", { "prompt": expected_prompt, "kwargs": { "name": "api-reviewer", "model": "anthropic/claude-opus-5", "thinking": "high", "host": "remote-box", }, }, ) self.assertEqual(result.rlm_child_id, "sub-a1b2c3d4") self.assertEqual(result.name, "api-reviewer") self.assertEqual(result.session_dir, Path("/tmp/parent/sub-a1b2c3d4")) self.assertEqual(result.model, "anthropic/claude-opus-5") def test_omits_unset_kwargs_from_host_payload(self) -> None: host_request = AsyncMock( return_value={ "rlm_child_id": "sub-a1b2c3d4", "name": "api-reviewer", "session_dir": "/tmp/parent/sub-a1b2c3d4", "model": "openai-codex/gpt-5.6-terra", } ) with patch.object(rlm_module, "host_request", host_request): result = asyncio.run( rlm_module.spawn("check the API", name="api-reviewer", model="openai-codex/gpt-5.6-terra") ) expected_prompt = rlm_module.compose_child_brief("check the API", name="api-reviewer") host_request.assert_awaited_once_with( "rlm.run", {"prompt": expected_prompt, "kwargs": {"name": "api-reviewer", "model": "openai-codex/gpt-5.6-terra"}}, ) self.assertEqual(result.model, "openai-codex/gpt-5.6-terra") def test_requires_a_model(self) -> None: host_request = AsyncMock() with patch.object(rlm_module, "host_request", host_request): with self.assertRaisesRegex(TypeError, r"missing 1 required keyword-only argument: 'model'"): asyncio.run(rlm_module.rlm.spawn("check the API", name="api-reviewer")) host_request.assert_not_awaited() def test_spawn_with_model_and_thinking(self) -> None: payload = { "rlm_child_id": "sub-a1b2c3d4", "name": "api-reviewer", "session_dir": "/tmp/parent/sub-a1b2c3d4", "model": "deepseek/deepseek-v4-flash", } host_request = AsyncMock(return_value=payload) spawns = (rlm_module.spawn, rlm_module.rlm.spawn) with patch.object(rlm_module, "host_request", host_request): for spawn in spawns: result = asyncio.run( spawn("check the API", name="api-reviewer", model="deepseek/deepseek-v4-flash", thinking="high") ) self.assertEqual(result.model, "deepseek/deepseek-v4-flash") def test_rejects_invalid_spawn_handle(self) -> None: base = { "rlm_child_id": "sub-a1b2c3d4", "name": "api-reviewer", "session_dir": "/tmp/parent/sub-a1b2c3d4", "model": "openai-codex/gpt-5.6-terra", } for missing in ("rlm_child_id", "name", "session_dir", "model"): payload = {k: v for k, v in base.items() if k != missing} host_request = AsyncMock(return_value=payload) with patch.object(rlm_module, "host_request", host_request): with self.assertRaisesRegex(RuntimeError, "invalid spawn handle"): asyncio.run(rlm_module.spawn("check the API", name="api-reviewer", model="openai-codex/gpt-5.6-terra")) def test_requires_an_explicit_child_name(self) -> None: host_request = AsyncMock() with patch.object(rlm_module, "host_request", host_request): with self.assertRaisesRegex(TypeError, r"missing 1 required keyword-only argument: 'name'"): asyncio.run(rlm_module.rlm.spawn("check the API", model="coder")) host_request.assert_not_awaited() def test_rejects_calling_rlm_directly_with_spawn_guidance(self) -> None: for target in (rlm_module.rlm, rlm_module): with self.assertRaisesRegex(TypeError, r"not callable; spawn a child with: handle = await rlm\.spawn\("): target("check the API") with self.assertRaisesRegex(AttributeError, r"rlm\.run was renamed; spawn a child with: handle = await rlm\.spawn\("): getattr(rlm_module.rlm, "run") def test_reports_models_through_host(self) -> None: payload = { "scouts": ["google-antigravity/gemini-3.8-flash-low", "openai-codex/gpt-5.6-luna"], "coders": ["google-antigravity/gemini-3.8-flash-high", "openai-codex/gpt-5.6-sol"], } host_request = AsyncMock(return_value=payload) with patch.object(rlm_module, "host_request", host_request): models = asyncio.run(rlm_module.rlm.models()) self.assertEqual(models["scouts"], payload["scouts"]) self.assertEqual(models.scouts, payload["scouts"]) self.assertEqual(models.coders[0], "google-antigravity/gemini-3.8-flash-high") self.assertIn("scouts", dir(models)) self.assertIn("coders", dir(models)) models.reviewers = ["test/reviewer"] self.assertEqual(models["reviewers"], ["test/reviewer"]) del models.reviewers self.assertNotIn("reviewers", models) host_request.assert_awaited_once_with("rlm.models") def test_deletes_subagent_by_name_through_host(self) -> None: deleted_payload = { "rlm_child_id": "sub-a1b2c3d4", "active_session_id": "active-child", "session_id": "session-child", "session_name": "api-reviewer", "session_dir": "/tmp/parent/sub-a1b2c3d4", "status": "completed", } host_request = AsyncMock(return_value={"subagent": deleted_payload}) with patch.object(rlm_module, "host_request", host_request): deleted = asyncio.run(rlm_module.rlm.delete_subagent(" api-reviewer ")) self.assertEqual(deleted.rlm_child_id, "sub-a1b2c3d4") self.assertEqual(deleted.session_name, "api-reviewer") host_request.assert_awaited_once_with( "rlm.delete_subagent", {"target": "api-reviewer"}, ) def test_deletes_subagent_object_by_child_id(self) -> None: subagent = rlm_module.RLMSubagent( rlm_child_id="sub-a1b2c3d4", active_session_id=None, session_id="session-child", session_name="api-reviewer", session_dir=Path("/tmp/parent/sub-a1b2c3d4"), status="running", ) host_request = AsyncMock( return_value={ "subagent": { "rlm_child_id": subagent.rlm_child_id, "active_session_id": subagent.active_session_id, "session_id": subagent.session_id, "session_name": subagent.session_name, "session_dir": str(subagent.session_dir), "status": subagent.status, } } ) with patch.object(rlm_module, "host_request", host_request): asyncio.run(rlm_module.delete_subagent(subagent)) host_request.assert_awaited_once_with( "rlm.delete_subagent", {"target": "sub-a1b2c3d4"}, ) def test_deletes_subagent_by_spawn_handle(self) -> None: handle = rlm_module.RLMSpawnHandle( rlm_child_id="sub-a1b2c3d4", name="api-reviewer", session_dir=Path("/tmp/parent/sub-a1b2c3d4"), model="deepseek/deepseek-v4-flash", ) deleted_payload = { "rlm_child_id": handle.rlm_child_id, "active_session_id": None, "session_id": None, "session_name": handle.name, "session_dir": str(handle.session_dir), "status": "completed", } host_request = AsyncMock(return_value={"subagent": deleted_payload}) with patch.object(rlm_module, "host_request", host_request): deleted = asyncio.run(rlm_module.rlm.delete_subagent(handle)) self.assertEqual(deleted.rlm_child_id, handle.rlm_child_id) self.assertEqual(deleted.session_name, handle.name) host_request.assert_awaited_once_with( "rlm.delete_subagent", {"target": "sub-a1b2c3d4"}, ) def test_rejects_invalid_delete_response_and_target(self) -> None: host_request = AsyncMock(return_value={"subagent": {"status": "completed"}}) with patch.object(rlm_module, "host_request", host_request): with self.assertRaisesRegex(RuntimeError, "rlm.delete_subagent entry is missing rlm_child_id"): asyncio.run(rlm_module.delete_subagent("api-reviewer")) with self.assertRaisesRegex(ValueError, "target must not be empty"): asyncio.run(rlm_module.delete_subagent(" ")) with self.assertRaisesRegex(TypeError, "target must be RLMSpawnHandle, RLMSubagent, or str"): asyncio.run(rlm_module.delete_subagent(123)) def test_rejects_invalid_registry_payload(self) -> None: host_request = AsyncMock(return_value={"subagents": [{"status": "completed"}]}) with patch.object(rlm_module, "host_request", host_request): with self.assertRaisesRegex(RuntimeError, "missing rlm_child_id"): asyncio.run(rlm_module.list_subagents()) def test_requires_a_default_session_name(self) -> None: host_request = AsyncMock( return_value={ "subagents": [ { "rlm_child_id": "sub-a1b2c3d4", "active_session_id": None, "session_id": "session-child", "session_dir": "/tmp/parent/sub-a1b2c3d4", "status": "running", } ] } ) with patch.object(rlm_module, "host_request", host_request): with self.assertRaisesRegex(RuntimeError, "missing session_name"): asyncio.run(rlm_module.list_subagents()) class RlmProgressNoteTest(unittest.TestCase): def test_progress_note_round_trips_an_accepted_note(self) -> None: host_request = AsyncMock(return_value={"accepted": True}) with patch.object(rlm_module, "host_request", host_request): result = asyncio.run(rlm_module.rlm.progress_note("building tests")) self.assertTrue(result.accepted) self.assertIsNone(result.retry_after_ms) host_request.assert_awaited_once_with("rlm.progress.note", {"message": "building tests"}) def test_progress_note_returns_the_throttle_hint(self) -> None: host_request = AsyncMock(return_value={"accepted": False, "retry_after_ms": 7000}) with patch.object(rlm_module, "host_request", host_request): result = asyncio.run(rlm_module.rlm.progress_note("second note")) self.assertFalse(result.accepted) self.assertEqual(result.retry_after_ms, 7000) def test_progress_note_validates_the_message(self) -> None: with self.assertRaisesRegex(TypeError, "message must be str"): asyncio.run(rlm_module.rlm.progress_note(42)) with self.assertRaisesRegex(ValueError, "empty"): asyncio.run(rlm_module.rlm.progress_note(" ")) with self.assertRaisesRegex(ValueError, "at most 512"): asyncio.run(rlm_module.rlm.progress_note("x" * 513)) # At the bound the message is accepted. host_request = AsyncMock(return_value={"accepted": True}) with patch.object(rlm_module, "host_request", host_request): result = asyncio.run(rlm_module.rlm.progress_note("x" * 512)) self.assertTrue(result.accepted) def test_progress_note_measures_utf16_length_at_the_boundary(self) -> None: # 512 astral characters are 1024 UTF-16 code units under the host's # message.length measure, so they must fail validation before the call. with self.assertRaisesRegex(ValueError, "at most 512"): asyncio.run(rlm_module.rlm.progress_note("🎉" * 512)) # A BMP character plus one astral character crosses the same bound. with self.assertRaisesRegex(ValueError, "at most 512"): asyncio.run(rlm_module.rlm.progress_note("a" * 511 + "🎉")) # 256 astral characters are exactly 512 UTF-16 code units: accepted. host_request = AsyncMock(return_value={"accepted": True}) with patch.object(rlm_module, "host_request", host_request): result = asyncio.run(rlm_module.rlm.progress_note("🎉" * 256)) self.assertTrue(result.accepted) host_request.assert_awaited_once_with("rlm.progress.note", {"message": "🎉" * 256}) # Lone surrogates are valid Python strings that json.dumps escapes and # the host accepts, so each must count as one UTF-16 code unit, not raise. host_request = AsyncMock(return_value={"accepted": True}) with patch.object(rlm_module, "host_request", host_request): result = asyncio.run(rlm_module.rlm.progress_note("\ud800")) self.assertTrue(result.accepted) host_request.assert_awaited_once_with("rlm.progress.note", {"message": "\ud800"}) with self.assertRaisesRegex(ValueError, "at most 512"): asyncio.run(rlm_module.rlm.progress_note("\ud800" * 513)) def test_progress_note_rejects_invalid_payloads(self) -> None: host_request = AsyncMock(return_value={"retry_after_ms": 1000}) with patch.object(rlm_module, "host_request", host_request): with self.assertRaisesRegex(RuntimeError, "accepted"): asyncio.run(rlm_module.rlm.progress_note("note")) host_request = AsyncMock(return_value={"accepted": True, "retry_after_ms": "soon"}) with patch.object(rlm_module, "host_request", host_request): with self.assertRaisesRegex(RuntimeError, "retry_after_ms"): asyncio.run(rlm_module.rlm.progress_note("note")) class RlmSubagentExtrasTest(unittest.TestCase): def test_lists_live_state_extras_from_host(self) -> None: host_request = AsyncMock( return_value={ "subagents": [ { "rlm_child_id": "sub-a1b2c3d4", "active_session_id": None, "session_id": "session-child", "session_name": "worker-a", "session_dir": "/tmp/parent/sub-a1b2c3d4", "status": "running", "activity": {"kind": "executing", "tool_name": "ipython"}, "tool_use_count": 3, "duration_ms": 4321, "answer_preview": "task finished cleanly", "replied_since_task": False, "progress_note": "halfway done", "label": "slow task", "last_activity_at": 1757700000000, "activity_stale_ms": None, } ] } ) with patch.object(rlm_module, "host_request", host_request): subagents = asyncio.run(rlm_module.list_subagents()) self.assertEqual(len(subagents), 1) entry = subagents[0] self.assertEqual(entry.activity.kind, "executing") self.assertEqual(entry.activity.tool_name, "ipython") self.assertEqual(entry.tool_use_count, 3) self.assertEqual(entry.duration_ms, 4321) self.assertEqual(entry.answer_preview, "task finished cleanly") self.assertFalse(entry.replied_since_task) self.assertEqual(entry.progress_note, "halfway done") self.assertEqual(entry.label, "slow task") self.assertEqual(entry.last_activity_at, 1757700000000) self.assertIsNone(entry.activity_stale_ms) def test_accepts_entries_missing_optional_extras(self) -> None: host_request = AsyncMock( return_value={ "subagents": [ { "rlm_child_id": "sub-old", "active_session_id": None, "session_id": None, "session_name": "worker-old", "session_dir": "/tmp/parent/sub-old", "status": "completed", } ] } ) with patch.object(rlm_module, "host_request", host_request): subagents = asyncio.run(rlm_module.list_subagents()) entry = subagents[0] self.assertIsNone(entry.activity) self.assertIsNone(entry.tool_use_count) self.assertIsNone(entry.duration_ms) self.assertIsNone(entry.answer_preview) self.assertIsNone(entry.replied_since_task) self.assertIsNone(entry.progress_note) self.assertIsNone(entry.label) self.assertIsNone(entry.last_activity_at) self.assertIsNone(entry.activity_stale_ms) def test_rejects_invalid_extras(self) -> None: base_entry = { "rlm_child_id": "sub-x", "active_session_id": None, "session_id": None, "session_name": "worker-x", "session_dir": "/tmp/parent/sub-x", "status": "running", } for extras, pattern in ( ({"activity": {"kind": "nonsense"}}, "invalid activity kind"), ({"activity": {"kind": "executing", "tool_name": 7}}, "invalid activity tool_name"), ({"tool_use_count": "three"}, "invalid tool_use_count"), ({"replied_since_task": "no"}, "invalid replied_since_task"), ({"progress_note": 42}, "invalid progress_note"), ({"last_activity_at": "soon"}, "invalid last_activity_at"), ): with self.subTest(extras=extras): host_request = AsyncMock(return_value={"subagents": [{**base_entry, **extras}]}) with patch.object(rlm_module, "host_request", host_request): with self.assertRaisesRegex(RuntimeError, pattern): asyncio.run(rlm_module.list_subagents()) class TestSpawnBriefComposition(unittest.TestCase): def test_composition_sections(self) -> None: brief = rlm_module.compose_child_brief( "run validation", name="worker-1", depth=1, read_only=True, deliverable="/tmp/report.md", evidence_bar="commands required", falsifier="speedup < 1", ) self.assertIn("You are child subagent 'worker-1' at depth 1.", brief) self.assertIn("Results arrive only through replies or files", brief) self.assertIn("# HARD RULES (READ-ONLY WORKER)", brief) self.assertIn("TASK:\nrun validation", brief) self.assertIn("DELIVERABLE: /tmp/report.md", brief) self.assertIn("/tmp/report.md.partial", brief) self.assertIn("EVIDENCE BAR:\ncommands required", brief) self.assertIn("FALSIFIER:\nspeedup < 1", brief) self.assertIn("FIRST ACTION: write /tmp/worker-1/status.md", brief) self.assertIn("LAST ACTION: reply to the parent", brief) minimal = rlm_module.compose_child_brief("run validation", name="worker-2", depth=1) self.assertNotIn("HARD RULES", minimal) self.assertNotIn("DELIVERABLE", minimal) self.assertNotIn("EVIDENCE BAR", minimal) self.assertNotIn("FALSIFIER", minimal) def test_lint_warning_behavior(self) -> None: import contextlib import io cases = [ ("plain prompt without path", None, None, None, True), ("plain prompt", "/tmp/out.md", None, None, False), ("plain prompt", None, "evidence", None, False), ("plain prompt", None, None, "falsifier", False), ("inspect src/index.ts", None, None, None, False), ("check /tmp/foo", None, None, None, False), ] for prompt, d, e, f, should_warn in cases: buf = io.StringIO() with contextlib.redirect_stderr(buf): rlm_module._check_brief_lint(prompt, name="w", deliverable=d, evidence_bar=e, falsifier=f) has_warning = "[WARNING] rlm.spawn: unlinted brief" in buf.getvalue() self.assertEqual(has_warning, should_warn, f"failed for {prompt=}") def test_spawn_passes_composed_prompt(self) -> None: host_request = AsyncMock( return_value={ "rlm_child_id": "sub-c1", "name": "scout", "session_dir": "/tmp/scout", "model": "m", } ) with patch.object(rlm_module, "host_request", host_request): asyncio.run( rlm_module.spawn( "profile memory", name="scout", model="m", read_only=True, deliverable="/tmp/mem.md", ) ) sent = host_request.await_args.args[1]["prompt"] self.assertIn("# HARD RULES", sent) self.assertIn("DELIVERABLE: /tmp/mem.md", sent) self.assertIn("/tmp/mem.md.partial", sent) if __name__ == "__main__": unittest.main()