aiturk-hermes-ide/tests/tools/test_protected_instruction_unattended.py

"""Protected instruction-file gate when no human can be reached.

A one-shot run (``chat -q`` / ``--query-file``) installs the CLI approval
callback but never starts the prompt UI. The gate used to treat the
registered callback as a reachable human and wait out ``approvals.timeout``
on a panel nobody could see. These tests pin the fixed contract:

* no reachable human -> BLOCKED at once, nothing written, the outcome is
  recorded for the launching process;
* the gate itself is unchanged: still asks a real human every time, still
  ignores yolo and ``approvals.single_query_mode``;
* an approval that arrives after the calling execute_code script stopped
  waiting is not applied.
"""

import json
import time

import pytest


@pytest.fixture(autouse=True)
def _gate_on(monkeypatch):
    import tools.approval as A
    import tools.file_tools as ft

    monkeypatch.setattr(ft, "_protected_instruction_config", lambda: (True, []))
    monkeypatch.delenv("HERMES_SINGLE_QUERY_SESSION", raising=False)
    A.drain_human_approval_required()
    A.set_approval_reply_deadline(None)
    yield
    A.drain_human_approval_required()
    A.set_approval_reply_deadline(None)


@pytest.fixture
def approvals():
    """CLI approval callback that records calls and gives a scripted answer."""
    from tools.terminal_tool import set_approval_callback

    state = {"calls": [], "answer": "deny"}

    def cb(command, description, **kwargs):
        state["calls"].append({"command": command, **kwargs})
        answer = state["answer"]
        if callable(answer):
            return answer()
        return answer

    set_approval_callback(cb)
    yield state
    set_approval_callback(None)


@pytest.fixture
def one_shot(monkeypatch):
    monkeypatch.setenv("HERMES_SINGLE_QUERY_SESSION", "1")


def _write(path, content="injected"):
    from tools.file_tools import write_file_tool

    return json.loads(write_file_tool(str(path), content))


def _records():
    import tools.approval as A

    return A.drain_human_approval_required()


# ---- one-shot run: callback registered, nobody to answer ------------------


class TestOneShotRun:
    def test_blocks_at_once_without_prompting(self, tmp_path, approvals, one_shot):
        target = tmp_path / "AGENTS.md"
        target.write_text("rules\n", encoding="utf-8")
        approvals["answer"] = "once"  # would be honored if anyone were asked

        started = time.monotonic()
        res = _write(target)
        elapsed = time.monotonic() - started

        assert res.get("error") and "BLOCKED" in res["error"]
        assert "APPROVAL REQUIRED" in res["error"]
        assert target.read_text(encoding="utf-8") == "rules\n"
        assert approvals["calls"] == [], "prompted on a surface nobody can see"
        assert elapsed < 5

    def test_outcome_is_recorded_for_the_launcher(self, tmp_path, approvals, one_shot):
        import tools.approval as A

        target = tmp_path / "CLAUDE.md"
        sid = A._approval_session_id.set("sess-123")
        tcid = A._approval_tool_call_id.set("call-9")
        try:
            _write(target)
        finally:
            A._approval_tool_call_id.reset(tcid)
            A._approval_session_id.reset(sid)

        records = _records()
        assert len(records) == 1
        rec = records[0]
        assert rec["kind"] == "protected_instruction_file"
        assert rec["reason"] == "no_interactive_user"
        assert rec["applied"] is False
        assert rec["session_id"] == "sess-123"
        assert rec["tool_call_id"] == "call-9"
        assert rec["request_id"]
        assert any("CLAUDE.md" in t for t in rec["targets"])
        assert _records() == [], "drain must clear"

    def test_yolo_does_not_open_the_gate(self, tmp_path, approvals, one_shot, monkeypatch):
        import tools.approval as A

        monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", True)
        target = tmp_path / "AGENTS.md"
        res = _write(target)
        assert res.get("error") and "BLOCKED" in res["error"]
        assert not target.exists()

    def test_single_query_mode_approve_does_not_open_the_gate(
        self, tmp_path, approvals, one_shot, monkeypatch
    ):
        import tools.approval as A

        real_cfg_get = A.cfg_get

        def fake_cfg_get(config, *keys, default=None):
            if keys == ("approvals", "single_query_mode"):
                return "approve"
            return real_cfg_get(config, *keys, default=default)

        monkeypatch.setattr(A, "cfg_get", fake_cfg_get)
        target = tmp_path / "SOUL.md"
        res = _write(target)
        assert res.get("error") and "BLOCKED" in res["error"]
        assert not target.exists()

    def test_regular_file_still_writes(self, tmp_path, approvals, one_shot):
        target = tmp_path / "notes.md"
        res = _write(target, "hello")
        assert not res.get("error"), res
        assert target.read_text(encoding="utf-8") == "hello"
        assert _records() == []

    def test_multi_file_patch_applies_nothing(self, tmp_path, approvals, one_shot):
        from tools.file_tools import patch_tool

        agents = tmp_path / "AGENTS.md"
        agents.write_text("rules\n", encoding="utf-8")
        plain = tmp_path / "plain.txt"
        plain.write_text("hello\n", encoding="utf-8")
        patch = (
            "*** Begin Patch\n"
            f"*** Update File: {plain}\n"
            "@@\n"
            "-hello\n"
            "+world\n"
            f"*** Update File: {agents}\n"
            "@@\n"
            "-rules\n"
            "+injected\n"
            "*** End Patch"
        )
        res = json.loads(patch_tool(mode="patch", patch=patch))
        assert res.get("error") and "BLOCKED" in res["error"]
        assert plain.read_text(encoding="utf-8") == "hello\n"
        assert agents.read_text(encoding="utf-8") == "rules\n"
        assert approvals["calls"] == []

    def test_retry_in_same_run_blocks_again(self, tmp_path, approvals, one_shot):
        target = tmp_path / "AGENTS.md"
        _write(target)
        res = _write(target)
        assert res.get("error") and "BLOCKED" in res["error"]
        assert not target.exists()
        assert len(_records()) == 2


