FabledScribe/tests/test_rule_ledger_ageing.py

"""Entries in the rule exclusion ledger age out (#3751).

WHY THIS EXISTS

#3749 clears the ledger when an EVENT destroys context — a compaction, a
/clear. This covers the case with no event at all: a long session where a rule
was named two hundred turns ago and has simply fallen out of attention. Same
argument #3702 made at the tier level (present in context and salient at the
moment are different properties), applied to time instead of to tier.

WHAT IS PINNED, AND WHAT DELIBERATELY IS NOT

Not the TTL's value. 2700 is a judgement made to be revised once there is
telemetry to revise it with, and a test asserting `== 2700` would turn a
legitimate tuning change into a red build — which is how a number nobody may
touch gets one. The tests read the constant out of the shell and assert the
PROPERTY around it, so any TTL works and only a broken comparison fails.

Runs the real shell, like `test_session_context_ledger` and the after-write
hook's tests, and deliberately with no SCRIBE_URL/SCRIBE_TOKEN: ageing is
local, keyless and networkless, and must work on an instance that is
unreachable or was never configured.
"""
from __future__ import annotations

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

import pytest

PLUGIN = Path(__file__).resolve().parents[1] / "plugin" / "hooks"
DEFS = PLUGIN / "scribe_defs.sh"


def _ttl() -> int:
    """The TTL as the shell defines it — never a copy of the number."""
    m = re.search(r"^_SCRIBE_RULE_TTL=(\d+)", DEFS.read_text(), re.M)
    assert m, "_SCRIBE_RULE_TTL is gone from scribe_defs.sh"
    return int(m.group(1))


def _live(ledger: Path) -> list[str]:
    """`scribe_rules_live` against a real ledger, as the hooks call it."""
    for tool in ("bash", "awk"):
        if shutil.which(tool) is None:
            pytest.skip(f"hook runtime tool {tool!r} not installed")
    proc = subprocess.run(
        ["bash", "-c", f'. "{DEFS}"; scribe_rules_live "{ledger}"'],
        capture_output=True, text=True, timeout=30,
        env={"PATH": os.environ["PATH"]},  # no credentials — see the docstring
    )
    assert proc.returncode == 0, proc.stderr
    out = proc.stdout.strip()
    return out.split(",") if out else []


def _write_ledger(ledger: Path, entries: list[tuple[str, int | None]]) -> None:
    """A ledger written in AGES rather than absolute stamps.

    Named for what it is: `tests/helpers._now` is a datetime for record
    fixtures and this file wants the shell's epoch, so they are deliberately
    not the same helper and deliberately not the same name. `time.time()`
    rather than shelling out to `date` — both read the one system clock, so
    the subprocess bought nothing.
    """
    now = int(time.time())
    ledger.write_text("".join(
        f"{rid}\n" if ago is None else f"{rid}\t{now - ago}\n"
        for rid, ago in entries
    ))


def test_an_old_entry_ages_out_and_a_recent_one_does_not(tmp_path):
    """BOTH DIRECTIONS IN ONE ASSERTION (rule 167), because either half alone
    passes against a broken implementation.

    A test that only checks the old id is gone passes against a helper that
    returns nothing at all — which would drop every exclusion and bring back
    the repetition the ledger exists to prevent. A test that only checks the
    recent id survives passes against the flat `tr '\\n' ','` read this
    replaces, i.e. against the defect itself.
    """
    ttl = _ttl()
    ledger = tmp_path / "s.rules.ids"
    _write_ledger(ledger, [("156", ttl + 600), ("168", 60)])

    assert _live(ledger) == ["168"], (
        f"expected the entry {ttl + 600}s old to age out and the 60s-old one "
        f"to survive (TTL {ttl}s). An empty list means nothing is excluded any "
        f"more; both ids means nothing ages."
    )


