feat(auto-agents): deterministic worker-side improvements

Four changes that move judgment off the LLM and into the dispatcher,
motivated by the live PR #30 escalation pilot (2026-05-12): Tier 0
spent 16 min and gave up on flaky-looking unrelated tests; Tier 2
spent 88 min discovering the PR was already correct and only needed
two compliance entries. Together these collapse the typical
"PR is correct, only needs compliance fixups" case from 88 min on
Kimi to ~5 min on gpt-5-mini at Tier 0.

- Outcome-JSON synthesis in dispatch_implementer
  (_synthesize_outcome_if_missing): synthesises a concrete outcome
  from terminal_state when the worker emits no contract JSON,
  routing the escalation predicate to ESCALATE instead of UNKNOWN.

- Diff-aware gate parser (_diff_aware_gate.py): pure-Python
  classifier that splits failing BDD scenarios into related vs.
  unrelated to the PR's changed files via a feature-stem heuristic.

- Flaky-test pre-flight (_implementer_gate_preflight.py): runs
  local_ci_gate.sh --fast twice and surfaces only the persistent
  failures. Off-by-default behind IMPLEMENTER_DISPATCHER_GATE_PREFLIGHT.

- Compliance gap detector (_implementer_compliance.py): deterministic
  check of CHANGELOG / CONTRIBUTORS / commit-footer / worktree-clean
  state. Result is embedded as a "Compliance gap report" stanza so
  the agent fills in known gaps instead of discovering them.

Both prompt stanzas are appended via _append_deterministic_stanzas,
flag-gated, skipped in dry-run, and skipped when no preclone exists.
Flag-off path is byte-equivalent to the pre-feature build.

Suite: 1382 passed, 3 skipped (+59 new tests over 4 modules).

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

CHANGELOG.md

@@ -7,6 +7,69 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Added
 