# ---- no channel at all ----------------------------------------------------


def test_no_channel_fails_closed_and_is_recorded(tmp_path):
    target = tmp_path / "AGENTS.md"
    res = _write(target)
    assert res.get("error") and "BLOCKED" in res["error"]
    assert not target.exists()
    records = _records()
    assert [r["reason"] for r in records] == ["no_approval_channel"]


def test_undeliverable_gateway_request_is_recorded(tmp_path):
    import tools.approval as A

    session_key = "protected-unattended-undelivered"
    token = A.set_current_session_key(session_key)
    try:
        def notify(approval_data):
            raise RuntimeError("client went away")

        A.register_gateway_notify(session_key, notify)
        try:
            target = tmp_path / "AGENTS.md"
            res = _write(target)
        finally:
            A.unregister_gateway_notify(session_key)
    finally:
        A.reset_current_session_key(token)

    assert res.get("error") and "BLOCKED" in res["error"]
    assert not target.exists()
    assert [r["reason"] for r in _records()] == ["approval_not_delivered"]


# ---- a reachable human: the gate behaves exactly as before -----------------


class TestReachableHuman:
    def test_allow_writes(self, tmp_path, approvals):
        target = tmp_path / "AGENTS.md"
        approvals["answer"] = "once"
        res = _write(target, "approved")
        assert not res.get("error"), res
        assert target.read_text(encoding="utf-8") == "approved"
        assert len(approvals["calls"]) == 1
        assert _records() == []

    @pytest.mark.parametrize("answer", ["deny", "timeout"])
    def test_deny_and_silence_block(self, tmp_path, approvals, answer):
        target = tmp_path / "AGENTS.md"
        approvals["answer"] = answer
        res = _write(target)
        assert res.get("error") and "BLOCKED" in res["error"]
        assert not target.exists()
        assert len(approvals["calls"]) == 1
        # A human was asked and did not consent: not an "unattended" outcome.
        assert _records() == []

    def test_prompt_that_dies_blocks(self, tmp_path, approvals):
        """Cancel / lost surface: the prompt raising is not consent."""
        def boom():
            raise RuntimeError("prompt surface lost")

        target = tmp_path / "AGENTS.md"
        approvals["answer"] = boom
        res = _write(target)
        assert res.get("error") and "BLOCKED" in res["error"]
        assert not target.exists()


# ---- approval that arrives after the caller stopped waiting ----------------


