"""Tiny rlm-compatible kernel shim for Prime Agent."""
from __future__ import annotations
import math
import sys
import types
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from .bash import BashHandle, BashResult, bash
from .harness import HarnessEntry, HarnessScope, HarnessState, RefinementEvent, get_harness_state
_MAX_SAFE_INTEGER = 2**53 - 1
@dataclass
class RLMSubagent:
rlm_child_id: str
name: str
session_dir: Path
model: str | None = None
result_path: Path | None = None
active_session_id: str | None = None
session_id: str | None = None
status: str = "admitted"
@property
def id(self) -> str:
return self.rlm_child_id
@property
def session_name(self) -> str:
return self.name
@property
def running(self) -> bool:
return self.status in {"admitted", "running"}
async def refresh(self) -> RLMSubagent:
"""Refresh this child snapshot in place."""
for child in await list_subagents():
if child.rlm_child_id == self.rlm_child_id:
self._merge(child)
return self
self.status = "missing"
return self
async def send(self, message: str) -> dict[str, Any]:
"""Send a direct message to this child."""
if self.status == "deleted":
raise RuntimeError(f"subagent {self.name!r} has been deleted")
return await send_child(self.name, message)
async def observe(self, limit: int = 8, max_chars: int = 800) -> dict[str, Any]:
"""Read bounded recent message previews from this child."""
return await observe(self.name, limit=limit, max_chars=max_chars)
async def info(self) -> dict[str, Any]:
"""Read the host's current summary for this child."""
return await agent_info(self.name)
async def result(
self,
fields: list[str] | tuple[str, ...] | None = None,
generation: int | None = None,
) -> RLMResult:
"""Read this child's admitted structured result."""
return await get_result(self, fields=fields, generation=generation)
async def delete(self) -> RLMSubagent:
"""Delete this child and update this handle in place."""
return await delete_subagent(self)
def _merge(self, child: RLMSubagent) -> None:
self.name = child.name
self.session_dir = child.session_dir
self.active_session_id = child.active_session_id
self.session_id = child.session_id
self.status = child.status
if child.model is not None:
self.model = child.model
if child.result_path is not None:
self.result_path = child.result_path
@dataclass(frozen=True)
class RLMResultSubmission:
rlm_child_id: str
generation: int
path: Path
sha256: str
@dataclass(frozen=True)
class RLMResult:
rlm_child_id: str
generation: int
path: Path
sha256: str
value: dict[str, Any]
@dataclass(frozen=True)
class RLMModel:
provider: str
id: str
name: str
selector: str
def _spawn_subagent_from_payload(payload: Any) -> RLMSubagent:
if not isinstance(payload, dict):
raise RuntimeError("rlm.run returned an invalid subagent")
child_id = payload.get("rlm_child_id")
name = payload.get("name")
session_dir = payload.get("session_dir")
model = payload.get("model")
result_path = payload.get("result_path")
if not all(isinstance(value, str) and value for value in (child_id, name, session_dir, model)):
raise RuntimeError("rlm.run returned an invalid subagent")
if result_path is not None and (not isinstance(result_path, str) or not result_path):
raise RuntimeError("rlm.run returned an invalid result_path")
return RLMSubagent(
rlm_child_id=child_id,
name=name,
session_dir=Path(session_dir),
model=model,
result_path=Path(result_path) if result_path is not None else None,
)
def _parse_host_reply(request_type: str, reply: dict[str, Any]) -> dict[str, Any]:
status = reply.get("status")
if status == "ok":
return reply["result"]
if status == "error":
raise RuntimeError(str(reply.get("error") or f"host request {request_type} failed"))
raise RuntimeError(f"host request {request_type} returned unexpected status: {status!r}")
async def host_request(request_type: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
"""Send a typed request to the Prime Agent host and await its reply.
This is the kernel side of the generic host bridge: Python skills call
``await host_request("<type>", {...})`` and the TypeScript host dispatches
on the type. Raises RuntimeError when the host reports an error or when no
handler for the type is registered in this session.
"""
if not isinstance(request_type, str) or not request_type:
raise TypeError("request_type must be a non-empty str")
if payload is not None and not isinstance(payload, dict):
raise TypeError(f"payload must be a dict or None, got {type(payload).__name__}")
from . import repl
# request_type goes last so a payload "type" key cannot reroute the request.
reply = await repl.host_request({**(payload or {}), "type": request_type})
return _parse_host_reply(request_type, reply)
def emit(data: dict[str, Any]) -> None:
"""Ship one display event (dict of MIME type -> JSON payload) to the host."""
from . import repl
repl.emit(data)
async def run(prompt: str, **kwargs: Any) -> RLMSubagent:
"""Spawn a recursive Prime Agent child and return once its task is admitted.
``model`` selects a child with an exact ``provider/model`` selector.
``thinking`` sets the child reasoning level (e.g. 'off', 'low', 'medium', 'high');
defaults to the parent level; levels invalid for the resolved model fail the spawn.
``result`` and ``context`` declare independent structured-result and context-admission contracts.
"""
if not isinstance(prompt, str):
raise TypeError(f"prompt must be str, got {type(prompt).__name__}")
unguarded = kwargs.get("unguarded")
if unguarded is not None and not isinstance(unguarded, bool):
raise TypeError("unguarded must be a bool")
if "result" not in kwargs and unguarded is not True:
raise ValueError(
"rlm requires a result contract so the child's communication with the parent is "
"schema-bound and context-guarded; pass result={...} or the explicit opt-out "
"unguarded=True"
)
_assert_json_value(kwargs, "kwargs")
payload = await host_request("rlm.run", {"prompt": prompt, "kwargs": kwargs})
return _spawn_subagent_from_payload(payload)
def _assert_json_value(value: Any, path: str = "value", depth: int = 0) -> None:
if depth > 32:
raise ValueError(f"{path} exceeds the maximum JSON nesting depth")
if value is None or isinstance(value, (str, bool)):
return
if isinstance(value, int):
if abs(value) > _MAX_SAFE_INTEGER:
raise ValueError(f"{path} integer exceeds the JavaScript safe range")
return
if isinstance(value, float):
if not math.isfinite(value):
raise ValueError(f"{path} must not contain a non-finite number")
if value.is_integer() and abs(value) > _MAX_SAFE_INTEGER:
raise ValueError(f"{path} integer exceeds the JavaScript safe range")
return
if isinstance(value, list):
for index, item in enumerate(value):
_assert_json_value(item, f"{path}[{index}]", depth + 1)
return
if isinstance(value, dict):
for key, item in value.items():
if not isinstance(key, str):
raise TypeError(f"{path} object keys must be strings")
_assert_json_value(item, f"{path}.{key}", depth + 1)
return
raise TypeError(f"{path} must contain only JSON values, got {type(value).__name__}")
def _result_submission_from_payload(payload: Any) -> RLMResultSubmission:
if not isinstance(payload, dict):
raise RuntimeError("rlm.submit_result returned an invalid receipt")
child_id = payload.get("child_id")
generation = payload.get("generation")
path = payload.get("path")
sha256 = payload.get("sha256")
if (
not isinstance(child_id, str)
or not child_id
or isinstance(generation, bool)
or not isinstance(generation, int)
or generation < 0
or generation > _MAX_SAFE_INTEGER
or not isinstance(path, str)
or not path
or not isinstance(sha256, str)
or not sha256
):
raise RuntimeError("rlm.submit_result returned an invalid receipt")
return RLMResultSubmission(
rlm_child_id=child_id, generation=generation, path=Path(path), sha256=sha256
)
async def submit_result(value: dict[str, Any]) -> RLMResultSubmission:
"""Submit this child's host-bound structured result once for the active task turn."""
if not isinstance(value, dict):
raise TypeError(f"value must be dict, got {type(value).__name__}")
_assert_json_value(value)
payload = await host_request("rlm.submit_result", {"value": value})
return _result_submission_from_payload(payload)
def _result_target(target: str | RLMSubagent) -> str:
if isinstance(target, RLMSubagent):
return target.rlm_child_id
if isinstance(target, str) and target.strip():
return target.strip()
raise TypeError("target must be a non-empty str or RLMSubagent")
async def get_result(
target: str | RLMSubagent,
fields: list[str] | tuple[str, ...] | None = None,
generation: int | None = None,
) -> RLMResult:
"""Read the schema-validated fields admitted by one direct child's contract."""
selector = _result_target(target)
if fields is not None:
if not isinstance(fields, (list, tuple)) or not all(
isinstance(field, str) and field for field in fields
):
raise TypeError("fields must be a list or tuple of non-empty strings")
if len(set(fields)) != len(fields):
raise ValueError("fields must not contain duplicates")
if generation is not None and (
isinstance(generation, bool)
or not isinstance(generation, int)
or generation < 0
or generation > _MAX_SAFE_INTEGER
):
raise TypeError("generation must be a non-negative int or None")
payload = await host_request(
"rlm.get_result",
{
"target": selector,
**({"fields": list(fields)} if fields is not None else {}),
**({"generation": generation} if generation is not None else {}),
},
)
child_id = payload.get("child_id")
result_generation = payload.get("generation")
path = payload.get("path")
sha256 = payload.get("sha256")
value = payload.get("result")
if (
not isinstance(child_id, str)
or not child_id
or isinstance(result_generation, bool)
or not isinstance(result_generation, int)
or result_generation < 0
or result_generation > _MAX_SAFE_INTEGER
or not isinstance(path, str)
or not path
or not isinstance(sha256, str)
or not sha256
or not isinstance(value, dict)
):
raise RuntimeError("rlm.get_result returned an invalid result")
return RLMResult(
rlm_child_id=child_id,
generation=result_generation,
path=Path(path),
sha256=sha256,
value=value,
)
def _model_from_payload(payload: Any) -> RLMModel:
if not isinstance(payload, dict):
raise RuntimeError("rlm.find_models returned an invalid model entry")
provider = payload.get("provider")
model_id = payload.get("id")
name = payload.get("name")
selector = payload.get("selector")
if not all(isinstance(value, str) and value for value in (provider, model_id, name, selector)):
raise RuntimeError("rlm.find_models returned an invalid model entry")
return RLMModel(provider=provider, id=model_id, name=name, selector=selector)
async def find_models(query: str = "", limit: int = 8) -> list[RLMModel]:
"""Search a bounded list of models backed by active user credentials."""
if not isinstance(query, str):
raise TypeError(f"query must be str, got {type(query).__name__}")
if not isinstance(limit, int):
raise TypeError(f"limit must be int, got {type(limit).__name__}")
payload = await host_request("rlm.find_models", {"query": query, "limit": limit})
models = payload.get("models")
if not isinstance(models, list):
raise RuntimeError("rlm.find_models returned an invalid models list")
return [_model_from_payload(model) for model in models]
def _subagent_from_payload(payload: Any, operation: str = "rlm.list_subagents") -> RLMSubagent:
if not isinstance(payload, dict):
raise RuntimeError(f"{operation} returned an invalid subagent entry")
child_id = payload.get("rlm_child_id")
active_session_id = payload.get("active_session_id")
session_id = payload.get("session_id")
name = payload.get("session_name")
session_dir = payload.get("session_dir")
status = payload.get("status")
model = payload.get("model")
result_path = payload.get("result_path")
if not isinstance(child_id, str) or not child_id:
raise RuntimeError(f"{operation} entry is missing rlm_child_id")
if active_session_id is not None and not isinstance(active_session_id, str):
raise RuntimeError(f"{operation} entry has invalid active_session_id")
if session_id is not None and not isinstance(session_id, str):
raise RuntimeError(f"{operation} entry has invalid session_id")
if not isinstance(name, str) or not name:
raise RuntimeError(f"{operation} entry is missing session_name")
if not isinstance(session_dir, str) or not session_dir:
raise RuntimeError(f"{operation} entry is missing session_dir")
if status not in {"running", "completed", "error"}:
raise RuntimeError(f"{operation} entry has invalid status")
if model is not None and (not isinstance(model, str) or not model):
raise RuntimeError(f"{operation} entry has invalid model")
if result_path is not None and (not isinstance(result_path, str) or not result_path):
raise RuntimeError(f"{operation} entry has invalid result_path")
return RLMSubagent(
rlm_child_id=child_id,
name=name,
session_dir=Path(session_dir),
model=model,
result_path=Path(result_path) if result_path is not None else None,
active_session_id=active_session_id,
session_id=session_id,
status=status,
)
async def list_subagents() -> list[RLMSubagent]:
"""List direct RLM children as the same type returned by `rlm()`."""
payload = await host_request("rlm.list_subagents")
entries = payload.get("subagents")
if not isinstance(entries, list):
raise RuntimeError("rlm.list_subagents returned an invalid subagents registry")
return [_subagent_from_payload(entry) for entry in entries]
async def family() -> dict[str, Any]:
"""List this agent's parent, siblings, and direct children."""
return await host_request("agent_message.list_agents")
async def send_parent(message: str) -> dict[str, Any]:
"""Send a direct message to this agent's parent."""
return await _send_agent_message(message, receiver_role="parent")
async def send_sibling(name: str, message: str) -> dict[str, Any]:
"""Send a direct message to a root sibling."""
return await _send_agent_message(message, receiver_role="sibling", receiver_name=name)
async def send_child(name: str, message: str) -> dict[str, Any]:
"""Send a direct message to a direct child."""
return await _send_agent_message(message, receiver_role="child", receiver_name=name)
async def observe(target: str, limit: int = 8, max_chars: int = 800) -> dict[str, Any]:
"""Read bounded recent message previews for a visible family agent."""
_validate_agent_target(target)
return await host_request(
"agent_observe.recent",
{"target": target, "limit": limit, "max_chars": max_chars},
)
async def agent_info(target: str) -> dict[str, Any]:
"""Read one visible family agent summary."""
_validate_agent_target(target)
return await host_request("agent_observe.get", {"target": target})
async def _send_agent_message(
message: str,
*,
receiver_role: str,
receiver_name: str | None = None,
) -> dict[str, Any]:
if not isinstance(message, str) or not message:
raise TypeError("message must be a non-empty str")
if receiver_role == "parent":
if receiver_name is not None:
raise ValueError("receiver_name must be omitted for parent messages")
elif receiver_role in {"sibling", "child"}:
if not isinstance(receiver_name, str) or not receiver_name.strip():
raise ValueError("receiver_name is required for sibling and child messages")
else:
raise ValueError("invalid receiver_role")
receipt = await host_request(
"agent_message.send",
{
"message": message,
"receiver_role": receiver_role,
"receiver_name": receiver_name,
},
)
_emit_agent_message(receipt, receiver_role)
return receipt
def _emit_agent_message(receipt: dict[str, Any], receiver_role: str) -> None:
try:
label = "Agent message queued" if receipt.get("deliveryStatus") == "queued" else "Agent message sent"
emit(
{
"application/vnd.prime-agent.agent-message+json": {
**receipt,
"receiverRole": receiver_role,
},
"text/plain": label,
}
)
except Exception:
pass
def _validate_agent_target(target: str) -> None:
if not isinstance(target, str) or not target.strip():
raise TypeError("target must be a non-empty str")
async def delete_subagent(target: str | RLMSubagent) -> RLMSubagent:
"""Delete a direct child and update an object target in place."""
if isinstance(target, RLMSubagent):
if target.status == "deleted":
return target
selector = target.rlm_child_id
elif isinstance(target, str):
selector = target.strip()
if not selector:
raise ValueError("target must not be empty")
else:
raise TypeError(f"target must be str or RLMSubagent, got {type(target).__name__}")
payload = await host_request("rlm.delete_subagent", {"target": selector})
deleted = _subagent_from_payload(payload.get("subagent"), "rlm.delete_subagent")
if isinstance(target, RLMSubagent):
target._merge(deleted)
target.status = "deleted"
return target
deleted.status = "deleted"
return deleted
class _HarnessProxy:
"""Resolve the harness state against the current environment on every access.
Session env vars may be applied after import, so a state bound at import
time could freeze an env-less resolution. Resolution must never raise (a
failure inside the kernel namespace would take down the kernel). When the
local store is genuinely unconfigured (no session env, e.g. --no-session)
reads see an empty view but local writes raise instructively instead of
vanishing on kernel exit; any other resolution failure degrades to a shared
in-memory store until local resolution starts succeeding.
"""
_fallback: HarnessState | None = None
_unpersisted: HarnessState | None = None
def _resolve(self) -> HarnessState:
try:
return get_harness_state()
except RuntimeError as exc:
if "Local harness state requires" in str(exc):
if _HarnessProxy._unpersisted is None:
_HarnessProxy._unpersisted = HarnessState(
in_memory=True,
local_write_error=(
f"{exc} This session has no persistent local harness store; "
"pass global_=True to persist across sessions."
),
)
return _HarnessProxy._unpersisted
return self._degraded()
except Exception: # pragma: no cover - harness access must never raise
return self._degraded()
@staticmethod
def _degraded() -> HarnessState:
if _HarnessProxy._fallback is None:
_HarnessProxy._fallback = HarnessState(in_memory=True)
return _HarnessProxy._fallback
def __getattr__(self, name: str) -> Any:
return getattr(self._resolve(), name)
def __repr__(self) -> str:
return repr(self._resolve())
_harness_state = _HarnessProxy()
class _RLMCallable:
harness = _harness_state
get_harness_state = staticmethod(get_harness_state)
async def run(self, prompt: str, **kwargs: Any) -> RLMSubagent:
return await run(prompt, **kwargs)
async def find_models(self, query: str = "", limit: int = 8) -> list[RLMModel]:
return await find_models(query, limit)
async def list_subagents(self) -> list[RLMSubagent]:
return await list_subagents()
async def family(self) -> dict[str, Any]:
return await family()
async def send_parent(self, message: str) -> dict[str, Any]:
return await send_parent(message)
async def send_sibling(self, name: str, message: str) -> dict[str, Any]:
return await send_sibling(name, message)
async def send_child(self, name: str, message: str) -> dict[str, Any]:
return await send_child(name, message)
async def observe(self, target: str, limit: int = 8, max_chars: int = 800) -> dict[str, Any]:
return await observe(target, limit, max_chars)
async def info(self, target: str) -> dict[str, Any]:
return await agent_info(target)
async def submit_result(self, value: dict[str, Any]) -> RLMResultSubmission:
return await submit_result(value)
async def get_result(
self,
target: str | RLMSubagent,
fields: list[str] | tuple[str, ...] | None = None,
generation: int | None = None,
) -> RLMResult:
return await get_result(target, fields, generation)
async def delete_subagent(self, target: str | RLMSubagent) -> RLMSubagent:
return await delete_subagent(target)
async def __call__(self, prompt: str, **kwargs: Any) -> RLMSubagent:
return await run(prompt, **kwargs)
rlm = _RLMCallable()
harness = _harness_state
class _CallableModule(types.ModuleType):
async def __call__(self, prompt: str, **kwargs: Any) -> RLMSubagent:
return await run(prompt, **kwargs)
sys.modules[__name__].__class__ = _CallableModule
__all__ = [
"BashHandle",
"BashResult",
"HarnessEntry",
"HarnessScope",
"HarnessState",
"McpIntegration",
"McpToolError",
"NotEnabled",
"RLMModel",
"RLMResult",
"RLMResultSubmission",
"RLMSubagent",
"RefinementEvent",
"bash",
"agent_info",
"delete_subagent",
"emit",
"family",
"find_models",
"get_harness_state",
"get_result",
"harness",
"host_request",
"list_subagents",
"observe",
"rlm",
"run",
"send_child",
"send_parent",
"send_sibling",
"submit_result",
]
# Lazily re-export the MCP base class. Kept lazy so `import rlm` never requires
# the optional `mcp` SDK — only integration packages that subclass it do.
_LAZY_MCP = {"McpIntegration", "McpToolError", "NotEnabled"}
def __getattr__(name: str) -> Any: # noqa: D401 - module-level lazy attr hook
if name in _LAZY_MCP:
from . import mcp_base
return getattr(mcp_base, name)
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")