FabledScribe/tests/test_pre_act_checkpoint.py

"""A rule put IN FRONT of an act, not beside its result.

WHY THIS EXISTS

Every other rule surface in this plugin returns `additionalContext`, which
Claude Code delivers alongside the tool RESULT. By the time the line is read
the call is written, so the rule reads as commentary on a decision already
made — and the common miss is acting on the thing in hand without reading
the contract around it.

A checkpoint spends the same retrieval differently. The hook returns a `deny`,
the act does not run, and the rule's own text can be read before the call
exists. The remedy is one `get_rule` call and the act may then be re-submitted
unchanged.

WHAT THIS PINS, AND WHAT IT DELIBERATELY DOES NOT

Pinned: the four conditions under which an act may be held, the two guards
that stop it recurring, and the shape of the envelope. Not pinned: the bar
itself (a tuning value, and a test asserting 0.80 would fail on every retune
while proving nothing) or the wording of the reason (prose, and it will be
rewritten). The cases below express scores relative to the constant.

THE ONE BOUNDARY THAT IS A STANDING DECISION, NOT A DESIGN CHOICE. Only the
ACTION arm can hold. `scribe_prior_art.sh` carries a tested property that it
never returns a permissionDecision — a recall aid may not stand in the way of
a write — and a checkpoint does not get to quietly overturn that. The write-path arm computes the same block and returns
it, so the decision can be revisited with evidence; the hook ignores it. The
last test here asserts that boundary holds, because it is exactly the kind of
property that erodes when someone extends the feature later.
"""
import json
import shutil
import subprocess
from pathlib import Path

import pytest

from scribe.services.plugin_context import (
    _CHECKPOINT_DEFAULT,
    CHECKPOINT_SESSION_CAP,
    _rule_band,
    checkpoint_for,
)
from tests.helpers import fake_rule

ROOT = Path(__file__).resolve().parents[1]
HOOKS = ROOT / "plugin" / "hooks"
DEFS = HOOKS / "scribe_defs.sh"
ACTION_HOOK = HOOKS / "scribe_tool_rules.sh"
WRITE_HOOK = HOOKS / "scribe_prior_art.sh"

WHERE = "this Bash call"
TRIGGER = "about to assert in a commit message what CI said"


def hit(score: float, rule_id: int, **attrs):
    # `attrs` overrides rather than duplicates: passing `when_to_apply=""` for
    # the no-trigger case alongside a hard-coded default is a duplicate keyword
    # and a TypeError, not a test.
    fields = {"id": rule_id, "title": f"rule {rule_id}", "when_to_apply": TRIGGER}
    fields.update(attrs)
    return (score, fake_rule(**fields))


def hold(kept, held=(), floor=None):
    return checkpoint_for(
        kept, held=set(held), floor=_CHECKPOINT_DEFAULT if floor is None else floor,
        where=WHERE,
    )


# ── The four conditions ───────────────────────────────────────────────────

def test_nothing_retrieved_holds_nothing():
    """The common case by a wide margin, and the one that must be cheapest."""
    assert hold([]) == {}


def test_a_hit_under_the_bar_is_a_hint_and_not_a_stop():
    """The hint arms keep working at their own floor; only a hit the corpus is
    confident about is allowed to cost a round trip."""
    assert hold([hit(_CHECKPOINT_DEFAULT - 0.001, 1)]) == {}


def test_a_hit_on_the_bar_holds_the_act():
    got = hold([hit(_CHECKPOINT_DEFAULT, 1)])
    assert got["rule_id"] == 1
    assert got["where"] == WHERE


def test_a_preference_never_holds_an_act():
    """A preference says how something has been done and following it is what
    keeps work consistent; a rule says what happens if you do not. Stopping an
    act over a preference would assert a force the record does not claim, and
    the renderer already keeps that distinction in the word that names it."""
    assert hold([hit(0.99, 2, kind="preference")]) == {}


def test_a_rule_the_session_already_opened_never_holds_an_act():
    """`held` is observable — a PostToolUse hook watches for the `get_rule`
    call — so this is a recorded event, not a model's self-report
    about its own context. A session that read the rule has already had the
    thing the checkpoint exists to produce, and holding it again would punish
    the behaviour being asked for."""
    assert hold([hit(0.99, 3)], held={3}) == {}


