Skip to content
Merged
Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
b9c1577
feat: support persistent execution workers outside the agent loop
hallerite Oct 7, 2026
ab4343b
fix: bound execution output and terminate flooding commands
hallerite Oct 8, 2026
aff5203
feat: proxy workspace tools over pluggable execution transports
hallerite Oct 13, 2026
c214953
fix: refuse rlm.services in split execution
hallerite Oct 29, 2026
762e0d4
fix: stop kernel commands with the kernel and keep worker control off…
hallerite Oct 29, 2026
665c113
fix: find a kernel's commands by an environment token
hallerite Oct 30, 2026
ff8767d
feat: announce the TCP worker's bound port
hallerite Oct 30, 2026
7cbe516
fix: stop native shell commands with their turn and session
hallerite Oct 31, 2026
d98bc27
fix: sync the session ledger to the execution worker incrementally
hallerite Oct 31, 2026
283d136
test: drive the split-execution checks through the engine
hallerite Oct 31, 2026
365b8f6
feat: run the supervisor's shell jobs, services and path watches in t…
hallerite Oct 31, 2026
9340d00
fix: keep a remote job's events that arrive before it is registered
hallerite Oct 31, 2026
9184e4a
fix: send split execution frames without Nagle delay
hallerite Oct 31, 2026
efac509
perf: report a short job's output and exit together under split execu…
hallerite Oct 31, 2026
6588547
fix: keep a long output's start and end instead of stopping it at 1 MiB
hallerite Nov 1, 2026
2e4629b
fix: show no more of a long output's ends than were kept
hallerite Nov 1, 2026
023b3e9
fix: stop what is attached to the kernel, and leave daemons running
hallerite Nov 1, 2026
3f76141
fix: give bash commands an empty stdin
hallerite Nov 1, 2026
89ad826
fix: leave a bash command's background job running at its timeout
hallerite Nov 1, 2026
4b0f6cf
fix: return a child's history in pages that each fit a frame
hallerite Nov 1, 2026
220db7d
docs: draw split execution, and say what runs in the workspace
hallerite Nov 1, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 40 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -550,3 +550,43 @@ stops with a failure event when the inbox event limit is reached. There are no
model-authored callbacks, event selectors, or output predicates.

Activity subscriptions become `completed` after their target permanently terminates, flushing pending activity and releasing their active slot. Watches of idle persistent agents remain active. Watching an already-finished target returns a completed subscription.

### Separate tool execution

The ACP runtime metadata accepts `execution_command` (a trusted argv launching
`python -m rlm.execution`) and `execution_cwd` (the task workspace). For example,
an orchestrator can start the worker with `docker exec -i <task-container>
/path/to/python -m rlm.execution`. Both processes must have the same nano-rlm
version installed. The orchestrator owns container provisioning and teardown.

The engine, model credentials, supervisor, and authoritative conversation logs
remain in the harness process. Native tools, persistent IPython kernels, and builtin skills run
in the worker environment. Recursive agents inherit the execution destination,
keep separate kernels, and share the task workspace. Communication and recursive
callbacks use worker stdio, so a task container can have networking disabled.
The command itself defines the isolation boundary; launching a local Python
worker alone provides process separation, not filesystem isolation.

ACP initialization advertises `ai.prime.rlm/execution-v1` with the supported
transports (`stdio`, `tcp`) and tools (`ipython`, `bash`, `edit`, `fetch`). Tool
calls cross the boundary as a name and arguments; filesystem operations happen
inside the worker. `ExecutionTransport` supplies `connect()` (reader and writer)
and `close()` independently of tool dispatch.

For a trusted private TCP network, launch `python -m rlm.execution --listen
0.0.0.0 --port 8765` with a random `RLM_EXECUTION_TOKEN` of at least 32 characters.
With `--port 0` the kernel picks a free port, which the worker prints on stderr as
`listening on HOST:PORT` once it accepts connections.
Set `execution_tcp` to `{"host": "worker-address", "port": 8765, "token": "..."}`
instead of `execution_command`. Each connection owns a separate execution session.
The token authenticates connections; this transport does not encrypt traffic, so
use a private network or an encrypted tunnel. The worker removes its token from
the environment before starting tools. The orchestrator owns listener teardown.

