from __future__ import annotations import asyncio import importlib import unittest from pathlib import Path from unittest.mock import AsyncMock, call, patch rlm_module = importlib.import_module("rlm") class RlmSubagentRegistryTest(unittest.TestCase): def test_lists_parent_scoped_subagents_from_host(self) -> None: host_request = AsyncMock( return_value={ "subagents": [ { "rlm_child_id": "sub-a1b2c3d4", "active_session_id": "active-child", "session_id": "session-child", "session_name": "subagent-check-api-a1b2c3d4", "session_dir": "/tmp/parent/sub-a1b2c3d4", "status": "completed", } ] } ) with patch.object(rlm_module, "host_request", host_request): subagents = asyncio.run(rlm_module.rlm.list_subagents()) self.assertEqual(len(subagents), 1) self.assertEqual(subagents[0].rlm_child_id, "sub-a1b2c3d4") self.assertEqual(subagents[0].active_session_id, "active-child") self.assertEqual(subagents[0].session_id, "session-child") self.assertEqual(subagents[0].session_name, "subagent-check-api-a1b2c3d4") self.assertEqual(subagents[0].session_dir, Path("/tmp/parent/sub-a1b2c3d4")) self.assertEqual(subagents[0].status, "completed") host_request.assert_awaited_once_with("rlm.list_subagents") def test_lists_failed_subagents_from_host(self) -> None: host_request = AsyncMock( return_value={ "subagents": [ { "rlm_child_id": "sub-failed", "active_session_id": None, "session_id": None, "session_name": "failed-worker", "session_dir": "/tmp/parent/sub-failed", "status": "error", } ] } ) with patch.object(rlm_module, "host_request", host_request): subagents = asyncio.run(rlm_module.rlm.list_subagents()) self.assertEqual(subagents[0].status, "error") def test_forwards_orchestrator_chosen_name_and_model_to_host(self) -> None: host_request = AsyncMock( return_value={ "rlm_child_id": "sub-a1b2c3d4", "name": "api-reviewer", "session_dir": "/tmp/parent/sub-a1b2c3d4", "model": "deepseek/deepseek-v4-flash", } ) with patch.object(rlm_module, "host_request", host_request): result = asyncio.run( rlm_module.rlm( "check the API", name="api-reviewer", model="deepseek/deepseek-v4-flash", result={"schema": {"type": "object"}, "required": False}, ) ) host_request.assert_awaited_once_with( "rlm.run", { "prompt": "check the API", "kwargs": { "name": "api-reviewer", "model": "deepseek/deepseek-v4-flash", "result": {"schema": {"type": "object"}, "required": False}, }, }, ) self.assertEqual(result.rlm_child_id, "sub-a1b2c3d4") self.assertEqual(result.name, "api-reviewer") self.assertEqual(result.model, "deepseek/deepseek-v4-flash") def test_parses_contract_result_path_from_subagent(self) -> None: host_request = AsyncMock( return_value={ "rlm_child_id": "sub-a1b2c3d4", "name": "contract-worker", "session_dir": "/tmp/parent/sub-a1b2c3d4", "model": "test/model", "result_path": "/tmp/parent/sub-a1b2c3d4/rlm-result.json", } ) with patch.object(rlm_module, "host_request", host_request): result = asyncio.run( rlm_module.rlm( "check", result={"schema": {"type": "object", "properties": {}}}, context={"result": {"fields": []}}, ) ) self.assertEqual(result.result_path, Path("/tmp/parent/sub-a1b2c3d4/rlm-result.json")) def test_rejects_host_bound_integers_outside_javascript_safe_range(self) -> None: too_large = 2**53 host_request = AsyncMock() with patch.object(rlm_module, "host_request", host_request): with self.assertRaisesRegex(ValueError, "safe range"): asyncio.run( rlm_module.rlm( "check", result={ "schema": { "type": "object", "properties": {"count": {"const": too_large}}, } }, ) ) with self.assertRaisesRegex(ValueError, "safe range"): asyncio.run(rlm_module.rlm.submit_result({"count": too_large})) host_request.assert_not_awaited() def test_submits_json_result_without_spoofable_identity(self) -> None: host_request = AsyncMock( return_value={ "child_id": "sub-a1b2c3d4", "generation": 0, "path": "/tmp/result.json", "sha256": "a" * 64, } ) with patch.object(rlm_module, "host_request", host_request): receipt = asyncio.run(rlm_module.rlm.submit_result({"status": "approved"})) self.assertEqual(receipt.rlm_child_id, "sub-a1b2c3d4") self.assertEqual(receipt.generation, 0) host_request.assert_awaited_once_with( "rlm.submit_result", {"value": {"status": "approved"}}, ) with self.assertRaisesRegex(ValueError, "non-finite"): asyncio.run(rlm_module.submit_result({"score": float("nan")})) with self.assertRaisesRegex(TypeError, "JSON values"): asyncio.run(rlm_module.submit_result({"value": {1, 2}})) def test_reads_only_requested_contract_result_fields(self) -> None: host_request = AsyncMock( return_value={ "child_id": "sub-a1b2c3d4", "generation": 1, "path": "/tmp/result.json", "sha256": "b" * 64, "result": {"status": "approved"}, } ) handle = rlm_module.RLMSubagent( rlm_child_id="sub-a1b2c3d4", name="worker", session_dir=Path("/tmp/worker"), model="test/model", ) with patch.object(rlm_module, "host_request", host_request): result = asyncio.run(rlm_module.rlm.get_result(handle, fields=("status",))) self.assertEqual(result.generation, 1) self.assertEqual(result.value, {"status": "approved"}) host_request.assert_awaited_once_with( "rlm.get_result", {"target": "sub-a1b2c3d4", "fields": ["status"]}, ) with self.assertRaisesRegex(ValueError, "duplicates"): asyncio.run(rlm_module.get_result(handle, fields=("status", "status"))) with self.assertRaisesRegex(TypeError, "generation"): asyncio.run(rlm_module.get_result(handle, generation=True)) def test_run_returns_mutable_subagent_with_handle_api(self) -> None: host_request = AsyncMock( return_value={ "rlm_child_id": "sub-a1b2c3d4", "name": "worker", "session_dir": "/tmp/parent/sub-a1b2c3d4", "model": "test/model", } ) with patch.object(rlm_module, "host_request", host_request): child = asyncio.run( rlm_module.rlm("check", result={"schema": {"type": "object"}, "required": False}) ) self.assertIsInstance(child, rlm_module.RLMSubagent) self.assertEqual(child.id, "sub-a1b2c3d4") self.assertEqual(child.session_name, "worker") self.assertTrue(child.running) for method in ("refresh", "send", "observe", "info", "result", "delete"): self.assertTrue(callable(getattr(child, method)), method) def test_requires_result_contract_unless_unguarded(self) -> None: host_request = AsyncMock( return_value={ "rlm_child_id": "sub-optout", "name": "optout", "session_dir": "/tmp/parent/sub-optout", "model": "test/model", } ) with patch.object(rlm_module, "host_request", host_request): with self.assertRaisesRegex(ValueError, "result contract"): asyncio.run(rlm_module.rlm("no schema")) with self.assertRaisesRegex(TypeError, "unguarded must be a bool"): asyncio.run(rlm_module.rlm("no schema", unguarded="yes")) # Explicit opt-out skips the requirement and passes through to the host. asyncio.run(rlm_module.rlm("no schema, opted out", unguarded=True)) self.assertEqual(host_request.await_count, 1) self.assertEqual( host_request.await_args.args[1], {"prompt": "no schema, opted out", "kwargs": {"unguarded": True}}, ) def test_subagent_send_and_observe_route_through_host(self) -> None: child = rlm_module.RLMSubagent( rlm_child_id="sub-a1b2c3d4", name="worker", session_dir=Path("/tmp/parent/sub-a1b2c3d4"), model="test/model", ) replies = [ {"deliveryStatus": "sent"}, {"target": "worker", "messages": []}, ] host_request = AsyncMock(side_effect=replies) with patch.object(rlm_module, "host_request", host_request), patch.object( rlm_module, "emit" ): receipt = asyncio.run(child.send("status?")) observed = asyncio.run(child.observe(limit=3, max_chars=64)) self.assertEqual(receipt, {"deliveryStatus": "sent"}) self.assertEqual(observed, {"target": "worker", "messages": []}) self.assertEqual( host_request.await_args_list, [ call("agent_message.send", {"message": "status?", "receiver_role": "child", "receiver_name": "worker"}), call("agent_observe.recent", {"target": "worker", "limit": 3, "max_chars": 64}), ], ) def test_subagent_send_rejects_deleted_children(self) -> None: child = rlm_module.RLMSubagent( rlm_child_id="sub-a1b2c3d4", name="worker", session_dir=Path("/tmp/parent/sub-a1b2c3d4"), status="deleted", ) with patch.object(rlm_module, "host_request", AsyncMock()) as host_request: with self.assertRaisesRegex(RuntimeError, "deleted"): asyncio.run(child.send("hello")) host_request.assert_not_awaited() def test_subagent_delete_merges_state_in_place(self) -> None: child = rlm_module.RLMSubagent( rlm_child_id="sub-a1b2c3d4", name="worker", session_dir=Path("/tmp/old"), model="test/model", status="running", ) host_request = AsyncMock( return_value={ "subagent": { "rlm_child_id": "sub-a1b2c3d4", "active_session_id": None, "session_id": "session-child", "session_name": "worker", "session_dir": "/tmp/parent/sub-a1b2c3d4", "status": "completed", } } ) with patch.object(rlm_module, "host_request", host_request): returned = asyncio.run(child.delete()) self.assertIs(returned, child) self.assertEqual(child.status, "deleted") self.assertEqual(child.session_id, "session-child") self.assertEqual(child.session_dir, Path("/tmp/parent/sub-a1b2c3d4")) with patch.object(rlm_module, "host_request", AsyncMock()) as host_request: asyncio.run(child.delete()) host_request.assert_not_awaited() def test_subagent_refresh_merges_registry_entry(self) -> None: child = rlm_module.RLMSubagent( rlm_child_id="sub-a1b2c3d4", name="worker", session_dir=Path("/tmp/old"), status="admitted", ) host_request = AsyncMock( return_value={ "subagents": [ { "rlm_child_id": "sub-a1b2c3d4", "active_session_id": "active-child", "session_id": "session-child", "session_name": "worker", "session_dir": "/tmp/parent/sub-a1b2c3d4", "status": "completed", } ] } ) with patch.object(rlm_module, "host_request", host_request): refreshed = asyncio.run(child.refresh()) self.assertIs(refreshed, child) self.assertEqual(child.status, "completed") self.assertEqual(child.active_session_id, "active-child") with patch.object(rlm_module, "host_request", AsyncMock(return_value={"subagents": []})): missing = asyncio.run(child.refresh()) self.assertIs(missing, child) self.assertEqual(child.status, "missing") def test_send_parent_sibling_child_route_through_host(self) -> None: host_request = AsyncMock(side_effect=[{"ok": True}, {"ok": True}, {"ok": True}]) with patch.object(rlm_module, "host_request", host_request), patch.object( rlm_module, "emit" ): asyncio.run(rlm_module.send_parent("to parent")) asyncio.run(rlm_module.send_sibling("beta", "to sibling")) asyncio.run(rlm_module.send_child("worker", "to child")) self.assertEqual( host_request.await_args_list, [ call("agent_message.send", {"message": "to parent", "receiver_role": "parent", "receiver_name": None}), call("agent_message.send", {"message": "to sibling", "receiver_role": "sibling", "receiver_name": "beta"}), call("agent_message.send", {"message": "to child", "receiver_role": "child", "receiver_name": "worker"}), ], ) def test_send_validates_message_and_receiver(self) -> None: with patch.object(rlm_module, "host_request", AsyncMock()) as host_request: with self.assertRaisesRegex(TypeError, "non-empty str"): asyncio.run(rlm_module.send_parent("")) with self.assertRaisesRegex(ValueError, "receiver_name is required"): asyncio.run(rlm_module._send_agent_message("hi", receiver_role="sibling")) with self.assertRaisesRegex(ValueError, "invalid receiver_role"): asyncio.run(rlm_module._send_agent_message("hi", receiver_role="cousin", receiver_name="x")) with self.assertRaisesRegex(TypeError, "non-empty str"): asyncio.run(rlm_module.observe(" ")) host_request.assert_not_awaited() def test_family_and_agent_info_route_through_host(self) -> None: host_request = AsyncMock(side_effect=[{"entries": []}, {"name": "worker"}]) with patch.object(rlm_module, "host_request", host_request): roster = asyncio.run(rlm_module.family()) info = asyncio.run(rlm_module.agent_info("worker")) self.assertEqual(roster, {"entries": []}) self.assertEqual(info, {"name": "worker"}) self.assertEqual( host_request.await_args_list, [call("agent_message.list_agents"), call("agent_observe.get", {"target": "worker"})], ) def test_finds_authenticated_models_through_host(self) -> None: host_request = AsyncMock( return_value={ "models": [ { "provider": "anthropic", "id": "claude-opus-4-7", "name": "Claude Opus 4.7", "selector": "anthropic/claude-opus-4-7", } ] } ) with patch.object(rlm_module, "host_request", host_request): models = asyncio.run(rlm_module.rlm.find_models("opus", limit=3)) self.assertEqual(models[0].provider, "anthropic") self.assertEqual(models[0].id, "claude-opus-4-7") self.assertEqual(models[0].name, "Claude Opus 4.7") self.assertEqual(models[0].selector, "anthropic/claude-opus-4-7") host_request.assert_awaited_once_with( "rlm.find_models", {"query": "opus", "limit": 3}, ) def test_rejects_invalid_model_search_input_and_response(self) -> None: with self.assertRaisesRegex(TypeError, "query must be str"): asyncio.run(rlm_module.find_models(123)) with self.assertRaisesRegex(TypeError, "limit must be int"): asyncio.run(rlm_module.find_models("opus", limit="3")) host_request = AsyncMock(return_value={"models": [{"provider": "anthropic"}]}) with patch.object(rlm_module, "host_request", host_request): with self.assertRaisesRegex(RuntimeError, "invalid model entry"): asyncio.run(rlm_module.find_models("opus")) 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", 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_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 str or RLMSubagent"): 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()) if __name__ == "__main__": unittest.main()