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) -> Path:
    """Run the SessionStart hook for real, with the ledger directories filled."""
    for tool in ("bash", "jq"):
        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")

    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"
        )