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