Supervisor shell jobs
(`rlm.shell`), services (`rlm.services`) and filesystem watchers
(`rlm.watch.path`) return explicit errors in this mode. `await child.history()` uses the supervisor; the current agent's local
history is a snapshot refreshed before each cell. Worker frames, including history
snapshots and outputs, are limited to 16 MiB. Kernel recovery preserves the existing
no-replay behavior. Worker disconnects fail the session; the orchestrator must stop
remaining task processes when the rollout ends.
63 changes: 40 additions & 23 deletions src/rlm/acp.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
from rlm.engine import RLMEngine
from rlm.config import (
ExecutionPolicy,
TcpExecutionConfig,
InvocationContext,
ProviderConfig,
RuntimeConfig,
Expand Down Expand Up @@ -75,6 +76,9 @@ class _RuntimeMetadata(_ContractModel):
leaf_append_to_system_prompt: str | None = None
skills: list[Annotated[str, Field(min_length=1)]]
builtin_tools: list[Annotated[str, Field(min_length=1)]] | None = None
execution_tcp: TcpExecutionConfig | None = None
execution_cwd: str | None = None
execution_command: list[Annotated[str, Field(min_length=1)]] | None = None
kernel_env: dict[str, str]
search_api_key: str | None
network_block: list[str] = Field(default_factory=list)
Expand Down Expand Up @@ -189,6 +193,33 @@ def _runtime_config(meta_kwargs: Any) -> tuple[RuntimeConfig, str]:
payload = _RuntimeMetadata.model_validate(
meta_kwargs.get(RUNTIME_METADATA_KEY),
)
result = (
RuntimeConfig(
model=payload.model,
provider=payload.provider,
invocation=InvocationContext(),
policy=payload.policy,
system_prompt_path=payload.system_prompt_path,
append_to_system_prompt=payload.append_to_system_prompt,
subagent_append_to_system_prompt=payload.subagent_append_to_system_prompt,
leaf_append_to_system_prompt=payload.leaf_append_to_system_prompt,
skills=tuple(payload.skills),
builtin_tools=(
tuple(payload.builtin_tools)
if payload.builtin_tools is not None
else None
),
kernel_env=tuple(payload.kernel_env.items()),
execution_tcp=payload.execution_tcp,
execution_cwd=payload.execution_cwd,
execution_command=tuple(payload.execution_command)
if payload.execution_command is not None
else None,
search_api_key=payload.search_api_key,
network_block=tuple(payload.network_block),
),
payload.session_id,
)
except ValidationError as error:
raise RequestError.invalid_params(
{
Expand All @@ -199,28 +230,7 @@ def _runtime_config(meta_kwargs: Any) -> tuple[RuntimeConfig, str]:
}
) from error

return (
RuntimeConfig(
model=payload.model,
provider=payload.provider,
invocation=InvocationContext(),
policy=payload.policy,
system_prompt_path=payload.system_prompt_path,
append_to_system_prompt=payload.append_to_system_prompt,
subagent_append_to_system_prompt=payload.subagent_append_to_system_prompt,
leaf_append_to_system_prompt=payload.leaf_append_to_system_prompt,
skills=tuple(payload.skills),
builtin_tools=(
tuple(payload.builtin_tools)
if payload.builtin_tools is not None
else None
),
kernel_env=tuple(payload.kernel_env.items()),
search_api_key=payload.search_api_key,
network_block=tuple(payload.network_block),
),
payload.session_id,
)
return result


def _mcp_servers(
Expand Down Expand Up @@ -293,7 +303,14 @@ async def initialize(
),
),
agent_info=Implementation(name="rlm", title="RLM", version=version("rlm")),
field_meta={CONTRACT_METADATA_KEY: True, "steering": {"supported": True}},
field_meta={
CONTRACT_METADATA_KEY: True,
"steering": {"supported": True},
"ai.prime.rlm/execution-v1": {
"transports": ["stdio", "tcp"],
"tools": ["ipython", "bash", "edit", "fetch"],
},
},
)

