cleveragents-core/tools/dispatch_implementer.py

#!/usr/bin/env python3
"""Deterministic dispatcher for implementation workers.

The dispatcher preserves the priority order originally established by
the (now-decommissioned) ``implementation-supervisor.md`` LLM agent:
fix failing PRs first, then PRs with unaddressed review feedback, then
new issue work. The worker remains the LLM boundary; Python owns
queueing, PR claims, watchdogs, and telemetry. The legacy supervisor
agent was deleted on 2026-05-09 — see ``CHANGELOG.md`` and
``docs/development/auto-agents-tier-2-3-plan.md`` for the rationale —
and this dispatcher is now the only orchestrator for implementer work.

Phase 2 (2026-05-09) adds optional pre-fetch parity with
``dispatch_review.py``:

- ``IMPLEMENTER_DISPATCHER_PREFETCH=1`` swaps the legacy title-only
  prompt for the rich pre-fetched prompt assembled in
  :mod:`_implementer_prompt`. The worker reads PR description, diff,
  CI status, comments, active REQUEST_CHANGES reviews, linked issues,
  and the Epic body without issuing a single in-session GET.
- ``IMPLEMENTER_DISPATCHER_PRECLONE=1`` (Phase 3) additionally
  pre-clones the PR's HEAD into ``/tmp/cleveragents-implementer-worktrees/``
  so the worker can ``edit`` against the worktree without spawning
  ``git-isolator-util``.

Both flags default to ``0`` (legacy behaviour) until a Phase 4
acceptance run flips them to default-on. The substrate is built; the
flags ungate adoption per the plan in
``docs/development/auto-agents-tier-2-3-plan.md``.
"""
from __future__ import annotations

import argparse
import logging
import os
import sys
from pathlib import Path
from typing import Any

_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,
)

_dispatch = _load_sibling("_dispatch_runtime", "_dispatch_runtime.py")
_claim_runtime = _load_sibling("_claim_runtime", "_claim_runtime.py")
_review_fetch = _load_sibling("_review_fetch", "_review_fetch.py")
_pr_clone = _load_sibling("_pr_clone", "_pr_clone.py")
_pr_diff = _load_sibling("_pr_diff", "_pr_diff.py")
_review_post = _load_sibling("_review_post", "_review_post.py")
_phase4_telemetry = _load_sibling(
    "_phase4_telemetry", "_phase4_telemetry.py"
)
_implementer_prefetch = _load_sibling(
    "_implementer_prefetch", "_implementer_prefetch.py"
)
_implementer_metadata_classifier = _load_sibling(
    "_implementer_metadata_classifier", "_implementer_metadata_classifier.py"
)
_implementer_prompt = _load_sibling(
    "_implementer_prompt", "_implementer_prompt.py"
)
_pr_context_sentinel = _load_sibling(
    "_pr_context_sentinel", "_pr_context_sentinel.py"
)
_worker_infra_seed = _load_sibling(
    "_worker_infra_seed", "_worker_infra_seed.py"
)
_implementer_escalation = _load_sibling(
    "_implementer_escalation", "_implementer_escalation.py"
)
_implementer_label_state = _load_sibling(
    "_implementer_label_state", "_implementer_label_state.py"
)
_implementer_compliance = _load_sibling(
    "_implementer_compliance", "_implementer_compliance.py"
)
_implementer_compliance_apply = _load_sibling(
    "_implementer_compliance_apply", "_implementer_compliance_apply.py"
)
_recent_push_cache = _load_sibling(
    "_recent_push_cache", "_recent_push_cache.py"
)
_implementer_gate_preflight = _load_sibling(
    "_implementer_gate_preflight", "_implementer_gate_preflight.py"
)
_opencode_worker = _load_sibling("_opencode_worker", "_opencode_worker.py")


_logger = logging.getLogger("dispatch_implementer")


DRIVER_NAME = "dispatch_implementer.py"
CLAIM_KIND = "implementer"

PREFETCH_ENV_VAR = "IMPLEMENTER_DISPATCHER_PREFETCH"
PRECLONE_ENV_VAR = "IMPLEMENTER_DISPATCHER_PRECLONE"

# In-cycle tier escalation (2026-05-12). When ON, the dispatcher runs
# the worker for Tier 0; if the worker fails in a way that escalation
# can help (see ``_implementer_escalation.decide``), the dispatcher
# re-runs the worker at Tier 1 within the same cycle — holding the
# claim throughout, refreshing TTL between attempts, resetting the
# worktree to the prefetched head_sha, and mutating the
# ``auto/last-attempt-tier-N`` labels for operator visibility. See
# ``docs/development/implementer-in-cycle-escalation-plan.md``.
#
# Default OFF: when unset (or falsy), the dispatcher runs in the
# pre-feature single-shot mode, byte-equivalent to the behaviour
# shipped before this commit.
ESCALATION_ENABLED_ENV_VAR = "IMPLEMENTER_ESCALATION_ENABLED"

# Tier 2 (``tier-2`` slot) kill-switch. Consulted only when
# ``IMPLEMENTER_ESCALATION_ENABLED`` is ON. **Default since
# 2026-05-12: ON.** Earlier iterations gated Tier 2 behind an opt-in
# flag, but that defeats the purpose of escalation — if Tier 1 fails,
# the PR stays stuck without trying the next model. The flag is now
# an opt-OUT kill-switch: set
# ``IMPLEMENTER_ESCALATION_TIER2_ENABLED=0`` to stop escalation at
# Tier 1 (used as a panic button if the Tier 2 model's wallclock
# turns out to be operationally painful in practice). The specific
# model behind the ``tier-2`` slot is configured in
# ``.opencode/models/tiers.yaml`` and can be swapped without code
# changes — the kill-switch is slot-based, not model-based.
#
# Operational impact: worst-case cycle wallclock rises from
# ``2 × worker_timeout_seconds`` (Tier 0→1 only) to
# ``3 × worker_timeout_seconds`` (~6 h). The claim's TTL refresh
# at each tier boundary keeps the merge-driver sweep from
# reclaiming the PR mid-cycle.
ESCALATION_TIER2_ENABLED_ENV_VAR = "IMPLEMENTER_ESCALATION_TIER2_ENABLED"

# Outcome-synthesis kill-switch. The escalation loop synthesises a
# concrete outcome (rebase-failed / timeout / transport-error) when
# the worker dies before emitting JSON, so the escalation predicate
# sees a competence-class signal instead of falling into the UNKNOWN
# bucket.
#
# **Default since 2026-05-13: ON when escalation is enabled.** Split
# from ``IMPLEMENTER_ESCALATION_ENABLED`` so an operator who suspects
# the synthesis is mis-routing a particular failure class can disable
# it without turning off the whole escalation feature. Set
# ``IMPLEMENTER_OUTCOME_SYNTHESIS=0`` to revert to the legacy
# "UNKNOWN-bucket wastes one retry" behaviour.
#
# **Operational cost when set to 0:** the escalation predicate sees
# UNKNOWN, which costs one wasted same-tier retry per cycle (the
# worker is re-spawned with the same tier hint and almost always
# fails the same way). Worst-case cycle wallclock rises by one
# worker_timeout (typically ~7200 s). Operators flipping this off
# should expect the cost and treat the flag as a panic-button, not
# a default-state.
OUTCOME_SYNTHESIS_ENV_VAR = "IMPLEMENTER_OUTCOME_SYNTHESIS"

# Compliance-gap detection kill-switch. The compliance section is
# computed by :func:`_compute_deterministic_sections` and serialised
# into the sentinel for the worker to read via
# ``implementer_pr_context.py --field compliance_gaps``.
#
# **Default since 2026-05-13: ON whenever escalation is enabled.**
# Set ``IMPLEMENTER_COMPLIANCE_GAPS_ENABLED=0`` to opt out.
#
# **Gating semantics: AND-only.** The compliance scan only runs when
# BOTH ``IMPLEMENTER_ESCALATION_ENABLED`` is on AND this flag is
# non-falsy. Setting this flag ON with escalation OFF does NOT
# enable the scan in isolation — compliance is meaningless outside
# the escalation loop's gap-filling flow (the worker has no retry
# path to fix discovered gaps). This flag is a kill-switch for the
# escalation-on case, not an independent feature toggle.
COMPLIANCE_GAPS_ENABLED_ENV_VAR = "IMPLEMENTER_COMPLIANCE_GAPS_ENABLED"

# Auto-fix kill switch. When this flag is non-falsy, the dispatcher
# applies trivial PR-compliance fixes (CONTRIBUTORS line, CHANGELOG
# stub, ISSUES CLOSED footer) deterministically without spawning the
# LLM worker. The worker is still called when there's real code work
# to do (failing tests, request-changes review feedback). Off-by-
# default because giving the dispatcher commit/push authority is a
# real responsibility expansion and operators should opt in.
#
# When this flag is OFF (default), the worker handles compliance
# fixes the way it has historically — slower and more expensive but
# with the LLM in the loop as a safety check.
#
# Operational consequences when set to 1:
# - ~5-13 min of LLM time saved per "metadata-only PR" cycle
# - Dispatcher pushes commits authored by the configured
#   ``GIT_USER_NAME`` / ``GIT_USER_EMAIL`` identity
# - Cycle archive's ``post_session_result`` gains an
#   ``auto_fix_report`` entry describing what was applied
AUTO_FIX_COMPLIANCE_ENV_VAR = "IMPLEMENTER_DISPATCHER_AUTO_FIX_COMPLIANCE"

# Estimator-driven adaptive tier selection (G11 harvest 2026-05-15).
# Default OFF. When OFF, every first-attempt invocation runs at
# tier-0 (the cheap-to-default slot) byte-for-byte to the pre-G11
# build: the dispatcher emits an explicit
# ``escalation_tier_hint: `0` `` line and ``tier-dispatcher`` skips
# the estimator. When ON, the dispatcher OMITS the hint on true
# first attempts (no prior ``auto/last-attempt-tier-N`` label, no
# in-cycle escalation already active), so ``tier-dispatcher`` falls
# through to ``estimator-implementation`` and a confident
# capability estimate decides the starting tier (-1..2). Cross-cycle
# resumption and in-cycle escalation continue to short-circuit the
# estimator by emitting their respective hints — the estimator is
# only consulted on TRUE first attempts.
ESTIMATOR_ENABLED_ENV_VAR = "IMPLEMENTER_ESTIMATOR_ENABLED"

# W8 harvest (2026-05-15) — loud-signal flag for degraded prompt
# assembly. When ON, the dispatcher emits a WARN log line naming
# the specific missing sections at the end of prompt assembly so an
# operator tailing the dispatcher log sees the silent-degradation
# signal without having to inspect telemetry JSONL. Default OFF
# preserves today's quiet behaviour byte-for-byte. The completeness
# signal is ALWAYS stashed on the per-item context (regardless of
# this flag) so future telemetry / status-comment paths can pick it
# up without needing operators to toggle the flag.
DEGRADED_PROMPT_LOG_ENV_VAR = "IMPLEMENTER_DEGRADED_PROMPT_LOG_ENABLED"


def _work_type_for_group(group_name: str) -> str:
    """Map dispatcher work-group names to the ``work_type`` constant
    that appears in the worker prompt header (see
    :func:`_implementer_prompt.assemble_header`). Kept here rather
    than imported because the prompt builder derives this from
    ``group.item_kind``; the sentinel needs the same value but
    receives the group name only.

    Falls back to ``"pr_fix"`` for unrecognised groups so a future
    work-group addition doesn't crash the sentinel write — the worst
    that happens is the sentinel reports a wrong ``work_type``,
    which the worker can sanity-check against its prompt header.
    """
    if group_name == "new_issue":
        return "issue_impl"
    return "pr_fix"


_TRUTHY_ENV_VALUES = frozenset({"1", "true", "yes", "on"})


def _env_truthy(name: str) -> bool:
    return os.environ.get(name, "").strip().lower() in _TRUTHY_ENV_VALUES


_FALSY_ENV_VALUES = frozenset({"0", "false", "no", "off"})


def _env_falsy_explicit(name: str) -> bool:
    """``True`` when ``name`` is set to one of the falsy literals.

    Used by :func:`_is_prefetch_enabled` and
    :func:`_is_preclone_enabled` to support the default-ON / opt-out
    model (set ``IMPLEMENTER_DISPATCHER_PREFETCH=0`` to explicitly
    disable the rich prompt path for one cycle). An unset variable is
    NOT falsy here — it falls through to the default.
    """
    return os.environ.get(name, "").strip().lower() in _FALSY_ENV_VALUES


def _is_prefetch_enabled() -> bool:
    """Return ``True`` when the dispatcher should use the rich
    pre-fetched prompt (PR description, diff, CI status, comments,
    linked issues, etc.).

    **Default since 2026-05-10: ON.** The post-mortem of the
    PR #30 implementer runs showed that without prefetch the
    ``task-implementor`` subagent burns ~3-4 minutes on
    ``curl`` + ``read`` calls just to discover the PR's head SHA,
    base branch, and changed files — all of which the dispatcher
    already has from its own Forgejo calls. The rich prompt
    eliminates those turns entirely. The legacy title-only prompt
    remains available behind an explicit
    ``IMPLEMENTER_DISPATCHER_PREFETCH=0`` opt-out (e.g. for
    bisecting a regression against the pre-Phase-4 behaviour).
    """
    if _env_falsy_explicit(PREFETCH_ENV_VAR):
        return False
    return True


def _is_escalation_enabled() -> bool:
    """Return ``True`` when the in-cycle tier escalation loop is on.

    Default OFF (per the plan's safety contract). When OFF, the
    dispatcher's post_session_action runs the legacy single-shot
    behaviour byte-equivalent to the pre-feature build. When ON for
    PR-shaped items, the post_session_action delegates to
    :func:`_post_session_action_with_escalation`, which observes the
    worker's outcome and, when escalation can help, re-dispatches at
    the next tier without releasing the implementer claim. See
    ``docs/development/implementer-in-cycle-escalation-plan.md``.
    """
    return _env_truthy(ESCALATION_ENABLED_ENV_VAR)


def _is_degraded_prompt_log_enabled() -> bool:
    """W8 harvest (2026-05-15): default OFF. When set to a truthy
    literal, the dispatcher emits a WARN log line naming the
    missing prefetch sections after each cycle where prefetch
    returned incomplete data."""
    return _env_truthy(DEGRADED_PROMPT_LOG_ENV_VAR)


def _is_implementer_estimator_enabled() -> bool:
    """Return ``True`` when first-attempt tier selection should defer
    to ``estimator-implementation`` instead of hard-coding tier-0.

    Default OFF. When OFF, the dispatcher emits an explicit
    ``escalation_tier_hint: `0` `` line on every true first attempt
    so the worker's ``tier-dispatcher`` short-circuits to the
    default slot — byte-for-byte to today's behaviour. When ON, the
    dispatcher OMITS the hint on first attempts (no
    ``auto/last-attempt-tier-N`` label and no in-cycle escalation
    seeding ``start_tier > 0``), letting ``tier-dispatcher`` run the
    estimator and choose an adaptive tier in ``-1`` .. ``2``.

    See ``docs/development/final-working-harvest-plan.md`` (G11).
    """
    return _env_truthy(ESTIMATOR_ENABLED_ENV_VAR)


