openworker/tests/test_approval_integrity.py

"""PR4 — the server validates approval answers, and autonomy changes are recorded.

`POST /v1/inbox/{id}/resolve` takes a raw resolution string, so without validation any
local API caller could mint a grant the UI deliberately never offers — e.g. a session-wide,
any-argument shell grant, which `ApprovalCard.tsx` withholds on purpose in favour of the
narrower command-scoped one.

See `ocw-context/docs/reviewed-auto-mode.md` Part 3.
"""

from __future__ import annotations

from types import SimpleNamespace

import pytest

from coworker.engine import ApprovalOutcome
from coworker.server.manager import SessionManager


@pytest.fixture
def manager(tmp_path):
    mgr = SessionManager(workspace=str(tmp_path), data_dir=tmp_path / "data")
    yield mgr
    mgr.audit_store.close()


def _request(tool: str, arguments=None, metadata=None):
    return SimpleNamespace(
        tool_name=tool, arguments=arguments or {}, metadata=metadata, reason=""
    )


def _stages(manager, session_id: str) -> list[str]:
    rows = manager.audit_store.list(session_id=session_id)
    return [r.get("stage") for r in rows]


# -- grants the UI never offers are downgraded ----------------------------------
def test_always_tool_refused_for_shell(manager):
    out = manager.approval_outcome(
        "always_tool", _request("run_shell", {"command": "rm -rf /"}), "s1"
    )
    assert out is ApprovalOutcome.ONCE, "a session-wide any-command shell grant must not stand"
    assert "grant_refused" in _stages(manager, "s1")


def test_always_tool_refused_for_connector_and_external(manager):
    connector = SimpleNamespace(requires_approval=True, category="connector")
    assert (
        manager.approval_outcome("always_tool", _request("gmail_send", {}, connector), "s2")
        is ApprovalOutcome.ONCE
    )
    # An MCP tool is not category=connector but is still external — same reasoning applies:
    # the grant would be unbounded over every future argument.
    mcp = SimpleNamespace(requires_approval=True, category="mcp")
    assert (
        manager.approval_outcome(
            "always_tool", _request("mcp__github__create_issue", {}, mcp), "s3"
        )
        is ApprovalOutcome.ONCE
    )


def test_always_tool_refused_for_save_skill(manager):
    assert (
        manager.approval_outcome("always_tool", _request("save_skill", {}), "s4")
        is ApprovalOutcome.ONCE
    )


def test_always_command_only_for_shell(manager):
    assert (
        manager.approval_outcome(
            "always_command", _request("write_file", {"path": "a"}), "s5"
        )
        is ApprovalOutcome.ONCE
    )


def test_always_domain_only_for_egress_with_a_url(manager):
    assert (
        manager.approval_outcome("always_domain", _request("write_file", {"path": "a"}), "s6")
        is ApprovalOutcome.ONCE
    )
    assert (
        manager.approval_outcome("always_domain", _request("web_fetch", {}), "s7")
        is ApprovalOutcome.ONCE
    )


# -- the legitimate grants still work -------------------------------------------
def test_legitimate_grants_survive(manager):
    assert (
        manager.approval_outcome(
            "always_tool", _request("write_file", {"path": "a.txt"}), "ok1"
        )
        is ApprovalOutcome.ALWAYS_TOOL
    )
    assert (
        manager.approval_outcome(
            "always_command", _request("run_shell", {"command": "npm test"}), "ok2"
        )
        is ApprovalOutcome.ALWAYS_COMMAND
    )
    assert (
        manager.approval_outcome(
            "always_domain", _request("web_fetch", {"url": "https://docs.python.org/3"}), "ok3"
        )
        is ApprovalOutcome.ALWAYS_DOMAIN
    )
    assert "grant_refused" not in _stages(manager, "ok1")


def test_plain_vocabularies_unchanged(manager):
    req = _request("write_file", {"path": "a.txt"})
    assert manager.approval_outcome("allow", req, "s8") is ApprovalOutcome.ONCE
    assert manager.approval_outcome("once", req, "s8") is ApprovalOutcome.ONCE
    assert manager.approval_outcome("deny", req, "s8") is ApprovalOutcome.DENY
    assert manager.approval_outcome("nonsense", req, "s8") is ApprovalOutcome.DENY


def test_always_maps_to_tool_grant_and_is_validated(manager):
    # The Inbox/channel vocabulary "always" maps to a tool grant — and is validated too, so
    # a Slack reply cannot mint what the in-app card would refuse.
    assert (
        manager.approval_outcome("always", _request("run_shell", {"command": "x"}), "s9")
        is ApprovalOutcome.ONCE
    )


# -- autonomy changes are recorded ----------------------------------------------
def test_unattended_transition_audited(manager):
    manager.set_unattended("s10", True)
    rows = manager.audit_store.list(session_id="s10")
    assert [r["stage"] for r in rows] == ["unattended_changed"]
    assert rows[0]["status"] == "raised"

    manager.set_unattended("s10", False)
    assert [r["stage"] for r in manager.audit_store.list(session_id="s10")][-1] == (
        "unattended_changed"
    )


def test_unattended_no_op_not_audited(manager):
    manager.set_unattended("s11", False)  # already off
    assert _stages(manager, "s11") == []


def test_mode_change_audited_with_direction(manager):
    manager.audit_autonomy_change("s12", "mode", "interactive", "auto")
    manager.audit_autonomy_change("s12", "mode", "auto", "plan")
    # AuditStore.list is newest-first (ORDER BY id DESC), so reverse for chronology.
    rows = list(reversed(manager.audit_store.list(session_id="s12")))
    assert [r["status"] for r in rows] == ["raised", "lowered"]
    assert rows[0]["reason"] == "mode: interactive → auto"