"""The plugin half of the rule↔moment signal (milestone 458 step 7), 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(): """Rule 33 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)