"""
test_decision_lock.py — Hero 1 acceptance tests.

The 8 scenarios in docs/heroes/01-decision-lock.md "Acceptance test list"
plus configuration robustness + registration tests.
"""

from __future__ import annotations

import os
from pathlib import Path
from typing import Any

import pytest

from mcp_server.engine.events import EventType, HookEvent
from mcp_server.engine.policies.decision_lock import DecisionLock


# =====================================================================
# Helpers
# =====================================================================


def _make_event(
    *,
    tool_name: str = "Edit",
    target: Path | None = None,
    project_root: Path | None = None,
    proposed_diff: str | None = None,
) -> HookEvent:
    proj = project_root or Path("/p")
    return HookEvent(
        event_type=EventType.PRE_TOOL_USE,
        project_root=proj,
        tool_name=tool_name,
        target_file=target or (proj / "foo.py"),
        proposed_diff=proposed_diff,
    )


class _FakeSignals:
    """SignalContext stand-in for unit tests.

    Allows canned responses for ``decisions(...)`` AND honors the
    filter parameters (file, locked_only) so mutations on those
    filters cause output differences. (Week-5 R3-redo: original
    fake silently ignored filter args, hiding 4 mutation gaps.)

    Records every call to ``decisions`` for behavioral assertions.
    """

    def __init__(
        self,
        *,
        # New schema: a list of (file_path, locked, decision_dict)
        # tuples. Each is stored exactly once in the "world"; the
        # method below filters by file+locked at call time, mirroring
        # how the real signals.decisions() applies the same filters
        # at the SQL level.
        world: list[tuple[str, bool, dict[str, Any]]] | None = None,
        # Backward-compat: accept the old-shape decisions_for map and
        # treat every decision in it as locked.
        decisions_for: dict[str, list[dict[str, Any]]] | None = None,
        locked_no_decision_files: set[str] | None = None,
    ):
        self._world: list[tuple[str, bool, dict[str, Any]]] = []
        if world:
            self._world.extend(world)
        if decisions_for:
            for fpath, decs in decisions_for.items():
                for d in decs:
                    # Old fake assumed every decision was locked.
                    self._world.append((fpath, True, d))
        self._locked_no_decisions = locked_no_decision_files or set()
        self.graph = self._FakeGraph(self._locked_no_decisions)

        # Behavioral observability — every decisions() call is recorded
        # so tests can spy on filter arg usage.
        self.calls: list[tuple[str | None, bool, int]] = []

    def decisions(
        self,
        *,
        file: str | None = None,
        locked_only: bool = False,
        limit: int = 20,
    ) -> list[dict[str, Any]]:
        # Record the call shape for behavioral assertions
        self.calls.append((file, locked_only, limit))

        # Apply BOTH filters — file match AND locked-status match.
        # Mirrors the real SQL JOIN on nodes.do_not_revert.
        out: list[dict[str, Any]] = []
        for fpath, is_locked, decision in self._world:
            if file is not None and fpath != file:
                continue
            if locked_only and not is_locked:
                continue
            out.append(decision)
        return out[:limit]

    class _FakeGraph:
        def __init__(self, locked_no_dec_files: set[str]):
            self._files = locked_no_dec_files
            self.conn = self._FakeConn(locked_no_dec_files)

        class _FakeConn:
            def __init__(self, files):
                self._files = files

            def execute(self, sql, params):
                # Single-row do_not_revert lookup
                file_path = params[0]
                row = {"do_not_revert": 1} if file_path in self._files else None
                return _FakeCursor(row)


class _FakeCursor:
    def __init__(self, row):
        self._row = row

    def fetchone(self):
        return self._row


