FabledGrimoire/tests/test_rule_moment_signal_hooks.py

"""The plugin half of the rule↔moment signal, run for real.

`scribe_moment.sh` writes every call to an acts ledger; `scribe_record_opened.sh`
sends the last few minutes of it with the rule a session opened; the server
answers with a proposal line when one is due. These drive both scripts through
bash with a stand-in `curl` on PATH that records what it was sent, so what is
pinned is what the scripts actually do — the window, the body's shape, the
reduction of a large event, the line coming back out — not what their source
looks like. `scribe_session_end.sh` removing the session's files is pinned the
same way.
"""
from __future__ import annotations

import inspect
import json
import os
import subprocess
import time
from pathlib import Path

import pytest

from scribe.routes import plugin as plugin_routes
from scribe.services import rule_moment_judgments as judgments_svc
from tests.helpers import need_tools

ROOT = Path(__file__).resolve().parents[1]
HOOKS = ROOT / "plugin" / "hooks"

FAKE_CURL = """#!/usr/bin/env bash
data=""
for a in "$@"; do
  if [ "$a" = "@-" ]; then data=$(cat); fi
done
printf '%s\\n' "$*" >> "$FAKE_CURL_LOG"
if [ -n "$data" ]; then printf '%s' "$data" > "$FAKE_CURL_BODY"; fi
if [ -n "${FAKE_CURL_REPLY:-}" ]; then printf '%s' "$FAKE_CURL_REPLY"; else printf '{}'; fi
"""


@pytest.fixture
def sandbox(tmp_path):
    need_tools("bash", "awk", "paste", "tail")
    bin_dir = tmp_path / "bin"
    bin_dir.mkdir()
    curl = bin_dir / "curl"
    curl.write_text(FAKE_CURL)
    curl.chmod(0o755)
    env = {
        "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}",
        "HOME": str(tmp_path), "TMPDIR": str(tmp_path),
        "SCRIBE_URL": "http://scribe.invalid", "SCRIBE_TOKEN": "t",
        "FAKE_CURL_LOG": str(tmp_path / "curl.log"),
        "FAKE_CURL_BODY": str(tmp_path / "curl.body"),
        "CLAUDE_PROJECT_DIR": str(tmp_path),
    }
    return tmp_path, env


def _run(script: str, event: dict, env: dict, **extra) -> subprocess.CompletedProcess:
    out = subprocess.run(
        ["bash", str(HOOKS / script)], input=json.dumps(event),
        capture_output=True, text=True, env={**env, **extra}, timeout=30,
    )
    assert out.returncode == 0, out.stderr
    return out


def _acts(tmp: Path, sid: str = "s1") -> list[tuple[int, dict]]:
    path = tmp / "scribe-moment" / f"{sid}.acts"
    rows = []
    for line in path.read_text().splitlines():
        stamp, _, act = line.partition("\t")
        rows.append((int(stamp), json.loads(act)))
    return rows


def test_every_call_is_written_to_the_acts_ledger_scribes_own_included(sandbox):
    tmp, env = sandbox
    push = {"session_id": "s1", "tool_name": "Bash",
            "tool_input": {"command": "git push origin dev"}}
    close = {"session_id": "s1", "tool_name": "mcp__plugin_scribe_scribe__update_task",
             "tool_input": {"task_id": 4, "status": "done"}}
    _run("scribe_moment.sh", push, env, FAKE_CURL_REPLY='{"tools":[]}')
    _run("scribe_moment.sh", close, env)
    acts = _acts(tmp)
    assert [a for _, a in acts] == [push, close]
    assert all(abs(stamp - time.time()) < 60 for stamp, _ in acts)


