feat(rlm): compose invariant child brief sections and warn on unlinted briefs
2539 unmodified linesExpand all2540254125422543254425442545254625472548103 unmodified linesExpand all2539 unmodified linesExpand all self.repl.execute("s6-import", "import rlm") events = self.repl.execute("s6", "rlm.spawn('x', name='n', role='coder')") text = self.traceback_text(events) self.assertEqual(one(events, "error")["ename"], "TypeError") self.assertIn("signature: rlm.spawn(prompt, *, name, model, thinking?=None, host?=None, workspace?=None, allow_push?=False)", text) self.assertIn("signature: rlm.spawn(prompt, *, name, model, thinking?=None, host?=None, workspace?=None, allow_push?=False, read_only?=False, deliverable?=None, evidence_bar?=None, falsifier?=None)", text) self.assertIn("nearest: 'role' -> model (rlm.spawn); use model=", text)
def test_namespace_skill_surface_answers_for_a_wrapped_module(self): setup = "\n".join(103 unmodified linesExpand all254 unmodified linesExpand all25525625725825925926026126226354 unmodified lines254 unmodified linesExpand all digest = surface.digest() self.assertIn("bash(command,", digest) self.assertIn("BashResult: ", digest) self.assertIn("BashHandle: ", digest) self.assertIn("rlm.spawn(prompt, *, name, model, thinking?=None, host?=None, workspace?=None, allow_push?=False)", digest) self.assertIn("rlm.spawn(prompt, *, name, model, thinking?=None, host?=None, workspace?=None, allow_push?=False, read_only?=False, deliverable?=None, evidence_bar?=None, falsifier?=None)", digest)
class RemedyTest(unittest.TestCase): SOURCE = DigestTest.SOURCE54 unmodified lines93 unmodified lines94959697989910010110110210310410510620 unmodified lines127128129130131132133132134135136137138352 unmodified linesExpand all49149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756893 unmodified lines 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": "check the API", "prompt": expected_prompt, "kwargs": { "name": "api-reviewer", "model": "anthropic/claude-opus-5", "thinking": "high",20 unmodified lines 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": "check the API", "kwargs": {"name": "api-reviewer", "model": "openai-codex/gpt-5.6-terra"}}, {"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:352 unmodified linesExpand all 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()1- Added structured brief composition (`read_only`, `deliverable`, `evidence_bar`, `falsifier`) to `rlm.spawn` with unlinted brief warnings and invariant action contracts.452 unmodified linesExpand all453454455456457458459460461462234 unmodified linesExpand all452 unmodified linesExpand all expect(prompt).toContain("name='worker'"); expect(prompt).toContain("session_dir"); expect(prompt).toContain("agent_observe"); expect(prompt).toContain("restricted to your parent, siblings, and direct children"); expect(prompt).toContain("Children write deliverables atomically"); expect(prompt).toContain("The parent must re-run the child's one-command gate"); });
test("omits ipython-only subagent guidance when ipython is inactive", () => { const harnessState: HarnessState = {234 unmodified linesExpand all12345678910250 unmodified linesExpand all2612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673681 unmodified line37037137237337437537637737837938038113 unmodified lines39539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244312 unmodified lines456457458459320460461462463464372 unmodified linesExpand all8378388398408418428438448458468478481 unmodified line85085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388463 unmodified lines94894995095195295395495595626 unmodified lines"""Tiny rlm-compatible kernel shim for Prime Agent."""
from __future__ import annotations
import osimport reimport sysimport typesfrom dataclasses import dataclassfrom pathlib import Path250 unmodified linesExpand all raise RuntimeError("rlm.models returned an invalid payload") return RLMModels(payload)
_PATH_LIKE_PATTERN = re.compile( r"(?:" r"/(?:[a-zA-Z0-9_.-]+/)*[a-zA-Z0-9_.-]+" r"|\./[a-zA-Z0-9_.-]+" r"|[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+" r"|[a-zA-Z0-9_.-]+\.(?:md|txt|json|yaml|yml|toml|py|ts|js|rs|go|diff|patch|log|csv)\b" r")")
def _has_path_like_token(text: str) -> bool: return bool(_PATH_LIKE_PATTERN.search(text))
def _warn_brief_lint(name: str) -> None: warning = ( f"[WARNING] rlm.spawn: unlinted brief for child {name!r}: missing deliverable (or output path), " "evidence_bar, and falsifier. Pass deliverable=..., evidence_bar=..., or falsifier=... " "(see delegation guidance in model prompt)." ) try: print(warning, file=sys.stderr) except Exception: pass
def _check_brief_lint( prompt: str, *, name: str, deliverable: str | None, evidence_bar: str | None, falsifier: str | None,) -> None: has_structured = bool(deliverable or evidence_bar or falsifier) if not has_structured and not _has_path_like_token(prompt): _warn_brief_lint(name)
def _current_child_depth() -> int: try: return int(os.environ.get("RLM_DEPTH", "0") or "0") + 1 except ValueError: return 1
def compose_child_brief( prompt: str, *, name: str, depth: int | None = None, read_only: bool = False, deliverable: str | None = None, evidence_bar: str | None = None, falsifier: str | None = None,) -> str: """Compose the invariant and structured parts of a child task brief.""" actual_depth = _current_child_depth() if depth is None else depth sections: list[str] = [ f"You are child subagent {name!r} at depth {actual_depth}.\n" "Results arrive only through replies or files, never as an rlm.spawn() return value." ]
if read_only: sections.append( "# HARD RULES (READ-ONLY WORKER)\n" f"- Do not modify anything outside your own scratch directory (/tmp/{name}).\n" "- Do not create, edit, delete, move, or rename any file in the repository.\n" "- Do not run state-changing git or jj commands (commit, reset, checkout, push, stash).\n" "- No builds, GPU work, or package installs that change state.\n" "- Reading and reasoning is the job. If changes are needed, describe them in the deliverable instead of making them." )
clean_prompt = prompt.strip() if clean_prompt.startswith("TASK:") or clean_prompt.startswith("# TASK"): sections.append(clean_prompt) else: sections.append(f"TASK:\n{clean_prompt}")
if deliverable is not None: sections.append( f"DELIVERABLE: {deliverable.strip()}\n" f"Write to {deliverable.strip()}.partial and rename it into place when finished, " "so a parent that reads the path early can never see a half-written artifact." )
if evidence_bar is not None: sections.append(f"EVIDENCE BAR:\n{evidence_bar.strip()}")
if falsifier is not None: sections.append(f"FALSIFIER:\n{falsifier.strip()}")
sections.append( f"FIRST ACTION: write /tmp/{name}/status.md with a 5-line plan, then update it after each step.\n" "LAST ACTION: reply to the parent via `await agent_message.send(message, receiver_role='parent')`." )
return "\n\n".join(sections)
async def spawn( prompt: str, *, name: str,1 unmodified line thinking: str | None = None, host: str | None = None, workspace: str | None = None, allow_push: bool = False, read_only: bool = False, deliverable: str | None = None, evidence_bar: str | None = None, falsifier: str | None = None,) -> RLMSpawnHandle: """Spawn a recursive Prime Agent child and return once its task is admitted.
``name`` and ``model`` are required. ``await rlm.models()`` must be called13 unmodified lines
``allow_push=True`` authorizes the child's skills to publish (``git push``, force resets) that a child would otherwise refuse; the child inherits it as ``RLM_ALLOW_PUSH=1``.
``read_only=True`` adds a HARD RULES block restricting the child to its scratch directory without modifying the repository, running state-changing git commands, or executing stateful builds.
``deliverable`` specifies the destination file for the child's report, instructing it to write to ``<path>.partial`` and rename atomically into place.
``evidence_bar`` defines the standard of proof required for claims.
``falsifier`` defines the pre-registered condition that refutes the hypothesis. """ if not isinstance(prompt, str): raise TypeError(f"prompt must be str, got {type(prompt).__name__}") if not isinstance(name, str) or not name: raise TypeError("name is required and must be a non-empty str") if not isinstance(model, str) or not model: raise TypeError("model is required and must be a non-empty str") if not isinstance(read_only, bool): raise TypeError(f"read_only must be a bool, got {type(read_only).__name__}") if deliverable is not None and (not isinstance(deliverable, str) or not deliverable.strip()): raise TypeError("deliverable must be a non-empty str when set") if evidence_bar is not None and (not isinstance(evidence_bar, str) or not evidence_bar.strip()): raise TypeError("evidence_bar must be a non-empty str when set") if falsifier is not None and (not isinstance(falsifier, str) or not falsifier.strip()): raise TypeError("falsifier must be a non-empty str when set") _check_brief_lint( prompt, name=name, deliverable=deliverable, evidence_bar=evidence_bar, falsifier=falsifier, ) composed_prompt = compose_child_brief( prompt, name=name, read_only=read_only, deliverable=deliverable, evidence_bar=evidence_bar, falsifier=falsifier, ) kwargs: dict[str, Any] = {"name": name, "model": model} if thinking is not None: kwargs["thinking"] = thinking if host is not None:12 unmodified lines raise ValueError("workspace='isolated' cannot be combined with a remote host") created = create_workspace(name) kwargs["cwd"] = str(created.path) try: payload = await host_request("rlm.run", {"prompt": prompt, "kwargs": kwargs}) payload = await host_request("rlm.run", {"prompt": composed_prompt, "kwargs": kwargs}) except BaseException: if created is not None: created.prune() raise372 unmodified linesExpand all thinking: str | None = None, host: str | None = None, workspace: str | None = None, allow_push: bool = False, read_only: bool = False, deliverable: str | None = None, evidence_bar: str | None = None, falsifier: str | None = None, ) -> RLMSpawnHandle: return await spawn( prompt, name=name,1 unmodified line thinking=thinking, host=host, workspace=workspace, allow_push=allow_push, read_only=read_only, deliverable=deliverable, evidence_bar=evidence_bar, falsifier=falsifier, )
def compose_child_brief( self, prompt: str, *, name: str, depth: int | None = None, read_only: bool = False, deliverable: str | None = None, evidence_bar: str | None = None, falsifier: str | None = None, ) -> str: return compose_child_brief( prompt, name=name, depth=depth, read_only=read_only, deliverable=deliverable, evidence_bar=evidence_bar, falsifier=falsifier, )
async def create_session( self, prompt: str, name: str | None = None,63 unmodified lines "RLMProgressNoteResult", "RLMSpawnHandle", "RLMSubagent", "RLMSubagentActivity", "compose_child_brief", "create_session", "RefinementEvent", "bash", "delete_subagent",26 unmodified lines255 unmodified linesExpand all2562572582592602612622632642653 unmodified lines255 unmodified linesExpand all } lines.push( "Fan-in results with `await rlm.collect(targets, timeout_ms=0)`: it returns typed snapshots of direct children (status, answer preview, error) without steering anyone; an explicit timeout blocks only that call until the children settle or the deadline passes.", "Large child outputs belong in files that you read selectively; `collect` snapshots are previews, not full results.", "Children write deliverables atomically (write to `<path>.partial` and rename into place).", "The parent must re-run the child's one-command gate (a deterministic-output hash, a build provenance check, or the arithmetic) before any child-reported number enters memory, a report, or a message to the user.", "Delegate parallel context-heavy research or independent implementation; do a single known lookup, edit, or command inline.", ); if (options.includeRefineExamples ?? true) { lines.push("Persist genuinely reusable delegation patterns with `await refine.run()`.");3 unmodified lines