cleveragents-core/tools/_dispatch_runtime.py

"""Shared runtime for deterministic review / implementation dispatchers.

This module is intentionally small: it reuses the Phase A
``_claim_runtime.py`` and ``_opencode_worker.py`` primitives, runs the
existing ``list_prs_*`` TypeScript wrappers, claims PR work before dispatch,
and records one SQLite row per dispatched item.  The LLM workers still make
the review / implementation decisions; Python owns only orchestration.
"""
from __future__ import annotations

import json
import logging
import os
import signal
import sqlite3
import subprocess
import sys
import threading
import time
import uuid
from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any

# Make ``tools/`` importable so we can pull the shared sibling loader by
# regular import rather than reimplementing the file-load dance inline.
# Both ``python tools/...`` invocation (sys.path[0] is already tools/) and
# pytest's ``spec_from_file_location`` loader (no implicit tools/ on path)
# work after this idempotent insert.
_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,
)

logger = logging.getLogger("dispatch_runtime")


_claim_runtime = _load_sibling("_claim_runtime", "_claim_runtime.py")
_forgejo_cache = _load_sibling("_forgejo_cache", "_forgejo_cache.py")
_opencode_worker = _load_sibling("_opencode_worker", "_opencode_worker.py")


REPO_ROOT = Path(__file__).resolve().parent.parent
SCRIPT_DIR = REPO_ROOT / ".opencode" / "skills" / "auto-agents-system" / "scripts"

API_BASE = _claim_runtime.API_BASE
REPO_OWNER = _claim_runtime.REPO_OWNER
REPO_NAME = _claim_runtime.REPO_NAME

CLAIM_COMMENT_MARKER = "<!-- claim_pr.ts: do-not-edit -->"

# Cap on the worker raw-response excerpt we quote into a Forgejo claim
# release comment. Worker output can be arbitrarily long and may contain
# tool-call traces, escape sequences, or partial Markdown that would
# break out of the surrounding code fence; truncate before sanitizing.
_RELEASE_DETAIL_MAX_CHARS = 500


def _sanitize_release_detail(raw: str | None) -> str:
    """Make a best-effort safe excerpt of a worker's raw response for
    inclusion in a Forgejo release comment.

    The release comment is operator-facing (and lands in a public
    Forgejo timeline), so we want a deterministic, readable, fence-safe
    excerpt rather than a verbatim dump. This:

    1. Returns ``""`` when the input is empty / None.
    2. Truncates to ``_RELEASE_DETAIL_MAX_CHARS`` *before* the rest of
       the cleanup so we never spend cycles on multi-MB inputs.
    3. Strips control characters except newline and tab — a stray
       carriage return or escape sequence could otherwise corrupt the
       comment renderer.
    4. Replaces backtick runs of length >= 3 so a worker emitting a
       triple-backtick code fence cannot close out the surrounding
       code fence in the release comment template.

    Note: sanitization is operator-facing hygiene, not a security
    control. The comment author identity is the bot, not the worker;
    nothing in the excerpt is interpreted as code.
    """
    if not raw:
        return ""
    excerpt = raw[:_RELEASE_DETAIL_MAX_CHARS]
    cleaned = "".join(
        ch
        for ch in excerpt
        if ch in ("\n", "\t") or (0x20 <= ord(ch) < 0x7F) or ord(ch) > 0x7F
    )
    # Single replace is sufficient: the substitute (modifier-letter
    # apostrophes) contains no backticks, so it can never produce a
    # new "```" run for a second pass to find.
    return cleaned.replace("```", "ʼʼʼ")


@dataclass(frozen=True)
class WorkGroup:
    name: str
    script_name: str
    item_kind: str
    claim_kind: str | None
    worker_agent: str
    tag_prefix: str
    prompt_factory: Callable[[DispatchConfig, dict[str, Any], WorkGroup], str]
    # Optional hook invoked by ``dispatch_one`` after the worker
    # session completes (and before the claim release). Positional
    # arguments: ``cfg``, ``item``, the parsed JSON the session
    # emitted (or None), the raw response text, the dispatcher-
    # derived ``terminal_state`` ("completed" / "timeout" /
    # "transport-error" / "dry-run"). Plus a single keyword-only
    # ``session_context: SessionContext`` carrying the work-group
    # name and the ISO-8601 timestamps + wall-clock that bracket
    # ``run_session_blocking`` (see :class:`SessionContext`).
    #
    # Hooks that don't need the context accept ``**_`` to absorb
    # the keyword. The dispatcher always passes ``session_context``;
    # ``**_kwargs`` only exists for back-compat with hooks written
    # against an earlier API shape.
    #
    # Returns a dict that gets merged into the dispatch outcome
    # under ``post_session_result``. Errors must be caught by the
    # action — exceptions propagate and would orphan the claim
    # release. The reviewer dispatcher uses this hook to POST the
    # worker's review verdict to Forgejo; the implementer dispatcher
    # uses it for cleanup + telemetry + status comments.
    post_session_action: (
        Callable[..., dict[str, Any]] | None
    ) = None


@dataclass(frozen=True)
class SessionContext:
    """Per-session context the dispatcher captures around
    ``run_session_blocking`` and threads into
    :meth:`WorkGroup.post_session_action`.

    Packing these four kwargs into a single dataclass replaces the
    earlier "pile of keyword-only arguments" call shape. The dataclass
    is intentionally frozen + slot-light: the dispatcher owns
    construction and the post-session action treats it as a read-only
    record.

    Fields:

    - ``work_group_name`` — canonical group name (one of
      ``failing_ci_pr`` / ``request_changes_pr`` / ``new_issue`` for
      the implementer; the reviewer-side groups follow the same
      convention). Equal to ``WorkGroup.name``.
    - ``session_started_at`` / ``session_completed_at`` — ISO-8601 UTC
      timestamps bracketing :func:`_opencode_worker.run_session_blocking`.
      Phase 4 telemetry's ``start_ts`` / ``end_ts`` keys derive from
      these so they reflect cycle wall-clock.
    - ``session_wallclock_seconds`` —
      ``SessionResult.wallclock_seconds`` echoed through; redundant
      with ``end - start`` but kept for cross-checking against the
      worker's self-reported timing.
    """

    work_group_name: str
    session_started_at: str
    session_completed_at: str
    session_wallclock_seconds: float


