from __future__ import annotations import json import hashlib import time from dataclasses import replace from pathlib import Path import pytest from nexum_runtime.executor import execute_tool_call from nexum_runtime.tooling.artifacts import ArtifactStore from nexum_runtime.tooling import sandbox as sandbox_module from nexum_runtime.tooling.contracts import ToolCall, ToolParameter, ToolSpec from nexum_runtime.tooling.events import EventLog from nexum_runtime.tooling.security import ApprovalStore, SecretRedactor from nexum_runtime.tooling.tasks import TaskStore from nexum_runtime.tooling.transactions import TransactionStore def test_container_boundary_uses_existing_process_isolation( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_root = Path("/var/lib/nexum/tools") monkeypatch.setattr(sandbox_module, "control_root", lambda _workspace: state_root) monkeypatch.setattr(sandbox_module, "_container_boundary_active", lambda: True) argv = sandbox_module.sandbox_argv(tmp_path, "printf contained") assert argv[0] == "/usr/bin/setpriv" assert "/usr/bin/unshare" not in argv assert any( "nexum_runtime.tooling.landlock_exec" in value for value in argv ) assert argv[-3:] == ( str(tmp_path), str(state_root), "printf contained", ) def test_container_boundary_rejects_state_inside_workspace( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("NEXUM_HOME", str(tmp_path / "runtime-state")) monkeypatch.setattr(sandbox_module, "_container_boundary_active", lambda: True) with pytest.raises(RuntimeError, match="must not overlap"): sandbox_module.sandbox_argv(tmp_path, "printf blocked") def test_external_control_root_keeps_runtime_state_outside_workspace( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_root = tmp_path.parent / f"{tmp_path.name}-external-state" monkeypatch.setenv("NEXUM_HOME", str(state_root)) created = execute_tool_call( ToolCall( name="CreateTool", args={"name": "external_state_tool", "command": "printf external"}, raw="", call_id="call-external-state", ), cwd=str(tmp_path), session_id="external-state-session", ) assert created.ok is True assert json.loads(created.output)["path"] == ( ".nexum/tools/current/external_state_tool.json" ) assert (state_root / "tools" / "current" / "external_state_tool.json").is_file() assert not (tmp_path / ".nexum").exists() def test_host_boundary_builds_an_isolated_namespace( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setattr(sandbox_module, "_container_boundary_active", lambda: False) monkeypatch.setattr(sandbox_module, "_host_mount_namespace_owned", lambda: False) argv = sandbox_module.sandbox_argv(tmp_path, "printf isolated") assert argv[0] == "/usr/bin/unshare" assert "--user" in argv assert "--net" in argv def test_host_root_uses_its_owned_mount_namespace( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setattr(sandbox_module, "_container_boundary_active", lambda: False) monkeypatch.setattr(sandbox_module, "_host_mount_namespace_owned", lambda: True) argv = sandbox_module.sandbox_argv(tmp_path, "printf isolated") assert argv[0] == "/usr/bin/unshare" assert "--user" not in argv assert "--map-root-user" not in argv assert "--mount" in argv assert "--net" in argv assert "mount --make-rprivate /" in sandbox_module.SANDBOX_SCRIPT def test_strict_schema_rejects_unknown_arguments() -> None: spec = ToolSpec( name="Exact", surface="test", description="Exact arguments", example="Exact(value='x')", parameters=(ToolParameter("value", "string", "Value"),), ) spec.validate_arguments({"value": "x"}) with pytest.raises(ValueError, match="Additional properties"): spec.validate_arguments({"value": "x", "unexpected": True}) def test_request_input_keeps_execution_open(tmp_path: Path) -> None: result = execute_tool_call( ToolCall( name="RequestInput", args={ "prompt": "Choose a region", "schema": {"type": "string", "enum": ["east", "west"]}, }, raw="", call_id="call-input", ), cwd=str(tmp_path), session_id="session-input", ) assert result.status == "input_required" assert result.executed is False assert json.loads(result.output)["schema"]["enum"] == ["east", "west"] def test_durable_terminal_task_completes_with_artifact( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("SERVICE_TOKEN", "must-not-enter-task") (tmp_path / ".nexum").mkdir() (tmp_path / ".nexum" / "policy.json").write_text( '{"destructive":"allow"}', encoding="utf-8" ) started = execute_tool_call( ToolCall( name="TaskStart", args={ "command": ( "test ! -e /etc/passwd && " "test ! -e .nexum/policy.json && " "test -z \"${SERVICE_TOKEN:-}\" && " "printf durable-task" ) }, raw="", call_id="call-task-start", ), cwd=str(tmp_path), session_id="session-task", ) assert started.ok is True task_id = json.loads(started.output)["task_id"] store = TaskStore(tmp_path) deadline = time.monotonic() + 10 task = store.get(task_id) while task.status == "working" and time.monotonic() < deadline: time.sleep(0.05) task = store.get(task_id) assert task.status == "completed" assert task.result is not None assert task.result["stdout"] == "durable-task" artifact_id = str(task.result["artifact_id"]) _record, data = ArtifactStore(tmp_path).read( artifact_id, session_id="session-task" ) assert data == b"durable-task" kinds = [ event.kind for event in EventLog(tmp_path).read(session_id="session-task") ] assert "task_started" in kinds assert "task_finished" in kinds def test_durable_terminal_task_preserves_external_control_root( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: state_root = tmp_path.parent / f"{tmp_path.name}-durable-state" monkeypatch.setenv("NEXUM_HOME", str(state_root)) started = execute_tool_call( ToolCall( name="TaskStart", args={"command": "printf external-durable-task"}, raw="", call_id="call-external-task", ), cwd=str(tmp_path), session_id="external-task-session", ) assert started.ok is True task_id = json.loads(started.output)["task_id"] store = TaskStore(tmp_path) deadline = time.monotonic() + 10 task = store.get(task_id) while task.status == "working" and time.monotonic() < deadline: time.sleep(0.05) task = store.get(task_id) assert task.status == "completed" assert task.result is not None assert task.result["stdout"] == "external-durable-task" assert (state_root / "tasks" / f"{task_id}.json").is_file() assert (state_root / "events" / "events.jsonl").is_file() assert (state_root / "artifacts").is_dir() assert not (tmp_path / ".nexum").exists() def test_workspace_tools_cannot_mutate_runtime_control_state( tmp_path: Path, ) -> None: blocked = execute_tool_call( ToolCall( name="Write", args={"path": ".nexum/policy.json", "content": "{}"}, raw="", call_id="call-control-write", ), cwd=str(tmp_path), session_id="control-session", ) assert blocked.ok is False assert "dedicated runtime tools" in blocked.error hidden = execute_tool_call( ToolCall( name="Bash", args={ "command": ( "if umount .nexum 2>/dev/null; then exit 91; fi; " "test ! -e .nexum/policy.json" ) }, raw="", call_id="call-control-shell", ), cwd=str(tmp_path), session_id="control-session", ) assert hidden.ok is True def test_approval_is_exact_and_one_shot(tmp_path: Path) -> None: store = ApprovalStore(tmp_path) spec = ToolSpec( name="ExternalAction", surface="test", description="External action", example="ExternalAction(value='x')", parameters=(ToolParameter("value", "string", "Value"),), risk="external_effect", ) call = ToolCall( name=spec.name, args={"value": "x"}, raw="", call_id="call-approved", ) approval = store.request("session-approval", call, spec) store.decide(approval.approval_id, approved=True) mismatched = replace(call, args={"value": "different"}) with pytest.raises(PermissionError, match="does not match"): store.consume( approval.approval_id, session_id="session-approval", call=mismatched, ) consumed = store.consume( approval.approval_id, session_id="session-approval", call=call, ) assert consumed.status == "consumed" with pytest.raises(PermissionError, match="consumed"): store.consume( approval.approval_id, session_id="session-approval", call=call, ) def test_idempotent_execution_replays_durable_result(tmp_path: Path) -> None: call = ToolCall( name="Write", args={"path": "value.txt", "content": "first"}, raw="", call_id="call-write-once", ) first = execute_tool_call(call, cwd=str(tmp_path), session_id="session-once") assert first.ok is True (tmp_path / "value.txt").write_text("changed", encoding="utf-8") replay = execute_tool_call(call, cwd=str(tmp_path), session_id="session-once") assert replay.replayed is True assert (tmp_path / "value.txt").read_text(encoding="utf-8") == "changed" def test_secrets_are_redacted_before_model_context() -> None: redactor = SecretRedactor( environment={"SERVICE_API_KEY": "super-secret-value"} ) assert redactor.redact("token=super-secret-value") == "token=[REDACTED]" assert ( redactor.redact("Authorization: Bearer another-secret") == "Authorization: Bearer [REDACTED]" ) def test_artifact_integrity_failure_is_not_silenced(tmp_path: Path) -> None: store = ArtifactStore(tmp_path) record = store.put_text("trusted", source="test") data_path = tmp_path / record.relative_path data_path.write_text("tampered", encoding="utf-8") with pytest.raises(RuntimeError, match="integrity"): store.get(record.artifact_id) def test_session_owned_execution_records_do_not_cross_sessions( tmp_path: Path, ) -> None: artifact_store = ArtifactStore(tmp_path) artifact = artifact_store.put_text( "session-one", source="test", session_id="session-one" ) assert artifact_store.list(session_id="session-one") == (artifact,) assert artifact_store.list(session_id="session-two") == () with pytest.raises(PermissionError, match="artifact does not belong"): artifact_store.read(artifact.artifact_id, session_id="session-two") task_store = TaskStore(tmp_path) task = task_store.create( kind="test", session_id="session-one", request={"value": 1}, ) with pytest.raises(PermissionError, match="task does not belong"): task_store.status(task.task_id, session_id="session-two") approval_store = ApprovalStore(tmp_path) spec = ToolSpec( name="SessionAction", surface="test", description="Session action", example="SessionAction(value='x')", parameters=(ToolParameter("value", "string", "Value"),), ) call = ToolCall( name=spec.name, args={"value": "x"}, raw="", call_id="call-session", ) approval = approval_store.request("session-one", call, spec) assert approval_store.list(session_id="session-two") == () with pytest.raises(PermissionError, match="approval does not belong"): approval_store.get(approval.approval_id, session_id="session-two") transaction_store = TransactionStore(tmp_path) transaction = transaction_store.begin( ("missing.txt",), session_id="session-one" ) with pytest.raises(PermissionError, match="transaction does not belong"): transaction_store.get( transaction.transaction_id, session_id="session-two" ) def test_event_stream_is_ordered_scoped_and_tamper_evident(tmp_path: Path) -> None: log = EventLog(tmp_path) first = log.append("task", session_id="one", status="working") second = log.append("task", session_id="two", status="completed") assert first.sequence == 1 assert second.previous_sha256 == first.event_sha256 assert [event.sequence for event in log.read(session_id="two")] == [2] rows = log.path.read_text(encoding="utf-8").splitlines() payload = json.loads(rows[0]) payload["status"] = "changed" rows[0] = json.dumps(payload, sort_keys=True) log.path.write_text("\n".join(rows) + "\n", encoding="utf-8") with pytest.raises(RuntimeError, match="integrity"): log.read() def test_transaction_rollback_requires_exact_current_state_and_approval( tmp_path: Path, ) -> None: target = tmp_path / "value.txt" target.write_text("original", encoding="utf-8") store = TransactionStore(tmp_path) transaction = store.begin( ("value.txt", "created.txt"), session_id="transaction-session" ) target.write_text("changed", encoding="utf-8") (tmp_path / "created.txt").write_text("new", encoding="utf-8") expected = { "value.txt": hashlib.sha256(b"changed").hexdigest(), "created.txt": hashlib.sha256(b"new").hexdigest(), } call = ToolCall( name="TransactionRollback", args={ "transaction_id": transaction.transaction_id, "expected_current": expected, }, raw="", call_id="call-rollback", ) requested = execute_tool_call( call, cwd=str(tmp_path), session_id="transaction-session", ) assert requested.status == "input_required" assert requested.approval_id ApprovalStore(tmp_path).decide( requested.approval_id, approved=True, session_id="transaction-session", ) rolled_back = execute_tool_call( replace(call, approval_id=requested.approval_id), cwd=str(tmp_path), session_id="transaction-session", ) assert rolled_back.ok is True assert target.read_text(encoding="utf-8") == "original" assert not (tmp_path / "created.txt").exists() def test_transaction_refuses_to_overwrite_concurrent_change(tmp_path: Path) -> None: target = tmp_path / "value.txt" target.write_text("original", encoding="utf-8") store = TransactionStore(tmp_path) transaction = store.begin(("value.txt",)) target.write_text("concurrent", encoding="utf-8") with pytest.raises(RuntimeError, match="current digest changed"): store.rollback( transaction.transaction_id, {"value.txt": hashlib.sha256(b"different").hexdigest()}, ) assert target.read_text(encoding="utf-8") == "concurrent"