cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

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

Browse Source 
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>
This commit is contained in:
Drew Morris
2026-05-06 23:12:09 -04:00
parent e247c10219
commit 593d142f6f
12 changed files with 2751 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.opencode/telemetry/server.py
+49 -4
View File
@@ -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} "
"(expected merge|conflict)",
"note": f"unknown driver: {driver!r} (expected "
"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
+118 -7
View File
@@ -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
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)
as deferred Tier 2 / Tier 3 effort, gated on operational telemetry from
this Tier 1 rollout.
to the reviewer / implementer pools.
### Tier 2 deterministic dispatchers (2026-05-06)
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`,
`flag_stale_prs.py`, `setup_auto_labels.py`, `setup_branch_protection.py`,
`dispatch_review.py`, `dispatch_implementer.py`, `verify_invariant.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
+81
View File
@@ -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
+10
View File
@@ -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_dispatch_runtime.py
+1017
View File
File diff suppressed because it is too large Load Diff
tests/auto_agents/test_telemetry_schema.py
+19 -5
View File
@@ -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):
"""The schema migration is bumped to v3 once the
``conflict_drive_cycles`` table lands (plan § 9 / Phase A.3)."""
def test_schema_version_is_four(cache):
"""The schema migration is bumped to v4 once the Tier 2 dispatcher
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
+843
View File
@@ -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
+50 -1
View File
@@ -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
+46
View File
@@ -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
+129 -45
View File
@@ -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
finally block still attempts session teardown.
Any HTTP / network error short-circuits to ``status="transport-error"``;
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(
outcome="transport-error",
return SessionResult(
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(
outcome="transport-error",
return SessionResult(
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(
outcome="transport-error",
return SessionResult(
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(
outcome="timeout",
return SessionResult(
status="timeout",
wallclock_seconds=_elapsed(),
session_id=session_id,
)
@@ -397,15 +426,14 @@ def run_worker_blocking(
session_id,
e,
)
return WorkerResult(
outcome="transport-error",
return SessionResult(
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(
outcome="transport-error",
return SessionResult(
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 {}
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_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],
parsed = _extract_last_json_object(raw_text)
return SessionResult(
status="completed",
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
+194
View File
@@ -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
+195
View File
@@ -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())