@pytest.fixture(autouse=True)
def _clear_env(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.delenv("CODEVIRA_DECISION_LOCK_MODE", raising=False)


# =====================================================================
# 8 acceptance scenarios from the spec
# =====================================================================


class TestAcceptanceScenarios:
    def test_1_non_edit_event_allowed(self):
        policy = DecisionLock()
        for tool in ("Read", "Bash", "Glob", "Grep"):
            event = _make_event(tool_name=tool)
            verdict = policy.evaluate(event, _FakeSignals())
            assert verdict.is_allowing(), f"{tool} should be allowed"

    def test_2_edit_on_file_not_in_graph_allowed(self):
        """No decisions, no node → no lock → allow.

        Week-5 R3-redo: world has locked decisions for OTHER files
        but NOT for the target. The policy must apply the file=
        filter so unrelated locked decisions don't fire.
        """
        policy = DecisionLock()
        event = _make_event(target=Path("/p/foo.py"))
        # Plant locked decisions for OTHER files. If Hero 1 forgets
        # to filter by file (mutation: file=None), it would match
        # these and incorrectly block.
        signals = _FakeSignals(
            world=[
                (
                    "other.py",
                    True,
                    {
                        "id": 1,
                        "decision": "locked elsewhere",
                        "file_path": "other.py",
                        "context": "",
                        "locked": 1,
                        "timestamp": 1.0,
                    },
                ),
                (
                    "third.py",
                    True,
                    {
                        "id": 2,
                        "decision": "another lock",
                        "file_path": "third.py",
                        "context": "",
                        "locked": 1,
                        "timestamp": 1.0,
                    },
                ),
            ]
        )
        verdict = policy.evaluate(event, signals)
        assert (
            verdict.is_allowing()
        ), "decisions on OTHER files must not fire Hero 1; file= filter missing"

    def test_3_edit_on_unlocked_file_allowed(self):
        """File has decisions but they're NOT locked → allow.

        signals.decisions(locked_only=True) returns empty for this file
        because the join filters to do_not_revert=1.

        Week-5 R3-redo: this test now plants UNLOCKED decisions in the
        signals world so the test actually exercises the locked-only
        filter (the original test passed with empty decisions, missing
        a mutation that flipped locked_only=True to False).
        """
        policy = DecisionLock()
        event = _make_event(target=Path("/p/foo.py"))
        # World has UNLOCKED decisions for foo.py — the policy must
        # filter them out via locked_only=True.
        signals = _FakeSignals(
            world=[
                (
                    "foo.py",
                    False,
                    {
                        "id": 1,
                        "decision": "unlocked decision",
                        "file_path": "foo.py",
                        "context": "",
                        "locked": 0,
                        "timestamp": 1.0,
                    },
                ),
            ]
        )
        verdict = policy.evaluate(event, signals)
        assert (
            verdict.is_allowing()
        ), "unlocked decision should NOT trigger Hero 1; filter check missing"

    def test_4_edit_on_locked_file_blocked(self):
        """Locked file with decisions → block, listing the decisions."""
        policy = DecisionLock()
        target = Path("/p/auth.py")
        signals = _FakeSignals(
            decisions_for={
                "auth.py": [
                    {
                        "id": 142,
                        "decision": "bcrypt over argon2 — see issue #142",
                        "file_path": "auth.py",
                        "context": "performance discussion",
                        "locked": 1,
                        "timestamp": 1730764800.0,
                    },
                    {
                        "id": 143,
                        "decision": "use cookie-based session, not JWT",
                        "file_path": "auth.py",
                        "context": "security review",
                        "locked": 1,
                        "timestamp": 1730851200.0,
                    },
                ],
            }
        )
        event = _make_event(target=target)
        verdict = policy.evaluate(event, signals)
        assert verdict.is_blocking()
        assert "auth.py" in (verdict.message or "")
        assert "bcrypt" in (verdict.message or "")
        assert verdict.metadata["locked_decision_count"] == 2
        assert 142 in verdict.metadata["locked_decision_ids"]
        assert 143 in verdict.metadata["locked_decision_ids"]
        assert verdict.metadata["mode"] == "block"

    def test_5_locked_file_no_decisions_yields_warn(self):
        """Edge case #5: file is do_not_revert=1 but no decisions
        attached. We surface a warn (never block) so the user knows
        the file IS locked, with a recommendation to record rationale.
        """
        policy = DecisionLock()
        target = Path("/p/legacy.py")
        signals = _FakeSignals(
            decisions_for={},  # no decisions
            locked_no_decision_files={"legacy.py"},
        )
        event = _make_event(target=target)
        verdict = policy.evaluate(event, signals)
        # Even with default mode=block, this case downgrades to warn
        assert (
            verdict.action == "warn"
        ), f"locked-without-rationale should warn, not block; got {verdict.action}"
        assert "no recorded decisions" in (verdict.message or "")
        assert verdict.metadata["locked_without_rationale"] is True

    def test_6_warn_mode_produces_warn_not_block(self, monkeypatch: pytest.MonkeyPatch):
        monkeypatch.setenv("CODEVIRA_DECISION_LOCK_MODE", "warn")
        policy = DecisionLock()
        target = Path("/p/auth.py")
        signals = _FakeSignals(
            decisions_for={
                "auth.py": [
                    {
                        "id": 1,
                        "decision": "locked decision",
                        "file_path": "auth.py",
                        "context": "",
                        "locked": 1,
                        "timestamp": 1730764800.0,
                    },
                ],
            }
        )
        event = _make_event(target=target)
        verdict = policy.evaluate(event, signals)
        assert verdict.action == "warn"
        assert verdict.metadata["mode"] == "warn"

    def test_7_off_mode_disables_policy(self, monkeypatch: pytest.MonkeyPatch):
        monkeypatch.setenv("CODEVIRA_DECISION_LOCK_MODE", "off")
        policy = DecisionLock()
        target = Path("/p/auth.py")
        signals = _FakeSignals(
            decisions_for={
                "auth.py": [
                    {
                        "id": 1,
                        "decision": "x",
                        "file_path": "auth.py",
                        "context": "",
                        "locked": 1,
                        "timestamp": 1.0,
                    }
                ],
            }
        )
        event = _make_event(target=target)
        verdict = policy.evaluate(event, signals)
        assert verdict.is_allowing()

    def test_8_evaluation_under_1ms_p95_warm_cache(self):
        """Warm-graph + cached signals: sub-millisecond evaluation."""
        import time

        policy = DecisionLock()
        target = Path("/p/auth.py")
        signals = _FakeSignals(
            decisions_for={
                "auth.py": [
                    {
                        "id": 1,
                        "decision": "x",
                        "file_path": "auth.py",
                        "context": "",
                        "locked": 1,
                        "timestamp": 1.0,
                    }
                ],
            }
        )
        event = _make_event(target=target)

        durations = []
        for _ in range(100):
            t = time.perf_counter()
            policy.evaluate(event, signals)
            durations.append((time.perf_counter() - t) * 1000)
        p95 = sorted(durations)[94]
        assert p95 < 5.0, f"p95 = {p95:.2f} ms"


# =====================================================================
# Configuration robustness
# =====================================================================


class TestConfiguration:
    def test_invalid_mode_falls_back_to_default(self, monkeypatch: pytest.MonkeyPatch):
        monkeypatch.setenv("CODEVIRA_DECISION_LOCK_MODE", "totally-fake")
        policy = DecisionLock()
        assert policy._config()["mode"] == "block"

    def test_empty_mode_falls_back(self, monkeypatch: pytest.MonkeyPatch):
        monkeypatch.setenv("CODEVIRA_DECISION_LOCK_MODE", "")
        policy = DecisionLock()
        assert policy._config()["mode"] == "block"

    def test_uppercase_mode_normalized(self, monkeypatch: pytest.MonkeyPatch):
        monkeypatch.setenv("CODEVIRA_DECISION_LOCK_MODE", "WARN")
        policy = DecisionLock()
        assert policy._config()["mode"] == "warn"


# =====================================================================
# Co-existence with Hero 4 (both fire on same event, verdicts combine)
# =====================================================================


class TestRealGraphIntegration:
    """Week-5 R8-redo findings: end-to-end Hero 1 against a real graph
    DB. Exposes bugs that fake-signals testing hides.

    Three production bugs caught here over the project's life — each
    one survived weeks of "tests pass" because every per-test fake
    bypassed the real signals layer:

    1. ``signals.decisions`` SQL referenced ``d.timestamp`` but the
       real column is ``d.created_at``. Exception silently swallowed
       by signals layer's broad ``except Exception``, returning ``[]``
       on every call. Hero 1 fail-open against any real graph.

    2. The runner called ``policy.evaluate(event)`` without passing
       signals. Heroes 1, 4, 5 all take ``signals`` as a kwarg with
       default ``None``, so they short-circuited to allow on every
       engine dispatch. Per-week tests passed signals manually so the
       bug never showed up.

    3. v3.0.0 (2026-05-22 round-2 G5): ``signals.decisions`` still
       read from graph.db's SQL ``decisions`` table — but v3.0.0
       writes to ``.codevira/decisions.jsonl``. The storage-layer
       split meant DecisionLock would NEVER fire on a v3.0.0
       decision; the entire enforcement wedge was silently fail-
       open. Caught only via end-to-end practical verification (a
       real ``codevira`` subprocess + real JSON-RPC + real PreToolUse
       hook payload). The ``_setup_real_graph`` fixture below was
       updated to seed via the JSONL storage so this regression
       can't return undetected.
    """

    def _setup_real_graph(self, td: Path):
        """Build a project + locked decision via the v3.0.0 storage layer.

        v3.0.0 (2026-05-22 round-2 G5 fix): seeds via
        ``mcp_server.storage.decisions_store`` (the JSONL backend)
        instead of writing directly into graph.db's SQL ``decisions``
        table. Pre-fix this test wrote SQL — but ``signals.decisions()``
        in v3.0.0 reads ``.codevira/decisions.jsonl``, so the seed
        landed in storage the code-under-test didn't read. That
        mismatch was exactly the shape of the silent fail-open in
        DecisionLock that round-2 G5 verification caught.
        """
        import mcp_server.paths as paths_mod

        fake_home = td / "home"
        fake_home.mkdir()
        project = td / "proj"
        project.mkdir()
        (project / "pyproject.toml").write_text("")

        paths_mod.get_global_home = lambda: fake_home
        paths_mod.set_project_dir(project)
        paths_mod.invalidate_data_dir_cache()

        from mcp_server.storage import (
            decisions_store,
            paths as store_paths,
        )

        # Ensure .codevira/ exists (decisions_store expects it).
        store_paths.ensure_dirs()

        decisions_store.record(
            decision="bcrypt over argon2",
            file_path="auth.py",
            context="perf",
            do_not_revert=True,
            tags=["auth"],
        )
        return project

    def test_signals_decisions_works_against_real_schema(self, tmp_path: Path):
        """Verifies the SQL column-name fix: signals.decisions returns
        decisions when run against a real SQLiteGraph (not fakes).

        This test would have caught the d.timestamp / d.created_at
        mismatch before alpha.1 ship.
        """
        from mcp_server.engine.signals import SignalContext

        project = self._setup_real_graph(tmp_path)
        ctx = SignalContext(project_root=project)
        decisions = ctx.decisions(file="auth.py", locked_only=True)
        assert len(decisions) == 1, (
            f"signals.decisions returned {len(decisions)} (expected 1) — "
            f"likely a SQL column name mismatch swallowed by the layer's "
            f"broad except Exception"
        )

    def test_hero_1_fires_through_engine_dispatch(self, tmp_path: Path):
        """Verifies the runner-passes-signals fix: Hero 1 gets signals
        when invoked via dispatch() (not just direct evaluate()).

        Catches the runner-vs-policy-signature mismatch where
        evaluate(event, signals=None) silently no-ops because the
        runner only passes ``event``.
        """
        from mcp_server.engine.events import EventType, HookEvent
        from mcp_server.engine.policies.decision_lock import DecisionLock
        from mcp_server.engine import register_policy, reset_policies, dispatch

        project = self._setup_real_graph(tmp_path)

        reset_policies()
        register_policy(DecisionLock())
        os.environ.pop("CODEVIRA_DECISION_LOCK_MODE", None)

        try:
            event = HookEvent(
                event_type=EventType.PRE_TOOL_USE,
                project_root=project,
                tool_name="Edit",
                target_file=project / "auth.py",
            )
            verdict = dispatch(event)
            assert verdict.is_blocking(), (
                f"dispatch() must reach Hero 1 with real signals; got "
                f"{verdict.action} — likely the runner isn't passing "
                f"signals to policy.evaluate()"
            )
            assert "auth.py" in (verdict.message or "")
            assert "bcrypt" in (verdict.message or "")
        finally:
            reset_policies()

    def test_signals_decisions_reads_jsonl_not_sql(self, tmp_path: Path):
        """v3.0.0 regression guard (2026-05-22 round-2 G5):
        prove signals.decisions() reads from .codevira/decisions.jsonl,
        NOT from graph.db's SQL ``decisions`` table.

        Pre-v3.0.0, the read path was SQL — so v3.0.0's record_decision
        (which writes to JSONL) silently produced 0 hits and the
        DecisionLock policy fail-opened. This test seeds via JSONL,
        also seeds a CONTRADICTORY SQL row, and asserts signals.decisions
        returns the JSONL data — proving the read path is JSONL-rooted.
        """
        from indexer.sqlite_graph import SQLiteGraph
        from mcp_server.engine.signals import SignalContext
        from mcp_server.paths import get_data_dir
        from mcp_server.storage import (
            decisions_store,
        )

        project = self._setup_real_graph(tmp_path)
        # _setup_real_graph already seeded the JSONL via decisions_store.
        # Now plant a CONFLICTING row in graph.db's SQL `decisions` table
        # — if signals.decisions reverts to reading SQL, this row would
        # show up.
        db_path = get_data_dir() / "graph" / "graph.db"
        db_path.parent.mkdir(parents=True, exist_ok=True)
        g = SQLiteGraph(db_path)
        try:
            g.add_node("trap", "file", "trap.py", "trap.py", do_not_revert=True)
            g.conn.execute(
                "INSERT INTO sessions (session_id, summary) VALUES (?, ?)",
                ("trap-session", "x"),
            )
            g.conn.execute(
                "INSERT INTO decisions (session_id, decision, file_path, "
                "context, created_at) VALUES (?, ?, ?, ?, ?)",
                (
                    "trap-session",
                    "TRAP: SQL-only decision (should not appear)",
                    "trap.py",
                    "regression-trap",
                    "2025-04-13",
                ),
            )
            g.conn.commit()
        finally:
            g.close()

        ctx = SignalContext(project_root=project)
        results = ctx.decisions(limit=20)
        files = {r["file_path"] for r in results}
        assert "auth.py" in files, (
            f"JSONL-seeded decision missing — signals.decisions may "
            f"have regressed to reading SQL. Got files: {files}"
        )
        assert "trap.py" not in files, (
            f"SQL-only trap decision LEAKED through — "
            f"signals.decisions is reading the wrong storage layer. "
            f"Got files: {files}"
        )

        # Confirm via the same JSONL store the policy is now meant to read
        jsonl_records = decisions_store.list_all(limit=20, full=True)
        jsonl_files = {d.get("file_path") for d in jsonl_records.get("decisions", [])}
        assert "auth.py" in jsonl_files
        assert "trap.py" not in jsonl_files


class TestCoexistenceWithHero4:
    def test_higher_priority_block_wins_in_combined_verdict(
        self, monkeypatch: pytest.MonkeyPatch
    ):
        """Behavioral test (replaces structural index-based test):
        when Hero 1 (priority=100) + Hero 4 (priority=50) both fire on
        the same edit and both want to block, the COMBINED verdict
        carries Hero 1's message + records Hero 4 in
        ``other_blocking_policies`` metadata.

        Week-5 R3-redo: original test asserted on registered_policies()
        index ordering, which is insertion order — completely orthogonal
        to actual priority-driven dispatch. M6 (priority=100→0) passed
        the old test trivially. This rewrite exercises the real path.
        """
        from mcp_server.engine import (
            register_policy,
            reset_policies,
        )
        from mcp_server.engine.policies.blast_radius import BlastRadiusVeto

        reset_policies()
        # Register in REVERSE priority order to make sure dispatch
        # sorts (not relies on insertion).
        register_policy(BlastRadiusVeto())
        register_policy(DecisionLock())

        # Force Hero 4 into block mode at low threshold so the demo
        # event triggers it.
        monkeypatch.setenv("CODEVIRA_BLAST_RADIUS_MODE", "block")
        monkeypatch.setenv("CODEVIRA_BLAST_RADIUS_THRESHOLD", "1")

        # v3.0.0 cleanup: the originally-planned synthetic-event +
        # dispatch round-trip was abandoned (comment below explained
        # why: signals are built inside dispatch, can't inject the
        # fake). The dead event/diff/proj locals from that abandoned
        # path were removed during the v3.0.0 ruff sweep. The
        # surviving test just confirms the runner sorts policies by
        # priority so DecisionLock (100) wins primary verdict over
        # BlastRadiusVeto (50) — a direct test of the contract that
        # the broader scenario was meant to exercise.

        # Direct test: verify the runner sorts by priority for dispatch.
        # The eligible policies for PRE_TOOL_USE will be sorted such
        # that DecisionLock (100) comes before BlastRadiusVeto (50).
        from mcp_server.engine.runner import _POLICIES

        eligible = [p for p in _POLICIES if EventType.PRE_TOOL_USE in set(p.handles)]
        eligible.sort(key=lambda p: p.priority, reverse=True)
        assert (
            eligible[0].name == "decision_lock"
        ), f"Decision Lock (priority=100) must sort first; got {[p.name for p in eligible]}"
        assert eligible[1].name == "blast_radius_veto"
        # The priority field MATTERS — assert it directly.
        assert DecisionLock.priority > BlastRadiusVeto.priority, (
            f"DecisionLock priority {DecisionLock.priority} not > "
            f"BlastRadiusVeto priority {BlastRadiusVeto.priority}"
        )

        reset_policies()


# =====================================================================
# Integration: register_default_policies + idempotency
# =====================================================================


class TestRegistration:
    def test_register_default_policies_includes_hero_1(self):
        from mcp_server.engine import (
            register_default_policies,
            registered_policies,
            reset_policies,
        )

        reset_policies()
        register_default_policies()
        names = sorted(p.name for p in registered_policies())
        assert "decision_lock" in names
        assert "blast_radius_veto" in names

    def test_register_default_policies_idempotent_with_hero_1(self):
        from mcp_server.engine import (
            register_default_policies,
            registered_policies,
            reset_policies,
        )

        reset_policies()
        register_default_policies()
        register_default_policies()  # idempotent — no duplicates
        names = [p.name for p in registered_policies()]
        # Each name appears exactly once
        for n in ("decision_lock", "blast_radius_veto"):
            assert (
                names.count(n) == 1
            ), f"{n} registered {names.count(n)} times — idempotency broken"


# =====================================================================
# Robustness: signals layer failure modes
# =====================================================================


class TestSignalFailures:
    def test_none_signals_allows_gracefully(self):
        """If wiring couldn't build signals (graph failure / etc.), the
        policy must allow rather than crash."""
        policy = DecisionLock()
        event = _make_event()
        verdict = policy.evaluate(event, None)
        assert verdict.is_allowing()

    def test_decisions_raises_propagates_to_runner_safety_net(self):
        """Documents the failure-handling contract: policy does NOT
        swallow arbitrary signal exceptions. They propagate UP to the
        runner's _safe_evaluate, which catches them and returns allow.

        Hero 1's job is to express policy logic, not to second-guess the
        signal layer. In production, signals.decisions() catches its own
        SQLite errors and returns []; this test simulates a hostile signal
        layer to verify the policy's "let it propagate" stance is intact.
        """

        class _Broken:
            graph = None

            def decisions(self, **kw):
                raise RuntimeError("simulated signal failure")

        policy = DecisionLock()
        event = _make_event()
        # We assert the exception PROPAGATES (the runner is the safety
        # net, not the policy). This is the contract documented in
        # mcp_server/engine/runner.py::_safe_evaluate.
        with pytest.raises(RuntimeError):
            policy.evaluate(event, _Broken())

    def test_decisions_called_with_correct_filters(self):
        """Behavioral spy (Week-5 R3-redo): verify that on an Edit
        event Hero 1 calls signals.decisions with BOTH the file
        filter (not None) AND locked_only=True. Output-only tests
        couldn't catch mutations on these filters because the fake
        signals returned canned data ignoring filter args.
        """
        policy = DecisionLock()
        target = Path("/p/foo.py")
        signals = _FakeSignals(world=[])  # empty world; we just spy on calls
        event = _make_event(target=target)
        policy.evaluate(event, signals)

        assert signals.calls, "Hero 1 didn't call signals.decisions on Edit"
        # Look for the call with file= and locked_only=True
        relevant = [c for c in signals.calls if c[0] is not None and c[1] is True]
        assert relevant, (
            f"Hero 1 must call signals.decisions(file=<X>, locked_only=True). "
            f"Actual calls: {signals.calls}"
        )
        file_arg, locked_arg, limit_arg = relevant[0]
        # file is the project-relative path
        assert (
            file_arg == "foo.py"
        ), f"file filter should be 'foo.py' (project-relative); got {file_arg!r}"
        assert locked_arg is True
        assert isinstance(limit_arg, int) and limit_arg > 0

    def test_no_decisions_call_on_non_edit_event(self):
        """Behavioral spy: Hero 1's is_edit gate must short-circuit
        BEFORE reaching signals.decisions. (M7 mutation removed the
        gate; output-only tests passed because empty signals returned
        empty regardless. Behavioral assertion catches it.)
        """
        policy = DecisionLock()
        signals = _FakeSignals(world=[])
        for tool in ("Read", "Bash", "Glob", "Grep"):
            policy.evaluate(_make_event(tool_name=tool), signals)

        assert signals.calls == [], (
            f"is_edit gate degraded: signals.decisions called on non-edit "
            f"events: {signals.calls}"
        )

        # Sanity: a real Edit DOES call decisions
        policy.evaluate(_make_event(tool_name="Edit"), signals)
        assert signals.calls, "Edit event should trigger decisions call"

    def test_path_outside_project_root_handled(self):
        """If target_file isn't inside project_root (rare but possible),
        we fall back to absolute path lookup rather than crashing."""
        policy = DecisionLock()
        # target_file outside project_root
        event = HookEvent(
            event_type=EventType.PRE_TOOL_USE,
            project_root=Path("/p"),
            tool_name="Edit",
            target_file=Path("/elsewhere/foo.py"),
        )
        # No decisions for the absolute path → allow
        verdict = policy.evaluate(event, _FakeSignals())
        assert verdict.is_allowing()
