mirror of
https://github.com/razzant/ouroboros.git
synced 2026-08-04 16:19:50 +00:00
543cc8376c
Windows: the python-resolver tests built fake venvs in POSIX layout (bin/python) while project_venv_python correctly looks for Scripts\python.exe — the fixture is now platform-aware; the isolated-checkout test repos pin core.autocrlf=false so the LF staged patch applies in the detached worktree on autocrlf=true runners. ui-smoke: the desktop chat scroll probe re-injects its bubbles after the viewport resize dance — a resize can re-render the chat from the (empty) real history and silently drop injected nodes, which made the post-resize assertion an environment-dependent flake. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
488 lines
17 KiB
Python
488 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import pathlib
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from ouroboros.python_interpreter import resolve_process_python
|
|
from ouroboros.tools.registry import ToolContext, ToolRegistry
|
|
|
|
|
|
def _executable(path: pathlib.Path) -> pathlib.Path:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
path.chmod(0o755)
|
|
return path
|
|
|
|
|
|
def _venv_python(env_root: pathlib.Path) -> pathlib.Path:
|
|
"""Platform-correct fake venv interpreter (Scripts\\python.exe on Windows)."""
|
|
from ouroboros.platform_layer import IS_WINDOWS
|
|
|
|
if IS_WINDOWS:
|
|
return _executable(env_root / "Scripts" / "python.exe")
|
|
return _executable(env_root / "bin" / "python")
|
|
|
|
|
|
def _context(
|
|
tmp_path: pathlib.Path,
|
|
*,
|
|
workspace: pathlib.Path | None = None,
|
|
workspace_mode: str = "",
|
|
) -> ToolContext:
|
|
repo = tmp_path / "system_repo"
|
|
data = tmp_path / "data"
|
|
repo.mkdir(exist_ok=True)
|
|
data.mkdir(exist_ok=True)
|
|
if workspace is not None:
|
|
workspace.mkdir(parents=True, exist_ok=True)
|
|
return ToolContext(
|
|
repo_dir=repo,
|
|
system_repo_dir=repo,
|
|
drive_root=data,
|
|
workspace_root=workspace,
|
|
workspace_mode=workspace_mode,
|
|
task_id="python-resolver-test",
|
|
)
|
|
|
|
|
|
def _tool_args(tool_name: str, *, cwd: str = "", token: str = "python") -> dict:
|
|
if tool_name == "run_command":
|
|
return {"cmd": [token, "-V"], "cwd": cwd}
|
|
if tool_name == "run_script":
|
|
return {"script": "print('ok')", "interpreter": token, "cwd": cwd}
|
|
if tool_name == "start_service":
|
|
return {"name": "svc", "cmd": [token, "-V"], "cwd": cwd}
|
|
if tool_name == "verify_and_record":
|
|
return {
|
|
"contract_kind": "explicit_command",
|
|
"check": [token, "-V"],
|
|
"cwd": cwd,
|
|
}
|
|
raise AssertionError(tool_name)
|
|
|
|
|
|
def _interpreter(tool_name: str, args: dict) -> str:
|
|
if tool_name in {"run_command", "start_service"}:
|
|
return str(args["cmd"][0])
|
|
if tool_name == "run_script":
|
|
return str(args["interpreter"])
|
|
return str(args["check"][0])
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"tool_name",
|
|
["run_command", "run_script", "start_service", "verify_and_record"],
|
|
)
|
|
def test_system_surfaces_use_validated_agent_python(tmp_path, monkeypatch, tool_name):
|
|
ctx = _context(tmp_path)
|
|
agent_python = _executable(tmp_path / "agent" / "bin" / "python")
|
|
monkeypatch.setenv("OUROBOROS_AGENT_PYTHON", str(agent_python))
|
|
|
|
resolved, trace = resolve_process_python(
|
|
ctx,
|
|
tool_name,
|
|
_tool_args(tool_name),
|
|
runtime_mode="advanced",
|
|
)
|
|
|
|
assert _interpreter(tool_name, resolved) == str(agent_python)
|
|
assert trace is not None
|
|
assert trace.surface == "system_repo"
|
|
assert trace.environment == "ouroboros_agent"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"tool_name",
|
|
["run_command", "run_script", "start_service", "verify_and_record"],
|
|
)
|
|
def test_external_workspace_prefers_project_venv(tmp_path, monkeypatch, tool_name):
|
|
workspace = tmp_path / "workspace"
|
|
ctx = _context(tmp_path, workspace=workspace, workspace_mode="external")
|
|
project_python = _venv_python(workspace / ".venv")
|
|
(workspace / ".venv" / "pyvenv.cfg").write_text("home = /usr/bin\n", encoding="utf-8")
|
|
monkeypatch.setenv(
|
|
"OUROBOROS_AGENT_PYTHON",
|
|
str(_executable(tmp_path / "agent" / "bin" / "python")),
|
|
)
|
|
|
|
resolved, trace = resolve_process_python(
|
|
ctx,
|
|
tool_name,
|
|
_tool_args(tool_name),
|
|
runtime_mode="advanced",
|
|
)
|
|
|
|
assert _interpreter(tool_name, resolved) == str(project_python)
|
|
assert trace is not None
|
|
assert trace.surface == "external_workspace"
|
|
assert trace.environment == "project_venv"
|
|
|
|
|
|
def test_project_venv_platform_layouts(tmp_path, monkeypatch):
|
|
from ouroboros import platform_layer
|
|
|
|
env_root = tmp_path / ".venv"
|
|
env_root.mkdir()
|
|
(env_root / "pyvenv.cfg").write_text("home = test\n", encoding="utf-8")
|
|
|
|
posix_python = env_root / "bin" / "python"
|
|
posix_python.parent.mkdir()
|
|
posix_python.write_text("", encoding="utf-8")
|
|
posix_python.chmod(0o755)
|
|
monkeypatch.setattr(platform_layer, "IS_WINDOWS", False)
|
|
assert platform_layer.project_venv_python(tmp_path) == str(posix_python)
|
|
|
|
windows_python = env_root / "Scripts" / "python.exe"
|
|
windows_python.parent.mkdir()
|
|
windows_python.write_text("", encoding="utf-8")
|
|
windows_python.chmod(0o755)
|
|
monkeypatch.setattr(platform_layer, "IS_WINDOWS", True)
|
|
assert platform_layer.project_venv_python(tmp_path) == str(windows_python)
|
|
|
|
|
|
def test_user_files_project_venv_and_missing_venv_path_fallback(tmp_path, monkeypatch):
|
|
user_root = tmp_path / "user_files"
|
|
project = user_root / "project"
|
|
project.mkdir(parents=True)
|
|
monkeypatch.setenv("OUROBOROS_USER_FILES_ROOT", str(user_root))
|
|
ctx = _context(tmp_path)
|
|
project_python = _venv_python(project / ".venv")
|
|
(project / ".venv" / "pyvenv.cfg").write_text("home = /usr/bin\n", encoding="utf-8")
|
|
|
|
resolved, trace = resolve_process_python(
|
|
ctx,
|
|
"run_command",
|
|
_tool_args("run_command", cwd=str(project)),
|
|
runtime_mode="advanced",
|
|
)
|
|
assert resolved["cmd"][0] == str(project_python)
|
|
assert trace is not None and trace.surface == "user_files"
|
|
|
|
(project / ".venv" / "pyvenv.cfg").unlink()
|
|
unresolved, fallback = resolve_process_python(
|
|
ctx,
|
|
"run_command",
|
|
_tool_args("run_command", cwd=str(project)),
|
|
runtime_mode="advanced",
|
|
)
|
|
assert unresolved["cmd"][0] == "python"
|
|
assert fallback is not None
|
|
assert fallback.environment == "target_path"
|
|
assert fallback.fallback_reason == "project_venv_unavailable"
|
|
|
|
|
|
def test_executor_uses_backend_python_but_unmapped_task_drive_uses_agent(tmp_path, monkeypatch):
|
|
workspace = tmp_path / "workspace"
|
|
ctx = _context(tmp_path, workspace=workspace, workspace_mode="external")
|
|
ctx.executor_ref = {
|
|
"type": "docker_exec",
|
|
"id": "programbench",
|
|
"container_name": "programbench",
|
|
"network": "none",
|
|
"workspace_host_path": str(workspace),
|
|
"workspace_backend_path": "/workspace",
|
|
}
|
|
_venv_python(workspace / ".venv")
|
|
(workspace / ".venv" / "pyvenv.cfg").write_text("home = /usr/bin\n", encoding="utf-8")
|
|
agent_python = _executable(tmp_path / "agent" / "bin" / "python")
|
|
monkeypatch.setenv("OUROBOROS_AGENT_PYTHON", str(agent_python))
|
|
|
|
resolved, trace = resolve_process_python(
|
|
ctx,
|
|
"run_command",
|
|
_tool_args("run_command"),
|
|
runtime_mode="advanced",
|
|
)
|
|
assert resolved["cmd"][0] == "python3"
|
|
assert trace is not None and trace.environment == "backend_path"
|
|
|
|
local, local_trace = resolve_process_python(
|
|
ctx,
|
|
"run_command",
|
|
_tool_args("run_command", cwd="task_drive"),
|
|
runtime_mode="advanced",
|
|
)
|
|
assert local["cmd"][0] == str(agent_python)
|
|
assert local_trace is not None and local_trace.surface == "task_drive"
|
|
|
|
|
|
def test_reviewed_skill_environment_precedes_executor(tmp_path, monkeypatch):
|
|
import ouroboros.marketplace.isolated_deps as isolated_deps
|
|
import ouroboros.skill_loader as skill_loader
|
|
import ouroboros.skill_readiness as skill_readiness
|
|
|
|
workspace = tmp_path / "workspace"
|
|
ctx = _context(tmp_path, workspace=workspace, workspace_mode="external")
|
|
ctx.task_metadata = {"source": "skill_scheduled_task", "skill": "demo"}
|
|
ctx.executor_ref = {
|
|
"type": "docker_exec",
|
|
"id": "executor",
|
|
"container_name": "executor",
|
|
"network": "none",
|
|
"workspace_host_path": str(workspace),
|
|
"workspace_backend_path": "/workspace",
|
|
}
|
|
skill_dir = tmp_path / "skill"
|
|
skill_dir.mkdir()
|
|
skill_python = _executable(skill_dir / ".ouroboros_env" / "python" / "bin" / "python")
|
|
loaded = SimpleNamespace(name="demo", skill_dir=skill_dir)
|
|
monkeypatch.setattr(skill_loader, "find_skill", lambda *args, **kwargs: loaded)
|
|
monkeypatch.setattr(
|
|
skill_readiness,
|
|
"skill_readiness_for_execution",
|
|
lambda *args, **kwargs: SimpleNamespace(ready=True),
|
|
)
|
|
monkeypatch.setattr(isolated_deps, "read_deps_state", lambda *args, **kwargs: {"status": "installed"})
|
|
monkeypatch.setattr(isolated_deps, "python_runtime_binary", lambda *args, **kwargs: skill_python)
|
|
|
|
resolved, trace = resolve_process_python(
|
|
ctx,
|
|
"run_command",
|
|
_tool_args("run_command"),
|
|
runtime_mode="advanced",
|
|
)
|
|
|
|
assert resolved["cmd"][0] == str(skill_python)
|
|
assert trace is not None
|
|
assert trace.environment == "isolated_skill"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("tool_name", "args"),
|
|
[
|
|
("run_command", {"cmd": ["/usr/bin/python", "-V"]}),
|
|
("run_command", {"cmd": ["python3.12", "-V"]}),
|
|
("run_command", {"cmd": ["sh", "-c", "python -V"]}),
|
|
("run_command", {"cmd": ["env", "python", "-V"]}),
|
|
("run_script", {"script": "pass", "interpreter": "/usr/bin/python"}),
|
|
("run_script", {"script": "pass", "interpreter": "python3.12"}),
|
|
("verify_and_record", {"contract_kind": "explicit_command", "check": "python -V"}),
|
|
("verify_and_record", {"contract_kind": "artifact_observation", "check": ["python", "-V"]}),
|
|
("remote_exec", {"cmd": ["python", "-V"]}),
|
|
],
|
|
)
|
|
def test_noneligible_invocations_are_byte_for_byte_unchanged(tmp_path, tool_name, args):
|
|
ctx = _context(tmp_path)
|
|
|
|
resolved, trace = resolve_process_python(
|
|
ctx,
|
|
tool_name,
|
|
args,
|
|
runtime_mode="advanced",
|
|
)
|
|
|
|
assert resolved == args
|
|
assert trace is None
|
|
|
|
|
|
def test_light_run_script_default_cwd_uses_task_drive_agent_python(tmp_path, monkeypatch):
|
|
ctx = _context(tmp_path)
|
|
agent_python = _executable(tmp_path / "agent" / "bin" / "python")
|
|
monkeypatch.setenv("OUROBOROS_AGENT_PYTHON", str(agent_python))
|
|
|
|
resolved, trace = resolve_process_python(
|
|
ctx,
|
|
"run_script",
|
|
{"script": "print('ok')"},
|
|
runtime_mode="light",
|
|
)
|
|
|
|
assert resolved["interpreter"] == str(agent_python)
|
|
assert trace is not None and trace.surface == "task_drive"
|
|
|
|
|
|
def test_registry_guard_and_handler_receive_same_resolved_verify_argv(tmp_path, monkeypatch):
|
|
import ouroboros.safety as safety
|
|
import ouroboros.tools.registry as registry_module
|
|
|
|
ctx = _context(tmp_path)
|
|
agent_python = _executable(tmp_path / "agent" / "bin" / "python")
|
|
monkeypatch.setenv("OUROBOROS_AGENT_PYTHON", str(agent_python))
|
|
monkeypatch.setenv("OUROBOROS_RUNTIME_MODE", "advanced")
|
|
monkeypatch.setattr(
|
|
safety,
|
|
"_run_llm_check",
|
|
lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("unexpected safety LLM call")),
|
|
)
|
|
|
|
registry = ToolRegistry(repo_dir=ctx.repo_dir, drive_root=ctx.drive_root)
|
|
registry.set_context(ctx)
|
|
captured: dict[str, list[str]] = {}
|
|
original_guard = registry_module.process_shell_guard_args
|
|
|
|
def capture_guard(name, args, **kwargs):
|
|
guarded = original_guard(name, args, **kwargs)
|
|
captured["guard"] = list(guarded["cmd"])
|
|
return guarded
|
|
|
|
def capture_handler(_ctx, contract_kind, check, **kwargs):
|
|
assert contract_kind == "explicit_command"
|
|
captured["handler"] = list(check)
|
|
return "ok"
|
|
|
|
monkeypatch.setattr(registry_module, "process_shell_guard_args", capture_guard)
|
|
monkeypatch.setattr(registry, "_run_shell_safety_check", lambda *args, **kwargs: "")
|
|
registry._entries["verify_and_record"].handler = capture_handler
|
|
|
|
result = registry.execute(
|
|
"verify_and_record",
|
|
{"contract_kind": "explicit_command", "check": ["python", "-m", "pytest", "--version"]},
|
|
)
|
|
|
|
expected = [str(agent_python), "-m", "pytest", "--version"]
|
|
assert result == "ok"
|
|
assert captured == {"guard": expected, "handler": expected}
|
|
events_path = ctx.drive_logs() / "events.jsonl"
|
|
event = json.loads(events_path.read_text(encoding="utf-8").splitlines()[-1])
|
|
assert event["type"] == "python_interpreter_resolution"
|
|
assert event["requested_interpreter"] == "python"
|
|
assert event["resolved_interpreter"] == str(agent_python)
|
|
assert "cmd" not in event
|
|
assert "check" not in event
|
|
|
|
|
|
def test_registry_uses_current_process_python_before_server_bootstrap(
|
|
tmp_path, monkeypatch,
|
|
):
|
|
import sys
|
|
|
|
ctx = _context(tmp_path)
|
|
monkeypatch.delenv("OUROBOROS_AGENT_PYTHON", raising=False)
|
|
monkeypatch.setenv("OUROBOROS_RUNTIME_MODE", "advanced")
|
|
monkeypatch.setattr("ouroboros.safety.check_safety", lambda *args, **kwargs: (True, ""))
|
|
registry = ToolRegistry(repo_dir=ctx.repo_dir, drive_root=ctx.drive_root)
|
|
registry.set_context(ctx)
|
|
observed = {}
|
|
|
|
def handler(_ctx, cmd, **_kwargs):
|
|
observed["cmd"] = cmd
|
|
return "ok"
|
|
|
|
registry._entries["run_command"].handler = handler
|
|
|
|
result = registry.execute("run_command", {"cmd": ["python", "-V"]})
|
|
|
|
assert result == "ok"
|
|
assert observed["cmd"][0] == str(pathlib.Path(sys.executable).absolute())
|
|
|
|
|
|
def test_run_script_accepts_registry_attested_versioned_agent_python(
|
|
tmp_path, monkeypatch,
|
|
):
|
|
import ouroboros.tools.shell as shell
|
|
|
|
ctx = _context(tmp_path)
|
|
versioned_python = _executable(tmp_path / "agent" / "bin" / "python3.12")
|
|
monkeypatch.setenv("OUROBOROS_AGENT_PYTHON", str(versioned_python))
|
|
monkeypatch.setenv("OUROBOROS_RUNTIME_MODE", "advanced")
|
|
monkeypatch.setattr("ouroboros.safety.check_safety", lambda *args, **kwargs: (True, ""))
|
|
monkeypatch.setattr(shell, "_run_shell", lambda *_args, **_kwargs: "ok")
|
|
|
|
registry = ToolRegistry(repo_dir=ctx.repo_dir, drive_root=ctx.drive_root)
|
|
registry.set_context(ctx)
|
|
monkeypatch.setattr(registry, "_run_shell_safety_check", lambda *args, **kwargs: "")
|
|
|
|
result = registry.execute(
|
|
"run_script",
|
|
{"script": "print('ok')", "interpreter": "python3"},
|
|
)
|
|
|
|
assert result.endswith("\nok")
|
|
assert "RUN_SCRIPT_BLOCKED" not in result
|
|
assert not hasattr(ctx, "_active_python_resolution")
|
|
|
|
|
|
def test_run_script_still_blocks_unattested_versioned_interpreter(tmp_path):
|
|
from ouroboros.tools.shell import _run_script
|
|
|
|
result = _run_script(
|
|
_context(tmp_path),
|
|
"print('unsafe')",
|
|
interpreter=str(tmp_path / "untrusted" / "python3.12"),
|
|
)
|
|
|
|
assert result.startswith("⚠️ RUN_SCRIPT_BLOCKED:")
|
|
|
|
|
|
def test_safety_fast_path_requires_matching_verified_resolver_provenance(tmp_path, monkeypatch):
|
|
import ouroboros.safety as safety
|
|
from ouroboros.python_interpreter import PythonResolutionTrace
|
|
|
|
agent_python = str(_executable(tmp_path / "agent" / "bin" / "python"))
|
|
verified = PythonResolutionTrace(
|
|
tool="run_command",
|
|
requested_interpreter="python",
|
|
resolved_interpreter=agent_python,
|
|
surface="system_repo",
|
|
environment="ouroboros_agent",
|
|
reason="agent_python",
|
|
)
|
|
calls: list[list[str]] = []
|
|
|
|
def llm_check(_tool, arguments, *_args, **_kwargs):
|
|
calls.append(list(arguments["cmd"]))
|
|
return False, "llm-called"
|
|
|
|
monkeypatch.setattr(safety, "_run_llm_check", llm_check)
|
|
safe = safety.check_safety(
|
|
"run_command",
|
|
{"cmd": [agent_python, "-m", "pytest", "-q"]},
|
|
python_resolution=verified,
|
|
)
|
|
arbitrary = safety.check_safety(
|
|
"run_command",
|
|
{"cmd": [agent_python, "-m", "pytest", "-q"]},
|
|
)
|
|
mismatch = safety.check_safety(
|
|
"run_command",
|
|
{"cmd": [str(tmp_path / "other" / "python"), "-m", "pytest", "-q"]},
|
|
python_resolution=verified,
|
|
)
|
|
|
|
assert safe == (True, "")
|
|
assert arbitrary == (False, "llm-called")
|
|
assert mismatch == (False, "llm-called")
|
|
assert len(calls) == 2
|
|
|
|
|
|
def test_verified_python_c_body_still_requires_safety_review(tmp_path, monkeypatch):
|
|
import ouroboros.safety as safety
|
|
from ouroboros.python_interpreter import PythonResolutionTrace
|
|
|
|
agent_python = str(_executable(tmp_path / "agent" / "bin" / "python"))
|
|
verified = PythonResolutionTrace(
|
|
tool="run_command",
|
|
requested_interpreter="python3",
|
|
resolved_interpreter=agent_python,
|
|
surface="system_repo",
|
|
environment="ouroboros_agent",
|
|
reason="agent_python",
|
|
)
|
|
monkeypatch.setattr(safety, "_run_llm_check", lambda *args, **kwargs: (False, "reviewed"))
|
|
|
|
result = safety.check_safety(
|
|
"run_command",
|
|
{"cmd": [agent_python, "-c", "print('not allowlisted by module')"]},
|
|
python_resolution=verified,
|
|
)
|
|
|
|
assert result == (False, "reviewed")
|
|
|
|
|
|
def test_resolution_trace_failure_is_fail_soft(tmp_path, monkeypatch):
|
|
import ouroboros.python_interpreter as resolver
|
|
|
|
trace = resolver.PythonResolutionTrace(
|
|
tool="run_command",
|
|
requested_interpreter="python",
|
|
resolved_interpreter="python3",
|
|
surface="executor",
|
|
environment="backend_path",
|
|
reason="executor_backend_python3",
|
|
)
|
|
monkeypatch.setattr(resolver, "append_jsonl", lambda *args, **kwargs: (_ for _ in ()).throw(OSError("full")))
|
|
|
|
resolver.record_python_resolution(_context(tmp_path), trace)
|