"""Tests for the OpenCode CLI backend.""" from __future__ import annotations import argparse import json import os import subprocess from pathlib import Path from unittest import mock import pytest from skillopt_sleep import cycle from skillopt_sleep.__main__ import _add_common, _cfg_from_args from skillopt_sleep.backend import ( _NO_WINDOW, _OPENCODE_SYNTHETIC_TOOL_QUERY, _OPENCODE_SYNTHETIC_TOOL_RESULT, DualBackend, MockBackend, OpenCodeCliBackend, OpenCodeError, _parse_opencode_jsonl_events, build_backend, get_backend, resolve_opencode_path, ) from skillopt_sleep.config import DEFAULTS, SleepConfig, load_config from skillopt_sleep.types import TaskRecord class _FakeProc: def __init__(self, stdout: str, stderr: str = "", returncode: int = 0) -> None: self.stdout = stdout self.stderr = stderr self.returncode = returncode def _event(event_type: str, *, session: str = "session-1", part=None) -> str: event = {"type": event_type, "sessionID": session} if part is not None: event["part"] = part return json.dumps(event) def _success_stream(*texts: str) -> str: lines = [ _event("step_start", part={"type": "step-start"}), *(_event("text", part={"type": "text", "text": text}) for text in texts), _event("step_finish", part={"type": "step-finish"}), ] return "\n".join(lines) def _tool_event( tool_id: str, *, call_id: str = "call-1", status: str = "completed", output: str | None = None, query=_OPENCODE_SYNTHETIC_TOOL_QUERY, ) -> str: state = {"status": status, "input": {"query": query}} if output is not None: state["output"] = output return _event( "tool_use", part={"type": "tool", "tool": tool_id, "callID": call_id, "state": state}, ) def _tool_success_stream(tool_id: str, text: str = "answer") -> str: return "\n".join( [ _event("step_start", part={"type": "step-start"}), _tool_event(tool_id, output=_OPENCODE_SYNTHETIC_TOOL_RESULT), _event("step_finish", part={"type": "step-finish"}), _event("step_start", part={"type": "step-start"}), _event("text", part={"type": "text", "text": text}), _event("step_finish", part={"type": "step-finish"}), ] ) def _resolved_mcp( *names: str, disabled: bool = False, snapshot: bool | None = None, ) -> str: mcp = { name: { "type": "local", "command": ["mcp-server"], **({"enabled": False} if disabled else {}), } for name in names } resolved = {"mcp": mcp} if snapshot is not None: resolved["snapshot"] = snapshot return json.dumps(resolved) def _successful_plain_results(*mcp_names: str, answer: str = "answer") -> list[_FakeProc]: return [ _FakeProc(_resolved_mcp(*mcp_names, snapshot=False)), _FakeProc(_resolved_mcp(*mcp_names, disabled=True, snapshot=False)), _FakeProc(_success_stream(answer)), ] def _assert_replay_permissions(config, env, agent_name: str, expected_permission: dict[str, str]) -> None: assert json.loads(env["OPENCODE_PERMISSION"]) == expected_permission assert config["permission"] == expected_permission assert config["agent"][agent_name]["permission"] == expected_permission def _assert_controlled_tool_environment(env, work: str) -> None: expected_npm = { "npm_config_audit": "false", "npm_config_cache": os.path.join(work, ".npm-cache"), "npm_config_fetch_retries": "0", "npm_config_fetch_retry_maxtimeout": "100", "npm_config_fetch_retry_mintimeout": "100", "npm_config_fetch_timeout": "1000", "npm_config_fund": "false", "npm_config_offline": "true", "npm_config_update_notifier": "false", } for key, value in expected_npm.items(): matches = { candidate: candidate_value for candidate, candidate_value in env.items() if candidate.upper() == key.upper() } assert matches == {key: value} assert env["OPENCODE_DISABLE_LSP_DOWNLOAD"] == "1" assert env["OPENCODE_DISABLE_MODELS_FETCH"] == "1" def _assert_replay_project_artifacts( work: str, *, project_id: str, tool_id: str, forbidden_text: str, ) -> None: root = Path(work) assert (root / ".git").is_dir() assert (root / ".git" / "opencode").read_text(encoding="ascii") == project_id + "\n" tool_files = list((root / ".opencode" / "tools").glob("*.js")) assert [path.stem for path in tool_files] == [tool_id] assert (root / ".opencode" / "node_modules").is_dir() package = json.loads((root / ".opencode" / "package.json").read_text()) lock = json.loads((root / ".opencode" / "package-lock.json").read_text()) assert package["dependencies"] == {"@opencode-ai/plugin": "0.0.0"} assert lock["packages"][""]["dependencies"] == package["dependencies"] source = tool_files[0].read_text(encoding="utf-8") assert forbidden_text not in source assert json.dumps(_OPENCODE_SYNTHETIC_TOOL_QUERY) in source assert json.dumps(_OPENCODE_SYNTHETIC_TOOL_RESULT) in source def test_resolve_opencode_path_precedence(monkeypatch): monkeypatch.setenv("SKILLOPT_SLEEP_OPENCODE_PATH", "env-opencode") with mock.patch("shutil.which", side_effect=lambda value: os.path.abspath(f"resolved-{value}")): assert resolve_opencode_path("explicit-opencode") == os.path.abspath("resolved-explicit-opencode") assert resolve_opencode_path() == os.path.abspath("resolved-env-opencode") def test_resolve_opencode_path_falls_back_to_command(monkeypatch): monkeypatch.delenv("SKILLOPT_SLEEP_OPENCODE_PATH", raising=False) with mock.patch("shutil.which", return_value=None): assert resolve_opencode_path() == "opencode" @pytest.mark.skipif(os.name != "nt", reason="Windows PATHEXT shim behavior") def test_resolve_opencode_path_preserves_windows_cmd_shim(monkeypatch): monkeypatch.delenv("SKILLOPT_SLEEP_OPENCODE_PATH", raising=False) executable = r"C:\npm\opencode.CMD" with mock.patch("shutil.which", return_value=executable): assert resolve_opencode_path() == executable @pytest.mark.parametrize("which_result", [None, os.path.join("bin", "opencode")]) @pytest.mark.parametrize("source", ["explicit", "environment"]) def test_resolve_opencode_path_anchors_relative_paths(monkeypatch, tmp_path, which_result, source): monkeypatch.chdir(tmp_path) relative = os.path.join("bin", "opencode") explicit = relative if source == "explicit" else "" if source == "environment": monkeypatch.setenv("SKILLOPT_SLEEP_OPENCODE_PATH", relative) with mock.patch("shutil.which", return_value=which_result): assert resolve_opencode_path(explicit) == os.path.abspath(relative) def test_constructor_uses_explicit_or_environment_model(monkeypatch): monkeypatch.setenv("SKILLOPT_SLEEP_OPENCODE_MODEL", "env/model") with mock.patch("shutil.which", return_value=None): assert OpenCodeCliBackend(model="explicit/model").model == "explicit/model" assert OpenCodeCliBackend().model == "env/model" def test_backend_runtime_uses_current_mutable_settings(): be = OpenCodeCliBackend(model="old/model", opencode_path="old-opencode", timeout=1) be.model = "new/model" be.opencode_path = "new-opencode" be.timeout = 7 with mock.patch( "skillopt_sleep.backend.subprocess.run", side_effect=_successful_plain_results(), ) as run: assert be._call("hello") == "answer" assert all(call.args[0][0] == "new-opencode" for call in run.call_args_list) assert all(call.kwargs["timeout"] == 7 for call in run.call_args_list) command = run.call_args_list[-1].args[0] assert command[command.index("--model") + 1] == "new/model" def test_parse_opencode_jsonl_collects_text_and_ignores_extensions(): raw = "\n".join( [ _event("step_start", part={"type": "step-start"}), _event("future_event"), _event("reasoning", part={"type": "reasoning", "text": "hidden"}), _event("text", part={"type": "text", "text": "first"}), "", _event("text", part={"type": "text", "text": "second"}), _event("step_finish", part={"type": "step-finish"}), ] ) assert _parse_opencode_jsonl_events(raw) == ("first\nsecond", [], "") @pytest.mark.parametrize( ("raw", "expected_code"), [ ("not json", "malformed_jsonl"), ('{"value":' + "9" * 5000 + "}", "malformed_jsonl"), (json.dumps(["not", "an", "object"]), "invalid_event"), (json.dumps({"type": "text"}), "invalid_event"), ( "\n".join( [ _event("step_start", session="a", part={"type": "step-start"}), _event("step_finish", session="b", part={"type": "step-finish"}), ] ), "mixed_session", ), (_event("error"), "error_event"), (_event("reasoning", part={"type": "text"}), "invalid_event"), ( "\n".join( [ _event("step_start", part={"type": "step-start"}), _event("tool_use", part={"type": "tool", "tool": "shell"}), ] ), "unexpected_tool_event", ), (_event("step_start", part={"type": "step-start"}), "incomplete_stream"), (_event("step_finish", part={"type": "step-finish"}), "incomplete_stream"), (_success_stream(" "), "empty_response"), ], ) def test_parse_opencode_jsonl_rejects_invalid_streams(raw, expected_code): assert _parse_opencode_jsonl_events(raw) == ("", [], expected_code) def test_parse_opencode_tool_jsonl_accepts_only_expected_completed_tools(): tool_id = "skillopt_replay_abc123" raw = _tool_success_stream(tool_id, "final answer") assert _parse_opencode_jsonl_events(raw, {tool_id}) == ( "final answer", [tool_id], "", ) @pytest.mark.parametrize( ("raw", "expected_code"), [ (_tool_success_stream("other"), "unexpected_tool_id"), ( "\n".join( [ _event("step_start", part={"type": "step-start"}), _tool_event("expected", status="error"), ] ), "tool_error", ), ( "\n".join( [ _event("step_start", part={"type": "step-start"}), _tool_event("expected", output="unexpected result"), _event("step_finish", part={"type": "step-finish"}), ] ), "invalid_tool_event", ), ( "\n".join( [ _event("step_start", part={"type": "step-start"}), _tool_event( "expected", query="task-derived query", output=_OPENCODE_SYNTHETIC_TOOL_RESULT, ), ] ), "invalid_tool_event", ), ( "\n".join( [ _event("step_start", part={"type": "step-start"}), _tool_event("expected", output=_OPENCODE_SYNTHETIC_TOOL_RESULT), _event("step_finish", part={"type": "step-finish"}), ] ), "missing_final_text", ), ], ) def test_parse_opencode_tool_jsonl_rejects_unverified_tools(raw, expected_code): assert _parse_opencode_jsonl_events(raw, {"expected"}) == ("", [], expected_code) def test_call_uses_stdin_temp_workspace_and_user_environment(monkeypatch, tmp_path): monkeypatch.setenv("OPENAI_API_KEY", "ambient-provider-key") monkeypatch.setenv("HOME", "/home/example") monkeypatch.setenv("PWD", "/original/project") monkeypatch.setenv("OLDPWD", "/previous/project") inline_config = json.dumps( { "default_agent": "unsafe-user-agent", "agent": {"unsafe-user-agent": {"permission": {"bash": "allow", "edit": "allow"}}}, } ) monkeypatch.setenv("OPENCODE_CONFIG_CONTENT", inline_config) config_path = str(tmp_path / "custom-opencode.json") config_dir = str(tmp_path / ".opencode") monkeypatch.setenv("OPENCODE_CONFIG", config_path) monkeypatch.setenv("OPENCODE_CONFIG_DIR", config_dir) for key in ( "NO_COLOR", "OPENCODE_DISABLE_AUTOUPDATE", "OPENCODE_DISABLE_EXTERNAL_SKILLS", "OPENCODE_DISABLE_PROJECT_CONFIG", "OPENCODE_DISABLE_SHARE", "OPENCODE_DISABLE_TERMINAL_TITLE", "OPENCODE_PURE", ): monkeypatch.setenv(key, "0") monkeypatch.setenv("OPENCODE_PERMISSION", '{"*":"allow"}') monkeypatch.delenv("OPENCODE_DISABLE_DEFAULT_PLUGINS", raising=False) captured = [] results = iter(_successful_plain_results("alpha", "beta")) def fake_run(cmd, **kwargs): snapshot = dict(kwargs) snapshot["env"] = kwargs["env"].copy() captured.append((cmd, snapshot)) assert os.path.isdir(kwargs["cwd"]) assert os.listdir(kwargs["cwd"]) == [] return next(results) executable = str(tmp_path / "opencode") be = OpenCodeCliBackend( model="provider/model", opencode_path=executable, timeout=37, ) with mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=fake_run): assert be._call("do the thing") == "answer" discovery_cmd, discovery_call = captured[0] verification_cmd, verification_call = captured[1] cmd, run_call = captured[2] expected_child_env = { "NO_COLOR": "1", "OPENCODE_DISABLE_AUTOUPDATE": "1", "OPENCODE_DISABLE_EXTERNAL_SKILLS": "1", "OPENCODE_DISABLE_PROJECT_CONFIG": "1", "OPENCODE_DISABLE_SHARE": "1", "OPENCODE_DISABLE_TERMINAL_TITLE": "1", "OPENCODE_PERMISSION": '{"*":"deny"}', "OPENCODE_PURE": "1", } for _, call in captured: assert call["capture_output"] is True assert call["creationflags"] == _NO_WINDOW assert call["text"] is True assert call["encoding"] == "utf-8" assert call["errors"] == "replace" assert call["timeout"] == be.timeout assert call["cwd"] == run_call["cwd"] assert call["env"]["PWD"] == call["cwd"] assert "OLDPWD" not in call["env"] for key, value in expected_child_env.items(): assert call["env"][key] == value assert discovery_cmd == [executable, "debug", "config", "--pure"] assert verification_cmd == discovery_cmd assert cmd[:5] == [ executable, "run", "--pure", "--format", "json", ] agent_name = cmd[cmd.index("--agent") + 1] assert agent_name.startswith("skillopt-sleep-") assert cmd[cmd.index("--title") + 1] == "skillopt-sleep" assert cmd[cmd.index("--dir") + 1] == run_call["cwd"] assert cmd[cmd.index("--model") + 1] == "provider/model" assert run_call["input"] == "do the thing" assert "input" not in discovery_call assert "input" not in verification_call assert "do the thing" not in cmd assert run_call["env"]["OPENAI_API_KEY"] == "ambient-provider-key" assert run_call["env"]["HOME"] == "/home/example" assert run_call["env"]["PWD"] != "/original/project" assert run_call["env"]["OPENCODE_CONFIG"] == config_path assert run_call["env"]["OPENCODE_CONFIG_DIR"] == config_dir discovered = json.loads(discovery_call["env"]["OPENCODE_CONFIG_CONTENT"]) assert "mcp" not in discovered injected = json.loads(run_call["env"]["OPENCODE_CONFIG_CONTENT"]) assert injected == { "snapshot": False, "agent": { agent_name: { "mode": "primary", "permission": {"*": "deny"}, } }, "mcp": { "alpha": {"enabled": False}, "beta": {"enabled": False}, }, } assert verification_call["env"]["OPENCODE_CONFIG_CONTENT"] == run_call["env"]["OPENCODE_CONFIG_CONTENT"] assert "OPENCODE_DISABLE_DEFAULT_PLUGINS" not in run_call["env"] assert not os.path.exists(run_call["cwd"]) assert be.last_call_error == "" def test_relative_opencode_config_paths_become_absolute(monkeypatch, tmp_path): monkeypatch.chdir(tmp_path) config_path = os.path.join("profiles", "opencode.json") config_dir = os.path.join("profiles", "opencode") monkeypatch.setenv("OPENCODE_CONFIG", config_path) monkeypatch.setenv("OPENCODE_CONFIG_DIR", config_dir) captured = [] results = iter(_successful_plain_results()) def fake_run(cmd, **kwargs): captured.append(kwargs["env"].copy()) return next(results) be = OpenCodeCliBackend(opencode_path=str(tmp_path / "opencode")) with mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=fake_run): assert be._call("hello") == "answer" assert all(env["OPENCODE_CONFIG"] == os.path.abspath(config_path) for env in captured) assert all(env["OPENCODE_CONFIG_DIR"] == os.path.abspath(config_dir) for env in captured) def test_empty_model_leaves_model_selection_to_opencode(monkeypatch): monkeypatch.delenv("SKILLOPT_SLEEP_OPENCODE_MODEL", raising=False) be = OpenCodeCliBackend(opencode_path="opencode") with mock.patch( "skillopt_sleep.backend.subprocess.run", side_effect=_successful_plain_results(), ) as run: assert be._call("hello") == "answer" assert "--model" not in run.call_args_list[-1].args[0] def test_each_call_uses_a_new_agent_name(): be = OpenCodeCliBackend(opencode_path="opencode") with ( mock.patch("secrets.token_hex", side_effect=["a" * 32, "b" * 32]), mock.patch( "skillopt_sleep.backend.subprocess.run", side_effect=_successful_plain_results() + _successful_plain_results(), ) as run, ): assert be._call("first") == "answer" assert be._call("second") == "answer" commands = [call.args[0] for call in run.call_args_list if call.args[0][1] == "run"] assert commands[0][commands[0].index("--agent") + 1] == ("skillopt-sleep-" + "a" * 32) assert commands[1][commands[1].index("--agent") + 1] == ("skillopt-sleep-" + "b" * 32) @pytest.mark.parametrize( ("run_result", "error_fragment"), [ (subprocess.TimeoutExpired("opencode", 1), "timed out"), (OSError("secret path"), "could not be executed"), (_FakeProc("misleading", returncode=2), "exited 2"), (_FakeProc("not json"), "malformed JSONL"), ], ) def test_call_records_process_and_protocol_failures(run_result, error_fragment): be = OpenCodeCliBackend(opencode_path="opencode", timeout=1) effects = _successful_plain_results()[:2] + [run_result] with mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=effects): assert be._call("hello") == "" assert error_fragment in be.last_call_error assert "secret path" not in be.last_call_error def test_call_handles_workspace_creation_failure_without_starting_child(): be = OpenCodeCliBackend(opencode_path="opencode") with ( mock.patch( "skillopt_sleep.backend.tempfile.TemporaryDirectory", side_effect=OSError("private workspace detail"), ), mock.patch("skillopt_sleep.backend.subprocess.run") as run, ): assert be._call("hello") == "" run.assert_not_called() assert "workspace could not be prepared" in be.last_call_error assert "private workspace detail" not in be.last_call_error @pytest.mark.parametrize( ("discovered_names", "final_mcp"), [ pytest.param( ("global-server",), { "global-server": { "type": "local", "command": ["mcp-server"], "enabled": True, } }, id="known-server-re-enabled", ), pytest.param( ("known-server",), { "known-server": {"enabled": False}, "new-server": { "type": "local", "command": ["mcp-server"], "enabled": True, }, }, id="new-server-enabled", ), ], ) def test_enabled_mcp_in_final_config_stops_before_model_call(discovered_names, final_mcp): discovered = _FakeProc(_resolved_mcp(*discovered_names)) verification = _FakeProc(json.dumps({"mcp": final_mcp})) be = OpenCodeCliBackend(opencode_path="opencode") with mock.patch( "skillopt_sleep.backend.subprocess.run", side_effect=[discovered, verification], ) as run: assert be._call("hello") == "" assert run.call_count == 2 assert all(call.args[0][1:3] == ["debug", "config"] for call in run.call_args_list) assert "disable every configured MCP server" in be.last_call_error def test_plain_replay_fails_if_snapshots_remain_enabled(): be = OpenCodeCliBackend(opencode_path="opencode") results = [ _FakeProc(_resolved_mcp(snapshot=False)), _FakeProc(_resolved_mcp(snapshot=True)), ] with mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=results) as run: assert be._call("hello") == "" assert run.call_count == 2 assert "disable session snapshots" in be.last_call_error @pytest.mark.parametrize( ("bad_result", "error_fragment"), [ (subprocess.TimeoutExpired("opencode", 1), "timed out"), (OSError("private config path"), "could not be completed"), (_FakeProc("", returncode=3), "exited 3"), (_FakeProc("not json"), "invalid configuration"), (_FakeProc('{"mcp":' + "9" * 5000 + "}"), "invalid configuration"), (_FakeProc(json.dumps(["not", "an", "object"])), "invalid configuration"), (_FakeProc(json.dumps({"mcp": []})), "invalid MCP configuration"), ( _FakeProc(json.dumps({"mcp": {"server": "not an object"}})), "invalid MCP configuration", ), ], ) def test_mcp_discovery_failure_stops_before_model_call(bad_result, error_fragment): be = OpenCodeCliBackend(opencode_path="opencode", timeout=1) with mock.patch( "skillopt_sleep.backend.subprocess.run", side_effect=[bad_result], ) as run: assert be._call("hello") == "" assert run.call_count == 1 assert run.call_args.args[0][1:3] == ["debug", "config"] assert error_fragment in be.last_call_error assert "private config path" not in be.last_call_error def test_mcp_verification_failure_stops_before_model_call(): be = OpenCodeCliBackend(opencode_path="opencode") with mock.patch( "skillopt_sleep.backend.subprocess.run", side_effect=[_FakeProc(_resolved_mcp("server")), _FakeProc("not json")], ) as run: assert be._call("hello") == "" assert run.call_count == 2 assert all(call.args[0][1:3] == ["debug", "config"] for call in run.call_args_list) assert "MCP verification" in be.last_call_error def test_resolved_config_secrets_are_not_exposed(caplog): secret = "resolved-provider-secret-74f92" discovered = _FakeProc( json.dumps( { "provider": {"custom": {"options": {"apiKey": secret}}}, "mcp": { "private-server": { "type": "remote", "url": "https://mcp.invalid", "headers": {"Authorization": secret}, } }, } ) ) verification = _FakeProc(secret, stderr=secret, returncode=5) be = OpenCodeCliBackend(opencode_path="opencode") with mock.patch( "skillopt_sleep.backend.subprocess.run", side_effect=[discovered, verification], ) as run: assert be._call("hello") == "" assert run.call_count == 2 assert secret not in be.last_call_error assert secret not in caplog.text assert secret not in run.call_args.kwargs["env"]["OPENCODE_CONFIG_CONTENT"] def test_mcp_checks_run_once_per_cache_miss(): be = OpenCodeCliBackend(opencode_path="opencode") results = _successful_plain_results("server") + _successful_plain_results("server") with mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=results) as run: assert be._cached_call("attempt:first", "first prompt") == "answer" assert be._cached_call("attempt:first", "first prompt") == "answer" assert be._cached_call("attempt:second", "second prompt") == "answer" commands = [call.args[0][1:3] for call in run.call_args_list] assert commands == [ ["debug", "config"], ["debug", "config"], ["run", "--pure"], ["debug", "config"], ["debug", "config"], ["run", "--pure"], ] def test_failed_call_is_not_cached(): be = OpenCodeCliBackend(opencode_path="opencode") with mock.patch.object(be, "_call", side_effect=["", "recovered"]) as call: assert be._cached_call("attempt:key", "prompt") == "" assert be._cached_call("attempt:key", "prompt") == "recovered" assert call.call_count == 2 def test_cached_success_clears_stale_call_error(): be = OpenCodeCliBackend(opencode_path="opencode") with mock.patch.object(be, "_call", return_value="answer") as call: assert be._cached_call("attempt:key", "prompt") == "answer" be.last_call_error = "unrelated later failure" assert be._cached_call("attempt:key", "prompt") == "answer" assert call.call_count == 1 assert be.last_call_error == "" def test_attempt_with_tools_requires_explicit_opt_in_without_starting_child(): be = OpenCodeCliBackend(opencode_path="opencode") with mock.patch("skillopt_sleep.backend.subprocess.run") as run: assert be.attempt_with_tools(mock.Mock(), "skill", "memory", ["search"]) == ("", []) run.assert_not_called() assert "opencode_tool_replay" in be.last_call_error @pytest.mark.parametrize("configured", [False, 0, 1, "false", "true", None]) def test_tool_replay_opt_in_requires_boolean_true(configured): be = OpenCodeCliBackend( opencode_path="opencode", tool_replay=configured, ) assert be.tool_replay is False def test_attempt_with_tools_builds_isolated_random_replay_request(monkeypatch, tmp_path): monkeypatch.setenv("OPENAI_API_KEY", "provider-secret-stays-in-env") monkeypatch.setenv("OPENCODE_DIRECT_TRACE", "1") for key in ("GIT_COMMON_DIR", "GIT_DIR", "GIT_OBJECT_DIRECTORY", "GIT_WORK_TREE"): monkeypatch.setenv(key, str(tmp_path / "outside-repository")) monkeypatch.setenv("git_dir", str(tmp_path / "lowercase-outside-repository")) hostile_npm = { "NPM_CONFIG_AUDIT": "true", "NpM_CoNfIg_CaChE": str(tmp_path / "outside-npm-cache"), "NPM_CONFIG_FETCH_RETRIES": "99", "npm_CONFIG_FETCH_RETRY_MAXTIMEOUT": "99999", "NPM_CONFIG_FETCH_RETRY_MINTIMEOUT": "99999", "NPM_CONFIG_FETCH_TIMEOUT": "99999", "nPm_CoNfIg_FuNd": "true", "NPM_CONFIG_OFFLINE": "false", "NPM_CONFIG_UPDATE_NOTIFIER": "true", } for key, value in hostile_npm.items(): monkeypatch.setenv(key, value) captured = [] project_id = "skillopt-sleep-" + "a" * 64 agent_name = "skillopt-sleep-" + "b" * 32 tool_id = "skillopt_replay_" + "c" * 32 expected_permission = {"*": "deny", tool_id: "allow"} intent = "OPENCODE_INTENT_SENTINEL_1937" secret_context = "task-secret-value-4819" skill = "OPENCODE_SKILL_SENTINEL_2864" memory = "OPENCODE_MEMORY_SENTINEL_7351" def fake_run(cmd, **kwargs): snapshot = dict(kwargs) snapshot["env"] = kwargs["env"].copy() captured.append((cmd, snapshot)) _assert_controlled_tool_environment(kwargs["env"], kwargs["cwd"]) if cmd[1:3] == ["debug", "config"]: config = json.loads(kwargs["env"]["OPENCODE_CONFIG_CONTENT"]) _assert_replay_permissions(config, kwargs["env"], agent_name, expected_permission) disabled = bool(config.get("mcp")) return _FakeProc( _resolved_mcp( "configured-server", disabled=disabled, snapshot=False, ) ) if cmd[1:3] == ["debug", "agent"]: config = json.loads(kwargs["env"]["OPENCODE_CONFIG_CONTENT"]) _assert_replay_permissions(config, kwargs["env"], agent_name, expected_permission) assert cmd[3] == agent_name agent = config["agent"][agent_name] assert agent["model"] == "provider/model" return _FakeProc(json.dumps({"tools": {"bash": False, tool_id: True}})) work = kwargs["cwd"] config = json.loads(kwargs["env"]["OPENCODE_CONFIG_CONTENT"]) _assert_replay_permissions(config, kwargs["env"], agent_name, expected_permission) assert config["snapshot"] is False assert cmd[cmd.index("--agent") + 1] == agent_name _assert_replay_project_artifacts( work, project_id=project_id, tool_id=tool_id, forbidden_text=secret_context, ) return _FakeProc(_tool_success_stream(tool_id)) be = OpenCodeCliBackend( model="provider/model", opencode_path="opencode", timeout=41, tool_replay=True, ) task = TaskRecord( id="opencode-tool-prompt", project=str(tmp_path), intent=intent, context_excerpt=secret_context, ) with ( mock.patch( "skillopt_sleep.backend.secrets.token_hex", side_effect=["a" * 64, "b" * 32, "c" * 32], ), mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=fake_run), ): response, called = be.attempt_with_tools(task, skill, memory, ["search"]) assert response == "answer" assert called == ["search"] assert be.last_call_error == "" assert [call[0][1:3] for call in captured] == [ ["debug", "config"], ["debug", "config"], ["debug", "agent"], ["run", "--pure"], ] run_call = captured[-1][1] assert run_call["env"]["OPENCODE_DISABLE_PROJECT_CONFIG"] == "0" assert run_call["env"]["OPENCODE_PURE"] == "1" assert run_call["env"]["OPENAI_API_KEY"] == "provider-secret-stays-in-env" assert "OPENCODE_DIRECT_TRACE" not in run_call["env"] for key in ("GIT_COMMON_DIR", "GIT_DIR", "GIT_OBJECT_DIRECTORY", "GIT_WORK_TREE"): assert not any(candidate.upper() == key for candidate in run_call["env"]) for sentinel in (intent, secret_context, skill, memory): assert sentinel in run_call["input"] assert '{"query":"synthetic"}' in run_call["input"] assert "do not call a tool merely because it is listed" in run_call["input"] assert "Learned preferences" in run_call["input"] assert "override earlier conflicting skill text" in run_call["input"] assert "Call every listed random internal ID" not in run_call["input"] assert not os.path.exists(run_call["cwd"]) @pytest.mark.parametrize( "tools", [ pytest.param(["../search"], id="invalid-prefix"), pytest.param([], id="empty"), pytest.param(["search", 123], id="non-string"), pytest.param(["x" * 129], id="too-long"), pytest.param([f"tool-{index}" for index in range(33)], id="too-many"), ], ) def test_attempt_with_tools_rejects_invalid_names_before_starting_child(tools, tmp_path): be = OpenCodeCliBackend(opencode_path="opencode", tool_replay=True) task = TaskRecord(id="invalid-tools", project=str(tmp_path), intent="answer") with mock.patch("skillopt_sleep.backend.subprocess.run") as run: assert be.attempt_with_tools(task, "", "", tools) == ("", []) run.assert_not_called() assert "invalid tool list" in be.last_call_error @pytest.mark.parametrize( ("child_result", "expected_fragment"), [ pytest.param( _FakeProc(json.dumps({"tools": {"expected": False}})), "restrict tools", id="expected-disabled", ), pytest.param( _FakeProc(json.dumps({"tools": {"expected": True, "bash": True}})), "restrict tools", id="extra-enabled", ), pytest.param( _FakeProc(json.dumps({"tools": {"expected": "allow"}})), "invalid configuration", id="non-boolean-status", ), pytest.param( _FakeProc(json.dumps({"tools": []})), "invalid configuration", id="invalid-tools-shape", ), pytest.param( subprocess.TimeoutExpired(cmd=["opencode"], timeout=1), "timed out", id="timeout", ), pytest.param(OSError("private child detail"), "could not be completed", id="spawn-error"), pytest.param(_FakeProc("", returncode=9), "failed", id="nonzero-exit"), pytest.param(_FakeProc("not-json"), "invalid configuration", id="invalid-json"), ], ) def test_tool_permission_verification_fails_closed(child_result, expected_fragment, tmp_path): backend = OpenCodeCliBackend(opencode_path=str(tmp_path / "opencode"), timeout=1) with ( mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=[child_result]) as run, pytest.raises(OpenCodeError) as error, ): backend._verify_tool_allowlist({}, str(tmp_path), "skillopt-sleep-agent", {"expected"}) run.assert_called_once() assert expected_fragment in str(error.value) assert "private child detail" not in str(error.value) def test_tool_replay_fails_before_model_if_snapshots_remain_enabled(): captured_work = "" def fake_run(cmd, **kwargs): nonlocal captured_work captured_work = kwargs["cwd"] assert cmd[1:3] == ["debug", "config"] snapshot = len(run.call_args_list) == 2 return _FakeProc(_resolved_mcp(snapshot=snapshot)) be = OpenCodeCliBackend(opencode_path="opencode", tool_replay=True) task = mock.Mock(intent="search", context_excerpt="") with mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=fake_run) as run: assert be.attempt_with_tools(task, "", "", ["search"]) == ("", []) assert run.call_count == 2 assert "disable session snapshots" in be.last_call_error assert captured_work and not os.path.exists(captured_work) @pytest.mark.parametrize( ("run_result", "error_fragment"), [ (subprocess.TimeoutExpired(cmd=["opencode"], timeout=1), "timed out"), (OSError("private run detail"), "could not be executed"), (_FakeProc("", returncode=9), "tool replay failed"), ], ) def test_tool_replay_handles_model_child_failures(run_result, error_fragment): captured_work = "" def fake_run(cmd, **kwargs): nonlocal captured_work captured_work = kwargs["cwd"] if cmd[1:3] == ["debug", "config"]: return _FakeProc(_resolved_mcp(snapshot=False)) if cmd[1:3] == ["debug", "agent"]: config = json.loads(kwargs["env"]["OPENCODE_CONFIG_CONTENT"]) permission = config["agent"][cmd[3]]["permission"] allowed = {name: True for name, value in permission.items() if value == "allow"} return _FakeProc(json.dumps({"tools": allowed})) if isinstance(run_result, BaseException): raise run_result return run_result be = OpenCodeCliBackend( opencode_path="opencode", timeout=1, tool_replay=True, ) task = mock.Mock(intent="search", context_excerpt="private prompt detail") with mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=fake_run) as run: assert be.attempt_with_tools(task, "", "", ["search"]) == ("", []) assert run.call_count == 4 assert be.tokens_used() > 0 assert error_fragment in be.last_call_error assert "private" not in be.last_call_error assert captured_work and not os.path.exists(captured_work) def test_tool_replay_maps_and_deduplicates_only_completed_requested_tools(tmp_path): search_id = "skillopt_replay_" + "c" * 32 lookup_id = "skillopt_replay_" + "d" * 32 def fake_run(cmd, **kwargs): if cmd[1:3] == ["debug", "config"]: return _FakeProc(_resolved_mcp(snapshot=False)) config = json.loads(kwargs["env"]["OPENCODE_CONFIG_CONTENT"]) if cmd[1:3] == ["debug", "agent"]: permission = config["agent"][cmd[3]]["permission"] assert permission == {"*": "deny", search_id: "allow", lookup_id: "allow"} return _FakeProc(json.dumps({"tools": {search_id: True, lookup_id: True}})) return _FakeProc( "\n".join( [ _event("step_start", part={"type": "step-start"}), _tool_event( lookup_id, call_id="call-1", output=_OPENCODE_SYNTHETIC_TOOL_RESULT, ), _tool_event( lookup_id, call_id="call-2", output=_OPENCODE_SYNTHETIC_TOOL_RESULT, ), _event("step_finish", part={"type": "step-finish"}), _event("step_start", part={"type": "step-start"}), _event("text", part={"type": "text", "text": "answer"}), _event("step_finish", part={"type": "step-finish"}), ] ) ) be = OpenCodeCliBackend(opencode_path="opencode", tool_replay=True) task = TaskRecord(id="partial-tool-replay", project=str(tmp_path), intent="use lookup") with ( mock.patch( "skillopt_sleep.backend.secrets.token_hex", side_effect=["a" * 64, "b" * 32, "c" * 32, "d" * 32], ), mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=fake_run), ): response, called = be.attempt_with_tools(task, "", "", ["search", "lookup"]) assert response == "answer" assert called == ["lookup"] def test_tool_replay_does_not_treat_a_self_report_as_a_tool_call(): def fake_run(cmd, **kwargs): if cmd[1:3] == ["debug", "config"]: return _FakeProc(_resolved_mcp(snapshot=False)) if cmd[1:3] == ["debug", "agent"]: config = json.loads(kwargs["env"]["OPENCODE_CONFIG_CONTENT"]) permission = config["agent"][cmd[3]]["permission"] return _FakeProc( json.dumps({"tools": {name: value == "allow" for name, value in permission.items() if name != "*"}}) ) return _FakeProc(_success_stream("TOOL_CALL: search")) be = OpenCodeCliBackend(opencode_path="opencode", tool_replay=True) task = mock.Mock(intent="answer", context_excerpt="") with mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=fake_run): assert be.attempt_with_tools(task, "", "", ["search"]) == ( "TOOL_CALL: search", [], ) assert be.last_call_error == "" def test_get_and_build_backend_route_opencode_path(): with mock.patch("shutil.which", return_value=None): for alias in ("opencode", "opencode_cli", "opencode-cli", "OPENCODE"): assert isinstance(get_backend(alias), OpenCodeCliBackend) single = build_backend( backend="opencode", opencode_path="custom-opencode", opencode_tool_replay=True, ) assert isinstance(single, OpenCodeCliBackend) assert single.opencode_path == "custom-opencode" assert single.tool_replay is True stringly_enabled = build_backend( backend="opencode", opencode_path="custom-opencode", opencode_tool_replay="true", ) assert stringly_enabled.tool_replay is False dual = build_backend( backend="mock", optimizer_backend="opencode", target_backend="opencode", opencode_path="custom-opencode", opencode_tool_replay=True, ) assert isinstance(dual, DualBackend) assert dual.optimizer.opencode_path == "custom-opencode" assert dual.target.opencode_path == "custom-opencode" assert dual.optimizer.tool_replay is True assert dual.target.tool_replay is True def test_cli_makes_relative_opencode_path_absolute(monkeypatch, tmp_path): monkeypatch.chdir(tmp_path) parser = argparse.ArgumentParser() _add_common(parser) relative = os.path.join("bin", "opencode") args = parser.parse_args( [ "--backend", "opencode", "--opencode-path", relative, "--opencode-tool-replay", ] ) monkeypatch.setattr("skillopt_sleep.config._user_config_path", lambda: None) cfg = _cfg_from_args(args) assert cfg.get("backend") == "opencode" assert cfg.get("opencode_path") == os.path.abspath(relative) assert cfg.get("opencode_tool_replay") is True def test_cli_without_tool_replay_flag_preserves_enabled_user_config(monkeypatch, tmp_path): parser = argparse.ArgumentParser() _add_common(parser) args = parser.parse_args(["--backend", "opencode"]) config_path = tmp_path / "config.json" config_path.write_text(json.dumps({"opencode_tool_replay": True}), encoding="utf-8") monkeypatch.setattr("skillopt_sleep.config._user_config_path", lambda: str(config_path)) cfg = _cfg_from_args(args) assert cfg.get("opencode_tool_replay") is True def test_cycle_diagnostic_build_forwards_opencode_settings(): cfg = load_config( backend="opencode", opencode_path="custom-opencode", opencode_tool_replay=True, ) with mock.patch("skillopt_sleep.cycle.build_backend", return_value=MockBackend()) as builder: cycle._make_model_key(cfg) builder.assert_called_once() assert builder.call_args.kwargs["opencode_path"] == "custom-opencode" assert builder.call_args.kwargs["opencode_tool_replay"] is True @pytest.mark.parametrize("configured", [True, "true"]) def test_runtime_cycle_build_forwards_opencode_settings(tmp_path, configured): project = tmp_path / "project" project.mkdir() cfg = SleepConfig( data={ **DEFAULTS, "backend": "opencode", "opencode_path": "custom-opencode", "opencode_tool_replay": configured, "projects": "invoked", "invoked_project": str(project), "state_dir": str(tmp_path / "state"), "claude_home": str(tmp_path / "claude-home"), "evidence_log": False, } ) with mock.patch("skillopt_sleep.cycle.build_backend", return_value=MockBackend()) as builder: cycle.run_sleep_cycle(cfg, seed_tasks=[], dry_run=True) assert builder.call_count == 1 assert builder.call_args.kwargs["opencode_path"] == "custom-opencode" assert builder.call_args.kwargs["opencode_tool_replay"] == configured @pytest.mark.parametrize( ("configured", "expected"), [(True, True), ("true", False)], ) def test_cycle_artifacts_record_only_a_boolean_tool_replay_opt_in( tmp_path, configured, expected, ): project = tmp_path / "project" project.mkdir() state_dir = tmp_path / "state" cfg = SleepConfig( data={ **DEFAULTS, "backend": "opencode", "opencode_tool_replay": configured, "projects": "invoked", "invoked_project": str(project), "state_dir": str(state_dir), "claude_home": str(tmp_path / "claude-home"), "evidence_log": True, "evolve_skill": False, "evolve_memory": False, } ) task = TaskRecord( id="opencode-artifacts", project=str(project), intent="record the resolved replay setting", reference_kind="exact", reference="expected response", split="val", ) with mock.patch("skillopt_sleep.cycle.build_backend", return_value=MockBackend()): outcome = cycle.run_sleep_cycle(cfg, seed_tasks=[task]) staging_dir = Path(outcome.staging_dir) records = [json.loads(line) for line in (staging_dir / "evidence.jsonl").read_text(encoding="utf-8").splitlines()] start = next(record for record in records if record["stage"] == "cycle" and record["event"] == "start") assert start["config"]["opencode_tool_replay"] is expected diagnostics_path = staging_dir / "diagnostics.json" diagnostics = json.loads(diagnostics_path.read_text(encoding="utf-8")) assert diagnostics["opencode_tool_replay"] is expected