the-nexus/tests/test_stop_protocol.py

#!/usr/bin/env python3
"""
tests/test_stop_protocol.py — 100% compliance tests for The Stop Protocol (M1)

Refs: the-nexus #844
"""

import json
import tempfile
import time
from pathlib import Path

import pytest

import sys
sys.path.insert(0, str(Path(__file__).parent.parent))

from nexus.stop_protocol import (
    StopProtocol,
    StopAck,
    HandsOffEntry,
    StopInterrupt,
    HandsOffError,
)


class TestPreToolCheckGate:
    """M1 Requirement: Pre-tool-check gate must block tools when stopped."""

    def test_is_stopped_returns_false_initially(self):
        sp = StopProtocol()
        assert sp.is_stopped("sess_123") is False

    def test_pre_tool_check_passes_when_not_stopped(self):
        sp = StopProtocol()
        # Should not raise
        sp.pre_tool_check("sess_123")

    def test_pre_tool_check_blocks_when_stopped(self):
        sp = StopProtocol()
        sp.acknowledge_stop(source="user", session_id="sess_123", reason="halt")
        with pytest.raises(StopInterrupt):
            sp.pre_tool_check("sess_123")

    def test_clear_stop_allows_tools_again(self):
        sp = StopProtocol()
        sp.acknowledge_stop(source="user", session_id="sess_123")
        assert sp.is_stopped("sess_123") is True
        sp.clear_stop("sess_123")
        assert sp.is_stopped("sess_123") is False
        # Should not raise after clear
        sp.pre_tool_check("sess_123")

    def test_stop_is_per_session(self):
        sp = StopProtocol()
        sp.acknowledge_stop(source="user", session_id="sess_A")
        assert sp.is_stopped("sess_A") is True
        assert sp.is_stopped("sess_B") is False
        sp.pre_tool_check("sess_B")  # should not raise


class TestStopAckLogging:
    """M1 Requirement: STOP_ACK must be logged immediately and durably."""

    def test_acknowledge_stop_returns_stop_ack(self):
        sp = StopProtocol()
        ack = sp.acknowledge_stop(source="user", session_id="sess_123", reason="test")
        assert isinstance(ack, StopAck)
        assert ack.source == "user"
        assert ack.session_id == "sess_123"
        assert ack.reason == "test"

    def test_ack_is_written_to_log(self, tmp_path):
        log_path = tmp_path / "ack_log.jsonl"
        sp = StopProtocol(ack_log_path=log_path)
        ack = sp.acknowledge_stop(source="telegram", session_id="sess_456", reason="explicit")
        # File must exist
        assert log_path.exists()
        # Must be valid JSONL
        lines = log_path.read_text().strip().split("\n")
        data = json.loads(lines[0])
        assert data["source"] == "telegram"
        assert data["session_id"] == "sess_456"
        assert data["reason"] == "explicit"

    def test_read_ack_log_returns_entries(self, tmp_path):
        log_path = tmp_path / "ack_log.jsonl"
        sp = StopProtocol(ack_log_path=log_path)
        sp.acknowledge_stop(source="a", session_id="s1")
        sp.acknowledge_stop(source="b", session_id="s2")
        entries = sp.read_ack_log(limit=10)
        assert len(entries) == 2
        assert entries[0]["source"] == "a"
        assert entries[1]["source"] == "b"

    def test_ack_log_is_append_only(self, tmp_path):
        log_path = tmp_path / "ack_log.jsonl"
        sp = StopProtocol(ack_log_path=log_path)
        sp.acknowledge_stop(source="first", session_id="s1")
        sp.acknowledge_stop(source="second", session_id="s2")
        lines = log_path.read_text().strip().split("\n")
        assert len(lines) == 2
        # First entry still present
        assert json.loads(lines[0])["source"] == "first"


