FabledScribe/tests/test_report_check_hook.py

"""The Stop hook that checks a task-closing reply for the completion sections
(milestone 409 step 5).

Runs the real shell against synthetic transcripts in the shape Claude Code
writes (one content block per JSONL line) and the shared HTTP sink. What it
pins: silence on every turn that closed nothing; a block only when the
instance recorded it, in the words the instance returned; one rewrite at most,
recorded; and no block from another plugin's loop or a failed task write.
"""
from __future__ import annotations

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

import pytest

from tests.helpers import http_sink

HOOK = Path(__file__).resolve().parents[1] / "plugin" / "hooks" / "scribe_report_check.sh"
TOOL = "mcp__plugin_scribe_scribe__update_task"
GOOD = ('**Where this sits:** milestone 12 "Move the backups offsite", step 3 of 5.\n'
        "**What now works:** the sync runs nightly.\n**Needs you:** nothing.\n**Next:** alerts.")
BAD = "All done, pushed it."
REASON = "SERVER REASON: rewrite as a completion report"


def _env(tmp_path, url="http://127.0.0.1:9"):
    for tool in ("jq", "curl", "bash"):
        if shutil.which(tool) is None:
            pytest.skip(f"hook runtime tool {tool!r} not installed")
    return {"PATH": os.environ["PATH"], "SCRIBE_URL": url, "SCRIBE_TOKEN": "t",
            "TMPDIR": str(tmp_path), "HOME": str(tmp_path)}


def _prompt(text="please finish it"):
    return {"type": "user", "message": {"role": "user", "content": text}}


def _tool_use(tid="toolu_1", status="done", name=TOOL, task_id=41):
    return {"type": "assistant", "message": {"content": [
        {"type": "tool_use", "id": tid, "name": name, "input": {"task_id": task_id, "status": status}}]}}


def _result(tid="toolu_1", is_error=False):
    return {"type": "user", "message": {"content": [
        {"type": "tool_result", "tool_use_id": tid, "is_error": is_error, "content": "{}"}]}}


def _text(text):
    return {"type": "assistant", "message": {"content": [{"type": "text", "text": text}]}}


def _transcript(tmp_path, lines):
    path = tmp_path / "t.jsonl"
    # Compact, like the file Claude Code writes ({"name":"…"}, no spaces).
    path.write_text("\n".join(json.dumps(line, separators=(",", ":")) for line in lines) + "\n")
    return path


def _run(env, transcript, active=False, session="s1"):
    out = subprocess.run(
        ["bash", str(HOOK)],
        input=json.dumps({"session_id": session, "transcript_path": str(transcript),
                          "cwd": str(transcript.parent), "hook_event_name": "Stop",
                          "stop_hook_active": active}),
        capture_output=True, text=True, env=env, timeout=30,
    )
    assert out.returncode == 0, out.stderr
    return out.stdout.strip()


def _closing_turn(reply):
    return [_prompt(), _text("On it."), _tool_use(), _result(), _text(reply)]


def test_a_turn_that_closed_nothing_is_silent_and_reports_nothing(tmp_path):
    with http_sink(b'{"status":"ok","reason":"x"}') as (port, seen):
        env = _env(tmp_path, f"http://127.0.0.1:{port}")
        t = _transcript(tmp_path, [_prompt(), _tool_use(status="in_progress"), _result(), _text(BAD)])
        assert _run(env, t) == ""
        assert seen == []


def test_a_complete_report_passes_silently_and_is_recorded(tmp_path):
    with http_sink(b'{"status":"ok"}') as (port, seen):
        env = _env(tmp_path, f"http://127.0.0.1:{port}")
        assert _run(env, _transcript(tmp_path, _closing_turn(GOOD))) == ""
        assert [q["outcome"] for q in seen] == [["passed"]]
        assert seen[0]["task_ids"] == ["41"]