async def new_session(
Expand Down
5 changes: 4 additions & 1 deletion src/rlm/agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,10 @@ async def info(self) -> AgentInfo:

async def history(self) -> History:
"""Read a fresh snapshot of this agent's local conversation history."""
return await history(session_dir=self.session_dir)
if not broker.is_active():
return await history(session_dir=self.session_dir)
payload = await broker.agent_request("agent.history", agent_id=self.id)
return History.from_events(self.session_dir, payload)

async def result(self, *, yield_after: float = 300) -> AgentResult:
"""Wait up to yield_after seconds for the child to finish, then return its latest state.
Expand Down
7 changes: 6 additions & 1 deletion src/rlm/broker.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ class BrokerAgentListRequest(TypedDict):

class BrokerAgentHandleRequest(TypedDict):
__pydantic_config__ = ConfigDict(extra="forbid", strict=True)
op: Literal["agent.info", "agent.cancel"]
op: Literal["agent.info", "agent.cancel", "agent.history"]
capability: Annotated[str, Field(min_length=1)]
scope_id: Annotated[str, Field(min_length=1)]
agent_id: Annotated[str, Field(min_length=1)]
Expand Down Expand Up @@ -403,6 +403,11 @@ def result_from_payload(value: dict[str, Any]) -> AgentResult:
raise RuntimeError("invalid response from RLM supervisor") from None


def is_active() -> bool:
"""Whether the current process is inside a brokered execution scope."""
return _endpoint is not None and _scope_id is not None


def configure(endpoint: BrokerEndpoint | None) -> None:
global _endpoint
_endpoint = endpoint
Expand Down
29 changes: 28 additions & 1 deletion src/rlm/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@

from __future__ import annotations

from pydantic import BaseModel, ConfigDict, Field, field_validator
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

from rlm.network import NetworkBlocklist
from rlm.semantic import ACP_EXTENSION_HEADER_NAMES
Expand Down Expand Up @@ -96,6 +96,14 @@ class ExecutionPolicy(_ConfigModel):
allow_git: bool = False


class TcpExecutionConfig(_ConfigModel):
"""Authenticated worker endpoint on a trusted private network."""

host: str = Field(min_length=1)
port: int = Field(gt=0, le=65535)
token: str = Field(min_length=32, repr=False)


class RuntimeConfig(_ConfigModel):
"""Configuration resolved once at an RLM process boundary."""

Expand All @@ -112,10 +120,29 @@ class RuntimeConfig(_ConfigModel):
"""Builtin tool set for every engine in the tree; None = the registry default
(`ipython` alone). Validated against the registry when the engine starts."""
kernel_env: tuple[tuple[str, str], ...] = Field(default=(), repr=False)
execution_cwd: str | None = None
execution_command: tuple[str, ...] | None = None
"""Trusted argv launching an rlm.execution worker in the task environment."""
execution_tcp: TcpExecutionConfig | None = None
search_api_key: str | None = Field(default=None, repr=False)
network_block: NetworkBlocklist = ()
"""Search-result destination rules shared by every agent in the session tree."""

@model_validator(mode="after")
def _validate_execution(self):
if self.execution_command is not None:
if not self.execution_command or any(
not arg for arg in self.execution_command
):
raise ValueError("execution_command must contain nonempty arguments")
if self.execution_tcp is not None:
raise ValueError("choose execution_command or execution_tcp")
return self

@property
def split_execution(self) -> bool:
return self.execution_command is not None or self.execution_tcp is not None

@property
def resolved_append_to_system_prompt(self) -> str | None:
"""The append for this engine's role in the session tree.
Expand Down
33 changes: 27 additions & 6 deletions src/rlm/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,7 @@ def __init__(
self.runtime_config = runtime_config
config = self.runtime_config
self.model = config.model
self.cwd = cwd or os.getcwd()
self.cwd = config.execution_cwd or cwd or os.getcwd()
self.exec_timeout = config.policy.exec_timeout
self.max_total_turns = config.policy.max_total_turns
self.max_tool_output_bytes = config.policy.max_tool_output_bytes
Expand Down Expand Up @@ -520,6 +520,10 @@ async def _start(self, prompt: str) -> None:
self._active_tools = get_active_builtin_tools(
self.exec_timeout, self.builtin_tools
)
if self.runtime_config.split_execution:
from rlm.execution import validate_tools

validate_tools(tool.name for tool in self._active_tools)
self._active_tool_schemas = [tool.schema() for tool in self._active_tools]

if self.compaction and self.summarize_at_tokens is None:
Expand Down Expand Up @@ -579,7 +583,17 @@ async def _start(self, prompt: str) -> None:
self._owns_supervisor = False
raise

self._repl = IPythonREPL(
repl_type = IPythonREPL
repl_options = {}
if self.runtime_config.split_execution:
from rlm.execution import RemoteREPL

repl_type = RemoteREPL
repl_options["command"] = self.runtime_config.execution_command
repl_options["tcp"] = self.runtime_config.execution_tcp
repl_options["tools"] = [tool.name for tool in self._active_tools]
self._repl = repl_type(
**repl_options,
cwd=self.cwd,
session=self.session,
kernel_env=self.kernel_env,
Expand Down Expand Up @@ -924,9 +938,14 @@ async def _run_loop(self) -> RLMResult:
if scope_id is not None:
repl.set_broker_scope(scope_id)
run = repl.run_in_thread if repl is not None else asyncio.to_thread
tool_task = asyncio.create_task(
run(tool.execute, tool_args, self._tool_context(messages))
)
if self.runtime_config.split_execution:
tool_task = asyncio.create_task(
run(repl.execute_tool, tool.name, tool_args)
)
else:
tool_task = asyncio.create_task(
run(tool.execute, tool_args, self._tool_context(messages))
)
try:
tool_result = await asyncio.shield(tool_task)
except asyncio.CancelledError:
Expand Down Expand Up @@ -1481,7 +1500,8 @@ def _load_system_prompt(self, active_tools: list[BuiltinTool]) -> str:
self.cwd,
discover_skills(self.session.dir),
depth=self.depth,
session_dir=str(self.session.dir),
Comment thread
cursor[bot] marked this conversation as resolved.
Resolved
session_dir=getattr(self._repl, "session_dir", None)
or str(self.session.dir),
allow_recursion=self.depth < self.max_depth,
delegation_prompt=self.runtime_config.policy.delegation_prompt,
allow_git=self.allow_git,
Expand All @@ -1495,6 +1515,7 @@ def _load_system_prompt(self, active_tools: list[BuiltinTool]) -> str:
if self.system_prompt_path
else None,
extra_instructions=self.append_to_system_prompt,
split_execution=self.runtime_config.split_execution,
agent_info=self._supervisor.agent_context(self._invocation_id)
if self._supervisor
else None,
Expand Down
Loading
Loading