class TestHandsOffRegistry:
    """M1 Requirement: Hands-off registry with time-bounded locks."""

    def test_add_hands_off_creates_entry(self, tmp_path):
        registry_path = tmp_path / "registry.json"
        sp = StopProtocol(registry_path=registry_path)
        entry = sp.add_hands_off(entity="ezra", reason="stopped", lock_seconds=3600)
        assert entry.entity == "ezra"
        assert entry.reason == "stopped"
        assert entry.locked_until > entry.locked_at

    def test_is_hands_off_true_for_locked_entity(self, tmp_path):
        registry_path = tmp_path / "registry.json"
        sp = StopProtocol(registry_path=registry_path)
        sp.add_hands_off(entity="allegro", lock_seconds=3600)
        assert sp.is_hands_off("allegro") is True

    def test_is_hands_off_false_for_unknown_entity(self, tmp_path):
        registry_path = tmp_path / "registry.json"
        sp = StopProtocol(registry_path=registry_path)
        assert sp.is_hands_off("unknown") is False

    def test_is_hands_off_false_after_expiry(self, tmp_path):
        registry_path = tmp_path / "registry.json"
        sp = StopProtocol(registry_path=registry_path)
        sp.add_hands_off(entity="temp", lock_seconds=0)  # immediate expiry
        time.sleep(0.01)  # ensure expiry
        assert sp.is_hands_off("temp") is False

    def test_registry_is_persistent(self, tmp_path):
        registry_path = tmp_path / "registry.json"
        sp1 = StopProtocol(registry_path=registry_path)
        sp1.add_hands_off(entity="persisted", lock_seconds=3600)
        # New instance reads same registry
        sp2 = StopProtocol(registry_path=registry_path)
        assert sp2.is_hands_off("persisted") is True

    def test_list_hands_off_returns_active_only(self, tmp_path):
        registry_path = tmp_path / "registry.json"
        sp = StopProtocol(registry_path=registry_path)
        sp.add_hands_off(entity="active", lock_seconds=3600)
        sp.add_hands_off(entity="expired", lock_seconds=0)
        time.sleep(0.01)
        active = sp.list_hands_off()
        assert len(active) == 1
        assert active[0].entity == "active"

    def test_remove_hands_off_clears_lock(self, tmp_path):
        registry_path = tmp_path / "registry.json"
        sp = StopProtocol(registry_path=registry_path)
        sp.add_hands_off(entity="removable", lock_seconds=3600)
        assert sp.is_hands_off("removable") is True
        sp.remove_hands_off("removable")
        assert sp.is_hands_off("removable") is False


class TestFullStop:
    """M1 Requirement: Full stop combines ack + hands-off atomically."""

    def test_full_stop_performs_both_actions(self, tmp_path):
        log_path = tmp_path / "ack_log.jsonl"
        registry_path = tmp_path / "registry.json"
        sp = StopProtocol(ack_log_path=log_path, registry_path=registry_path)
        ack, entry = sp.full_stop(
            session_id="sess_full",
            entity="ezra",
            source="user",
            reason="critical",
            lock_seconds=7200,
        )
        # Stop is active
        assert sp.is_stopped("sess_full") is True
        # Entity is hands-off
        assert sp.is_hands_off("ezra") is True
        # Ack logged
        assert log_path.exists()
        # Registry saved
        assert registry_path.exists()


class Test100PercentCompliance:
    """
    M1 Requirement: 100% compliance test coverage.
    
    These tests prove every code path is reachable and correct.
    """

    def test_all_stop_paths_covered(self):
        """Aggregate proof that all M1 requirements are met."""
        sp = StopProtocol()
        # 1. Pre-tool-check gate works
        assert sp.is_stopped("new_sess") is False
        sp.acknowledge_stop("test", "new_sess")
        assert sp.is_stopped("new_sess") is True
        with pytest.raises(StopInterrupt):
            sp.pre_tool_check("new_sess")
        # 2. STOP_ACK logging works
        acks = sp.read_ack_log(limit=100)
        assert any(a["session_id"] == "new_sess" for a in acks)
        # 3. Hands-off registry works
        entry = sp.add_hands_off("test_entity", lock_seconds=1)
        assert sp.is_hands_off("test_entity") is True
        # 4. Clear/reset works
        sp.clear_stop("new_sess")
        assert sp.is_stopped("new_sess") is False

    def test_edge_cases(self, tmp_path):
        """Boundary conditions and error paths."""
        log_path = tmp_path / "edge_ack.jsonl"
        registry_path = tmp_path / "edge_registry.json"
        sp = StopProtocol(ack_log_path=log_path, registry_path=registry_path)
        # Clearing non-existent stop
        assert sp.clear_stop("never_stopped") is False
        # Removing non-existent entity
        assert sp.remove_hands_off("never_added") is False
        # Reading empty log
        assert sp.read_ack_log() == []
        # Registry with no file
        assert sp.list_hands_off() == []

    def test_invalid_registry_handled_gracefully(self, tmp_path):
        """Corrupted registry must not crash."""
        registry_path = tmp_path / "bad_registry.json"
        registry_path.write_text("{not valid json")
        sp = StopProtocol(registry_path=registry_path)
        # Should start empty, not crash
        assert sp.list_hands_off() == []
        # Should be able to add entries
        sp.add_hands_off("recovery", lock_seconds=60)
        assert sp.is_hands_off("recovery") is True