Excuses (5) every reflect Backlog item carries a handle

engine/skills/reflect/SKILL.md

@@ -151,6 +151,8 @@ Normally the step-5 worktree agent does this. If the parent must apply (worktree
 
 Short list, no preamble. Every applied, backlogged, or routed line carries the three parts from step 4 — what happened, fix, catch — plus the catch's backtest result (`fired` / `silent` / `unchecked`, or `catch: none — <reason>`):
 
+Before presenting the summary, run `python3 engine/skills/reflect/scripts/check_backlog_handles.py` on the draft summary; it must exit 0, and exit 2 is unchecked rather than a pass.
+
 - Edits applied: `<skill path>` — what changed, one line each.
 - New skills created: `<skill path>` — one line each (rare).
 - Backlogged: `<what to build>` — one line each, tagged with its tier and the evidence that motivated it.

engine/skills/reflect/scripts/check_backlog_handles.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import re
+import sys
+
+OK = 0
+FAIL = 1
+UNCHECKED = 2
+
+SECTION_ALIASES = {
+    "accepted": "accepted",
+    "backlog": "backlog",
+    "backlogged": "backlog",
+    "rejected": "rejected",
+    "route to automate me": "route-to-automate-me",
+    "route-to-automate-me": "route-to-automate-me",
+    "route to automate-me": "route-to-automate-me",
+    "route-to-automate me": "route-to-automate-me",
+    "routed to automate me": "route-to-automate-me",
+    "routed to automate-me": "route-to-automate-me",
+}
+
+BULLET_RE = re.compile(r"^\s*(?:[-*+]|\d+[.)])\s+(?P<item>\S.*)$")
+MARKDOWN_HEADING_RE = re.compile(r"^\s{0,3}#{1,6}\s+(?P<title>.*?)\s*#*\s*$")
+BOLD_HEADING_RE = re.compile(r"^\s{0,3}(?:\*\*|__)(?P<title>.*?)(?:\*\*|__):?\s*$")
+PLAIN_HEADING_RE = re.compile(r"^\s{0,3}(?P<title>[A-Za-z][A-Za-z` -]+):\s*$")
+WORKFLOW_RE = re.compile(r"\bwf-\d+(?:-\d+)+\b", re.IGNORECASE)
+TASK_RE = re.compile(
+    r"\b(?:task(?:\s+id)?\s*[:#]\s*|task\s+id\s+|task-)[A-Za-z0-9][A-Za-z0-9._/-]*\b",
+    re.IGNORECASE,
+)
+TRACKER_RE = re.compile(
+    r"(?:(?<!\w)#\d+\b|\b(?:issue|pr|pull\s+request)\s*#?\d+\b|/(?:issues|pull)/\d+\b)",
+    re.IGNORECASE,
+)
+DECLINED_RE = re.compile(r"\bdeclined by user\b", re.IGNORECASE)
+
+
+def normalized_heading(title: str) -> str:
+    title = title.strip().strip(":").strip()
+    title = title.replace("`", "")
+    title = re.sub(r"\s+", " ", title)
+    return title.lower()
+
+
+def section_for(line: str) -> str | None:
+    for pattern in (MARKDOWN_HEADING_RE, BOLD_HEADING_RE, PLAIN_HEADING_RE):
+        match = pattern.match(line)
+        if not match:
+            continue
+        return SECTION_ALIASES.get(normalized_heading(match.group("title")))
+    return None
+
+
+def has_handle(item: str) -> bool:
+    return any(
+        pattern.search(item)
+        for pattern in (WORKFLOW_RE, TASK_RE, TRACKER_RE, DECLINED_RE)
+    )
+
+
+def parse_backlog_items(text: str) -> tuple[str, list[str], str | None]:
+    current_section = None
+    saw_section = False
+    saw_backlog = False
+    backlog_text_without_bullets = False
+    items: list[list[str]] = []
+    for raw in text.splitlines():
+        section = section_for(raw)
+        if section:
+            current_section = section
+            saw_section = True
+            saw_backlog = saw_backlog or section == "backlog"
+            continue
+        if current_section != "backlog":
+            continue
+        stripped = raw.strip()
+        if not stripped:
+            continue
+        bullet = BULLET_RE.match(raw)
+        if bullet:
+            items.append([bullet.group("item").strip()])
+        elif items and raw[:1].isspace():
+            items[-1].append(stripped)
+        else:
+            backlog_text_without_bullets = True
+    if not saw_section:
+        return "unchecked", [], "no recognized reflect findings section"
+    if saw_backlog and backlog_text_without_bullets and not items:
+        return "unchecked", [], "Backlog section is not itemized"
+    return "ok", [" ".join(parts) for parts in items], None
+
+
+def read_input(path: str | None) -> str:
+    if path:
+        with open(path, encoding="utf-8") as handle:
+            return handle.read()
+    return sys.stdin.read()
+
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("summary", nargs="?")
+    args = ap.parse_args(argv)
+    try:
+        text = read_input(args.summary)
+    except OSError as exc:
+        print(f"unchecked  cannot read summary: {exc}")
+        return UNCHECKED
+    status, items, reason = parse_backlog_items(text)
+    if status == "unchecked":
+        print(f"unchecked  cannot parse reflect summary: {reason}")
+        return UNCHECKED
+    offenders = [item for item in items if not has_handle(item)]
+    if offenders:
+        print(f"fail     {len(offenders)} Backlog item(s) missing a task, issue, PR, or declined-by-user handle")
+        for item in offenders:
+            print(f"- {item}")
+        return FAIL
+    print("ok       every Backlog item has a durable handle")
+    return OK
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

engine/skills/reflect/scripts/tests/test_check_backlog_handles.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import textwrap
+import unittest
+
+SCRIPTS = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+SCRIPT = os.path.join(SCRIPTS, "check_backlog_handles.py")
+
+
+def run_summary(text: str, *args: str) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        [sys.executable, SCRIPT, *args],
+        input=textwrap.dedent(text).lstrip(),
+        capture_output=True,
+        text=True,
+    )
+
+
+class TestBacklogHandleCheck(unittest.TestCase):
+    def test_backlog_item_without_handle_fails_and_names_item(self):
+        res = run_summary(
+            """
+            ## Accepted
+            - What happened: one covered thing.
+
+            ## Backlog
+            - Build a land-stack no-op fix so queueing an empty stack exits cleanly.
+
+            ## Rejected
+            - Not relevant.
+            """
+        )
+        self.assertEqual(res.returncode, 1)
+        self.assertIn("land-stack no-op fix", res.stdout)
+
+    def test_backlog_item_with_invoker_workflow_id_passes(self):
+        res = run_summary(
+            """
+            ## Backlog
+            - Build a land-stack no-op fix. Invoker workflow wf-1789281139235-17 owns it.
+            """
+        )
+        self.assertEqual(res.returncode, 0)
+
+    def test_summary_with_only_accepted_items_passes(self):
+        res = run_summary(
+            """
+            ## Accepted
+            - What happened: the session skipped proof.
+              Fix: tighten the verifier.
+              Catch: test_check_verifier.py fired.
+            """
+        )
+        self.assertEqual(res.returncode, 0)
+
+    def test_route_and_rejected_items_without_handles_pass(self):
+        res = run_summary(
+            """
+            ## Route-to-automate-me
+            - Preserve the user's preferred review order.
+
+            ## Rejected
+            - One-off timeout in a closed environment.
+            """
+        )
+        self.assertEqual(res.returncode, 0)
+
+    def test_unparseable_input_exits_unchecked(self):
+        res = run_summary("not a reflect summary at all\n")
+        self.assertEqual(res.returncode, 2)
+        self.assertIn("unchecked", res.stdout)
+
+    def test_declined_by_user_is_an_explicit_handle(self):
+        res = run_summary(
+            """
+            ## Backlog
+            - Build a worker for optional cleanup, declined by user.
+            """
+        )
+        self.assertEqual(res.returncode, 0)
+
+    def test_file_argument_is_supported(self):
+        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8") as handle:
+            handle.write("## Backlog\n- Fix the checker in PR #541.\n")
+            handle.flush()
+            res = subprocess.run(
+                [sys.executable, SCRIPT, handle.name],
+                capture_output=True,
+                text=True,
+            )
+        self.assertEqual(res.returncode, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()