"""Agent loop + delegation + compaction tests, with the LLM mocked offline.""" import json from pathlib import Path from types import SimpleNamespace import pytest import openai4s.agent.compaction as comp_mod import openai4s.agent.delegation as deleg_mod import openai4s.agent.loop as loop_mod from openai4s.agent import Agent from openai4s.agent.delegation import DelegationError, DelegationRunner from openai4s.config import Config, RoadmapFeatureFlags, get_config from openai4s.kernel.readiness import EnvironmentReadinessError class ScriptedLLM: """Returns queued replies in order; each call pops one.""" def __init__(self, replies): self._replies = list(replies) self.calls = [] def __call__(self, messages, cfg, **kw): self.calls.append(messages) content = ( self._replies.pop(0) if self._replies else ("```python\nhost.submit_output({}, ['Finished the task'])\n```") ) return { "content": content, "reasoning": None, "usage": {}, "finish_reason": "stop", "raw": {}, } def _tool_block(json_body: str) -> str: """Build a fenced ```tool block the way tests build ```python cells: three backticks + 'tool' + newline + the JSON + newline + three backticks.""" return "```" + "tool\n" + json_body + "\n" + "```" def test_detached_call_budget_is_keyed_on_the_session_root(monkeypatch): """A delegated child must not get its own slots. The per-session bound exists because the team quota gate reads a ledger that has not been charged for calls still in flight, and that ledger charges the *session root*. Keying on `frame_id` gave every `host.delegate` child a private scope, so one Stop on a fan-out could leave up to the fan-out cap billing while the session's own four stayed free. """ agent = Agent(use_skills=False, allow_delegate=False) class _Frames: def __init__(self): self.asked = [] def resolve_frame_scope(self, frame_id, **_kwargs): self.asked.append(frame_id) # what the store answers for a delegated child frame return {"frame_id": frame_id, "root_frame_id": "root-session"} frames = _Frames() agent.dispatcher = SimpleNamespace(store=frames) agent.frame_id = "child-frame-7" assert agent._provider_call_scope() == "root-session" assert frames.asked == ["child-frame-7"] # No frame at all keeps only the process-wide bound. agent.frame_id = None assert agent._provider_call_scope() is None # A store that cannot resolve must not fail the turn; the frame is the # honest fallback, and it still bounds *something*. agent.frame_id = "solo-frame" agent.dispatcher = SimpleNamespace(store=object()) assert agent._provider_call_scope() == "solo-frame" def test_code_as_action_cycle(monkeypatch): scripted = ScriptedLLM( [ "Let me compute it.\n```python\nprint(6 * 7)\n```", "```python\nhost.submit_output({'answer': 42}, ['Computed the answer'])\n```", ] ) monkeypatch.setattr(loop_mod, "chat", scripted) agent = Agent(use_skills=False, allow_delegate=False) result = agent.run("compute 6*7 and submit") # Completion is signalled through host.submit_output, not a text convention. assert result["stop_reason"] == "submitted" assert result["submitted_output"]["output"] == {"answer": 42} # 2 assistant turns happened assert len(scripted.calls) == 2 def test_cli_save_artifact_resolves_relative_to_actual_kernel_cwd( monkeypatch, tmp_path ): scripted = ScriptedLLM( [ "```python\n" "open('cli-result.txt', 'w').write('science')\n" "saved = host.save_artifact('cli-result.txt')\n" "print(saved['version_id'])\n" "```", "```python\n" "host.submit_output({'saved': True}, ['Saved the CLI artifact'])\n" "```", ] ) monkeypatch.setattr(loop_mod, "chat", scripted) monkeypatch.chdir(tmp_path) agent = Agent(use_skills=False, allow_delegate=False, max_turns=3) result = agent.run("write and save a relative artifact") assert result["stop_reason"] == "submitted" artifact = agent.dispatcher.store.artifact_by_filename( "cli-result.txt", agent.frame_id, strict=True ) assert artifact is not None metadata = agent.dispatcher.store.version_meta(artifact["latest_version_id"]) assert metadata["path"] == str(tmp_path / "cli-result.txt") assert Path(metadata["snapshot_path"]).read_text() == "science" def test_no_code_block_nudge(monkeypatch): scripted = ScriptedLLM( [ "I think the answer is 42.", # no code -> nudge "```python\nhost.submit_output({'a': 1}, ['Answered the question'])\n```", ] ) monkeypatch.setattr(loop_mod, "chat", scripted) result = Agent(use_skills=False, allow_delegate=False).run("hi") assert result["stop_reason"] == "submitted" def test_cli_non_scientific_finalize_uses_live_catalog_and_engine_result(monkeypatch): seen_tools = [] arguments = { "summary": "The requested explanation was completed.", "completion_bullets": ["Completed the requested explanation"], } def finalize_chat(messages, cfg, **kwargs): del messages, cfg seen_tools.append([tool.name for tool in kwargs["tools"]]) call = { "id": "final-cli", "wire_id": "wire-final-cli", "name": "finalize_response", "ordinal": 0, "raw_arguments": json.dumps(arguments), "arguments": arguments, "parse_error": None, "provider_meta": {"provider": "test"}, } return { "content": "", "tool_calls": [call], "assistant_message": { "role": "assistant", "content": "", "tool_calls": [call], }, } monkeypatch.setattr(loop_mod, "chat", finalize_chat) def unexpected_kernel(*_args, **_kwargs): raise AssertionError("a structured-finalize-only CLI turn spawned a kernel") monkeypatch.setattr(loop_mod, "Kernel", unexpected_kernel) agent = Agent( use_skills=False, allow_delegate=False, max_turns=1, ) result = agent.run( "Explain the already-known result without scientific computation" ) assert "finalize_response" in seen_tools[0] assert "bash" not in seen_tools[0] assert "submit_output" not in seen_tools[0] assert result["stop_reason"] == "submitted" assert result["submitted_output"] == { "output": {"summary": "The requested explanation was completed."}, "completion_bullets": ["Completed the requested explanation"], } assert result["final_message"] == "The requested explanation was completed." assert [ group["kind"] for group in agent.dispatcher.store.list_action_groups(agent.frame_id) ] == ["user", "finalize", "terminal"] @pytest.mark.parametrize("run_tool_first", [False, True]) def test_stage1_missing_profile_keeps_cli_control_plane_kernel_lazy( tmp_path, monkeypatch, run_tool_first ): """Readiness is a Code Cell gate, never a task or control-plane gate.""" replies = [] if run_tool_first: replies.append(("list_dir", {"path": "."})) replies.append( ( "finalize_response", { "summary": "The control-only request was completed.", "completion_bullets": ["Completed without a science runtime"], }, ) ) def control_chat(messages, cfg, **kwargs): del messages, cfg, kwargs name, arguments = replies.pop(0) call = { "id": f"call-{name}", "wire_id": f"wire-{name}", "name": name, "ordinal": 0, "raw_arguments": json.dumps(arguments), "arguments": arguments, "parse_error": None, "provider_meta": {"provider": "test"}, } return { "content": "", "tool_calls": [call], "assistant_message": { "role": "assistant", "content": "", "tool_calls": [call], }, } def forbidden_readiness(**_kwargs): raise AssertionError("a control-only turn probed scientific readiness") def forbidden_kernel(*_args, **_kwargs): raise AssertionError("a control-only turn spawned a kernel") monkeypatch.setattr(loop_mod, "chat", control_chat) monkeypatch.setattr(loop_mod, "Kernel", forbidden_kernel) monkeypatch.setattr( "openai4s.kernel.readiness.standard_profile_readiness", forbidden_readiness, ) cfg = Config( data_dir=tmp_path / "data", llm=get_config().llm, roadmap_features=RoadmapFeatureFlags(stage1_trusted_delivery=True), ) agent = Agent( cfg=cfg, workspace=tmp_path / "workspace", use_skills=False, allow_delegate=False, max_turns=2, ) result = agent.run("complete through the control plane") assert result["stop_reason"] == "submitted" assert result["final_message"] == "The control-only request was completed." assert agent.dispatcher.store.cell_count(agent.frame_id) == 0 assert replies == [] @pytest.mark.parametrize(("language", "fence"), [("python", "python"), ("r", "r")]) def test_stage1_cli_cell_readiness_refuses_before_safety_or_worker( tmp_path, monkeypatch, language, fence ): readiness = { "state": "needs_repair", "ready": False, "missing_environments": [], "missing_packages": {"python": ["numpy"], "r": ["r-base"]}, "remediation": None, } scripted = ScriptedLLM([f"```{fence}\nprint(42)\n```"]) monkeypatch.setattr(loop_mod, "chat", scripted) monkeypatch.setattr( "openai4s.kernel.readiness.standard_profile_readiness", lambda **_kwargs: readiness, ) def forbidden_kernel(*_args, **_kwargs): raise AssertionError(f"a refused {language} Cell spawned a worker") monkeypatch.setattr(loop_mod, "Kernel", forbidden_kernel) import openai4s.kernel.r_kernel as r_kernel_mod monkeypatch.setattr(r_kernel_mod, "spawn_r_kernel", forbidden_kernel) cfg = Config( data_dir=tmp_path / "data", llm=get_config().llm, roadmap_features=RoadmapFeatureFlags(stage1_trusted_delivery=True), ) agent = Agent( cfg=cfg, workspace=tmp_path / "workspace", use_skills=False, allow_delegate=False, max_turns=1, ) monkeypatch.setattr( agent, "_pre_exec_gate", lambda *_args, **_kwargs: (_ for _ in ()).throw( AssertionError("readiness refusal happened after the safety gate") ), ) with pytest.raises(EnvironmentReadinessError) as refused: agent.run("run a scientific Cell") assert refused.value.readiness == readiness assert agent.dispatcher.store.cell_count(agent.frame_id) == 0 assert agent._foreground_kernel is None def test_submit_output_soft_fail_does_not_complete(monkeypatch): """host.submit_output with invalid completion_bullets soft-fails (the dispatcher returns {'error': ...} → RuntimeError in the cell) and the task does NOT end; a subsequent valid submit_output is what completes it.""" scripted = ScriptedLLM( [ "```python\n" "try:\n" " host.submit_output({'a': 1}, [])\n" "except RuntimeError as e:\n" " print('SOFT-FAIL:', e)\n" "```", "```python\nhost.submit_output({'a': 1}, ['Computed the answer'])\n```", ] ) monkeypatch.setattr(loop_mod, "chat", scripted) agent = Agent(use_skills=False, allow_delegate=False, max_turns=4) result = agent.run("submit twice") # the invalid submit did not stop the loop — the valid one did assert result["stop_reason"] == "submitted" assert len(scripted.calls) == 2 assert result["submitted_output"]["output"] == {"a": 1} assert result["submitted_output"]["completion_bullets"] == ["Computed the answer"] obs = [t["content"] for t in result["transcript"] if t["role"] == "observation"] assert any( "SOFT-FAIL:" in o and "completion_bullets must be a list of 1-4 items" in o for o in obs ) class _MeteredLLM(ScriptedLLM): """A ScriptedLLM whose every reply bills a fixed canonical usage.""" def __init__(self, replies, *, raise_on_call=None, after_call=None): super().__init__(replies) self.raise_on_call = raise_on_call self.after_call = after_call self.observed_status: list[str | None] = [] self.agent = None def __call__(self, messages, cfg, **kw): if self.agent is not None: frame = self.agent.dispatcher.store.get_frame(self.agent.frame_id) self.observed_status.append(frame and frame.get("status")) if self.raise_on_call is not None: raise self.raise_on_call reply = super().__call__(messages, cfg, **kw) reply["usage"] = { "prompt_tokens": 100, "completion_tokens": 7, "input_tokens": 100, "output_tokens": 7, } if self.after_call is not None: self.after_call() return reply class _SettableCancellation: def __init__(self, cancelled: bool = False) -> None: self.flag = cancelled def cancelled(self) -> bool: return self.flag def _root_frame_row(agent): frame = agent.dispatcher.store.get_frame(agent.frame_id) assert frame is not None return frame def test_cli_root_frame_is_done_with_token_totals_after_submit(monkeypatch): """A root Agent owns the turn frame it opened, so it must close it. The frame used to keep the repository default ``processing`` forever with NULL token columns, so `GET /frames//status` on the same data dir answered "processing" for a finished CLI run and per-session token accounting silently skipped every CLI run. """ scripted = _MeteredLLM( ["```python\nhost.submit_output({'a': 1}, ['Computed the answer'])\n```"] ) monkeypatch.setattr(loop_mod, "chat", scripted) agent = Agent(use_skills=False, allow_delegate=False, max_turns=3) result = agent.run("submit once") assert result["stop_reason"] == "submitted" frame = _root_frame_row(agent) assert frame["status"] == "done" assert frame["input_tokens"] == 100 assert frame["output_tokens"] == 7 def test_cli_root_frame_is_failed_after_max_turns(monkeypatch): scripted = _MeteredLLM(["Still thinking."] * 5) monkeypatch.setattr(loop_mod, "chat", scripted) agent = Agent(use_skills=False, allow_delegate=False, max_turns=2) result = agent.run("never finishes") assert result["stop_reason"] == "max_turns" frame = _root_frame_row(agent) assert frame["status"] == "failed" assert frame["input_tokens"] == 200 assert frame["output_tokens"] == 14 def test_cli_root_frame_is_cancelled_when_cancelled_before_the_run(monkeypatch): scripted = _MeteredLLM(["never called"]) monkeypatch.setattr(loop_mod, "chat", scripted) agent = Agent( use_skills=False, allow_delegate=False, max_turns=3, cancellation=_SettableCancellation(cancelled=True), ) result = agent.run("cancelled up front") assert result["stop_reason"] == "cancelled" assert scripted.calls == [] # The Web gateway's vocabulary for a cancelled turn, not a second one. assert _root_frame_row(agent)["status"] == "cancelled" def test_cli_root_frame_is_cancelled_when_the_engine_stops_on_cancel(monkeypatch): cancellation = _SettableCancellation() scripted = _MeteredLLM( ["Still thinking."] * 5, after_call=lambda: setattr(cancellation, "flag", True), ) monkeypatch.setattr(loop_mod, "chat", scripted) agent = Agent( use_skills=False, allow_delegate=False, max_turns=5, cancellation=cancellation, ) result = agent.run("stopped mid-run") assert result["stop_reason"] == "cancelled" assert len(scripted.calls) == 1 assert _root_frame_row(agent)["status"] == "cancelled" # The one billed reply lands on the frame whether the engine received it # or ChatModel abandoned it to its (asynchronous) late-accounting sink. import time deadline = time.monotonic() + 10 while _root_frame_row(agent)["input_tokens"] is None: assert time.monotonic() < deadline, "the billed reply was never metered" time.sleep(0.05) assert _root_frame_row(agent)["input_tokens"] == 100 def test_cli_root_frame_is_failed_when_the_run_raises(monkeypatch): scripted = _MeteredLLM([], raise_on_call=RuntimeError("provider exploded")) monkeypatch.setattr(loop_mod, "chat", scripted) agent = Agent(use_skills=False, allow_delegate=False, max_turns=3) with pytest.raises(RuntimeError, match="provider exploded"): agent.run("the provider raises") assert _root_frame_row(agent)["status"] == "failed" def test_cli_root_frame_is_cancelled_on_ctrl_c(monkeypatch): """Ctrl-C is the CLI user's Stop: it re-raises, and the frame says so.""" def interrupted(self, messages, **_kwargs): raise KeyboardInterrupt monkeypatch.setattr(loop_mod.AgentEngine, "run", interrupted) agent = Agent(use_skills=False, allow_delegate=False, max_turns=3) with pytest.raises(KeyboardInterrupt): agent.run("interrupted") assert _root_frame_row(agent)["status"] == "cancelled" def test_reused_cli_agent_reopens_its_frame_for_the_next_run(monkeypatch): """A reused Agent must not report the previous run's terminal mid-run.""" scripted = _MeteredLLM(["Still thinking."] * 2) monkeypatch.setattr(loop_mod, "chat", scripted) agent = Agent(use_skills=False, allow_delegate=False, max_turns=2) scripted.agent = agent assert agent.run("first")["stop_reason"] == "max_turns" assert _root_frame_row(agent)["status"] == "failed" scripted._replies = ["Still thinking."] * 2 scripted.observed_status.clear() assert agent.run("second")["stop_reason"] == "max_turns" assert scripted.observed_status == ["processing", "processing"] frame = _root_frame_row(agent) assert frame["status"] == "failed" assert frame["input_tokens"] == 400 def test_agent_given_a_frame_does_not_write_that_frame_s_status(monkeypatch): """A delegated child's status belongs to the DelegationRunner, and an embedder's frame to the embedder: only a frame the Agent opened is its.""" from openai4s.store import get_store store = get_store(get_config().db_path) frame_id = store.new_frame(kind="turn", model="m", depth=1) store.update_frame(frame_id, status="awaiting_user_response") scripted = _MeteredLLM(["Still thinking."] * 3) monkeypatch.setattr(loop_mod, "chat", scripted) agent = Agent( use_skills=False, allow_delegate=False, max_turns=2, frame_id=frame_id ) assert agent.run("child")["stop_reason"] == "max_turns" frame = store.get_frame(frame_id) assert frame["status"] == "awaiting_user_response" assert frame["input_tokens"] is None assert frame["output_tokens"] is None def test_max_turns_stop(monkeypatch): # never calls submit_output -> should stop at max_turns scripted = ScriptedLLM(["```python\nx = 1\n```"] * 10) monkeypatch.setattr(loop_mod, "chat", scripted) agent = Agent(use_skills=False, allow_delegate=False, max_turns=3) result = agent.run("loop forever") assert result["stop_reason"] == "max_turns" def test_repeated_native_tool_trips_no_progress(monkeypatch, tmp_path): """Same native tool+args three times must stop as no_progress, not completed.""" provider_calls = [] def chat(messages, cfg, **kwargs): del messages, cfg, kwargs index = len(provider_calls) call = { "id": f"loop-{index}", "wire_id": f"loop-{index}", "name": "list_dir", "ordinal": 0, "raw_arguments": '{"path":"."}', "arguments": {"path": "."}, "parse_error": None, "provider_meta": {"provider": "test"}, } provider_calls.append(call) return { "content": "", "tool_calls": [call], "assistant_message": { "role": "assistant", "content": "", "tool_calls": [call], }, } monkeypatch.setattr(loop_mod, "chat", chat) agent = Agent( use_skills=False, allow_delegate=False, max_turns=10, workspace=tmp_path, ) result = agent.run("list the workspace repeatedly") assert result["stop_reason"] == "no_progress" assert result["submitted_output"] is None assert len(provider_calls) == 3 groups = agent.dispatcher.store.list_action_groups(agent.frame_id) kinds = [group["kind"] for group in groups] assert kinds.count("native_tools") == 3 assert kinds[-1] == "terminal" terminal = groups[-1]["events"][0] assert terminal["type"] == "failed" assert terminal["result"]["reason"] == "no_progress" assert "completion" not in terminal["result"] # ---- R execution channel (```r) ------------------------------------------- class _FakeRKernel: """Stands in for the persistent R kernel in loop tests (no R needed).""" def __init__(self): self.cells = [] self.down = False def is_alive(self): return not self.down def execute(self, code, origin="agent", on_chunk=None): self.cells.append(code) return { "stdout": "[1] 42\n", "stderr": "", "error": None, "interrupted": False, "trace": {"error_lineno": None, "error_call": None}, "usage": {}, } def shutdown(self): self.down = True def test_r_cell_routes_to_r_kernel_and_is_non_terminal(monkeypatch): """An ```r cell runs on the (lazily spawned) R kernel, its observation is fed back, and — R being an analysis channel with no host object — the task still completes only through a python host.submit_output cell. The R kernel is shut down with the run.""" import openai4s.kernel.r_kernel as rk_mod fake = _FakeRKernel() spawns = [] def fake_spawn(**kw): spawns.append(kw) return fake monkeypatch.setattr(rk_mod, "spawn_r_kernel", fake_spawn) scripted = ScriptedLLM( [ "R first.\n```r\nx <- 42\nprint(x)\n```", "```python\nhost.submit_output({'a': 1}, ['Analyzed in R'])\n```", ] ) monkeypatch.setattr(loop_mod, "chat", scripted) result = Agent(use_skills=False, allow_delegate=False, max_turns=4).run("use R") assert result["stop_reason"] == "submitted" assert fake.cells == ["x <- 42\nprint(x)\n"] assert len(spawns) == 1 # lazy: spawned exactly once, on first ```r cell obs = [t["content"] for t in result["transcript"] if t["role"] == "observation"] assert any("[1] 42" in o for o in obs) assert fake.down # run-scoped lifecycle def test_r_cell_without_r_soft_fails_into_observation(monkeypatch): """No R interpreter -> the ```r cell yields an ERROR observation (never a crash), and the model can fall back to python and still finish.""" import openai4s.kernel.r_kernel as rk_mod def no_r(**kw): raise RuntimeError("no R interpreter available: build the 'r' env") monkeypatch.setattr(rk_mod, "spawn_r_kernel", no_r) scripted = ScriptedLLM( [ "```r\n1 + 1\n```", "```python\nhost.submit_output({'a': 1}, ['Fell back to python'])\n```", ] ) monkeypatch.setattr(loop_mod, "chat", scripted) result = Agent(use_skills=False, allow_delegate=False, max_turns=4).run("try R") assert result["stop_reason"] == "submitted" obs = [t["content"] for t in result["transcript"] if t["role"] == "observation"] assert any("R kernel unavailable" in o for o in obs) # ---- ReAct tool surface (```tool) ---------------------------------------- def test_react_tool_call_then_submit(monkeypatch, tmp_path): """Happy ReAct path: a ```tool turn runs a read-only tool through the REAL HostDispatcher (whose workspace is a per-test tmp dir), its result is fed back as ONE '[Tool Results]' observation, and the loop CONTINUES to the next turn (it does not nudge or end) until a later python cell submits output.""" scripted = ScriptedLLM( [ # `list_dir` runs cleanly offline: the dispatcher auto-creates the # workspace dir and lists it (empty here) — no network, no fixtures. "Let me look around first.\n" + _tool_block('{"name": "list_dir", "arguments": {"path": "."}}'), "```python\nhost.submit_output({}, ['done'])\n```", ] ) monkeypatch.setattr(loop_mod, "chat", scripted) result = Agent( use_skills=False, allow_delegate=False, max_turns=4, workspace=tmp_path ).run("list the workspace, then submit") # completion still flows ONLY through host.submit_output assert result["stop_reason"] == "submitted" # the tool result came back as one observation Turn, tagged [Tool Results] obs = [t["content"] for t in result["transcript"] if t["role"] == "observation"] assert any(o.startswith("[Tool Results]") for o in obs) assert any("[Tool: list_dir]" in o for o in obs) # the tool observation was fed back and the loop continued (>=2 chat calls): # it neither nudged nor ended on the tool turn. assert len(scripted.calls) >= 2 def test_react_malformed_tool_block_surfaces_error(monkeypatch): """Malformed ReAct path: a ```tool block with invalid JSON is surfaced as a '[Tool error]' observation (the loop does not crash), and a later python cell still completes the task.""" scripted = ScriptedLLM( [ _tool_block("{not valid json,}"), "```python\nhost.submit_output({}, ['done'])\n```", ] ) monkeypatch.setattr(loop_mod, "chat", scripted) result = Agent(use_skills=False, allow_delegate=False, max_turns=4).run( "bad tool, then submit" ) assert result["stop_reason"] == "submitted" obs = [t["content"] for t in result["transcript"] if t["role"] == "observation"] # the parse error was fed back, not raised assert any("[Tool error]" in o for o in obs) def test_code_cell_wins_over_embedded_tool_fence(monkeypatch): """Fence-collision guard: a ```python cell whose body QUOTES a ```tool block (e.g. writing docs about the tool syntax) runs the CELL — the embedded tool is never executed and the turn is not hijacked into a tool turn.""" doc = ( "```python\n" "readme = '''\nUsage example:\n" + _tool_block('{"name": "bash", "arguments": {"command": "echo pwned"}}') + "\n'''\nprint('wrote', len(readme), 'chars')\n" "host.submit_output({'readme': readme}, ['documented'])\n" "```" ) scripted = ScriptedLLM([doc]) monkeypatch.setattr(loop_mod, "chat", scripted) result = Agent(use_skills=False, allow_delegate=False, max_turns=1).run( "write the docs and submit" ) assert result["stop_reason"] == "submitted" assert len(scripted.calls) == 1 # the embedded fence did not truncate/error assert '"name": "bash"' in result["submitted_output"]["output"]["readme"] obs = [t["content"] for t in result["transcript"] if t["role"] == "observation"] # the embedded ```tool must NOT have been executed as a tool call assert not any("[Tool Results]" in o for o in obs) assert not any("[Tool: bash]" in o for o in obs) assert not any("ERROR" in o for o in obs) def test_four_backtick_python_fence_is_complete_and_wins_over_inner_tool(): outer = "`" * 4 reply = ( outer + "python\nreadme = '''\n" + _tool_block('{"name": "bash", "arguments": {"command": "echo pwned"}}') + "\n'''\nhost.submit_output({'readme': readme}, ['done'])\n" + outer ) code = loop_mod._extract_code(reply) assert code is not None compile(code, "", "exec") assert "host.submit_output" in code assert loop_mod.parse_tool_calls(reply) == ([], []) # ---- compaction ---------------------------------------------------------- def test_estimate_tokens_monotonic(): small = [{"role": "user", "content": "x"}] big = [{"role": "user", "content": "x" * 4000}] assert comp_mod.estimate_tokens(big) > comp_mod.estimate_tokens(small) def test_context_estimate_accounts_for_structured_components_independently(): estimate = comp_mod.estimate_context( [ { "role": "assistant", "content": [ {"type": "text", "text": "look"}, {"type": "image_url", "image_url": {"url": "https://x/i"}}, ], "tool_calls": [{"id": "c1", "name": "lookup", "arguments": {}}], "wire_state": {"response_id": "resp-1", "cursor": 9}, } ], tool_schemas=[{"name": "lookup", "parameters": {"type": "object"}}], ) assert estimate.text > 0 assert estimate.images >= comp_mod.IMAGE_TOKEN_ESTIMATE assert estimate.tool_calls > 0 assert estimate.tool_schemas > 0 assert estimate.wire_state > 0 components = estimate.as_dict() assert components.pop("total") == sum(components.values()) def test_context_estimate_separates_tool_results_and_artifact_refs(): estimate = comp_mod.estimate_context( [ { "role": "tool", "content": "large structured result", "artifact_refs": [{"artifact_id": "a-1", "version_id": "v-1"}], } ] ) assert estimate.tool_results > 0 assert estimate.artifact_refs > 0 assert estimate.text == 0 def test_large_tool_result_is_content_addressed_and_recoverable(tmp_path): content = "measurement," * 100 messages = [ {"role": "system", "content": "system"}, {"role": "user", "content": "task"}, { "role": "tool", "tool_call_id": "call-1", "name": "measure", "content": content, }, ] projected = comp_mod.externalize_large_outputs( messages, tmp_path, threshold_chars=32, preview_chars=24, archive_metadata={"branch": "branch-a", "ledger_cursor": 41}, ) reference = projected[2]["content_archive"] assert len(reference["sha256"]) == 64 assert reference["sha256"] in projected[2]["content"] assert "measurement" in projected[2]["content"] assert comp_mod.load_archived_content(tmp_path, reference["sha256"]) == content blob = json.loads((tmp_path / reference["archive_ref"]).read_text("utf-8")) assert blob["metadata"]["branch"] == "branch-a" assert blob["metadata"]["ledger_cursor"] == 41 def test_large_tool_result_can_become_a_versioned_artifact_reference(): calls = [] def archive(content, message, metadata): calls.append((content, message, metadata)) return {"artifact_id": "a-context", "version_id": "v-context"} projected = comp_mod.externalize_large_outputs( [{"role": "tool", "content": "x" * 100}], None, threshold_chars=20, artifact_archiver=archive, ) assert calls and calls[0][2]["original_chars"] == 100 assert projected[0]["artifact_refs"][0]["artifact_id"] == "a-context" assert "version_id: v-context" in projected[0]["content"] def test_code_and_observation_are_one_atomic_compaction_segment(): messages = [ {"role": "system", "content": "system"}, {"role": "user", "content": "task"}, {"role": "assistant", "content": "```python\nx = 1\n```"}, {"role": "user", "content": "[Observation]\n(no output)"}, {"role": "assistant", "content": "next"}, ] segments = comp_mod.segment_messages(messages) assert any( segment.kind == "code_observation" and (segment.start, segment.end) == (2, 4) for segment in segments ) # A fixed two-message tail would begin at the observation; V2 expands it # back to include the source cell that produced that observation. assert comp_mod.safe_keep_recent(messages, minimum=2) == 3 def test_should_compact_uses_window(monkeypatch): cfg = get_config() # ~1000 tokens of content msgs = [{"role": "user", "content": "x" * 4000}] * 10 # Tiny window -> should compact; huge window -> should not. monkeypatch.setattr(cfg, "context_window_tokens", 100) monkeypatch.setattr(cfg, "compaction_trigger_ratio", 0.75) assert comp_mod.should_compact(msgs, cfg) is True monkeypatch.setattr(cfg, "context_window_tokens", 10_000_000) assert comp_mod.should_compact(msgs, cfg) is False def test_compact_shrinks_and_preserves_head(monkeypatch): monkeypatch.setattr(comp_mod, "chat", ScriptedLLM(["SUMMARY TEXT"])) msgs = ( [{"role": "system", "content": "sys"}, {"role": "user", "content": "task"}] + [{"role": "assistant", "content": f"a{i}"} for i in range(6)] + [{"role": "user", "content": f"o{i}"} for i in range(6)] ) out = comp_mod.compact(msgs, get_config(), keep_recent=4) assert len(out) < len(msgs) assert out[0]["content"] == "sys" # system preserved assert out[1]["content"] == "task" # original task preserved assert "SUMMARY TEXT" in out[2]["content"] # summary injected assert out[-1]["content"] == "o5" # most recent kept verbatim def test_compact_handoff_is_structured_and_never_invents_kernel_continuity( monkeypatch, tmp_path ): monkeypatch.setattr(comp_mod, "chat", ScriptedLLM(["Finished old analysis."])) messages = [ {"role": "system", "content": "sys"}, {"role": "user", "content": "task"}, {"role": "assistant", "content": "old work"}, {"role": "user", "content": "old result"}, {"role": "assistant", "content": "recent"}, ] archives = [] out = comp_mod.compact( messages, get_config(), keep_recent=1, archive_dir=tmp_path, archive_metadata={ "branch": "experiment-b", "ledger_cursor": 73, "recovery_pointer": {"checkpoint": "cp-4"}, }, archive_sink=archives.append, ) handoff = out[2]["content"] for field in comp_mod.HANDOFF_FIELDS: assert f"## {field}" in handoff assert "Unknown" in handoff assert "NOT assumed to exist" in handoff assert "namespace still holds" not in handoff archive_path = next( path for path in tmp_path.glob("compaction-*.json") if path.is_file() ) archive = json.loads(archive_path.read_text("utf-8")) assert archive["schema_version"] == 2 assert archive["metadata"]["branch"] == "experiment-b" assert archive["metadata"]["ledger_cursor"] == 73 assert archive["metadata"]["recovery_pointer"] == {"checkpoint": "cp-4"} assert archives[0]["archive_id"] == archive["archive_id"] assert archives[0]["context_estimate_before"]["total"] > 0 def test_compact_handoff_marks_restarted_generation_as_non_persistent(monkeypatch): monkeypatch.setattr(comp_mod, "chat", ScriptedLLM(["Summary"])) messages = [ {"role": "system", "content": "sys"}, {"role": "user", "content": "task"}, {"role": "assistant", "content": "old"}, {"role": "user", "content": "old result"}, {"role": "assistant", "content": "recent"}, ] out = comp_mod.compact( messages, get_config(), keep_recent=1, archive_metadata={ "active_kernel_generation": "python:8", "previous_kernel_generation": "python:7", }, ) assert "python:8" in out[2]["content"] assert "previous: python:7" in out[2]["content"] assert "variables from earlier generations are NOT available" in out[2]["content"] # ---- delegation ---------------------------------------------------------- @pytest.mark.parametrize( "second_spec", [ {"request": "async", "wait": False}, {"request": ["fanout-a", "fanout-b"], "wait": True}, {"request": "serial", "wait": True}, ], ) def test_reused_agent_recreates_its_delegation_runner(monkeypatch, second_spec): """Every Agent run gets a live pool, including parallel second runs.""" observed_runners = [] delegated_results = [] chat_calls = 0 agent = None def fake_run_one(self, child): assert child.begin(1) out = { "child_id": child.child_id, "stop_reason": "submitted", "task_status": "completed", "output": {"task": child.spec["request"]}, } assert child.finish_done(out) return out monkeypatch.setattr(deleg_mod.DelegationRunner, "_run_one", fake_run_one) def finalize_reply(call_index): arguments = { "summary": f"run {call_index} complete", "completion_bullets": [f"Completed run {call_index}"], } call = { "id": f"final-{call_index}", "wire_id": f"wire-final-{call_index}", "name": "finalize_response", "ordinal": 0, "raw_arguments": json.dumps(arguments), "arguments": arguments, "parse_error": None, "provider_meta": {"provider": "test"}, } return { "content": "", "tool_calls": [call], "assistant_message": { "role": "assistant", "content": "", "tool_calls": [call], }, } def chat_with_second_run_delegation(messages, cfg, **kwargs): nonlocal chat_calls del messages, cfg, kwargs chat_calls += 1 assert agent is not None runner = agent._delegation_runner assert runner is not None observed_runners.append(runner) if chat_calls == 2: result = runner(dict(second_spec)) if second_spec.get("wait") is False: result = runner.collect({"child_ids": [result["child_id"]]})[0] delegated_results.append(result) return finalize_reply(chat_calls) monkeypatch.setattr(loop_mod, "chat", chat_with_second_run_delegation) agent = Agent(use_skills=False, allow_delegate=True, max_turns=1) first = agent.run("first run") assert first["stop_reason"] == "submitted" assert agent._delegation_runner is None second = agent.run("second run") assert second["stop_reason"] == "submitted" assert len(observed_runners) == 2 assert observed_runners[0] is not observed_runners[1] assert all(runner._pool._shutdown for runner in observed_runners) assert delegated_results if isinstance(delegated_results[0], list): assert len(delegated_results[0]) == 2 else: assert delegated_results[0]["task_status"] == "completed" def test_delegate_fanout_cap(): runner = DelegationRunner(get_config()) with pytest.raises(DelegationError): runner({"request": ["t"] * (deleg_mod.FANOUT_CAP + 1)}) def test_delegate_single_and_list(monkeypatch): # Stub the leaf Agent.run so no real LLM/kernel is used. def fake_run(self, task): return { "stop_reason": "final", "submitted_output": { "output": {"echo": task}, "completion_bullets": ["ok"], }, "final_message": "FINAL", } monkeypatch.setattr(loop_mod.Agent, "run", fake_run) runner = DelegationRunner(get_config()) one = runner({"request": "do X"}) assert isinstance(one, dict) assert one["output"] == {"echo": "do X"} many = runner({"request": ["A", "B", "C"]}) assert isinstance(many, list) and len(many) == 3 assert {m["output"]["echo"] for m in many} == {"A", "B", "C"} def test_delegate_output_schema_uses_shared_completion_validation(monkeypatch): def fake_run(self, task): output = {"x": 1} if task == "valid" else {"y": 1} return { "stop_reason": "submitted", "submitted_output": { "output": output, "completion_bullets": ["Computed the result"], }, "final_message": None, } monkeypatch.setattr(loop_mod.Agent, "run", fake_run) runner = DelegationRunner(get_config()) schema = {"type": "object", "required": ["x"]} invalid = runner({"request": "invalid", "output_schema": schema}) assert invalid["error"] == ( "output_schema violation: output missing required field 'x'" ) assert runner.children()[0]["status"] == "failed" valid = runner({"request": "valid", "output_schema": schema}) assert "error" not in valid assert valid["output"] == {"x": 1} assert runner.children()[1]["status"] == "done" def test_web_delegated_child_runs_in_parent_session_workspace(monkeypatch, tmp_path): """A Web-delegated child's relative writes land in the parent session's workspace, never in the daemon's launch directory (its process cwd).""" daemon_cwd = tmp_path / "daemon-cwd" workspace = tmp_path / "session-workspace" daemon_cwd.mkdir() workspace.mkdir() scripted = ScriptedLLM( [ "```python\n" "open('kernel-out.txt', 'w').write('from kernel cwd')\n" "host.write_file('host-out.txt', 'from host files')\n" "host.submit_output({'wrote': True}, ['Wrote both files'])\n" "```", ] ) monkeypatch.setattr(loop_mod, "chat", scripted) monkeypatch.chdir(daemon_cwd) # The gateway wires the runner with the session workspace (_wire_delegation). runner = DelegationRunner(get_config(), workspace=workspace) try: result = runner({"request": "write files via relative paths"}) finally: runner.close() assert result["stop_reason"] == "submitted" # Kernel cwd and the dispatcher's file service both anchor to the workspace. assert (workspace / "kernel-out.txt").read_text() == "from kernel cwd" assert (workspace / "host-out.txt").read_text() == "from host files" assert not (daemon_cwd / "kernel-out.txt").exists() assert not (daemon_cwd / "host-out.txt").exists() def test_cli_delegated_child_keeps_process_cwd(monkeypatch, tmp_path): """Without an explicit workspace (the CLI path) a delegated child still resolves relative writes against the process cwd, exactly as before.""" cli_cwd = tmp_path / "cli-cwd" cli_cwd.mkdir() scripted = ScriptedLLM( [ "```python\n" "open('kernel-out.txt', 'w').write('from cli cwd')\n" "host.submit_output({'wrote': True}, ['Wrote the file'])\n" "```", ] ) monkeypatch.setattr(loop_mod, "chat", scripted) monkeypatch.chdir(cli_cwd) runner = DelegationRunner(get_config()) try: result = runner({"request": "write a file via a relative path"}) finally: runner.close() assert result["stop_reason"] == "submitted" assert (cli_cwd / "kernel-out.txt").read_text() == "from cli cwd" def test_delegating_agent_threads_workspace_to_its_nested_runner(tmp_path): """A child Agent hands its own workspace to the runner it builds for grandchildren, so the whole delegation subtree stays anchored.""" agent = Agent(use_skills=False, workspace=tmp_path) try: assert agent._delegation_runner is not None assert agent._delegation_runner.workspace == tmp_path finally: agent._delegation_runner.close() def test_delegate_session_cap(monkeypatch): def fake_run(self, task): return {"stop_reason": "final", "submitted_output": None, "final_message": None} monkeypatch.setattr(loop_mod.Agent, "run", fake_run) runner = DelegationRunner(get_config()) runner._spawned = deleg_mod.SESSION_CAP # pretend we're at the cap with pytest.raises(DelegationError): runner({"request": "one more"}) if __name__ == "__main__": raise SystemExit(pytest.main([__file__, "-v"]))