FabledScribe/tests/test_session_slippage_readout.py

"""Which rules fired this session, which changed an action, which did not (#4216).

WHY THIS CANNOT BE ASKED OF THE SERVER

`rule_usage_events` has no session column — it is per user over a window — so
a SESSION-scoped answer has to be assembled where a session is a thing that
exists. That is the plugin, from four ledgers four hooks already write:

    .rules.ids      an arm NAMED the rule           (a teaser was shown)
    .opened.ids     the session called get_rule     (#4100)
    .acted.ids      the session called rule_outcome (#4216, new here)
    .checkpoint.ids the rule HELD an act            (#4214)

EVERY LINE IS AN OBSERVED TOOL CALL. Nothing asks the model what it followed;
milestone 386 ruled that out because a model asked "did you apply rule 156?"
will say yes. These record what happened.

WHY THE COMPACTION SEAM

A rule read and left unresolved is invisible by construction — it looks
exactly like a rule that worked. The compaction is where that invisibility
becomes permanent: the turns holding the evidence are summarised away, and an
unjudged thing that survives as nothing is how a decision quietly becomes
nobody's. A PreCompact hook's stdout becomes the summariser's instructions
(#3680), so this does not say "you slipped" — it says which ids must be
carried through, which is the one thing a summary can do about it.

WHAT IS PINNED: the arithmetic, and which conditions produce which lines. NOT
pinned: the wording.
"""
from __future__ import annotations

import json
import os
import shutil
import subprocess
import time
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
HOOKS = ROOT / "plugin" / "hooks"
DEFS = HOOKS / "scribe_defs.sh"
PRECOMPACT = HOOKS / "scribe_precompact_preserve.sh"
RECORDER = HOOKS / "scribe_record_outcome.sh"
HOOKS_JSON = HOOKS / "hooks.json"


def _need(*tools):
    for t in tools:
        if shutil.which(t) is None:
            pytest.skip(f"hook runtime tool {t!r} not installed")


def sh(script: str) -> str:
    _need("bash", "awk")
    r = subprocess.run(
        ["bash", "-c", f'set -uo pipefail\n. "{DEFS}"\n{script}'],
        capture_output=True, text=True, timeout=30,
    )
    assert r.returncode == 0, f"exit {r.returncode}: {r.stderr}"
    return r.stdout


def ledgers(tmp_path, *, named=(), opened=(), acted=(), held=()) -> Path:
    """The four ledgers, written THROUGH THE REAL WRITERS.

    Not hand-rolled files: `scribe_rules_append` is what creates the evidence
    twin these read from, so a fixture that wrote the bytes itself would keep
    passing if the twin stopped being written — which is the whole defect
    #4217 found.
    """
    d = tmp_path / "scribe-priorart"
    d.mkdir(parents=True, exist_ok=True)
    script = []
    for name, ids in (("rules", named), ("opened", opened), ("acted", acted)):
        if ids:
            body = "".join(f"{i}\n" for i in ids)
            script.append(
                f"printf '%s' '{body}' | scribe_rules_append \"{d}/s.{name}.ids\""
            )
    for i in held:
        script.append(f'scribe_checkpoint_allowed "{d}/s.checkpoint.ids" {i} >/dev/null')
    if script:
        sh("\n".join(script))
    return d


def readout(d: Path) -> str:
    return sh(f'scribe_slippage_lines "{d}" "s"')


# ── The arithmetic ────────────────────────────────────────────────────────

def minus(a: str, b: str) -> str:
    return sh(f'scribe_ids_minus "{a}" "{b}"').strip()


def test_the_difference_keeps_order_and_drops_members():
    assert minus("1 9 34 156 173", "9 34 156") == "1 173"


def test_an_empty_difference_prints_nothing():
    assert minus("9 34", "9 34") == ""


def test_an_empty_minuend_is_not_an_error():
    assert minus("", "9") == ""


def test_a_repeated_id_is_counted_once():
    """The ledgers are append-only, so the same rule can appear many times."""
    assert minus("9 9 34 9", "") == "9 34"


# ── What the readout says ─────────────────────────────────────────────────

def test_the_readout_names_what_was_read(tmp_path):
    out = readout(ledgers(tmp_path, named=[1, 9], opened=[9]))
    assert "read: 9" in out


def test_a_rule_named_and_never_opened_is_named_as_such(tmp_path):
    """The arm talking to nobody. Not an accusation — a teaser skimmed past
    leaves nothing behind — but it is the number that says whether the arm is
    earning its place."""
    out = readout(ledgers(tmp_path, named=[1, 9, 173], opened=[9]))
    assert "never opened" in out
    line = next(ln for ln in out.splitlines() if "never opened" in ln)
    assert "1" in line and "173" in line
    assert " 9" not in line.split(":")[1], "an opened rule is not also unread"


