feat(auto-agents): Tier 2 deterministic review/implementer dispatchers

Replaces the long-running pr-review-supervisor / implementation-
supervisor LLM polling loops with host-level Python dispatchers that
own queueing, claim ownership, watchdogs, and SQLite telemetry. The
LLM workers retain sole responsibility for review judgment and code
generation; Python owns only orchestration. The hard merge invariant
is unaffected — these dispatchers do not touch master.

Driver surface
- _opencode_worker.run_session_blocking — outcome-agnostic OpenCode
  session lifecycle (completed / timeout / transport-error). Used
  directly by reviewer / implementer dispatchers whose workers do
  not emit the conflict-driver JSON exit schema.
  run_worker_blocking is now a thin wrapper that adds the
  conflict-specific JSON-outcome classification.
- _dispatch_runtime — shared Python runtime (work-group polling,
  claim helpers, dispatch loop, telemetry). Pre-checks issue labels
  before claiming and refuses when any auto/claimed-* is already
  present, distinguishing already-claimed vs labels-fetch-failed
  vs claim-failed terminal states. Frozen DispatchConfig.
- dispatch_review.py / dispatch_implementer.py — per-pipeline work
  groups, prompts, and CLIs. Each refuses startup when a competing
  AUTO-REV-SUP / AUTO-IMP-SUP legacy supervisor is live on the
  same OpenCode server (override:
  {REVIEW,IMPLEMENTER}_DISPATCHER_ALLOW_SUPERVISOR_COEXIST=1).
- _loader.py — shared sibling-module loader; replaces the three
  duplicated copies in the dispatcher entry points.

Operational hardening
- run_outer_loop tracks consecutive cycle exceptions against
  cycle_failure_budget (default 5, env-tunable per driver) and
  exits 2 on exhaustion for supervisor-driven restart.
- _sanitize_release_detail strips control bytes and neutralises
  triple-backtick fences before quoting worker raw_response in
  Forgejo claim-release comments.
- scripts/opencode-builder.sh: OPENCODE_BUILDER_SERVER_ONLY=1 keeps
  only the OpenCode HTTP API up so the Python dispatchers own
  queue orchestration without auto-agents running concurrently.

Telemetry
- _forgejo_cache.py schema v4: dispatch_review_cycles,
  dispatch_implementer_cycles. One row per cycle with cycle_id,
  driver, candidates_count, claims_acquired, swept_count,
  processed_count, terminal_state, worker_outcome, session_id,
  worker_wallclock_seconds, raw.
- .opencode/telemetry/server.py wires the new tables into
  /api/cycles?driver=dispatch_review|dispatch_implementer and
  surfaces a composite terminal_state/worker_outcome 24h breakdown
  so dashboards can distinguish session-level vs work-level
  outcomes.

Tests
- 31 new tests in tests/auto_agents/test_dispatch_runtime.py
  covering: candidate priority/dedup, claim/release labels,
  foreign-claim refusal, same-kind-claim refusal,
  labels-fetch-failed terminal state, sanitization, supervisor
  coexistence guard (pass/refuse/override/unreachable-server),
  session timeout / transport-error propagation,
  JSON-vs-no-JSON worker exits, cycle failure budget exit and
  reset, heartbeat cadence, end-to-end --once --dry-run /
  --status CLI smoke, and full prompt-snapshot tests for
  _review_prompt and _implementation_prompt (PR-fix + issue-impl).
- test_telemetry_schema.py asserts schema v4 and the presence of
  the two new dispatcher cycle tables.

338 auto_agents tests pass (was 322 before Tier 2). Conflict driver
regression suite unchanged. Dispatchers run cleanly under
--status / --once --dry-run with no Forgejo or OpenCode HTTP traffic.

Co-authored-by: Cursor <cursoragent@cursor.com>

.opencode/telemetry/server.py

@@ -52,7 +52,8 @@ Endpoints (all JSON unless noted)
 - ``GET /static/<file>``          app.js / style.css
 - ``GET /api/meta``               repo target, identity, paths in use
 - ``GET /api/health``             daemon PIDs, heartbeats, OpenCode health
-- ``GET /api/cycles?driver=...``  merge/conflict cycle rows
+- ``GET /api/cycles?driver=...``  merge / conflict / dispatch_review /
+  dispatch_implementer cycle rows
 - ``GET /api/prs?label=...``      open PRs by Forgejo label (live API)
 - ``GET /api/velocity?window=...`` window stats from cache
 - ``GET /api/sessions``           OpenCode sessions
@@ -455,11 +456,37 @@ def _api_cycles(driver: str, limit: int) -> dict[str, Any]:
                 " ORDER BY started_at DESC LIMIT ?",
                 (limit,),
             )
+        elif driver in ("dispatch_review", "dispatch_implementer"):
+            # Tier 2 dispatcher telemetry. ``terminal_state`` describes
+            # the *session lifecycle* (completed / timeout /
+            # transport-error / already-claimed / labels-fetch-failed
+            # / claim-failed / dry-run) while ``worker_outcome`` is the
+            # worker's own JSON-emitted outcome when present (for
+            # implementer workers). Both are surfaced so an operator
+            # can distinguish "the OpenCode session ended cleanly" from
+            # "the worker reported success / failure for the actual
+            # work" — these are independent dimensions per the
+            # session-status / JSON-outcome separation introduced in
+            # the dispatcher hardening pass.
+            table = (
+                "dispatch_review_cycles"
+                if driver == "dispatch_review"
+                else "dispatch_implementer_cycles"
+            )
+            cur = conn.execute(
+                f"SELECT cycle_id, started_at, ended_at, driver,"
+                f" candidates_count, claims_acquired, swept_count,"
+                f" processed_count, terminal_state, worker_outcome,"
+                f" session_id, worker_wallclock_seconds, raw"
+                f" FROM {table}"
+                f" ORDER BY started_at DESC LIMIT ?",
+                (limit,),
+            )
         else:
             return {
                 "rows": [],
-                "note": f"unknown driver: {driver!r} "
+                "note": f"unknown driver: {driver!r} (expected "
-                "(expected merge|conflict)",
+                "merge|conflict|dispatch_review|dispatch_implementer)",
             }
         rows = _rows_to_dicts(list(cur.fetchall()))
         # outcome breakdown over the LAST 24 H, regardless of limit, so
@@ -471,13 +498,31 @@ def _api_cycles(driver: str, limit: int) -> dict[str, Any]:
                 " WHERE started_at > datetime('now','-24 hours')"
                 " GROUP BY terminal_state ORDER BY n DESC"
             ).fetchall()
-        else:
+        elif driver == "conflict":
             counts = conn.execute(
                 "SELECT outcome AS k, COUNT(*) AS n FROM"
                 " conflict_drive_cycles"
                 " WHERE started_at > datetime('now','-24 hours')"
                 " GROUP BY outcome ORDER BY n DESC"
             ).fetchall()
