hermes: finalize completed review verdicts in the goal judge

The local goal judge scored every worker report against "did the reviewed
implementation reach a shippable state". A read-only reviewer that returned a
completed BLOCK verdict with findings was therefore resumed turn after turn with
an instruction to repair code it was forbidden to touch (observed live on
t_dbdcd739), burning subscription capacity and risking an unbounded loop.

Completion is now judged against the action the card assigned:

* Cards declare their role explicitly with Hermes-Task-Role / Hermes-Expected-
  Output metadata. Pre-contract cards fall back to a narrow inference that needs
  a read-only scope, a requested verdict, no requested mutation deliverable, and
  a report that changed no files.
* Role resolution reads only the card itself. Prior attempts, parent results,
  cross-task history and comments appended to the worker context can no longer
  reassign the role.
* Review, audit and diagnostic cards finalize deterministically on a truthful
  SHIP or BLOCK verdict with evidence, and fail closed on a missing,
  unrecognized or self-contradictory verdict, on a BLOCK without findings, and
  on a verdict without evidence. Every rejection reason carries the read-only
  guard, so a resumed review is never told to edit the reviewed implementation.
* Implementation cards keep the fail-closed model judge unchanged, including the
  unfinished-work heuristic and judge-unavailable rejection.
* All judge reasons are bounded, single-line and secret-redacted before they
  reach Kanban metadata, comments and continuation prompts.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

services/hermes/scripts/cli_lane_goal.py

@@ -1,5 +1,24 @@
 #!/usr/bin/env python3
-"""Fail-closed completion checks for durable external Kanban workers."""
+"""Role-aware completion checks for durable external Kanban workers.
+
+Completion is judged against the action the card assigned, not against the
+health of whatever the card asked the worker to look at.
+
+* Implementation cards keep the fail-closed local goal judge: a report is
+  accepted only when the model agrees every explicit requirement finished.
+* Review, audit, and diagnostic cards deliver a *verdict*. A completed report
+  carrying a truthful SHIP or BLOCK verdict with evidence finalizes the card
+  even when the verdict says the reviewed artifact must not ship. Findings are
+  the deliverable, not a task blocker, so the card is never resumed with an
+  instruction to repair an implementation the reviewer may not touch.
+
+The task role comes from explicit card metadata (``Hermes-Task-Role: review``)
+whenever the card author supplies it. Cards written before that contract fall
+back to a deliberately narrow inference that requires a read-only scope, a
+requested verdict, no requested mutation deliverable, and a report that changed
+no files. Review completion is fail-closed on a missing, unrecognized, or
+self-contradictory verdict and on missing evidence.
+"""
 
 from __future__ import annotations
 
@@ -24,6 +43,30 @@ GOAL_JUDGE_TIMEOUT_SECONDS = max(
     min(180, int(os.environ.get("HERMES_GOAL_JUDGE_TIMEOUT_SECONDS", "120"))),
 )
 RESULT_STATUSES = frozenset({"completed", "blocked", "incomplete"})
