FabledScribe/tests/test_session_ledger_clear.py

"""A compaction clears every session ledger, by convention not by list (#4101).

WHY THIS EXISTS

Milestone 386 established the claim: a ledger describes what a session HOLDS,
so the events that destroy context must destroy it too, or the records it names
become permanently unreachable mid-session. `scribe_session_context.sh`
implemented that — for the rules ledger, by name.

Five ledgers live in that directory and two were on the list. The note arms'
three (`.ids`, `.sync.ids`, `.derive.ids`) were left, under a comment asserting
this was "a decision rather than an oversight". It was not a decision, and the
arms left out are the ones where it costs most:

  - their exclusions are HARD — `exclude_ids` is passed into
    `semantic_search_notes` itself, so a surfaced note leaves the result set
    entirely, with no weaker rendering to fall back to the way #3750 gave a
    repeated rule one;
  - and they never AGE — #3751's TTL was added to the rules ledger only.

Hard, permanent and never cleared: a note surfaced in a session's first minute
is unreachable for the rest of it, through any number of compactions.

WHAT THIS PINS

The list was the bug, so the fix cannot be a longer list and neither can the
test. Both sides are asserted:

  1. BEHAVIOUR — the hook is run on a real `compact` event with all five
     ledgers on disk, and all five are gone afterwards. Run rather than
     grepped, because grepping for the names is the hand-maintained pattern
     this step removes.
  2. THE CONVENTION THE BEHAVIOUR RESTS ON — every per-session ledger any hook
     builds is named `<sid>[.<kind>].ids`. That is what makes a sixth ledger
     covered on the day it is written, and it is the assumption that would
     rot silently, because a ledger named outside it simply never clears and
     nothing says so.

And the negative: `<sid>.unreached` is not a ledger of held context but a
record that the instance could not be reached, and a glob that swept it away
would make a hook forget an outage it is meant to report (#2932).
"""
import json
import os
import re
import shutil
import subprocess
from pathlib import Path

import pytest

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

# Every ledger the hooks write today, and where. Named here so a failure READS
# as "this one survived", but never used to build the hook's own delete list —
# the convention tests below are what keep this roster honest as it grows.
LEDGERS = {
    "scribe-priorart": (".ids", ".rules.ids", ".opened.ids", ".sync.ids",
                        ".derive.ids"),
    "scribe-autoinject": (".ids",),
}


def _swept_dirs() -> set[str]:
    """The directories the SHIPPED hook sweeps, read from the hook itself.

    Read rather than restated: a test carrying its own copy of the roster would
    agree with itself forever, which is precisely the failure that let the
    auto-inject ledger sit in a directory nothing cleared.
    """
    line = re.search(r'SCRIBE_LEDGER_DIRS="([^"]*)"', DEFS.read_text())
    assert line, "SCRIBE_LEDGER_DIRS is gone; the clear has no roster"
    return set(line.group(1).split())


def _run_session_start(source: str, tmp: Path, extra: list[str] | None = None) -> Path:
    """Run the SessionStart hook for real, with the ledger directories filled.

    `extra` names further files to plant in `scribe-priorart` — used to put
    an evidence twin in front of the sweep.
    """
    for tool in ("bash",):
        if shutil.which(tool) is None:
            pytest.skip(f"hook runtime tool {tool!r} not installed")

    for dirname, suffixes in LEDGERS.items():
        state = tmp / dirname
        state.mkdir(parents=True, exist_ok=True)
        for suffix in suffixes:
            (state / f"s1{suffix}").write_text("42\t1789600000\n")
    # Not a ledger: an outage marker that must outlive the clear.
    (tmp / "scribe-priorart" / "s1.unreached").write_text("1\n")
    for name in extra or ():
        (tmp / "scribe-priorart" / name).write_text("42\t1789600000\n")

    env = {"PATH": os.environ["PATH"], "HOME": str(tmp), "TMPDIR": str(tmp)}
    out = subprocess.run(
        ["bash", str(SESSION_START)],
        input=json.dumps({"session_id": "s1", "source": source}),
        capture_output=True, text=True, env=env, timeout=60,
    )
    assert out.returncode == 0, out.stderr
    return tmp


@pytest.mark.parametrize("source", ["compact", "clear"])
def test_a_context_destroying_source_clears_every_ledger(source, tmp_path):
    """The step, stated as behaviour: all of them, in both directories."""
    root = _run_session_start(source, tmp_path)
    survived = [f"{d}/s1{s}" for d, suffixes in LEDGERS.items()
                for s in suffixes if (root / d / f"s1{s}").exists()]
    assert not survived, (
        f"{survived} survived a {source!r} that destroyed what they describe"
    )


@pytest.mark.parametrize("source", ["startup", "resume"])
def test_a_source_that_kept_the_context_keeps_the_ledgers(source, tmp_path):
    """The mirror error, and the more expensive one.

    `resume` genuinely restores the context, so the ledger still describes what
    the session holds; clearing there would re-surface every record after a
    restore that lost nothing. A blanket glob makes over-clearing cheap to
    write, which is exactly why this direction needs a test of its own.
    """
    root = _run_session_start(source, tmp_path)
    for dirname, suffixes in LEDGERS.items():
        for suffix in suffixes:
            assert (root / dirname / f"s1{suffix}").exists(), (
                f"{dirname}/s1{suffix} was cleared on {source!r}, which lost "
                "no context"
            )