+        else:
+            table = (
+                "dispatch_review_cycles"
+                if driver == "dispatch_review"
+                else "dispatch_implementer_cycles"
+            )
+            # Composite breakdown: dashboards must show both terminal
+            # state and worker outcome side by side, since a session
+            # that ``completed`` with ``worker_outcome="rebase-failed"``
+            # is operationally different from one that ``completed``
+            # with no JSON exit at all.
+            counts = conn.execute(
+                f"SELECT terminal_state || '/' ||"
+                f"   COALESCE(worker_outcome, '-') AS k,"
+                f" COUNT(*) AS n FROM {table}"
+                f" WHERE started_at > datetime('now','-24 hours')"
+                f" GROUP BY k ORDER BY n DESC"
+            ).fetchall()
         return {
             "rows": rows,
             "outcome_breakdown_24h": _rows_to_dicts(list(counts)),

AGENTS.md

@@ -565,11 +565,93 @@ primary release mechanism — under normal operation, claims are released
 within seconds of work completing, not at the end of the TTL window.
 
 This is the minimum viable extension of the deterministic merge driver
-to the reviewer / implementer pools. It does **not** replace the LLM
+to the reviewer / implementer pools.
-supervisor sessions that dispatch workers; that work is captured in
+
-[`docs/development/auto-agents-tier-2-3-plan.md`](docs/development/auto-agents-tier-2-3-plan.md)
+### Tier 2 deterministic dispatchers (2026-05-06)
-as deferred Tier 2 / Tier 3 effort, gated on operational telemetry from
+
-this Tier 1 rollout.
+The reviewer and implementer queues now have host-level Python
+dispatchers:
+
+- [`tools/dispatch_review.py`](tools/dispatch_review.py) polls the same
+  five review work groups the old `pr-review-supervisor` used, claims
+  PRs with `auto/claimed-reviewer`, invokes `pr-review-worker` through
+  OpenCode HTTP, releases the claim, and records
+  `dispatch_review_cycles` telemetry.
+- [`tools/dispatch_implementer.py`](tools/dispatch_implementer.py) polls
+  failing-CI PRs, unaddressed `REQUEST_CHANGES` PRs, then open issues,
+  claims PR work with `auto/claimed-implementer`, invokes
+  `implementation-worker`, releases the claim, and records
+  `dispatch_implementer_cycles` telemetry. Issue implementation is not
+  claimed because no PR exists yet.
+
+The LLM workers still perform the irreducible work: code review judgment
+and code generation. The dispatchers replace only the long-running LLM
+supervisor loops with deterministic queueing, liveness, claim, timeout,
+and telemetry behavior. During rollout, do not run the old supervisor
+sessions and these dispatchers against the same repo at the same time
+unless the run is explicitly observe-only; they poll the same candidate
+wrappers and would compete for the same work.
+
+**Claim ownership semantics.** Forgejo's label-add API is idempotent —
+it returns 200 whether the label was newly attached or already
+present — so the dispatchers cannot infer ownership from the add call
+alone. Before each claim, `_dispatch_runtime.claim_work_item` GETs the
+issue's labels and refuses (`terminal_state="already-claimed"`) when
+*any* `auto/claimed-*` label is already attached. This means a PR
+already held by the merge driver, the conflict driver, or a sibling
+dispatcher is left untouched: the worker is not invoked and the
+release path is not run, so we never strip a claim we did not just
+attach. The pre-check has the same TOCTOU window as `claim_pr.ts`
+(documented in `conflict-drive-plan.md` § 3.3.1); the worker layer's
+idempotent claim helper deduplicates work in the rare racing case.
+
+**Worker session outcomes.** The dispatchers call
+`_opencode_worker.run_session_blocking`, an outcome-agnostic entry point
+whose `SessionResult.status` is one of `completed` / `timeout` /
+`transport-error`. Reviewer / implementer workers do **not** need to
+emit the conflict-driver `{outcome: resolved | unresolvable |
+rebase-failed}` JSON to be classified as successful — a clean session
+exit is sufficient. The conflict driver still uses
+`run_worker_blocking`, which thin-wraps the session runner and adds the
+JSON outcome classification.
+
+**Cycle failure budget.** Each dispatcher's outer loop tracks
+consecutive cycle exceptions against `cycle_failure_budget` (default 5,
+overridable via `REVIEW_DISPATCHER_CYCLE_FAILURE_BUDGET` /
+`IMPLEMENTER_DISPATCHER_CYCLE_FAILURE_BUDGET`). Transient failures —
+`tsx`-time exceptions, malformed JSON from a work-group script, Forgejo
+4xx blips — are logged and the loop continues. Once the budget is
+exhausted the driver exits with code 2 so a supervising launcher
+(systemd, `opencode-builder.sh`, etc.) can restart it with fresh
+state, matching the behavior of `merge_drive.py` and
+`conflict_drive.py`.
+
+**Mechanical guard against legacy-supervisor coexistence.** Each
+dispatcher CLI checks the OpenCode `/session` listing at startup and
+refuses to start when a competing legacy LLM supervisor session is
+live (`[AUTO-REV-SUP]` for `dispatch_review.py`, `[AUTO-IMP-SUP]` for
+`dispatch_implementer.py`). The check runs after `--status` /
+`--dry-run` short-circuits, so operator inspection commands still
+work. Override only for observe-only deployments by setting
+`REVIEW_DISPATCHER_ALLOW_SUPERVISOR_COEXIST=1` or
+`IMPLEMENTER_DISPATCHER_ALLOW_SUPERVISOR_COEXIST=1`. A transient
+OpenCode unreachability at startup is treated as "no coexistence
+detected" — the cycle failure budget surfaces persistent server
+outages on the first cycle attempt.
+
+**Claim-time terminal states.** In addition to the session lifecycle
+states above, the dispatcher records:
+
+- `claim-failed` — Forgejo refused our label-add (e.g., the label is
+  not defined on the repo).
+- `already-claimed` — another worker already holds an
+  `auto/claimed-*` label on the item; we did not claim and did not
+  release.
+- `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.
 
 ### Tier 0A audit findings (2026-05-02)
 
@@ -657,6 +739,24 @@ you add a new env var to the driver, add it here too.
 | `MERGE_DRIVER_LOG_LEVEL` | no | `INFO` | Python logging level for `merge_drive.py`: `DEBUG`/`INFO`/`WARNING`/`ERROR`. Stderr is the destination unless a parent process (systemd, supervisord) has already configured the root logger. |
 | `VERIFY_INVARIANT_LOG_LEVEL` | no | `INFO` (or `DEBUG` if `--progress` is passed) | Python logging level for `verify_invariant.py`. The `--progress` CLI flag flips the default to `DEBUG`; setting this env var overrides both. |
 
+The Tier 2 dispatchers follow the same conventions:
+`REVIEW_DISPATCHER_*` and `IMPLEMENTER_DISPATCHER_*` cover
+`LOCK_PATH`, `HEARTBEAT_PATH`, `CYCLE_SECONDS`,
+`MAX_ITEMS_PER_CYCLE`, `WORKER_TIMEOUT_SECONDS`,
+`CLAIM_TTL_SECONDS`, `API_RETRIES`, `REQUEST_TIMEOUT_S`,
+`SCRIPT_TIMEOUT_SECONDS`, `CYCLE_FAILURE_BUDGET`, and `LOG_LEVEL`.
+`dispatch_review.py` reads `FORGEJO_REVIEWER_PAT` (falling back to
+`GITEA_TOKEN` for local tests); `dispatch_implementer.py` reads
+`FORGEJO_PAT` (falling back to `GITEA_TOKEN`). Both read
+`OPENCODE_SERVER_URL` (default
+`http://127.0.0.1:4096`) and write heartbeat files through the same
+fallback chain as `merge_drive.py`.
+
+For deterministic-dispatcher runs, start OpenCode in server-only mode:
+`OPENCODE_BUILDER_SERVER_ONLY=1 bash scripts/opencode-builder.sh`. That
+keeps the HTTP worker API available without launching the legacy
+`auto-agents` supervisor fleet.
+
 ### Bot-generated commit conventions
 
 The merge driver commits via the Forgejo API merge endpoint. Specific
@@ -699,6 +799,16 @@ discover the system from `AGENTS.md` alone:
   subagent only when the rebase actually conflicts, force-pushes with
   `--force-with-lease`, escalates to `auto/needs-implementer` after
   three definite failures in 24h. Single-instance lock + heartbeat.
+- [`tools/dispatch_review.py`](tools/dispatch_review.py) — Tier 2:
+  deterministic reviewer dispatcher. Polls existing review work-group
+  wrappers, claims `auto/claimed-reviewer`, invokes `pr-review-worker`,
+  releases claims, and records `dispatch_review_cycles`.
+- [`tools/dispatch_implementer.py`](tools/dispatch_implementer.py) —
+  Tier 2: deterministic implementer dispatcher. Polls existing
+  implementer / issue work-group wrappers, claims
+  `auto/claimed-implementer` for PR work, invokes
+  `implementation-worker`, releases claims, and records
+  `dispatch_implementer_cycles`.
 - [`tools/inject_synthetic_conflict.py`](tools/inject_synthetic_conflict.py)
   — local-only test helper: opens a PR on a fork with a guaranteed
   add/add or modify-vs-stub conflict (modes: `trivial`,
@@ -739,8 +849,9 @@ discover the system from `AGENTS.md` alone:
 ### Targeting a personal-fork test repo
 
 All Python pipeline tools (`merge_drive.py`, `conflict_drive.py`,
-`verify_invariant.py`, `forgejo_audit.py`, `audit_branch_protection.py`,
+`dispatch_review.py`, `dispatch_implementer.py`, `verify_invariant.py`,
-`flag_stale_prs.py`, `setup_auto_labels.py`, `setup_branch_protection.py`,
+`forgejo_audit.py`, `audit_branch_protection.py`, `flag_stale_prs.py`,
+`setup_auto_labels.py`, `setup_branch_protection.py`,
 `migrate_to_new_driver.py`, `inject_synthetic_conflict.py`)
 read `FORGEJO_OWNER` / `FORGEJO_REPO` /
 `FORGEJO_API_BASE` (and `FORGEJO_ORG` / `FORGEJO_DEFAULT_BRANCH` /

CHANGELOG.md

@@ -31,6 +31,87 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Added
 
+- **Tier 2 deterministic review / implementer dispatchers** (2026-05-06).
+  Added `tools/dispatch_review.py` and `tools/dispatch_implementer.py`
+  plus a shared `_dispatch_runtime.py` so the reviewer and implementer
+  queues can be polled by host-level Python drivers instead of long-lived
+  LLM supervisor sessions. The dispatchers reuse the existing `list_prs_*`
+  / `list_issues` work-group wrappers, claim PR work with
+  `auto/claimed-reviewer` or `auto/claimed-implementer` before invoking
+  the single-shot LLM workers through OpenCode HTTP, release claims in a
+  deterministic finally block, and write `dispatch_review_cycles` /
+  `dispatch_implementer_cycles` telemetry. `opencode-builder.sh` now has
+  `OPENCODE_BUILDER_SERVER_ONLY=1` so operators can run only the OpenCode
+  server while the Python dispatchers own queue orchestration. Same-day
+  hardening on the initial dispatcher landing:
+  - `_opencode_worker.py` gained an outcome-agnostic
+    `run_session_blocking` that returns a `SessionResult` describing the
+    OpenCode session lifecycle (`completed` / `timeout` / `transport-error`)
+    without forcing the conflict-driver `{outcome: resolved|...}` JSON
+    schema. `run_worker_blocking` is now a thin wrapper that adds the
+    conflict-specific JSON classification, so reviewer / implementer
+    workers no longer get misclassified as `worker-failed` when their
+    successful runs do not emit a JSON object.
+  - `_dispatch_runtime.claim_work_item` pre-checks the issue's labels
+    and refuses to claim when *any* `auto/claimed-*` label is already
+    present (foreign or sibling). This closes the latent ownership
+    bug where Forgejo's idempotent label-add could let one dispatcher
+    silently inherit another worker's live claim and then strip it in
+    its finally block. The refusing path now reports
+    `terminal_state="already-claimed"` and never invokes the worker
+    or the release helper.
+  - `_dispatch_runtime.run_outer_loop` now wraps each cycle in a
+    `cycle_failure_budget` (default 5, env-tunable per driver):
+    transient script / API failures are logged and retried, but a
+    persistently-broken cycle exits with code 2 so a supervising
+    launcher can restart the driver with fresh state. The status
+    payload exposes the new knob.
+  - Test coverage extended to cover foreign-claim refusal,
+    same-kind-claim refusal, session timeout / transport-error
+    propagation, JSON-vs-no-JSON worker exits, the cycle failure
+    budget, and the heartbeat cadence.
+  - Refuse claims when the labels GET fails on auth / permission
+    (4xx) instead of proceeding with a silent label attach; the
+    new `labels-fetch-failed` terminal state captures the HTTP
+    status in `claim_result.label_fetch_status` for operator
+    triage.
+  - `_sanitize_release_detail` strips control characters and
+    neutralises triple-backtick code fences so a worker's raw
+    response cannot break the surrounding fenced template in the
+    Forgejo claim-release comment.
+  - `_dispatch_runtime.assert_no_legacy_supervisor` queries
+    OpenCode `/session` at dispatcher startup and refuses to launch
+    when a competing legacy LLM supervisor (`[AUTO-REV-SUP]` /
+    `[AUTO-IMP-SUP]`) is live. Override with
+    `REVIEW_DISPATCHER_ALLOW_SUPERVISOR_COEXIST=1` /
+    `IMPLEMENTER_DISPATCHER_ALLOW_SUPERVISOR_COEXIST=1` for
+    observe-only deployments. A transient OpenCode outage at
+    startup is treated as "no coexistence detected" and the
+    cycle-failure budget surfaces persistent server problems on
+    the first cycle.
+  - `tools/_loader.py` consolidates the previously-duplicated
+    `_load_sibling` file-load helper used by `_dispatch_runtime.py`,
+    `dispatch_review.py`, and `dispatch_implementer.py`. The
+    drivers now bootstrap `tools/` onto `sys.path` once and import
+    the helper by name.
+  - `DispatchConfig` is now a frozen dataclass; tests pass
+    `cycle_failure_budget` through the fixture rather than
+    mutating the instance.
+  - `.opencode/telemetry/server.py` exposes
+    `dispatch_review_cycles` and `dispatch_implementer_cycles` via
+    the existing `/api/cycles?driver=...` endpoint and surfaces a
+    composite `terminal_state/worker_outcome` 24 h breakdown so a
+    dashboard can distinguish "session completed but worker
+    reported failure" from "session completed and worker reported
+    success" from "session completed and worker emitted no JSON".
+  - Snapshot tests for `_review_prompt` and
+    `_implementation_prompt` (PR-fix and issue-impl variants)
+    catch encoding regressions like the em-dash slip caught
+    manually during the initial Tier 2 landing. End-to-end
+    `--once --dry-run` and `--status` CLI tests cover argparse,
+    env-var defaulting, table routing, and the dry-run
+    short-circuit without making any HTTP calls.
+
 - **`.opencode/telemetry/`: pipeline telemetry console** (2026-05-06).
   A small, dependency-free HTTP server that consolidates every silo
   the auto-agents pipeline emits into a single browser tab. Phase B

scripts/opencode-builder.sh

@@ -18,6 +18,7 @@ HOST="${OPENCODE_HOST:-127.0.0.1}"
 BASE="http://${HOST}:${PORT}"
 AGENT="auto-agents"
 PROMPT="Complete the current project's milestones up to and including v3.7.0 to a production ready state"
+SERVER_ONLY="${OPENCODE_BUILDER_SERVER_ONLY:-0}"
 HEALTH_TIMEOUT=60  # seconds to wait for the server to become healthy
 MAX_IDLE_PER_MINUTE=10
 
@@ -192,6 +193,15 @@ else
   log "Server is ready."
 fi
 
+if [[ "$SERVER_ONLY" == "1" || "$SERVER_ONLY" == "true" || "$SERVER_ONLY" == "yes" ]]; then
+  log "Server-only mode enabled; not launching auto-agents or LLM supervisors."
+  log "Use tools/dispatch_review.py and tools/dispatch_implementer.py as the deterministic dispatchers."
+  while ! $STOP_LOOP; do
+    sleep 5
+  done
+  exit 0
+fi
+
 # ── Create a session ──────────────────────────────────────────────────────
 create_session
 

tests/auto_agents/test_telemetry_schema.py

@@ -7,7 +7,8 @@ Covers:
 - ``record_conflict_drive_cycle`` and its retry-budget /
   conflict_drive_cycles_in_window helpers (Tier 1.5 — schema v3 added by
   ``conflict-drive-plan.md`` § 9 / Phase A.3).
-- Schema migration from v1 → v3.
+- Dispatcher telemetry tables (Tier 2 — schema v4).
+- Schema migration from v1 → v4.
 """
 
 from __future__ import annotations
@@ -151,13 +152,26 @@ def test_record_ci_gate_event_and_llm_activity(cache):
     assert llm[0]["decision"] == "approved"
 
 
-def test_schema_version_is_three(cache):
+def test_schema_version_is_four(cache):
-    """The schema migration is bumped to v3 once the
+    """The schema migration is bumped to v4 once the Tier 2 dispatcher
-    ``conflict_drive_cycles`` table lands (plan § 9 / Phase A.3)."""
+    telemetry tables land."""
     row = cache._conn.execute(
         "SELECT MAX(version) AS v FROM schema_version"
     ).fetchone()
-    assert row["v"] == 3
+    assert row["v"] == 4
+
+
+def test_dispatcher_cycle_tables_exist(cache):
+    """Happy path: both deterministic dispatcher telemetry tables are
+    created by the cache schema."""
+    tables = {
+        row["name"]
+        for row in cache._conn.execute(
+            "SELECT name FROM sqlite_master WHERE type = 'table'"
+        ).fetchall()
+    }
+    assert "dispatch_review_cycles" in tables
+    assert "dispatch_implementer_cycles" in tables
 
 
 def test_failure_unknown_pr_numbers_shape(cache):

tools/_dispatch_runtime.py

@@ -0,0 +1,843 @@
+"""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
+
+# 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_chars: list[str] = []
+    for ch in excerpt:
+        if ch in ("\n", "\t"):
+            cleaned_chars.append(ch)
+            continue
+        if ord(ch) < 0x20 or ord(ch) == 0x7F:
+            continue
+        cleaned_chars.append(ch)
+    cleaned = "".join(cleaned_chars)
+    while "```" in cleaned:
+        cleaned = cleaned.replace("```", "ʼʼʼ")
+    return cleaned
+
+
+@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]
+
+
+@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}
+
+
+def detect_legacy_supervisor_sessions(
+    cfg: DispatchConfig, *, supervisor_tags: list[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: if the
+    OpenCode server is not reachable we return ``[]`` rather than
+    block startup, since blocking on a transient network blip would
+    just trade one failure mode for another. The dispatcher's own
+    cycle-failure budget will surface a persistent server outage on
+    the next cycle.
+    """
+    try:
+        sessions = _opencode_worker._request(
+            "GET", f"{cfg.server_url}/session"
+        )
+    except Exception as exc:  # noqa: BLE001 — best-effort startup check
+        logger.warning(
+            "supervisor-coexistence check failed (server_url=%s): %s",
+            cfg.server_url,
+            exc,
+        )
+        return []
+    if not isinstance(sessions, list):
+        return []
+    needles = [f"[{tag}]" for tag in supervisor_tags]
+    matches: list[dict[str, Any]] = []
+    for sess in sessions:
+        if not isinstance(sess, dict):
+            continue
+        title = str(sess.get("title") or "")
+        if any(needle in title for needle in needles):
+            matches.append(sess)
+    return matches
+
+
+def assert_no_legacy_supervisor(
+    cfg: DispatchConfig,
+    *,
+    driver_name: str,
+    supervisor_tags: list[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
+    titles = ", ".join(
+        f"#{i+1} {m.get('title')!r}" for i, m in enumerate(matches[:5])
+    )
+    raise SystemExit(
+        f"{driver_name}: refusing to start — {len(matches)} legacy "
+        f"supervisor session(s) detected on OpenCode at {cfg.server_url}: "
+        f"{titles}. Stop them (or run scripts/opencode-builder.sh with "
+        f"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.
+    - ``claim-failed`` — Forgejo refused our label-add (label undefined
+      in the repo). 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`` 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)
+        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,
+        )
+        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
+        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,
+        }
+    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()
+    swept = sweep_own_claims(cfg, sweep_claim_kind) if sweep_claim_kind else []
+    candidates, group_counts = collect_candidates(cfg, groups)
+    processed: list[dict[str, Any]] = []
+    claims_acquired = 0
+    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})
+    ended_at = _now()
+    record_cycle(
+        cfg,
+        cycle_id=cycle_id,
+        started_at=started_at,
+        ended_at=ended_at,
+        driver_name=driver_name,
+        candidates_count=len(candidates),
+        group_counts=group_counts,
+        claims_acquired=claims_acquired,
+        swept=swept,
+        processed=processed,
+    )
+    return {
+        "cycle_id": cycle_id,
+        "started_at": started_at,
+        "ended_at": ended_at,
+        "candidates_count": len(candidates),
+        "group_counts": group_counts,
+        "claims_acquired": claims_acquired,
+        "swept": swept,
+        "processed": processed,
+    }
+
+
+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:
+    if table_name not in {"dispatch_review_cycles", "dispatch_implementer_cycles"}:
+        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.execute(
+            f"""
+            CREATE TABLE IF NOT EXISTS {table_name} (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                cycle_id TEXT NOT NULL,
+                started_at TEXT NOT NULL,
+                ended_at TEXT NOT NULL,
+                driver TEXT NOT NULL,
+                candidates_count INTEGER NOT NULL,
+                claims_acquired INTEGER NOT NULL,
+                swept_count INTEGER NOT NULL,
+                processed_count INTEGER NOT NULL,
+                terminal_state TEXT,
+                worker_outcome TEXT,
+                session_id TEXT,
+                worker_wallclock_seconds REAL,
+                raw TEXT NOT NULL
+            )
+            """
+        )
+        conn.execute(
+            f"CREATE INDEX IF NOT EXISTS idx_{table_name}_started_at "
+            f"ON {table_name}(started_at)"
+        )
+        conn.execute(
+            f"CREATE INDEX IF NOT EXISTS idx_{table_name}_terminal "
+            f"ON {table_name}(terminal_state)"
+        )
+
+
+def record_cycle(
+    cfg: DispatchConfig,
+    *,
+    cycle_id: str,
+    started_at: str,
+    ended_at: str,
+    driver_name: str,
+    candidates_count: int,
+    group_counts: dict[str, int],
+    claims_acquired: int,
+    swept: list[int],
+    processed: list[dict[str, Any]],
+) -> None:
+    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 "")
+        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:
+        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,
+                started_at,
+                ended_at,
+                driver_name,
+                candidates_count,
+                claims_acquired,
+                len(swept),
+                len(processed),
+                terminal_state,
+                worker_outcome,
+                session_id,
+                worker_wallclock,
+                json.dumps(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))

tools/_forgejo_cache.py

@@ -96,7 +96,9 @@ else:
 # Schema versions:
 #   1 — original (commits, pulls, reachability, sync_meta)
 #   2 — Tier 1B telemetry: merge_cycle, ci_gate_events, llm_activity
-SCHEMA_VERSION = 3
+#   3 — Tier 1.5 telemetry: conflict_drive_cycles
+#   4 — Tier 2 telemetry: dispatch_review_cycles, dispatch_implementer_cycles
+SCHEMA_VERSION = 4
 
 # Safety ceiling on pagination during a single sync pass.
 SYNC_PAGE_HARD_LIMIT = 4000  # 4000 * 50 = 200,000 rows; far above anything real.
@@ -300,6 +302,53 @@ CREATE TABLE IF NOT EXISTS conflict_drive_cycles (
 );
 CREATE INDEX IF NOT EXISTS idx_cdc_pr_number_started ON conflict_drive_cycles(pr_number, started_at);
 CREATE INDEX IF NOT EXISTS idx_cdc_started_at        ON conflict_drive_cycles(started_at);
+
+-- ─── Tier 2 dispatcher telemetry ─────────────────────────────────────────
+--
+-- One row per dispatcher cycle. ``raw`` contains the per-group counts, swept
+-- claims, and per-item worker details; the promoted columns cover dashboard
+-- and operator queries without JSON parsing.
+CREATE TABLE IF NOT EXISTS dispatch_review_cycles (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    cycle_id TEXT NOT NULL,
+    started_at TEXT NOT NULL,
+    ended_at TEXT NOT NULL,
+    driver TEXT NOT NULL,
+    candidates_count INTEGER NOT NULL,
+    claims_acquired INTEGER NOT NULL,
+    swept_count INTEGER NOT NULL,
+    processed_count INTEGER NOT NULL,
+    terminal_state TEXT,
+    worker_outcome TEXT,
+    session_id TEXT,
+    worker_wallclock_seconds REAL,
+    raw TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_dispatch_review_cycles_started_at
+    ON dispatch_review_cycles(started_at);
+CREATE INDEX IF NOT EXISTS idx_dispatch_review_cycles_terminal
+    ON dispatch_review_cycles(terminal_state);
+
+CREATE TABLE IF NOT EXISTS dispatch_implementer_cycles (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    cycle_id TEXT NOT NULL,
+    started_at TEXT NOT NULL,
+    ended_at TEXT NOT NULL,
+    driver TEXT NOT NULL,
+    candidates_count INTEGER NOT NULL,
+    claims_acquired INTEGER NOT NULL,
+    swept_count INTEGER NOT NULL,
+    processed_count INTEGER NOT NULL,
+    terminal_state TEXT,
+    worker_outcome TEXT,
+    session_id TEXT,
+    worker_wallclock_seconds REAL,
+    raw TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_dispatch_implementer_cycles_started_at
+    ON dispatch_implementer_cycles(started_at);
+CREATE INDEX IF NOT EXISTS idx_dispatch_implementer_cycles_terminal
+    ON dispatch_implementer_cycles(terminal_state);
 """
 
 

tools/_loader.py

@@ -0,0 +1,46 @@
+"""Shared sibling-module loader for the auto-agents Python tooling.
+
+Every driver and runtime helper in :mod:`tools` needs to load its
+peer modules without forcing callers to install the package. The
+historical pattern was to redeclare ``_load_sibling`` in each entry
+point; that left three independent copies in
+``_dispatch_runtime.py``, ``dispatch_review.py``, and
+``dispatch_implementer.py`` (plus near-identical copies elsewhere).
+
+This module collapses those copies into one helper. Behavior is
+identical: load by file path relative to ``tools/``, cache the result
+in :data:`sys.modules` under the requested name, and return the
+cached instance on subsequent calls.
+"""
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+from types import ModuleType
+
+
+TOOLS_DIR = Path(__file__).resolve().parent
+
+
+def load_sibling(module_name: str, filename: str) -> ModuleType:
+    """Load ``filename`` from ``tools/`` as ``module_name``.
+
+    Returns the cached module from :data:`sys.modules` if already
+    loaded; otherwise compiles and registers a fresh instance. Used by
+    drivers that need to import other ``tools/`` files without making
+    ``tools`` an installable package.
+    """
+    cached = sys.modules.get(module_name)
+    if cached is not None:
+        return cached
+    path = TOOLS_DIR / filename
+    spec = importlib.util.spec_from_file_location(module_name, path)
+    if spec is None or spec.loader is None:
+        raise ImportError(
+            f"cannot build module spec for {module_name} at {path}"
+        )
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = mod
+    spec.loader.exec_module(mod)
+    return mod

tools/_opencode_worker.py

@@ -69,6 +69,34 @@ class WorkerResult:
     raw_response: str = ""
 
 
+SessionStatus = Literal["completed", "timeout", "transport-error"]
+
+
+@dataclass
+class SessionResult:
+    """Outcome-agnostic result from running a one-shot OpenCode session.
+
+    Workers that do not emit a fixed JSON schema (review, implementation)
+    use this directly; the conflict driver keeps using
+    :func:`run_worker_blocking`, which classifies the parsed JSON into a
+    domain-specific :class:`WorkerResult.outcome`.
+
+    ``status`` describes only the *session* lifecycle: did the session
+    finish normally (``completed``), hit the watchdog
+    (``timeout``), or fail at the OpenCode HTTP layer
+    (``transport-error``). ``parsed_json`` is the last parseable JSON
+    object found in the final assistant message, or ``None`` if the
+    worker did not emit one — which is the normal case for review /
+    implementation workers.
+    """
+
+    status: SessionStatus
+    wallclock_seconds: float = 0.0
+    session_id: str = ""
+    raw_response: str = ""
+    parsed_json: dict[str, Any] | None = None
+
+
 # ─── HTTP layer (kept intentionally minimal — no shared retries) ──────────
 #
 # We deliberately do NOT reuse ``_claim_runtime``'s state-change retry
@@ -271,7 +299,7 @@ def _is_idle(server_url: str, session_id: str) -> bool:
 # ─── Public entry point ────────────────────────────────────────────────────
 
 
-def run_worker_blocking(
+def run_session_blocking(
     *,
     server_url: str,
     agent: str,
@@ -279,9 +307,15 @@ def run_worker_blocking(
     prompt: str,
     timeout_seconds: int = 900,
     poll_interval_seconds: float = 2.0,
-) -> WorkerResult:
+) -> SessionResult:
     """Spin up a one-shot OpenCode agent session, wait for it to finish,
-    and return a typed :class:`WorkerResult`.
+    and return a typed :class:`SessionResult`.
+
+    This is the outcome-agnostic entry point used by drivers whose
+    workers do not emit a fixed ``{"outcome": ...}`` JSON schema (review,
+    implementation). For the conflict driver, see
+    :func:`run_worker_blocking`, which thin-wraps this function and adds
+    the conflict-specific JSON-outcome classification.
 
     Lifecycle (matches plan § 6.2):
 
@@ -292,28 +326,27 @@ def run_worker_blocking(
     2. ``POST /session/{id}/prompt_async`` with the agent name + prompt.
     3. Poll ``GET /session/status`` every ``poll_interval_seconds`` until
        the session is idle. The watchdog wallclock at ``timeout_seconds``
-       short-circuits with ``outcome=timeout``; the session is asked to
+       short-circuits with ``status="timeout"``; the session is asked to
        abort and DELETEd in the finally block.
     4. ``GET /session/{id}/message`` to read the final assistant turn.
     5. Tolerant JSON extraction (last parseable ``{...}`` substring of the
-       last assistant text) → ``outcome`` / ``files_touched``.
+       last assistant text) → ``parsed_json``. May be ``None`` for
+       workers that do not emit a JSON object — that is not a failure.
     6. ``DELETE /session/{id}`` in a finally block. On DELETE failure log
        a warning and continue — the session leaks until OpenCode server
        restart. Acceptable for this driver's expected dispatch rate
        (single-digit per day per plan § 6.2).
 
-    Any HTTP / network error short-circuits to ``transport-error``; the
+    Any HTTP / network error short-circuits to ``status="transport-error"``;
-    finally block still attempts session teardown.
+    the finally block still attempts session teardown.
     """
     started_at = time.monotonic()
     session_id = ""
-    raw_text = ""
 
     def _elapsed() -> float:
         return time.monotonic() - started_at
 
     try:
-        # 1. Create session.
         title = f"[{tag}] {agent}"
         try:
             session = _request(
@@ -321,8 +354,8 @@ def run_worker_blocking(
             )
         except _TRANSPORT_EXC as e:
             logger.warning("OpenCode session create failed: %s", e)
-            return WorkerResult(
+            return SessionResult(
-                outcome="transport-error",
+                status="transport-error",
                 wallclock_seconds=_elapsed(),
                 raw_response=str(e),
             )
@@ -330,8 +363,8 @@ def run_worker_blocking(
             logger.warning(
                 "OpenCode session create returned malformed payload: %r", session
             )
-            return WorkerResult(
+            return SessionResult(
-                outcome="transport-error",
+                status="transport-error",
                 wallclock_seconds=_elapsed(),
                 raw_response=json.dumps(session) if session is not None else "",
             )
@@ -343,7 +376,6 @@ def run_worker_blocking(
             agent,
         )
 
-        # 2. Dispatch the prompt.
         try:
             _request(
                 "POST",
@@ -354,14 +386,13 @@ def run_worker_blocking(
             logger.warning(
                 "OpenCode prompt_async failed for session %s: %s", session_id, e
             )
-            return WorkerResult(
+            return SessionResult(
-                outcome="transport-error",
+                status="transport-error",
                 wallclock_seconds=_elapsed(),
                 session_id=session_id,
                 raw_response=str(e),
             )
 
-        # 3. Poll until idle or timeout.
         deadline = started_at + timeout_seconds
         while True:
             if time.monotonic() >= deadline:
@@ -370,8 +401,6 @@ def run_worker_blocking(
                     timeout_seconds,
                     session_id,
                 )
-                # Best-effort abort. Ignore failures; teardown follows in
-                # finally either way.
                 try:
                     _request(
                         "POST",
@@ -383,8 +412,8 @@ def run_worker_blocking(
                         session_id,
                         e,
                     )
-                return WorkerResult(
+                return SessionResult(
-                    outcome="timeout",
+                    status="timeout",
                     wallclock_seconds=_elapsed(),
                     session_id=session_id,
                 )
@@ -397,15 +426,14 @@ def run_worker_blocking(
                     session_id,
                     e,
                 )
-                return WorkerResult(
+                return SessionResult(
-                    outcome="transport-error",
+                    status="transport-error",
                     wallclock_seconds=_elapsed(),
                     session_id=session_id,
                     raw_response=str(e),
                 )
             time.sleep(poll_interval_seconds)
 
-        # 4. Fetch messages and extract JSON.
         try:
             messages = _request(
                 "GET", f"{server_url}/session/{session_id}/message"
@@ -416,8 +444,8 @@ def run_worker_blocking(
                 session_id,
                 e,
             )
-            return WorkerResult(
+            return SessionResult(
-                outcome="transport-error",
+                status="transport-error",
                 wallclock_seconds=_elapsed(),
                 session_id=session_id,
                 raw_response=str(e),
@@ -425,31 +453,16 @@ def run_worker_blocking(
         if not isinstance(messages, list):
             messages = []
         raw_text = _last_assistant_text(messages)
-        parsed = _extract_last_json_object(raw_text) or {}
+        parsed = _extract_last_json_object(raw_text)
-        outcome_str = parsed.get("outcome")
+        return SessionResult(
-        if outcome_str not in ("resolved", "unresolvable", "rebase-failed"):
+            status="completed",
-            logger.warning(
-                "OpenCode worker returned unexpected outcome %r — classifying "
-                "as rebase-failed (session=%s)",
-                outcome_str,
-                session_id,
-            )
-            outcome_str = "rebase-failed"
-        files_touched = parsed.get("files_touched") or []
-        if not isinstance(files_touched, list):
-            files_touched = []
-        return WorkerResult(
-            outcome=outcome_str,  # type: ignore[arg-type]
-            files_touched=[str(f) for f in files_touched],
             wallclock_seconds=_elapsed(),
             session_id=session_id,
             raw_response=raw_text,
+            parsed_json=parsed,
         )
 
     finally:
-        # 6. Best-effort teardown. A failed DELETE is logged + ignored;
-        # the session leaks until OpenCode server restart. Acceptable for
-        # the driver's expected single-digit dispatch rate.
         if session_id:
             try:
                 _request("DELETE", f"{server_url}/session/{session_id}")
@@ -460,3 +473,74 @@ def run_worker_blocking(
                     session_id,
                     e,
                 )
+
+
+def run_worker_blocking(
+    *,
+    server_url: str,
+    agent: str,
+    tag: str,
+    prompt: str,
+    timeout_seconds: int = 900,
+    poll_interval_seconds: float = 2.0,
+) -> WorkerResult:
+    """Run a one-shot OpenCode agent and classify the result against the
+    conflict-driver outcome schema.
+
+    Thin wrapper around :func:`run_session_blocking` that interprets the
+    last parseable JSON object (per plan § 6.2 step 5) into one of:
+
+    - ``resolved`` / ``unresolvable`` / ``rebase-failed`` from the
+      worker's structured exit, or
+    - ``rebase-failed`` if the session completed but the JSON contract
+      was violated (no JSON, missing ``outcome`` key, or unrecognized
+      value), or
+    - ``timeout`` / ``transport-error`` propagated from the session
+      lifecycle.
+
+    Drivers whose workers do not emit this schema (review, implementer)
+    should call :func:`run_session_blocking` directly and treat
+    ``status == "completed"`` as success.
+    """
+    session = run_session_blocking(
+        server_url=server_url,
+        agent=agent,
+        tag=tag,
+        prompt=prompt,
+        timeout_seconds=timeout_seconds,
+        poll_interval_seconds=poll_interval_seconds,
+    )
+    if session.status == "timeout":
+        return WorkerResult(
+            outcome="timeout",
+            wallclock_seconds=session.wallclock_seconds,
+            session_id=session.session_id,
+            raw_response=session.raw_response,
+        )
+    if session.status == "transport-error":
+        return WorkerResult(
+            outcome="transport-error",
+            wallclock_seconds=session.wallclock_seconds,
+            session_id=session.session_id,
+            raw_response=session.raw_response,
+        )
+    parsed = session.parsed_json or {}
+    outcome_str = parsed.get("outcome")
+    if outcome_str not in ("resolved", "unresolvable", "rebase-failed"):
+        logger.warning(
+            "OpenCode worker returned unexpected outcome %r — classifying "
+            "as rebase-failed (session=%s)",
+            outcome_str,
+            session.session_id,
+        )
+        outcome_str = "rebase-failed"
+    files_touched = parsed.get("files_touched") or []
+    if not isinstance(files_touched, list):
+        files_touched = []
+    return WorkerResult(
+        outcome=outcome_str,  # type: ignore[arg-type]
+        files_touched=[str(f) for f in files_touched],
+        wallclock_seconds=session.wallclock_seconds,
+        session_id=session.session_id,
+        raw_response=session.raw_response,
+    )

tools/dispatch_implementer.py

@@ -0,0 +1,194 @@
+#!/usr/bin/env python3
+"""Deterministic dispatcher for implementation workers.
+
+The dispatcher preserves the existing priority order from
+``implementation-supervisor``: 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.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+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 load_sibling as _load_sibling  # noqa: E402  type: ignore[import-not-found]
+
+
+_dispatch = _load_sibling("_dispatch_runtime", "_dispatch_runtime.py")
+
+
+DRIVER_NAME = "dispatch_implementer.py"
+CLAIM_KIND = "implementer"
+
+
+def _implementation_prompt(cfg: Any, item: dict[str, Any], group: Any) -> str:
+    work_type = "issue_impl" if group.item_kind == "issue" else "pr_fix"
+    title = str(item.get("title") or "")
+    number = int(item["number"])
+    claim_note = (
+        "The deterministic dispatcher already claimed "
+        "`auto/claimed-implementer` before starting this worker. If your "
+        "startup claim step sees the label already present, treat that as "
+        "success and continue normally. Still run your release step before "
+        "exiting; the dispatcher will also release in its finally block."
+        if work_type == "pr_fix"
+        else "No PR exists yet for this issue work, so there is no "
+        "`auto/claimed-implementer` claim to acquire before dispatch."
+    )
+    return f"""Implement or fix the indicated issue or pull request.
+
+forgejo_url: `{cfg.forgejo_url}`
+forgejo_owner: `{cfg.owner}`
+forgejo_repo: `{cfg.repo}`
+
+work_type: {json.dumps(work_type)}
+work_number: {number}
+work_title: {json.dumps(title)}
+
+PR Compliance Checklist (MANDATORY - complete ALL items before creating a PR):
+[ ] 1. CHANGELOG.md — add entry under [Unreleased] section
+[ ] 2. CONTRIBUTORS.md — add or update contribution entry
+[ ] 3. Commit footer — include `ISSUES CLOSED: #<issue-number>` in the commit message
+[ ] 4. CI passes — all quality gates and tests green before requesting review
+[ ] 5. BDD/Behave tests — added or updated for the changed behaviour
+[ ] 6. Epic reference — PR description references the parent Epic issue number
+[ ] 7. Labels — applied via forgejo-label-manager: State/In Review, Priority/<level>, MoSCoW/<level>, Type/<type>
+[ ] 8. Milestone — PR assigned to the earliest open milestone matching the issue
+
+{claim_note}
+
+When the implementation work is complete, include exactly one JSON object in
+your final response:
+{{"outcome": "resolved", "files_touched": ["path/changed"]}}
+
+If you cannot complete the implementation because of an unrecoverable setup,
+API, or repository problem, include:
+{{"outcome": "rebase-failed", "files_touched": []}}
+"""
+
+
+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.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.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,
+    ),
+]
+
+
+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")
+        ),
+    )
+
+
+SUPERVISOR_TAGS = ["AUTO-IMP-SUP"]
+SUPERVISOR_OVERRIDE_ENV = "IMPLEMENTER_DISPATCHER_ALLOW_SUPERVISOR_COEXIST"
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--once", action="store_true", help="run one cycle and exit")
+    parser.add_argument("--status", action="store_true", help="print config and exit")
+    parser.add_argument("--dry-run", action="store_true", help="claim nothing and do not dispatch")
+    args = parser.parse_args()
+    _dispatch._configure_logging("IMPLEMENTER_DISPATCHER_LOG_LEVEL")
+    cfg = load_config(dry_run=args.dry_run)
+    if args.status:
+        _dispatch.json_line(_dispatch.status_payload(cfg, driver_name=DRIVER_NAME))
+        return 0
+    if not args.dry_run:
+        _dispatch.assert_no_legacy_supervisor(
+            cfg,
+            driver_name=DRIVER_NAME,
+            supervisor_tags=SUPERVISOR_TAGS,
+            override_env=SUPERVISOR_OVERRIDE_ENV,
+        )
+    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())

tools/dispatch_review.py

@@ -0,0 +1,195 @@
+#!/usr/bin/env python3
+"""Deterministic dispatcher for PR review workers.
+
+Replaces the long-running ``pr-review-supervisor`` polling session with a
+host-level Python loop.  It keeps review judgment in ``pr-review-worker``;
+this file owns queue polling, pre-dispatch claims, worker watchdogs, and
+cycle telemetry.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+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 load_sibling as _load_sibling  # noqa: E402  type: ignore[import-not-found]
+
+
+_dispatch = _load_sibling("_dispatch_runtime", "_dispatch_runtime.py")
+
+
+DRIVER_NAME = "dispatch_review.py"
+CLAIM_KIND = "reviewer"
+
+
+def _review_prompt(cfg: Any, item: dict[str, Any], group: Any) -> str:
+    review_type = {
+        "addressed_changes_ci_passing": "re_review",
+        "addressed_changes_ci_failing": "re_review",
+        "no_active_review_ci_passing": "first_review",
+        "no_active_review_ci_failing": "first_review",
+        "missing_ci_checks": "ci_flag",
+    }[group.name]
+    head = item.get("head") if isinstance(item.get("head"), dict) else {}
+    branch_name = str(head.get("ref") or item.get("branch_name") or "")
+    head_sha = str(head.get("sha") or item.get("head_sha") or "")
+    priority_label = item.get("priority_label")
+    return f"""Review the indicated Pull Request.
+
+forgejo_url: `{cfg.forgejo_url}`
+forgejo_owner: `{cfg.owner}`
+forgejo_repo: `{cfg.repo}`
+
+pr_number: {int(item["number"])}
+pr_title: {json.dumps(str(item.get("title") or ""))}
+branch_name: {json.dumps(branch_name)}
+head_sha: {json.dumps(head_sha)}
+ci_status: {json.dumps(str(item.get("ci_status") or ""))}
+priority_rank: {int(item.get("priority_rank", 6))}
+priority_label: {json.dumps(priority_label)}
+review_type: {json.dumps(review_type)}
+
+The deterministic dispatcher already claimed `auto/claimed-reviewer` before
+starting this worker. If your startup claim step sees the label already
+present, treat that as success and continue normally. Still run your release
+step before exiting; the dispatcher will also release in its finally block.
+
+When the review is fully submitted, include exactly one JSON object in your
+final response:
+{{"outcome": "resolved", "files_touched": []}}
+
+If you cannot complete the review because of an unrecoverable setup or API
+problem, include:
+{{"outcome": "rebase-failed", "files_touched": []}}
+"""
+
+
+WORK_GROUPS = [
+    _dispatch.WorkGroup(
+        name="addressed_changes_ci_passing",
+        script_name="list_prs_addressed_changes_ci_passing",
+        item_kind="pr",
+        claim_kind=CLAIM_KIND,
+        worker_agent="pr-review-worker",
+        tag_prefix="AUTO-REV",
+        prompt_factory=_review_prompt,
+    ),
+    _dispatch.WorkGroup(
+        name="no_active_review_ci_passing",
+        script_name="list_prs_no_active_review_ci_passing",
+        item_kind="pr",
+        claim_kind=CLAIM_KIND,
+        worker_agent="pr-review-worker",
+        tag_prefix="AUTO-REV",
+        prompt_factory=_review_prompt,
+    ),
+    _dispatch.WorkGroup(
+        name="addressed_changes_ci_failing",
+        script_name="list_prs_addressed_changes_ci_failing",
+        item_kind="pr",
+        claim_kind=CLAIM_KIND,
+        worker_agent="pr-review-worker",
+        tag_prefix="AUTO-REV",
+        prompt_factory=_review_prompt,
+    ),
+    _dispatch.WorkGroup(
+        name="no_active_review_ci_failing",
+        script_name="list_prs_no_active_review_ci_failing",
+        item_kind="pr",
+        claim_kind=CLAIM_KIND,
+        worker_agent="pr-review-worker",
+        tag_prefix="AUTO-REV",
+        prompt_factory=_review_prompt,
+    ),
+    _dispatch.WorkGroup(
+        name="missing_ci_checks",
+        script_name="list_prs_missing_ci_checks",
+        item_kind="pr",
+        claim_kind=CLAIM_KIND,
+        worker_agent="pr-review-worker",
+        tag_prefix="AUTO-REV",
+        prompt_factory=_review_prompt,
+    ),
+]
+
+
+def load_config(*, dry_run: bool = False) -> Any:
+    token = _dispatch.load_secret("FORGEJO_REVIEWER_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(
+            "REVIEW_DISPATCHER_LOCK_PATH", "review-dispatcher.lock"
+        ),
+        heartbeat_path=_dispatch.resolve_lock_or_heartbeat(
+            "REVIEW_DISPATCHER_HEARTBEAT_PATH", "review-dispatcher.heartbeat"
+        ),
+        cycle_interval_seconds=int(os.environ.get("REVIEW_DISPATCHER_CYCLE_SECONDS", "300")),
+        max_items_per_cycle=int(os.environ.get("REVIEW_DISPATCHER_MAX_ITEMS_PER_CYCLE", "1")),
+        worker_timeout_seconds=int(os.environ.get("REVIEW_DISPATCHER_WORKER_TIMEOUT_SECONDS", "1800")),
+        claim_ttl_seconds=int(os.environ.get("REVIEW_DISPATCHER_CLAIM_TTL_SECONDS", "1800")),
+        api_retries=int(os.environ.get("REVIEW_DISPATCHER_API_RETRIES", "3")),
+        request_timeout_s=int(os.environ.get("REVIEW_DISPATCHER_REQUEST_TIMEOUT_S", "30")),
+        script_timeout_seconds=int(os.environ.get("REVIEW_DISPATCHER_SCRIPT_TIMEOUT_SECONDS", "120")),
+        table_name="dispatch_review_cycles",
+        dry_run=dry_run,
+        cycle_failure_budget=int(
+            os.environ.get("REVIEW_DISPATCHER_CYCLE_FAILURE_BUDGET", "5")
+        ),
+    )
+
+
+SUPERVISOR_TAGS = ["AUTO-REV-SUP"]
+SUPERVISOR_OVERRIDE_ENV = "REVIEW_DISPATCHER_ALLOW_SUPERVISOR_COEXIST"
+
+
+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("REVIEW_DISPATCHER_LOG_LEVEL")
+    cfg = load_config(dry_run=args.dry_run)
+    if args.status:
+        _dispatch.json_line(_dispatch.status_payload(cfg, driver_name=DRIVER_NAME))
+        return 0
+    if not args.dry_run:
+        _dispatch.assert_no_legacy_supervisor(
+            cfg,
+            driver_name=DRIVER_NAME,
+            supervisor_tags=SUPERVISOR_TAGS,
+            override_env=SUPERVISOR_OVERRIDE_ENV,
+        )
+    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())