FabledScribe/tests/test_session_context_ledger.py

"""The SessionStart hook clears the rule ledger exactly when context dies (#3749).

WHY THIS EXISTS

The prior-art and tool-rule hooks record every rule id they have named in
`<state>/<sid>.rules.ids` and hand it back as `exclude_rule_ids`, so a rule is
surfaced once per session and then goes quiet. That is correct while the
session still holds what it was told.

A compaction breaks that assumption in the worst available way: it summarizes
the earlier injections out of context and does not touch the filesystem. The
rule ends up absent from context AND still excluded — unreachable for the rest
of the session. The compaction banner tells the model to re-pull its
*always-on* rules, but a rule an arm surfaced is conditional and is not in that
set, so it has no other way back. The rules most likely to be in that state are
the ones that fire most often.

WHAT THIS PINS

Not "the ledger is cleared" — that would pass against a hook which deletes it
on every source, and deleting on `resume` is its own defect: the context was
genuinely restored there, so re-surfacing every rule is the mirror error.

What is pinned is the DISCRIMINATION. The whole source table is asserted in one
statement, so a blanket delete (all False) and a no-op (all True) both fail,
and neither can be made to pass by editing one case.

Runs the real shell against a temp TMPDIR, like the after-write hook's tests.
Deliberately with no SCRIBE_URL/SCRIBE_TOKEN in the environment: clearing the
ledger is local, keyless and networkless, and must still happen on an instance
that is unreachable or unconfigured.
"""
from __future__ import annotations

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

import pytest

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


def _env(tmp_path):
    """Near-namesake of test_after_write_hook's `_env`, and deliberately not it:
    that one needs git and curl and SUPPLIES credentials, because the behaviour
    it tests is a network round-trip. This one must prove the opposite — that
    the clear happens with no credentials and no network at all — so sharing a
    helper would mean testing this case in an environment that cannot show it.
    """
    for tool in ("jq", "bash"):
        if shutil.which(tool) is None:
            pytest.skip(f"hook runtime tool {tool!r} not installed")
    # No SCRIBE_URL / SCRIBE_TOKEN on purpose — see the module docstring.
    return {"PATH": os.environ["PATH"], "TMPDIR": str(tmp_path),
            "HOME": str(tmp_path)}


def _ledger(tmp_path, sid: str, name: str = "rules.ids") -> Path:
    d = tmp_path / "scribe-priorart"
    d.mkdir(exist_ok=True)
    f = d / f"{sid}.{name}"
    f.write_text("156\n168\n")
    return f


def _fire(source: str, sid: str, env) -> None:
    subprocess.run(
        ["bash", str(HOOK)],
        input=json.dumps({"source": source, "session_id": sid}),
        capture_output=True, text=True, env=env, timeout=30,
    )


def test_the_rules_ledger_survives_exactly_when_the_context_does(tmp_path):
    """The whole source table, in one assertion, so it cannot be half-satisfied.

    `startup` is listed even though it is a no-op against a session id that has
    never been seen: it is asserted here so that a future change which starts
    clearing indiscriminately fails on a case somebody would otherwise call
    harmless.
    """
    env = _env(tmp_path)
    survived = {}
    for source in ("compact", "clear", "resume", "startup"):
        sid = f"sess-{source}"
        ledger = _ledger(tmp_path, sid)
        _fire(source, sid, env)
        survived[source] = ledger.exists()

    assert survived == {
        "compact": False,
        "clear": False,
        "resume": True,
        "startup": True,
    }, (
        f"got {survived}. A rule surfaced before a compaction is summarized "
        f"out of context while its id stays on the exclusion ledger, so it "
        f"becomes unreachable for the rest of the session — that is what the "
        f"compact/clear cases prevent. The resume case is the other half: the "
        f"context came back intact there, and re-surfacing every rule after a "
        f"restore that lost nothing is the same defect from the other side. "
        f"All-False means something is deleting unconditionally; all-True "
        f"means the clear never runs."
    )


def test_the_note_ledgers_are_cleared_too_now_that_a_repeat_is_shown(tmp_path):
    """The decision this test used to pin, reversed — and why it could be.

    It read: the note arms were deliberately left out of #3749, whether a
    surfaced NOTE should come back after a compaction is a different question,
    and a glob over `<sid>.*` would decide it by accident. That was right when
    it was written, and the reason was the note arms' behaviour rather than the
    note arms' nature: a note repeat was WITHHELD, so clearing their ledger
    meant re-injecting whole menu lines the session had already been given.
    Against that, leaving them was the cheaper mistake.

    #4101 removed the premise. A note repeat is now rendered again with a
    `seen` marker instead of dropped from the search, so the ledger no longer
    decides whether a record is reachable — only whether its line says the
    session has met it before. Across a compaction that marker becomes a false
    statement: it tells a freshly-summarised session it has already seen a
    record that is nowhere in its context. Keeping the ledger buys nothing and
    asserts something untrue.

    Kept as a named test rather than deleted, because the roster it guards
    moved to tests/test_session_ledger_clear.py and what is worth keeping HERE
    is the record that the answer changed, with the reason attached — a
    deleted test takes the old decision's reasoning with it.
    """
    env = _env(tmp_path)
    sid = "sess-scope"
    rules = _ledger(tmp_path, sid, "rules.ids")
    notes = _ledger(tmp_path, sid, "ids")
    sync = _ledger(tmp_path, sid, "sync.ids")
    derive = _ledger(tmp_path, sid, "derive.ids")

    _fire("compact", sid, env)

    assert not rules.exists(), "the rule ledger should have been cleared"
    assert not (notes.exists() or sync.exists() or derive.exists()), (
        "a note ledger outlived the context it describes, so its records are "
        "marked `seen` to a session that can no longer see them"
    )


def test_a_compact_without_a_session_id_is_survivable(tmp_path):
    """Defensive, because this hook's contract is fail-open.

    An event with no `session_id` names no ledger. The hook must not error, and
    must not fall back to a wildcard — clearing every session's ledger on the
    machine because this one event was malformed is the worst available
    reading of "best effort".
    """
    env = _env(tmp_path)
    other = _ledger(tmp_path, "someone-elses-session")

    proc = subprocess.run(
        ["bash", str(HOOK)],
        input=json.dumps({"source": "compact"}),
        capture_output=True, text=True, env=env, timeout=30,
    )

    assert proc.returncode == 0, proc.stderr
    assert other.exists(), (
        "an event with no session id cleared a ledger belonging to a different "
        "session"
    )