@dataclass(frozen=True)
class DispatchConfig:
    token: str
    forgejo_url: str
    owner: str
    repo: str
    server_url: str
    lock_path: Path
    heartbeat_path: Path
    cycle_interval_seconds: int
    max_items_per_cycle: int
    worker_timeout_seconds: int
    claim_ttl_seconds: int
    api_retries: int
    request_timeout_s: int
    script_timeout_seconds: int
    table_name: str
    dry_run: bool = False
    cycle_failure_budget: int = 5


def _configure_logging(env_var: str) -> None:
    level_name = os.environ.get(env_var, "INFO").upper()
    level = getattr(logging, level_name, logging.INFO)
    if not logging.getLogger().handlers:
        logging.basicConfig(
            level=level,
            format="%(asctime)s %(name)s %(levelname)s %(message)s",
            stream=sys.stderr,
        )
    else:
        logging.getLogger().setLevel(level)


def _read_dotenv_value(name: str) -> str | None:
    for path in (REPO_ROOT / ".devcontainer" / ".env", REPO_ROOT / ".env"):
        if not path.exists():
            continue
        for line in path.read_text().splitlines():
            stripped = line.strip()
            if not stripped or stripped.startswith("#") or "=" not in stripped:
                continue
            key, value = stripped.split("=", 1)
            if key.strip() == name:
                return value.strip().strip('"').strip("'")
    return None


def load_secret(*names: str) -> str:
    for name in names:
        value = os.environ.get(name) or _read_dotenv_value(name)
        if value:
            return value
    joined = " / ".join(names)
    raise SystemExit(f"missing required token ({joined})")


def derive_forgejo_url() -> str:
    explicit = os.environ.get("FORGEJO_URL")
    if explicit:
        return explicit.rstrip("/")
    if API_BASE.endswith("/api/v1"):
        return API_BASE[: -len("/api/v1")]
    return API_BASE.rstrip("/")


def resolve_lock_or_heartbeat(env_var: str, basename: str) -> Path:
    explicit = os.environ.get(env_var)
    if explicit:
        return Path(explicit)
    candidates: list[Path | None] = [
        Path("/var/run") / basename,
        Path(os.environ.get("XDG_RUNTIME_DIR", "")) / basename
        if os.environ.get("XDG_RUNTIME_DIR")
        else None,
        Path("/tmp") / basename,
    ]
    for candidate in candidates:
        if candidate is None:
            continue
        try:
            candidate.parent.mkdir(parents=True, exist_ok=True)
            candidate.touch(exist_ok=True)
            return candidate
        except (OSError, PermissionError):
            continue
    return Path("/tmp") / basename


def run_list_script(
    script_name: str,
    cfg: DispatchConfig,
    *,
    token: str | None = None,
) -> list[dict[str, Any]]:
    script_path = SCRIPT_DIR / f"{script_name}.ts"
    if not script_path.exists():
        raise RuntimeError(f"work-group script does not exist: {script_path}")
    cmd = [
        "npx",
        "--yes",
        "tsx",
        str(script_path),
        "--url",
        cfg.forgejo_url,
        "--pat",
        token or cfg.token,
        "--owner",
        cfg.owner,
        "--repo",
        cfg.repo,
    ]
    proc = subprocess.run(
        cmd,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=cfg.script_timeout_seconds,
        check=False,
    )
    if proc.returncode != 0:
        raise RuntimeError(
            f"{script_name} exited {proc.returncode}: {proc.stderr.strip()}"
        )
    try:
        payload = json.loads(proc.stdout or "[]")
    except json.JSONDecodeError as exc:
        raise RuntimeError(
            f"{script_name} emitted invalid JSON: {proc.stdout[:500]!r}"
        ) from exc
    if not isinstance(payload, list):
        raise RuntimeError(f"{script_name} emitted non-list JSON")
    return [item for item in payload if isinstance(item, dict)]


def collect_candidates(
    cfg: DispatchConfig,
    groups: list[WorkGroup],
) -> tuple[list[tuple[WorkGroup, dict[str, Any]]], dict[str, int]]:
    candidates: list[tuple[WorkGroup, dict[str, Any]]] = []
    counts: dict[str, int] = {}
    seen: set[tuple[str, int]] = set()
    for group in groups:
        items = run_list_script(group.script_name, cfg)
        counts[group.name] = len(items)
        for item in items:
            number = _item_number(item)
            if number is None:
                continue
            key = (group.item_kind, number)
            if key in seen:
                continue
            seen.add(key)
            candidates.append((group, item))
    return candidates, counts


def _item_number(item: dict[str, Any]) -> int | None:
    try:
        number = int(item.get("number"))
    except (TypeError, ValueError):
        return None
    return number if number > 0 else None


def _claim_label(kind: str) -> str:
    return f"auto/claimed-{kind}"