def test_a_rule_read_with_no_outcome_is_the_headline(tmp_path):
    """THE MILESTONE'S WHOLE SUBJECT. Read and unresolved is arithmetically
    identical to read and followed, and this is the only place that difference
    gets carried across the seam."""
    out = readout(ledgers(tmp_path, named=[9, 34], opened=[9, 34], acted=[34]))
    assert "READ WITH NO OUTCOME RECORDED" in out
    line = next(ln for ln in out.splitlines() if "NO OUTCOME" in ln)
    assert "9" in line
    assert "rule_outcome" in line, "a finding with no remedy is a complaint"


def test_a_rule_that_held_an_act_is_reported_separately(tmp_path):
    """The strongest evidence a rule changed something: it stopped a call
    before it ran (#4214). Kept apart from `read` because reading a rule and
    having it alter what you did are different claims."""
    out = readout(ledgers(tmp_path, named=[156], opened=[156], held=[156]))
    assert "held an act" in out and "156" in out


def test_a_session_that_resolved_everything_makes_no_accusation(tmp_path):
    """Traffic is always reported — it is what the static instructions above
    already ask for in prose. The SUBTRACTIONS are conditional, so a clean
    session gets no scolding. '0 rules unresolved' on every compaction is how
    a readout teaches its reader to skip it."""
    out = readout(ledgers(tmp_path, named=[7], opened=[7], acted=[7]))
    assert "read: 7" in out
    assert "NO OUTCOME" not in out
    assert "never opened" not in out


def test_a_session_no_rule_touched_says_nothing_at_all(tmp_path):
    """Rule 115's reasoning: a fresh install must not be told something is
    wrong when the truth is that nothing has happened yet."""
    d = tmp_path / "scribe-priorart"
    d.mkdir(parents=True)
    assert readout(d).strip() == ""


# ── The hook that carries it ──────────────────────────────────────────────

def run_precompact(event: dict, tmpdir: Path) -> subprocess.CompletedProcess:
    _need("bash")
    env = dict(os.environ)
    env["TMPDIR"] = str(tmpdir)
    return subprocess.run(["bash", str(PRECOMPACT)], input=json.dumps(event),
                          capture_output=True, text=True, timeout=30, env=env)


def test_the_compaction_hook_appends_the_readout(tmp_path):
    ledgers(tmp_path, named=[1, 9], opened=[9], acted=[])
    r = run_precompact({"session_id": "s", "trigger": "manual"}, tmp_path)
    assert r.returncode == 0
    assert "Preserve the following literally" in r.stdout, "static half intact"
    assert "READ WITH NO OUTCOME RECORDED" in r.stdout


def test_the_compaction_hook_still_exits_zero_with_no_session(tmp_path):
    """An event with no session_id has no ledgers to read. The static
    instructions still go out — they are the part that matters most, and
    losing them because a measurement was unavailable would be the worse
    trade."""
    r = run_precompact({"trigger": "auto"}, tmp_path)
    assert r.returncode == 0
    assert "Preserve the following literally" in r.stdout


def test_the_compaction_hook_never_emits_a_json_envelope(tmp_path):
    """Regression guard on the change that added the readout: for PreCompact
    the envelope is pasted into the summariser's prompt rather than
    unwrapped."""
    ledgers(tmp_path, named=[1], opened=[1])
    r = run_precompact({"session_id": "s", "trigger": "manual"}, tmp_path)
    assert not r.stdout.lstrip().startswith("{")


# ── The recorder that makes `acted` mean anything ─────────────────────────

def run_recorder(event: dict, tmpdir: Path) -> subprocess.CompletedProcess:
    _need("bash")
    env = {"PATH": os.environ["PATH"], "HOME": str(tmpdir), "TMPDIR": str(tmpdir)}
    return subprocess.run(["bash", str(RECORDER)], input=json.dumps(event),
                          capture_output=True, text=True, timeout=30, env=env)


def test_declaring_an_outcome_is_recorded(tmp_path):
    r = run_recorder(
        {"session_id": "s", "tool_input": {"rule_id": 156, "outcome": "applied"}},
        tmp_path,
    )
    assert r.returncode == 0
    assert (tmp_path / "scribe-priorart" / "s.acted.ids").read_text().startswith("156\t")


def test_the_entry_is_stamped_like_its_siblings(tmp_path):
    """One reader ages all three ledgers, so all three carry a timestamp."""
    run_recorder({"session_id": "s", "tool_input": {"rule_id": 9}}, tmp_path)
    line = (tmp_path / "scribe-priorart" / "s.acted.ids").read_text().strip()
    assert "\t" in line and line.split("\t")[1].isdigit()