def test_the_outage_marker_is_not_swept_with_them(tmp_path):
    """`.unreached` records that the instance was down, not what was surfaced.

    Different lifetime, different question. #2932's whole point is that "we
    checked and found nothing" and "we never managed to check" must stay
    distinguishable, and a clear that took this file out would quietly answer
    the second with the first.
    """
    root = _run_session_start("compact", tmp_path)
    assert (root / "scribe-priorart" / "s1.unreached").exists()


# ── the two assumptions the sweep rests on ─────────────────────────────────
#
# Both are checked against the hooks rather than trusted, because neither fails
# loudly: a ledger outside them simply never clears, on the arm whose author had
# no reason to know a convention existed.

def _dir_vars(text: str) -> dict[str, str]:
    """`var="${TMPDIR:-/tmp}/<name>"` → {var: name}, per script."""
    return dict(re.findall(
        r'(\w+)="\$\{TMPDIR:-/tmp\}/([A-Za-z0-9_-]+)"', text))


def _session_paths(text: str):
    """Every `$<var>/${safe_sid}<suffix>` a script composes."""
    return re.findall(r'\$(\w+)"?/\$\{safe_sid\}([A-Za-z0-9_.]*)', text)


def test_every_directory_holding_a_ledger_is_one_the_clear_visits():
    """The failure that shipped once already, in the same step that fixed it.

    The first cut swept `scribe-priorart` alone. Every ledger NAMED in the
    hooks was there, so it read as complete — and `scribe_autoinject.sh` keeps
    its note ledger in `scribe-autoinject`, which meant the arm that fires most
    was the only one still carrying the bug. Nothing said so: the clear ran,
    found nothing to remove, and exited 0.
    """
    swept = _swept_dirs()
    offenders = []
    for script in sorted(HOOKS.glob("*.sh")):
        text = script.read_text()
        dirs = _dir_vars(text)
        for var, suffix in _session_paths(text):
            if not suffix.endswith(".ids"):
                continue
            where = dirs.get(var)
            if where not in swept:
                offenders.append(f"{script.name}: ${var} -> {where or '?'}")

    assert not offenders, (
        "these ledgers live in directories the compaction clear never visits, "
        f"so they outlive the context they describe: {offenders}. Add the "
        "directory to SCRIBE_LEDGER_DIRS in scribe_defs.sh."
    )


def test_every_session_file_in_a_swept_directory_follows_the_convention():
    """The other half: the sweep matches `.ids`, so a ledger must be named it.

    Stated as its own test because the two assumptions fail differently — this
    one lets a ledger sit in the right directory and still never clear.
    """
    # Session files in a swept directory that are NOT per-session ledgers.
    NOT_LEDGERS = {".unreached"}

    swept = _swept_dirs()
    offenders = []
    for script in sorted(HOOKS.glob("*.sh")):
        text = script.read_text()
        dirs = _dir_vars(text)
        for var, suffix in _session_paths(text):
            if dirs.get(var) not in swept:
                continue          # a different directory, a different question
            if suffix in NOT_LEDGERS or suffix.endswith(".ids"):
                continue
            offenders.append(f"{script.name}: ${var}/${{safe_sid}}{suffix}")

    assert not offenders, (
        "these session files sit in a swept directory but clear on no "
        f"compaction, because the sweep matches `.ids`: {offenders}"
    )


def test_the_clear_is_derived_and_not_a_list_of_names():
    """The regression that would look like a fix.

    Appending an `rm -f` per ledger passes every behavioural test above while
    rebuilding the trap for the next one. The property worth keeping is that
    the hook names no ledger at all.
    """
    sh = SESSION_START.read_text()
    block = sh.split('case "$source" in')[1].split("esac")[0]
    assert "scribe_clear_session_ledgers" in block
    for suffix in {s for suffixes in LEDGERS.values() for s in suffixes}:
        assert suffix not in block, (
            f"the clear names {suffix} again — a list, not a convention"
        )


def test_the_evidence_twin_is_not_swept_with_them(tmp_path):
    """The second thing that survives, and for the `.unreached` reason (#4217).

    `scribe_rules_append` writes each ledger twice: `<sid>.<kind>.ids`, which
    says what this context HOLDS and must be forgotten here, and
    `<sid>.<kind>.keep.ids`, which says what HAPPENED and no compaction makes
    untrue. The session-end readout is built on the second, and it runs AT the
    compaction — so a sweep that took both would leave the readout reporting
    only the stretch since the last one, while reading as though it had
    reported the session.

    Measured before the split, on the instance this was built on: six sessions,
    208 `get_rule` calls, 3 surviving ledger entries.

    The test above asserts every exclusion ledger dies; this asserts its twin
    does not. Neither is complete alone, and the sweep is one glob away from
    either mistake.
    """
    root = _run_session_start("compact", tmp_path, extra=["s1.rules.keep.ids"])
    assert (root / "scribe-priorart" / "s1.rules.keep.ids").exists(), (
        "the evidence twin was swept with the exclusion ledgers — the readout "
        "at this seam now reports a session it cannot see"
    )