+REVIEW_ROLE = "review"
+IMPLEMENTATION_ROLE = "implementation"
+TASK_ROLES = frozenset({REVIEW_ROLE, IMPLEMENTATION_ROLE})
+ROLE_ALIASES = {
+    "audit": REVIEW_ROLE,
+    "assessment": REVIEW_ROLE,
+    "diagnosis": REVIEW_ROLE,
+    "diagnostic": REVIEW_ROLE,
+    "inspection": REVIEW_ROLE,
+    "review": REVIEW_ROLE,
+    "reviewer": REVIEW_ROLE,
+    "verdict": REVIEW_ROLE,
+    "build": IMPLEMENTATION_ROLE,
+    "change": IMPLEMENTATION_ROLE,
+    "implement": IMPLEMENTATION_ROLE,
+    "implementation": IMPLEMENTATION_ROLE,
+    "repair": IMPLEMENTATION_ROLE,
+}
+JUDGE_REASON_LIMIT = 600
+MIN_REVIEW_SUMMARY_CHARS = 80
+READ_ONLY_GUARD = (
+    "the assigned action is a read-only review: report the verdict and evidence, "
+    "and do not modify the reviewed implementation"
+)
 GOAL_JUDGE_SCHEMA: dict[str, Any] = {
     "type": "object",
     "additionalProperties": False,
@@ -43,6 +86,60 @@ UNFINISHED_EVIDENCE = re.compile(
     r"\b(?:tests?|checks?|build|validation|verification|commit|push)\b)",
     re.IGNORECASE | re.DOTALL,
 )
+# The worker context appends prior attempts, parent results, cross-task history
+# and comments after the card the worker was actually given. Role resolution
+# reads only the card itself so an unrelated sentence quoted from an earlier run
+# cannot reassign the task role.
+HISTORY_SECTION = re.compile(
+    r"^##\s+(?:Prior attempts\b|Parent task results\b|Recent work by\b|Comment thread\b)",
+    re.IGNORECASE | re.MULTILINE,
+)
+# Explicit, machine-readable card metadata. This is the supported contract;
+# every pattern below it only keeps pre-contract cards working.
+ROLE_DIRECTIVE = re.compile(
+    r"^[\s>#*\-]*(?:hermes[\s_-]*)?task[\s_-]*role\s*[:=]\s*([A-Za-z_-]+)\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+OUTPUT_DIRECTIVE = re.compile(
+    r"^[\s>#*\-]*(?:hermes[\s_-]*)?expected[\s_-]*output\s*[:=]\s*([A-Za-z_-]+)\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+READ_ONLY_SCOPE = re.compile(
+    r"(?:\b(?:do|does|must|should|shall|may|will)\s+not\s+(?:\w+\s+){0,3}?"
+    r"(?:edit|modify|change|alter|patch|rewrite|implement|repair)\b"
+    r"|\bdon'?t\s+(?:\w+\s+){0,3}?(?:edit|modify|change|alter|patch)\b"
+    r"|\bwithout\s+(?:editing|modifying|changing|altering|patching)\b"
+    r"|\bno\s+(?:code|file|source|implementation)\s+(?:edits?|changes?|modifications?)\b"
+    r"|\bmake\s+no\s+(?:code|file|source)?\s*changes\b"
+    r"|\bread[\s-]?only\s+(?:scope|review|audit|analysis|assessment|inspection|diagnos\w+)\b)",
+    re.IGNORECASE,
+)
+VERDICT_DELIVERABLE = re.compile(
+    r"(?:\bship\s+or\s+block\b|\bblock\s+or\s+ship\b|\bship\s*/\s*block\b"
+    r"|\breturn\s+(?:a\s+|one\s+|the\s+)?(?:strict\s+)?verdict\b"
+    r"|\b(?:review|audit|release|diagnostic)\s+verdict\b|\bverdict\s*[:=])",
+    re.IGNORECASE,
+)
+MUTATION_DELIVERABLE = re.compile(
+    r"(?:\bpushed\s+(?:sha|commit|head|branch|revision)\b"
+    r"|\bpush\s+(?:the\s+|your\s+|one\s+)?(?:branch|commit|change\w*|fix\w*)\b"
+    r"|\bcommit\s+and\s+push\b"
+    r"|\bopen\s+(?:a|an|one|the)\s+(?:new\s+)?(?:draft\s+)?(?:pr\b|pull\s+request)"
+    r"|\bdraft\s+pr\s+only\b)",
+    re.IGNORECASE,
+)
+VERDICT_DIRECTIVE = re.compile(
+    r"\b(?:final\s+|strict\s+|overall\s+)?(?:verdict|recommendation|decision)\s*"
+    r"(?:is\s*)?[:=—-]\s*[*_`\s]*(ship|block)\b",
+    re.IGNORECASE,
+)
+VERDICT_TOKEN = re.compile(r"(?<![A-Za-z])(SHIP|BLOCK)(?![A-Za-z])")
+CONTROL_CHARACTERS = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
+SECRET_LIKE = re.compile(
+    r"(?:gh[pousr]_[A-Za-z0-9]{16,}|xox[abprs]-[A-Za-z0-9-]{10,}"
+    r"|sk-[A-Za-z0-9_-]{16,}|AKIA[0-9A-Z]{16}"
+    r"|eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,})"
+)
 
 
 def _bounded(value: str, limit: int) -> str:
@@ -55,17 +152,138 @@ def _bounded(value: str, limit: int) -> str:
     return f"{value[:head]}{marker}{value[-(remaining - head):]}"
 
 
-def unfinished_result_reason(result: dict[str, Any]) -> str | None:
+def sanitize_reason(value: Any, limit: int = JUDGE_REASON_LIMIT) -> str:
+    """Return one bounded, single-line, secret-free judge reason."""
+    text = SECRET_LIKE.sub("[redacted]", str(value or ""))
+    text = " ".join(CONTROL_CHARACTERS.sub(" ", text).split())
+    if not text:
+        return "local judge supplied no reason"
+    return text if len(text) <= limit else f"{text[: limit - 3].rstrip()}..."
+
+
+def _strings(value: Any) -> list[str]:
+    """Return the non-empty string entries of a reported evidence list."""
+    if not isinstance(value, list):
+        return []
+    return [str(item).strip() for item in value if str(item).strip()]
+
+
+def card_scope(objective: str) -> str:
+    """Return the card the worker was assigned, without appended run history."""
+    text = str(objective or "")
+    match = HISTORY_SECTION.search(text)
+    return text[: match.start()] if match else text
+
+
+def _declared_role(objective: str) -> str | None:
+    """Return the role a card declared through explicit worker metadata."""
+    for pattern in (ROLE_DIRECTIVE, OUTPUT_DIRECTIVE):
+        roles = {
+            ROLE_ALIASES[value]
+            for value in (
+                match.group(1).strip().lower() for match in pattern.finditer(objective)
+            )
+            if value in ROLE_ALIASES
+        }
+        if len(roles) == 1:
+            return roles.pop()
+        if roles:
+            # A card that declares two different roles has no usable contract.
+            # Hold it to the stricter implementation regime rather than letting
+            # the more permissive review path win an ambiguous declaration.
+            return IMPLEMENTATION_ROLE
+    return None
+
+
+def task_role(objective: str, result: dict[str, Any] | None = None) -> tuple[str, str]:
+    """Resolve the assigned task role and how that resolution was reached."""
+    text = card_scope(objective)
+    declared = _declared_role(text)
+    if declared is not None:
+        return declared, "directive"
+    changed = _strings((result or {}).get("changed_files"))
+    if (
+        not changed
+        and READ_ONLY_SCOPE.search(text)
+        and VERDICT_DELIVERABLE.search(text)
+        and not MUTATION_DELIVERABLE.search(text)
+    ):
+        return REVIEW_ROLE, "inferred"
+    return IMPLEMENTATION_ROLE, "default"
+
+
+def review_verdict(result: dict[str, Any]) -> tuple[str | None, str | None]:
+    """Return the declared SHIP/BLOCK verdict, or why it cannot be trusted."""
+    declared: set[str] = set()
+    field = result.get("verdict")
+    if field is not None:
+        value = str(field).strip().upper()
+        if value not in {"SHIP", "BLOCK"}:
+            return None, f"review verdict field is not SHIP or BLOCK: {value[:40]!r}"
+        declared.add(value)
+    summary = str(result.get("summary") or "")
+    declared.update(match.upper() for match in VERDICT_DIRECTIVE.findall(summary))
+    if len(declared) > 1:
+        return None, "review report declares both SHIP and BLOCK verdicts"
+    if declared:
+        return declared.pop(), None
+    tokens = {match.upper() for match in VERDICT_TOKEN.findall(summary)}
+    if len(tokens) > 1:
+        return None, "review summary names both SHIP and BLOCK without a verdict line"
+    if tokens:
+        return tokens.pop(), None
+    return None, "review report declares no explicit SHIP or BLOCK verdict"
+
+
+def review_completion_problem(result: dict[str, Any]) -> str | None:
+    """Fail closed on a review report that does not carry a usable verdict."""
+    verdict, problem = review_verdict(result)
+    if problem or verdict is None:
+        return problem or "review report declares no explicit SHIP or BLOCK verdict"
+    summary = str(result.get("summary") or "").strip()
+    if len(summary) < MIN_REVIEW_SUMMARY_CHARS:
+        return f"{verdict} verdict carries no reviewable rationale in its summary"
+    findings = _strings(result.get("findings"))
+    if verdict == "BLOCK" and not findings:
+        return "BLOCK verdict reports no findings to justify it"
+    if not findings and not _strings(result.get("tests_run")) and not _strings(
+        result.get("artifacts")
+    ):
+        return f"{verdict} verdict reports no findings, tests, or artifacts as evidence"
+    return None
+
+
+def _reported_role(result: dict[str, Any]) -> str:
+    """Classify a report shape when no objective is available to the caller."""
+    if _strings(result.get("changed_files")):
+        return IMPLEMENTATION_ROLE
+    if result.get("verdict") is not None or VERDICT_DIRECTIVE.search(
+        str(result.get("summary") or "")
+    ):
+        return REVIEW_ROLE
+    return IMPLEMENTATION_ROLE
+
+
+def unfinished_result_reason(
+    result: dict[str, Any],
+    *,
+    role: str | None = None,
+) -> str | None:
     """Reject self-contradictory completion reports without model inference."""
     status = str(result.get("status") or "")
     summary = str(result.get("summary") or "").strip()
     blockers = result.get("blockers")
     if status == "incomplete":
-        return summary or "worker explicitly reported incomplete work"
+        return sanitize_reason(summary or "worker explicitly reported incomplete work")
     if status != "completed":
         return None
     if isinstance(blockers, list) and any(str(item).strip() for item in blockers):
         return "worker reported blockers while claiming completion"
+    if (role or _reported_role(result)) == REVIEW_ROLE:
+        # A review's prose describes the reviewed artifact, so the unfinished
+        # work heuristic below would read the artifact's state as the review's
+        # own. The verdict contract is the deterministic gate instead.
+        return review_completion_problem(result)
     tests = result.get("tests_run")
     evidence = [summary]
     if isinstance(tests, list):
@@ -76,18 +294,14 @@ def unfinished_result_reason(result: dict[str, Any]) -> str | None:
     return None
 
 
-def judge_goal_completion(
+def _completion_claimed(result: dict[str, Any]) -> bool:
-    objective: str,
+    """Report whether the worker actually claimed a finished task."""
-    result: dict[str, Any],
+    return str(result.get("status") or "") == "completed"
-    *,
+
-    url: str = GOAL_JUDGE_URL,
+
-    open_request: Callable[..., Any] = urllib.request.urlopen,
+def _judge_payload(objective: str, result: dict[str, Any], role: str) -> dict[str, Any]:
-) -> tuple[bool, str]:
+    """Build the fail-closed local judge request for an implementation card."""
-    """Use the local model as a fail-closed judge for explicit goal cards."""
+    return {
-    deterministic_reason = unfinished_result_reason(result)
-    if deterministic_reason:
-        return False, deterministic_reason
-    payload = {
         "model": GOAL_JUDGE_MODEL,
         "messages": [
             {
@@ -98,13 +312,16 @@ def judge_goal_completion(
                     "Return complete only when the report gives concrete evidence that every "
                     "explicit requirement finished. Return continue when any work, test, command, "
                     "commit, push, review, or verification is pending, in progress, omitted, or "
-                    "inconclusive. Do not trust the report status field."
+                    "inconclusive. Defects the worker reports about some other artifact are "
+                    "findings, not unfinished work on this task. Do not trust the report status "
+                    "field."
                 ),
             },
             {
                 "role": "user",
                 "content": json.dumps(
                     {
+                        "task_role": role,
                         "objective": _bounded(objective, 6000),
                         "worker_report": _bounded(
                             json.dumps(result, default=str), 4000
@@ -126,9 +343,40 @@ def judge_goal_completion(
             },
         },
     }
+
+
+def judge_goal_completion(
+    objective: str,
+    result: dict[str, Any],
+    *,
+    url: str = GOAL_JUDGE_URL,
+    open_request: Callable[..., Any] = urllib.request.urlopen,
+) -> tuple[bool, str]:
+    """Judge one worker report against the action its card actually assigned."""
+    role, source = task_role(objective, result)
+    deterministic_reason = unfinished_result_reason(result, role=role)
+    if deterministic_reason is None and not _completion_claimed(result):
+        deterministic_reason = (
+            f"worker reported status {str(result.get('status') or 'unknown')!r} "
+            "rather than a finished task"
+        )
+    if deterministic_reason:
+        if role == REVIEW_ROLE:
+            return False, sanitize_reason(f"{READ_ONLY_GUARD}; {deterministic_reason}")
+        return False, sanitize_reason(deterministic_reason)
+    if role == REVIEW_ROLE:
+        # The requested deliverable is present and self-consistent. A verdict
+        # that the reviewed artifact must not ship is a finished review, so the
+        # card finalizes instead of resuming a worker that may not repair it.
+        verdict, _ = review_verdict(result)
+        findings = len(_strings(result.get("findings")))
+        return True, sanitize_reason(
+            f"review deliverable complete: {verdict} verdict with {findings} finding(s); "
+            f"task role {role} resolved by {source}"
+        )
     request = urllib.request.Request(
         url,
-        data=json.dumps(payload).encode("utf-8"),
+        data=json.dumps(_judge_payload(objective, result, role)).encode("utf-8"),
         headers={"Content-Type": "application/json"},
         method="POST",
     )
@@ -139,7 +387,7 @@ def judge_goal_completion(
         verdict = json.loads(content)
         if verdict.get("verdict") not in {"complete", "continue"}:
             raise ValueError("local judge returned an invalid verdict")
-        reason = str(verdict.get("reason") or "local judge supplied no reason")
+        reason = sanitize_reason(verdict.get("reason"))
         return verdict.get("verdict") == "complete", reason
     except (
         AttributeError,
@@ -151,4 +399,6 @@ def judge_goal_completion(
         json.JSONDecodeError,
         urllib.error.URLError,
     ) as error:
-        return False, f"local completion judge unavailable: {type(error).__name__}: {error}"
+        return False, sanitize_reason(
+            f"local completion judge unavailable: {type(error).__name__}: {error}"
+        )

testing/tests/test_hermes_cli_review_goal.py

@@ -0,0 +1,461 @@
+"""Task-role and verdict-contract tests for the Hermes goal judge."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+
+SCRIPT = Path(__file__).parents[2] / "services/hermes/scripts/cli_lane_goal.py"
+SPEC = importlib.util.spec_from_file_location("cli_lane_review_goal_test", SCRIPT)
+assert SPEC and SPEC.loader
+goal = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = goal
+SPEC.loader.exec_module(goal)
+
+
+REVIEW_CARD = """# Kanban task t_review: Independent read-only review of the worker pool
+
+Assignee: cli-claude-xhigh
+
+## Body
+Perform a fresh independent read-only release review of the pool implementation.
+Do not edit files, comment, commit, push, merge, publish, or deploy.
+Return strict SHIP or BLOCK with exact file/line and reproducible inputs.
+"""
+IMPLEMENTATION_CARD = """# Kanban task t_impl: Repair the pool blockers
+
+Assignee: cli-auto
+
+## Body
+Repair every reported blocker, run the focused suites, and open a draft PR.
+Return exact pushed SHA and evidence.
+"""
+POLLUTED_CARD = REVIEW_CARD + """
+## Recent work by @cli-claude-xhigh
+- t_prior: Implemented and pushed the pool. PR #18 is open at the exact pushed SHA.
+
+## Comment thread
+- shepherd: commit and push the repairs before returning.
+"""
+
+
+def _review(**overrides):
+    value = {
+        "status": "completed",
+        "summary": (
+            "Independent read-only review of PR #18 at head 2000252. Verdict: BLOCK. "
+            "Five reproducible P0 defects fire on the coordinator's first tick and the "
+            "review worktree was left pristine."
+        ),
+        "changed_files": [],
+        "tests_run": ["pytest testing/tests/test_pool.py: 82 passed"],
+        "artifacts": ["/opt/data/workspace/evidence/t_review/RUN.md"],
+        "findings": ["P0-1 BLOCKER - coordinator.py:248 compares int to str run ids."],
+        "blockers": [],
+    }
+    value.update(overrides)
+    return value
+
+
+def _implementation(**overrides):
+    value = {
+        "status": "completed",
+        "summary": "All acceptance criteria passed and the branch was pushed.",
+        "changed_files": ["src/example.py"],
+        "tests_run": ["pytest -q: 12 passed"],
+        "artifacts": [],
+        "findings": [],
+        "blockers": [],
+    }
+    value.update(overrides)
+    return value
+
+
+def _no_judge(*_args, **_kwargs):
+    raise AssertionError("the model judge must not be consulted for a review card")
+
+
+def test_completed_block_review_finalizes_without_consulting_the_model():
+    accepted, reason = goal.judge_goal_completion(
+        REVIEW_CARD,
+        _review(),
+        open_request=_no_judge,
+    )
+
+    assert accepted is True
+    assert "BLOCK verdict with 1 finding(s)" in reason
+    assert "resolved by inferred" in reason
+
+
+def test_completed_ship_review_finalizes_on_its_own_evidence():
+    accepted, reason = goal.judge_goal_completion(
+        REVIEW_CARD,
+        _review(
+            summary=(
+                "Independent read-only review of PR #18 at head 2000252. Verdict: SHIP. "
+                "Every prioritized boundary was exercised and no defect survived."
+            ),
+            findings=[],
+        ),
+        open_request=_no_judge,
+    )
+
+    assert accepted is True
+    assert "SHIP verdict with 0 finding(s)" in reason
+
+
+@pytest.mark.parametrize(
+    ("role_line", "expected"),
+    [
+        ("Hermes-Task-Role: review", goal.REVIEW_ROLE),
+        ("Task-Role: audit", goal.REVIEW_ROLE),
+        ("- Hermes Task Role = diagnostic", goal.REVIEW_ROLE),
+        ("Hermes-Expected-Output: verdict", goal.REVIEW_ROLE),
+        ("Hermes-Task-Role: implementation", goal.IMPLEMENTATION_ROLE),
+    ],
+)
+def test_explicit_card_metadata_decides_the_task_role(role_line, expected):
+    card = IMPLEMENTATION_CARD.replace("## Body\n", f"## Body\n{role_line}\n")
+
+    assert goal.task_role(card, _review())[0] == expected
+    assert goal.task_role(card, _review())[1] == "directive"
+
+
+def test_explicit_review_metadata_survives_a_changed_file_report():
+    card = REVIEW_CARD.replace("## Body\n", "## Body\nHermes-Task-Role: review\n")
+
+    accepted, reason = goal.judge_goal_completion(
+        card,
+        _review(changed_files=["reports/review.md"]),
+        open_request=_no_judge,
+    )
+
+    assert accepted is True
+    assert "resolved by directive" in reason
+
+
+def test_appended_run_history_cannot_reassign_the_card_role():
+    assert goal.task_role(POLLUTED_CARD, _review()) == (goal.REVIEW_ROLE, "inferred")
+    assert "Recent work by" not in goal.card_scope(POLLUTED_CARD)
+    assert "Do not edit files" in goal.card_scope(POLLUTED_CARD)
+
+
+def test_implementation_card_never_infers_a_review_role():
+    assert goal.task_role(IMPLEMENTATION_CARD, _review()) == (
+        goal.IMPLEMENTATION_ROLE,
+        "default",
+    )
+
+
+@pytest.mark.parametrize(
+    ("result", "expected"),
+    [
+        (_review(summary="Verdict: BLOCK."), "carries no reviewable rationale"),
+        (_review(findings=[]), "BLOCK verdict reports no findings"),
+        (
+            _review(summary=_review()["summary"].replace("Verdict: BLOCK.", "")),
+            "declares no explicit SHIP or BLOCK verdict",
+        ),
+        (
+            _review(verdict="SHIP"),
+            "declares both SHIP and BLOCK verdicts",
+        ),
+        (
+            _review(verdict="maybe"),
+            "review verdict field is not SHIP or BLOCK",
+        ),
+        (
+            _review(
+                summary=(
+                    "The reviewed change is fine overall. SHIP is defensible but BLOCK "
+                    "is also arguable given the unresolved capacity question."
+                ),
+            ),
+            "names both SHIP and BLOCK without a verdict line",
+        ),
+        (
+            _review(
+                summary=(
+                    "Independent read-only review of PR #18. Verdict: SHIP. Every "
+                    "prioritized boundary was exercised and nothing survived."
+                ),
+                findings=[],
+                tests_run=[],
+                artifacts=[],
+            ),
+            "no findings, tests, or artifacts as evidence",
+        ),
+    ],
+)
+def test_malformed_review_completion_fails_closed(result, expected):
+    accepted, reason = goal.judge_goal_completion(
+        REVIEW_CARD,
+        result,
+        open_request=_no_judge,
+    )
+
+    assert accepted is False
+    assert expected in reason
+    assert reason.startswith(goal.READ_ONLY_GUARD)
+
+
+def test_rejected_review_never_demands_an_implementation_edit():
+    accepted, reason = goal.judge_goal_completion(
+        REVIEW_CARD,
+        _review(findings=[]),
+        open_request=_no_judge,
+    )
+
+    assert accepted is False
+    assert "do not modify the reviewed implementation" in reason
+
+
+def test_incomplete_review_turn_is_resumed_under_the_read_only_guard():
+    accepted, reason = goal.judge_goal_completion(
+        REVIEW_CARD,
+        _review(status="incomplete", summary="Only two of six boundaries were read."),
+        open_request=_no_judge,
+    )
+
+    assert accepted is False
+    assert reason.startswith(goal.READ_ONLY_GUARD)
+    assert "Only two of six boundaries were read." in reason
+
+
+def test_blocked_review_reports_its_own_obstacle_not_a_verdict():
+    result = _review(
+        status="blocked",
+        summary="The pull request head could not be resolved from any SCM mirror.",
+        blockers=["Gitea API is unreachable from the review worktree."],
+    )
+
+    assert goal.unfinished_result_reason(result, role=goal.REVIEW_ROLE) is None
+    assert goal.unfinished_result_reason(result) is None
+    accepted, reason = goal.judge_goal_completion(
+        REVIEW_CARD, result, open_request=_no_judge
+    )
+    assert accepted is False
+    assert "rather than a finished task" in reason
+
+
+def test_conflicting_role_directives_hold_the_card_to_implementation():
+    card = REVIEW_CARD.replace(
+        "## Body\n",
+        "## Body\nHermes-Task-Role: review\nHermes-Task-Role: implementation\n",
+    )
+
+    assert goal.task_role(card, _review()) == (goal.IMPLEMENTATION_ROLE, "directive")
+
+
+def test_incomplete_review_summary_is_bounded_before_it_leaves_the_judge():
+    accepted, reason = goal.judge_goal_completion(
+        REVIEW_CARD,
+        _review(status="incomplete", summary="pending\n" + "y" * 4000),
+        open_request=_no_judge,
+    )
+
+    assert accepted is False
+    assert reason.startswith(goal.READ_ONLY_GUARD)
+    assert len(reason) <= goal.JUDGE_REASON_LIMIT
+
+
+def test_review_claiming_completion_with_blockers_is_rejected():
+    accepted, reason = goal.judge_goal_completion(
+        REVIEW_CARD,
+        _review(blockers=["the diff could not be resolved"]),
+        open_request=_no_judge,
+    )
+
+    assert accepted is False
+    assert "worker reported blockers while claiming completion" in reason
+
+
+def test_review_prose_about_the_reviewed_artifact_is_not_unfinished_work():
+    result = _review(
+        summary=(
+            "Independent read-only review. Verdict: BLOCK. The reviewed branch's CI "
+            "build is still running and its migration remains pending, which is why "
+            "the change is unfit to ship."
+        ),
+        tests_run=["upstream pipeline: build is still running on the reviewed head"],
+    )
+
+    assert goal.unfinished_result_reason(result, role=goal.REVIEW_ROLE) is None
+    accepted, _reason = goal.judge_goal_completion(
+        REVIEW_CARD,
+        result,
+        open_request=_no_judge,
+    )
+    assert accepted is True
+
+
+def test_implementation_prose_about_unfinished_work_is_still_rejected():
+    result = _implementation(summary="The broad rerun remains active before the push.")
+
+    assert goal.unfinished_result_reason(result)
+    assert goal.unfinished_result_reason(result, role=goal.IMPLEMENTATION_ROLE)
+
+
+class JudgeResponse:
+    def __init__(self, verdict: str, reason: str):
+        self.body = json.dumps(
+            {"choices": [{"message": {"content": json.dumps(
+                {"verdict": verdict, "reason": reason}
+            )}}]}
+        ).encode()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return False
+
+    def read(self):
+        return self.body
+
+
+def test_implementation_card_keeps_the_fail_closed_model_judge():
+    observed = {}
+
+    def request(req, timeout):
+        observed["payload"] = json.loads(req.data)
+        observed["timeout"] = timeout
+        return JudgeResponse("continue", "the remote head was never verified")
+
+    accepted, reason = goal.judge_goal_completion(
+        IMPLEMENTATION_CARD,
+        _implementation(),
+        open_request=request,
+    )
+
+    assert accepted is False
+    assert reason == "the remote head was never verified"
+    role = json.loads(observed["payload"]["messages"][1]["content"])["task_role"]
+    assert role == goal.IMPLEMENTATION_ROLE
+    assert observed["timeout"] == 120
+
+
+def test_implementation_completion_still_finalizes_on_an_accepting_judge():
+    accepted, reason = goal.judge_goal_completion(
+        IMPLEMENTATION_CARD,
+        _implementation(),
+        open_request=lambda *_a, **_k: JudgeResponse("complete", "every criterion has evidence"),
+    )
+
+    assert accepted is True
+    assert reason == "every criterion has evidence"
+
+
+def test_ambiguous_card_without_a_verdict_report_stays_on_the_model_judge():
+    ambiguous = """# Kanban task t_mixed: Review and repair the pool
+
+## Body
+Review the pool, then repair what you find. Do not modify unrelated files.
+Return a verdict and open a draft PR with the repairs.
+"""
+
+    accepted, _reason = goal.judge_goal_completion(
+        ambiguous,
+        _review(),
+        open_request=lambda *_a, **_k: JudgeResponse("continue", "the repairs are missing"),
+    )
+
+    assert goal.task_role(ambiguous, _review())[0] == goal.IMPLEMENTATION_ROLE
+    assert accepted is False
+
+
+def test_judge_reason_is_bounded_sanitized_and_secret_free():
+    leaky = "line one\r\n\x07 token ghp_" + "A" * 24 + " " + "x" * 900
+
+    accepted, reason = goal.judge_goal_completion(
+        IMPLEMENTATION_CARD,
+        _implementation(),
+        open_request=lambda *_a, **_k: JudgeResponse("continue", leaky),
+    )
+
+    assert accepted is False
+    assert len(reason) <= goal.JUDGE_REASON_LIMIT
+    assert "ghp_" not in reason
+    assert "[redacted]" in reason
+    assert "\n" not in reason and "\r" not in reason and "\x07" not in reason
+
+
+def test_unavailable_judge_still_fails_closed_for_implementation_cards():
+    accepted, reason = goal.judge_goal_completion(
+        IMPLEMENTATION_CARD,
+        _implementation(),
+        open_request=lambda *_a, **_k: JudgeResponse("unknown", "bad"),
+    )
+
+    assert accepted is False
+    assert "local completion judge unavailable" in reason
+
+
+def test_bare_uppercase_verdict_token_is_accepted_as_a_last_resort():
+    result = _review(
+        summary=(
+            "Independent read-only review of PR #18 at head 2000252 concludes BLOCK "
+            "because five reproducible P0 defects fire on the first coordinator tick."
+        ),
+    )
+
+    assert goal.review_verdict(result) == ("BLOCK", None)
+    accepted, _reason = goal.judge_goal_completion(
+        REVIEW_CARD, result, open_request=_no_judge
+    )
+    assert accepted is True
+
+
+def test_unknown_role_directive_falls_through_to_the_next_contract_surface():
+    card = IMPLEMENTATION_CARD.replace(
+        "## Body\n",
+        "## Body\nHermes-Task-Role: gardener\nHermes-Expected-Output: verdict\n",
+    )
+
+    assert goal.task_role(card, _review()) == (goal.REVIEW_ROLE, "directive")
+
+
+def test_unrecognized_directives_leave_the_role_to_inference():
+    card = REVIEW_CARD.replace("## Body\n", "## Body\nHermes-Task-Role: gardener\n")
+
+    assert goal.task_role(card, _review()) == (goal.REVIEW_ROLE, "inferred")
+
+
+def test_malformed_evidence_fields_are_read_without_raising():
+    assert goal._strings("not-a-list") == []
+    result = _implementation(tests_run="pytest -q", summary="Everything shipped.")
+
+    assert goal.unfinished_result_reason(result) is None
+
+
+def test_implementation_rejection_reason_carries_no_read_only_guard():
+    accepted, reason = goal.judge_goal_completion(
+        IMPLEMENTATION_CARD,
+        _implementation(blockers=["the remote was unreachable"]),
+        open_request=_no_judge,
+    )
+
+    assert accepted is False
+    assert reason == "worker reported blockers while claiming completion"
+
+
+def test_oversized_objectives_and_empty_reasons_stay_bounded():
+    compacted = goal._bounded("a" * 9000, 6000)
+
+    assert len(compacted) == 6000
+    assert "[goal evidence compacted]" in compacted
+    assert goal.sanitize_reason("   ") == "local judge supplied no reason"
+
+
+def test_judging_never_mutates_the_reported_result():
+    result = _review()
+    snapshot = json.dumps(result, sort_keys=True)
+
+    goal.judge_goal_completion(REVIEW_CARD, result, open_request=_no_judge)
+
+    assert json.dumps(result, sort_keys=True) == snapshot

testing/tests/test_hermes_cli_review_lane.py

@@ -0,0 +1,427 @@
+"""Lane-level goal-loop behaviour for review and implementation cards."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+from contextlib import nullcontext
+from pathlib import Path
+from types import SimpleNamespace
+
+
+SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts"
+sys.path.insert(0, str(SCRIPTS))
+
+
+def _load(name: str):
+    spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py")
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+lanes = _load("cli_lane_runner")
+# The single-claim loop lives in the runner today and in ``cli_lane_execution``
+# once the lane modules are split. Patch whichever module owns it so this file
+# describes lane behaviour rather than one file layout.
+execution = sys.modules.get("cli_lane_execution", lanes)
+goal = sys.modules["cli_lane_goal"]
+
+REVIEW_CARD = """# Kanban task t_review: Independent read-only review of the worker pool
+
+## Body
+Perform a fresh independent read-only release review of the pool implementation.
+Do not edit files, comment, commit, push, merge, publish, or deploy.
+Return strict SHIP or BLOCK with exact file/line and reproducible inputs.
+"""
+IMPLEMENTATION_CARD = """# Kanban task t_impl: Repair the pool blockers
+
+## Body
+Repair every reported blocker, run the focused suites, and open a draft PR.
+Return exact pushed SHA and evidence.
+"""
+BLOCK_REVIEW = {
+    "status": "completed",
+    "summary": (
+        "Independent read-only review of PR #18 at head 2000252. Verdict: BLOCK. "
+        "Five reproducible P0 defects fire on the coordinator's first tick and the "
+        "review worktree was left pristine."
+    ),
+    "changed_files": [],
+    "tests_run": ["pytest testing/tests/test_pool.py: 82 passed"],
+    "artifacts": [],
+    "findings": ["P0-1 BLOCKER - coordinator.py:248 compares int to str run ids."],
+    "blockers": [],
+}
+
+
+class _Connection:
+    def close(self) -> None:
+        return None
+
+
+class _Lane:
+    """One instrumented direct-CLI-lane claim over a fake Kanban board."""
+
+    def __init__(self, tmp_path: Path, monkeypatch, *, card: str, task, reports):
+        self.calls: list[tuple[str, dict]] = []
+        self.comments: list[str] = []
+        self.prompts: list[str] = []
+        self.routes: list[tuple[str, dict]] = []
+        self.heartbeats: list[tuple[str, object]] = []
+        self.reports = list(reports)
+        self.task = task
+        self.state_root = tmp_path / "cli-lanes"
+        self.workspace = tmp_path
+        self.db = SimpleNamespace(
+            scoped_current_board=lambda _board: nullcontext(),
+            connect=lambda board=None, **_kwargs: _Connection(),
+            get_task=lambda _conn, _task_id: self.task,
+            worker_log_path=lambda _task_id, board: tmp_path / "worker.log",
+            _resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_card"),
+            set_branch_name=lambda *_args: None,
+            set_workspace_path=lambda *_args: None,
+            build_worker_context=lambda *_args: card,
+            heartbeat_worker=lambda _conn, _task_id, *, note, expected_run_id: (
+                self.heartbeats.append((note, expected_run_id)) or True
+            ),
+            add_comment=lambda _conn, _task_id, _author, body: self.comments.append(body),
+            complete_task=self._complete_task,
+            block_task=self._block_task,
+            reclaim_task=self._reclaim_task,
+        )
+        monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=self.db))
+        # The lane derives durable journal identity from ``STATE_ROOT``, so
+        # relocate the root itself rather than stubbing ``state_path``.
+        for module in list(sys.modules.values()):
+            if getattr(module, "__name__", "").startswith("cli_lane") and hasattr(
+                module, "STATE_ROOT"
+            ):
+                monkeypatch.setattr(module, "STATE_ROOT", self.state_root)
+        monkeypatch.setattr(execution, "fresh_unavailable_provider", lambda: None)
+        monkeypatch.setattr(execution, "select_route", self._select_route)
+        monkeypatch.setattr(execution, "run_provider", self._run_provider)
+
+    def _complete_task(
+        self,
+        _conn,
+        task_id,
+        *,
+        result,
+        summary,
+        metadata,
+        expected_run_id=None,
+        replay_ended_run_id=None,
+    ):
+        self.calls.append(
+            (
+                "complete",
+                {
+                    "task_id": task_id,
+                    "result": result,
+                    "summary": summary,
+                    "metadata": metadata,
+                    "expected_run_id": expected_run_id or replay_ended_run_id,
+                },
+            )
+        )
+        self.task = SimpleNamespace(
+            **{
+                **vars(self.task),
+                "status": "done",
+                "result": result,
+                "completed_run_id": expected_run_id or replay_ended_run_id,
+            }
+        )
+        return True
+
+    def _block_task(self, _conn, task_id, *, reason, kind, expected_run_id=None):
+        self.calls.append(
+            (
+                "block",
+                {
+                    "task_id": task_id,
+                    "reason": reason,
+                    "kind": kind,
+                    "expected_run_id": expected_run_id,
+                },
+            )
+        )
+        return True
+
+    def _reclaim_task(self, _conn, _task_id, *, reason=None, expected_run_id=None):
+        return True
+
+    @property
+    def state_file(self) -> Path:
+        return self.state_root / "titan-iac/t_card.json"
+
+    @staticmethod
+    def _route(provider: str, effort: str = "xhigh"):
+        return lanes.Route(
+            provider,
+            f"{provider}-model",
+            effort,
+            f"{provider}-{effort}",
+            "switchyard",
+            "routed",
+            1,
+            (),
+        )
+
+    def _select_route(self, _prompt, assignee, **kwargs):
+        self.routes.append((assignee, kwargs))
+        if assignee.startswith("cli-codex"):
+            return self._route("codex")
+        return self._route("claude")
+
+    def _run_provider(self, *args, **_kwargs):
+        self.prompts.append(args[1])
+        args[6]("working")
+        return self.reports.pop(0)
+
+    @property
+    def terminal(self) -> tuple[str, dict]:
+        assert len(self.calls) == 1, self.calls
+        return self.calls[0]
+
+    @property
+    def state(self) -> dict:
+        return json.loads(self.state_file.read_text(encoding="utf-8"))
+
+    def rejections(self) -> list[str]:
+        return [item for item in self.comments if "Goal completion rejected" in item]
+
+
+def _task(**overrides):
+    value = {
+        "id": "t_card",
+        "status": "running",
+        "current_run_id": 23,
+        "completed_run_id": None,
+        "result": None,
+        "assignee": "cli-claude-xhigh",
+        "max_runtime_seconds": 600,
+        "goal_mode": True,
+        "goal_max_turns": 8,
+    }
+    value.update(overrides)
+    return SimpleNamespace(**value)
+
+
+def _result(**overrides):
+    value = dict(BLOCK_REVIEW)
+    value.update(overrides)
+    return lanes.ProcessResult(0, "review turn", value, False)
+
+
+def _no_judge(*_args, **_kwargs):
+    raise AssertionError("the model judge must not be consulted for a review card")
+
+
+def test_completed_block_review_finalizes_on_its_first_turn(tmp_path, monkeypatch):
+    monkeypatch.setattr(goal.urllib.request, "urlopen", _no_judge)
+    lane = _Lane(
+        tmp_path,
+        monkeypatch,
+        card=REVIEW_CARD,
+        task=_task(),
+        reports=[_result()],
+    )
+
+    execution.execute_claim("titan-iac", "t_card")
+
+    action, kwargs = lane.terminal
+    assert action == "complete"
+    assert kwargs["expected_run_id"] == 23
+    assert json.loads(kwargs["result"])["findings"] == BLOCK_REVIEW["findings"]
+    assert kwargs["metadata"]["goal_turn"] == 1
+    assert "BLOCK verdict with 1 finding(s)" in kwargs["metadata"]["goal_judge_reason"]
+    assert lane.rejections() == []
+    assert len(lane.prompts) == 1
+    assert lane.reports == []
+
+
+def test_review_verdict_reaches_kanban_without_hidden_mutation(tmp_path, monkeypatch):
+    monkeypatch.setattr(goal.urllib.request, "urlopen", _no_judge)
+    lane = _Lane(
+        tmp_path,
+        monkeypatch,
+        card=REVIEW_CARD,
+        task=_task(),
+        reports=[_result()],
+    )
+
+    execution.execute_claim("titan-iac", "t_card")
+
+    _action, kwargs = lane.terminal
+    stored = json.loads(kwargs["result"])
+    assert stored["status"] == "completed"
+    assert stored["summary"] == BLOCK_REVIEW["summary"]
+    assert stored["blockers"] == []
+    assert kwargs["summary"] == BLOCK_REVIEW["summary"]
+    assert kwargs["metadata"]["findings"] == BLOCK_REVIEW["findings"]
+    assert len(kwargs["metadata"]["goal_judge_reason"]) <= goal.JUDGE_REASON_LIMIT
+
+
+def test_reviewer_provider_fallback_still_finalizes_one_verdict(tmp_path, monkeypatch):
+    monkeypatch.setattr(goal.urllib.request, "urlopen", _no_judge)
+    lane = _Lane(
+        tmp_path,
+        monkeypatch,
+        card=REVIEW_CARD,
+        task=_task(assignee="cli-auto"),
+        reports=[
+            lanes.ProcessResult(1, "usage limit reached", None, True),
+            _result(),
+        ],
+    )
+
+    execution.execute_claim("titan-iac", "t_card")
+
+    action, kwargs = lane.terminal
+    assert action == "complete"
+    assert kwargs["metadata"]["provider"] == "codex"
+    assert kwargs["metadata"]["goal_turn"] == 1
+    assert any("Provider fallback: claude -> codex" in item for item in lane.comments)
+    assert lane.rejections() == []
+    assert len(lane.prompts) == 2
+
+
+def test_review_restart_resumes_the_next_turn_and_completes_once(tmp_path, monkeypatch):
+    monkeypatch.setattr(goal.urllib.request, "urlopen", _no_judge)
+    lane = _Lane(
+        tmp_path,
+        monkeypatch,
+        card=REVIEW_CARD,
+        task=_task(),
+        reports=[_result()],
+    )
+    lane.state_file.parent.mkdir(parents=True, exist_ok=True)
+    lane.state_file.write_text(
+        json.dumps({"goal_turn": 3, "current_route": {"provider": "codex"}}),
+        encoding="utf-8",
+    )
+
+    execution.execute_claim("titan-iac", "t_card")
+
+    action, kwargs = lane.terminal
+    assert action == "complete"
+    assert kwargs["metadata"]["goal_turn"] == 4
+    assert lane.state["goal_turn"] == 4
+    assert any("Restart-time provider change" in item for item in lane.comments)
+
+
+def test_malformed_review_is_resumed_without_demanding_an_edit(tmp_path, monkeypatch):
+    monkeypatch.setattr(goal.urllib.request, "urlopen", _no_judge)
+    lane = _Lane(
+        tmp_path,
+        monkeypatch,
+        card=REVIEW_CARD,
+        task=_task(goal_max_turns=2),
+        reports=[
+            _result(summary="Review done.", findings=[]),
+            _result(),
+        ],
+    )
+
+    execution.execute_claim("titan-iac", "t_card")
+
+    action, kwargs = lane.terminal
+    assert action == "complete"
+    assert kwargs["metadata"]["goal_turn"] == 2
+    assert len(lane.rejections()) == 1
+    assert goal.READ_ONLY_GUARD in lane.rejections()[0]
+    assert "do not modify the reviewed implementation" in lane.prompts[1]
+    assert "Goal-loop continuation" in lane.prompts[1]
+
+
+def test_exhausted_review_budget_blocks_with_a_read_only_reason(tmp_path, monkeypatch):
+    monkeypatch.setattr(goal.urllib.request, "urlopen", _no_judge)
+    lane = _Lane(
+        tmp_path,
+        monkeypatch,
+        card=REVIEW_CARD,
+        task=_task(goal_max_turns=1),
+        reports=[_result(summary="Review done.", findings=[])],
+    )
+
+    execution.execute_claim("titan-iac", "t_card")
+
+    action, kwargs = lane.terminal
+    assert action == "block"
+    assert kwargs["expected_run_id"] == 23
+    assert goal.READ_ONLY_GUARD in kwargs["reason"]
+
+
+def test_implementation_goal_loop_is_preserved(tmp_path, monkeypatch):
+    verdicts = iter([(False, "the remote head was never verified"), (True, "all criteria hold")])
+    contexts: list[str] = []
+
+    def judge(objective, *_args, **_kwargs):
+        contexts.append(objective)
+        return next(verdicts)
+
+    monkeypatch.setattr(goal, "judge_goal_completion", judge)
+    implementation = {
+        "status": "completed",
+        "summary": "Focused tests passed and the branch was pushed.",
+        "changed_files": ["src/a.py"],
+        "tests_run": ["pytest focused: passed"],
+        "artifacts": [],
+        "findings": [],
+        "blockers": [],
+    }
+    lane = _Lane(
+        tmp_path,
+        monkeypatch,
+        card=IMPLEMENTATION_CARD,
+        task=_task(assignee="cli-auto", goal_max_turns=3),
+        reports=[
+            lanes.ProcessResult(0, "turn one", dict(implementation), False),
+            lanes.ProcessResult(0, "turn two", dict(implementation), False),
+        ],
+    )
+
+    execution.execute_claim("titan-iac", "t_card")
+
+    action, kwargs = lane.terminal
+    assert action == "complete"
+    assert kwargs["metadata"]["goal_turn"] == 2
+    assert len(lane.rejections()) == 1
+    assert "the remote head was never verified" in lane.rejections()[0]
+    assert "prior rejected reports" in contexts[1]
+
+
+def test_incomplete_implementation_never_finalizes(tmp_path, monkeypatch):
+    lane = _Lane(
+        tmp_path,
+        monkeypatch,
+        card=IMPLEMENTATION_CARD,
+        task=_task(assignee="cli-auto", goal_max_turns=1),
+        reports=[
+            lanes.ProcessResult(
+                0,
+                "turn one",
+                {
+                    "status": "incomplete",
+                    "summary": "The focused suite is still running.",
+                    "changed_files": ["src/a.py"],
+                    "tests_run": [],
+                    "artifacts": [],
+                    "findings": [],
+                    "blockers": [],
+                },
+                False,
+            )
+        ],
+    )
+
+    execution.execute_claim("titan-iac", "t_card")
+
+    action, kwargs = lane.terminal
+    assert action == "block"
+    assert "The focused suite is still running." in kwargs["reason"]