def _is_outcome_synthesis_enabled() -> bool:
    """Return ``True`` when the dispatcher should synthesise an
    outcome JSON for workers that exited without emitting one.

    Default-ON when ``IMPLEMENTER_OUTCOME_SYNTHESIS`` is unset; set
    to a falsy literal (``0`` / ``false`` / ``no`` / ``off``) to
    revert to legacy behaviour. Independent of
    :func:`_is_escalation_enabled` so the kill-switch can be flipped
    without disabling the whole escalation feature. The legacy path
    (``_post_session_action`` for non-PR work / when escalation is
    off) never calls the synthesiser regardless of this flag.
    """
    if _env_falsy_explicit(OUTCOME_SYNTHESIS_ENV_VAR):
        return False
    return True


def _is_compliance_gaps_enabled() -> bool:
    """Return ``True`` when the dispatcher should run the compliance
    gap detector and serialise the result into the sentinel.

    Default-ON when ``IMPLEMENTER_COMPLIANCE_GAPS_ENABLED`` is unset
    (and escalation is enabled — the caller's responsibility); set
    to a falsy literal to opt out. The split from
    :func:`_is_escalation_enabled` lets an operator disable just
    the compliance scan if e.g. ``git`` is unreliable on the
    worktree, without losing tier escalation.
    """
    if _env_falsy_explicit(COMPLIANCE_GAPS_ENABLED_ENV_VAR):
        return False
    return True


def _is_auto_fix_compliance_enabled() -> bool:
    """Return ``True`` when the dispatcher should apply deterministic
    compliance fixes (CONTRIBUTORS line, CHANGELOG stub, ISSUES CLOSED
    footer) without spawning the LLM worker.

    **Default OFF.** Enabling this gives the dispatcher commit/push
    authority for trivial PR-hygiene changes. Set
    ``IMPLEMENTER_DISPATCHER_AUTO_FIX_COMPLIANCE=1`` to opt in.

    See :data:`AUTO_FIX_COMPLIANCE_ENV_VAR` for full rationale.
    """
    return _env_truthy(AUTO_FIX_COMPLIANCE_ENV_VAR)


def _is_tier2_enabled() -> bool:
    """Return ``True`` when Tier 2 (the ``tier-2`` slot) is part of
    the escalation ladder.

    Consulted only when :func:`_is_escalation_enabled` is True.
    **Default since 2026-05-12: ON.** The flag is now an opt-OUT
    kill-switch — set ``IMPLEMENTER_ESCALATION_TIER2_ENABLED=0`` to
    cap the ladder at Tier 1 (used as a panic button if whatever
    model currently occupies the ``tier-2`` slot proves
    operationally painful). The earlier opt-IN default defeated the
    purpose of escalation because a Tier-1 failure would leave the
    PR stuck without ever trying the next model.
    """
    return not _env_falsy_explicit(ESCALATION_TIER2_ENABLED_ENV_VAR)


def _max_tier_for_cycle() -> int:
    """Maximum tier index the escalation loop will walk.

    Returns ``2`` (Tier 0 → 1 → 2) by default. Returns ``1`` only
    when an operator has explicitly set the Tier-2 kill-switch
    (``IMPLEMENTER_ESCALATION_TIER2_ENABLED=0``).
    """
    return 2 if _is_tier2_enabled() else 1


def _is_preclone_enabled() -> bool:
    """Return ``True`` when the dispatcher should pre-clone the PR's
    head into ``/tmp/<repo>`` and pass the working-copy path to the
    worker.

    **Default since 2026-05-10: ON.** Symmetric to
    :func:`_is_prefetch_enabled`: the pre-clone eliminates the
    ``git-isolator-util`` subagent's clone-then-checkout dance for
    ``pr_fix`` work (the typical ~60-100 s of network + git I/O).
    The worker's prompt carries the ``## Pre-cloned working copy``
    section with a working ``repo_dir`` path so the worker proceeds
    directly to the patch step. Set
    ``IMPLEMENTER_DISPATCHER_PRECLONE=0`` to fall back to the
    in-session clone via ``git-isolator-util``.
    """
    if _env_falsy_explicit(PRECLONE_ENV_VAR):
        return False
    return True


# ─── Legacy prompt (preserved for ``IMPLEMENTER_DISPATCHER_PREFETCH=0``) ────


def _legacy_implementation_prompt(
    cfg: Any, item: dict[str, Any], group: Any
) -> str:
    """Title-only prompt produced by the dispatcher before Phase 2.

    Kept available so an operator who flips
    ``IMPLEMENTER_DISPATCHER_PREFETCH=0`` (Phase 4 rollback) gets the
    same prompt shape the worker historically saw — Forgejo GETs +
    ``git-isolator-util`` clones inside the session.

    Header assembly is delegated to
    :func:`_implementer_prompt.assemble_header` so a future tweak to
    the per-section heading wording (forgejo_url / work_type
    template, claim notes, PR Compliance Checklist) lands in one
    place and propagates to both the legacy and prefetch branches.
    The output-JSON contract is appended verbatim from
    :data:`_implementer_prompt.OUTPUT_CONTRACT`. The final ``\\n`` is
    preserved for byte-equivalence with the historical snapshot
    (operators relying on ``--once`` log greps key off the trailing
    newline).

    The header builder + claim-note constants are public symbols on
    :mod:`_implementer_prompt`; cross-module access is intentional
    rather than a privacy pierce.
    """
    claim_note = (
        _implementer_prompt.PR_CLAIM_NOTE
        if group.item_kind != "issue"
        else _implementer_prompt.ISSUE_CLAIM_NOTE
    )
    header = _implementer_prompt.assemble_header(cfg, item, group, claim_note)
    return f"{header}\n\n{_implementer_prompt.OUTPUT_CONTRACT}\n"


# ─── Pre-fetch + pre-clone prompt (Phase 2 + 3 path) ────────────────────────


def _head_ref_from_result(result: Any) -> str:
    """Pick the PR's head-branch short name out of the prefetch
    result without raising when any layer of the
    ``pr_details.head.ref`` chain is absent. Returns ``""`` when
    the dispatcher's PR-details fetch failed; the empty branch
    field is harmless (worker falls back to its prompt-derived
    name)."""
    pr_details = getattr(result, "pr_details", None) or {}
    head = pr_details.get("head") if isinstance(pr_details, dict) else None
    if not isinstance(head, dict):
        return ""
    ref = head.get("ref")
    return ref if isinstance(ref, str) else ""


def _build_clone_section(
    cfg: Any,
    pr_number: int,
    head_sha: str,
    *,
    head_ref: str = "",
) -> tuple[str, Any]:
    """Run the Phase 3 pre-clone (when ``IMPLEMENTER_DISPATCHER_PRECLONE=1``)
    and return ``(section_text, clone_handle)``.

    ``head_ref`` is the PR's head-branch short name, sourced from
    the dispatcher's freshly-fetched ``pr_details.head.ref``. It
    flows through to the workspace handoff sentinel's ``branch``
    field so the worker can ``git push`` back to the right branch
    without an extra Forgejo GET. Empty string is fine — the
    sentinel records an empty branch and the worker falls back to
    its prompt-derived branch name.

    On dry-run / disabled / failure: returns the no-handle stanza
    text and a ``None`` handle. The dispatcher's finally block treats
    ``None`` as "nothing to clean up" and the worker prompt's
    ``## Pre-cloned working copy`` section text already documents
    the fall-through to ``git-isolator-util``.

    Why this lives here rather than in :mod:`_implementer_prompt`:
    the prompt builder is pure (no I/O), so it can be unit-tested
    without monkeypatching git / the network. The clone is a side
    effect that needs to live in the dispatcher driver where the
    cleanup hook also lives.
    """
    if cfg.dry_run or not _is_preclone_enabled() or not head_sha:
        return _pr_diff.build_clone_section(None, head_sha or ""), None
    handle = _pr_clone.prepare_pr_worktree(
        cfg, pr_number, head_sha, head_ref=head_ref, kind="implementer"
    )
    return _pr_diff.build_clone_section(handle, head_sha or ""), handle


def _rebuild_prompt_from_cached_result(
    cfg: Any,
    item: dict[str, Any],
    group: Any,
    cached_result: Any,
    cached_clone: Any,
) -> str | None:
    """Re-render the prompt text from a cached prefetch result.

    B3 (2026-05-13): the escalation loop's ESCALATE branch was
    re-running the entire prefetch + gate-preflight + compliance
    scan for every tier transition (~7 min per transition on a
    comment-heavy PR). The prefetch result and det_sections don't
    change between tiers within a single cycle — the worktree
    reset puts everything back at the prefetched head_sha — so
    rebuild the prompt TEXT from cached pieces and skip the I/O.

    Returns the prompt string on cache hit; ``None`` if the
    cached pieces are incomplete (caller falls back to a full
    prefetch).

    Note: the sentinel write + det_sections compute also runs
    inside the cached path so a Tier 1+ attempt sees up-to-date
    on-disk handoff. The expensive part — Forgejo fetches,
    bare-mirror refresh, gate-preflight — is what we skip.
    """
    if cached_result is None:
        return None
    clone_section = ""
    if cached_clone is not None:
        # Same shape as _build_clone_section but without re-cloning.
        try:
            clone_section = _pr_diff.build_clone_section(
                cached_clone,
                getattr(cached_result, "head_sha", "") or "",
            )
        except Exception as exc:
            _logger.warning(
                "B3 cache path: failed to render clone section from "
                "cached handle; falling through to full prefetch: %s",
                exc,
            )
            return None
    if group.name == "failing_ci_pr":
        text = _implementer_prompt.build_pr_fix_prompt(
            cfg, item, group, cached_result, clone_section,
        )
    elif group.name == "request_changes_pr":
        text = _implementer_prompt.build_request_changes_prompt(
            cfg, item, group, cached_result, clone_section,
        )
    elif group.name == "new_issue":
        text = _implementer_prompt.build_new_issue_prompt(
            cfg, item, group, cached_result,
        )
    else:
        return None

    # Append the deterministic-sections stanzas the same way the
    # full prefetch path does. The cached det_sections live on
    # the item context (stashed during the original cycle's
    # prefetch).
    existing_context = item.get("_dispatcher_implementer_context") or {}
    det_sections = existing_context.get("_deterministic_sections", {})
    text = _append_deterministic_stanzas(
        text, cfg, item, group, cached_result, cached_clone,
        det_sections=det_sections,
    )
    return text