def _existing_claim_labels(
    number: int, cfg: DispatchConfig
) -> tuple[list[str] | None, int]:
    """Return ``(labels, status)`` for issue/PR ``number``.

    ``labels`` is the list of every ``auto/claimed-*`` label currently
    attached, or ``None`` when the fetch failed at a level that means we
    cannot trust "no claim is present" (auth, repo-gone, persistent
    5xx). ``status`` is the HTTP status code from
    ``_claim_runtime.idempotent_get`` (already retried for 5xx /
    transport errors).

    The caller refuses the claim on ``None`` instead of proceeding,
    closing the latent bug where a 401 / 403 silently let us POST a
    claim label we had no business attaching.
    """
    response = _claim_runtime.get(
        f"/repos/{cfg.owner}/{cfg.repo}/issues/{number}/labels",
        cfg,
    )
    status = int(response.get("status") or 0)
    if status != 200:
        return None, status
    body = response.get("body")
    if not isinstance(body, list):
        return None, status
    return (
        [
            str(entry.get("name"))
            for entry in body
            if isinstance(entry, dict)
            and isinstance(entry.get("name"), str)
            and str(entry.get("name")).startswith("auto/claimed-")
        ],
        status,
    )


def claim_work_item(
    number: int,
    cfg: DispatchConfig,
    *,
    claim_kind: str,
    driver_name: str,
) -> dict[str, Any]:
    """Attempt to acquire the ``auto/claimed-{claim_kind}`` label on
    ``number``.

    Forgejo's label-add API is idempotent — it returns 200 whether the
    label was newly attached or already present — so we cannot
    distinguish ownership at the API layer alone. To avoid releasing a
    claim we did not acquire (and thereby disrupting another worker's
    in-flight session), we pre-check the issue's labels and refuse when
    *any* ``auto/claimed-*`` label is already present. This still has a
    TOCTOU race window (a sibling driver could attach the label between
    our GET and POST), which we accept and document in the same way
    ``conflict_drive.py`` § 3.3.1 documents the same race; the worker
    layer's idempotent claim helper deduplicates work even in the rare
    racing case.
    """
    if cfg.dry_run:
        return {"applied": False, "dry_run": True, "number": number}
    existing, label_status = _existing_claim_labels(number, cfg)
    if existing is None:
        logger.warning(
            "skip claim of #%s — labels GET returned status=%s (cannot "
            "verify ownership; treating as foreign claim)",
            number,
            label_status,
        )
        return {
            "applied": False,
            "number": number,
            "reason": "labels-fetch-failed",
            "label_fetch_status": label_status,
        }
    if existing:
        logger.info(
            "skip claim of #%s — already-claimed labels=%s",
            number,
            existing,
        )
        return {
            "applied": False,
            "number": number,
            "reason": "already-claimed",
            "existing_labels": existing,
        }
    label = _claim_label(claim_kind)
    if not _claim_runtime._add_label(number, label, cfg):
        return {"applied": False, "number": number, "reason": "label-not-found"}
    ttl_until = (
        datetime.now(UTC) + timedelta(seconds=cfg.claim_ttl_seconds)
    ).isoformat()
    body = (
        f"{CLAIM_COMMENT_MARKER}\n\n"
        f"Claimed by `{driver_name}` (pid {os.getpid()}) until `{ttl_until}`.\n\n"
        "This claim is advisory and will be released when the worker exits, "
        "or after the TTL by a sibling driver's expired-claim sweep."
    )
    _claim_runtime.post(
        f"/repos/{cfg.owner}/{cfg.repo}/issues/{number}/comments",
        cfg,
        {"body": body},
    )
    return {"applied": True, "number": number, "ttl_until": ttl_until}


# ─── In-flight-claim registry for signal-handler cleanup (Phase 5a) ─────────
#
# When the dispatcher receives SIGTERM / SIGINT mid-cycle the worker session
# can still be 30+ minutes from completion, but the operator (systemd, a
# launcher script, or a human pressing Ctrl-C) expects the ``auto/claimed-*``
# label to clear quickly so the next dispatcher invocation — or a sibling
# bot — can pick the work up. We record the in-flight claim metadata in a
# module-level slot so the signal handler can release the claim
# synchronously without waiting for the worker session to finish.
#
# Threading note: the dispatcher is single-threaded (workers run via
# blocking subprocess / blocking HTTP poll), so a plain dict + lock is
# sufficient. ``threading.Lock`` is reentrant-safe enough for the
# Python-level signal-delivery model — handlers run between bytecodes
# in the main thread, so a brief lock-then-clear-then-release sequence
# under the lock cannot deadlock against itself. We deliberately keep
# the critical section tiny (snapshot + flag flip) so the signal
# handler doesn't hold the lock during the actual HTTP call.
_INFLIGHT_CLAIM_LOCK = threading.Lock()
_INFLIGHT_CLAIM: dict[str, Any] | None = None
# Stop event registered by ``run_outer_loop`` / ``run_one_cycle``-via-CLI;
# the signal handler sets it so the outer loop exits cleanly after the
# in-flight cycle returns.
_REGISTERED_STOP_EVENT: Any | None = None
# Records whether ``install_signal_handlers`` has already run in this
# process so a second call (e.g. ``run_outer_loop`` after a one-shot
# ``run_one_cycle`` in the same process for a test) is a no-op rather
# than overwriting the previous SIGTERM handler.
_SIGNAL_HANDLERS_INSTALLED = False


def _register_inflight_claim(
    *,
    number: int,
    cfg: DispatchConfig,
    claim_kind: str,
    driver_name: str,
) -> None:
    """Stamp the in-flight claim metadata so the signal handler can
    release it before the dispatcher exits. Called by ``dispatch_one``
    after a successful claim, paired with
    :func:`_deregister_inflight_claim` in the finally block.
    """
    global _INFLIGHT_CLAIM
    with _INFLIGHT_CLAIM_LOCK:
        _INFLIGHT_CLAIM = {
            "number": number,
            "cfg": cfg,
            "claim_kind": claim_kind,
            "driver_name": driver_name,
        }