class TestLateApproval:
    def test_late_cli_approval_is_not_applied(self, tmp_path, approvals):
        import tools.approval as A

        target = tmp_path / "AGENTS.md"
        target.write_text("rules\n", encoding="utf-8")

        def answer_after_deadline():
            # The human says yes, but only once the execute_code script that
            # asked for this write has already timed out.
            A.set_approval_reply_deadline(time.monotonic() - 1)
            return "once"

        A.set_approval_reply_deadline(time.monotonic() + 300)
        approvals["answer"] = answer_after_deadline
        res = _write(target)

        assert res.get("error") and "BLOCKED" in res["error"]
        assert "NOT applied" in res["error"]
        assert target.read_text(encoding="utf-8") == "rules\n"

    def test_late_approval_does_not_carry_to_the_next_request(self, tmp_path, approvals):
        import tools.approval as A

        target = tmp_path / "AGENTS.md"
        A.set_approval_reply_deadline(time.monotonic() - 1)
        approvals["answer"] = "once"
        assert "BLOCKED" in _write(target).get("error", "")
        assert not target.exists()

        # A fresh call (new deadline) asks again; the old yes is not reused.
        A.set_approval_reply_deadline(time.monotonic() + 300)
        approvals["answer"] = "deny"
        assert "BLOCKED" in _write(target).get("error", "")
        assert not target.exists()
        assert len(approvals["calls"]) == 2

    def test_in_time_approval_is_applied(self, tmp_path, approvals):
        import tools.approval as A

        target = tmp_path / "AGENTS.md"
        A.set_approval_reply_deadline(time.monotonic() + 300)
        approvals["answer"] = "once"
        res = _write(target, "approved")
        assert not res.get("error"), res
        assert target.read_text(encoding="utf-8") == "approved"

    def test_late_gateway_approval_is_not_applied(self, tmp_path):
        import tools.approval as A

        session_key = "protected-unattended-late"
        token = A.set_current_session_key(session_key)
        try:
            def notify(approval_data):
                A.resolve_gateway_approval(session_key, "once")

            A.register_gateway_notify(session_key, notify)
            try:
                A.set_approval_reply_deadline(time.monotonic() - 1)
                target = tmp_path / "AGENTS.md"
                res = _write(target)
            finally:
                A.unregister_gateway_notify(session_key)
        finally:
            A.reset_current_session_key(token)

        assert res.get("error") and "NOT applied" in res["error"]
        assert not target.exists()

    def test_no_deadline_outside_execute_code(self):
        import tools.approval as A

        assert A.approval_reply_window_remaining() is None
        assert A.approval_arrived_too_late() is False

    def test_rpc_dispatch_binds_and_restores_the_deadline(self):
        import tools.approval as A
        import tools.code_execution_tool as cet

        start = time.monotonic()
        previous = cet._bind_rpc_reply_deadline(start)
        try:
            remaining = A.approval_reply_window_remaining()
            assert remaining is not None
            assert 0 < remaining <= cet._RPC_REPLY_TIMEOUT_SECONDS
        finally:
            cet._restore_rpc_reply_deadline(previous)
        assert A.approval_reply_window_remaining() is None

    def test_deadline_is_per_thread(self):
        import threading

        import tools.approval as A

        A.set_approval_reply_deadline(time.monotonic() - 1)
        seen = {}

        def other():
            seen["late"] = A.approval_arrived_too_late()

        worker = threading.Thread(target=other)
        worker.start()
        worker.join()
        assert seen["late"] is False
        assert A.approval_arrived_too_late() is True


# ---- what the launching process sees ---------------------------------------


def test_recorded_outcomes_are_bounded():
    import tools.approval as A

    for i in range(A._HUMAN_APPROVAL_REQUIRED_MAX + 20):
        A.note_human_approval_required("protected_instruction_file", [f"f{i}"], "x")
    records = A.drain_human_approval_required()
    assert len(records) == A._HUMAN_APPROVAL_REQUIRED_MAX
    assert records[-1]["targets"] == [f"f{A._HUMAN_APPROVAL_REQUIRED_MAX + 19}"]


def test_one_shot_exit_reports_approval_required_on_stderr(capsys):
    import cli as cli_module
    import tools.approval as A

    A.note_human_approval_required(
        "protected_instruction_file", ["AGENTS.md"], "no_interactive_user"
    )
    cli_module._emit_approval_required("sess-42")

    err = capsys.readouterr().err
    lines = [l for l in err.splitlines() if l.startswith("approval_required: ")]
    assert len(lines) == 1
    payload = json.loads(lines[0].split(": ", 1)[1])
    assert payload["session_id"] == "sess-42"
    assert payload["applied"] is False
    assert payload["reason"] == "no_interactive_user"
    assert payload["resume"] == "interactive"
    assert payload["targets"] == ["AGENTS.md"]

    cli_module._emit_approval_required("sess-42")
    assert "approval_required" not in capsys.readouterr().err


# ---- execute_code RPC: the dispatcher really binds the window ---------------


def _drive_rpc_once(dispatch, stop_event=None):
    """Send one tool call through the real RPC server loop; return its reply."""
    import socket
    import threading

    import tools.code_execution_tool as cet

    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.bind(("127.0.0.1", 0))
    server.listen(1)
    stop_event = stop_event or threading.Event()
    worker = threading.Thread(
        target=cet._rpc_server_loop,
        args=(server, "task", [], [0], 5, frozenset({"patch"}), stop_event, "tok", dispatch),
        daemon=True,
    )
    worker.start()
    client = socket.create_connection(server.getsockname(), timeout=10)
    try:
        client.sendall(
            (json.dumps({"tool": "patch", "args": {}, "token": "tok"}) + "\n").encode()
        )
        buf = b""
        while not buf.endswith(b"\n"):
            chunk = client.recv(65536)
            if not chunk:
                break
            buf += chunk
    finally:
        client.close()
        stop_event.set()
        worker.join(timeout=10)
        server.close()
    return buf.decode().strip()