def _prefetch_prompt(
    cfg: Any, item: dict[str, Any], group: Any
) -> str:
    """Run the appropriate per-work-group fetcher and assemble the
    rich prompt. Stamps ``item["_dispatcher_implementer_context"]``
    so a future ``post_session_action`` (Phase 5b operator-status
    comments) can read the same context without re-fetching.

    Falls back to the legacy prompt ONLY in dry-run mode (for unit
    tests) — otherwise the dispatcher must produce a prefetch-shaped
    prompt every cycle when the flag is enabled. We do NOT silently
    degrade to the legacy prompt on fetch failure: that produces an
    underspecified prompt the worker will paper over.

    The clone handle is stamped onto the item alongside the prefetch
    result so the dispatcher's finally block can clean it up.
    """
    pr_number = int(item.get("number") or 0)
    # B3 (2026-05-13): when this is a re-entry from the escalation
    # loop's ESCALATE branch (Tier N → N+1), the prefetch result and
    # det_sections are already on the item context — re-running the
    # entire prefetch + gate-preflight + compliance scan would burn
    # ~7 minutes per tier transition on top of comment-cache wins
    # from C5. Cached path: when ``_dispatcher_implementer_context``
    # already has a populated ``result`` AND the worktree is still
    # at the original head_sha (worktree-reset between tiers makes
    # this invariant hold), just rebuild the prompt text from the
    # cached pieces and skip the I/O.
    existing_context = item.get("_dispatcher_implementer_context")
    if isinstance(existing_context, dict) and existing_context.get("result"):
        cached_result = existing_context["result"]
        cached_clone = existing_context.get("clone_handle")
        cached_text = _rebuild_prompt_from_cached_result(
            cfg, item, group, cached_result, cached_clone,
        )
        if cached_text is not None:
            _logger.info(
                "prefetch cache hit for PR #%s — reused prefetch result + "
                "det_sections from prior attempt (B3)",
                pr_number,
            )
            return cached_text

    # Per-cycle worker-infra seed: copy ``tools/local_ci_gate.sh`` and
    # the implementer-side helper scripts into ``/tmp/local_tools/``
    # (defaults; see ``_worker_infra_seed.py``). The worker invokes
    # them via absolute ``/tmp/local_tools/...`` paths because the
    # fork's master branch does NOT carry these files — keeping them
    # in a separate /tmp tree means the worker never accidentally
    # ``git add``s them into the PR. Seeding is best-effort: a copy
    # failure logs WARNING and the worker falls back to its own
    # diagnostic path (cwd doesn't contain the script → exit 2 from
    # ``local_ci_gate.sh``, reported verbatim in the attempt comment).
    # Gated by ``WORKER_INFRA_DISABLE=1`` for tests; the autouse
    # pytest fixture sets that env var so unit tests don't write
    # to ``/tmp/local_tools`` on the developer's box.
    if not cfg.dry_run:
        try:
            result_seed = _worker_infra_seed.seed_worker_infra()
            if not result_seed.skipped:
                _logger.info(
                    "worker-infra seed: copied %d file(s) (%d bytes) "
                    "from %s to %s",
                    result_seed.files_copied,
                    result_seed.bytes_copied,
                    result_seed.source_root,
                    result_seed.dest_dir,
                )
        except Exception as e:
            # Best-effort: a seed failure should not crash the
            # dispatcher; the worker will report the missing-script
            # error verbatim on its first invocation attempt.
            _logger.warning(
                "worker-infra seed failed (worker will fall back to "
                "cwd-relative invocation, which exits 2 on a fresh "
                "clone): %s", e,
            )
    if group.name in ("failing_ci_pr",):
        result = _implementer_prefetch.fetch_pr_fix_context(cfg, item)
        clone_section, clone_handle = _build_clone_section(
            cfg, pr_number, result.head_sha,
            head_ref=_head_ref_from_result(result),
        )
        text = _implementer_prompt.build_pr_fix_prompt(
            cfg, item, group, result, clone_section
        )
    elif group.name in ("request_changes_pr",):
        result = _implementer_prefetch.fetch_request_changes_context(cfg, item)
        clone_section, clone_handle = _build_clone_section(
            cfg, pr_number, result.head_sha,
            head_ref=_head_ref_from_result(result),
        )
        text = _implementer_prompt.build_request_changes_prompt(
            cfg, item, group, result, clone_section
        )
    elif group.name in ("new_issue",):
        result = _implementer_prefetch.fetch_new_issue_context(cfg, item)
        # No PR exists yet — no clone, no head_sha.
        clone_handle = None
        text = _implementer_prompt.build_new_issue_prompt(cfg, item, group, result)
    else:
        # Unknown group — preserve safety by falling back to legacy.
        _logger.warning(
            "_prefetch_prompt: unknown group=%s; falling back to legacy prompt",
            group.name,
        )
        return _legacy_implementation_prompt(cfg, item, group)
    item["_dispatcher_implementer_context"] = {
        "result": result,
        "clone_handle": clone_handle,
    }
    # PR-context sentinel: write a parallel on-disk handoff so a
    # downstream subagent (typically ``task-implementor`` at depth 3
    # of the ``task`` tool chain) can re-read the prefetched API data
    # via ``tools/implementer_pr_context.py`` instead of either
    # (a) hoping the prompt sections survived the intermediate tier
    # agents' summarisation, or (b) re-issuing the Forgejo curls.
    # Best-effort: a write failure logs WARNING and the worker falls
    # back to the prompt content (or to curl).
    #
    # Dry-run gate: dry-run is the operator-visible "preview, no I/O"
    # contract. The fetchers themselves already short-circuit in
    # dry-run and return empty results — writing a sentinel for an
    # empty result would (a) violate the no-I/O invariant and (b)
    # leave a stale empty-shaped sentinel on disk that the next real
    # cycle would have to overwrite. The workspace handoff is
    # gated upstream in ``_build_clone_section``; this is the matching
    # gate for the PR-context handoff.
    # Compute deterministic sections (gate-preflight + compliance
    # gaps) BEFORE the sentinel write so they land in the same
    # payload the worker reads via ``implementer_pr_context.py``.
    # ``_compute_deterministic_sections`` returns a dict keyed by
    # ``compliance_gaps`` / ``gate_preflight`` (each value either a
    # plain dict or ``None``). Skipped in dry-run / when no worktree
    # was materialised.
    det_sections = _compute_deterministic_sections(
        cfg, item, group, result, clone_handle,
    )
    # Stash on the item context so the short-circuit logic
    # (``_maybe_short_circuit``) can read the deterministic-section
    # results without re-running compliance / preflight scans.
    context = item.get("_dispatcher_implementer_context")
    if isinstance(context, dict):
        context["_deterministic_sections"] = det_sections
        # G1 harvest (2026-05-15): record the metadata-only
        # classification on the context so cycle telemetry and any
        # future grooming-diversion path can observe how often the
        # heuristic fires on real PRs. No behaviour change here —
        # the dispatcher still spawns the worker for every cycle
        # regardless of this flag. Actual diversion is deferred
        # until either a dedicated grooming driver consumes
        # ``tools/groom_label_inference``'s rule engine or the
        # implementer-worker prompt is updated to take a no-clone
        # fast path on the sentinel.
        try:
            context["metadata_only_candidate"] = _classify_metadata_only(
                item, result,
            )
        except Exception as exc:  # noqa: BLE001 — best-effort observability
            _logger.warning(
                "metadata-only classification raised for #%s: %s",
                pr_number, exc,
            )
            context["metadata_only_candidate"] = False

        # W8 harvest (2026-05-15): translate the prefetch carrier's
        # per-section completion flags into a single
        # ``{degraded, missing_sections, error_kinds}`` signal and
        # stash it on the context so downstream telemetry / status
        # comments can surface it. When
        # ``IMPLEMENTER_DEGRADED_PROMPT_LOG_ENABLED=1`` (default OFF)
        # AND the cycle is degraded, also emit a WARN log naming
        # the specific missing sections so an operator tailing the
        # dispatcher log can see the silent-degradation signal
        # without parsing JSONL telemetry.
        try:
            completeness = _implementer_prefetch.assess_prompt_completeness(
                result,
            )
            context["prompt_completeness"] = completeness
            if completeness["degraded"] and _is_degraded_prompt_log_enabled():
                _logger.warning(
                    "prompt assembly DEGRADED for #%s — missing "
                    "sections=%s; error_kinds=%s. Cycle will proceed "
                    "(crash-safety) but the worker may emit a "
                    "lower-quality verdict on partial context.",
                    pr_number,
                    completeness["missing_sections"],
                    completeness["error_kinds"],
                )
        except Exception as exc:  # noqa: BLE001 — best-effort observability
            _logger.warning(
                "prompt completeness assessment raised for #%s: %s",
                pr_number, exc,
            )

    # C4: look up any recent dispatcher push for THIS head_sha so
    # we can warn the worker not to re-do the same fix.
    recent_push = None
    head_sha = getattr(result, "head_sha", "") or ""
    if head_sha:
        try:
            recent_push = _recent_push_cache.lookup_recent_push(
                pr_number, head_sha,
            )
        except Exception as exc:
            _logger.warning(
                "recent-push cache lookup raised for #%s: %s",
                pr_number, exc,
            )

    if not cfg.dry_run:
        try:
            _pr_context_sentinel.write(
                pr_number=pr_number,
                work_type=_work_type_for_group(group.name),
                work_group=group.name,
                result=result,
                item=item,
                compliance_gaps=det_sections.get("compliance_gaps"),
                gate_preflight=det_sections.get("gate_preflight"),
                recent_implementer_push=recent_push,
            )
        except Exception as e:  # noqa: BLE001 — best-effort
            _logger.warning(
                "PR context sentinel write failed for PR #%s: %s",
                pr_number, e,
            )

    # Append a brief sentinel-pointer stanza + (when the flags are
    # on) a short summary the agent can react to even if upstream
    # tier agents summarise the prompt body. The authoritative copy
    # lives in the sentinel; the prompt-side stanza is a fallback +
    # nudge to read the sentinel.
    text = _append_deterministic_stanzas(
        text, cfg, item, group, result, clone_handle,
        det_sections=det_sections,
    )
    return text


def _compute_deterministic_sections(
    cfg: Any,
    item: dict[str, Any],
    group: Any,
    result: Any,
    clone_handle: Any,
) -> dict[str, Any]:
    """Run gate-preflight + compliance-gap detection and return a
    dict with the two keys ``gate_preflight`` and ``compliance_gaps``
    (each value is the section dict or absent).

    Run-order rationale: gate-preflight first because the dispatcher
    pays its wallclock cost (two ``--fast`` runs) regardless of what
    compliance reports. Compliance is sub-100-ms and never blocks.

    Skipped in dry-run / when no worktree was materialised — both
    sections need the pre-cloned worktree. Returns an empty dict in
    those cases (caller does not write either key into the sentinel).
    """
    out: dict[str, Any] = {}
    if cfg.dry_run:
        return out
    worktree = getattr(clone_handle, "path", None)
    if not worktree:
        return out
    from pathlib import Path
    worktree_path = Path(str(worktree))
    if not worktree_path.exists():
        return out

    pr_number = int(item.get("number") or 0)

    # ─── 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,
            )
            if isinstance(classification, dict):
                # P3: cross-check local preflight against the
                # dispatcher's prefetched remote CI state. If the
                # local --fast gates say "all clean" but remote CI
                # is failing, the failing job is something --fast
                # doesn't exercise (e.g. e2e_tests or coverage).
                # Surface the divergence so the worker doesn't trust
                # preflight alone — without this, the worker reads
                # "preflight clean" + "compliance clean" and emits
                # `resolved` while CI is still red (the 2026-05-13
                # PR #30 attempt 1 / PR #28 cycle 2 failure mode).
                remote_ci_state = _extract_remote_ci_state(result)
                classification["remote_ci_state"] = remote_ci_state
                preflight_clean = (
                    classification.get("failures_total", 0) == 0
                    and not classification.get("preflight_timeout", False)
                )
                if preflight_clean and remote_ci_state == "failure":
                    classification["diverges_from_remote_ci"] = True
                else:
                    classification["diverges_from_remote_ci"] = False
                out["gate_preflight"] = classification
        except Exception as exc:
            _logger.warning(
                "gate pre-flight failed for #%s "
                "(continuing without the section): %s",
                pr_number, exc,
            )

    # ─── Compliance gap detection ──────────────────────────────
    # Two-flag gate: requires the escalation feature to be on
    # (compliance is meaningless outside the escalation loop's
    # gap-filling flow) AND the per-feature opt-out
    # ``IMPLEMENTER_COMPLIANCE_GAPS_ENABLED`` to be non-falsy
    # (default-ON). The split lets an operator disable just the
    # compliance scan without losing tier escalation — useful
    # when ``git`` is misbehaving on the pre-cloned worktree.
    if _is_escalation_enabled() and _is_compliance_gaps_enabled():
        try:
            git_user_email = (
                os.environ.get("GIT_USER_EMAIL")
                or getattr(cfg, "git_user_email", "")
                or ""
            )
            gaps, masked_checks = (
                _implementer_compliance.check_compliance_gaps_with_masking(
                    worktree_path, git_user_email,
                )
            )
            if isinstance(gaps, dict):
                out["compliance_gaps"] = {
                    "gaps": gaps,
                    "pr_number": pr_number or None,
                    "git_user_email": git_user_email,
                    # Masking-aware count: a downstream consumer
                    # reading just ``gaps_open_count`` on a
                    # fully-masked worktree would otherwise see 0
                    # (because masked checks return True) and read
                    # "all clean" on a tree the dispatcher couldn't
                    # actually inspect. Passing the masked set
                    # excludes those keys from the count entirely.
                    "gaps_open_count": _implementer_compliance.gaps_open_count(
                        gaps, masked_checks=masked_checks,
                    ),
                    # The masked-checks list lets the renderer hedge
                    # the "all passed" verdict when git itself failed
                    # on one or more checks. Sorted for deterministic
                    # serialisation in the sentinel.
                    "masked_checks": sorted(masked_checks),
                }
        except Exception as exc:
            _logger.warning(
                "compliance gap detection failed for #%s "
                "(continuing without the section): %s",
                pr_number, exc,
            )

    return out