def test_holding_one_rule_does_not_excuse_another():
    assert hold([hit(0.99, 4)], held={3})["rule_id"] == 4


def test_only_the_bands_top_hit_may_hold():
    """`_rule_band` keeps a SET so an act can surface several rules, but the
    ranker's confidence claim attaches to its first element only. A stop
    raised on the fourth line of a band is a stop justified by a score nobody
    claimed — so a preference on top ends the question rather than deferring
    to the rule behind it."""
    band = [hit(0.99, 5, kind="preference"), hit(0.985, 6)]
    assert hold(band) == {}


def test_position_decides_not_score():
    """Deliberately falsifiable from the other side: hits arrive ordered, and
    this reads `kept[0]` rather than re-maximising. A version that took the
    highest score would pick 8 here."""
    assert hold([hit(0.985, 7), hit(0.99, 8)])["rule_id"] == 7


def test_the_band_trims_before_the_checkpoint_sees_it():
    """The two instruments compose in one direction only: the band decides
    what is close enough to show, and the checkpoint reads what survived."""
    kept = _rule_band([hit(0.83, 11), hit(0.81, 12), hit(0.70, 13)])
    assert len(kept) == 2
    assert hold(kept, floor=0.80)["rule_id"] == 11


def test_a_floor_of_zero_disables_rather_than_holding_everything():
    """The failure direction that matters. A bar read as 0 — a cleared
    setting, a bad parse — must turn the feature OFF, never hold the first
    command of every session behind whatever ranked first."""
    assert hold([hit(0.99, 9)], floor=0.0) == {}


# ── What the held act is told ─────────────────────────────────────────────

def test_the_reason_names_the_one_call_that_clears_it():
    """A stop whose remedy is vague costs more than the miss it prevents."""
    reason = hold([hit(0.99, 42)])["reason"]
    assert "get_rule(42)" in reason
    assert "rule 42" in reason
    assert TRIGGER in reason


def test_the_reason_says_the_act_may_proceed_unchanged():
    """Nothing here knows whether the act is wrong, and saying so is what
    keeps the stop from reading as an accusation the system cannot support."""
    assert "re-submit" in hold([hit(0.99, 42)])["reason"]


def test_a_rule_with_no_trigger_renders_without_an_empty_bracket():
    assert "()" not in hold([hit(0.99, 10, when_to_apply="")])["reason"]


def test_every_held_act_carries_a_rendered_reason():
    """Rendered at the one call site inside `checkpoint_for`, never by each
    arm: two arms that each remember to render it are two arms that can stop
    agreeing on what a stop says."""
    assert hold([hit(0.99, 1)])["reason"].strip()


# ── The two guards, in the shell that enforces them ───────────────────────

def sh(script: str) -> subprocess.CompletedProcess:
    for tool in ("bash", "awk"):
        if shutil.which(tool) is None:
            pytest.skip(f"hook runtime tool {tool!r} not installed")
    return subprocess.run(
        ["bash", "-c", f'set -uo pipefail\n. "{DEFS}"\n{script}'],
        capture_output=True, text=True,
    )


@pytest.fixture()
def ledger(tmp_path):
    return tmp_path / "sid.checkpoint.ids"


def allowed(ledger, rule_id) -> bool:
    r = sh(f'scribe_checkpoint_allowed "{ledger}" "{rule_id}" && echo YES || echo NO')
    assert r.returncode == 0, r.stderr
    return r.stdout.strip().endswith("YES")


def test_a_rule_may_hold_at_most_one_act_per_session(ledger):
    """Once the remedy has been offered, repeating it turns a reader who
    decided the rule does not apply into a reader who cannot proceed."""
    assert allowed(ledger, 101)
    assert not allowed(ledger, 101)
    assert allowed(ledger, 102), "a different rule is a different claim"


def test_a_session_cannot_be_held_more_than_the_cap(ledger):
    """The guard on the worst case, not a tuning value: a mis-set floor or a
    corpus that suddenly resembles everything must degrade to a noisy session,
    never to one that cannot make progress."""
    for n in range(CHECKPOINT_SESSION_CAP):
        assert allowed(ledger, 200 + n)
    assert not allowed(ledger, 999)


