diff --git a/modules/runtime/__init__.py b/modules/runtime/__init__.py index e69de29..1c5d45a 100644 --- a/modules/runtime/__init__.py +++ b/modules/runtime/__init__.py @@ -0,0 +1 @@ +from modules.runtime.cli_adapter import validate_spec, dry_run, replay diff --git a/modules/runtime/cli_adapter.py b/modules/runtime/cli_adapter.py new file mode 100644 index 0000000..ca3a6f9 --- /dev/null +++ b/modules/runtime/cli_adapter.py @@ -0,0 +1,209 @@ +"""Python adapter to the Rust runtime CLI for executing and validating agent contracts.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +from pathlib import Path +from typing import Any + +def find_ctxt_bin(custom_bin: str | None = None) -> str: + """Locate the ctxt binary from custom argument, COMPTEXT_CLI_BIN env var, or system PATH.""" + if custom_bin: + return custom_bin + if "COMPTEXT_CLI_BIN" in os.environ: + return os.environ["COMPTEXT_CLI_BIN"] + path_bin = shutil.which("ctxt") + if path_bin: + return path_bin + return "ctxt" + +def find_config_path(custom_config: str | None = None) -> str | None: + """Locate the config path from custom argument, COMPTEXT_CONFIG_PATH env var, or workspace fallback.""" + if custom_config: + return custom_config + if "COMPTEXT_CONFIG_PATH" in os.environ: + return os.environ["COMPTEXT_CONFIG_PATH"] + + # Fallback lookup in worktrees structure + try: + current_file = Path(__file__).resolve() + # modules/runtime/cli_adapter.py -> parent x4 is worktree root parent (worktrees) + worktrees_dir = current_file.parent.parent.parent.parent + cli_worktree = worktrees_dir / "comptext-cli" + cand = cli_worktree / "comptext.example.toml" + if cand.exists(): + return str(cand) + except Exception: + pass + + return None + +def get_allowlisted_env() -> dict[str, str]: + """Build a restricted environment mapping containing only allowlisted keys.""" + allowlist = { + "PATH", "SYSTEMROOT", "TEMP", "TMP", "USERPROFILE", + "HOME", "LANG", "LC_ALL" + } + filtered = {} + for k, v in os.environ.items(): + if k in allowlist or k.startswith("COMPTEXT_"): + filtered[k] = v + return filtered + +def parse_error_envelope(stdout: str, stderr: str, returncode: int) -> dict[str, Any]: + """Parse stdout/stderr robustly to extract or construct a valid ErrorEnvelope dict.""" + # 1. Try parsing stderr + if stderr and stderr.strip(): + try: + data = json.loads(stderr.strip()) + if isinstance(data, dict) and "contract_name" in data: + return data + except Exception: + pass + + # 2. Try parsing stdout + if stdout and stdout.strip(): + try: + data = json.loads(stdout.strip()) + if isinstance(data, dict) and "contract_name" in data: + return data + except Exception: + pass + + # 3. Fallback bounded ErrorEnvelope + max_len = 1024 + stderr_bounded = stderr[:max_len] + ("..." if len(stderr) > max_len else "") + stdout_bounded = stdout[:max_len] + ("..." if len(stdout) > max_len else "") + + return { + "contract_name": "error-envelope", + "schema_version": "v1", + "error_code": "NON_ZERO_EXIT", + "message": f"Command exited with status {returncode}", + "details": {"stderr": stderr_bounded, "stdout": stdout_bounded} + } + +def run_ctxt_cmd( + args: list[str], + bin_path: str | None = None, + config_path: str | None = None, + timeout: float | None = None, + cwd: str | None = None, +) -> dict[str, Any]: + """Execute the ctxt CLI with --json and return parsed output or a structured ErrorEnvelope.""" + resolved_bin = find_ctxt_bin(bin_path) + resolved_config = find_config_path(config_path) + + cmd = [resolved_bin] + if resolved_config: + cmd.extend(["--config", resolved_config]) + cmd.append("--json") + cmd.extend(args) + + if timeout is None: + try: + timeout = float(os.environ.get("COMPTEXT_CLI_TIMEOUT", 30)) + except ValueError: + timeout = 30.0 + + env = get_allowlisted_env() + execution_cwd = cwd or os.getcwd() + + try: + res = subprocess.run( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + shell=False, + timeout=timeout, + cwd=execution_cwd, + env=env + ) + except subprocess.TimeoutExpired as e: + return { + "contract_name": "error-envelope", + "schema_version": "v1", + "error_code": "TIMEOUT", + "message": f"Command timed out after {timeout} seconds: {e}", + "details": None + } + except Exception as e: + return { + "contract_name": "error-envelope", + "schema_version": "v1", + "error_code": "EXECUTION_FAILED", + "message": f"Failed to execute ctxt binary: {e}", + "details": None + } + + if res.returncode == 0: + try: + return json.loads(res.stdout) + except Exception as e: + return { + "contract_name": "error-envelope", + "schema_version": "v1", + "error_code": "MALFORMED_OUTPUT", + "message": f"Stdout could not be parsed as JSON: {e}", + "details": {"stdout": res.stdout[:1024]} + } + else: + return parse_error_envelope(res.stdout, res.stderr, res.returncode) + +def validate_spec( + spec_path: str, + bin_path: str | None = None, + config_path: str | None = None, + timeout: float | None = None, +) -> dict[str, Any]: + """Validate an AgentSpec file using the Rust CLI.""" + return run_ctxt_cmd( + ["agent", "validate-spec", spec_path], + bin_path=bin_path, + config_path=config_path, + timeout=timeout, + ) + +def dry_run( + spec_path: str, + out_evidence: str, + out_replay: str, + bin_path: str | None = None, + config_path: str | None = None, + timeout: float | None = None, +) -> dict[str, Any]: + """Execute an AgentSpec in dry-run mode using the Rust CLI.""" + return run_ctxt_cmd( + [ + "agent", "dry-run", + "--spec", spec_path, + "--out-evidence", out_evidence, + "--out-replay", out_replay + ], + bin_path=bin_path, + config_path=config_path, + timeout=timeout, + ) + +def replay( + replay_path: str, + evidence_path: str, + bin_path: str | None = None, + config_path: str | None = None, + timeout: float | None = None, +) -> dict[str, Any]: + """Verify logged evidence against a replay manifest using the Rust CLI.""" + return run_ctxt_cmd( + [ + "agent", "replay", + "--replay", replay_path, + "--evidence", evidence_path + ], + bin_path=bin_path, + config_path=config_path, + timeout=timeout, + ) diff --git a/tests/runtime/test_cli_adapter.py b/tests/runtime/test_cli_adapter.py new file mode 100644 index 0000000..8082531 --- /dev/null +++ b/tests/runtime/test_cli_adapter.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +import json +import os +import subprocess +from unittest.mock import patch, MagicMock + +from modules.runtime.cli_adapter import ( + find_ctxt_bin, + find_config_path, + get_allowlisted_env, + parse_error_envelope, + run_ctxt_cmd, + validate_spec, + dry_run, + replay, +) + +def test_get_allowlisted_env(monkeypatch): + monkeypatch.setenv("PATH", "/usr/bin") + monkeypatch.setenv("SECRET_TOKEN", "supersecret") + monkeypatch.setenv("COMPTEXT_CLI_BIN", "/custom/bin/ctxt") + + env = get_allowlisted_env() + assert "PATH" in env + assert "COMPTEXT_CLI_BIN" in env + assert "SECRET_TOKEN" not in env + +def test_find_ctxt_bin_custom_override(monkeypatch): + monkeypatch.setenv("COMPTEXT_CLI_BIN", "/env/bin/ctxt") + assert find_ctxt_bin("/override/bin/ctxt") == "/override/bin/ctxt" + assert find_ctxt_bin(None) == "/env/bin/ctxt" + +def test_parse_error_envelope_from_stderr(): + stderr = json.dumps({ + "contract_name": "error-envelope", + "schema_version": "v1", + "error_code": "CUSTOM_ERR", + "message": "Error from stderr", + "details": None + }) + res = parse_error_envelope("", stderr, 1) + assert res["error_code"] == "CUSTOM_ERR" + assert res["message"] == "Error from stderr" + +def test_parse_error_envelope_from_stdout(): + stdout = json.dumps({ + "contract_name": "error-envelope", + "schema_version": "v1", + "error_code": "STDOUT_ERR", + "message": "Error from stdout", + "details": None + }) + res = parse_error_envelope(stdout, "", 1) + assert res["error_code"] == "STDOUT_ERR" + +def test_parse_error_envelope_fallback(): + res = parse_error_envelope("raw stdout", "raw stderr", 1) + assert res["contract_name"] == "error-envelope" + assert res["error_code"] == "NON_ZERO_EXIT" + assert res["details"]["stderr"] == "raw stderr" + assert res["details"]["stdout"] == "raw stdout" + +@patch("subprocess.run") +def test_run_ctxt_cmd_success(mock_run): + mock_res = MagicMock() + mock_res.returncode = 0 + mock_res.stdout = json.dumps({"ok": True, "validated": True}) + mock_run.return_value = mock_res + + res = run_ctxt_cmd(["agent", "validate-spec", "spec.json"], bin_path="/bin/ctxt") + assert res["ok"] is True + assert res["validated"] is True + + # Verify subprocess.run args + args, kwargs = mock_run.call_args + assert kwargs["shell"] is False + assert args[0][0] == "/bin/ctxt" + assert "--json" in args[0] + assert "agent" in args[0] + assert "validate-spec" in args[0] + assert "spec.json" in args[0] + +@patch("subprocess.run") +def test_run_ctxt_cmd_timeout(mock_run): + mock_run.side_effect = subprocess.TimeoutExpired(cmd="ctxt", timeout=5.0) + + res = validate_spec("spec.json", bin_path="/bin/ctxt", timeout=5.0) + assert res["contract_name"] == "error-envelope" + assert res["error_code"] == "TIMEOUT" diff --git a/tests/runtime/test_cli_adapter_integration.py b/tests/runtime/test_cli_adapter_integration.py new file mode 100644 index 0000000..d2e4565 --- /dev/null +++ b/tests/runtime/test_cli_adapter_integration.py @@ -0,0 +1,89 @@ +from __future__ import annotations + +import json +import os +import shutil +from pathlib import Path +import pytest +from modules.runtime.cli_adapter import validate_spec, dry_run, replay, find_ctxt_bin + +def get_integration_ctxt_bin() -> str | None: + """Find ctxt binary for integration tests or return None if not built/available.""" + if "COMPTEXT_CLI_BIN" in os.environ: + cand = Path(os.environ["COMPTEXT_CLI_BIN"]) + if cand.exists(): + return str(cand) + + # Try looking relative to worktree setup + repo_root = Path(__file__).resolve().parent.parent.parent + base_dir = repo_root.parent.parent + worktree_cli = base_dir / "worktrees" / "comptext-cli" + + candidates = [ + worktree_cli / "target" / "debug" / "ctxt.exe", + worktree_cli / "target" / "debug" / "ctxt", + worktree_cli / "target" / "release" / "ctxt.exe", + worktree_cli / "target" / "release" / "ctxt", + ] + for cand in candidates: + if cand.exists() and cand.is_file(): + return str(cand) + + path_bin = shutil.which("ctxt") + if path_bin: + return path_bin + + return None + +def test_python_cli_adapter_golden_path_integration(tmp_path: Path) -> None: + bin_path = get_integration_ctxt_bin() + if not bin_path: + pytest.skip("ctxt binary not built or available for integration test.") + + # 1. Prepare paths + spec_file = tmp_path / "spec.json" + evidence_file = tmp_path / "evidence.jsonl" + replay_file = tmp_path / "replay.json" + + # 2. Write valid AgentSpec + spec_data = { + "contract_name": "agent-spec", + "schema_version": "v1", + "agent_spec_id": "python-adapter-test", + "intent": "validate", + "goal": "Verify Python subprocess CLI adapter integration", + "pipeline": ["echo-step"], + "outputs": [{"kind": "json", "path": "evidence/run.json"}] + } + spec_file.write_text(json.dumps(spec_data)) + + # 3. Test validate_spec + res_val = validate_spec(str(spec_file), bin_path=bin_path) + assert res_val.get("ok") is True + assert res_val.get("contract_name") == "agent-spec" + assert res_val.get("validated") is True + + # 4. Test dry_run + res_run = dry_run(str(spec_file), str(evidence_file), str(replay_file), bin_path=bin_path) + assert res_run.get("ok") is True + assert res_run.get("status") == "success" + assert "deterministic_root_hash" in res_run + assert "execution_chain_hash" in res_run + + assert evidence_file.exists() + assert replay_file.exists() + assert Path(str(evidence_file) + ".completion.json").exists() + + # 5. Test replay success + res_replay = replay(str(replay_file), str(evidence_file), bin_path=bin_path) + assert res_replay.get("ok") is True + assert res_replay.get("verified") is True + + # 6. Test replay failure on mutated evidence + lines = evidence_file.read_text().splitlines() + mutated_lines = [line.replace("fixture.echo", "malicious.tool") for line in lines] + evidence_file.write_text("\n".join(mutated_lines) + "\n") + + res_replay_fail = replay(str(replay_file), str(evidence_file), bin_path=bin_path) + assert res_replay_fail.get("contract_name") == "error-envelope" + assert res_replay_fail.get("error_code") == "REPLAY_VERIFICATION_FAILED"