def _append_deterministic_stanzas(
    prompt: str,
    cfg: Any,
    item: dict[str, Any],
    group: Any,
    result: Any,
    clone_handle: Any,
    *,
    det_sections: dict[str, Any] | None = None,
) -> str:
    """Append a brief sentinel-pointer stanza (plus, when the
    matching flag is on, a condensed inline summary) for each
    deterministic section the dispatcher computed.

    Why a short stanza vs. the full markdown
    ----------------------------------------
    The previous design embedded the full classification + per-gap
    hints in the prompt body. ``task-implementor.md:305`` is explicit
    that intermediate tier agents routinely summarise non-diff
    sections away before the worker sees them. The sentinel survives
    because the worker reads it directly off disk.

    This stanza is intentionally short (one paragraph + one bullet
    list per section) so its survival probability through tier-agent
    summarisation is high. It tells the agent: (a) the read command,
    (b) the headline counts, (c) the per-gap hints. The full
    classification stays in the sentinel where the agent can drill 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 compliance reports "all gaps
    closed" the agent's exit path is clean even if 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

    # When the caller did not pre-compute, do it here so legacy
    # callers (and test fixtures that hand-roll a clone_handle)
    # still produce a fully-populated stanza. Production callers
    # in :func:`_prefetch_prompt` pass ``det_sections`` so the same
    # computation runs once per cycle and lands in BOTH the sentinel
    # and the prompt — no double work. An earlier iteration emitted
    # a WARNING on this path to catch production-callsite regression;
    # removed because (a) the sole production caller is one line of
    # code, easily audited, and (b) the noise leaked into tests and
    # broke ``pytest -W error`` setups.
    if det_sections is None:
        det_sections = _compute_deterministic_sections(
            cfg, item, group, result, clone_handle,
        )

    pr_number = int(item.get("number") or 0)
    extras: list[str] = []

    # ─── Gate pre-flight summary (always before compliance) ─────
    # ``_compute_deterministic_sections`` only sets ``gate_preflight``
    # when ``is_preflight_enabled()`` is True; ``run_preflight`` then
    # always returns at least ``{"preflight_enabled": True, ...}``
    # so the dict-shape check alone is sufficient. (An earlier
    # iteration OR'd over three truthy fields as defensive coverage
    # against an unreachable caller — removed for clarity.)
    preflight = det_sections.get("gate_preflight")
    if isinstance(preflight, dict):
        extras.append(_render_preflight_pointer_stanza(preflight, pr_number))

    # ─── Compliance summary (always after gate-preflight) ───────
    # P8 fix (2026-05-13): the compliance renderer needs the
    # preflight payload to decide whether "all compliance gaps
    # closed" is sufficient grounds to direct emit-resolved. When
    # preflight shows related-to-diff failures, divergence from
    # remote CI, or its own timeout, the green-path directive is
    # actively misleading — every run-2 disputed-resolved attempt
    # hit this. ``preflight`` is None when the section wasn't
    # computed this cycle (flag off / dry-run); the renderer treats
    # that as "no preflight signal to block on" and behaves exactly
    # as before.
    compliance = det_sections.get("compliance_gaps")
    if isinstance(compliance, dict) and isinstance(
        compliance.get("gaps"), dict
    ):
        masked_checks = compliance.get("masked_checks") or []
        extras.append(_render_compliance_pointer_stanza(
            compliance["gaps"], pr_number,
            masked_checks=list(masked_checks),
            preflight=preflight if isinstance(preflight, dict) else None,
        ))

    if not extras:
        return prompt
    return prompt + "\n\n" + "\n\n".join(extras) + "\n"


def _render_preflight_pointer_stanza(
    preflight: dict[str, Any], pr_number: int,
) -> str:
    """Brief gate-preflight summary + sentinel-read hint.

    Surfaces preflight-timeout explicitly: the underlying
    :func:`_implementer_gate_preflight._run_gate_once` returns
    ``preflight_timeout=True`` on subprocess.TimeoutExpired. Without
    surfacing it here, the renderer would emit a clean "no failures"
    summary on a silent timeout (the previous bug).

    Defensive short-circuit: ``preflight_enabled=False`` is the
    documented disabled-shape from :func:`run_preflight`. Production
    never lands here (``_compute_deterministic_sections`` already
    gates on ``is_preflight_enabled()`` before computing), but a
    future caller writing the disabled-shape into the sentinel
    would otherwise trigger the misleading "no persistent failures"
    green-light line. Treat disabled-shape as "no stanza."
    """
    if not preflight.get("preflight_enabled"):
        return ""
    statuses = preflight.get("gate_statuses") or {}
    failures_total = int(preflight.get("failures_total") or 0)
    related = preflight.get("related") or []
    unrelated = preflight.get("unrelated") or []
    timeout = bool(preflight.get("preflight_timeout"))
    lines = ["## Pre-flight gate summary (read sentinel for detail)"]
    lines.append("")
    if timeout:
        # Timeout payloads zero the counts in the orchestrator so
        # the renderer can't double-message; this stanza tells the
        # worker the classification is suppressed and why.
        lines.append(
            "**Pre-flight timed out** — the dispatcher could not "
            "deterministically classify failures this cycle. Treat "
            "every gate failure you see in-session as potentially real."
        )
        lines.append(
            "_Classification suppressed: persistent-failure counts "
            "are not reported on timeout._"
        )
    if statuses:
        roll = ", ".join(
            f"`{g}`={s}" for g, s in sorted(statuses.items())
        )
        lines.append(f"Gate roll-up: {roll}")
    if failures_total and not timeout:
        # Only emit the count outside the timeout path. The
        # timeout branch suppresses counts on purpose; this guard
        # is a belt-and-braces defence against a future caller
        # bypassing the orchestrator's zeroing.
        lines.append(
            f"Persistent failures: **{failures_total}** "
            f"(related-to-diff={len(related)}, unrelated={len(unrelated)})."
        )
    elif not timeout:
        lines.append(
            "No persistent failures across two pre-flight runs — you "
            "do NOT need to re-run `local_ci_gate.sh --fast` unless "
            "you change files."
        )
    # P8/P3 surface (2026-05-13): when local preflight passes but
    # remote CI says "failure", the failing job is something --fast
    # doesn't run (e2e_tests, coverage_report). Without this line
    # the worker reads "preflight clean" and (combined with a clean
    # compliance check) emits resolved — every run-2 cycle hit this.
    # The sentinel always carried ``diverges_from_remote_ci`` since
    # P3 landed; this just surfaces it in the prose so the worker
    # sees it without having to read the sentinel.
    if preflight.get("diverges_from_remote_ci") and not timeout:
        lines.append(
            "**Pre-flight diverges from remote CI** "
            "(`diverges_from_remote_ci=true`): local `--fast` gates "
            "pass but the PR's remote CI is failing. The failing "
            "check is something `--fast` does not exercise (likely "
            "`e2e_tests` or `coverage_report`). Read "
            f"`--field ci` to identify the failing check and address "
            "it. **Do NOT emit `{\"outcome\": \"resolved\"}` until "
            "remote CI is green.**"
        )
    elif (
        preflight.get("remote_ci_state") == "failure"
        and failures_total
        and not timeout
    ):
        # Both local and remote are red — note the remote state so
        # the worker isn't tempted to treat the unrelated locals as
        # the whole picture.
        lines.append(
            "Remote CI is also failing — check `--field ci` for "
            "which specific checks remote reports as failed before "
            "deciding whether your local-related failures are the "
            "whole story."
        )
    lines.append(
        "Full classification (per-scenario paths + run metadata) is "
        "in the PR-context sentinel: "
        f"`python3 tools/implementer_pr_context.py read --pr {pr_number} "
        "--field gate_preflight`"
    )
    return "\n".join(lines)


def _render_compliance_pointer_stanza(
    gaps: dict[str, bool], pr_number: int,
    *,
    masked_checks: list[str] | None = None,
    preflight: dict[str, Any] | None = None,
) -> str:
    """Brief compliance summary + sentinel-read hint.

    Mirrors :func:`_implementer_compliance.render_prompt_stanza` but
    condensed: this is the SHORT version that survives tier-agent
    summarisation. The full per-gap hints are still rendered by the
    compliance module's own helper if the worker chooses to fetch
    them.

    ``masked_checks`` carries the list of check names where the
    underlying git call failed and the function fell back to the
    "treat as passing" policy. When non-empty the renderer
    explicitly downgrades the "all checks passed" verdict to "all
    OBSERVABLE checks passed; N were masked because git failed"
    and refuses to direct the worker to exit. Without this hedge
    the worker would receive a confident `resolved` directive on
    a worktree the dispatcher couldn't actually inspect — see
    plan v4 §"Compliance masking footgun".

    ``preflight`` is the optional gate-preflight section dict (same
    shape as ``run_preflight`` returns). When provided, the renderer
    consults it to decide whether the green-path "emit resolved"
    directive is safe — if preflight shows related-to-diff failures,
    divergence from remote CI, or its own timeout, the directive is
    SUPPRESSED and replaced with an explicit "do not emit resolved
    until preflight is clean" line. This is the P8 fix (2026-05-13):
    every run-2 disputed-resolved attempt was caused by emitting the
    green directive while preflight was reporting failure. ``None``
    (the default) preserves pre-fix behaviour for callers that don't
    have a preflight payload (legacy fixtures, dispatcher cycles
    where preflight didn't run).
    """
    masked_checks = masked_checks or []
    all_closed = _implementer_compliance.all_gaps_closed(gaps)
    missing = [k for k, v in gaps.items() if not v]
    # P8: derive the "preflight blocks emit-resolved" predicate.
    # When preflight is absent / disabled / not provided, the
    # predicate is False and the renderer behaves as before.
    preflight_block_reasons = _compliance_preflight_block_reasons(preflight)
    lines = ["## Compliance gap report (read sentinel for hints)"]
    lines.append("")
    if all_closed and not masked_checks and not preflight_block_reasons:
        lines.append(
            "**All compliance checks passed.** The PR is complete — "
            "do NOT re-apply or re-edit the code fix. Verify quality "
            "gates and emit `{\"outcome\": \"resolved\", \"files_touched\": []}`."
        )
    elif all_closed and not masked_checks and preflight_block_reasons:
        # P8 fix branch: compliance is clean but preflight has
        # evidence the PR is NOT actually done. Spell out exactly
        # what's wrong so the worker can act on it. The wording
        # explicitly forbids emit-resolved and explains the cost
        # of emitting it anyway (the dispatcher will catch the
        # lie via ``outcome_disputed=True`` and escalate the tier).
        joined = "; also ".join(preflight_block_reasons)
        lines.append(
            "**Compliance items are complete, but " + joined + ".**"
        )
        lines.append(
            "Do NOT emit `{\"outcome\": \"resolved\"}` yet — fix the "
            "underlying failure(s), verify the quality gates pass, "
            "then commit and push the real fix. Emitting `resolved` "
            "without advancing HEAD will be detected by the "
            "dispatcher as a disputed-resolved and the work will be "
            "escalated to the next tier (wasting your model's "
            "budget)."
        )
    elif all_closed and masked_checks:
        # Hedge: every check that COULD be evaluated passed, but
        # one or more underlying ``git`` calls failed so the
        # dispatcher could not actually inspect those signals.
        # Do NOT direct the worker to exit — the masked check
        # could be hiding a real gap.
        joined_masked = ", ".join(f"`{m}`" for m in masked_checks)
        lines.append(
            "**All OBSERVABLE compliance checks passed, but "
            f"{len(masked_checks)} check(s) were MASKED** because "
            f"`git` itself failed: {joined_masked}. Do NOT exit "
            "with `resolved` — re-verify the masked check(s) "
            "in-session (run `git status` / `git log -1`) before "
            "deciding whether the PR is actually complete."
        )
    else:
        joined = ", ".join(f"`{m}`" for m in missing)
        lines.append(f"Gaps to fill ({len(missing)}): {joined}.")
        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."
        )
        if masked_checks:
            joined_masked = ", ".join(f"`{m}`" for m in masked_checks)
            lines.append(
                f"_Note: {len(masked_checks)} check(s) were masked "
                f"because git failed: {joined_masked}. The values "
                "shown for those keys may be stale — re-verify in-session._"
            )
    lines.append(
        "Per-gap hints + ground-truth dict are in the PR-context "
        f"sentinel: `python3 tools/implementer_pr_context.py read "
        f"--pr {pr_number} --field compliance_gaps`"
    )
    return "\n".join(lines)


def _compliance_preflight_block_reasons(
    preflight: dict[str, Any] | None,
) -> list[str]:
    """Return a list of human-readable reasons preflight blocks the
    "emit resolved" directive, or empty when nothing blocks.

    P8 fix (2026-05-13). Used by
    :func:`_render_compliance_pointer_stanza` to decide whether the
    green-path verdict is safe. Three independent block conditions
    are returned as separate strings so the renderer can join them
    with semicolons; each is phrased to continue a sentence that
    starts with "Compliance items are complete, but …".

    Block conditions:

    - **related-to-diff preflight failures** — the local gates show
      failures whose paths intersect the PR's diff. Highest-confidence
      signal that HEAD is not actually fixed.
    - **diverges_from_remote_ci** — local preflight passes but
      remote CI reports failure. The failing check is something
      ``--fast`` doesn't exercise (e2e_tests, coverage_report).
    - **preflight_timeout** — the dispatcher couldn't classify
      anything because the gate run timed out; local state is
      unverified.

    ``None`` / ``preflight_enabled=False`` / empty dict all return
    an empty list (no signal to act on).
    """
    if not isinstance(preflight, dict):
        return []
    if not preflight.get("preflight_enabled"):
        return []
    reasons: list[str] = []
    if preflight.get("preflight_timeout"):
        reasons.append(
            "pre-flight timed out (`preflight_timeout=true`) — local "
            "gate state is unverified this cycle. Re-run "
            "`bash /tmp/local_tools/tools/local_ci_gate.sh --fast "
            "--repo-root {repo_dir}` in-session before deciding"
        )
        # When preflight timed out the related/divergence signals
        # are zeroed by the orchestrator — don't double-message.
        return reasons
    related = preflight.get("related") or []
    if related:
        reasons.append(
            f"pre-flight reports **{len(related)}** failure(s) "
            "**related to this diff** — the code in HEAD is not "
            "actually passing the gates (look at `--field gate_preflight` "
            "for the per-scenario paths)"
        )
    if preflight.get("diverges_from_remote_ci"):
        reasons.append(
            "**local pre-flight passes but remote CI is failing** "
            "(`diverges_from_remote_ci=true`) — the failing job is "
            "something `--fast` does not exercise (`e2e_tests` or "
            "`coverage_report`). Fetch the failing check from "
            "`--field ci` and address it"
        )
    return reasons


def _extract_remote_ci_state(result: Any) -> str:
    """Pull the remote CI aggregate state from the prefetch result.

    The dispatcher's :mod:`_implementer_prefetch` fetches the
    PR's HEAD CI status; this just normalises the value into one
    of ``"success"``, ``"failure"``, ``"pending"``, or ``"unknown"``.
    Used by :func:`_compute_deterministic_sections` (P3) to
    cross-check the local preflight result against what remote CI
    actually says.
    """
    ci_status = getattr(result, "ci_status", None)
    if isinstance(ci_status, dict):
        state = ci_status.get("state")
        if isinstance(state, str) and state:
            return state.lower()
    if isinstance(ci_status, str) and ci_status:
        return ci_status.lower()
    return "unknown"


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


# C2 harvest (2026-05-15): the classifier itself lives in
# :mod:`_implementer_metadata_classifier` to start the long-overdue
# decomposition of this 3.2k-line driver toward the ~500-line module
# budget the rest of ``tools/`` honours. The leading-underscore alias
# below preserves the in-module name so existing tests continue to
# call ``driver._classify_metadata_only(...)`` without churn.
_classify_metadata_only = _implementer_metadata_classifier.classify_metadata_only


def _maybe_short_circuit(
    cfg: Any, item: dict[str, Any], context: dict[str, Any],
) -> None:
    """Decide whether the dispatcher can satisfy this PR's cycle
    without spawning the LLM worker, and if so stash a synthetic
    ``SessionResult`` on ``context["_short_circuit_result"]`` for
    the dispatch runtime to consume.

    Two short-circuit paths:

    1. **P0 — skip-when-green.** If compliance scan says all checks
       passed, gate-preflight says no failures, AND remote CI status
       is ``success``, then this PR is genuinely done and the
       listing script's "failing CI" snapshot was stale (CI flipped
       between snapshot and claim). Emit ``no_changes_needed`` and
       release. Fires whenever escalation is on and conditions hold.

    2. **A — auto-fix-compliance.** If compliance scan reports gaps
       that the dispatcher can fix deterministically (CONTRIBUTORS,
       CHANGELOG bullet, ISSUES CLOSED footer), gate-preflight is
       clean, AND remote CI failure looks attributable only to those
       missing items, then apply the fixes + push + emit
       ``resolved`` synthetically. Gated by
       ``IMPLEMENTER_DISPATCHER_AUTO_FIX_COMPLIANCE=1`` because it
       gives the dispatcher commit/push authority.

    Both paths are no-ops when:
    - dry-run
    - escalation feature disabled
    - the item isn't PR-shaped
    - the prefetch result is missing
    - any required signal (compliance, preflight, CI) is absent or
      ambiguous
    """
    if cfg.dry_run:
        return
    if not _is_escalation_enabled():
        return
    if not isinstance(item.get("head"), dict):
        return
    if not isinstance(context, dict):
        return
    result = context.get("result")
    if result is None:
        return
    clone_handle = context.get("clone_handle")
    if clone_handle is None or not getattr(clone_handle, "path", None):
        return

    pr_number = int(item.get("number") or 0)
    remote_ci_state = _extract_remote_ci_state(result)

    # Pull the deterministic sections we already computed for the
    # sentinel. Re-running them would double the compute cost; both
    # are stashed on the context by ``_prefetch_prompt``.
    det_sections = context.get("_deterministic_sections") or {}
    compliance = det_sections.get("compliance_gaps") or {}
    preflight = det_sections.get("gate_preflight") or {}
    gaps: dict[str, bool] = compliance.get("gaps") or {}
    masked_checks = compliance.get("masked_checks") or []
    if not gaps:
        # No compliance scan ran — be conservative and dispatch
        # the worker. Most commonly this means the preclone was
        # disabled or failed.
        return

    preflight_clean = bool(
        preflight
        and preflight.get("failures_total", 0) == 0
        and not preflight.get("preflight_timeout", False)
    )
    compliance_clean = (
        all(bool(v) for v in gaps.values()) and not masked_checks
    )

    # ─── P0: everything green ──────────────────────────────────
    if (
        compliance_clean
        and preflight_clean
        and remote_ci_state == "success"
    ):
        _logger.info(
            "P0 short-circuit for PR #%s: compliance clean + preflight "
            "clean + remote CI success — skipping LLM worker",
            pr_number,
        )
        context["_short_circuit_result"] = {
            "status": "completed",
            "wallclock_seconds": 0.0,
            "session_id": "",
            "raw_response": (
                '{"outcome": "no_changes_needed", "files_touched": [], '
                '"_dispatcher_short_circuit": "P0"}'
            ),
            "parsed_json": {
                "outcome": "no_changes_needed",
                "files_touched": [],
                "_dispatcher_short_circuit": "P0",
            },
        }
        return

    # ─── A: auto-fix compliance gaps ───────────────────────────
    # Only fires when (a) the operator opted in via the flag, (b)
    # preflight is clean (so failing CI is plausibly about
    # compliance-shaped checks, not test failures), (c) compliance
    # has at least one gap, (d) none of the gaps are masked (we
    # don't have reliable signal to fix what we can't see).
    if (
        not _is_auto_fix_compliance_enabled()
        or masked_checks
        or compliance_clean
        or not preflight_clean
    ):
        return

    # Don't auto-fix when remote CI is passing — there's nothing to
    # fix in that case (P0 above would have caught the truly green
    # case; this guards against weird-state cycles where CI is
    # green but the dispatcher's compliance scan found something).
    if remote_ci_state == "success":
        return

    worktree = Path(str(clone_handle.path))
    pr_title = ""
    pr_details = getattr(result, "pr_details", None)
    if isinstance(pr_details, dict):
        pr_title = str(pr_details.get("title") or "")
    if not pr_title:
        pr_title = str(item.get("title") or f"PR #{pr_number}")

    linked = getattr(result, "linked_issues", None) or []
    linked_issue_numbers: list[int] = []
    for li in linked:
        if isinstance(li, dict):
            n = li.get("number")
            if isinstance(n, int) and n > 0:
                linked_issue_numbers.append(n)

    git_user_name = (
        os.environ.get("GIT_USER_NAME")
        or getattr(cfg, "git_user_name", "")
        or "CleverThis"
    )
    git_user_email = (
        os.environ.get("GIT_USER_EMAIL")
        or getattr(cfg, "git_user_email", "")
        or "hal9000@cleverthis.com"
    )
    branch = ""
    if isinstance(pr_details, dict):
        head = pr_details.get("head") or {}
        if isinstance(head, dict):
            branch = str(head.get("ref") or "")
    if not branch:
        _logger.warning(
            "auto-fix short-circuit for PR #%s aborted: missing head_ref "
            "(can't push without the branch name); falling through to worker",
            pr_number,
        )
        return

    try:
        report = _implementer_compliance_apply.apply_compliance_fixes(
            worktree, gaps,
            git_user_name=git_user_name,
            git_user_email=git_user_email,
            pr_title=pr_title,
            pr_number=pr_number,
            linked_issue_numbers=linked_issue_numbers,
        )
    except Exception as exc:
        _logger.warning(
            "auto-fix apply raised for PR #%s; falling through to worker: %s",
            pr_number, exc,
        )
        return

    if not report.all_applied:
        _logger.info(
            "auto-fix could not deterministically resolve all gaps for PR "
            "#%s (%d failures); falling through to worker",
            pr_number,
            sum(1 for r in report.per_gap if r.error),
        )
        return

    if not report.any_committed:
        # All gaps were already satisfied (idempotent re-run). Push
        # is unnecessary — the worktree is the same as origin.
        _logger.info(
            "auto-fix found no actual gaps to fix on PR #%s after recheck "
            "(idempotent); short-circuit emits no_changes_needed",
            pr_number,
        )
        context["_short_circuit_result"] = {
            "status": "completed",
            "raw_response": (
                '{"outcome": "no_changes_needed", "files_touched": [], '
                '"_dispatcher_short_circuit": "A_idempotent"}'
            ),
            "parsed_json": {
                "outcome": "no_changes_needed",
                "files_touched": [],
                "_dispatcher_short_circuit": "A_idempotent",
            },
        }
        return

    # Log the apply phase BEFORE the push so a subsequent push failure
    # still leaves a trace of what the deterministic applier did.
    _logger.info(
        "auto-fix for PR #%s applied %d compliance fix(es), committed %s — "
        "pushing to branch %s",
        pr_number,
        sum(1 for r in report.per_gap if r.applied and not r.error),
        report.final_head_sha[:12] if report.final_head_sha else "<unknown>",
        branch,
    )

    # Push the deterministic commit. The pre-cloned worktree's ``origin``
    # is a bare HTTPS URL with no embedded credentials (see
    # ``_pr_clone._clone_url``), so the push MUST run under the
    # credential-bearing git env — the GIT_ASKPASS shim that supplies
    # the Forgejo PAT, identical to the env the pre-clone/fetch already
    # use. Without it the push fails rc=128 "could not read Username"
    # and the auto-fix silently falls through to the worker.
    push_result = _implementer_compliance_apply.push_branch(
        worktree, branch,
        git_env=_pr_clone._git_env(cfg),
    )
    if not push_result.applied:
        _logger.warning(
            "auto-fix push failed for PR #%s (%s); falling through to worker",
            pr_number, push_result.error,
        )
        return

    _logger.info(
        "auto-fix short-circuit for PR #%s: applied %d compliance fix(es) "
        "+ pushed %s",
        pr_number,
        sum(1 for r in report.per_gap if r.applied and not r.error),
        report.final_head_sha[:12] if report.final_head_sha else "<unknown>",
    )
    context["_short_circuit_result"] = {
        "status": "completed",
        "raw_response": (
            '{"outcome": "resolved", "files_touched": ["CHANGELOG.md", '
            '"CONTRIBUTORS.md"], "_dispatcher_short_circuit": "A_auto_fix", '
            '"_auto_fix_head_sha": "' + report.final_head_sha + '"}'
        ),
        "parsed_json": {
            "outcome": "resolved",
            "files_touched": [
                r.gap_name for r in report.per_gap
                if r.applied and not r.error
            ],
            "_dispatcher_short_circuit": "A_auto_fix",
            "_auto_fix_head_sha": report.final_head_sha,
        },
    }
    # Stash the apply report for downstream telemetry / cycle archive
    # consumers that want the audit trail.
    context["_auto_fix_report"] = {
        "all_applied": report.all_applied,
        "any_committed": report.any_committed,
        "final_head_sha": report.final_head_sha,
        "per_gap": [
            {
                "gap_name": r.gap_name,
                "applied": r.applied,
                "error": r.error,
            }
            for r in report.per_gap
        ],
    }


def _implementation_prompt_dispatch(
    cfg: Any, item: dict[str, Any], group: Any
) -> str:
    """Top-level prompt factory honouring two flag families:

    - ``IMPLEMENTER_DISPATCHER_PREFETCH`` — ``=1`` produces the
      rich pre-fetched prompt; anything else produces the legacy
      title-only prompt.
    - ``IMPLEMENTER_ESCALATION_ENABLED`` — when ``=1`` AND the item
      is a PR (not a new_issue), the dispatcher reads
      ``auto/last-attempt-tier-N`` from Forgejo and seeds
      ``start_tier``. When ``start_tier > 0``, the prompt carries
      an explicit ``escalation_tier_hint: N`` line so the worker
      jumps straight to the right model — closing the cross-cycle
      resumption loop the 2026-05-12 review pointed at. The seeded
      tier is also stashed on ``item["_dispatcher_implementer_context"]``
      so :func:`_post_session_action_with_escalation` knows where
      the cycle started without re-fetching the labels.
    """
    if _is_prefetch_enabled():
        base_prompt = _prefetch_prompt(cfg, item, group)
    else:
        base_prompt = _legacy_implementation_prompt(cfg, item, group)

    estimator_enabled = _is_implementer_estimator_enabled()
    is_pr_shape = isinstance(item.get("head"), dict)

    # When in-cycle escalation is OFF or the item is not a PR there
    # is no escalation context to manage and no ``release_claim_on_exit``
    # directive to emit. The estimator flag still controls whether
    # the worker should fall through to ``estimator-implementation``
    # on the first attempt (G11 harvest 2026-05-15).
    if not _is_escalation_enabled() or not is_pr_shape:
        if estimator_enabled:
            # OMIT the hint so ``tier-dispatcher`` runs the estimator
            # and chooses an adaptive tier.
            return base_prompt
        # Default OFF: emit an explicit hint=0 so the worker passes
        # it through and ``tier-dispatcher`` short-circuits to
        # ``tier-0``. Byte-equivalent to today's flag-OFF prompt
        # after the worker's no-hint→default-0 fallback is removed
        # (see ``implementation-worker.md`` G11 edit).
        return f"{base_prompt}\n\nescalation_tier_hint: `0`\n"

    pr_number = int(item.get("number") or 0)
    start_tier = _read_start_tier_from_labels(cfg, pr_number)

    # Stash on the per-item context so the post-session action can
    # initialise its loop counter without re-fetching labels.
    # ``_prefetch_prompt`` already places ``_dispatcher_implementer_context``
    # on the item; this just adds a key. Defensive against unknown
    # work groups whose prompt branches did not populate the
    # context (the ``_prefetch_prompt`` fallback path).
    context = item.get("_dispatcher_implementer_context")
    if not isinstance(context, dict):
        context = {}
        item["_dispatcher_implementer_context"] = context
    context["start_tier"] = start_tier

    # Tell the worker to skip its session-end claim release.
    # When escalation is on, the dispatcher holds the claim across
    # all tier attempts in this cycle and the dispatch_one finally
    # block does the actual release. Without this directive, the
    # worker's release between tiers leaves a 1–3 s window where
    # another driver could grab the PR. See plan #5.
    extras = ["release_claim_on_exit: false"]
    if start_tier > 0:
        # Cross-cycle resumption: ``auto/last-attempt-tier-N`` labels
        # name the next tier explicitly. The estimator cannot override
        # label-driven resumption — even when ``estimator_enabled`` is
        # True the hint is emitted so the worker resumes at the
        # correct tier.
        extras.append(f"escalation_tier_hint: `{start_tier}`")
    elif not estimator_enabled:
        # Estimator OFF on a true first attempt (no resumption labels):
        # emit the explicit hint=0 fallback so the worker short-circuits
        # to ``tier-0`` — byte-for-byte the legacy behaviour.
        extras.append("escalation_tier_hint: `0`")
    # else: estimator ON, first attempt — OMIT the hint so the worker
    # falls through to ``estimator-implementation``.
    final_prompt = f"{base_prompt}\n\n" + "\n".join(extras) + "\n"

    # P0 / A: deterministic short-circuit.
    # If the prefetch + deterministic sections + remote CI state all
    # agree that no LLM work is needed (or that the only needed work
    # is mechanical compliance fixes the dispatcher can apply itself),
    # stash a synthetic SessionResult on the context. The dispatch
    # runtime's short-circuit hook consumes it and skips the worker.
    # Off by default (gated on IMPLEMENTER_DISPATCHER_AUTO_FIX_COMPLIANCE
    # for the auto-fix path; the no-op skip case fires whenever
    # escalation is on and conditions are met).
    _maybe_short_circuit(cfg, item, context)
    return final_prompt


# ─── Post-session action: cleanup pre-cloned worktree ───────────────────────


# Terminal states for which the dispatcher posts an operator-status
# comment on the PR timeline (Phase 5b). ``unknown`` is intentionally
# omitted: the dispatcher genuinely doesn't know what happened, and
# posting "Worker session ended without verdict" on a PR that may
# have produced a fine fix the next cycle picks up causes operator-
# noise spam (each ``unknown`` cycle's ``outcome_reason`` differs in
# wall-clock metadata, so the fingerprint dedup at
# ``post_implementer_status_comment`` does not catch the repeats).
_NON_PUSHING_TERMINAL_STATES = frozenset({
    "timeout",
    "transport-error",
})


def _should_post_status(
    parsed_json: dict[str, Any] | None, terminal_state: str
) -> tuple[bool, str, str]:
    """Decide whether the dispatcher should post an operator-status
    comment for this cycle, and what (outcome, reason) tuple to use.

    Returns ``(post, outcome, reason)``:

    - ``post=True`` when the cycle ended without a successful PR
      push (timeout, transport error, worker JSON ``rebase-failed``,
      or no parsed JSON at all). The implementer worker's success
      path posts its own attempt comment, so we don't double-post
      on ``outcome=resolved``.
    - ``post=False`` for the success path or for already-claimed /
      claim-failed cycles where there is nothing useful to surface
      to the operator.
    """
    if terminal_state in _NON_PUSHING_TERMINAL_STATES:
        return True, terminal_state, "Worker session ended without verdict"
    if isinstance(parsed_json, dict):
        outcome = str(parsed_json.get("outcome") or "")
        if outcome and outcome != "resolved":
            # Most commonly ``rebase-failed`` per the worker's output
            # contract. Surface the worker's own outcome verbatim so
            # the fingerprint dedup handles repeat failures cleanly.
            files_touched = parsed_json.get("files_touched")
            reason = (
                f"Worker reported outcome={outcome!r}; "
                f"files_touched={files_touched}"
            )
            return True, outcome, reason
    # ``completed`` + ``outcome=resolved`` (or no JSON, but a
    # ``completed`` terminal state) is the worker's success path. The
    # worker already posts its own attempt comment; the dispatcher
    # does not need to post a status comment.
    return False, "", ""


def _resolve_work_group_name(
    item: dict[str, Any], explicit: str | None = None
) -> str:
    """Return the precise work-group name for ``item`` (e.g.
    ``failing_ci_pr`` / ``request_changes_pr`` / ``new_issue``).

    Production path: ``_dispatch_runtime.dispatch_one`` passes the
    canonical name explicitly via the ``work_group_name`` keyword
    argument; ``explicit`` is non-None and we return it verbatim.

    Test / direct-invocation fallback: when the helper is called
    without an explicit name (older tests, direct unit-test
    invocation), fall back to the item-shape heuristic. The
    heuristic returns *canonical* group names — ``failing_ci_pr``
    for PR-shaped items (the priority-zero bucket; the conservative
    default since fixing failing CI is the dispatcher's first
    responsibility) and ``new_issue`` for issue-shaped items. We
    deliberately avoid invented names like ``pr_work`` so the
    Phase 4 fixture and decision-gate thresholds (which key off the
    canonical names) cannot end up with a fallback row that does
    not match any threshold.
    """
    if isinstance(explicit, str) and explicit:
        return explicit
    head = item.get("head")
    return "failing_ci_pr" if isinstance(head, dict) else "new_issue"


def _resolve_pre_session_head_sha(
    item: dict[str, Any], context: dict[str, Any] | None
) -> str:
    """Return the freshly-fetched head_sha from the prefetch result,
    falling back to ``item.head.sha`` for the legacy path. Empty
    string when nothing is available — the telemetry row carries
    that as ``head_sha_advanced=None``.
    """
    if isinstance(context, dict):
        result_obj = context.get("result")
        if hasattr(result_obj, "head_sha"):
            sha = str(result_obj.head_sha or "")
            if sha:
                return sha
    head = item.get("head")
    if isinstance(head, dict):
        return str(head.get("sha") or "")
    return ""


def _fetch_post_session_head_sha(cfg: Any, pr_number: int) -> str:
    """Best-effort GET ``/repos/.../pulls/{n}`` after the worker
    session to detect a head_sha advance (i.e. the worker pushed at
    least one commit). Returns empty string on failure / dry-run /
    issue items.

    Phase 4 telemetry's ``head_sha_advanced`` keys off this. We
    deliberately do NOT raise — a transient 5xx during shutdown
    must not orphan the claim release.
    """
    if cfg.dry_run or pr_number <= 0:
        return ""
    try:
        response = _claim_runtime.get(
            f"/repos/{cfg.owner}/{cfg.repo}/pulls/{int(pr_number)}", cfg
        )
    except Exception as exc:
        _logger.warning(
            "post-session head_sha fetch failed for #%s: %s",
            pr_number,
            exc,
        )
        return ""
    if int(response.get("status") or 0) != 200:
        return ""
    body = response.get("body") or {}
    head = body.get("head") if isinstance(body, dict) else None
    if isinstance(head, dict):
        return str(head.get("sha") or "")
    return ""


def _record_phase4_telemetry(
    cfg: Any,
    item: dict[str, Any],
    *,
    parsed_json: dict[str, Any] | None,
    raw_response: str,
    terminal_state: str,
    context: dict[str, Any] | None,
    work_group_name: str,
    session_started_at: str | None,
    session_completed_at: str | None,
    session_wallclock_seconds: float | None,
    subagent_max_depth: int | None = None,
    tier_attempt_index: int | None = None,
    escalation_action: str | None = None,
    escalation_tier_hint: int | None = None,
    outcome_synthesised: bool | None = None,
) -> dict[str, Any] | None:
    """Extract a Phase 4 telemetry row + write it to the JSONL sink.

    The timing fields (``session_started_at``,
    ``session_completed_at``, ``session_wallclock_seconds``) come
    from :func:`_dispatch_runtime.dispatch_one` which brackets
    ``run_session_blocking`` with two ``_now()`` calls. When the
    helper is invoked outside the runtime (direct unit tests), the
    callers pass ``None`` for the three fields and the row is
    populated with a single ``now_iso()`` snapshot — analysts who
    run aggregation against the JSONL sink will see ``None`` /
    ``0``-second rows for those direct invocations and can filter
    them out by absence of timing data.

    Returns ``{"row": <row>, "sink": <path or None>}`` on success or
    ``None`` if extraction itself raised (kept best-effort so a bug
    in the regex extractor cannot orphan a claim release).
    """
    pr_number = int(item.get("number") or 0)
    pre_sha = _resolve_pre_session_head_sha(item, context)
    post_sha = _fetch_post_session_head_sha(cfg, pr_number) if pre_sha else ""
    fallback_now = _phase4_telemetry.now_iso()
    start_ts = session_started_at or fallback_now
    end_ts = session_completed_at or fallback_now
    try:
        row = _phase4_telemetry.extract_phase4_telemetry(
            cycle_id=None,
            pr_number=pr_number,
            work_group=work_group_name,
            start_ts=start_ts,
            end_ts=end_ts,
            wall_clock_seconds=session_wallclock_seconds,
            parsed_json=parsed_json,
            raw_response=raw_response,
            terminal_state=terminal_state,
            pre_session_head_sha=pre_sha,
            post_session_head_sha=post_sha or None,
            subagent_max_depth=subagent_max_depth,
            tier_attempt_index=tier_attempt_index,
            escalation_action=escalation_action,
            escalation_tier_hint=escalation_tier_hint,
            outcome_synthesised=outcome_synthesised,
        )
    except Exception as exc:
        _logger.warning(
            "phase4 telemetry extraction raised for #%s: %s",
            pr_number,
            exc,
        )
        return None
    sink = _phase4_telemetry.write_telemetry_jsonl(row)
    # C4 (2026-05-13): if this attempt produced a push, record it
    # so the NEXT cycle for the same PR knows the dispatcher
    # itself just contributed the current head_sha. The next
    # cycle's prefetch surfaces this to the worker so it doesn't
    # re-do the same fix when CI is still failing for a different
    # reason (PR #28 cycle 2 failure mode from the live test).
    if (
        post_sha
        and pre_sha
        and post_sha != pre_sha
    ):
        try:
            outcome_str = ""
            if isinstance(parsed_json, dict):
                outcome_str = str(parsed_json.get("outcome") or "")
            _recent_push_cache.record_push(
                pr_number, post_sha,
                terminal_state=terminal_state,
                outcome=outcome_str,
            )
        except Exception as exc:
            _logger.warning(
                "recent-push cache record failed for #%s: %s",
                pr_number, exc,
            )
    return {"row": row, "sink": str(sink) if sink else None}


def _cleanup_clone_handle(
    cfg: Any, item: dict[str, Any], context: dict[str, Any] | None
) -> tuple[bool, str | None]:
    """Cleanup the pre-cloned worktree (when one was created in
    Phase 3) and return ``(attempted, error_string)``. Best-effort
    — failures (the worker ``rm -rf``'d the dir, disk full, etc.)
    are logged + reported but do NOT raise.

    Also removes the PR-context sentinel — even when no clone was
    materialised the dispatcher may have written one (the two
    sentinels are independent), so this runs unconditionally for
    any item with a PR number. ``handle.cleanup`` already removes
    the workspace sentinel; the PR-context sentinel lives in a
    different directory and needs its own cleanup call.
    """
    pr_number = int(item.get("number") or 0)
    if pr_number > 0:
        try:
            _pr_context_sentinel.delete(pr_number)
        except Exception as e:  # noqa: BLE001 — best-effort cleanup
            _logger.warning(
                "PR context sentinel cleanup raised for #%s: %s",
                pr_number, e,
            )
    if not isinstance(context, dict):
        return False, None
    handle = context.get("clone_handle")
    if handle is None:
        return False, None
    try:
        handle.cleanup(cfg)
    except Exception as exc:
        error_repr = f"{type(exc).__name__}: {exc}"
        _logger.warning(
            "implementer pre-clone cleanup raised for #%s: %s",
            item.get("number"),
            exc,
        )
        return True, error_repr
    return True, None


def _maybe_post_status_comment(
    cfg: Any,
    item: dict[str, Any],
    *,
    parsed_json: dict[str, Any] | None,
    terminal_state: str,
    context: dict[str, Any] | None,
    tier: int | None = None,
) -> dict[str, Any] | None:
    """Post a per-fingerprint operator-status comment when the cycle
    ended without a successful PR push.

    Tri-state return value (chosen so cycle-archive consumers can
    distinguish "not applicable" from "ran and skipped" without
    inspecting the Forgejo POST log):

    - ``None`` — gate did not fire (dry-run, issue work item, or
      worker reported a successful resolved push). The cycle archive
      reads ``post_session_result.status_comment is None`` as
      "no status comment expected for this cycle."
    - ``{..., "skipped_duplicate": True}`` — gate fired but a
      matching fingerprint already existed on the timeline; no POST
      was issued.
    - ``{..., "skipped_duplicate": False, "status": 2xx, ...}`` —
      gate fired AND posted; ``status`` is the Forgejo HTTP code.
    - ``{..., "status": 0, "error": "..."}`` — POST raised; we
      swallow + log so claim release isn't orphaned.
    """
    pr_number = int(item.get("number") or 0)
    item_kind_is_pr = bool(item.get("head"))
    if cfg.dry_run or pr_number <= 0 or not item_kind_is_pr:
        return None
    should_post, outcome, reason = _should_post_status(parsed_json, terminal_state)
    if not should_post:
        return None
    pr_comments: list[dict[str, Any]] | None = None
    if isinstance(context, dict):
        result = context.get("result")
        if hasattr(result, "pr_comments"):
            pr_comments = list(result.pr_comments) if result.pr_comments else []
    try:
        post_result = _review_post.post_implementer_status_comment(
            cfg,
            pr_number,
            outcome=outcome,
            outcome_reason=reason,
            terminal_state=terminal_state,
            pr_comments=pr_comments,
            tier=tier,
        )
        return {
            "fingerprint": post_result.get("fingerprint"),
            "skipped_duplicate": post_result.get("skipped_duplicate"),
            "status": post_result.get("status"),
        }
    except Exception as exc:
        _logger.warning(
            "post_implementer_status_comment failed for #%s: %s",
            pr_number,
            exc,
        )
        return {
            "fingerprint": None,
            "skipped_duplicate": False,
            "status": 0,
            "error": f"{type(exc).__name__}: {exc}",
        }


def _post_session_action(
    cfg: Any,
    item: dict[str, Any],
    parsed_json: dict[str, Any] | None,
    raw_response: str,
    terminal_state: str,
    *,
    session_context: Any | None = None,
    work_group_name: str | None = None,
    session_started_at: str | None = None,
    session_completed_at: str | None = None,
    session_wallclock_seconds: float | None = None,
    subagent_max_depth: int | None = None,
) -> dict[str, Any]:
    """Cleanup + status-comment hook the dispatcher runs after the
    worker session.

    Three responsibilities, each in its own helper:

    1. :func:`_record_phase4_telemetry` — extract a Phase 4 row from
       the session output and write it to the JSONL sink (when the
       :data:`_phase4_telemetry.PHASE4_TELEMETRY_ENV_VAR` env var
       points at a directory).
    2. :func:`_cleanup_clone_handle` — remove the pre-cloned
       worktree (when one was created — Phase 3 gating).
    3. :func:`_maybe_post_status_comment` — post a per-fingerprint
       operator-status comment when the cycle ended without a
       successful PR push (Phase 5b).

    Inputs come from :func:`_dispatch_runtime.dispatch_one`:

    - ``session_context`` (preferred) — a
      :class:`_dispatch_runtime.SessionContext` dataclass packing
      ``work_group_name`` plus session-timing fields. The dispatcher
      always passes this kwarg; direct test callers may pass the
      legacy flat kwargs instead (see below).
    - ``work_group_name`` / ``session_started_at`` /
      ``session_completed_at`` / ``session_wallclock_seconds`` —
      legacy flat kwargs. Used only when ``session_context`` is
      absent (back-compat for tests written against the previous
      API). New code should always pass ``session_context``.

    The status comment is only posted for items that had a real PR
    number — ``new_issue`` cycles that never opened a PR have no
    timeline to post on. Skipped on dry-run.

    Returns a dict that gets merged into the cycle archive under
    ``post_session_result`` so an operator can see whether cleanup +
    status fired. Errors propagate as logged warnings; raising would
    orphan the claim release.
    """
    if session_context is not None:
        # Unpack the dataclass; the flat-kwargs path is only the
        # back-compat tail for direct test callers.
        work_group_name = session_context.work_group_name
        session_started_at = session_context.session_started_at
        session_completed_at = session_context.session_completed_at
        session_wallclock_seconds = session_context.session_wallclock_seconds
        # ``subagent_max_depth`` lives on SessionContext but is new
        # (Tier-1 R2): older tests may construct a SessionContext
        # without it. ``getattr`` with the default preserves them.
        subagent_max_depth = getattr(
            session_context, "subagent_max_depth", None
        )
    context = item.get("_dispatcher_implementer_context")
    context_dict = context if isinstance(context, dict) else None
    resolved_group_name = _resolve_work_group_name(item, work_group_name)
    out: dict[str, Any] = {
        "cleanup_attempted": False,
        "cleanup_error": None,
        "status_comment": None,
        "phase4_telemetry": None,
        "work_group_name": resolved_group_name,
    }
    # Surface the dispatcher's deterministic-fix report (P0 / A) when
    # the cycle was short-circuited by ``_maybe_short_circuit``. An
    # operator inspecting cycle archives sees ``auto_fix_report`` ==
    # None on a normal worker cycle and a populated dict when the
    # dispatcher applied compliance fixes itself.
    out["auto_fix_report"] = (
        context_dict.get("_auto_fix_report") if context_dict else None
    )
    out["short_circuit"] = bool(
        context_dict and context_dict.get("_short_circuit_consumed")
    )
    out["phase4_telemetry"] = _record_phase4_telemetry(
        cfg,
        item,
        parsed_json=parsed_json,
        raw_response=raw_response,
        terminal_state=terminal_state,
        context=context_dict,
        work_group_name=resolved_group_name,
        session_started_at=session_started_at,
        session_completed_at=session_completed_at,
        session_wallclock_seconds=session_wallclock_seconds,
        subagent_max_depth=subagent_max_depth,
    )
    out["cleanup_attempted"], out["cleanup_error"] = _cleanup_clone_handle(
        cfg, item, context_dict
    )
    out["status_comment"] = _maybe_post_status_comment(
        cfg,
        item,
        parsed_json=parsed_json,
        terminal_state=terminal_state,
        context=context_dict,
    )
    return out


# ─── In-cycle tier escalation (flag-gated) ──────────────────────────────────


def _fetch_pr_state(cfg: Any, pr_number: int) -> str:
    """Best-effort GET of the PR state ("open" / "closed" / "merged").

    The escalation loop calls this between tier attempts so the
    predicate (:func:`_implementer_escalation.decide`) can end the
    cycle early when the PR is closed or merged by a human mid-cycle.
    On any failure (dry-run, transient 5xx, network), this returns
    ``"open"`` — the conservative default that lets the loop continue
    rather than spuriously ending the cycle on a transient fetch
    blip. The plan's tier-stable class names this explicitly.
    """
    if cfg.dry_run or pr_number <= 0:
        return "open"
    try:
        response = _claim_runtime.get(
            f"/repos/{cfg.owner}/{cfg.repo}/pulls/{int(pr_number)}", cfg
        )
    except Exception as exc:
        _logger.warning(
            "PR state fetch failed for #%s (treating as 'open'): %s",
            pr_number, exc,
        )
        return "open"
    if int(response.get("status") or 0) != 200:
        return "open"
    body = response.get("body") or {}
    if not isinstance(body, dict):
        return "open"
    state = body.get("state")
    if isinstance(state, str) and state:
        # Forgejo exposes "merged" via the boolean `merged` flag on
        # state=closed PRs; honour that distinction so the predicate
        # can map both to tier-stable without false-negatives.
        if state == "closed" and body.get("merged") is True:
            return "merged"
        return state
    return "open"


def _compute_head_sha_tristate(
    cfg: Any, pr_number: int, pre_sha: str
) -> bool | None:
    """Tri-state computation of ``head_sha_advanced``.

    - ``True``  — verified the worker pushed (pre != post, both set).
    - ``False`` — verified the worker did NOT push (pre == post).
    - ``None``  — fetch failed or pre_sha unknown; the predicate
      treats this distinctly (e.g. ``outcome=resolved`` + ``None`` →
      :data:`_implementer_escalation.EscalationAction.RETRY_POST_FETCH`).

    Mirrors the dispatcher's existing head_sha_advanced derivation
    (in :mod:`_phase4_telemetry`) but exposes the tri-state for the
    escalation predicate rather than collapsing ``None`` to ``False``.
    """
    if not pre_sha:
        return None
    post_sha = _fetch_post_session_head_sha(cfg, pr_number)
    if not post_sha:
        return None
    return pre_sha != post_sha


def _reset_worktree_to_pinned_sha(handle: Any) -> bool:
    """Reset the pre-cloned worktree to the SHA captured at
    prefetch time so a subsequent tier starts from the same baseline
    Tier 0 saw.

    PD P0 (plan v2 critique): a plain ``git reset --hard HEAD`` is
    ambiguous if the worker pushed during Tier N — ``HEAD`` may
    have advanced via the worker's session-end ``git fetch``. The
    pinned ``handle.head_sha`` is the prefetch-time snapshot stored
    on the :class:`_pr_clone.WorktreeHandle`. Reset target is that
    SHA explicitly.

    Returns ``True`` on success, ``False`` on any subprocess failure
    (logged WARNING; escalation continues — the next session's
    ``implementer-workspace.py discover`` will see the vanished /
    dirty worktree and fall through to ``git-isolator-util``, the
    legacy pre-pre-clone code path).

    Run-11 deep inspection: the prior worker session can ``rm -rf``
    its own worktree (the task-implementor prompt at one point
    explicitly instructed it to — fixed in the same commit as this
    function). Defend against that and other state-corruption cases:

    - Missing worktree dir: log clearly and return False without
      shelling out to ``git`` against a non-existent path.
    - Stale ``.git/index.lock`` from a SIGKILL'd worker git op:
      remove it so the reset isn't blocked by an orphan lock.
    - On ``git`` failure, surface the captured stderr in the
      warning — the bare ``CalledProcessError.__str__`` only shows
      the exit code, which is useless for diagnosis.

    Note: the remote branch may carry commits the worker pushed in
    Tier N. The plan's silent-worst-case rule
    (``head_sha_advanced=True`` + non-success outcome → END_CYCLE)
    short-circuits that case before we reach this reset.
    """
    import subprocess
    from pathlib import Path as _Path

    if handle is None:
        return False
    path = getattr(handle, "path", None)
    pinned_sha = getattr(handle, "head_sha", None)
    if not path or not pinned_sha:
        return False
    sha_short = str(pinned_sha)[:12]

    if not _Path(path).is_dir():
        _logger.warning(
            "worktree disappeared before reset for SHA %s at %s — "
            "the prior worker session likely deleted it. Escalation "
            "will continue; the next session's "
            "``implementer-workspace.py discover`` will fall through "
            "to ``git-isolator-util`` for a fresh clone.",
            sha_short, path,
        )
        return False

    # Stale-lock cleanup. A SIGKILL'd or crashed git op in the
    # previous session can leave ``.git/<name>.lock`` files behind
    # (``index.lock`` is the common one, but ``HEAD.lock``,
    # ``MERGE_MSG.lock``, ``ORIG_HEAD.lock`` are also possible),
    # which then blocks every subsequent git operation with
    # ``Unable to create '...lock'``. The locks have no legitimate
    # concurrent owner here (the worker session has already
    # terminated by the time the dispatcher is between tiers), so
    # removing them is safe and matches what an operator would do
    # by hand. The glob covers every ``*.lock`` under the worktree's
    # ``.git/`` rather than just ``index.lock`` so any combination
    # of crashed git ops leaves a clean slate for the reset.
    try:
        git_dir = _Path(path) / ".git"
        if git_dir.is_dir():
            for lock in git_dir.glob("*.lock"):
                if lock.is_file():
                    lock.unlink()
                    _logger.info(
                        "removed stale .git/%s at %s before reset",
                        lock.name, path,
                    )
    except OSError:
        pass

    try:
        subprocess.run(
            ["git", "-C", str(path), "reset", "--hard", str(pinned_sha)],
            check=True, capture_output=True, timeout=30, text=True,
        )
        subprocess.run(
            ["git", "-C", str(path), "clean", "-xfdq"],
            check=True, capture_output=True, timeout=30, text=True,
        )
        return True
    except subprocess.CalledProcessError as exc:
        # ``CalledProcessError.__str__`` only shows the exit code —
        # include the actual stderr so the operator can act on it
        # (vanished SHA vs. lock contention vs. permissions issue,
        # etc.).
        _logger.warning(
            "worktree reset to pinned SHA %s failed at %s: "
            "exit=%s stderr=%r; escalation continues",
            sha_short, path, exc.returncode,
            (exc.stderr or "").strip()[:400],
        )
        return False
    except subprocess.SubprocessError as exc:
        # Timeout / missing binary / etc. — same fail-soft policy.
        _logger.warning(
            "worktree reset to pinned SHA %s raised %s at %s; "
            "escalation continues",
            sha_short, type(exc).__name__, path,
        )
        return False


def _read_start_tier_from_labels(cfg: Any, pr_number: int) -> int:
    """Read ``auto/last-attempt-tier-N`` from the PR's labels and
    return the tier the dispatcher should START this cycle at.

    Policy: ``start_tier = min(labeled_tier + 1, max_tier)``. The
    label captures "we already tried this tier and it failed (or
    crashed mid-run)"; the next cycle skips ahead one tier so the
    same model isn't asked to redo the same work. When the labeled
    tier equals ``max_tier``, the cap stops further escalation —
    the cycle just retries the top tier (effectively a no-op
    promotion).

    Returns ``0`` when the PR has no attempt label, when the fetch
    fails, or in dry-run mode. The label is observability AND
    seed-state (per the 2026-05-12 decision: we wrote it for a
    reason; use it).

    Conservative side: a dispatcher that crashes between LAUNCH and
    a successful end-of-cycle could leave a label that overshoots
    the actual work done. Worst case: the next cycle skips ahead
    and tries a higher tier than strictly needed. The cost is one
    higher-tier worker call vs. the savings of skipping a known-
    failed tier — favourable on average.
    """
    if cfg.dry_run or pr_number <= 0:
        return 0
    try:
        response = _claim_runtime.get(
            f"/repos/{cfg.owner}/{cfg.repo}/issues/{int(pr_number)}/labels",
            cfg,
        )
    except Exception as exc:
        _logger.warning(
            "label fetch for start-tier failed for #%s "
            "(starting at Tier 0): %s",
            pr_number, exc,
        )
        return 0
    if int(response.get("status") or 0) != 200:
        return 0
    body = response.get("body")
    if not isinstance(body, list):
        return 0
    # Find the highest labelled tier (defensive — a PR shouldn't
    # carry multiple last-attempt-tier-* labels because
    # ``apply_attempt_label`` clears siblings, but the same PR
    # processed by two dispatcher versions could have stale ones).
    # Bound-check the parsed integer against the known label set
    # so an operator-introduced ``auto/last-attempt-tier-9`` typo
    # doesn't silently get capped — explicitly skip out-of-range
    # values and log so the stale-label state is operator-visible.
    valid_tier_range = range(len(_implementer_label_state.ATTEMPT_TIER_LABELS))
    highest_labelled = -1
    for label in body:
        if not isinstance(label, dict):
            continue
        name = label.get("name")
        if not isinstance(name, str):
            continue
        if not name.startswith("auto/last-attempt-tier-"):
            continue
        try:
            n = int(name.rsplit("-", 1)[-1])
        except ValueError:
            continue
        if n not in valid_tier_range:
            _logger.warning(
                "PR #%s carries out-of-range attempt label %r "
                "(valid: %s); skipping for start-tier seed",
                pr_number, name, list(valid_tier_range),
            )
            continue
        if n > highest_labelled:
            highest_labelled = n
    if highest_labelled < 0:
        return 0
    return min(highest_labelled + 1, _max_tier_for_cycle())


def _build_prompt_for_tier(
    cfg: Any, item: dict[str, Any], group: Any, tier: int
) -> str:
    """Build the worker prompt for an explicit tier attempt.

    Always rebuilds the base prompt fresh (no recursion through
    :func:`_implementation_prompt_dispatch`) so the
    ``escalation_tier_hint`` line is added cleanly without risking
    a stale inherited hint from cross-cycle resumption. Tier 1+
    append the explicit hint. Tier 0 emission depends on the G11
    estimator flag: with the flag OFF (default) an explicit
    ``escalation_tier_hint: `0` `` line is emitted to preserve the
    legacy short-circuit; with the flag ON the line is omitted so
    ``tier-dispatcher`` runs the estimator instead.

    Used by:
    - The escalation loop in :func:`_post_session_action_with_escalation`
      for tier > 0 attempts.
    - :func:`_implementation_prompt_dispatch` for the initial cycle
      prompt when cross-cycle resumption seeds ``start_tier > 0``.
    """
    if _is_prefetch_enabled():
        base_prompt = _prefetch_prompt(cfg, item, group)
    else:
        base_prompt = _legacy_implementation_prompt(cfg, item, group)
    # The escalation loop only spawns subsequent tier sessions when
    # ``_is_escalation_enabled()`` is True and the item is
    # PR-shaped. In every case where this helper runs the
    # release-skip directive applies, so include it unconditionally
    # — the dispatcher holds the claim across the whole cycle.
    extras = ["release_claim_on_exit: false"]
    if tier > 0:
        extras.append(f"escalation_tier_hint: `{tier}`")
    elif not _is_implementer_estimator_enabled():
        # G11 (2026-05-15): emit the explicit hint=0 fallback so the
        # worker short-circuits to ``tier-0`` when the estimator flag
        # is OFF. The worker no longer defaults to ``0`` on its own.
        extras.append("escalation_tier_hint: `0`")
    # else (tier == 0 AND estimator enabled): omit the hint so the
    # worker falls through to ``estimator-implementation``.
    return f"{base_prompt}\n\n" + "\n".join(extras) + "\n"


def _per_tier_worker_timeout(cfg: Any, tier: int) -> int:
    """Resolve the worker timeout for ``tier``.

    Per-tier override via ``IMPLEMENTER_DISPATCHER_WORKER_TIMEOUT_TIER_{N}_SECONDS``;
    falls back to ``cfg.worker_timeout_seconds`` (the global
    setting). Empirical observation (2026-05-13): the higher tier
    slots (``tier-1``, ``tier-2``) historically ran significantly
    slower than the default slot (``tier-0``) and the global 1800s
    limit forced repeated timeouts on PR #30 attempts 4-5 in the
    live test. Recommended values when the global is set to a
    Tier-0-friendly default of 900s: Tier 0=900s, Tier 1=1800s,
    Tier 2=3600s. The actual model behind each slot is configured
    in ``.opencode/models/tiers.yaml``; the per-tier timeout knob
    is slot-based, not model-based, so it stays correct across
    model swaps.

    The override is read at session-spawn time so an operator can
    bump the value without restarting the dispatcher (next worker
    invocation picks it up).
    """
    env_name = f"IMPLEMENTER_DISPATCHER_WORKER_TIMEOUT_TIER_{int(tier)}_SECONDS"
    raw = os.environ.get(env_name)
    if raw:
        try:
            return max(60, int(raw))
        except ValueError:
            _logger.warning(
                "%s=%r is not an int; falling back to global timeout",
                env_name, raw,
            )
    return int(cfg.worker_timeout_seconds)


def _run_worker_at_tier(
    cfg: Any,
    group: Any,
    item: dict[str, Any],
    tier: int,
    *,
    tag: str,
    on_poll: Any,
    redact_values: list[str],
) -> Any:
    """Spawn a worker session for ``tier`` and return its
    :class:`_opencode_worker.SessionResult`.

    Wraps :func:`_opencode_worker.run_session_blocking` so the
    escalation loop can call it for each subsequent tier without
    duplicating the redact / heartbeat plumbing. ``tag`` /
    ``on_poll`` / ``redact_values`` are reused from the original
    ``dispatch_one`` invocation that spawned Tier 0 — the worker
    session for Tier N inherits the same tag so an operator
    grepping the cycle archive sees all attempts under one PR.

    Timeout is resolved per-tier (B2 fix) — Tier 1+ models legitimately
    need more wallclock than Tier 0. See
    :func:`_per_tier_worker_timeout`.
    """
    prompt = _build_prompt_for_tier(cfg, item, group, tier)
    return _opencode_worker.run_session_blocking(
        server_url=cfg.server_url,
        agent=group.worker_agent,
        tag=tag,
        prompt=prompt,
        timeout_seconds=_per_tier_worker_timeout(cfg, tier),
        on_poll=on_poll,
        redact_values=redact_values,
    )


def _terminal_state_from_session(session: Any) -> str:
    """Mirror dispatch_one's terminal_state derivation for a session
    that the escalation loop spawned (Tier 1+). ``completed`` for the
    happy path, the raw ``status`` otherwise (``timeout`` /
    ``transport-error``)."""
    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 (as ``outcome_synthesised``) so an operator analysing the
    JSONL sink can distinguish cycles where the worker explicitly
    reported failure from cycles where the dispatcher synthesised a
    verdict.

    Gated by :func:`_is_outcome_synthesis_enabled`. When that returns
    False, the function passes through whatever ``parsed_json`` was
    (including ``None``) with ``was_synthesized=False``; the
    downstream escalation predicate will then see the legacy
    UNKNOWN bucket and behave byte-equivalent to the pre-feature
    build.
    """
    if isinstance(parsed_json, dict):
        outcome = parsed_json.get("outcome")
        if isinstance(outcome, str) and outcome:
            return parsed_json, False
    if not _is_outcome_synthesis_enabled():
        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],
    parsed_json: dict[str, Any] | None,
    raw_response: str,
    terminal_state: str,
    *,
    session_context: Any | None = None,
    work_group_name: str | None = None,
    session_started_at: str | None = None,
    session_completed_at: str | None = None,
    session_wallclock_seconds: float | None = None,
    subagent_max_depth: int | None = None,
) -> dict[str, Any]:
    """Flag=1 post-session action: in-cycle tier escalation.

    Runs immediately after the dispatcher's Tier 0 worker session.
    Performs the same per-attempt work as
    :func:`_post_session_action` (Phase 4 telemetry, status comment),
    then evaluates :func:`_implementer_escalation.decide`. If
    escalation is warranted, runs subsequent tiers in the same
    process — refreshing the claim TTL via
    :func:`_claim_runtime.claim_pr` (the agent-side
    ``claim_pr.ts --action claim`` does NOT bump TTL on re-claim, see
    plan §"Audit finding"), resetting the pre-cloned worktree to the
    prefetched head_sha, and mutating the
    ``auto/last-attempt-tier-N`` labels for operator visibility.

    Falls back to the legacy single-shot path
    (:func:`_post_session_action`) when the item is not PR-shaped
    (``new_issue`` work group) — the plan explicitly scopes
    escalation to PR work for v1.

    The escalation loop is bounded by the failure-class budgets
    encoded in :mod:`_implementer_escalation`: at most
    ``max_tier + 1`` worker sessions per cycle, with 2 retries at
    same tier for transport-class failures (default; see
    :data:`_implementer_escalation.BUDGET_PER_FAILURE_CLASS`). The
    worst-case cycle wallclock is ``(max_tier + 1) * worker_timeout``
    which v1 caps at 2 × 7200s = 4h with Tier 0→1 only.
    """
    # Issue work bypasses the escalation harness — no PR exists yet
    # to mutate labels on or hold a claim across.
    is_pr_shape = isinstance(item.get("head"), dict)
    if not is_pr_shape:
        return _post_session_action(
            cfg, item, parsed_json, raw_response, terminal_state,
            session_context=session_context,
            work_group_name=work_group_name,
            session_started_at=session_started_at,
            session_completed_at=session_completed_at,
            session_wallclock_seconds=session_wallclock_seconds,
            subagent_max_depth=subagent_max_depth,
        )

    if session_context is not None:
        work_group_name = session_context.work_group_name
        session_started_at = session_context.session_started_at
        session_completed_at = session_context.session_completed_at
        session_wallclock_seconds = session_context.session_wallclock_seconds
        subagent_max_depth = getattr(
            session_context, "subagent_max_depth", None
        )
        # Heartbeat callback threaded from dispatch_one (P0 fix
        # 2026-05-12). Without it, Tier 1+ sessions ran with a
        # no-op on_poll and the heartbeat file went stale for the
        # entire session duration — systemd watchdog could SIGTERM
        # the dispatcher mid-escalation. ``None`` (back-compat for
        # tests that constructed SessionContext without it) falls
        # back to a no-op closure that still satisfies the worker's
        # callable contract.
        heartbeat_callback = (
            getattr(session_context, "heartbeat_callback", None)
            or (lambda: None)
        )
    else:
        heartbeat_callback = lambda: None  # noqa: E731 — legacy path

    pr_number = int(item.get("number") or 0)
    context_dict = item.get("_dispatcher_implementer_context")
    if not isinstance(context_dict, dict):
        context_dict = None
    resolved_group_name = _resolve_work_group_name(item, work_group_name)
    clone_handle = (
        context_dict.get("clone_handle") if context_dict else None
    )
    pre_sha = _resolve_pre_session_head_sha(item, context_dict)

    # Tier 0 per-attempt work: phase4 row + per-tier status comment.
    # We reuse the existing helpers; the only difference vs. the
    # legacy path is that we mark the attempt index for telemetry
    # (TODO #6 wires this into the phase4 schema; until then the
    # extractor silently ignores the kwarg).
    attempts: list[dict[str, Any]] = []

    _logger.info(
        "escalation: starting in-cycle loop for PR #%s (max_tier=%d)",
        pr_number, _max_tier_for_cycle(),
    )

    def _record_attempt(
        tier_idx: int,
        parsed: dict[str, Any] | None,
        raw: str,
        term_state: str,
        wallclock: float | None,
        depth: int | None,
        action_str: str | None = None,
        outcome_synthesised: bool | None = None,
    ) -> dict[str, Any]:
        """Per-attempt side effects: telemetry write + per-tier
        status comment. Returns a dict for the attempts log."""
        phase4 = _record_phase4_telemetry(
            cfg, item,
            parsed_json=parsed,
            raw_response=raw,
            terminal_state=term_state,
            context=context_dict,
            work_group_name=resolved_group_name,
            session_started_at=session_started_at,
            session_completed_at=session_completed_at,
            session_wallclock_seconds=wallclock,
            subagent_max_depth=depth,
            tier_attempt_index=tier_idx,
            escalation_action=action_str,
            escalation_tier_hint=tier_idx,
            outcome_synthesised=outcome_synthesised,
        )
        status_comment = _maybe_post_status_comment(
            cfg, item,
            parsed_json=parsed,
            terminal_state=term_state,
            context=context_dict,
            tier=tier_idx,
        )
        return {
            "tier": tier_idx,
            "terminal_state": term_state,
            "escalation_action": action_str,
            "phase4_telemetry": phase4,
            "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
    # PR has never escalated before (or in dry-run / fetch-failed
    # cases), it defaults to 0 — byte-equivalent to a first-attempt
    # cycle.
    start_tier = 0
    if isinstance(context_dict, dict):
        try:
            stashed = int(context_dict.get("start_tier") or 0)
        except (TypeError, ValueError):
            stashed = 0
        start_tier = max(0, min(stashed, _max_tier_for_cycle()))
    if not cfg.dry_run:
        # Best-effort: a label-add failure (unprovisioned label,
        # transient Forgejo 5xx, ValueError on out-of-range tier)
        # must not kill the cycle before the post-session telemetry
        # write, the cleanup, and the claim release fire. The
        # symmetric ``clear_attempt_labels`` call at end-of-cycle
        # is already wrapped; mirror that here.
        try:
            _implementer_label_state.apply_attempt_label(
                cfg, pr_number, start_tier,
            )
        except Exception as exc:
            _logger.warning(
                "apply_attempt_label(%d) failed for #%s "
                "(continuing without per-tier label observability): %s",
                start_tier, pr_number, exc,
            )
    head_sha_advanced = _compute_head_sha_tristate(cfg, pr_number, pre_sha)
    pr_state = _fetch_pr_state(cfg, pr_number)
    action = _implementer_escalation.decide(
        parsed_json=parsed_json,
        terminal_state=terminal_state,
        head_sha_advanced=head_sha_advanced,
        pr_state=pr_state,
        transport_retries_used=0,
        current_tier=start_tier,
        max_tier=_max_tier_for_cycle(),
    )
    attempts.append(_record_attempt(
        start_tier, parsed_json, raw_response, terminal_state,
        session_wallclock_seconds, subagent_max_depth,
        action_str=action.value,
        outcome_synthesised=t0_outcome_synthesised,
    ))

    EscAction = _implementer_escalation.EscalationAction
    current_tier = start_tier
    transport_retries = 0
    # ``last_*`` carry the most-recent worker session's outputs so
    # the loop's decide() call sees fresh data each iteration.
    last_parsed = parsed_json
    last_raw = raw_response
    last_terminal_state = terminal_state
    last_wallclock = session_wallclock_seconds
    last_depth = subagent_max_depth
    last_outcome_synthesised = t0_outcome_synthesised
    final_action = action

    # The terminal actions (SUCCESS, END_CYCLE, EXHAUSTED) skip the
    # loop entirely. The non-terminal actions enter the while loop.
    while action not in (
        EscAction.SUCCESS, EscAction.END_CYCLE, EscAction.EXHAUSTED
    ):
        if action == EscAction.RETRY_POST_FETCH:
            # One re-fetch of the head_sha; no worker re-run, no
            # tier advance. Re-decide.
            head_sha_advanced = _compute_head_sha_tristate(cfg, pr_number, pre_sha)
            pr_state = _fetch_pr_state(cfg, pr_number)
            action = _implementer_escalation.decide(
                parsed_json=last_parsed,
                terminal_state=last_terminal_state,
                head_sha_advanced=head_sha_advanced,
                pr_state=pr_state,
                transport_retries_used=transport_retries,
                current_tier=current_tier,
                max_tier=_max_tier_for_cycle(),
            )
            final_action = action
            continue

        if action == EscAction.RETRY_SAME_TIER:
            transport_retries += 1
            _implementer_escalation.sleep_for_retry(transport_retries)
            # No claim refresh, no worktree reset — this is a
            # same-tier retry by design (transport-class failure).
            new_session = _run_worker_at_tier(
                cfg, _work_group_for_item(item, resolved_group_name), item,
                current_tier,
                tag=_tag_for_item(item, resolved_group_name),
                on_poll=heartbeat_callback,
                redact_values=[cfg.token] if getattr(cfg, "token", None) else [],
            )
            last_parsed = new_session.parsed_json
            last_terminal_state = _terminal_state_from_session(new_session)
            last_parsed, last_outcome_synthesised = _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:
            # ESCALATE: advance tier, refresh claim, reset worktree,
            # apply new label, run worker.
            current_tier += 1
            transport_retries = 0
            if not cfg.dry_run:
                try:
                    _implementer_label_state.apply_attempt_label(
                        cfg, pr_number, current_tier,
                    )
                except Exception as exc:
                    _logger.warning(
                        "apply_attempt_label(%d) failed mid-escalation "
                        "for #%s (continuing escalation without "
                        "per-tier label observability): %s",
                        current_tier, pr_number, exc,
                    )
                # Refresh the visible TTL comment so the merge-driver
                # sweep's expiry math advances. The agent-side
                # claim_pr.ts release between sessions briefly drops
                # the label; the race window is small and the
                # implementer dispatcher's single-instance lock makes
                # a sibling-dispatcher takeover practically impossible
                # — see plan v2 §"Concurrency" and the legacy-removal
                # milestone for future hardening.
                try:
                    _claim_runtime.claim_pr(
                        pr_number, cfg, driver_name=DRIVER_NAME,
                    )
                except Exception as exc:
                    _logger.warning(
                        "claim refresh between tiers failed for #%s "
                        "(escalation continues; merge-driver sweep "
                        "remains the safety net): %s",
                        pr_number, exc,
                    )
                _reset_worktree_to_pinned_sha(clone_handle)
            new_session = _run_worker_at_tier(
                cfg, _work_group_for_item(item, resolved_group_name), item,
                current_tier,
                tag=_tag_for_item(item, resolved_group_name),
                on_poll=heartbeat_callback,
                redact_values=[cfg.token] if getattr(cfg, "token", None) else [],
            )
            last_parsed = new_session.parsed_json
            last_terminal_state = _terminal_state_from_session(new_session)
            last_parsed, last_outcome_synthesised = _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

        head_sha_advanced = _compute_head_sha_tristate(cfg, pr_number, pre_sha)
        pr_state = _fetch_pr_state(cfg, pr_number)
        action = _implementer_escalation.decide(
            parsed_json=last_parsed,
            terminal_state=last_terminal_state,
            head_sha_advanced=head_sha_advanced,
            pr_state=pr_state,
            transport_retries_used=transport_retries,
            current_tier=current_tier,
            max_tier=_max_tier_for_cycle(),
        )
        attempts.append(_record_attempt(
            current_tier, last_parsed, last_raw, last_terminal_state,
            last_wallclock, last_depth,
            action_str=action.value,
            outcome_synthesised=last_outcome_synthesised,
        ))
        final_action = action

    # End-of-cycle cleanup: clone handle, sentinel, attempt labels.
    cleanup_attempted, cleanup_error = _cleanup_clone_handle(
        cfg, item, context_dict,
    )
    if not cfg.dry_run:
        try:
            _implementer_label_state.clear_attempt_labels(cfg, pr_number)
        except Exception as exc:
            _logger.warning(
                "clear_attempt_labels failed for #%s: %s",
                pr_number, exc,
            )

    return {
        "escalation_enabled": True,
        "final_action": final_action.value,
        "final_tier": current_tier,
        "attempts": attempts,
        "cleanup_attempted": cleanup_attempted,
        "cleanup_error": cleanup_error,
        "work_group_name": resolved_group_name,
    }


def _work_group_for_item(item: dict[str, Any], group_name: str) -> Any:
    """Look up a :class:`_dispatch.WorkGroup` by name so the
    escalation loop's worker-spawn helper has the same group object
    that ``dispatch_one`` used for Tier 0. Falls back to the first
    PR-shaped group on miss — the escalation loop only runs for
    PR-shaped items, so this fallback is a defensive safety net.
    """
    for g in WORK_GROUPS:
        if g.name == group_name:
            return g
    for g in WORK_GROUPS:
        if g.item_kind == "pr":
            return g
    return WORK_GROUPS[0]


def _tag_for_item(item: dict[str, Any], group_name: str) -> str:
    """Tag used for the OpenCode session for tier > 0 attempts.

    Mirrors ``_dispatch_runtime._tag_for`` but accessible from
    inside ``dispatch_implementer.py`` without re-importing the
    helper. Kept lockstep with that function — both produce
    ``AUTO-IMP-PR-{number}`` for PR work.
    """
    group = _work_group_for_item(item, group_name)
    number = int(item.get("number") or 0)
    suffix = "PR" if group.item_kind == "pr" else "ISSUE"
    return f"{group.tag_prefix}-{suffix}-{number}"


def _dispatch_post_session_action(
    cfg: Any,
    item: dict[str, Any],
    parsed_json: dict[str, Any] | None,
    raw_response: str,
    terminal_state: str,
    *,
    session_context: Any | None = None,
    **kwargs: Any,
) -> dict[str, Any]:
    """Route to the legacy or the escalation-aware post-session
    action based on the runtime flag.

    Checked at CALL TIME (not import time) so test code can
    monkeypatch the env var per-test and toggle the path. The
    fallback to the legacy path is unconditional when the flag is
    OFF, preserving byte-equivalence with the pre-feature build.
    """
    if _is_escalation_enabled():
        return _post_session_action_with_escalation(
            cfg, item, parsed_json, raw_response, terminal_state,
            session_context=session_context,
            **kwargs,
        )
    return _post_session_action(
        cfg, item, parsed_json, raw_response, terminal_state,
        session_context=session_context,
        **kwargs,
    )


WORK_GROUPS = [
    _dispatch.WorkGroup(
        name="failing_ci_pr",
        script_name="list_prs_ci_failing",
        item_kind="pr",
        claim_kind=CLAIM_KIND,
        worker_agent="implementation-worker",
        tag_prefix="AUTO-IMP",
        prompt_factory=_implementation_prompt_dispatch,
        post_session_action=_dispatch_post_session_action,
    ),
    _dispatch.WorkGroup(
        name="request_changes_pr",
        script_name="list_prs_changes_requested",
        item_kind="pr",
        claim_kind=CLAIM_KIND,
        worker_agent="implementation-worker",
        tag_prefix="AUTO-IMP",
        prompt_factory=_implementation_prompt_dispatch,
        post_session_action=_dispatch_post_session_action,
    ),
    _dispatch.WorkGroup(
        name="new_issue",
        script_name="list_issues",
        item_kind="issue",
        claim_kind=None,
        worker_agent="implementation-worker",
        tag_prefix="AUTO-IMP",
        prompt_factory=_implementation_prompt_dispatch,
        post_session_action=_dispatch_post_session_action,
    ),
]


def load_config(*, dry_run: bool = False) -> Any:
    token = _dispatch.load_secret("FORGEJO_PAT", "GITEA_TOKEN")
    return _dispatch.DispatchConfig(
        token=token,
        forgejo_url=_dispatch.derive_forgejo_url(),
        owner=os.environ.get("FORGEJO_OWNER", _dispatch.REPO_OWNER),
        repo=os.environ.get("FORGEJO_REPO", _dispatch.REPO_NAME),
        server_url=os.environ.get("OPENCODE_SERVER_URL", "http://127.0.0.1:4096").rstrip(
            "/"
        ),
        lock_path=_dispatch.resolve_lock_or_heartbeat(
            "IMPLEMENTER_DISPATCHER_LOCK_PATH", "implementer-dispatcher.lock"
        ),
        heartbeat_path=_dispatch.resolve_lock_or_heartbeat(
            "IMPLEMENTER_DISPATCHER_HEARTBEAT_PATH",
            "implementer-dispatcher.heartbeat",
        ),
        cycle_interval_seconds=int(
            os.environ.get("IMPLEMENTER_DISPATCHER_CYCLE_SECONDS", "120")
        ),
        max_items_per_cycle=int(
            os.environ.get("IMPLEMENTER_DISPATCHER_MAX_ITEMS_PER_CYCLE", "1")
        ),
        worker_timeout_seconds=int(
            os.environ.get("IMPLEMENTER_DISPATCHER_WORKER_TIMEOUT_SECONDS", "7200")
        ),
        claim_ttl_seconds=int(
            os.environ.get("IMPLEMENTER_DISPATCHER_CLAIM_TTL_SECONDS", "7200")
        ),
        api_retries=int(os.environ.get("IMPLEMENTER_DISPATCHER_API_RETRIES", "3")),
        request_timeout_s=int(
            os.environ.get("IMPLEMENTER_DISPATCHER_REQUEST_TIMEOUT_S", "30")
        ),
        script_timeout_seconds=int(
            os.environ.get("IMPLEMENTER_DISPATCHER_SCRIPT_TIMEOUT_SECONDS", "120")
        ),
        table_name="dispatch_implementer_cycles",
        dry_run=dry_run,
        cycle_failure_budget=int(
            os.environ.get("IMPLEMENTER_DISPATCHER_CYCLE_FAILURE_BUDGET", "5")
        ),
        linked_issue_policy=_review_fetch.normalize_linked_issue_policy(
            os.environ.get("IMPLEMENTER_DISPATCHER_LINKED_ISSUE_POLICY"),
        ),
    )


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--once", action="store_true", help="run one cycle and exit"
    )
    parser.add_argument(
        "--status", action="store_true", help="print config and exit"
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="claim nothing and do not dispatch",
    )
    args = parser.parse_args()
    _dispatch._configure_logging("IMPLEMENTER_DISPATCHER_LOG_LEVEL")
    cfg = load_config(dry_run=args.dry_run)
    if args.status:
        _dispatch.json_line(_dispatch.status_payload(cfg, driver_name=DRIVER_NAME))
        return 0
    if not args.dry_run:
        # Worktree janitor (R3, 2026-05-16): sweep stale + corrupted
        # implementer worktrees BEFORE the main loop runs a new
        # cycle. Prior runs that SIGTERM-terminated mid-session
        # leave orphaned worktree dirs whose mirror bookkeeping
        # collides with the next ``git worktree add``; the janitor
        # + the in-prepare auto-prune-on-failure together close
        # that loop. See PR #29's 4-orphan accumulation in
        # run-15/16/17/18 for the live evidence.
        try:
            _pr_clone.prune_orphan_worktrees(cfg, kind="implementer")
        except Exception as exc:  # noqa: BLE001
            _logger.warning(
                "implementer worktree janitor raised at startup; "
                "continuing: %s: %s", type(exc).__name__, exc,
            )
    if args.once:
        _dispatch.json_line(
            _dispatch.run_one_cycle(
                cfg,
                WORK_GROUPS,
                driver_name=DRIVER_NAME,
                sweep_claim_kind=CLAIM_KIND,
            )
        )
        return 0
    _dispatch.run_outer_loop(
        cfg,
        WORK_GROUPS,
        driver_name=DRIVER_NAME,
        sweep_claim_kind=CLAIM_KIND,
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())