feat: controller-owned allocator API on control-plane DB (Closes #600)
Add gitea_allocate_next_work and allocator_service routing policy on top of the #613 ControlPlaneDB substrate. Workers get atomic assign+lease results (or wait/no_safe_work/terminal-path outcomes) without self-selecting work via file locks or comment-only leases. #612 remains downstream. Closes #600
This commit is contained in:
@@ -21,6 +21,7 @@ import json
|
||||
import functools
|
||||
import contextlib
|
||||
import subprocess
|
||||
import uuid
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
|
||||
@@ -832,6 +833,8 @@ import review_workflow_boundary # noqa: E402
|
||||
import review_workflow_load # noqa: E402
|
||||
import mcp_session_state # noqa: E402
|
||||
import stale_review_decision_lock # noqa: E402
|
||||
import allocator_service # noqa: E402
|
||||
import control_plane_db # noqa: E402
|
||||
import agent_temp_artifacts
|
||||
import issue_lock_worktree # noqa: E402
|
||||
import issue_lock_provenance # noqa: E402
|
||||
@@ -11109,6 +11112,310 @@ def gitea_capability_stop_terminal_report() -> dict:
|
||||
})
|
||||
|
||||
|
||||
def _control_plane_db_or_error() -> tuple[Any | None, list[str]]:
|
||||
"""Open the #613 control-plane DB substrate; fail closed on errors."""
|
||||
try:
|
||||
db = control_plane_db.ControlPlaneDB()
|
||||
return db, []
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return None, [
|
||||
f"control-plane DB substrate unavailable: {_redact(str(exc))} "
|
||||
"(fail closed, #613/#600)"
|
||||
]
|
||||
|
||||
|
||||
def _allocator_candidates_from_gitea(
|
||||
*,
|
||||
remote: str,
|
||||
host: str | None,
|
||||
org: str,
|
||||
repo: str,
|
||||
include_issues: bool = True,
|
||||
include_prs: bool = True,
|
||||
limit: int = 50,
|
||||
) -> tuple[list[Any], list[str]]:
|
||||
"""Build allocator candidates from live Gitea open issues/PRs."""
|
||||
reasons: list[str] = []
|
||||
candidates: list[Any] = []
|
||||
try:
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return [], [f"failed to resolve Gitea target: {_redact(str(exc))}"]
|
||||
|
||||
if include_prs:
|
||||
try:
|
||||
prs = api_get_all(
|
||||
f"{repo_api_url(h, o, r)}/pulls?state=open", auth
|
||||
) or []
|
||||
except Exception as exc: # noqa: BLE001
|
||||
reasons.append(f"failed to list open PRs: {_redact(str(exc))}")
|
||||
prs = []
|
||||
for pr in prs[: max(1, int(limit))]:
|
||||
if not isinstance(pr, dict):
|
||||
continue
|
||||
number = pr.get("number")
|
||||
if number is None:
|
||||
continue
|
||||
head = pr.get("head") if isinstance(pr.get("head"), dict) else {}
|
||||
head_sha = head.get("sha") or pr.get("head_sha")
|
||||
labels = []
|
||||
for lab in pr.get("labels") or []:
|
||||
if isinstance(lab, dict) and lab.get("name"):
|
||||
labels.append(str(lab["name"]))
|
||||
elif isinstance(lab, str):
|
||||
labels.append(lab)
|
||||
# Lightweight review signals (best-effort; fail soft into reviewer path).
|
||||
rc_current = False
|
||||
approval_current = False
|
||||
approval_stale = False
|
||||
try:
|
||||
feedback = gitea_get_pr_review_feedback(
|
||||
int(number), remote=remote, org=o, repo=r
|
||||
)
|
||||
if feedback.get("success"):
|
||||
rc_current = bool(
|
||||
feedback.get("has_blocking_change_requests")
|
||||
and not feedback.get("review_feedback_stale")
|
||||
)
|
||||
# Stale RC means author pushed; reviewer still next.
|
||||
if feedback.get("has_blocking_change_requests") and feedback.get(
|
||||
"review_feedback_stale"
|
||||
):
|
||||
approval_stale = False
|
||||
except Exception:
|
||||
pass
|
||||
mergeable = bool(pr.get("mergeable"))
|
||||
try:
|
||||
candidates.append(
|
||||
allocator_service.WorkCandidate(
|
||||
kind="pr",
|
||||
number=int(number),
|
||||
state="open",
|
||||
labels=tuple(labels),
|
||||
title=str(pr.get("title") or ""),
|
||||
priority=10 if rc_current else 5,
|
||||
head_sha=head_sha,
|
||||
request_changes_current_head=rc_current,
|
||||
approval_on_current_head=approval_current,
|
||||
approval_stale=approval_stale,
|
||||
mergeable=mergeable,
|
||||
)
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
reasons.append(
|
||||
f"skipped invalid PR candidate #{number}: {_redact(str(exc))}"
|
||||
)
|
||||
|
||||
if include_issues:
|
||||
try:
|
||||
issues = api_get_all(
|
||||
f"{repo_api_url(h, o, r)}/issues?state=open&type=issues", auth
|
||||
) or []
|
||||
except Exception as exc: # noqa: BLE001
|
||||
# Fallback without type filter
|
||||
try:
|
||||
issues = api_get_all(
|
||||
f"{repo_api_url(h, o, r)}/issues?state=open", auth
|
||||
) or []
|
||||
except Exception as exc2: # noqa: BLE001
|
||||
reasons.append(
|
||||
f"failed to list open issues: {_redact(str(exc2))}"
|
||||
)
|
||||
issues = []
|
||||
for issue in issues[: max(1, int(limit))]:
|
||||
if not isinstance(issue, dict):
|
||||
continue
|
||||
# Pull requests also appear in /issues on Gitea — skip them.
|
||||
if issue.get("pull_request") is not None:
|
||||
continue
|
||||
number = issue.get("number")
|
||||
if number is None:
|
||||
continue
|
||||
labels = []
|
||||
for lab in issue.get("labels") or []:
|
||||
if isinstance(lab, dict) and lab.get("name"):
|
||||
labels.append(str(lab["name"]).lower())
|
||||
elif isinstance(lab, str):
|
||||
labels.append(lab.lower())
|
||||
body = str(issue.get("body") or "")
|
||||
title = str(issue.get("title") or "")
|
||||
blocked = "status:blocked" in labels
|
||||
# Explicit downstream dependency: #612 waits on #600 allocator.
|
||||
dep_unmet = False
|
||||
dep_reason = None
|
||||
if int(number) == 612:
|
||||
dep_unmet = True
|
||||
dep_reason = (
|
||||
"issue#612 (incident bridge) remains downstream of #600 "
|
||||
"allocator and must not be allocated until #600 is complete"
|
||||
)
|
||||
# Generic body markers for blocked-on unfinished deps.
|
||||
lower_body = body.lower()
|
||||
if "blocked on #600" in lower_body or "downstream of #600" in lower_body:
|
||||
dep_unmet = True
|
||||
dep_reason = dep_reason or (
|
||||
f"issue#{number} body marks dependency on #600"
|
||||
)
|
||||
try:
|
||||
candidates.append(
|
||||
allocator_service.WorkCandidate(
|
||||
kind="issue",
|
||||
number=int(number),
|
||||
state="open",
|
||||
labels=tuple(labels),
|
||||
title=title,
|
||||
priority=20 if "status:ready" in labels else 1,
|
||||
blocked=blocked,
|
||||
dependency_unmet=dep_unmet,
|
||||
dependency_reason=dep_reason,
|
||||
)
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
reasons.append(
|
||||
f"skipped invalid issue candidate #{number}: {_redact(str(exc))}"
|
||||
)
|
||||
|
||||
return candidates, reasons
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_allocate_next_work(
|
||||
apply: bool = False,
|
||||
role: str | None = None,
|
||||
session_id: str | None = None,
|
||||
remote: str = "dadeschools",
|
||||
host: str | None = None,
|
||||
org: str | None = None,
|
||||
repo: str | None = None,
|
||||
include_issues: bool = True,
|
||||
include_prs: bool = True,
|
||||
candidates_json: str | None = None,
|
||||
limit: int = 50,
|
||||
) -> dict:
|
||||
"""Controller-owned next-work allocator using the #613 control-plane DB (#600).
|
||||
|
||||
Workers must not self-select exclusive work under the standard multi-LLM
|
||||
workflow. Call this tool instead.
|
||||
|
||||
*apply=false* (default): dry-run selection only — no assignment/lease.
|
||||
*apply=true*: atomically assign + lease the selected Gitea issue/PR via
|
||||
``ControlPlaneDB.assign_and_lease`` (never file locks or comment-only
|
||||
leases as the coordination source).
|
||||
|
||||
Outcomes include: ``assigned_work``, ``preview``, ``wait``,
|
||||
``blocked_by_terminal_path``, ``no_safe_work``, ``role_ineligible``.
|
||||
|
||||
*candidates_json* may inject a JSON list of candidate dicts (tests /
|
||||
controller overrides). When omitted, open issues/PRs are loaded from Gitea.
|
||||
|
||||
#612 remains downstream: raw monitoring incidents are never candidates.
|
||||
"""
|
||||
read_block = _profile_operation_gate("gitea.read")
|
||||
if read_block:
|
||||
return {
|
||||
"success": False,
|
||||
"outcome": allocator_service.OUTCOME_NO_SAFE,
|
||||
"apply": bool(apply),
|
||||
"reasons": read_block,
|
||||
"permission_report": _permission_block_report("gitea.read"),
|
||||
"assignment": None,
|
||||
"substrate": "control_plane_db",
|
||||
"file_lock_only": False,
|
||||
"comment_lease_only": False,
|
||||
}
|
||||
|
||||
try:
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
except ValueError as exc:
|
||||
return {
|
||||
"success": False,
|
||||
"outcome": allocator_service.OUTCOME_NO_SAFE,
|
||||
"reasons": [str(exc)],
|
||||
"assignment": None,
|
||||
"substrate": "control_plane_db",
|
||||
}
|
||||
|
||||
profile = get_profile()
|
||||
profile_name = (profile.get("profile_name") or "").strip() or None
|
||||
active_role = _profile_role_kind(profile)
|
||||
role_in = (role or active_role or "").strip() or "author"
|
||||
|
||||
try:
|
||||
username = _authenticated_username(h)
|
||||
except Exception:
|
||||
username = None
|
||||
|
||||
db, db_errs = _control_plane_db_or_error()
|
||||
if db is None:
|
||||
return {
|
||||
"success": False,
|
||||
"outcome": allocator_service.OUTCOME_NO_SAFE,
|
||||
"apply": bool(apply),
|
||||
"reasons": db_errs,
|
||||
"assignment": None,
|
||||
"substrate": "control_plane_db",
|
||||
"file_lock_only": False,
|
||||
"comment_lease_only": False,
|
||||
}
|
||||
|
||||
inv_reasons: list[str] = []
|
||||
candidates: list[Any] = []
|
||||
if candidates_json:
|
||||
try:
|
||||
raw = json.loads(candidates_json)
|
||||
if not isinstance(raw, list):
|
||||
raise ValueError("candidates_json must be a JSON list")
|
||||
for item in raw:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
candidates.append(allocator_service.candidate_from_dict(item))
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return {
|
||||
"success": False,
|
||||
"outcome": allocator_service.OUTCOME_NO_SAFE,
|
||||
"apply": bool(apply),
|
||||
"reasons": [
|
||||
f"invalid candidates_json: {_redact(str(exc))} (fail closed)"
|
||||
],
|
||||
"assignment": None,
|
||||
"substrate": "control_plane_db",
|
||||
}
|
||||
else:
|
||||
candidates, inv_reasons = _allocator_candidates_from_gitea(
|
||||
remote=remote,
|
||||
host=host,
|
||||
org=o,
|
||||
repo=r,
|
||||
include_issues=include_issues,
|
||||
include_prs=include_prs,
|
||||
limit=limit,
|
||||
)
|
||||
|
||||
sid = (session_id or "").strip() or (
|
||||
f"{profile_name or 'session'}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
|
||||
)
|
||||
result = allocator_service.allocate_next_work(
|
||||
db,
|
||||
session_id=sid,
|
||||
role=role_in,
|
||||
remote=remote if remote in REMOTES else remote,
|
||||
org=o,
|
||||
repo=r,
|
||||
candidates=candidates,
|
||||
apply=bool(apply),
|
||||
profile_name=profile_name,
|
||||
username=username,
|
||||
)
|
||||
if inv_reasons:
|
||||
result.setdefault("inventory_warnings", inv_reasons)
|
||||
result["candidate_count"] = len(candidates)
|
||||
result["inventory_source"] = (
|
||||
"candidates_json" if candidates_json else "gitea_live"
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
# ── Entry point ───────────────────────────────────────────────────────────────
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user