FabledScribe/tests/test_rule_opened_ledger.py

"""Named is not read: the ledger's third state (#4100).

WHY THIS EXISTS

Milestone 386 made a repeat REFERENCED rather than withheld, and the line it
chose says "You saw it earlier this session". That claim was never checked.
The arms emit a TEASER — title, trigger, `get_rule(N)` — so a session can be
shown a rule twenty times and never read a word of it, and a compaction
summarises the teaser away leaving nothing at all. The server was asserting
something about the reader's context it had no way to know.

`.opened.ids` is the observable half, written by a PostToolUse hook from the
`get_rule` call itself. That is why this is not the self-report 386 rejected:
the objection was to ASKING a model about its own context, and a tool call is
an event the harness reports whether anyone asks or not.

WHAT THIS PINS

The three states and their three lines, the hook that records an open, and the
fact that BOTH ledgers die together on a compaction — a session told "you
opened it earlier" about a rule that was just summarised out of its context
would be a more confident version of the bug this removes.
"""
from __future__ import annotations

import json
import os
import re
import shutil
import subprocess
from pathlib import Path
from unittest.mock import MagicMock

import pytest

from scribe.services.plugin_context import _rule_hint_line

ROOT = Path(__file__).resolve().parents[1]
HOOKS = ROOT / "plugin" / "hooks"
RECORDER = HOOKS / "scribe_record_opened.sh"


def _rule(rid=156, kind="rule"):
    return MagicMock(id=rid, kind=kind, title="A wait with no deadline is a bug",
                     when_to_apply="crossing a process boundary")


# ── the three lines ────────────────────────────────────────────────────────

def test_a_rule_never_surfaced_is_offered_as_new():
    line = _rule_hint_line(_rule(), where="here", seen=False, held=False)
    assert "not in this session's loaded set" in line


def test_a_rule_named_but_not_opened_says_so_and_still_invites():
    """The state that did not exist. It must NOT claim the reader saw it, and
    must still carry the pull pointer — a session that skipped the teaser is
    in nearly the position of one that was never shown it."""
    line = _rule_hint_line(_rule(), where="here", seen=True, held=False)
    assert "not opened" in line
    assert "get_rule(156)" in line
    assert "You saw it earlier" not in line, (
        "the middle state is claiming the reader read something they did not"
    )


def test_a_rule_the_session_opened_is_described_as_opened():
    line = _rule_hint_line(_rule(), where="here", seen=True, held=True)
    assert "opened it earlier" in line
    assert "get_rule(156)" in line


def test_the_three_states_produce_three_different_lines():
    """Guard against a refactor collapsing two branches: each state has to be
    distinguishable, or the distinction this milestone bought is gone while
    every individual assertion above still passes."""
    lines = {
        _rule_hint_line(_rule(), where="here", seen=s, held=h)
        for s, h in ((False, False), (True, False), (True, True))
    }
    assert len(lines) == 3


def test_held_outranks_seen_regardless_of_kind():
    """`kind` moves the head and the ledger moves the tail; #3497's history is
    the two being reasoned about together and one of them being forgotten."""
    for kind in ("rule", "preference"):
        line = _rule_hint_line(_rule(kind=kind), where="here", seen=True, held=True)
        assert "opened it earlier" in line


# ── the recorder ───────────────────────────────────────────────────────────

def _run_recorder(event: dict, tmp: Path) -> Path:
    for tool in ("bash", "jq"):
        if shutil.which(tool) is None:
            pytest.skip(f"hook runtime tool {tool!r} not installed")
    env = {"PATH": os.environ["PATH"], "HOME": str(tmp), "TMPDIR": str(tmp)}
    out = subprocess.run(["bash", str(RECORDER)], input=json.dumps(event),
                         capture_output=True, text=True, env=env, timeout=30)
    assert out.returncode == 0, out.stderr
    return tmp / "scribe-priorart" / "s1.opened.ids"


def test_opening_a_rule_is_recorded(tmp_path):
    led = _run_recorder(
        {"session_id": "s1", "tool_name": "mcp__scribe__get_rule",
         "tool_input": {"rule_id": 156}}, tmp_path)
    assert led.exists()
    assert led.read_text().split("\t")[0] == "156"


def test_the_entry_is_stamped_so_it_ages_like_the_naming_ledger(tmp_path):
    """Both ledgers are read by `scribe_rules_live`, which ages on that stamp.
    An unstamped entry never expires — bounded, but it would mean an opened
    rule stays 'opened' for a session's whole life."""
    led = _run_recorder(
        {"session_id": "s1", "tool_name": "mcp__scribe__get_rule",
         "tool_input": {"rule_id": 9}}, tmp_path)
    parts = led.read_text().strip().split("\t")
    assert len(parts) == 2 and parts[1].isdigit()


@pytest.mark.parametrize("event", [
    {"session_id": "s1", "tool_name": "mcp__scribe__get_rule", "tool_input": {}},
    {"session_id": "s1", "tool_name": "mcp__scribe__get_rule",
     "tool_input": {"rule_id": "../../etc"}},
    {"tool_name": "mcp__scribe__get_rule", "tool_input": {"rule_id": 5}},
])
def test_an_event_with_nothing_usable_records_nothing_and_still_exits_zero(event, tmp_path):
    """A hook that fails a tool call over bookkeeping is worse than one that
    misses an id: the cost of a miss is one extra line, in the direction that
    shows more rather than less."""
    led = _run_recorder(event, tmp_path)
    assert not led.exists()


# ── the two ledgers stay in step ───────────────────────────────────────────

def test_a_compaction_clears_both_ledgers():
    """The one that would be worst to get wrong. `.opened.ids` describes a
    context the compaction just destroyed, so keeping it while clearing the
    naming ledger would have the arms telling a freshly-summarised session
    "you opened it earlier" about a rule now nowhere in its context.
    """
    sh = (HOOKS / "scribe_session_context.sh").read_text()
    block = sh.split("case \"$source\" in")[1].split("esac")[0]
    assert "compact|clear)" in block
    for led in (".rules.ids", ".opened.ids"):
        assert re.search(rf"rm -f .*{re.escape(led)}", block), (
            f"{led} survives a compaction that destroyed what it describes"
        )


def test_the_recorder_is_registered_on_the_get_rule_tool():
    hooks = json.loads((HOOKS / "hooks.json").read_text())["hooks"]
    posts = hooks["PostToolUse"]
    mine = [b for b in posts
            if any("scribe_record_opened.sh" in h["command"] for h in b["hooks"])]
    assert len(mine) == 1, "the opened-recorder is not registered exactly once"
    # An MCP tool's server segment varies with how the plugin was installed, so
    # the matcher must not pin one spelling of it.
    matcher = mine[0]["matcher"]
    assert re.fullmatch(matcher, "mcp__plugin_scribe_scribe__get_rule"), matcher
    assert re.fullmatch(matcher, "mcp__scribe__get_rule"), matcher
    assert not re.fullmatch(matcher, "Bash"), matcher