def test_rpc_server_loop_binds_reply_window_around_dispatch():
    import tools.approval as A
    import tools.code_execution_tool as cet

    seen = {}

    def dispatch(tool_name, tool_args):
        seen["remaining"] = A.approval_reply_window_remaining()
        seen["late"] = A.approval_arrived_too_late()
        return json.dumps({"ok": True})

    reply = _drive_rpc_once(dispatch)
    assert json.loads(reply) == {"ok": True}
    assert seen["late"] is False
    assert 0 < seen["remaining"] <= cet._RPC_REPLY_TIMEOUT_SECONDS


def test_rpc_approval_after_script_ended_is_late():
    """Script killed by its own run timeout while the prompt is still open."""
    import threading

    import tools.approval as A

    stop_event = threading.Event()
    seen = {}

    def dispatch(tool_name, tool_args):
        seen["before"] = A.approval_arrived_too_late()
        stop_event.set()  # the script ended while a human was deciding
        seen["after"] = A.approval_arrived_too_late()
        return json.dumps({"ok": True})

    _drive_rpc_once(dispatch, stop_event)
    assert seen == {"before": False, "after": True}


def test_protected_write_through_rpc_is_dropped_when_script_ended(tmp_path, approvals):
    """End to end through the RPC loop: late yes -> BLOCKED, file untouched."""
    import threading

    from tools.file_tools import write_file_tool
    from tools.terminal_tool import _get_approval_callback, set_approval_callback

    target = tmp_path / "AGENTS.md"
    target.write_text("rules\n", encoding="utf-8")
    stop_event = threading.Event()
    callback = _get_approval_callback()

    def answer_after_script_died():
        stop_event.set()
        return "once"

    approvals["answer"] = answer_after_script_died

    def dispatch(tool_name, tool_args):
        set_approval_callback(callback)  # the RPC thread carries the cell's callback
        try:
            return write_file_tool(str(target), "injected")
        finally:
            set_approval_callback(None)

    reply = json.loads(_drive_rpc_once(dispatch, stop_event))
    assert "NOT applied" in reply.get("error", "")
    assert target.read_text(encoding="utf-8") == "rules\n"
    assert len(approvals["calls"]) == 1


def test_retired_kernel_cell_counts_as_gone():
    import tools.approval as A

    cell = {"active": True}
    previous = A.also_gone_when(lambda: not cell["active"])
    try:
        assert A.approval_arrived_too_late() is False
        cell["active"] = False
        assert A.approval_arrived_too_late() is True
    finally:
        A.restore_approval_reply_deadline(previous)
    assert A.approval_arrived_too_late() is False


# ---- an automatic approver is not a person ----------------------------------


def test_subagent_auto_approve_cannot_open_the_gate(tmp_path):
    from tools.delegate_tool import _subagent_auto_approve
    from tools.terminal_tool import set_approval_callback

    target = tmp_path / "AGENTS.md"
    set_approval_callback(_subagent_auto_approve)
    try:
        res = _write(target)
    finally:
        set_approval_callback(None)
    assert res.get("error") and "BLOCKED" in res["error"]
    assert not target.exists()
    assert [r["reason"] for r in _records()] == ["no_interactive_user"]


def test_blocked_message_survives_braces_in_target(tmp_path):
    import tools.file_tools as ft

    res = ft._request_protected_instruction_approval(["{be}/AGENTS.md"])
    assert res.startswith("BLOCKED") and "{be}/AGENTS.md" in res


# ---- sibling gates in a one-shot run ----------------------------------------


def test_computer_use_does_not_wait_in_one_shot_run(one_shot, monkeypatch):
    import tools.computer_use.tool as cu

    calls = []
    monkeypatch.setattr(cu, "_approval_callback", lambda *a, **k: calls.append(a) or "approve_once")
    res = cu._request_approval("click", {}, session_id="unattended-test")
    assert res is not None and "one-shot run" in json.loads(res)["error"]
    assert calls == []


def test_computer_use_still_asks_when_interactive(monkeypatch):
    import tools.computer_use.tool as cu

    calls = []
    monkeypatch.setattr(cu, "_approval_callback", lambda *a, **k: calls.append(a) or "approve_once")
    assert cu._request_approval("click", {}, session_id="attended-test") is None
    assert len(calls) == 1


def test_memory_write_prompt_unavailable_in_one_shot_run(approvals, monkeypatch):
    import tools.write_approval as wa

    assert wa._interactive_approval_available() is True
    monkeypatch.setenv("HERMES_SINGLE_QUERY_SESSION", "1")
    assert wa._interactive_approval_available() is False


def test_one_shot_marker_is_not_inherited_by_child_processes(one_shot):
    from tools.environments.local import _make_run_env

    assert "HERMES_SINGLE_QUERY_SESSION" not in _make_run_env({})