def _deregister_inflight_claim() -> dict[str, Any] | None:
    """Pop and return the in-flight claim metadata. Called by
    ``dispatch_one``'s finally before the regular ``release_work_item``
    runs so the signal handler does not double-release. Returns the
    previous registration (or ``None`` if the signal handler already
    cleared it).
    """
    global _INFLIGHT_CLAIM
    with _INFLIGHT_CLAIM_LOCK:
        previous = _INFLIGHT_CLAIM
        _INFLIGHT_CLAIM = None
        return previous


def _signal_release_handler(signum: int, _frame: Any) -> None:
    """SIGTERM / SIGINT handler: release any in-flight claim and request
    cooperative shutdown, then re-raise as SystemExit so the outer
    loop's ``finally lock.release()`` runs.

    Sequence:

    1. Snapshot ``_INFLIGHT_CLAIM`` under the lock and atomically clear
       it. This is the slot ``dispatch_one``'s finally block
       *would* otherwise read; clearing here means the finally path
       sees ``None`` and does NOT post a second release comment.
    2. Set the registered stop event so the outer loop exits between
       cycles (or, if we re-raise into the cycle, the cycle's
       exception handler counts it against the failure budget).
    3. Issue the release HTTP call. We log + swallow exceptions so a
       transient network blip during shutdown can't loop forever.
    4. Re-raise as ``SystemExit`` with code 130 (SIGINT convention)
       or 143 (SIGTERM convention) so the process exit reflects the
       signal that triggered it.

    Re-raising rather than returning matters because the in-flight
    worker session is blocked on an HTTP poll inside
    ``_opencode_worker.run_session_blocking``; that loop checks the
    stop event at most once per poll interval. SystemExit unwinds
    through every ``try: ... finally:`` in the call chain (including
    the worker poll loop's), so the dispatcher exits within seconds
    rather than waiting for the worker timeout.
    """
    global _INFLIGHT_CLAIM
    snapshot: dict[str, Any] | None
    with _INFLIGHT_CLAIM_LOCK:
        snapshot = _INFLIGHT_CLAIM
        # Clear the slot under the lock so a concurrent
        # ``dispatch_one`` finally that runs after we drop the lock
        # but before we issue the HTTP call sees ``None`` and skips
        # its own release. The HTTP call itself is intentionally
        # OUTSIDE the lock — it can take seconds and we don't want
        # to block the lock that long. A follow-up signal during
        # the release would re-snapshot ``None`` and skip cleanly.
        if snapshot is not None:
            _INFLIGHT_CLAIM = None
    if _REGISTERED_STOP_EVENT is not None:
        try:
            _REGISTERED_STOP_EVENT.set()
        except Exception:
            logger.exception(
                "signal handler: setting stop event raised; "
                "continuing with claim release"
            )
    # Resolve the human-readable signal name once so both the
    # logger output AND the Forgejo release-comment ``terminal_state``
    # field show ``signal-SIGTERM`` / ``signal-SIGINT`` rather than
    # the bare numeric ``signal-15`` / ``signal-2``. The ``Signals``
    # enum was added in Python 3.5; falling back to the numeric
    # form on platforms with bespoke signal numbers (Windows,
    # mostly) keeps the dispatcher portable.
    try:
        signal_name = signal.Signals(signum).name
    except (ValueError, AttributeError):
        signal_name = f"sig{int(signum)}"
    if snapshot is not None:
        try:
            release_work_item(
                snapshot["number"],
                snapshot["cfg"],
                claim_kind=snapshot["claim_kind"],
                driver_name=snapshot["driver_name"],
                terminal_state=f"signal-{signal_name}",
                detail=(
                    f"released by dispatcher signal handler "
                    f"({signal_name}, signum={signum}); cooperative "
                    f"shutdown initiated"
                ),
            )
            logger.warning(
                "signal handler released claim for #%s (%s, signum=%s)",
                snapshot["number"],
                signal_name,
                signum,
            )
        except Exception as exc:
            logger.exception(
                "signal handler: release_work_item failed for #%s: %s",
                snapshot["number"],
                exc,
            )
    # POSIX convention: ``128 + signum``. SIGINT (=2) → 130,
    # SIGTERM (=15) → 143; everything else picks up the same rule.
    raise SystemExit(128 + int(signum))


def install_signal_handlers(stop: Any) -> None:
    """Install SIGTERM / SIGINT handlers for cooperative dispatcher shutdown.

    Idempotent: a second call from the same process is a no-op (the
    test suite runs many ``run_outer_loop`` invocations against
    different fake clocks; we don't want each to clobber the
    handler from the previous test).

    The handler is intentionally installed in the main thread by the
    main thread — Python's ``signal`` module raises ``ValueError`` if
    called from a non-main thread, so a future architecture that
    wraps ``run_outer_loop`` in a thread would need to install the
    handler before spawning the thread.
    """
    global _REGISTERED_STOP_EVENT, _SIGNAL_HANDLERS_INSTALLED
    _REGISTERED_STOP_EVENT = stop
    if _SIGNAL_HANDLERS_INSTALLED:
        return
    try:
        signal.signal(signal.SIGTERM, _signal_release_handler)
        signal.signal(signal.SIGINT, _signal_release_handler)
        _SIGNAL_HANDLERS_INSTALLED = True
    except ValueError as exc:
        # Hit when called from a non-main thread. The dispatcher's
        # production paths always install from the main thread, but
        # the test suite occasionally invokes ``run_outer_loop`` from
        # a worker thread (e.g. when a fixture wraps it for
        # parallelism). We log + continue without handlers in that
        # case rather than failing the whole loop — the test would
        # have asserted the handler installation explicitly if it
        # cared.
        logger.warning(
            "install_signal_handlers: signal.signal raised %s "
            "(probably called from non-main thread); continuing "
            "without dispatcher signal handlers",
            exc,
        )


