feat(409): a Stop hook checks that a reply closing a task has the completion sections (#4014)

Everything else Scribe gives an agent arrives before the reply is written. A
Stop hook is the one moment the finished reply exists, so it is the last
chance to fix a report the operator can't read, and the only place adherence
to the shape can be measured.

- plugin/hooks/scribe_report_check.sh (Stop): deterministic, no model call.
  1. Did this turn close a task? That means an update_task/create_task call
     with status "done" since the turn's prompt, whose tool_result is not an
     error. Otherwise it stays silent, which covers most turns (one grep).
  2. Does the reply that ends the turn say where the work sits (a record by
     id and title, or step N of M), what needs the operator, and what comes
     next? Matched on those words, not on exact headings.
  3. If sections are missing, it blocks once. With stop_hook_active set, a
     rewrite is recorded (passed_after_rewrite / missing_after_rewrite) and
     never blocked again. A block loop started by another plugin (no marker
     from this hook) is left alone.
- Measured: every checked reply is reported to GET /api/plugin/report-check
  (passed / blocked / after rewrite). Turns that close nothing are not
  reported; they would cost a request per turn and add nothing to the rate.
  Outcomes go to app_logs as category "plugin", action "report_check".
- It blocks only when the block was recorded, and only in the server's words.
  The endpoint returns the block reason, so the hook carries timing and
  transport only (PACKAGING.md), and an unconfigured or unreachable instance
  never stops a session.
- The transcript format is read from real transcripts and marked in the hook
  as observed rather than documented. The Stop contract (transcript_path,
  stop_hook_active, decision/reason, no matcher, SubagentStop separate) was
  checked against the Claude Code hooks docs. A prompt-type hook was not
  needed: the deterministic check passed a real completion report from this
  session and blocked a stripped one.
- A pipefail trap was caught while exercising the hook: `tail | grep -q`
  reports failure exactly when grep matches, because tail dies of SIGPIPE.
  The prefilter reads through process substitution; the section checks use
  here-strings.
- Tests: an end-to-end hook suite over synthetic transcripts and the shared
  HTTP sink (silence, pass, server-worded block, rewrite recorded, foreign
  loop, errored write, earlier turn, unwritten reply, no recorded check, bare
  id), and service tests for the reason wording and the outcome record. Smoke
  event added to check_plugin; README and PACKAGING list the hook and
  endpoint. Plugin version minted.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

src/scribe/routes/plugin.py

@@ -14,6 +14,7 @@ from scribe.auth import admin_required, get_current_user_id, login_required
 from scribe.config import Config
 from scribe.services import plugin_context as plugin_ctx_svc
 from scribe.services import repo_bindings as repo_bindings_svc
+from scribe.services import report_check as report_check_svc
 from scribe.services.settings import get_admin_setting, set_setting
 
 plugin_bp = Blueprint("plugin", __name__, url_prefix="/api/plugin")
@@ -257,6 +258,45 @@ def _parse_shapes(raw: str) -> list[tuple[str, str]]:
     return out
 
 
+@plugin_bp.get("/report-check")
+@login_required
+async def report_check():
+    """Record what a Stop hook found in a reply that closed a task (milestone 409 step 5).
+
+    The hook decides which completion sections the reply lacks — a local check
+    of text it can read — and reports the outcome here. For `blocked` the
+    response carries the `reason` to send the agent back with: the words are
+    the server's, so every client's hook says the same thing (plugin/PACKAGING.md).
+    A hook blocks only on a `reason` it received, which means only on a block
+    that was recorded.
+
+    A GET for the reason every plugin endpoint is one: a read-scoped key must
+    be enough to run the plugin, and this records telemetry the way /retrieve
+    records a retrieval log.
+
+    Query:
+        outcome (str)   — passed | blocked | passed_after_rewrite |
+                          missing_after_rewrite. Anything else is a 400.
+        missing (opt)   — comma-separated sections the reply lacked:
+                          "where it sits", "needs you", "next".
+        task_ids (opt)  — comma-separated ids of the tasks the turn closed.
+        repo (opt)      — working repo remote, resolved like the other arms.
+    """
+    outcome = (request.args.get("outcome") or "").strip()
+    if outcome not in report_check_svc.OUTCOMES:
+        return jsonify({"error": f"outcome must be one of {list(report_check_svc.OUTCOMES)}"}), 400
+    missing = [m for m in (request.args.get("missing") or "").split(",") if m.strip()]
+    task_ids = _int_list(request.args.get("task_ids"))[:20]
+    project_id, _repo, _unbound = await _project_scope()
+    await report_check_svc.record_report_check(
+        g.user.id, outcome, missing=missing, task_ids=task_ids, project_id=project_id or None,
+    )
+    body: dict = {"status": "ok"}
+    if outcome == "blocked":
+        body["reason"] = report_check_svc.block_reason(missing)
+    return jsonify(body)
+
+
 @plugin_bp.get("/processes")
 @login_required
 async def process_manifest():

src/scribe/services/report_check.py

@@ -0,0 +1,80 @@
+"""The report-shape check: what the plugin's Stop hook found, and what it says.
+
+WHY THIS EXISTS (milestone 409 step 5)
+
+Everything that helps an agent write a readable completion report arrives
+BEFORE the reply is written. A client's Stop hook is the one moment the
+finished reply exists, so it checks that a reply closing a task carries the
+completion sections (where the work sits, what needs the operator, what comes
+next), and reports what it found here. Two jobs live on this side:
+
+  - RECORDING the outcome, so the rate of `blocked` among checked replies is a
+    number milestone 409's last step can read rather than an impression.
+  - OWNING THE WORDS the agent is sent back with. A hook carries timing and
+    transport only (plugin/PACKAGING.md); guidance text comes from the server,
+    so a second client's hook gets the same instruction by calling the same
+    endpoint, and the wording changes in one place.
+
+app_logs rather than a table of its own: one small event with a JSON detail is
+what that table holds, it already has retention and an admin viewer, and
+nothing here needs a join. If the numbers earn a readout, that is the moment to
+decide whether they earn a table.
+"""
+from __future__ import annotations
+
+import json
+
+from scribe.models import async_session
+from scribe.models.app_log import AppLog
+
+OUTCOMES = ("passed", "blocked", "passed_after_rewrite", "missing_after_rewrite")
+
+# The sections a hook may name as missing, in the order the reason lists them.
+# Anything else a client sends is dropped rather than echoed into an
+# instruction the agent will follow.
+SECTIONS = ("where it sits", "needs you", "next")
+
+
+def known_sections(missing: list[str]) -> list[str]:
+    wanted = {m.strip().lower() for m in missing}
+    return [s for s in SECTIONS if s in wanted]
+
+
+def block_reason(missing: list[str]) -> str:
+    """What the agent is told when its completion report is sent back.
+
+    Names what is missing and points at the reporting-back skill for the shape
+    rather than restating it — the skill owns the shape (decision #4027).
+    """
+    listed = ", ".join(known_sections(missing)) or "the completion sections"
+    return (
+        f"This turn closed a Scribe task, and the reply that ends it is missing: {listed}. "
+        "The operator reads this reply to find out where the work stands. Rewrite it as a "
+        "completion report (the reporting-back skill has the shape): where it sits — the task "
+        "or milestone by id and title, from `placement` — what now works, what needs them "
+        "(or \"nothing\"), and what comes next."
+    )
+
+
+async def record_report_check(
+    user_id: int | None,
+    outcome: str,
+    *,
+    missing: list[str] | None = None,
+    task_ids: list[int] | None = None,
+    project_id: int | None = None,
+) -> None:
+    if outcome not in OUTCOMES:
+        raise ValueError(f"unknown report-check outcome {outcome!r}")
+    details: dict = {"outcome": outcome, "missing": known_sections(missing or []),
+                     "task_ids": list(task_ids or [])}
+    if project_id:
+        details["project_id"] = project_id
+    async with async_session() as session:
+        session.add(AppLog(
+            category="plugin",
+            user_id=user_id,
+            action="report_check",
+            details=json.dumps(details),
+        ))
+        await session.commit()