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