def test_a_missing_section_blocks_once_in_the_servers_words_then_records_the_rewrite(tmp_path):
    reply = json.dumps({"status": "ok", "reason": REASON}).encode()
    with http_sink(reply) as (port, seen):
        env = _env(tmp_path, f"http://127.0.0.1:{port}")
        out = json.loads(_run(env, _transcript(tmp_path, _closing_turn(BAD))))
        assert out == {"decision": "block", "reason": REASON}
        assert seen[0]["outcome"] == ["blocked"]
        assert seen[0]["missing"] == ["where it sits,needs you,next"]

        # The rewrite: Claude Code sets stop_hook_active; the hook records and never blocks again.
        rewritten = _transcript(tmp_path, _closing_turn(BAD) + [_text(GOOD)])
        assert _run(env, rewritten, active=True) == ""
        assert seen[1]["outcome"] == ["passed_after_rewrite"]
        assert _run(env, rewritten, active=True) == ""
        assert len(seen) == 2


def test_a_rewrite_that_still_misses_is_recorded_and_not_blocked(tmp_path):
    reply = json.dumps({"status": "ok", "reason": REASON}).encode()
    with http_sink(reply) as (port, seen):
        env = _env(tmp_path, f"http://127.0.0.1:{port}")
        t = _transcript(tmp_path, _closing_turn(BAD))
        _run(env, t)
        assert _run(env, t, active=True) == ""
        assert [q["outcome"][0] for q in seen] == ["blocked", "missing_after_rewrite"]


def test_another_hooks_block_loop_is_left_alone(tmp_path):
    with http_sink(b'{"status":"ok","reason":"x"}') as (port, seen):
        env = _env(tmp_path, f"http://127.0.0.1:{port}")
        assert _run(env, _transcript(tmp_path, _closing_turn(BAD)), active=True) == ""
        assert seen == []


def test_a_task_write_that_failed_closed_nothing(tmp_path):
    with http_sink(b'{"status":"ok","reason":"x"}') as (port, seen):
        env = _env(tmp_path, f"http://127.0.0.1:{port}")
        t = _transcript(tmp_path, [_prompt(), _tool_use(), _result(is_error=True), _text(BAD)])
        assert _run(env, t) == ""
        assert seen == []


def test_a_task_closed_in_an_earlier_turn_does_not_count(tmp_path):
    with http_sink(b'{"status":"ok","reason":"x"}') as (port, seen):
        env = _env(tmp_path, f"http://127.0.0.1:{port}")
        t = _transcript(tmp_path, _closing_turn(GOOD) + [_prompt("thanks, what else?"), _text(BAD)])
        assert _run(env, t) == ""
        assert seen == []


def test_a_reply_not_yet_written_is_not_judged(tmp_path):
    with http_sink(b'{"status":"ok","reason":"x"}') as (port, seen):
        env = _env(tmp_path, f"http://127.0.0.1:{port}")
        t = _transcript(tmp_path, [_prompt(), _tool_use(), _result()])
        assert _run(env, t) == ""
        assert seen == []


def test_no_block_without_a_recorded_check(tmp_path):
    t = _transcript(tmp_path, _closing_turn(BAD))
    # Unreachable instance.
    assert _run(_env(tmp_path), t) == ""
    # An instance that answered but returned no reason.
    with http_sink(b'{"status":"ok"}') as (port, seen):
        assert _run(_env(tmp_path, f"http://127.0.0.1:{port}"), t, session="s2") == ""
        assert seen[0]["outcome"] == ["blocked"]


def test_a_bare_id_does_not_count_as_placing_the_work(tmp_path):
    reply = json.dumps({"status": "ok", "reason": REASON}).encode()
    with http_sink(reply) as (port, seen):
        env = _env(tmp_path, f"http://127.0.0.1:{port}")
        bare = "Closed #41.\n**Needs you:** nothing.\n**Next:** #42."
        assert json.loads(_run(env, _transcript(tmp_path, _closing_turn(bare))))["decision"] == "block"
        assert seen[0]["missing"] == ["where it sits"]