def release_work_item(
    number: int,
    cfg: DispatchConfig,
    *,
    claim_kind: str,
    driver_name: str,
    terminal_state: str,
    detail: str = "",
) -> dict[str, Any]:
    if cfg.dry_run:
        return {"released": False, "dry_run": True, "number": number}
    label = _claim_label(claim_kind)
    removed = _claim_runtime._remove_label(number, label, cfg)
    body = (
        f"{CLAIM_COMMENT_MARKER}\n\n"
        f"Released by `{driver_name}` (pid {os.getpid()}). "
        f"terminal_state=`{terminal_state}`"
        + (f"\n\nDetail: {detail}" if detail else "")
    )
    _claim_runtime.post(
        f"/repos/{cfg.owner}/{cfg.repo}/issues/{number}/comments",
        cfg,
        {"body": body},
    )
    return {"released": removed, "number": number}


def sweep_own_claims(cfg: DispatchConfig, claim_kind: str) -> list[int]:
    return _claim_runtime.sweep_expired_claims(
        cfg,
        session_pr_numbers=set(),
        label=_claim_label(claim_kind),
    )


def dispatch_one(
    cfg: DispatchConfig,
    group: WorkGroup,
    item: dict[str, Any],
    *,
    driver_name: str,
) -> dict[str, Any]:
    """Claim, dispatch, and release a single work item.

    Terminal states (recorded as ``terminal_state`` in telemetry):

    - ``invalid-item`` — the work-group script returned an item with no
      usable ``number``.
    - ``already-claimed`` — another worker already owns an
      ``auto/claimed-*`` label on this item; we did NOT acquire a claim
      and do NOT release one.
    - ``labels-fetch-failed`` — the labels GET returned non-200 (auth /
      permission / repo-gone). Treated as a foreign claim: refused
      rather than silently attaching a label we have no permission to
      manage. The HTTP status is captured in
      ``claim_result.label_fetch_status`` for operator triage.
    - ``claim-failed`` — Forgejo accepted the labels GET but refused
      our label-add (label undefined in the repo, etc.). No release
      attempted.
    - ``dry-run`` — ``cfg.dry_run`` is set; no worker was dispatched.
    - ``completed`` — the OpenCode session reached idle without timing
      out or hitting a transport error. The worker's domain-specific
      result (review submitted, code generated, etc.) is judged by side
      effects, not by a JSON outcome key. Reviewer / implementer
      workers do NOT emit the conflict-driver JSON schema.
    - ``timeout`` / ``transport-error`` — propagated verbatim from
      :class:`_opencode_worker.SessionResult.status`.

    The release in the finally block runs only when ``claimed`` is
    true (i.e. :func:`claim_work_item` returned ``applied=True``). This
    is what makes ``already-claimed`` and ``labels-fetch-failed`` safe:
    we never remove a label we did not just attach.
    """
    number = _item_number(item)
    if number is None:
        return {"terminal_state": "invalid-item", "item": item}
    claimed = False
    claim_result: dict[str, Any] | None = None
    started = time.monotonic()
    session = None
    terminal_state = "unknown"
    try:
        if group.claim_kind is not None:
            claim_result = claim_work_item(
                number,
                cfg,
                claim_kind=group.claim_kind,
                driver_name=driver_name,
            )
            claimed = bool(claim_result.get("applied"))
            if not claimed and not cfg.dry_run:
                reason = claim_result.get("reason")
                if reason == "already-claimed":
                    terminal_state = "already-claimed"
                elif reason == "labels-fetch-failed":
                    terminal_state = "labels-fetch-failed"
                else:
                    terminal_state = "claim-failed"
                return {
                    "terminal_state": terminal_state,
                    "claim_result": claim_result,
                    "item_number": number,
                }
            # Register the in-flight claim so a SIGTERM / SIGINT
            # handler can release it before this dispatcher exits.
            # Skipped on dry-run (no real claim was acquired) and on
            # the ``new_issue`` work group whose ``claim_kind`` is
            # ``None`` (the outer ``if group.claim_kind is not None``
            # already gates this branch correctly). The matching
            # deregister fires in the finally block before
            # ``release_work_item`` so the signal handler's
            # synchronous release and the finally's release cannot
            # both post release comments.
            if claimed and not cfg.dry_run:
                _register_inflight_claim(
                    number=number,
                    cfg=cfg,
                    claim_kind=group.claim_kind,
                    driver_name=driver_name,
                )
        prompt = group.prompt_factory(cfg, item, group)
        if cfg.dry_run:
            terminal_state = "dry-run"
            return {
                "terminal_state": terminal_state,
                "claim_result": claim_result,
                "item_number": number,
                "prompt": prompt,
            }
        tag = _tag_for(group, number)
        # Refresh our liveness heartbeat once per polling iteration so a
        # 30-minute review session doesn't look like a hung dispatcher
        # process to the launcher / systemd unit. Without this, the
        # watchdog above the dispatcher would SIGTERM us mid-cycle and
        # orphan both the OpenCode session and the auto/claimed-* lock
        # we hold on this PR. Errors from the callback are swallowed
        # inside run_session_blocking so a transient disk-full / EROFS
        # on the heartbeat write cannot mask a successful worker
        # completion.
        heartbeat_path = cfg.heartbeat_path

        def _refresh_heartbeat() -> None:
            _claim_runtime.write_heartbeat(heartbeat_path)

        # Bracket ``run_session_blocking`` with ISO-8601 UTC
        # timestamps so the post-session hook (Phase 4 telemetry,
        # specifically) can record actual cycle wall-clock instead
        # of post-hoc ``now_iso()`` placeholders. The worker's
        # self-reported ``session.wallclock_seconds`` is also
        # threaded through for cross-checking.
        session_started_at = _now()
        session = _opencode_worker.run_session_blocking(
            server_url=cfg.server_url,
            agent=group.worker_agent,
            tag=tag,
            prompt=prompt,
            timeout_seconds=cfg.worker_timeout_seconds,
            on_poll=_refresh_heartbeat,
        )
        session_completed_at = _now()
        if session.status == "completed":
            terminal_state = "completed"
        else:
            terminal_state = session.status  # "timeout" | "transport-error"
        worker_outcome: str | None = None
        if isinstance(session.parsed_json, dict):
            outcome_value = session.parsed_json.get("outcome")
            if isinstance(outcome_value, str) and outcome_value:
                worker_outcome = outcome_value
        post_session_result: dict[str, Any] | None = None
        if group.post_session_action is not None:
            # Pack the work-group name and session-timing context
            # into a single ``SessionContext`` dataclass so the post-
            # session action sees a single read-only record instead
            # of a fan of four loose kwargs (and so future fields
            # land in one place). Earlier iteration stamped these
            # onto ``item`` directly, then a flat ``**kwargs`` —
            # both made the data flow hard to follow.
            session_context = SessionContext(
                work_group_name=group.name,
                session_started_at=session_started_at,
                session_completed_at=session_completed_at,
                session_wallclock_seconds=session.wallclock_seconds,
            )
            try:
                post_session_result = group.post_session_action(
                    cfg,
                    item,
                    session.parsed_json,
                    session.raw_response,
                    terminal_state,
                    session_context=session_context,
                )
            except Exception as exc:
                logger.exception(
                    "%s post_session_action raised on item #%s: %s",
                    group.name,
                    number,
                    exc,
                )
                post_session_result = {
                    "review_action": "failed",
                    "review_action_reason": (
                        f"post_session_action_exception: "
                        f"{type(exc).__name__}: {exc}"
                    ),
                }
        return {
            "terminal_state": terminal_state,
            "session_status": session.status,
            "worker_outcome": worker_outcome,
            "session_id": session.session_id,
            "worker_wallclock_seconds": session.wallclock_seconds,
            "raw_response": session.raw_response,
            "parsed_json": session.parsed_json,
            "claim_result": claim_result,
            "item_number": number,
            "post_session_result": post_session_result,
            # Surfaces ``error_kind`` from _opencode_worker so the
            # cycle log can pivot on transport-create-session vs
            # transport-poll-status vs startup-grace-elapsed without
            # parsing free-text raw_response. ``None`` on
            # ``completed``.
            "session_error_kind": session.error_kind,
        }
    finally:
        # Deregister the in-flight claim BEFORE issuing the regular
        # release so a signal that fires between ``_register_inflight_claim``
        # and here triggers exactly one release path, not two. The
        # signal handler clears the slot and posts the release; this
        # path checks the previous registration and only posts when
        # the slot is still our claim (i.e. no signal raced us). If
        # the snapshot is None the signal handler already released —
        # we skip the duplicate release comment.
        previous_registration = _deregister_inflight_claim()
        if (
            claimed
            and group.claim_kind is not None
            and previous_registration is not None
        ):
            release_work_item(
                number,
                cfg,
                claim_kind=group.claim_kind,
                driver_name=driver_name,
                terminal_state=terminal_state,
                detail=_sanitize_release_detail(
                    session.raw_response if session is not None else ""
                ),
            )
        elapsed = time.monotonic() - started
        logger.info(
            "%s item #%s finished terminal_state=%s elapsed=%.1fs",
            group.name,
            number,
            terminal_state,
            elapsed,
        )


