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 sqlite3
import subprocess
import sys
import time
import uuid
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Callable, Iterable

# 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 load_sibling as _load_sibling  # noqa: E402  type: ignore[import-not-found]


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). Receives the
    # cfg, item, the parsed JSON the session emitted (or None), the
    # raw response text, and the session's terminal_state. Returns a
    # dict that gets merged into the dispatch outcome under
    # ``post_session_result`` for cycle telemetry. 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 leaves it None.
    post_session_action: (
        Callable[
            ["DispatchConfig", dict[str, Any], dict[str, Any] | None, str, str],
            dict[str, Any],
        ]
        | None
    ) = None


@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(timezone.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}


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}


_SUPERVISOR_TRUNCATION_LIMIT = 5


def detect_legacy_supervisor_sessions(
    cfg: DispatchConfig, *, supervisor_tags: Iterable[str]
) -> list[dict[str, Any]]:
    """Query OpenCode for live sessions whose title contains any of
    ``supervisor_tags`` (matched as a literal ``[TAG]`` prefix).

    Used by each dispatcher's startup so a deterministic dispatcher
    refuses to coexist with the legacy LLM supervisor session that
    polls the same work groups. The check is best-effort:
    :func:`_opencode_worker.list_sessions` returns ``[]`` on transport
    failure rather than raising, so a transient network blip at
    startup is treated as "no coexistence detected". The dispatcher's
    own cycle-failure budget will surface a persistent server outage
    on the next cycle.
    """
    sessions = _opencode_worker.list_sessions(cfg.server_url)
    needles = [f"[{tag}]" for tag in supervisor_tags]
    return [
        sess
        for sess in sessions
        if any(needle in str(sess.get("title") or "") for needle in needles)
    ]


def assert_no_legacy_supervisor(
    cfg: DispatchConfig,
    *,
    driver_name: str,
    supervisor_tags: Iterable[str],
    override_env: str,
) -> None:
    """Refuse startup when a competing legacy LLM supervisor is live.

    Intended to be called from each dispatcher's CLI ``main`` after
    config load and before any cycle work. The override env var lets an
    operator explicitly run both layers in observe-only configurations
    (e.g. while migrating one repo at a time); the default posture is
    to fail loudly and let the operator decide.
    """
    if os.environ.get(override_env, "").lower() in ("1", "true", "yes"):
        logger.warning(
            "%s skipping legacy-supervisor coexistence check because "
            "%s is set",
            driver_name,
            override_env,
        )
        return
    matches = detect_legacy_supervisor_sessions(
        cfg, supervisor_tags=supervisor_tags
    )
    if not matches:
        return
    shown = matches[:_SUPERVISOR_TRUNCATION_LIMIT]
    titles = ", ".join(f"#{i+1} {m.get('title')!r}" for i, m in enumerate(shown))
    elision = (
        f" (showing first {_SUPERVISOR_TRUNCATION_LIMIT})"
        if len(matches) > _SUPERVISOR_TRUNCATION_LIMIT
        else ""
    )
    raise SystemExit(
        f"{driver_name}: refusing to start — {len(matches)} legacy "
        f"supervisor session(s) detected on OpenCode at {cfg.server_url}: "
        f"{titles}{elision}. Stop them (or run scripts/opencode-builder.sh "
        f"with OPENCODE_BUILDER_SERVER_ONLY=1) and retry, or set "
        f"{override_env}=1 to override (observe-only deployments only)."
    )


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

        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,
        )
        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:
            try:
                post_session_result = group.post_session_action(
                    cfg,
                    item,
                    session.parsed_json,
                    session.raw_response,
                    terminal_state,
                )
            except Exception as exc:  # noqa: BLE001 — claim release MUST run
                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:
        if claimed and group.claim_kind 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()
    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(timezone.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))