--help`. Other listed Python tools are IPython-only."
)
if _should_include_git_history_guard(active_tools, allow_git):
- parts.append(GIT_HISTORY_GUARD_PROMPT)
+ parts.append(
+ GIT_HISTORY_GUARD_PROMPT
+ if network_guidance
+ else OFFLINE_GIT_HISTORY_GUARD_PROMPT
+ )
return "\n\n".join(parts)
diff --git a/tests/e2e/__init__.py b/tests/e2e/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py
new file mode 100644
index 00000000..9f182685
--- /dev/null
+++ b/tests/e2e/conftest.py
@@ -0,0 +1,364 @@
+"""End-to-end runs against a hosted model.
+
+Each test starts `rlm --acp` the way an orchestrator does, gives it a task in a
+throwaway workspace, and checks what the agent left there: files, its ledger and the
+session snapshot. Where the model could get an answer without the feature under test,
+the answer exists only behind that feature (a stamp only a tool call returns, a
+codeword only a skill holds, a token only a sub-agent's history shows), so a pass proves
+the feature ran. The suite runs only with `-m e2e`, needs `PRIME_API_KEY` (plus
+`PRIME_TEAM_ID` for a team's models), and skips without it. `RLM_E2E_MODEL` and
+`RLM_E2E_BASE_URL` choose the model.
+
+Every test runs in three placements: `local` (tools in the agent's process tree),
+`stdio` (a worker the agent starts, over its stdin and stdout) and `tcp` (a worker
+over TCP, started on its own with no credentials in its environment). The last two
+are split execution.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import json
+import os
+import re
+import secrets
+import subprocess
+import sys
+from collections.abc import AsyncIterator, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import pytest
+from acp import PROTOCOL_VERSION, spawn_agent_process, text_block
+
+from rlm.acp import (
+ ACP_SEMANTIC_EDGES_METADATA_KEY,
+ RUNTIME_METADATA_KEY,
+ SESSION_METADATA_KEY,
+)
+
+MODEL = os.environ.get("RLM_E2E_MODEL") or "internal/glm-5.3-fast"
+BASE_URL = os.environ.get("RLM_E2E_BASE_URL") or "https://api.pinference.ai/api/v1"
+MODES = ("local", "stdio", "tcp")
+PROMPT_TIMEOUT = 600
+
+
+def pytest_collection_modifyitems(items):
+ for item in items:
+ if "tests/e2e/" in item.nodeid.replace(os.sep, "/"):
+ item.add_marker(pytest.mark.e2e)
+
+
+@pytest.fixture(scope="session", autouse=True)
+def install_fixture_skills():
+ """The agent sees the skills a real install has, not the unit tests' fixtures."""
+ yield
+
+
+@pytest.fixture(params=MODES)
+def mode(request) -> str:
+ return request.param
+
+
+@dataclass
+class Turn:
+ stop_reason: str
+ answer: str
+ meta: dict[str, Any]
+
+
+class _Client:
+ def __init__(self) -> None:
+ self.messages: list[str] = []
+
+ async def session_update(self, session_id: str, update: Any, **kwargs: Any) -> None:
+ content = getattr(update, "content", None)
+ if getattr(update, "session_update", None) == "agent_message_chunk" and content:
+ self.messages.append(getattr(content, "text", ""))
+
+
+class AgentSession:
+ """One ACP session: prompt, steer or cancel it while it lives, then read what it
+ left behind."""
+
+ def __init__(
+ self,
+ connection: Any,
+ session_id: str,
+ client: _Client,
+ workspace: Path,
+ home: Path,
+ ) -> None:
+ self._connection = connection
+ self._client = client
+ self.session_id = session_id
+ self.workspace = workspace
+ self.home = home
+ self.turns: list[Turn] = []
+ self.running: asyncio.Task[Turn] | None = None
+ self._calls_before_running = 0
+ self.closed_meta: dict[str, Any] | None = None
+
+ async def prompt(self, text: str) -> Turn:
+ """Run one prompt to its end."""
+ self._client.messages.clear()
+ response = await asyncio.wait_for(
+ self._connection.prompt(
+ session_id=self.session_id, prompt=[text_block(text)]
+ ),
+ PROMPT_TIMEOUT,
+ )
+ turn = Turn(
+ response.stop_reason,
+ "".join(self._client.messages),
+ response.field_meta or {},
+ )
+ self.turns.append(turn)
+ return turn
+
+ def start(self, text: str) -> asyncio.Task[Turn]:
+ """Start a prompt and return while it runs."""
+ self._calls_before_running = len(self.tool_calls())
+ self.running = asyncio.create_task(self.prompt(text))
+ return self.running
+
+ async def called(self, name: str, timeout: float = 300) -> None:
+ """Wait until the running prompt's agent has called the tool `name`."""
+ deadline = asyncio.get_running_loop().time() + timeout
+ while name not in self.tool_calls()[self._calls_before_running :]:
+ assert self.running is not None and not self.running.done(), (
+ f"the prompt ended without calling {name}: {self.describe()}"
+ )
+ assert asyncio.get_running_loop().time() < deadline, (
+ f"no {name} call within {timeout}s: {self.describe()}"
+ )
+ await asyncio.sleep(0.2)
+
+ async def steer(self, text: str) -> dict[str, Any]:
+ return await self._connection.ext_method(
+ "session/steering",
+ {"sessionId": self.session_id, "prompt": [{"type": "text", "text": text}]},
+ )
+
+ async def cancel(self) -> None:
+ await self._connection.cancel(session_id=self.session_id)
+
+ @property
+ def session_dir(self) -> Path:
+ (session_dir,) = (self.home / "sessions").iterdir()
+ return session_dir
+
+ @property
+ def records(self) -> list[dict[str, Any]]:
+ """The root agent's ledger as written so far."""
+ return self.ledger(self.session_dir)
+
+ def children(self) -> list[Path]:
+ """Session directories of the root agent's sub-agents."""
+ return sorted(
+ path for path in self.session_dir.iterdir() if path.name.startswith("sub-")
+ )
+
+ @staticmethod
+ def ledger(session_dir: Path) -> list[dict[str, Any]]:
+ path = session_dir / "messages.jsonl"
+ if not path.exists():
+ return []
+ records = []
+ for line in path.read_text().splitlines():
+ with contextlib.suppress(
+ json.JSONDecodeError
+ ): # a line still being written
+ records.append(json.loads(line))
+ return records
+
+ @property
+ def snapshot(self) -> dict[str, Any]:
+ """The authoritative snapshot `session/close` returned."""
+ assert self.closed_meta is not None, "the session is still open"
+ return self.closed_meta[SESSION_METADATA_KEY]
+
+ @property
+ def answer(self) -> str:
+ return self.turns[-1].answer if self.turns else ""
+
+ @property
+ def stop_reason(self) -> str:
+ return self.turns[-1].stop_reason if self.turns else ""
+
+ def read(self, name: str) -> str:
+ """A file the agent left in the workspace, stripped; empty when it left none."""
+ path = self.workspace / name
+ return path.read_text().strip() if path.exists() else ""
+
+ def tool_calls(self) -> list[str]:
+ """Names of the tools the root agent called, in order."""
+ return [call["function"]["name"] for call in self.tool_call_records()]
+
+ def tool_call_records(self) -> list[dict[str, Any]]:
+ return [
+ call
+ for record in self.records
+ if record.get("type") == "assistant"
+ for call in record["message"].get("tool_calls") or ()
+ ]
+
+ def tool_results(self) -> list[str]:
+ """What the root agent's tool calls returned, in order."""
+ return [
+ record["content"]
+ for record in self.records
+ if record.get("type") == "tool_result"
+ ]
+
+ def edges(self) -> set[str]:
+ """Types of the semantic edges the session published."""
+ metas = [turn.meta for turn in self.turns] + [self.closed_meta or {}]
+ return {
+ edge["type"]
+ for meta in metas
+ for edge in (meta.get(ACP_SEMANTIC_EDGES_METADATA_KEY) or {}).get(
+ "edges", ()
+ )
+ }
+
+ def describe(self) -> str:
+ stops = [turn.stop_reason for turn in self.turns]
+ return (
+ f"stops {stops}; tools {self.tool_calls()}; answer {self.answer[-500:]!r}"
+ )
+
+
+def _provider() -> dict[str, Any]:
+ headers = {}
+ if os.environ.get("PRIME_TEAM_ID"):
+ headers["X-Prime-Team-ID"] = os.environ["PRIME_TEAM_ID"]
+ return {
+ "base_url": BASE_URL,
+ "api_key": os.environ["PRIME_API_KEY"],
+ "headers": headers,
+ "max_retries": 5,
+ }
+
+
+def _start_tcp_worker(workspace: Path, token: str) -> subprocess.Popen:
+ """A worker the way a remote workspace runs one: its own process, its own
+ environment, none of the agent's credentials."""
+ env = {
+ key: value
+ for key, value in os.environ.items()
+ if not key.startswith(("PRIME_", "RLM_E2E_"))
+ }
+ env["RLM_EXECUTION_TOKEN"] = token
+ return subprocess.Popen(
+ [sys.executable, "-m", "rlm.execution", "--listen", "127.0.0.1", "--port", "0"],
+ cwd=workspace,
+ env=env,
+ stderr=subprocess.PIPE,
+ text=True,
+ )
+
+
+async def _listening_port(worker: subprocess.Popen) -> int:
+ line = await asyncio.wait_for(asyncio.to_thread(worker.stderr.readline), 60)
+ match = re.search(r"listening on \S+:(\d+)", line)
+ if not match:
+ raise RuntimeError(f"execution worker did not start: {line!r}")
+ return int(match.group(1))
+
+
+@pytest.fixture
+def agent(tmp_path, mode):
+ """Open a session on a fresh workspace, in this test's placement. Closing it
+ records the final snapshot."""
+ if not os.environ.get("PRIME_API_KEY"):
+ pytest.skip("needs PRIME_API_KEY")
+ workers: list[subprocess.Popen] = []
+
+ @contextlib.asynccontextmanager
+ async def open_session(
+ *,
+ files: dict[str, str] | None = None,
+ builtin_tools: list[str] | None = None,
+ skills: tuple[str, ...] = ("edit",),
+ skill_docs: dict[str, str] | None = None,
+ mcp_servers: Sequence[Any] = (),
+ policy: dict[str, Any] | None = None,
+ ) -> AsyncIterator[AgentSession]:
+ workspace = tmp_path / "workspace"
+ workspace.mkdir()
+ for name, text in (files or {}).items():
+ (workspace / name).write_text(text)
+ home = tmp_path / "rlm"
+ for name, text in (skill_docs or {}).items():
+ (home / "skills" / name).mkdir(parents=True)
+ (home / "skills" / name / "SKILL.md").write_text(text)
+ runtime: dict[str, Any] = {
+ "session_id": "e2e",
+ "model": MODEL,
+ "provider": _provider(),
+ "policy": {"max_depth": 1, "exec_timeout": 120, **(policy or {})},
+ "system_prompt_path": None,
+ "append_to_system_prompt": None,
+ "skills": list(skills),
+ "builtin_tools": builtin_tools,
+ "kernel_env": {},
+ "search_api_key": None,
+ }
+ if mode == "stdio":
+ runtime["execution_command"] = [sys.executable, "-m", "rlm.execution"]
+ runtime["execution_cwd"] = str(workspace)
+ elif mode == "tcp":
+ token = secrets.token_hex(32)
+ worker = _start_tcp_worker(workspace, token)
+ workers.append(worker)
+ port = await _listening_port(worker)
+ runtime["execution_tcp"] = {
+ "host": "127.0.0.1",
+ "port": port,
+ "token": token,
+ }
+ runtime["execution_cwd"] = str(workspace)
+ client = _Client()
+ executable = str(Path(sys.executable).parent / "rlm")
+ env = {**os.environ, "RLM_HOME": str(home)}
+ async with spawn_agent_process(client, executable, "--acp", env=env) as (
+ connection,
+ _process,
+ ):
+ await connection.initialize(PROTOCOL_VERSION)
+ created = await connection.new_session(
+ cwd=str(workspace),
+ mcp_servers=list(mcp_servers),
+ **{RUNTIME_METADATA_KEY: runtime},
+ )
+ session = AgentSession(
+ connection, created.session_id, client, workspace, home
+ )
+ try:
+ yield session
+ finally:
+ if session.running is not None and not session.running.done():
+ session.running.cancel()
+ await asyncio.gather(session.running, return_exceptions=True)
+ closed = await connection.close_session(created.session_id)
+ session.closed_meta = closed.field_meta
+
+ yield open_session
+ for worker in workers:
+ worker.kill()
+ worker.wait()
+
+
+@pytest.fixture
+def run_agent(agent):
+ """Run one prompt to its end in a fresh session."""
+
+ async def run(prompt: str, **options: Any) -> AgentSession:
+ async with agent(**options) as session:
+ await session.prompt(prompt)
+ return session
+
+ return run
diff --git a/tests/e2e/fixtures/desk_mcp.py b/tests/e2e/fixtures/desk_mcp.py
new file mode 100644
index 00000000..0e15cd94
--- /dev/null
+++ b/tests/e2e/fixtures/desk_mcp.py
@@ -0,0 +1,25 @@
+"""A stdio MCP server whose results only a real call produces: `echo` stamps what it
+returns, and `lookup` fails with a token in its error."""
+
+from mcp.server.fastmcp import FastMCP
+
+ECHO_STAMP = "ok-7f3"
+ERROR_TOKEN = "err-5c2"
+
+server = FastMCP("desk")
+
+
+@server.tool()
+def echo(message: str) -> str:
+ """Echo the message back, stamped so the caller can prove the tool ran."""
+ return f"{message} [{ECHO_STAMP}]"
+
+
+@server.tool()
+def lookup(order_id: str) -> str:
+ """Look up an order by its id."""
+ raise ValueError(f"Order {order_id} not found [{ERROR_TOKEN}]")
+
+
+if __name__ == "__main__":
+ server.run()
diff --git a/tests/e2e/test_agents.py b/tests/e2e/test_agents.py
new file mode 100644
index 00000000..57d93a74
--- /dev/null
+++ b/tests/e2e/test_agents.py
@@ -0,0 +1,67 @@
+"""The supervisor's API from a cell: sub-agents, their history, persistent agents, and
+background jobs the agent waits on."""
+
+from __future__ import annotations
+
+import json
+import re
+
+
+async def test_subagent_does_part_of_the_work(run_agent):
+ outcome = await run_agent(
+ "Spawn a sub-agent with rlm.agent.spawn and have it write the sum of the integers "
+ "from 1 to 100 to child.txt. Wait for it to finish, then write the word done to "
+ "answer.txt."
+ )
+ assert outcome.read("child.txt") == "5050", outcome.describe()
+ assert outcome.read("answer.txt") == "done", outcome.describe()
+ assert outcome.children(), outcome.describe()
+ assert "subagent_call" in outcome.edges(), outcome.edges()
+
+
+async def test_parent_reads_a_childs_history(run_agent):
+ outcome = await run_agent(
+ "Spawn a sub-agent whose task is: 'Run `import secrets; print(secrets.token_hex(4))` "
+ "in ipython, then answer with only the word done.' Its answer will not contain the "
+ "token, so once it finishes, read its conversation with `await child.history()` and "
+ "write the token it printed to answer.txt."
+ )
+ (child,) = outcome.children()
+ printed = {
+ token
+ for record in outcome.ledger(child)
+ if record.get("type") == "tool_result"
+ for token in re.findall(r"\b[0-9a-f]{8}\b", record["content"])
+ }
+ assert outcome.read("answer.txt") in printed, (printed, outcome.describe())
+ assert any(
+ "history(" in json.loads(call["function"]["arguments"]).get("code", "")
+ for call in outcome.tool_call_records()
+ if call["function"]["name"] == "ipython"
+ ), outcome.describe()
+
+
+async def test_persistent_agent_takes_a_second_instruction(run_agent):
+ outcome = await run_agent(
+ "Spawn a persistent sub-agent with `rlm.agent.spawn(task=..., name='scribe', "
+ "persistent=True)` whose task is to write the word alpha to a.txt, and wait for its "
+ "result. Then give the same agent a second instruction with `await scribe.send(...)`: "
+ "write the word beta to b.txt. Wait for that result too, then write the word done "
+ "to answer.txt."
+ )
+ assert outcome.read("a.txt") == "alpha", outcome.describe()
+ assert outcome.read("b.txt") == "beta", outcome.describe()
+ assert outcome.read("answer.txt") == "done", outcome.describe()
+ assert len(outcome.children()) == 1, outcome.children()
+
+
+async def test_wait_returns_when_a_background_job_ends(run_agent):
+ outcome = await run_agent(
+ "Start the command `sleep 10; echo finished` as a background job with "
+ "`await rlm.shell.run(command, yield_after=0)`. Do not poll it: call the wait tool, "
+ "which returns when the job completes. Then write the job's output to answer.txt."
+ )
+ assert outcome.read("answer.txt") == "finished", outcome.describe()
+ assert (
+ "New supervisor events or instructions are available." in outcome.tool_results()
+ ), outcome.tool_results()
diff --git a/tests/e2e/test_session.py b/tests/e2e/test_session.py
new file mode 100644
index 00000000..38e0e58e
--- /dev/null
+++ b/tests/e2e/test_session.py
@@ -0,0 +1,94 @@
+"""The session as a client drives it: several prompts on one kernel, steering into a
+waiting agent, cancelling a turn, tool timeouts, and the budgets that end a run."""
+
+from __future__ import annotations
+
+import asyncio
+import time
+
+import pytest
+
+
+async def test_second_prompt_continues_the_session(agent):
+ async with agent(files={"secret.txt": "marmalade\n"}) as session:
+ await session.prompt(
+ "In ipython, store the contents of secret.txt, stripped, in a variable named "
+ "word, then delete secret.txt. Do not print word."
+ )
+ assert not (session.workspace / "secret.txt").exists(), session.describe()
+ await session.prompt("Write the value of your variable word to answer.txt.")
+ assert session.read("answer.txt") == "marmalade", session.describe()
+ assert [turn.stop_reason for turn in session.turns] == ["end_turn", "end_turn"]
+ assert "continuation" in session.edges(), session.edges()
+
+
+async def test_steering_wakes_a_waiting_agent(agent):
+ async with agent() as session:
+ running = session.start(
+ "Before doing anything else, call the wait tool and keep waiting until the user "
+ "sends you a message. It contains a code word: write the code word to answer.txt."
+ )
+ await session.called("wait")
+ assert await session.steer("The code word is heron-913.") == {
+ "outcome": "injected"
+ }
+ await running
+ assert session.read("answer.txt") == "heron-913", session.describe()
+ assert "A new user message is available." in session.tool_results()
+
+
+async def test_cancelled_turn_leaves_the_session_usable(agent):
+ async with agent(policy={"exec_timeout": 300}) as session:
+ running = session.start(
+ "Run `import time; time.sleep(300)` in one ipython call."
+ )
+ await session.called("ipython")
+ await asyncio.sleep(3) # let the cell start sleeping
+ started = time.monotonic()
+ await session.cancel()
+ assert (await running).stop_reason == "cancelled"
+ await session.prompt("Write the word ok to answer.txt.")
+ elapsed = time.monotonic() - started
+ assert session.read("answer.txt") == "ok", session.describe()
+ # Left to run, the cell would hold the kernel for the rest of its 300 s.
+ assert elapsed < 120, f"cancel and second prompt took {elapsed:.0f}s"
+
+
+async def test_tool_timeout_is_reported_and_work_goes_on(run_agent):
+ outcome = await run_agent(
+ "Run `import time; time.sleep(60)` in one ipython call, without a timeout "
+ "argument. When the call comes back, write the word continued to answer.txt.",
+ policy={"exec_timeout": 5},
+ )
+ assert outcome.read("answer.txt") == "continued", outcome.describe()
+ assert any("timed out after 5s" in result for result in outcome.tool_results()), (
+ outcome.tool_results()
+ )
+
+
+CAPS = [
+ pytest.param(
+ {"max_total_turns": 1},
+ "max_total_turns",
+ "max_turn_requests",
+ id="max-total-turns",
+ ),
+ pytest.param(
+ {"max_total_tokens": 1}, "max_total_tokens", "max_tokens", id="max-total-tokens"
+ ),
+ pytest.param({"max_tokens": 1}, "token_budget", "max_tokens", id="max-tokens"),
+]
+
+
+@pytest.mark.parametrize("cap,last_stop_reason,acp_stop_reason", CAPS)
+async def test_budget_stops_the_run(run_agent, cap, last_stop_reason, acp_stop_reason):
+ """Each budget ends a task that wants three model calls after the first."""
+ outcome = await run_agent(
+ "Use three separate ipython calls: write 1 to a.txt, then 2 to b.txt, then 3 "
+ "to c.txt.",
+ policy=cap,
+ )
+ assert outcome.snapshot["last_stop_reason"] == last_stop_reason, outcome.describe()
+ assert outcome.stop_reason == acp_stop_reason, outcome.describe()
+ assert outcome.snapshot["turns"] == 1, outcome.snapshot
+ assert not outcome.read("b.txt"), outcome.describe()
diff --git a/tests/e2e/test_tasks.py b/tests/e2e/test_tasks.py
new file mode 100644
index 00000000..84ddfc5b
--- /dev/null
+++ b/tests/e2e/test_tasks.py
@@ -0,0 +1,108 @@
+"""Tasks a hosted model completes with nano-rlm, each checked by what it leaves behind."""
+
+from __future__ import annotations
+
+
+async def test_writes_files(run_agent):
+ outcome = await run_agent(
+ "Write the prime numbers below 50 to primes.txt, one per line, in ascending order."
+ )
+ assert outcome.read("primes.txt").split() == [
+ "2",
+ "3",
+ "5",
+ "7",
+ "11",
+ "13",
+ "17",
+ "19",
+ "23",
+ "29",
+ "31",
+ "37",
+ "41",
+ "43",
+ "47",
+ ], outcome.describe()
+
+
+async def test_kernel_keeps_state_between_calls(run_agent):
+ outcome = await run_agent(
+ "Use two separate ipython tool calls. In the first, read the integer in seed.txt "
+ "into a variable named seed and print nothing else. In the second, without "
+ "reading any file, write seed * 3 to answer.txt.",
+ files={"seed.txt": "1234\n"},
+ )
+ assert outcome.read("answer.txt") == "3702", outcome.describe()
+ assert outcome.tool_calls().count("ipython") >= 2, outcome.describe()
+
+
+async def test_edit_skill_fixes_a_bug(run_agent):
+ outcome = await run_agent(
+ "calc.py has a bug: add() should return the sum of its arguments. Fix it with "
+ "the edit skill, changing only the broken line.",
+ files={
+ "calc.py": "def add(a, b):\n return a - b\n\n\ndef double(x):\n return 2 * x\n"
+ },
+ )
+ source = outcome.read("calc.py")
+ namespace: dict = {}
+ exec(source, namespace)
+ assert namespace["add"](2, 3) == 5, outcome.describe()
+ assert "def double(x):\n return 2 * x" in source
+
+
+async def test_native_bash_tool(run_agent):
+ outcome = await run_agent(
+ "With the bash tool, count the lines of data.csv that contain 'error' and write "
+ "the count to answer.txt.",
+ files={
+ "data.csv": "".join(
+ f"{i},{'error' if i % 7 == 0 else 'ok'}\n" for i in range(1, 101)
+ )
+ },
+ builtin_tools=["ipython", "bash", "edit"],
+ )
+ assert outcome.read("answer.txt") == "14", outcome.describe()
+ assert "bash" in outcome.tool_calls(), outcome.describe()
+
+
+async def test_long_output_shows_its_end(run_agent):
+ outcome = await run_agent(
+ "Run `seq 1 300000` in a shell and look at its output. Write the last number in "
+ "that output to answer.txt."
+ )
+ assert outcome.read("answer.txt") == "300000", outcome.describe()
+
+
+async def test_service_answers_requests(run_agent):
+ outcome = await run_agent(
+ "Serve the working directory over HTTP with `python -m http.server` on a free "
+ "port, as a service that keeps running, then fetch /hello.txt from it and write "
+ "the response body to answer.txt.",
+ files={"hello.txt": "hello from the workspace\n"},
+ )
+ assert outcome.read("answer.txt") == "hello from the workspace", outcome.describe()
+
+
+async def test_kernel_recovers_after_it_dies(run_agent):
+ outcome = await run_agent(
+ "Run `import os; os._exit(1)` in the ipython tool; it kills the kernel. After "
+ "the kernel restarts, write the word recovered to answer.txt."
+ )
+ assert outcome.read("answer.txt") == "recovered", outcome.describe()
+ assert any(record.get("type") == "kernel_recovery" for record in outcome.records), (
+ outcome.describe()
+ )
+
+
+async def test_work_continues_past_a_compaction(run_agent):
+ outcome = await run_agent(
+ "Use three separate tool calls. First read the code word in secret.txt. Then run "
+ "`seq 1 3000` in a shell and print all of its output. Then write the code word "
+ "to answer.txt.",
+ files={"secret.txt": "marmalade\n"},
+ policy={"summarize_at_tokens": 8000},
+ )
+ assert outcome.read("answer.txt") == "marmalade", outcome.describe()
+ assert outcome.snapshot["metrics"]["num_compactions"] >= 1, outcome.describe()
diff --git a/tests/e2e/test_tools.py b/tests/e2e/test_tools.py
new file mode 100644
index 00000000..e5edb345
--- /dev/null
+++ b/tests/e2e/test_tools.py
@@ -0,0 +1,95 @@
+"""What a client equips the agent with: MCP servers, `SKILL.md` skills and the native
+fetch tool. Each answer exists only behind the feature."""
+
+from __future__ import annotations
+
+import http.server
+import sys
+import threading
+from pathlib import Path
+
+import pytest
+from acp.schema import McpServerStdio
+
+from .fixtures.desk_mcp import ECHO_STAMP, ERROR_TOKEN
+
+DESK = McpServerStdio(
+ name="desk",
+ command=sys.executable,
+ args=[str(Path(__file__).parent / "fixtures" / "desk_mcp.py")],
+ env=[],
+)
+
+CODEWORD = "amber-falcon-482"
+CODEWORD_SKILL = f"""---
+name: project-codeword
+description: Look up the project codeword. Use this whenever you are asked for the project codeword.
+---
+
+# Project codeword
+
+The project codeword is `{CODEWORD}`. Reply with it verbatim when asked.
+"""
+
+PAGE_STAMP = "pg-41c"
+
+
+async def test_mcp_tool(run_agent):
+ outcome = await run_agent(
+ "Find the echo tool of the desk MCP server with rlm.tools, call it with the "
+ "message 'hello world', and write exactly what it returns to answer.txt.",
+ mcp_servers=[DESK],
+ )
+ answer = outcome.read("answer.txt")
+ assert "hello world" in answer and ECHO_STAMP in answer, outcome.describe()
+
+
+async def test_mcp_tool_error_reaches_the_model(run_agent):
+ outcome = await run_agent(
+ "Call the lookup tool of the desk MCP server with the order id A-17, and write "
+ "the error message it gives to answer.txt.",
+ mcp_servers=[DESK],
+ )
+ assert ERROR_TOKEN in outcome.read("answer.txt"), outcome.describe()
+
+
+async def test_skill_instructions(run_agent):
+ outcome = await run_agent(
+ "What is the project codeword? Look it up with your project-codeword skill and "
+ "write it to answer.txt.",
+ skill_docs={"project-codeword": CODEWORD_SKILL},
+ )
+ assert outcome.read("answer.txt") == CODEWORD, outcome.describe()
+
+
+@pytest.fixture
+def notice_url():
+ """A page on this machine, so the fetch reaches it from any placement."""
+
+ class Notice(http.server.BaseHTTPRequestHandler):
+ def do_GET(self):
+ body = f"The notice stamp is {PAGE_STAMP}.
"
+ self.send_response(200)
+ self.send_header("Content-Type", "text/html")
+ self.end_headers()
+ self.wfile.write(body.encode())
+
+ def log_message(self, *args):
+ pass
+
+ server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Notice)
+ thread = threading.Thread(target=server.serve_forever, daemon=True)
+ thread.start()
+ yield f"http://127.0.0.1:{server.server_address[1]}/notice.html"
+ server.shutdown()
+ thread.join()
+
+
+async def test_fetch_tool(run_agent, notice_url):
+ outcome = await run_agent(
+ f"Read {notice_url} with the fetch tool and write the notice stamp it shows to "
+ "answer.txt.",
+ builtin_tools=["ipython", "fetch"],
+ )
+ assert outcome.read("answer.txt") == PAGE_STAMP, outcome.describe()
+ assert "fetch" in outcome.tool_calls(), outcome.describe()
diff --git a/tests/test_acp.py b/tests/test_acp.py
index 8c5b34eb..0336457c 100644
--- a/tests/test_acp.py
+++ b/tests/test_acp.py
@@ -60,6 +60,7 @@ def _runtime_metadata(**overrides: Any) -> dict[str, Any]:
"max_concurrent_subagents": 4,
"max_subagent_calls": None,
"allow_git": False,
+ "network_guidance": False,
},
"system_prompt_path": None,
"append_to_system_prompt": None,
@@ -163,6 +164,7 @@ def execution_snapshot(self) -> dict[str, Any]:
"compaction_tail_tokens": 12000,
"compaction_prompt_tokens": 4000,
"allow_git": False,
+ "network_guidance": False,
"harness_enabled": True,
"harness_global": False,
"auto_refine": False,
diff --git a/tests/test_compaction.py b/tests/test_compaction.py
index 3f09e32c..8d046493 100644
--- a/tests/test_compaction.py
+++ b/tests/test_compaction.py
@@ -1,5 +1,6 @@
from __future__ import annotations
+import itertools
import json
import re
import sys
@@ -87,6 +88,11 @@ def test_overflow_detection_is_status_gated():
assert is_context_overflow(error)
+# Discovered context windows are cached per base URL for the whole process, and an
+# id() can be reused once its object is freed, so each client gets a fresh number.
+_client_numbers = itertools.count()
+
+
class _ScriptedClient(DummyClient):
def __init__(
self,
@@ -97,7 +103,7 @@ def __init__(
super().__init__([])
self.actions = list(actions)
self.max_model_len = max_model_len
- self.base_url = f"http://scripted-{id(self)}"
+ self.base_url = f"http://scripted-{next(_client_numbers)}"
@property
def models(self):
diff --git a/tests/test_config.py b/tests/test_config.py
index 1980a890..e263d87c 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -23,6 +23,7 @@ def test_runtime_config_redacts_secrets():
max_compactions=3,
max_compaction_attempts=7,
allow_git=True,
+ network_guidance=False,
),
skills=("search", "edit"),
kernel_env=(("TASK_TOKEN", "task-secret"),),
@@ -44,6 +45,7 @@ def test_runtime_config_redacts_secrets():
max_concurrent_subagents=4,
max_subagent_calls=None,
allow_git=True,
+ network_guidance=False,
)
assert config.skills == ("search", "edit")
assert config.kernel_env == (("TASK_TOKEN", "task-secret"),)
diff --git a/tests/test_prompt.py b/tests/test_prompt.py
index 6f88ae1e..b0ec3bb2 100644
--- a/tests/test_prompt.py
+++ b/tests/test_prompt.py
@@ -13,7 +13,10 @@
EDIT_SKILL_PROMPT,
GIT_HISTORY_GUARD_PROMPT,
IPYTHON_CONTROL_PROMPT,
+ KERNEL_PACKAGES_PROMPT,
+ KERNEL_PREINSTALLED_PROMPT,
MCP_DISCOVERY_PROMPT,
+ OFFLINE_GIT_HISTORY_GUARD_PROMPT,
SEARCH_SKILL_PROMPT,
build_system_prompt,
resolve_prompts,
@@ -30,6 +33,7 @@ def _prompt(
*,
installed_skills: list[str] | None = None,
allow_git: bool = False,
+ network_guidance: bool = True,
mcp_servers: list[dict] | None = None,
) -> str:
return build_system_prompt(
@@ -37,6 +41,7 @@ def _prompt(
installed_skills or [],
allow_recursion=False,
allow_git=allow_git,
+ network_guidance=network_guidance,
active_tools=active_tools,
mcp_servers=mcp_servers,
)
@@ -79,6 +84,19 @@ def test_git_history_guard_prompt_included_for_shell_tools():
assert "`git log --all`" in prompt and "`clone`" in prompt
+def test_network_guidance_off_drops_only_network_dependent_sentences():
+ online = _prompt([_Tool("ipython")])
+ offline = _prompt([_Tool("ipython")], network_guidance=False)
+
+ assert KERNEL_PACKAGES_PROMPT in online and GIT_HISTORY_GUARD_PROMPT in online
+ assert KERNEL_PREINSTALLED_PROMPT in offline
+ assert OFFLINE_GIT_HISTORY_GUARD_PROMPT in offline
+ assert "uv pip install" not in offline
+ assert "online solutions" not in offline
+ assert "Do not cheat by copying or inferring solutions" in offline
+ assert "`git log --all`" in offline and "`clone`" in offline
+
+
def test_git_history_guard_prompt_omitted_when_unrestricted():
assert GIT_HISTORY_GUARD_PROMPT not in _prompt([_Tool("ipython")], allow_git=True)
+
-
-
+
+
+