def test_a_repeatedly_surfaced_rule_appears_once_in_the_exclusion_list(tmp_path):
    """List hygiene, and it is not cosmetic.

    The ledger is append-only, so a rule surfaced five times has five lines.
    The output is spliced straight into `&exclude_rule_ids=` and parsed by the
    server as a list; duplicates make that list grow without bound over a long
    session, and a trailing or doubled comma is an empty element the parser has
    to decide what to do with. Neither is visible from the hint.

    (There was a boundary test here — `ttl` vs `ttl + 1`. It was racy by
    construction: a ledger written at T is read at T+n, so the entry ages by
    however long the subprocess took and the two cases swap. A one-second
    distinction on a 45-minute window is also not a behaviour anyone can
    observe, so it was pinning a flake rather than a property.)
    """
    ledger = tmp_path / "s.rules.ids"
    _write_ledger(ledger, [("156", 60), ("168", 30), ("156", 10), ("156", 5)])

    live = _live(ledger)
    assert live == ["156", "168"], (
        f"got {live}: an id repeated in the ledger must appear once in the "
        f"exclusion list, in first-seen order"
    )
    raw = subprocess.run(
        ["bash", "-c", f'. "{DEFS}"; scribe_rules_live "{ledger}"'],
        capture_output=True, text=True, timeout=30,
        env={"PATH": os.environ["PATH"]},
    ).stdout.strip()
    assert ",," not in raw and not raw.endswith(","), (
        f"the exclusion list has an empty element: {raw!r}"
    )


def test_a_rule_that_ages_out_and_returns_does_not_ping_pong(tmp_path):
    """THE FAILURE THIS COSTS THE MOST TO GET WRONG.

    The ledger is append-only, so a rule that ages out, gets surfaced fresh and
    is appended again has TWO lines. An implementation reading the FIRST leaves
    it permanently expired — and it then re-announces itself on every single
    call for the rest of the session. The mechanism meant to quieten things
    becomes the loudest thing in the hint.
    """
    ttl = _ttl()
    ledger = tmp_path / "s.rules.ids"
    _write_ledger(ledger, [("156", ttl + 600)])

    subprocess.run(
        ["bash", "-c", f'. "{DEFS}"; printf \'156\\n\' | scribe_rules_append "{ledger}"'],
        capture_output=True, text=True, timeout=30,
        env={"PATH": os.environ["PATH"]},
    )

    assert _live(ledger) == ["156"], (
        "a re-surfaced rule read as still expired; the most recent entry for "
        "an id must win, or it re-announces itself on every call from here on"
    )


def test_a_bare_id_from_the_old_format_is_treated_as_live(tmp_path):
    """The compatibility decision, asserted rather than described.

    Every session in flight when this ships has a ledger of bare ids. Reading
    unknown as EXPIRED would make all of them re-announce every rule they had
    already been told, all at once — the exact noise this feature exists to
    prevent, delivered by the feature itself on the day it ships.

    Unknown means "not measured", never "old" — the same discipline the
    nullable retrieval_logs columns use. Those entries never age, which is
    bounded, because the session ends.
    """
    ttl = _ttl()
    ledger = tmp_path / "s.rules.ids"
    ledger.write_text("156\n168\n")
    assert _live(ledger) == ["156", "168"]

    # And a bare id keeps its own meaning next to a stale stamped one, rather
    # than the file falling back to one format or the other wholesale.
    _write_ledger(ledger, [("9", None), ("156", ttl + 600)])
    assert _live(ledger) == ["9"]


def test_a_missing_or_empty_ledger_excludes_nothing(tmp_path):
    """Fail-open, like every other path in these hooks: no ledger means no
    exclusions, not an error and not a stray comma the server would parse as
    an id."""
    assert _live(tmp_path / "never-written.rules.ids") == []
    empty = tmp_path / "empty.rules.ids"
    empty.write_text("")
    assert _live(empty) == []


def test_both_hooks_read_the_ledger_through_the_shared_helper():
    """Structural, because the defect this would reintroduce is invisible.

    Two hooks share one ledger so a rule named by one arm is not re-offered by
    the other. If either goes back to reading the file flat, that hook stops
    ageing while its sibling keeps ageing — the two disagree about which rules
    are live, and nothing in the output says so.

    Pinned on the flat-read SHAPE rather than on a helper name: a hook that
    stops calling the helper but still ages correctly some other way is not a
    defect, and a rename of the helper is not one either.
    """
    for name in ("scribe_prior_art.sh", "scribe_tool_rules.sh"):
        src = (PLUGIN / name).read_text()
        rule_reads = [
            ln for ln in src.splitlines()
            if "rulefile" in ln and "tr " in ln and r"'\n' ','" in ln
        ]
        assert not rule_reads, (
            f"{name} reads the rule ledger flat again: {rule_reads}. That "
            f"hook's exclusions would stop ageing while its sibling's keep "
            f"ageing, and the two would disagree silently about which rules "
            f"this session still holds."
        )
        assert "scribe_rules_live" in src, (
            f"{name} no longer reads the rule ledger through the shared "
            f"helper, so the two hooks can drift on the ledger format"
        )