def test_a_refused_hold_is_not_written_to_the_ledger(ledger):
    """Otherwise the cap eats itself: refusals would count toward it and the
    ledger would grow without a single act ever being held."""
    for n in range(CHECKPOINT_SESSION_CAP):
        allowed(ledger, 200 + n)
    allowed(ledger, 999)
    assert len(ledger.read_text().split()) == CHECKPOINT_SESSION_CAP


@pytest.mark.parametrize("rule_id", ["", "abc", "  "])
def test_an_unreadable_rule_id_refuses_rather_than_holding(ledger, rule_id):
    """Fails CLOSED in the direction that costs nothing. A garbled id cannot
    be written to the ledger, so allowing it would be a stop that recurs
    forever with no way to clear it."""
    assert not allowed(ledger, rule_id)


def test_no_ledger_file_refuses_rather_than_holding(tmp_path):
    """A session with no id gets no ledger, and a stop that cannot be recorded
    is a stop that cannot be capped."""
    assert not allowed("", 101)


def test_the_deny_envelope_is_valid_json_carrying_the_reason():
    r = sh('scribe_json_deny PreToolUse "read \\"rule 9\\" first — then re-submit"')
    assert r.returncode == 0, r.stderr
    out = json.loads(r.stdout)["hookSpecificOutput"]
    assert out["hookEventName"] == "PreToolUse"
    assert out["permissionDecision"] == "deny"
    # Quotes and an em dash survive the escaper — the reason is prose and will
    # contain both, and a broken envelope is silently ignored by the harness.
    assert '"rule 9"' in out["permissionDecisionReason"]
    assert "re-submit" in out["permissionDecisionReason"]


# ── The boundary that is a recorded decision ──────────────────────────────

def test_only_the_action_hook_can_hold_an_act():
    """THE GUARD ON THE RECORDED DECISION, and the reason it is here rather
    than left to memory.

    `scribe_prior_art.sh` runs before every Write and Edit and has never been
    able to stop one. That is a standing decision — a recall aid may not stand
    in the way of the work — and `test_hook_never_returns_a_permission_
    decision` in test_write_path_trigger.py holds the other half of it.

    This milestone has a live argument for extending the checkpoint to writes:
    three of its seven misses were file edits and none of them are reachable
    from the command side. That argument is exactly why this assertion exists.
    A feature with a good reason to spread is the kind that spreads without
    anyone deciding to, and the decision here is the operator's to revisit.
    """
    assert "scribe_json_deny" in ACTION_HOOK.read_text()
    write_code = [
        ln for ln in WRITE_HOOK.read_text().splitlines()
        if not ln.lstrip().startswith("#")
    ]
    assert not any("scribe_json_deny" in ln for ln in write_code)
    assert not any("permissionDecision" in ln for ln in write_code)


def test_the_action_hook_caps_every_hold_it_emits():
    """Structural, and able to fail: the deny and the ledger call
    must appear together. A deny emitted outside the guard is a session that
    can be held by the same rule on every command, which is the one outcome
    both guards exist to prevent."""
    text = ACTION_HOOK.read_text()
    code = [ln for ln in text.splitlines() if not ln.lstrip().startswith("#")]
    denies = [i for i, ln in enumerate(code) if "scribe_json_deny" in ln]
    assert denies, "the action arm no longer emits a hold at all"
    for i in denies:
        window = "\n".join(code[max(0, i - 6):i])
        assert "scribe_checkpoint_allowed" in window, (
            "a hold is emitted without passing the per-rule and per-session "
            "guards first"
        )


def test_the_action_hook_records_what_was_surfaced_whichever_way_it_renders():
    """The ledger append sits BEFORE the checkpoint branch. What the server
    chose to surface happened whichever way this hook then renders it, and
    doing the bookkeeping inside one branch is how the two arms' ledgers came
    to disagree once already."""
    code = [
        ln for ln in ACTION_HOOK.read_text().splitlines()
        if not ln.lstrip().startswith("#")
    ]
    append = next(i for i, ln in enumerate(code) if "scribe_rules_append" in ln)
    deny = next(i for i, ln in enumerate(code) if "scribe_json_deny" in ln)
    assert append < deny


def test_the_action_hook_is_still_shell_valid():
    subprocess.run(["bash", "-n", str(ACTION_HOOK)], check=True)