def _tag_for(group: WorkGroup, number: int) -> str:
    suffix = "PR" if group.item_kind == "pr" else "ISSUE"
    return f"{group.tag_prefix}-{suffix}-{number}"


def run_one_cycle(
    cfg: DispatchConfig,
    groups: list[WorkGroup],
    *,
    driver_name: str,
    sweep_claim_kind: str | None,
) -> dict[str, Any]:
    cycle_id = str(uuid.uuid4())
    started_at = _now()
    # Insert the in-flight row before any work so the telemetry console
    # can show this cycle as "running" the moment it starts. Without
    # this, the cycle would only appear after a multi-minute worker
    # session finished, leaving the dashboard looking like the
    # dispatcher had stopped working. The companion ``finish_cycle`` call
    # in the finally block UPDATEs the same row with aggregate counts.
    begin_cycle(cfg, cycle_id=cycle_id, started_at=started_at, driver_name=driver_name)
    swept: list[int] = []
    candidates: list[tuple[WorkGroup, dict[str, Any]]] = []
    group_counts: dict[str, int] = {}
    processed: list[dict[str, Any]] = []
    claims_acquired = 0
    try:
        swept = sweep_own_claims(cfg, sweep_claim_kind) if sweep_claim_kind else []
        candidates, group_counts = collect_candidates(cfg, groups)
        for group, item in candidates[: cfg.max_items_per_cycle]:
            outcome = dispatch_one(cfg, group, item, driver_name=driver_name)
            if (outcome.get("claim_result") or {}).get("applied"):
                claims_acquired += 1
            processed.append({"group": group.name, **outcome})
        return {
            "cycle_id": cycle_id,
            "started_at": started_at,
            "ended_at": None,  # set by the finally block via finish_cycle
            "candidates_count": len(candidates),
            "group_counts": group_counts,
            "claims_acquired": claims_acquired,
            "swept": swept,
            "processed": processed,
        }
    finally:
        ended_at = _now()
        # Use a defensive try/except so a finish-time write failure
        # (disk full, transient SQLite lock contention, etc.) does not
        # mask whatever exception actually drove us into the finally
        # block. The in-flight row stays in place if the UPDATE fails
        # — operators can spot the orphan via ``ended_at IS NULL`` and
        # the next ensure_cycle_table run will not delete it.
        try:
            finish_cycle(
                cfg,
                cycle_id=cycle_id,
                ended_at=ended_at,
                candidates_count=len(candidates),
                group_counts=group_counts,
                claims_acquired=claims_acquired,
                swept=swept,
                processed=processed,
                # Pass started_at + driver_name so the INSERT-fallback
                # path (which fires only if begin_cycle's row got lost
                # — e.g. SQLite WAL corruption + recovery between
                # begin_cycle and finish_cycle) still records an
                # accurate started_at and driver instead of synthesising
                # zero-duration ``unknown`` rows. Under the normal
                # UPDATE path these kwargs are unused.
                started_at=started_at,
                driver_name=driver_name,
            )
        except sqlite3.Error:
            logger.exception(
                "%s finish_cycle UPDATE failed for cycle_id=%s — orphan "
                "in-flight row will be visible in dispatch_*_cycles",
                driver_name,
                cycle_id,
            )