+- **Deterministic worker-side improvements (2026-05-13).** Four
+  changes that move judgment off the LLM and into the dispatcher,
+  motivated by the live PR #30 escalation pilot (2026-05-12) where
+  Tier 0 spent 16 min and gave up on flaky-looking unrelated test
+  failures, while Tier 2 spent 88 min discovering that the PR was
+  already correct and only needed two compliance-checklist entries.
+  Together these collapse the typical "PR is correct, only needs
+  compliance fixups" case from 88 min on Kimi to ~5 min on
+  gpt-5-mini at Tier 0.
+
+  - **Outcome-JSON synthesis** (`dispatch_implementer._synthesize_outcome_if_missing`):
+    when the worker terminates without emitting the contract
+    `{"outcome": ...}` JSON (the gpt-5-mini failure mode in
+    production), the dispatcher synthesises one from the terminal
+    state — `completed` → `rebase-failed`, `timeout` → `timeout`,
+    `transport-error` → `transport-error`. The escalation predicate
+    then routes to ESCALATE (competence class) instead of the
+    UNKNOWN bucket that costs a wasted same-tier retry. The
+    synthesised flag is captured on the Phase 4 row so analysts
+    can distinguish worker-reported from dispatcher-inferred
+    outcomes. Always-on when `IMPLEMENTER_ESCALATION_ENABLED=1`.
+
+  - **Diff-aware gate parser** (`tools/_diff_aware_gate.py`):
+    pure-Python classifier that takes `local_ci_gate.sh` output +
+    the PR's changed-file list and splits failing BDD scenarios
+    into "related to your diff" vs. "unrelated" buckets. The
+    feature-stem heuristic catches the common case
+    (`features/auto_debug_cli_coverage.feature` ↔
+    `src/cleveragents/auto_debug/cli.py`). Used directly by the
+    flaky pre-flight; available as a standalone helper for the
+    worker.
+
+  - **Flaky-test pre-flight** (`tools/_implementer_gate_preflight.py`):
+    when `IMPLEMENTER_DISPATCHER_GATE_PREFLIGHT=1` is set, the
+    dispatcher runs `local_ci_gate.sh --fast` against the
+    pre-cloned worktree TWICE before handing the prompt to the
+    worker. Only the persistent failures (intersection of run 1
+    and run 2) are surfaced — flaky failures that cleared on
+    retry are filtered out. The classified-as-unrelated bucket
+    carries an explicit "Do NOT bail" instruction the worker can
+    cite. Off-by-default. Cost: ~6 min of dispatcher wallclock
+    per cycle, buys back significantly more (Tier 0 stops bailing
+    on flaky unrelated failures). The motivating production case
+    (PR #30 Run 1) is captured as a regression test.
+
+  - **Compliance gap detector** (`tools/_implementer_compliance.py`):
+    deterministic check of the four PR Compliance Checklist items
+    against the pre-cloned worktree — worktree clean,
+    `CHANGELOG.md [Unreleased]` non-empty, `CONTRIBUTORS.md`
+    contains the author email, HEAD commit has `ISSUES CLOSED:`
+    footer. Result is embedded in the worker prompt as a
+    "Compliance gap report (deterministic)" stanza. When all
+    gaps are closed, the agent's instruction is to emit
+    `{"outcome": "resolved", ...}` without re-touching the code;
+    when some are open, per-gap fill-in hints are inlined.
+    Activated alongside `IMPLEMENTER_ESCALATION_ENABLED=1`.
+
+  Both prompt stanzas (gate pre-flight, compliance gap) are
+  appended via `_append_deterministic_stanzas` in
+  `_prefetch_prompt`. They're independently gated and skipped in
+  dry-run / when no preclone worktree exists. Flag-off path is
+  byte-equivalent to the pre-feature build.
+
 - **In-cycle tier escalation for the implementer dispatcher
   (2026-05-12).** Replaces the cross-cycle `auto/last-attempt-tier-N`
   scheme planned in `auto-agents-tier-2-3-plan.md § Phase 5c` with a

tests/auto_agents/test_diff_aware_gate.py

@@ -0,0 +1,181 @@
+"""Unit tests for ``tools/_diff_aware_gate.py``.
+
+The classifier is pure-function — every test here constructs a
+synthetic ``gate_output`` string + a synthetic ``changed_files``
+list and asserts the classification dict has the expected shape.
+
+The load-bearing motivating case (PR #30 Run 1, 2026-05-12) is
+captured as a named regression test: gpt-5-mini ran ``--fast``,
+the unit_tests gate failed with 6 BDD scenarios that were all in
+CLI-related feature files (``automation_profile_cli.feature``,
+``cli_extensions.feature``, ``cli_output_formats.feature``,
+``plan_cli_commands_r2.feature``, ``plan_cli_coverage_boost.feature``,
+``plan_prompt_command.feature``), and the PR's actual diff only
+touched ``src/cleveragents/langgraph/graph.py`` (a LangGraph
+subscription fix). All six failures should classify as
+``unrelated_to_diff`` so the dispatcher knows not to escalate on
+them.
+"""
+from __future__ import annotations
+
+import pytest
+
+from .conftest import load_tool_module
+
+
+@pytest.fixture
+def gate():
+    return load_tool_module("_diff_aware_gate")
+
+
+# ─── Parser ─────────────────────────────────────────────────────────
+
+
+class TestParseFailingScenarios:
+    def test_extracts_path_and_line_from_behave_summary(self, gate):
+        text = """
+        Failing scenarios:
+          features/foo.feature:42  Scenario: did the thing
+          features/sub/bar.feature:7  Scenario: another
+        """
+        out = gate.parse_failing_scenarios(text)
+        assert out == [
+            {"path": "features/foo.feature", "line": "42"},
+            {"path": "features/sub/bar.feature", "line": "7"},
+        ]
+
+    def test_dedupes_repeated_failures(self, gate):
+        # Same failure reported inline + in footer should appear once
+        text = """
+        FAIL features/foo.feature:42
+        Failing scenarios:
+          features/foo.feature:42
+        """
+        out = gate.parse_failing_scenarios(text)
+        assert out == [{"path": "features/foo.feature", "line": "42"}]
+
+    def test_empty_output_returns_empty_list(self, gate):
+        assert gate.parse_failing_scenarios("") == []
+        assert gate.parse_failing_scenarios(None) == []  # type: ignore[arg-type]
+
+    def test_ignores_non_feature_paths(self, gate):
+        # Random log noise mentioning ":42" or other path patterns
+        # must NOT match — only features/...feature:N counts.
+        text = "Some unrelated log line about /tmp/foo.py:42 happened"
+        assert gate.parse_failing_scenarios(text) == []
+
+
+class TestParseGateStatuses:
+    def test_extracts_gate_statuses(self, gate):
+        text = """
+lint: PASS
+typecheck: PASS
+unit_tests: FAIL (6 scenarios)
+integration_tests: SKIP
+"""
+        statuses = gate.parse_gate_statuses(text)
+        assert statuses == {
+            "lint": "PASS",
+            "typecheck": "PASS",
+            "unit_tests": "FAIL",
+            "integration_tests": "SKIP",
+        }
+
+    def test_handles_mixed_case(self, gate):
+        text = "Lint: pass\nUnit_Tests: Fail"
+        statuses = gate.parse_gate_statuses(text)
+        assert statuses == {"lint": "PASS", "unit_tests": "FAIL"}
+
+
+# ─── Relatedness heuristic ─────────────────────────────────────────
+
+
+class TestRelatedness:
+    def test_feature_file_directly_changed_is_related(self, gate):
+        # Touched the feature itself → trivially related
+        out = gate.classify_failures(
+            "Failing scenarios:\n  features/foo.feature:7",
+            changed_files=["features/foo.feature"],
+        )
+        assert out["failures_related_to_diff"] == 1
+        assert out["failures_unrelated_to_diff"] == 0
+
+    def test_feature_stem_matches_changed_source(self, gate):
+        # Feature ``auto_debug_cli_coverage`` ~ touched
+        # ``src/cleveragents/auto_debug/cli.py``
+        out = gate.classify_failures(
+            "Failing scenarios:\n  features/auto_debug_cli_coverage.feature:42",
+            changed_files=["src/cleveragents/auto_debug/cli.py"],
+        )
+        assert out["failures_related_to_diff"] == 1
+
+    def test_pr30_run1_regression_case(self, gate):
+        # The motivating production failure case from 2026-05-12:
+        # PR #30 (langgraph fix) Run 1 (gpt-5-mini, Tier 0) had six
+        # failing CLI scenarios none of which exercise langgraph
+        # code. Tier 0 gave up; this classifier would have told it
+        # "all unrelated, ignore them".
+        text = """
+unit_tests: FAIL
+Failing scenarios:
+  features/automation_profile_cli.feature:74
+  features/cli_extensions.feature:358
+  features/cli_output_formats.feature:46
+  features/plan_cli_commands_r2.feature:224
+  features/plan_cli_coverage_boost.feature:90
+  features/plan_prompt_command.feature:30
+"""
+        out = gate.classify_failures(
+            text,
+            changed_files=["src/cleveragents/langgraph/graph.py"],
+        )
+        assert out["failures_total"] == 6
+        assert out["failures_related_to_diff"] == 0
+        assert out["failures_unrelated_to_diff"] == 6
+
+    def test_no_failures_with_passing_gates(self, gate):
+        out = gate.classify_failures(
+            "lint: PASS\ntypecheck: PASS\nunit_tests: PASS",
+            changed_files=["any.py"],
+        )
+        assert out["failures_total"] == 0
+        assert out["gate_statuses"]["lint"] == "PASS"
+
+
+# ─── Prompt rendering ──────────────────────────────────────────────
+
+
+class TestRenderPromptStanza:
+    def test_empty_classification_renders_empty_string(self, gate):
+        out = gate.render_prompt_stanza(
+            {"failures_total": 0, "gate_statuses": {}}
+        )
+        assert out == ""
+
+    def test_all_gates_pass_renders_no_failures_message(self, gate):
+        classification = gate.classify_failures(
+            "lint: PASS\ntypecheck: PASS\nunit_tests: PASS\nintegration_tests: PASS",
+            changed_files=["any.py"],
+        )
+        stanza = gate.render_prompt_stanza(classification)
+        assert "Pre-flight quality gates" in stanza
+        assert "No failures persisted" in stanza
+        assert "lint" in stanza and "PASS" in stanza
+
+    def test_mixed_failures_renders_both_buckets(self, gate):
+        text = """
+unit_tests: FAIL
+Failing scenarios:
+  features/langgraph_subscription.feature:10
+  features/automation_profile_cli.feature:74
+"""
+        out = gate.classify_failures(
+            text, changed_files=["src/cleveragents/langgraph/graph.py"],
+        )
+        stanza = gate.render_prompt_stanza(out)
+        assert "Related to your diff" in stanza
+        assert "langgraph_subscription" in stanza
+        assert "Unrelated to your diff" in stanza
+        assert "automation_profile_cli" in stanza
+        # The unrelated section should carry the "don't bail" guidance
+        assert "Do NOT bail" in stanza

tests/auto_agents/test_implementer_compliance.py

@@ -0,0 +1,270 @@
+"""Unit tests for ``tools/_implementer_compliance.py``.
+
+The detector reads from a real on-disk worktree (via ``git`` and
+file reads). Tests use ``tmp_path`` to build small fixture trees
+with controlled CHANGELOG / CONTRIBUTORS content and a real git
+repo, then assert each check returns the expected bool.
+
+The motivating production case (PR #30 Run 7, 2026-05-12) had:
+- worktree_clean: True (Kimi pushed, then the worktree is clean
+  at the new HEAD)
+- changelog_unreleased_nonempty: False at the START (this was the
+  gap Kimi had to fill)
+- contributors_has_author: False at the START
+- commit_has_issues_closed: depends on whether the commit's footer
+  was set correctly
+
+This test module pins the bool semantics for each.
+"""
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from .conftest import load_tool_module
+
+
+@pytest.fixture
+def compliance():
+    return load_tool_module("_implementer_compliance")
+
+
+@pytest.fixture
+def git_repo(tmp_path):
+    """A minimal git repo with one commit. Returns the path."""
+    subprocess.run(["git", "init", "-q", str(tmp_path)], check=True)
+    subprocess.run(
+        ["git", "-C", str(tmp_path), "config", "user.email", "test@example.com"],
+        check=True,
+    )
+    subprocess.run(
+        ["git", "-C", str(tmp_path), "config", "user.name", "Test"],
+        check=True,
+    )
+    (tmp_path / "README.md").write_text("# test\n")
+    subprocess.run(
+        ["git", "-C", str(tmp_path), "add", "."], check=True
+    )
+    subprocess.run(
+        ["git", "-C", str(tmp_path), "commit", "-q", "-m", "initial"],
+        check=True,
+    )
+    return tmp_path
+
+
+# ─── worktree_clean ────────────────────────────────────────────────
+
+
+class TestWorktreeClean:
+    def test_clean_returns_true(self, compliance, git_repo):
+        assert compliance.check_worktree_clean(git_repo) is True
+
+    def test_uncommitted_change_returns_false(self, compliance, git_repo):
+        (git_repo / "README.md").write_text("# dirty\n")
+        assert compliance.check_worktree_clean(git_repo) is False
+
+    def test_untracked_file_returns_false(self, compliance, git_repo):
+        (git_repo / "new.py").write_text("print()\n")
+        assert compliance.check_worktree_clean(git_repo) is False
+
+    def test_non_git_dir_returns_false(self, compliance, tmp_path):
+        not_a_repo = tmp_path / "nope"
+        not_a_repo.mkdir()
+        assert compliance.check_worktree_clean(not_a_repo) is False
+
+
+# ─── changelog_unreleased_nonempty ─────────────────────────────────
+
+
+class TestChangelogUnreleased:
+    def test_missing_file_returns_false(self, compliance, tmp_path):
+        assert compliance.check_changelog_unreleased_nonempty(tmp_path) is False
+
+    def test_no_unreleased_section_returns_false(self, compliance, tmp_path):
+        (tmp_path / "CHANGELOG.md").write_text(
+            "# Changelog\n\n## [1.0.0] - 2026-01-01\n- old entry\n"
+        )
+        assert compliance.check_changelog_unreleased_nonempty(tmp_path) is False
+
+    def test_empty_unreleased_section_returns_false(self, compliance, tmp_path):
+        (tmp_path / "CHANGELOG.md").write_text(
+            "# Changelog\n\n## [Unreleased]\n\n## [1.0.0]\n- old\n"
+        )
+        assert compliance.check_changelog_unreleased_nonempty(tmp_path) is False
+
+    def test_unreleased_with_bullet_returns_true(self, compliance, tmp_path):
+        (tmp_path / "CHANGELOG.md").write_text(
+            "# Changelog\n\n## [Unreleased]\n- New feature\n\n## [1.0.0]\n"
+        )
+        assert compliance.check_changelog_unreleased_nonempty(tmp_path) is True
+
+    def test_unreleased_with_only_subheader_returns_false(self, compliance, tmp_path):
+        (tmp_path / "CHANGELOG.md").write_text(
+            "# Changelog\n\n## [Unreleased]\n\n### Added\n\n## [1.0.0]\n"
+        )
+        # subheader without content → still empty
+        assert compliance.check_changelog_unreleased_nonempty(tmp_path) is False
+
+    def test_unreleased_with_paragraph_returns_true(self, compliance, tmp_path):
+        (tmp_path / "CHANGELOG.md").write_text(
+            "# Changelog\n\n## [Unreleased]\nFixed the bug.\n\n## [1.0.0]\n"
+        )
+        assert compliance.check_changelog_unreleased_nonempty(tmp_path) is True
+
+    def test_horizontal_rule_does_not_count(self, compliance, tmp_path):
+        (tmp_path / "CHANGELOG.md").write_text(
+            "# Changelog\n\n## [Unreleased]\n---\n\n## [1.0.0]\n"
+        )
+        assert compliance.check_changelog_unreleased_nonempty(tmp_path) is False
+
+
+# ─── contributors_has_author ───────────────────────────────────────
+
+
+class TestContributorsHasAuthor:
+    def test_missing_file_returns_false(self, compliance, tmp_path):
+        assert (
+            compliance.check_contributors_has_author(tmp_path, "x@y.com")
+            is False
+        )
+
+    def test_author_present_returns_true(self, compliance, tmp_path):
+        (tmp_path / "CONTRIBUTORS.md").write_text(
+            "# Contributors\n- Alice <alice@example.com>\n"
+        )
+        assert (
+            compliance.check_contributors_has_author(
+                tmp_path, "alice@example.com"
+            )
+            is True
+        )
+
+    def test_case_insensitive(self, compliance, tmp_path):
+        (tmp_path / "CONTRIBUTORS.md").write_text(
+            "- Alice <Alice@Example.Com>\n"
+        )
+        assert (
+            compliance.check_contributors_has_author(
+                tmp_path, "alice@example.com"
+            )
+            is True
+        )
+
+    def test_author_absent_returns_false(self, compliance, tmp_path):
+        (tmp_path / "CONTRIBUTORS.md").write_text(
+            "- Bob <bob@example.com>\n"
+        )
+        assert (
+            compliance.check_contributors_has_author(
+                tmp_path, "alice@example.com"
+            )
+            is False
+        )
+
+    def test_empty_email_returns_false(self, compliance, tmp_path):
+        (tmp_path / "CONTRIBUTORS.md").write_text("- Alice\n")
+        assert compliance.check_contributors_has_author(tmp_path, "") is False
+
+
+# ─── commit_has_issues_closed ──────────────────────────────────────
+
+
+class TestCommitHasIssuesClosed:
+    def test_commit_without_footer_returns_false(self, compliance, git_repo):
+        # ``git_repo`` fixture's commit message is just "initial"
+        assert compliance.check_commit_has_issues_closed(git_repo) is False
+
+    def test_commit_with_footer_returns_true(self, compliance, git_repo):
+        # Amend the commit message
+        subprocess.run(
+            ["git", "-C", str(git_repo), "commit", "--amend",
+             "-m", "fix: thing\n\nISSUES CLOSED: #42"],
+            check=True,
+            capture_output=True,
+        )
+        assert compliance.check_commit_has_issues_closed(git_repo) is True
+
+    def test_case_insensitive_footer(self, compliance, git_repo):
+        subprocess.run(
+            ["git", "-C", str(git_repo), "commit", "--amend",
+             "-m", "fix: thing\n\nissues closed: 42"],
+            check=True,
+            capture_output=True,
+        )
+        assert compliance.check_commit_has_issues_closed(git_repo) is True
+
+
+# ─── Aggregate check + rendering ───────────────────────────────────
+
+
+class TestCheckComplianceGaps:
+    def test_returns_all_four_keys(self, compliance, git_repo):
+        gaps = compliance.check_compliance_gaps(git_repo, "x@y.com")
+        assert set(gaps.keys()) == {
+            "worktree_clean",
+            "changelog_unreleased_nonempty",
+            "contributors_has_author",
+            "commit_has_issues_closed",
+        }
+
+    def test_pr30_initial_state_three_gaps(self, compliance, git_repo):
+        # Simulate the PR #30 starting state: code is committed
+        # (clean worktree), but no CHANGELOG entry, no CONTRIBUTORS
+        # attribution, no ISSUES CLOSED footer. This is what Kimi
+        # discovered manually in 76 minutes.
+        gaps = compliance.check_compliance_gaps(
+            git_repo, "hal9000@cleverthis.com",
+        )
+        assert gaps["worktree_clean"] is True  # nothing uncommitted
+        assert gaps["changelog_unreleased_nonempty"] is False
+        assert gaps["contributors_has_author"] is False
+        assert gaps["commit_has_issues_closed"] is False
+
+
+class TestAllGapsClosed:
+    def test_all_true_returns_true(self, compliance):
+        assert compliance.all_gaps_closed(
+            {"a": True, "b": True, "c": True}
+        ) is True
+
+    def test_any_false_returns_false(self, compliance):
+        assert compliance.all_gaps_closed(
+            {"a": True, "b": False}
+        ) is False
+
+    def test_empty_dict_returns_false(self, compliance):
+        # Defensive: an empty dict is "no checks ran", which we
+        # treat as not-closed.
+        assert compliance.all_gaps_closed({}) is False
+
+
+class TestRenderPromptStanza:
+    def test_empty_dict_renders_empty(self, compliance):
+        assert compliance.render_prompt_stanza({}) == ""
+
+    def test_all_passing_includes_success_directive(self, compliance):
+        out = compliance.render_prompt_stanza({
+            "worktree_clean": True,
+            "changelog_unreleased_nonempty": True,
+            "contributors_has_author": True,
+            "commit_has_issues_closed": True,
+        })
+        assert "All checks passed" in out
+        assert "\"outcome\": \"resolved\"" in out
+        assert "Do NOT re-apply" in out
+
+    def test_missing_includes_per_gap_hints(self, compliance):
+        out = compliance.render_prompt_stanza({
+            "worktree_clean": True,
+            "changelog_unreleased_nonempty": False,
+            "contributors_has_author": False,
+            "commit_has_issues_closed": True,
+        }, pr_number=30)
+        # The two missing gaps should each get a hint line
+        assert "CHANGELOG.md" in out
+        assert "CONTRIBUTORS.md" in out
+        # Closed gaps shouldn't get a hint
+        assert "amend the HEAD commit message" not in out
+        assert "Gaps to fill (2)" in out

tests/auto_agents/test_implementer_escalation_integration.py

@@ -823,6 +823,178 @@ class TestTier2DefaultOn:
         assert driver._max_tier_for_cycle() == 1
 
 
+class TestDeterministicStanzas:
+    """``_append_deterministic_stanzas`` adds the compliance-gap and
+    gate-preflight stanzas to the prefetched prompt when their flags
+    are on. The legacy (no-flags) path must be byte-equivalent to
+    the pre-feature build — no stanzas appended.
+
+    Pinned plan-2026-05-13 invariant: each stanza is independently
+    gated. Compliance behind ``IMPLEMENTER_ESCALATION_ENABLED``,
+    gate-preflight behind ``IMPLEMENTER_DISPATCHER_GATE_PREFLIGHT``.
+    """
+
+    def test_no_flags_appends_nothing(
+        self, driver, cfg, monkeypatch, tmp_path
+    ):
+        # Both flags off → returned prompt is byte-equivalent.
+        monkeypatch.delenv("IMPLEMENTER_ESCALATION_ENABLED", raising=False)
+        monkeypatch.delenv(
+            "IMPLEMENTER_DISPATCHER_GATE_PREFLIGHT", raising=False,
+        )
+        item = _pr_item()
+        result = type("FakeResult", (), {"diff": ""})()
+        handle = type(
+            "FakeHandle", (), {"path": str(tmp_path)},
+        )()
+        out = driver._append_deterministic_stanzas(
+            "BASE PROMPT", cfg, item, _pr_group(driver), result, handle,
+        )
+        assert out == "BASE PROMPT"
+
+    def test_dry_run_skips_stanzas(
+        self, driver, cfg, monkeypatch, tmp_path
+    ):
+        # Even with flags on, dry-run never appends.
+        monkeypatch.setenv("IMPLEMENTER_ESCALATION_ENABLED", "1")
+        monkeypatch.setenv("IMPLEMENTER_DISPATCHER_GATE_PREFLIGHT", "1")
+        from .conftest import make_dispatch_config
+        dry_cfg = make_dispatch_config(tmp_path, dry_run=True)
+        item = _pr_item()
+        handle = type(
+            "FakeHandle", (), {"path": str(tmp_path)},
+        )()
+        out = driver._append_deterministic_stanzas(
+            "BASE PROMPT", dry_cfg, item, _pr_group(driver), None, handle,
+        )
+        assert out == "BASE PROMPT"
+
+    def test_no_clone_handle_skips_stanzas(
+        self, driver, cfg, monkeypatch
+    ):
+        # When preclone is disabled / failed, no worktree → no
+        # filesystem to check → skip both stanzas.
+        monkeypatch.setenv("IMPLEMENTER_ESCALATION_ENABLED", "1")
+        item = _pr_item()
+        out = driver._append_deterministic_stanzas(
+            "BASE PROMPT", cfg, item, _pr_group(driver), None, None,
+        )
+        assert out == "BASE PROMPT"
+
+    def test_escalation_flag_on_appends_compliance_stanza(
+        self, driver, cfg, monkeypatch, tmp_path
+    ):
+        import subprocess
+        # Build a real git repo so the compliance detector can run
+        subprocess.run(["git", "init", "-q", str(tmp_path)], check=True)
+        subprocess.run(
+            ["git", "-C", str(tmp_path), "config", "user.email", "x@y.com"],
+            check=True,
+        )
+        subprocess.run(
+            ["git", "-C", str(tmp_path), "config", "user.name", "Test"],
+            check=True,
+        )
+        (tmp_path / "README.md").write_text("init\n")
+        subprocess.run(
+            ["git", "-C", str(tmp_path), "add", "."], check=True,
+        )
+        subprocess.run(
+            ["git", "-C", str(tmp_path), "commit", "-q", "-m", "init"],
+            check=True,
+        )
+
+        monkeypatch.setenv("IMPLEMENTER_ESCALATION_ENABLED", "1")
+        monkeypatch.delenv(
+            "IMPLEMENTER_DISPATCHER_GATE_PREFLIGHT", raising=False,
+        )
+        monkeypatch.setenv("GIT_USER_EMAIL", "hal9000@cleverthis.com")
+        item = _pr_item()
+        result = type("FakeResult", (), {"diff": ""})()
+        handle = type(
+            "FakeHandle", (), {"path": str(tmp_path)},
+        )()
+        out = driver._append_deterministic_stanzas(
+            "BASE PROMPT", cfg, item, _pr_group(driver), result, handle,
+        )
+        # Compliance stanza header should be present
+        assert "Compliance gap report" in out
+        # The git_repo fixture has no CHANGELOG → gap reported
+        assert "changelog_unreleased_nonempty" in out
+        # Gate-preflight stanza should NOT be present (flag off)
+        assert "Pre-flight quality gates" not in out
+
+
+def _pr_group(driver):
+    """Return the failing_ci_pr WorkGroup from the driver's
+    WORK_GROUPS list — same one dispatch_one uses for PR work."""
+    return next(g for g in driver.WORK_GROUPS if g.name == "failing_ci_pr")
+
+
+class TestOutcomeSynthesis:
+    """``_synthesize_outcome_if_missing`` post-processes worker
+    output so a worker that gave up without emitting the contract
+    JSON gets a concrete outcome string. Otherwise the escalation
+    predicate falls into the UNKNOWN bucket which costs one wasted
+    same-tier retry per cycle (the gpt-5-mini failure mode observed
+    in Run 1 on PR #30, 2026-05-12).
+    """
+
+    def test_passthrough_when_outcome_present(self, driver):
+        parsed, synth = driver._synthesize_outcome_if_missing(
+            {"outcome": "resolved", "files_touched": ["a.py"]},
+            "completed",
+        )
+        assert synth is False
+        assert parsed == {"outcome": "resolved", "files_touched": ["a.py"]}
+
+    def test_synthesise_completed_to_rebase_failed(self, driver):
+        parsed, synth = driver._synthesize_outcome_if_missing(
+            None, "completed",
+        )
+        assert synth is True
+        assert parsed["outcome"] == "rebase-failed"
+        assert parsed["_synthesized"] is True
+        assert parsed["files_touched"] == []
+
+    def test_synthesise_timeout(self, driver):
+        parsed, synth = driver._synthesize_outcome_if_missing(
+            None, "timeout",
+        )
+        assert synth is True
+        assert parsed["outcome"] == "timeout"
+
+    def test_synthesise_transport_error(self, driver):
+        parsed, synth = driver._synthesize_outcome_if_missing(
+            None, "transport-error",
+        )
+        assert synth is True
+        assert parsed["outcome"] == "transport-error"
+
+    def test_unknown_terminal_state_synthesises_unknown(self, driver):
+        parsed, synth = driver._synthesize_outcome_if_missing(
+            None, "weird-new-state-from-future",
+        )
+        assert synth is True
+        assert parsed["outcome"] == "unknown"
+
+    def test_empty_dict_synthesises(self, driver):
+        # parsed_json is a dict but the outcome key is absent
+        parsed, synth = driver._synthesize_outcome_if_missing(
+            {}, "completed",
+        )
+        assert synth is True
+        assert parsed["outcome"] == "rebase-failed"
+
+    def test_non_string_outcome_synthesises(self, driver):
+        # outcome present but not a string (LLM-output noise)
+        parsed, synth = driver._synthesize_outcome_if_missing(
+            {"outcome": True}, "completed",
+        )
+        assert synth is True
+        assert parsed["outcome"] == "rebase-failed"
+
+
 class TestStatusCommentFingerprintTier:
     """Direct unit-ish tests for ``_review_post.post_implementer_status_comment``'s
     tier-in-fingerprint behaviour. The integration suite stubs the

tests/auto_agents/test_implementer_gate_preflight.py

@@ -0,0 +1,197 @@
+"""Unit tests for ``tools/_implementer_gate_preflight.py``.
+
+The pre-flight is gated behind ``IMPLEMENTER_DISPATCHER_GATE_PREFLIGHT``
+and runs ``local_ci_gate.sh --fast`` twice via subprocess. We stub
+the subprocess via the ``runner`` parameter on :func:`run_preflight`
+so tests are fully hermetic — no real gates run.
+
+Key behaviours pinned:
+
+- **Flag off** → returns an empty/disabled classification.
+- **Flag on, run 1 passes** → run 2 is skipped (no redundant work).
+- **Flag on, run 1 fails, run 2 passes** → flaky failures are
+  filtered out; the agent sees zero persistent failures.
+- **Flag on, both runs fail with the same scenarios** → those
+  scenarios surface as persistent failures.
+- **Flag on, runs fail with DIFFERENT scenarios** → only the
+  intersection is persistent.
+
+The motivating regression case (PR #30 Run 1, 2026-05-12) is
+covered: six CLI-feature scenarios that flake intermittently, on
+a PR that only touches LangGraph code, should classify as zero
+persistent-related failures when retried.
+"""
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from .conftest import load_tool_module
+
+
+@pytest.fixture
+def preflight():
+    return load_tool_module("_implementer_gate_preflight")
+
+
+def _fake_runner(stdouts: list[str], returncodes: list[int]):
+    """Return a runner callable that yields the scripted outputs in
+    order. Each call pops the next ``(stdout, returncode)`` pair."""
+    calls = iter(zip(stdouts, returncodes))
+
+    def runner(cmd, env, timeout):
+        out, rc = next(calls)
+        return subprocess.CompletedProcess(
+            args=cmd, returncode=rc, stdout=out, stderr=""
+        )
+
+    return runner
+
+
+# ─── Flag gating ───────────────────────────────────────────────────
+
+
+class TestFlagGating:
+    def test_disabled_by_default(self, preflight, monkeypatch):
+        monkeypatch.delenv(preflight.PREFLIGHT_ENABLED_ENV_VAR, raising=False)
+        result = preflight.run_preflight(
+            Path("/tmp/fake"), changed_files=["x.py"],
+        )
+        assert result["preflight_enabled"] is False
+        assert result["failures_total"] == 0
+        assert result["runs"] == []
+
+    def test_truthy_env_var_enables(self, preflight, monkeypatch):
+        for val in ("1", "true", "TRUE", "yes", "on"):
+            monkeypatch.setenv(preflight.PREFLIGHT_ENABLED_ENV_VAR, val)
+            assert preflight.is_preflight_enabled() is True
+
+    def test_falsy_env_var_disables(self, preflight, monkeypatch):
+        for val in ("", "0", "false", "no", "off"):
+            monkeypatch.setenv(preflight.PREFLIGHT_ENABLED_ENV_VAR, val)
+            assert preflight.is_preflight_enabled() is False
+
+
+# ─── Persistent-failure detection ──────────────────────────────────
+
+
+class TestPersistentFailures:
+    def test_run1_passing_skips_run2(self, preflight, monkeypatch):
+        monkeypatch.setenv(preflight.PREFLIGHT_ENABLED_ENV_VAR, "1")
+        runner = _fake_runner(
+            stdouts=["lint: PASS\nunit_tests: PASS\n"],
+            returncodes=[0],
+        )
+        result = preflight.run_preflight(
+            Path("/tmp"), changed_files=["x.py"], runner=runner,
+        )
+        # Only one run recorded; failures_total = 0
+        assert len(result["runs"]) == 1
+        assert result["failures_total"] == 0
+        assert result["gate_statuses"]["lint"] == "PASS"
+
+    def test_run1_fails_run2_passes_filters_flaky(
+        self, preflight, monkeypatch
+    ):
+        # Six BDD failures on run 1 — all gone on run 2 → flaky.
+        # Persistent failures = empty.
+        monkeypatch.setenv(preflight.PREFLIGHT_ENABLED_ENV_VAR, "1")
+        run1_out = """
+unit_tests: FAIL
+Failing scenarios:
+  features/automation_profile_cli.feature:74
+  features/cli_extensions.feature:358
+  features/cli_output_formats.feature:46
+"""
+        run2_out = "lint: PASS\nunit_tests: PASS\n"
+        runner = _fake_runner(
+            stdouts=[run1_out, run2_out], returncodes=[1, 0],
+        )
+        result = preflight.run_preflight(
+            Path("/tmp"),
+            changed_files=["src/cleveragents/langgraph/graph.py"],
+            runner=runner,
+        )
+        assert len(result["runs"]) == 2
+        assert result["failures_total"] == 0
+        assert result["failures_unrelated_to_diff"] == 0
+
+    def test_both_runs_fail_same_scenarios_surfaces_persistent(
+        self, preflight, monkeypatch
+    ):
+        # Two runs with identical failures → all persistent
+        monkeypatch.setenv(preflight.PREFLIGHT_ENABLED_ENV_VAR, "1")
+        same = """
+unit_tests: FAIL
+Failing scenarios:
+  features/real_bug.feature:10
+  features/related.feature:20
+"""
+        runner = _fake_runner(stdouts=[same, same], returncodes=[1, 1])
+        result = preflight.run_preflight(
+            Path("/tmp"),
+            changed_files=["src/cleveragents/real_bug/mod.py"],
+            runner=runner,
+        )
+        assert result["failures_total"] == 2
+        # ``real_bug`` feature stem ↔ ``real_bug/mod.py`` → related
+        assert result["failures_related_to_diff"] == 1
+        # ``related.feature`` doesn't intersect → unrelated
+        assert result["failures_unrelated_to_diff"] == 1
+
+    def test_runs_fail_different_scenarios_surfaces_intersection(
+        self, preflight, monkeypatch
+    ):
+        # Run 1 fails A,B,C; run 2 fails B,C,D. Persistent = B,C.
+        monkeypatch.setenv(preflight.PREFLIGHT_ENABLED_ENV_VAR, "1")
+        run1 = """
+unit_tests: FAIL
+Failing scenarios:
+  features/a.feature:1
+  features/b.feature:2
+  features/c.feature:3
+"""
+        run2 = """
+unit_tests: FAIL
+Failing scenarios:
+  features/b.feature:2
+  features/c.feature:3
+  features/d.feature:4
+"""
+        runner = _fake_runner(stdouts=[run1, run2], returncodes=[1, 1])
+        result = preflight.run_preflight(
+            Path("/tmp"), changed_files=["x.py"], runner=runner,
+        )
+        assert result["failures_total"] == 2  # only b, c persisted
+
+    def test_pr30_regression_case_zero_persistent_when_retry_clears(
+        self, preflight, monkeypatch
+    ):
+        # Reconstruct the PR #30 Run 1 (2026-05-12) failure shape:
+        # six CLI-feature scenarios fail on run 1, all pass on run 2,
+        # PR's actual diff was LangGraph. Persistent-related = 0
+        # → agent gets a clean prompt and doesn't bail.
+        monkeypatch.setenv(preflight.PREFLIGHT_ENABLED_ENV_VAR, "1")
+        run1 = """
+unit_tests: FAIL
+Failing scenarios:
+  features/automation_profile_cli.feature:74
+  features/cli_extensions.feature:358
+  features/cli_output_formats.feature:46
+  features/plan_cli_commands_r2.feature:224
+  features/plan_cli_coverage_boost.feature:90
+  features/plan_prompt_command.feature:30
+"""
+        run2 = "lint: PASS\ntypecheck: PASS\nunit_tests: PASS\nintegration_tests: PASS"
+        runner = _fake_runner(stdouts=[run1, run2], returncodes=[1, 0])
+        result = preflight.run_preflight(
+            Path("/tmp"),
+            changed_files=["src/cleveragents/langgraph/graph.py"],
+            runner=runner,
+        )
+        # All gates pass on second attempt → zero persistent failures.
+        assert result["failures_total"] == 0
+        assert result["runs"][0]["failures"] == 6
+        assert result["runs"][1]["failures"] == 0

tools/_diff_aware_gate.py

@@ -0,0 +1,257 @@
+"""Diff-aware classifier for ``local_ci_gate.sh`` quality-gate
+output.
+
+Cross-references the failing BDD scenarios reported by the gate
+runner against the PR's changed-files list, splitting failures
+into two buckets:
+
+- ``related_failures`` — failures in feature files whose paths
+  match (or are colocated with) files the PR touches. Almost
+  always real regressions caused by the PR.
+- ``unrelated_failures`` — failures in feature files the PR did
+  not touch. Strongly correlated with environmental flakiness or
+  CI-side intermittency — these are the failures Tier 0 historically
+  bailed out on (e.g., PR #30 Run 1, 2026-05-12, where six CLI
+  scenarios failed unrelated to the LangGraph fix and gpt-5-mini
+  gave up rather than re-running them).
+
+Used by the dispatcher's pre-flight pre-fetch path (Heavy flaky
+pre-flight, plan §"#1 Heavy") to verify a failure is persistent
+before surfacing it to the worker. Also useful as a standalone
+helper the implementer worker can call mid-session.
+
+This module is pure-Python and has no I/O of its own — callers
+provide the gate output text and the list of changed files. Easy
+to unit-test.
+"""
+from __future__ import annotations
+
+import re
+from typing import Iterable
+
+
+# Pattern for failing-BDD-scenario lines in ``behave`` output:
+#
+#     features/path/to/file.feature:42  Scenario: did the thing
+#
+# behave reports failures as "Failing scenarios:" header followed by
+# indented ``path:line`` lines. The trailing scenario name is
+# optional and may contain anything; we only capture the file path
+# and line number which are the load-bearing identifiers.
+_BEHAVE_FAILURE_RE = re.compile(
+    r"^\s*(?P<path>features/[^\s:]+\.feature):(?P<line>\d+)",
+    re.MULTILINE,
+)
+
+# When a gate reports "FAILED" via the local_ci_gate.sh wrapper the
+# header line looks like ``CI / unit_tests*  FAIL  ...``. Used as a
+# coarse status indicator for gates the parser doesn't deep-parse.
+_GATE_STATUS_RE = re.compile(
+    r"^(?P<gate>lint|typecheck|unit_tests|integration_tests|"
+    r"e2e_tests|coverage)\s*:\s*(?P<status>PASS|FAIL|SKIP)",
+    re.MULTILINE | re.IGNORECASE,
+)
+
+
+def parse_failing_scenarios(gate_output: str) -> list[dict[str, str]]:
+    """Extract failing scenario paths + line numbers from gate output.
+
+    Returns a list of ``{"path": str, "line": str}`` dicts in the
+    order they appear in the output. Duplicates are de-duplicated
+    by ``(path, line)`` because a behave summary tends to list each
+    failure both inline and in the "Failing scenarios:" footer.
+    """
+    seen: set[tuple[str, str]] = set()
+    out: list[dict[str, str]] = []
+    for m in _BEHAVE_FAILURE_RE.finditer(gate_output or ""):
+        key = (m.group("path"), m.group("line"))
+        if key in seen:
+            continue
+        seen.add(key)
+        out.append({"path": m.group("path"), "line": m.group("line")})
+    return out
+
+
+def parse_gate_statuses(gate_output: str) -> dict[str, str]:
+    """Map gate name → status (PASS/FAIL/SKIP) parsed from output.
+
+    Coarse-grained — used by the dispatcher to surface a high-level
+    gate roll-up in the worker prompt without exposing the full raw
+    output. The agent reads this dict and can decide whether to
+    drill into specific failures.
+    """
+    out: dict[str, str] = {}
+    for m in _GATE_STATUS_RE.finditer(gate_output or ""):
+        out[m.group("gate").lower()] = m.group("status").upper()
+    return out
+
+
+def _normalise_path(p: str) -> str:
+    """Strip leading ``./`` and trailing whitespace from a path."""
+    return (p or "").strip().removeprefix("./")
+
+
+def _feature_likely_related(
+    feature_path: str, changed_files: Iterable[str]
+) -> bool:
+    """Heuristic: a feature is likely related to a PR's changes if
+    EITHER the feature file itself was changed OR the feature's
+    name-stem appears in a changed source-file path.
+
+    Examples:
+    - feature ``features/auto_debug_cli_coverage.feature`` is related
+      to a PR touching ``src/cleveragents/auto_debug/cli.py`` (stem
+      ``auto_debug`` appears in both).
+    - feature ``features/plan_cli_commands_r2.feature`` is unrelated
+      to a PR touching ``langgraph/graph.py`` (no overlap).
+
+    This is intentionally simple — false positives (calling
+    something related when it isn't) are cheap (we just don't
+    auto-suppress that failure). False negatives (calling something
+    unrelated when it IS) are also cheap because the worker still
+    sees the failure list and can decide.
+    """
+    feature_path_norm = _normalise_path(feature_path)
+    changed_norm = {_normalise_path(c) for c in changed_files}
+
+    # Direct hit — feature file itself changed.
+    if feature_path_norm in changed_norm:
+        return True
+
+    # Extract the stem of the feature name (strip extension + path).
+    # ``features/auto_debug_cli_coverage.feature`` → ``auto_debug_cli_coverage``.
+    stem = feature_path_norm.split("/")[-1]
+    if stem.endswith(".feature"):
+        stem = stem[: -len(".feature")]
+    if not stem:
+        return False
+
+    # Split the stem into prefix tokens — the most-specific token
+    # first. We compare just the first 2-3 tokens because feature
+    # filenames are often verbose ("plan_cli_commands_r2_boost") and
+    # matching the WHOLE stem against a path would miss legitimate
+    # matches like ``plan_cli/commands.py``.
+    tokens = stem.split("_")
+    # Take the first 2 tokens as the "salient prefix" — the bit
+    # that's most likely to map to a source-tree path component.
+    prefix_candidates = []
+    if tokens:
+        prefix_candidates.append(tokens[0])
+    if len(tokens) >= 2:
+        prefix_candidates.append("_".join(tokens[:2]))
+
+    for prefix in prefix_candidates:
+        if not prefix:
+            continue
+        for changed in changed_norm:
+            # Match against PATH SEGMENTS — avoid matching substrings
+            # of unrelated names (``plan`` matches ``plans/`` but
+            # not ``misplanned``).
+            segments = changed.replace(".", "/").split("/")
+            if prefix in segments:
+                return True
+    return False
+
+
+def classify_failures(
+    gate_output: str, changed_files: Iterable[str]
+) -> dict[str, object]:
+    """Top-level classifier: parse + relate.
+
+    Returns a structured dict the dispatcher embeds in the worker
+    prompt:
+
+    .. code-block:: python
+
+        {
+            "gate_statuses": {"lint": "PASS", "unit_tests": "FAIL", ...},
+            "failures_total": 6,
+            "failures_related_to_diff": 0,
+            "failures_unrelated_to_diff": 6,
+            "related": [{"path": ..., "line": ...}, ...],
+            "unrelated": [{"path": ..., "line": ...}, ...],
+        }
+
+    The agent's instruction (in implementation-worker / task-implementor
+    prompts): if ``failures_related_to_diff == 0`` and the gate
+    pre-flight already retried persistent failures, the run is
+    PROBABLY flaky — focus on compliance gaps / fill-the-blanks
+    rather than diagnosing the unrelated failures.
+    """
+    statuses = parse_gate_statuses(gate_output)
+    failures = parse_failing_scenarios(gate_output)
+    changed = list(changed_files)
+    related: list[dict[str, str]] = []
+    unrelated: list[dict[str, str]] = []
+    for f in failures:
+        if _feature_likely_related(f["path"], changed):
+            related.append(f)
+        else:
+            unrelated.append(f)
+    return {
+        "gate_statuses": statuses,
+        "failures_total": len(failures),
+        "failures_related_to_diff": len(related),
+        "failures_unrelated_to_diff": len(unrelated),
+        "related": related,
+        "unrelated": unrelated,
+    }
+
+
+def render_prompt_stanza(classification: dict[str, object]) -> str:
+    """Render the classification result as a markdown stanza for the
+    worker prompt. Kept here (next to the classifier) so the
+    rendered shape stays lockstep with the data shape.
+
+    Returns the empty string when there are zero failures to surface
+    — the dispatcher embeds the stanza unconditionally when the
+    classification ran, but the agent reads "empty stanza = all
+    gates passed in pre-flight" without a special-case.
+    """
+    total = classification.get("failures_total", 0)
+    related = classification.get("related") or []
+    unrelated = classification.get("unrelated") or []
+    statuses = classification.get("gate_statuses") or {}
+    if not total and not statuses:
+        return ""
+
+    lines = ["## Pre-flight quality gates (verified persistent)\n"]
+    if statuses:
+        lines.append("Gate summary:")
+        for gate, status in sorted(statuses.items()):
+            lines.append(f"- `{gate}`: **{status}**")
+        lines.append("")
+
+    if total == 0:
+        lines.append(
+            "_No failures persisted after the pre-flight retry — "
+            "you do NOT need to re-run `local_ci_gate.sh --fast` "
+            "yourself unless you change files._"
+        )
+        return "\n".join(lines)
+
+    lines.append(
+        f"Failures total: **{total}** "
+        f"(related to your diff: {len(related)}, "
+        f"unrelated: {len(unrelated)})."
+    )
+    lines.append("")
+    if related:
+        lines.append("**Related to your diff** — investigate these:")
+        for f in related:
+            lines.append(f"- `{f['path']}:{f['line']}`")
+        lines.append("")
+    if unrelated:
+        lines.append(
+            "**Unrelated to your diff** — these were retried by "
+            "the dispatcher's pre-flight and persisted, but they "
+            "exercise code your PR did not touch. Likely "
+            "environmental / pre-existing. Do NOT bail on the "
+            "cycle for these — focus on the related failures "
+            "above (if any) and on compliance gaps (next section). "
+            "If you have time, you may re-run these locally to "
+            "confirm reproducibility before exiting."
+        )
+        for f in unrelated:
+            lines.append(f"- `{f['path']}:{f['line']}`")
+    return "\n".join(lines)

tools/_implementer_compliance.py

@@ -0,0 +1,264 @@
+"""Compliance gap detector for the implementer dispatcher.
+
+Walks the pre-cloned worktree and checks PR Compliance Checklist
+items mechanically:
+
+- ``worktree_clean`` — no uncommitted changes (``git status
+  --porcelain`` empty). When True, the agent doesn't need to
+  re-apply the code fix; whatever's at HEAD is already what gets
+  pushed.
+- ``changelog_unreleased_nonempty`` — the ``[Unreleased]`` section
+  in CHANGELOG.md has at least one entry.
+- ``contributors_has_author`` — the git author email appears in
+  CONTRIBUTORS.md.
+- ``commit_has_issues_closed`` — the last commit's message body
+  contains an ``ISSUES CLOSED: #N`` footer.
+
+The detector emits a structured dict that gets embedded into the
+worker prompt under a ``## Compliance gap report (deterministic)``
+stanza. The agent's instruction is: when ALL bools are true, the
+PR is done; emit ``{"outcome": "resolved", ...}``. When some are
+false, fill ONLY the missing items — do NOT re-touch the code fix.
+
+Motivation (plan §"#2", 2026-05-13): in production on 2026-05-12,
+the actual fix on PR #30 was just two missing compliance items
+(CHANGELOG + CONTRIBUTORS) — Kimi at Tier 2 spent 76 minutes
+DISCOVERING this and applying it. Detection takes <100 ms. Letting
+the worker fill in known gaps (instead of having it explore the
+worktree to FIND them) collapses the work to a one- or two-turn
+fill-in-the-blanks.
+
+The module has no LLM dependency and is pure read-only filesystem
+work — easy to unit-test against fixture trees.
+"""
+from __future__ import annotations
+
+import re
+import subprocess
+from pathlib import Path
+from typing import Any
+
+
+def _git(args: list[str], cwd: Path, timeout: int = 10) -> tuple[int, str]:
+    """Run a ``git`` subprocess inside ``cwd`` and return
+    ``(returncode, stdout)``. Stderr is discarded — for these
+    read-only commands we only care about success + stdout."""
+    try:
+        result = subprocess.run(
+            ["git", *args],
+            cwd=str(cwd),
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            check=False,
+        )
+    except (subprocess.TimeoutExpired, FileNotFoundError):
+        return -1, ""
+    return result.returncode, (result.stdout or "")
+
+
+def check_worktree_clean(worktree: Path) -> bool:
+    """``True`` iff ``git status --porcelain`` is empty.
+
+    Empty = no uncommitted changes = whatever's at HEAD is what
+    would get pushed if the agent ran ``git push``. The agent should
+    NOT re-apply code changes when this is True; the diff is already
+    present in the committed history.
+    """
+    rc, out = _git(["status", "--porcelain"], worktree)
+    if rc != 0:
+        return False
+    return out.strip() == ""
+
+
+# Regex to find the ``[Unreleased]`` section and the next ``## `` or
+# end-of-file. Multi-line, ungreedy. The section is "non-empty" if
+# anything but whitespace + horizontal rules appears between the
+# header and the next section.
+_UNRELEASED_SECTION_RE = re.compile(
+    r"^##\s*\[Unreleased\][^\n]*\n(?P<body>.*?)(?=^##\s|\Z)",
+    re.MULTILINE | re.DOTALL | re.IGNORECASE,
+)
+
+
+def check_changelog_unreleased_nonempty(worktree: Path) -> bool:
+    """``True`` iff ``CHANGELOG.md`` has a ``[Unreleased]`` section
+    with at least one non-trivial line.
+
+    "Non-trivial" excludes lines that are just whitespace, comments,
+    or ``---`` horizontal-rule separators. A section that only
+    contains a placeholder ``###`` subheader with no content under
+    it counts as empty.
+
+    Returns ``False`` if the file is missing or unreadable — the
+    agent's job in that case is to either create the section or
+    flag the structural problem.
+    """
+    changelog = worktree / "CHANGELOG.md"
+    try:
+        text = changelog.read_text(encoding="utf-8")
+    except (OSError, UnicodeDecodeError):
+        return False
+    m = _UNRELEASED_SECTION_RE.search(text)
+    if m is None:
+        return False
+    body = m.group("body")
+    for line in body.splitlines():
+        stripped = line.strip()
+        if not stripped:
+            continue
+        if stripped.startswith("#"):
+            # subheader like "### Added" / "### Changed" — only
+            # counts as content if accompanied by an entry under it
+            continue
+        if set(stripped) <= {"-", "="}:
+            # horizontal rule
+            continue
+        # Found a real content line (bullet, paragraph, etc.)
+        return True
+    return False
+
+
+def check_contributors_has_author(
+    worktree: Path, git_user_email: str
+) -> bool:
+    """``True`` iff ``CONTRIBUTORS.md`` contains the author's email
+    (case-insensitive)."""
+    if not git_user_email:
+        return False
+    contributors = worktree / "CONTRIBUTORS.md"
+    try:
+        text = contributors.read_text(encoding="utf-8")
+    except (OSError, UnicodeDecodeError):
+        return False
+    return git_user_email.lower() in text.lower()
+
+
+_ISSUES_CLOSED_RE = re.compile(
+    r"^\s*ISSUES\s*CLOSED\s*:\s*#?\d+", re.MULTILINE | re.IGNORECASE
+)
+
+
+def check_commit_has_issues_closed(worktree: Path) -> bool:
+    """``True`` iff the HEAD commit's message body contains
+    ``ISSUES CLOSED: #N``.
+
+    Reads the full message via ``git log -1 --pretty=%B``. Case-
+    insensitive. The actual project convention uses uppercase
+    ``ISSUES CLOSED:`` but we accept variants like
+    ``Issues closed: 42`` for robustness.
+    """
+    rc, body = _git(["log", "-1", "--pretty=%B", "HEAD"], worktree)
+    if rc != 0:
+        return False
+    return _ISSUES_CLOSED_RE.search(body) is not None
+
+
+def check_compliance_gaps(
+    worktree: Path, git_user_email: str = ""
+) -> dict[str, bool]:
+    """Run all four compliance checks and return a structured dict.
+
+    Each value is a strict ``bool``. The agent's prompt instruction
+    is to inspect this dict and fill in the missing items only.
+    """
+    return {
+        "worktree_clean": check_worktree_clean(worktree),
+        "changelog_unreleased_nonempty": check_changelog_unreleased_nonempty(
+            worktree
+        ),
+        "contributors_has_author": check_contributors_has_author(
+            worktree, git_user_email
+        ),
+        "commit_has_issues_closed": check_commit_has_issues_closed(worktree),
+    }
+
+
+def all_gaps_closed(gaps: dict[str, bool]) -> bool:
+    """``True`` iff every check passed. Convenience for callers that
+    want a single "is the PR done?" signal."""
+    return all(bool(v) for v in gaps.values()) and bool(gaps)
+
+
+def render_prompt_stanza(
+    gaps: dict[str, bool], pr_number: int | None = None
+) -> str:
+    """Render the compliance-gap report as a markdown stanza for
+    the worker prompt.
+
+    The instruction text tells the agent how to react:
+
+    - All True → emit the success-JSON and exit (no code changes
+      needed).
+    - Some False → fill the missing items only.
+
+    Keeps the rendering co-located with the data shape so a future
+    schema change lands in one place.
+    """
+    if not gaps:
+        return ""
+
+    lines = ["## Compliance gap report (deterministic)\n"]
+    lines.append(
+        "The dispatcher checked the PR Compliance Checklist items "
+        "mechanically against the pre-cloned worktree. Treat these "
+        "as ground truth — they're cheaper to trust than re-deriving."
+    )
+    lines.append("")
+    for key, value in sorted(gaps.items()):
+        marker = "✅" if value else "❌"
+        lines.append(f"- {marker} `{key}`: **{value}**")
+    lines.append("")
+
+    if all_gaps_closed(gaps):
+        lines.append(
+            "**All checks passed.** The PR is already complete — "
+            "the code fix is in HEAD and every Compliance Checklist "
+            "item is satisfied. Do NOT re-apply or re-edit the "
+            "code. Verify the local quality gates (if you have not "
+            "already) and emit:"
+        )
+        lines.append(
+            "    {\"outcome\": \"resolved\", \"files_touched\": []}"
+        )
+    else:
+        missing = [k for k, v in gaps.items() if not v]
+        lines.append(
+            f"**Gaps to fill ({len(missing)}):** "
+            + ", ".join(f"`{m}`" for m in missing)
+        )
+        lines.append("")
+        lines.append(
+            "Fill ONLY the missing items — the existing code fix in "
+            "HEAD is correct and should not be re-touched unless a "
+            "related-to-diff gate failure tells you otherwise."
+        )
+        # Per-gap hints
+        if not gaps.get("changelog_unreleased_nonempty", True):
+            lines.append(
+                "- For `changelog_unreleased_nonempty`: add an entry "
+                "under the `[Unreleased]` section of `CHANGELOG.md` "
+                "describing the change in this PR."
+            )
+        if not gaps.get("contributors_has_author", True):
+            lines.append(
+                "- For `contributors_has_author`: add a line to "
+                "`CONTRIBUTORS.md` attributing the change to the "
+                "author email in your prompt."
+            )
+        if not gaps.get("commit_has_issues_closed", True):
+            tail = (
+                f" referencing PR #{pr_number}'s linked issue"
+                if pr_number else ""
+            )
+            lines.append(
+                "- For `commit_has_issues_closed`: amend the HEAD "
+                f"commit message to add an `ISSUES CLOSED: #N` footer{tail}."
+            )
+        if not gaps.get("worktree_clean", True):
+            lines.append(
+                "- For `worktree_clean`: there are uncommitted "
+                "changes in the worktree — commit them (or stash "
+                "if intentional) before pushing."
+            )
+    return "\n".join(lines)

tools/_implementer_gate_preflight.py

@@ -0,0 +1,242 @@
+"""Heavy flaky-test pre-flight for the implementer dispatcher.
+
+Runs ``local_ci_gate.sh --fast`` against the pre-cloned worktree
+TWICE and classifies failures as persistent (failed both runs)
+vs. flaky (passed the second run). The persistent-failure list is
+then surfaced in the worker prompt via
+:mod:`_diff_aware_gate.render_prompt_stanza`, split by
+related-to-diff vs unrelated-to-diff buckets.
+
+Motivation (plan §"#1 Heavy", 2026-05-13): in production on
+2026-05-12, gpt-5-mini at Tier 0 ran ``--fast`` once on PR #30,
+observed six CLI-feature scenarios failing in unit_tests, decided
+they were "probably flaky or environmental" but unrelated to the
+LangGraph fix in the diff — and then **gave up**. Kimi (Tier 2)
+ran the SAME ``--fast`` seven hours later and got all four gates
+PASS. The dispatcher running ``--fast`` twice up front would have
+caught the flakiness deterministically; gpt-5-mini would have
+either seen empty failures (flakes self-cleared on retry) OR seen
+a clean "unrelated to your diff" classification and known not to
+bail.
+
+Gating:
+- Always-off by default (no behaviour change).
+- Activated via ``IMPLEMENTER_DISPATCHER_GATE_PREFLIGHT=1``.
+- Independent of the in-cycle escalation flag; this is a strict
+  upgrade to the worker's pre-fetched context that has no
+  byte-equivalence concerns of its own (it ADDS a prompt stanza
+  rather than modifying existing ones).
+
+Cost:
+- Two ``local_ci_gate.sh --fast`` invocations per dispatched PR
+  cycle. First-cold-cache run is ~5 min (per the R1 entry in
+  CHANGELOG); second is warm-cache, typically <1 min. Net
+  pre-flight overhead: ~6 min per cycle. Buys back significantly
+  more (Tier 0 attempts no longer bail on flaky unrelated tests).
+
+The module has NO I/O of its own beyond ``subprocess.run`` calls
+on the operator's local checkout — no Forgejo calls, no model
+calls. Easy to unit-test by stubbing out the runner via the
+``runner`` parameter on :func:`run_preflight`.
+"""
+from __future__ import annotations
+
+import logging
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+from typing import Callable
+
+_TOOLS_DIR = str(Path(__file__).resolve().parent)
+if _TOOLS_DIR not in sys.path:
+    sys.path.insert(0, _TOOLS_DIR)
+from _loader import (  # noqa: E402  type: ignore[import-not-found]
+    load_sibling as _load_sibling,
+)
+
+_diff_aware_gate = _load_sibling("_diff_aware_gate", "_diff_aware_gate.py")
+
+_logger = logging.getLogger("implementer_gate_preflight")
+
+
+# Env var controlling whether the pre-flight runs at all. Default
+# OFF so the dispatcher's pre-feature behaviour is byte-equivalent.
+# Set ``IMPLEMENTER_DISPATCHER_GATE_PREFLIGHT=1`` to activate.
+PREFLIGHT_ENABLED_ENV_VAR = "IMPLEMENTER_DISPATCHER_GATE_PREFLIGHT"
+
+# Per-invocation timeout for ``local_ci_gate.sh --fast``. 20 minutes
+# matches the R1 ``timeout: 1200000`` recommendation baked into
+# the ``quality-gates`` skill — the first cold-cache run can
+# legitimately take ~5 min and we want to avoid spurious aborts.
+_DEFAULT_GATE_TIMEOUT_SECONDS = 1200
+
+# Wrapper-relative path to the gate script (matches what the worker
+# would also use). ``local_ci_gate.sh`` is in the repo's ``tools/``
+# dir and the dispatcher's CWD is the repo root, but to keep this
+# robust against different CWDs we resolve it at call time.
+_GATE_SCRIPT = "tools/local_ci_gate.sh"
+
+
+def is_preflight_enabled() -> bool:
+    """Return ``True`` when the dispatcher should run the gate
+    pre-flight. Off-by-default."""
+    raw = os.environ.get(PREFLIGHT_ENABLED_ENV_VAR, "").strip().lower()
+    return raw in {"1", "true", "yes", "on"}
+
+
+_TRUTHY = frozenset({"1", "true", "yes", "on"})
+
+
+def _run_gate_once(
+    worktree: Path,
+    runner: Callable[[list[str], dict[str, str], int], subprocess.CompletedProcess] | None = None,
+    *,
+    extra_env: dict[str, str] | None = None,
+) -> tuple[int, str]:
+    """Run ``local_ci_gate.sh --fast`` once against ``worktree``.
+
+    Returns ``(returncode, combined_stdout_stderr)``. Caller decides
+    what to do with the result — this function does not raise on
+    non-zero exit codes (the gate exits 1 on test failure, which is
+    the case we WANT to capture and parse).
+
+    The ``runner`` indirection exists so unit tests can substitute
+    a function that returns scripted output without invoking the
+    real script.
+    """
+    cmd = [_GATE_SCRIPT, "--fast", "--repo-root", str(worktree)]
+    env = {**os.environ, **(extra_env or {})}
+
+    if runner is not None:
+        result = runner(cmd, env, _DEFAULT_GATE_TIMEOUT_SECONDS)
+        # subprocess.CompletedProcess fields: returncode + stdout/stderr
+        out = (result.stdout or "") + (result.stderr or "")
+        return result.returncode, out
+
+    started = time.monotonic()
+    try:
+        result = subprocess.run(
+            cmd,
+            env=env,
+            capture_output=True,
+            text=True,
+            timeout=_DEFAULT_GATE_TIMEOUT_SECONDS,
+            check=False,
+        )
+    except subprocess.TimeoutExpired:
+        elapsed = time.monotonic() - started
+        _logger.warning(
+            "gate pre-flight timed out after %.1fs running %s",
+            elapsed, " ".join(cmd),
+        )
+        return -1, f"<TIMEOUT after {elapsed:.1f}s>"
+    return result.returncode, (result.stdout or "") + (result.stderr or "")
+
+
+def _failed_scenario_keys(gate_output: str) -> set[tuple[str, str]]:
+    """Helper: extract the ``(path, line)`` set of failing scenarios
+    so two runs can be intersected. Mirrors
+    :func:`_diff_aware_gate.parse_failing_scenarios` but returns a
+    set for membership testing."""
+    failures = _diff_aware_gate.parse_failing_scenarios(gate_output)
+    return {(f["path"], f["line"]) for f in failures}
+
+
+def run_preflight(
+    worktree: Path,
+    changed_files: list[str],
+    *,
+    runner: Callable[[list[str], dict[str, str], int], subprocess.CompletedProcess] | None = None,
+) -> dict[str, object]:
+    """Run the gate twice, build a persistent-failure classification.
+
+    Returns the dict shape :func:`_diff_aware_gate.classify_failures`
+    produces, but with an additional ``"runs"`` key carrying per-run
+    summaries:
+
+    .. code-block:: python
+
+        {
+            "gate_statuses": {...},
+            "failures_total": int,
+            "failures_related_to_diff": int,
+            "failures_unrelated_to_diff": int,
+            "related": [...],
+            "unrelated": [...],
+            "runs": [
+                {"run": 1, "returncode": int, "failures": int},
+                {"run": 2, "returncode": int, "failures": int},
+            ],
+            "preflight_enabled": True,
+        }
+
+    Persistence rule: a scenario is "persistent" if it appears in
+    the failure list of BOTH runs (same ``(path, line)`` key).
+    Scenarios that fail run 1 but pass run 2 are considered flaky
+    and are NOT surfaced to the agent.
+
+    When run 1 already passes (returncode 0, no failures), run 2 is
+    skipped — the gates passed and there's nothing to re-verify.
+    """
+    if not is_preflight_enabled():
+        return {
+            "preflight_enabled": False,
+            "failures_total": 0,
+            "failures_related_to_diff": 0,
+            "failures_unrelated_to_diff": 0,
+            "related": [],
+            "unrelated": [],
+            "gate_statuses": {},
+            "runs": [],
+        }
+
+    rc1, out1 = _run_gate_once(worktree, runner=runner)
+    failures1 = _failed_scenario_keys(out1)
+    run_summaries: list[dict[str, object]] = [
+        {"run": 1, "returncode": rc1, "failures": len(failures1)}
+    ]
+
+    if rc1 == 0 and not failures1:
+        # All gates passed on the first attempt — no need to re-run.
+        classification = _diff_aware_gate.classify_failures(
+            out1, changed_files
+        )
+        classification.update({"runs": run_summaries, "preflight_enabled": True})
+        return classification
+
+    rc2, out2 = _run_gate_once(worktree, runner=runner)
+    failures2 = _failed_scenario_keys(out2)
+    run_summaries.append(
+        {"run": 2, "returncode": rc2, "failures": len(failures2)}
+    )
+
+    # Persistent = present in BOTH runs. Synthesize a synthetic
+    # "gate output" containing only the persistent failures so the
+    # classifier doesn't accidentally see flaky ones too. The
+    # synthetic output uses the same shape parse_failing_scenarios
+    # consumes (``features/path.feature:N``).
+    persistent = failures1 & failures2
+    synthetic_lines = [f"  {p}:{n}" for (p, n) in sorted(persistent)]
+    # Preserve the gate-status header from the second run's output
+    # so the classifier still emits a gate_statuses summary.
+    statuses_text = "\n".join(
+        line for line in (out2 or "").splitlines()
+        if any(g in line.lower() for g in (
+            "lint:", "typecheck:", "unit_tests:",
+            "integration_tests:", "e2e_tests:", "coverage:",
+        ))
+    )
+    synthetic_output = (
+        statuses_text
+        + "\n\nFailing scenarios (persistent across two runs):\n"
+        + "\n".join(synthetic_lines)
+    )
+    classification = _diff_aware_gate.classify_failures(
+        synthetic_output, changed_files
+    )
+    classification.update(
+        {"runs": run_summaries, "preflight_enabled": True}
+    )
+    return classification

tools/dispatch_implementer.py

@@ -70,6 +70,15 @@ _implementer_escalation = _load_sibling(
 _implementer_label_state = _load_sibling(
     "_implementer_label_state", "_implementer_label_state.py"
 )
+_implementer_compliance = _load_sibling(
+    "_implementer_compliance", "_implementer_compliance.py"
+)
+_implementer_gate_preflight = _load_sibling(
+    "_implementer_gate_preflight", "_implementer_gate_preflight.py"
+)
+_diff_aware_gate = _load_sibling(
+    "_diff_aware_gate", "_diff_aware_gate.py"
+)
 _opencode_worker = _load_sibling("_opencode_worker", "_opencode_worker.py")
 
 
@@ -445,9 +454,135 @@ def _prefetch_prompt(
                 "PR context sentinel write failed for PR #%s: %s",
                 pr_number, e,
             )
+
+    # Append deterministic-improvement stanzas (plan 2026-05-13).
+    # Both are gated and skipped in dry-run / when no worktree was
+    # materialised. Each appends a self-contained markdown section
+    # the worker can read without changing the existing prompt body.
+    text = _append_deterministic_stanzas(
+        text, cfg, item, group, result, clone_handle,
+    )
     return text
 
 
+def _append_deterministic_stanzas(
+    prompt: str,
+    cfg: Any,
+    item: dict[str, Any],
+    group: Any,
+    result: Any,
+    clone_handle: Any,
+) -> str:
+    """Append the gate-preflight + compliance-gap stanzas to the
+    prefetch prompt when their respective flags are on.
+
+    Both stanzas need the pre-cloned worktree's path — when the
+    preclone is disabled or failed, we skip silently (the worker
+    falls back to its in-session discovery flow). Both are off by
+    default so the legacy prompt is byte-equivalent for operators
+    who haven't opted in.
+
+    Order: gate-preflight stanza first (it tells the agent which
+    test failures matter), then compliance stanza (it tells the
+    agent what to fill in). When the compliance stanza reports
+    "all gaps closed" the agent's exit path is clean even if the
+    gate-preflight surfaced unrelated flakes.
+    """
+    if cfg.dry_run:
+        return prompt
+    worktree = getattr(clone_handle, "path", None)
+    if not worktree:
+        return prompt
+    from pathlib import Path
+    worktree_path = Path(str(worktree))
+    if not worktree_path.exists():
+        return prompt
+
+    pr_number = int(item.get("number") or 0)
+    extras: list[str] = []
+
+    # ─── Gate pre-flight (off-by-default; runs --fast twice) ────
+    if _implementer_gate_preflight.is_preflight_enabled():
+        try:
+            changed_files = _collect_changed_files_from_result(result)
+            classification = _implementer_gate_preflight.run_preflight(
+                worktree_path, changed_files,
+            )
+            stanza = _diff_aware_gate.render_prompt_stanza(classification)
+            if stanza:
+                extras.append(stanza)
+        except Exception as exc:
+            _logger.warning(
+                "gate pre-flight failed for #%s "
+                "(continuing without the stanza): %s",
+                pr_number, exc,
+            )
+
+    # ─── Compliance gap detection (gated on escalation flag) ───
+    if _is_escalation_enabled():
+        try:
+            git_user_email = (
+                os.environ.get("GIT_USER_EMAIL")
+                or getattr(cfg, "git_user_email", "")
+                or ""
+            )
+            gaps = _implementer_compliance.check_compliance_gaps(
+                worktree_path, git_user_email,
+            )
+            stanza = _implementer_compliance.render_prompt_stanza(
+                gaps, pr_number=pr_number or None,
+            )
+            if stanza:
+                extras.append(stanza)
+        except Exception as exc:
+            _logger.warning(
+                "compliance gap detection failed for #%s "
+                "(continuing without the stanza): %s",
+                pr_number, exc,
+            )
+
+    if not extras:
+        return prompt
+    return prompt + "\n\n" + "\n\n".join(extras) + "\n"
+
+
+def _collect_changed_files_from_result(result: Any) -> list[str]:
+    """Best-effort extraction of the PR's changed-file list from the
+    prefetch result. Used by the gate pre-flight to classify gate
+    failures as related/unrelated to the diff.
+
+    The :mod:`_implementer_prefetch` result carries diff bytes; we
+    parse the standard ``diff --git a/path b/path`` headers to
+    enumerate file paths. Returns an empty list when no diff is
+    available — the classifier then treats every failure as
+    "unrelated", which is the safe direction (false negatives don't
+    cause regressions, they just over-report unrelated failures).
+    """
+    diff_text = ""
+    for attr in ("diff", "diff_text", "unified_diff", "pr_diff"):
+        v = getattr(result, attr, None)
+        if isinstance(v, str) and v:
+            diff_text = v
+            break
+        if isinstance(v, bytes):
+            diff_text = v.decode("utf-8", errors="replace")
+            break
+    if not diff_text:
+        return []
+    import re as _re
+    paths: list[str] = []
+    seen: set[str] = set()
+    # Match ``diff --git a/<path> b/<path>`` lines
+    for m in _re.finditer(
+        r"^diff --git a/(\S+) b/\S+", diff_text, _re.MULTILINE,
+    ):
+        p = m.group(1)
+        if p not in seen:
+            seen.add(p)
+            paths.append(p)
+    return paths
+
+
 def _implementation_prompt_dispatch(
     cfg: Any, item: dict[str, Any], group: Any
 ) -> str:
@@ -1190,6 +1325,59 @@ def _terminal_state_from_session(session: Any) -> str:
     return "completed" if session.status == "completed" else session.status
 
 
+# Map worker terminal_state → synthesized outcome string when the
+# worker emitted no parseable JSON. The escalation predicate's
+# UNKNOWN bucket (1 retry then escalate) is the right fallback for
+# "we don't know what happened", but the more common case in
+# production is "worker gave up and narrated a failure without
+# emitting the contract JSON" (gpt-5-mini's failure mode on PR #30,
+# Run 1). Synthesizing a concrete outcome lets ``decide()`` route
+# to ESCALATE-as-competence-failure (skip the wasted same-tier
+# retry) and gives the status-comment fingerprint a non-empty
+# reason string. The synthesis is gated behind escalation so the
+# flag=0 path is byte-equivalent to the pre-feature build.
+_SYNTHESIZED_OUTCOME_FROM_TERMINAL_STATE: dict[str, str] = {
+    "completed": "rebase-failed",
+    "timeout": "timeout",
+    "transport-error": "transport-error",
+}
+
+
+def _synthesize_outcome_if_missing(
+    parsed_json: dict[str, Any] | None,
+    terminal_state: str,
+) -> tuple[dict[str, Any] | None, bool]:
+    """Return ``(possibly-synthesized parsed_json, was_synthesized)``.
+
+    When the worker emitted a parseable JSON with a string
+    ``outcome``, pass through unchanged. Otherwise synthesize a
+    ``{"outcome": "<mapped>", "files_touched": [], "_synthesized":
+    True}`` dict so the escalation predicate sees a competence-class
+    signal instead of falling into the UNKNOWN bucket (which costs
+    one wasted same-tier retry per cycle).
+
+    The ``_synthesized`` flag is captured in the Phase 4 telemetry
+    row so an operator analysing the JSONL sink can distinguish
+    cycles where the worker explicitly reported failure from cycles
+    where the dispatcher synthesised a verdict.
+    """
+    if isinstance(parsed_json, dict):
+        outcome = parsed_json.get("outcome")
+        if isinstance(outcome, str) and outcome:
+            return parsed_json, False
+    synthesized_outcome = _SYNTHESIZED_OUTCOME_FROM_TERMINAL_STATE.get(
+        terminal_state, "unknown"
+    )
+    return (
+        {
+            "outcome": synthesized_outcome,
+            "files_touched": [],
+            "_synthesized": True,
+        },
+        True,
+    )
+
+
 def _post_session_action_with_escalation(
     cfg: Any,
     item: dict[str, Any],
@@ -1331,6 +1519,14 @@ def _post_session_action_with_escalation(
             "status_comment": status_comment,
         }
 
+    # Synthesise an outcome JSON when the worker produced none —
+    # turns the dispatcher's UNKNOWN-bucket waste into a clean
+    # competence-class signal. Recorded on the per-attempt
+    # telemetry row as ``outcome_synthesised=true``.
+    parsed_json, _t0_outcome_synthesised = _synthesize_outcome_if_missing(
+        parsed_json, terminal_state,
+    )
+
     # ─── Initial attempt (cross-cycle resumption aware) ────────
     # ``start_tier`` was seeded by :func:`_implementation_prompt_dispatch`
     # via the ``auto/last-attempt-tier-N`` label fetch. When this
@@ -1425,8 +1621,11 @@ def _post_session_action_with_escalation(
                 redact_values=[cfg.token] if getattr(cfg, "token", None) else [],
             )
             last_parsed = new_session.parsed_json
-            last_raw = new_session.raw_response
             last_terminal_state = _terminal_state_from_session(new_session)
+            last_parsed, _ = _synthesize_outcome_if_missing(
+                last_parsed, last_terminal_state,
+            )
+            last_raw = new_session.raw_response
             last_wallclock = new_session.wallclock_seconds
             last_depth = new_session.subagent_max_depth
         else:
@@ -1474,8 +1673,11 @@ def _post_session_action_with_escalation(
                 redact_values=[cfg.token] if getattr(cfg, "token", None) else [],
             )
             last_parsed = new_session.parsed_json
-            last_raw = new_session.raw_response
             last_terminal_state = _terminal_state_from_session(new_session)
+            last_parsed, _ = _synthesize_outcome_if_missing(
+                last_parsed, last_terminal_state,
+            )
+            last_raw = new_session.raw_response
             last_wallclock = new_session.wallclock_seconds
             last_depth = new_session.subagent_max_depth