@pytest.mark.parametrize("event", [
    {}, {"session_id": "s"}, {"tool_input": {"rule_id": 9}},
    {"session_id": "s", "tool_input": {"rule_id": "not-a-number"}},
])
def test_an_unusable_event_records_nothing_and_still_exits_zero(event, tmp_path):
    """A bookkeeping failure must never turn a successful tool call into a
    hook error."""
    r = run_recorder(event, tmp_path)
    assert r.returncode == 0
    assert not (tmp_path / "scribe-priorart" / "s.acted.ids").exists()


def test_the_recorder_is_registered_on_the_rule_outcome_tool():
    """A ledger nothing writes to reads as 'nothing was acted on', which is
    the exact false finding this milestone exists to avoid producing."""
    cfg = json.loads(HOOKS_JSON.read_text())
    entries = cfg["hooks"]["PostToolUse"]
    assert any(
        "rule_outcome" in e.get("matcher", "")
        and any("scribe_record_outcome.sh" in h["command"] for h in e["hooks"])
        for e in entries
    )


def test_the_recorder_is_shell_valid():
    _need("bash")
    subprocess.run(["bash", "-n", str(RECORDER)], check=True)


# ── Evidence outlives the compaction; exclusions do not (#4217) ────────────
#
# THE DEFECT THIS PINS, measured before it was fixed: across six sessions on
# the instance this was built on, 208 `get_rule` calls produced 3 surviving
# ledger entries. `.opened.ids` was doing two jobs with opposite lifetimes —
# "this context holds the rule" (must be forgotten at a compaction, and three
# hooks depend on that) and "this was opened" (which no compaction makes
# untrue). The sweep, correct for the first, was deleting the second.

def test_the_readout_is_unchanged_by_the_sweep(tmp_path):
    """The readout runs AT the compaction. If the sweep took its inputs it
    would report only the last stretch of a session and read as though it had
    reported all of it."""
    d = ledgers(tmp_path, named=[9, 34, 156], opened=[9, 156], acted=[156], held=[156])
    before = readout(d)
    sh(f'TMPDIR="{tmp_path}" scribe_clear_session_ledgers s')
    assert readout(d) == before
    assert "READ WITH NO OUTCOME RECORDED: 9" in readout(d)


def test_the_sweep_still_clears_what_the_context_no_longer_holds(tmp_path):
    """The other half, and it must keep working: after a compaction the agent
    genuinely does not hold what it was shown, so an arm that stayed quiet on
    the strength of the old ledger would be silent about a rule the context
    has lost."""
    d = ledgers(tmp_path, named=[9], opened=[9])
    sh(f'TMPDIR="{tmp_path}" scribe_clear_session_ledgers s')
    assert not (d / "s.rules.ids").exists()
    assert not (d / "s.opened.ids").exists()
    assert (d / "s.opened.keep.ids").exists()


def test_the_evidence_twin_is_not_aged_out(tmp_path):
    """An exclusion ledger ages by TTL, and should: a rule named two hours ago
    is not in this context. A rule OPENED two hours ago was still opened."""
    d = tmp_path / "scribe-priorart"
    d.mkdir(parents=True, exist_ok=True)
    stale = int(time.time()) - 99999
    (d / "s.opened.keep.ids").write_text(f"9\t{stale}\n")
    (d / "s.rules.keep.ids").write_text(f"9\t{stale}\n")
    assert "read: 9" in readout(d)


def test_every_ledger_written_through_the_appender_gets_a_twin(tmp_path):
    """Derived in one place rather than listed at the call sites — a list is
    what broke this before, and the next ledger somebody adds should be born
    on the right side without anyone remembering to say so."""
    d = tmp_path / "scribe-priorart"
    d.mkdir(parents=True, exist_ok=True)
    sh(f"""printf '7\n' | scribe_rules_append "{d}/s.brandnew.ids" """)
    assert (d / "s.brandnew.keep.ids").read_text().startswith("7\t")
    sh(f'TMPDIR="{tmp_path}" scribe_clear_session_ledgers s')
    assert (d / "s.brandnew.keep.ids").exists()


def test_a_checkpoint_budget_is_restored_by_a_compaction(tmp_path):
    """The cap counts the SWEPT file on purpose. After a compaction this
    context has not read the rule, so the budget to stop an act on it is
    honestly fresh — while the record that a stop already happened stays."""
    d = tmp_path / "scribe-priorart"
    d.mkdir(parents=True, exist_ok=True)
    f = d / "s.checkpoint.ids"
    sh(f'scribe_checkpoint_allowed "{f}" 156 >/dev/null')
    r = subprocess.run(
        ["bash", "-c", f'. "{DEFS}"\nscribe_checkpoint_allowed "{f}" 156'],
        capture_output=True, text=True, timeout=30,
    )
    assert r.returncode != 0, "a rule does not get to stop the same session twice"
    sh(f'TMPDIR="{tmp_path}" scribe_clear_session_ledgers s')
    sh(f'scribe_checkpoint_allowed "{f}" 156 >/dev/null')
    assert (d / "s.checkpoint.keep.ids").read_text().count("156") == 2