def run_outer_loop(
    cfg: DispatchConfig,
    groups: list[WorkGroup],
    *,
    driver_name: str,
    sweep_claim_kind: str | None,
    stop: Any | None = None,
) -> None:
    """Hold the single-instance lock and run cycles until ``stop`` fires.

    Each cycle's exceptions are logged and counted against
    ``cfg.cycle_failure_budget``. After that many consecutive failures
    we exit with code 2 — matching ``merge_drive.py`` / ``conflict_drive.py``
    so an upstream supervisor (systemd, a launcher script, etc.)
    restarts us with fresh state instead of letting a partial-output
    work-group script wedge the loop indefinitely. A successful cycle
    resets the counter.
    """
    lock = _claim_runtime.SingleInstanceLock(cfg.lock_path)
    if not lock.acquire():
        raise SystemExit(f"another {driver_name} instance holds {cfg.lock_path}")
    stop = stop or _claim_runtime.StopEvent()
    # Install SIGTERM / SIGINT handlers so an operator stop (systemd,
    # Ctrl-C, supervisor restart) releases any in-flight
    # ``auto/claimed-*`` label before the dispatcher exits. Without
    # this, the merge driver's ``sweep_expired_claims`` would only
    # release the orphan label after ``claim_ttl_seconds`` (default
    # 7200 s = 2 h) — long enough for a sibling dispatcher to wait
    # noticeably for the next cycle. Test seam: a test that calls
    # ``run_outer_loop`` with its own ``stop`` event still gets the
    # handler installation, but tests that exercise the handler
    # directly should call :func:`install_signal_handlers` themselves
    # against a controlled stop event.
    install_signal_handlers(stop)
    consecutive_failures = 0
    try:
        while not stop.is_set():
            try:
                run_one_cycle(
                    cfg,
                    groups,
                    driver_name=driver_name,
                    sweep_claim_kind=sweep_claim_kind,
                )
                consecutive_failures = 0
            except Exception:
                consecutive_failures += 1
                logger.exception(
                    "%s cycle failed (consecutive=%s/%s)",
                    driver_name,
                    consecutive_failures,
                    cfg.cycle_failure_budget,
                )
                if consecutive_failures >= cfg.cycle_failure_budget:
                    logger.error(
                        "%s exceeded cycle failure budget (%s); exiting "
                        "for supervisor restart",
                        driver_name,
                        cfg.cycle_failure_budget,
                    )
                    raise SystemExit(2)
            _claim_runtime.write_heartbeat(cfg.heartbeat_path)
            stop.sleep(cfg.cycle_interval_seconds)
    finally:
        lock.release()


def _now() -> str:
    return datetime.now(UTC).isoformat()


def ensure_cycle_table(table_name: str) -> None:
    """Create or repair the per-driver cycle table.

    Delegates to :func:`_forgejo_cache.ensure_dispatch_cycle_schema`
    so the v5 schema (``ended_at`` nullable, UNIQUE INDEX on
    ``cycle_id``) is defined in exactly one place. Both this function
    and ``ForgejoCache._migrate_to_v5_in_flight_rows`` import their
    DDL from there, eliminating drift risk if a future migration adds
    a column.

    The dispatcher writes cycle rows via raw ``sqlite3.connect`` rather
    than going through ``ForgejoCache``, so this function must run the
    creation + migration itself; ``ForgejoCache`` may not have been
    opened in this process before the first dispatcher cycle.
    """
    if table_name not in _forgejo_cache.DISPATCH_CYCLE_TABLES:
        raise ValueError(f"unexpected dispatch table: {table_name}")
    path = _forgejo_cache.DEFAULT_CACHE_PATH
    path.parent.mkdir(parents=True, exist_ok=True)
    with sqlite3.connect(path) as conn:
        conn.row_factory = sqlite3.Row
        _forgejo_cache.ensure_dispatch_cycle_schema(conn, table_name)


def begin_cycle(
    cfg: DispatchConfig,
    *,
    cycle_id: str,
    started_at: str,
    driver_name: str,
) -> None:
    """Insert an in-flight row for ``cycle_id`` (``ended_at IS NULL``).

    Pairs with :func:`finish_cycle`, which UPDATEs the same row with the
    final ``ended_at`` and aggregate fields when the cycle resolves.
    Inserting at start fixes the telemetry-blind-spot where a multi-
    minute review session would not appear anywhere in the cycle table
    until it finished — making the dispatcher look idle even though it
    was actively polling an OpenCode worker.

    The cycle row carries placeholder counts (``0``) and an empty raw
    blob until :func:`finish_cycle` overwrites them. The UNIQUE index
    on ``cycle_id`` prevents accidental double-inserts; a re-entrant
    crash-loop that attempted ``begin_cycle`` twice would surface a
    ``sqlite3.IntegrityError`` rather than silently fork the row.
    """
    ensure_cycle_table(cfg.table_name)
    raw_placeholder = json.dumps(
        {"group_counts": {}, "swept": [], "processed": [], "in_flight": True},
        sort_keys=True,
    )
    with sqlite3.connect(_forgejo_cache.DEFAULT_CACHE_PATH) as conn:
        conn.execute(
            f"""
            INSERT INTO {cfg.table_name} (
                cycle_id, started_at, ended_at, driver, candidates_count,
                claims_acquired, swept_count, processed_count, terminal_state,
                worker_outcome, session_id, worker_wallclock_seconds, raw
            ) VALUES (?, ?, NULL, ?, 0, 0, 0, 0, NULL, NULL, NULL, NULL, ?)
            """,
            (
                cycle_id,
                started_at,
                driver_name,
                raw_placeholder,
            ),
        )


