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")
_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_prompt = _load_sibling(
    "_implementer_prompt", "_implementer_prompt.py"
)
_pr_context_sentinel = _load_sibling(
    "_pr_context_sentinel", "_pr_context_sentinel.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"


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_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 _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)
    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).
    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,
        )
    except Exception as e:  # noqa: BLE001 — best-effort
        _logger.warning(
            "PR context sentinel write failed for PR #%s: %s",
            pr_number, e,
        )
    return text


def _implementation_prompt_dispatch(
    cfg: Any, item: dict[str, Any], group: Any
) -> str:
    """Top-level prompt factory honouring the
    ``IMPLEMENTER_DISPATCHER_PREFETCH`` flag. ``=1`` produces the
    Phase 2 prompt; anything else produces the legacy prompt."""
    if _is_prefetch_enabled():
        return _prefetch_prompt(cfg, item, group)
    return _legacy_implementation_prompt(cfg, item, group)


# ─── 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,
) -> 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,
        )
    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)
    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,
) -> 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,
        )
        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,
) -> 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
    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,
    }
    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,
    )
    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


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


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 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())