def test_the_open_itself_is_not_an_act(sandbox):
    tmp, env = sandbox
    _run("scribe_moment.sh", {"session_id": "s1", "tool_name": "Bash",
                              "tool_input": {"command": "ls"}}, env,
         FAKE_CURL_REPLY='{"tools":[]}')
    _run("scribe_moment.sh", {"session_id": "s1", "tool_name": "mcp__scribe__get_rule",
                              "tool_input": {"rule_id": 11}}, env)
    assert [a["tool_name"] for _, a in _acts(tmp)] == ["Bash"]


def test_a_large_event_is_kept_as_its_tool_name(sandbox):
    tmp, env = sandbox
    big = {"session_id": "s1", "tool_name": "Write",
           "tool_input": {"file_path": "x", "content": "y" * 40000}}
    _run("scribe_moment.sh", big, env, FAKE_CURL_REPLY='{"tools":[]}')
    [(_, act)] = _acts(tmp)
    assert act == {"tool_name": "Write", "tool_input": {}}


def test_an_open_sends_the_window_and_prints_the_answer(sandbox):
    tmp, env = sandbox
    ledger = tmp / "scribe-moment"
    ledger.mkdir()
    now = int(time.time())
    old = {"tool_name": "Bash", "tool_input": {"command": "make old"}}
    recent = {"tool_name": "Bash", "tool_input": {"command": "git push"}}
    (ledger / "s1.acts").write_text(
        f"{now - 1000}\t{json.dumps(old)}\n{now - 5}\t{json.dumps(recent)}\n"
    )
    line = "> Rule #11 has been opened just after `work.deliver` in 3 distinct sessions"
    out = _run("scribe_record_opened.sh",
               {"session_id": "s1", "tool_name": "mcp__scribe__get_rule",
                "tool_input": {"rule_id": 11}}, env,
               FAKE_CURL_REPLY=json.dumps({"moments": ["work.deliver"], "context": line}))

    body = json.loads((tmp / "curl.body").read_text())
    assert body == {"rule_id": 11, "session_id": "s1", "acts": [recent]}
    assert "/api/plugin/rule-opened" in (tmp / "curl.log").read_text()
    printed = json.loads(out.stdout)["hookSpecificOutput"]
    assert printed["hookEventName"] == "PostToolUse"
    assert printed["additionalContext"] == line
    # The open is still recorded in its own ledger, as before.
    assert (tmp / "scribe-priorart" / "s1.opened.ids").read_text().startswith("11\t")


def test_an_open_with_nothing_in_the_window_sends_nothing(sandbox):
    tmp, env = sandbox
    ledger = tmp / "scribe-moment"
    ledger.mkdir()
    (ledger / "s1.acts").write_text(
        f"{int(time.time()) - 1000}\t{json.dumps({'tool_name': 'Bash', 'tool_input': {}})}\n"
    )
    out = _run("scribe_record_opened.sh",
               {"session_id": "s1", "tool_name": "mcp__scribe__get_rule",
                "tool_input": {"rule_id": 11}}, env)
    assert out.stdout == ""
    assert not (tmp / "curl.log").exists()


def test_session_end_removes_the_sessions_moment_files_and_no_others(sandbox):
    tmp, env = sandbox
    ledger = tmp / "scribe-moment"
    ledger.mkdir()
    for name in ("s1.acts", "s1.tools", "s2.acts"):
        (ledger / name).write_text("x")
    _run("scribe_session_end.sh", {"session_id": "s1", "reason": "clear"}, env)
    assert sorted(p.name for p in ledger.iterdir()) == ["s2.acts"]


def test_the_hook_and_the_route_agree_on_the_body():
    """A contract across the shell/Python seam: a renamed field fails silently."""
    src = (HOOKS / "scribe_record_opened.sh").read_text()
    route = inspect.getsource(plugin_routes.rule_opened)
    assert "/api/plugin/rule-opened" in src
    for field in ("rule_id", "session_id", "acts"):
        assert f'"{field}"' in src and f'"{field}"' in route
    assert "'.context'" in src
    assert '"context"' in inspect.getsource(judgments_svc.opened_after)