def finish_cycle(
    cfg: DispatchConfig,
    *,
    cycle_id: str,
    ended_at: str,
    candidates_count: int,
    group_counts: dict[str, int],
    claims_acquired: int,
    swept: list[int],
    processed: list[dict[str, Any]],
    started_at: str | None = None,
    driver_name: str | None = None,
) -> None:
    """Overwrite the in-flight row for ``cycle_id`` with the resolved
    aggregate fields.

    Falls back to ``INSERT`` if no row exists for ``cycle_id`` — this
    handles drivers that never called :func:`begin_cycle` (e.g. test
    fixtures that build a complete row in one shot) and keeps the
    function compatible with the legacy ``record_cycle`` semantics.

    The optional ``started_at`` and ``driver_name`` kwargs feed the
    INSERT-fallback path. When provided, the synthesised row carries
    accurate timestamps + driver attribution. When omitted, the row
    uses ``ended_at`` as both ``started_at`` (collapsing wallclock
    duration to zero) and ``"unknown"`` for the driver, and the raw
    JSON blob carries a ``"synthetic_started_at": true`` marker so
    cycle-time analytics can exclude the row from duration stats.
    """
    ensure_cycle_table(cfg.table_name)
    terminal_state = None
    worker_outcome = None
    session_id = None
    worker_wallclock = None
    if processed:
        last = processed[-1]
        terminal_state = str(last.get("terminal_state") or "") or None
        worker_outcome = (
            str(last.get("worker_outcome")) if last.get("worker_outcome") else None
        )
        session_id = str(last.get("session_id")) if last.get("session_id") else None
        worker_wallclock = last.get("worker_wallclock_seconds")
    raw = {
        "group_counts": group_counts,
        "swept": swept,
        "processed": processed,
    }
    with sqlite3.connect(_forgejo_cache.DEFAULT_CACHE_PATH) as conn:
        cursor = conn.execute(
            f"""
            UPDATE {cfg.table_name}
            SET ended_at = ?,
                candidates_count = ?,
                claims_acquired = ?,
                swept_count = ?,
                processed_count = ?,
                terminal_state = ?,
                worker_outcome = ?,
                session_id = ?,
                worker_wallclock_seconds = ?,
                raw = ?
            WHERE cycle_id = ?
            """,
            (
                ended_at,
                candidates_count,
                claims_acquired,
                len(swept),
                len(processed),
                terminal_state,
                worker_outcome,
                session_id,
                worker_wallclock,
                json.dumps(raw, sort_keys=True),
                cycle_id,
            ),
        )
        if cursor.rowcount == 0:
            # No matching in-flight row — fall back to a full INSERT
            # so callers that skip begin_cycle (legacy tests,
            # crash-recovery paths that lost their begin_cycle row)
            # still get a complete record. Use ``started_at`` /
            # ``driver_name`` if the caller passed them (modern
            # ``run_one_cycle`` does); otherwise stamp the row with the
            # ``ended_at`` value for ``started_at`` (collapsing the
            # cycle to zero duration) and ``"unknown"`` for the
            # driver, plus a ``synthetic_started_at`` flag in raw so
            # cycle-time analytics can spot and exclude these rows.
            insert_started_at = started_at if started_at is not None else ended_at
            insert_driver = driver_name if driver_name is not None else "unknown"
            insert_raw = dict(raw)
            if started_at is None:
                insert_raw["synthetic_started_at"] = True
            conn.execute(
                f"""
                INSERT INTO {cfg.table_name} (
                    cycle_id, started_at, ended_at, driver, candidates_count,
                    claims_acquired, swept_count, processed_count, terminal_state,
                    worker_outcome, session_id, worker_wallclock_seconds, raw
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    cycle_id,
                    insert_started_at,
                    ended_at,
                    insert_driver,
                    candidates_count,
                    claims_acquired,
                    len(swept),
                    len(processed),
                    terminal_state,
                    worker_outcome,
                    session_id,
                    worker_wallclock,
                    json.dumps(insert_raw, sort_keys=True),
                ),
            )




def status_payload(cfg: DispatchConfig, *, driver_name: str) -> dict[str, Any]:
    ensure_cycle_table(cfg.table_name)
    return {
        "driver": driver_name,
        "owner": cfg.owner,
        "repo": cfg.repo,
        "forgejo_url": cfg.forgejo_url,
        "server_url": cfg.server_url,
        "lock_path": str(cfg.lock_path),
        "heartbeat_path": str(cfg.heartbeat_path),
        "cycle_interval_seconds": cfg.cycle_interval_seconds,
        "max_items_per_cycle": cfg.max_items_per_cycle,
        "worker_timeout_seconds": cfg.worker_timeout_seconds,
        "claim_ttl_seconds": cfg.claim_ttl_seconds,
        "cycle_failure_budget": cfg.cycle_failure_budget,
        "table_name": cfg.table_name,
        "dry_run": cfg.dry_run,
    }


def json_line(data: Any) -> None:
    print(json.dumps(data, indent=2, sort_keys=True))