Nexum-Expanded/runtime/tests/tooling/test_execution_services.py

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"