diff --git a/allocator_service.py b/allocator_service.py new file mode 100644 index 0000000..4a45564 --- /dev/null +++ b/allocator_service.py @@ -0,0 +1,610 @@ +"""Controller-owned work allocator policy (#600). + +Builds on the #613 control-plane DB substrate (``ControlPlaneDB.assign_and_lease``). + +Workers must not self-select exclusive work under the standard multi-LLM +workflow. They call ``gitea_allocate_next_work`` which: + +1. Inspects candidate Gitea issues/PRs (never raw monitoring incidents). +2. Applies ADR routing policy (role, terminal path, leases, blocked, deps). +3. Atomically assigns + leases the selected item in one DB transaction. + +This module is pure selection + substrate orchestration. Gitea I/O for live +inventory lives in the MCP tool wrapper so tests can inject candidates. + +#612 remains downstream: bridge-created Gitea issues become candidates only +after they exist as normal issues; this module never assigns incidents. +""" + +from __future__ import annotations + +import os +import uuid +from dataclasses import dataclass, field +from typing import Any, Sequence + +from control_plane_db import ( + ControlPlaneDB, + ControlPlaneError, + InvalidWorkKindError, + LeaseRequiredError, + WORK_KINDS, +) + +# Outcomes required by #600 / ADR §5. +OUTCOME_ASSIGNED = "assigned_work" +OUTCOME_WAIT = "wait" +OUTCOME_BLOCKED_TERMINAL = "blocked_by_terminal_path" +OUTCOME_BLOCKED_LEASE = "blocked_by_active_lease" +OUTCOME_NEEDS_CONTROLLER = "needs_controller" +OUTCOME_NO_SAFE = "no_safe_work" +OUTCOME_ROLE_INELIGIBLE = "role_ineligible" +OUTCOME_PREVIEW = "preview" # dry-run only (apply=false) + +ROLE_AUTHOR = "author" +ROLE_REVIEWER = "reviewer" +ROLE_MERGER = "merger" +ROLE_RECONCILER = "reconciler" +ROLE_CONTROLLER = "controller" + +VALID_ROLES = frozenset( + {ROLE_AUTHOR, ROLE_REVIEWER, ROLE_MERGER, ROLE_RECONCILER, ROLE_CONTROLLER} +) + +# Default action matrices by role (mutation gate will re-check). +ROLE_ACTIONS: dict[str, tuple[tuple[str, ...], tuple[str, ...]]] = { + ROLE_AUTHOR: ( + ("implement", "comment", "push", "create_pr"), + ("approve", "merge", "request_changes", "self_select_without_assignment"), + ), + ROLE_REVIEWER: ( + ("review", "comment", "approve", "request_changes"), + ("merge", "push", "create_pr", "self_select_without_assignment"), + ), + ROLE_MERGER: ( + ("merge", "comment"), + ("approve", "request_changes", "push", "create_pr", "self_select_without_assignment"), + ), + ROLE_RECONCILER: ( + ("comment", "diagnose", "cleanup"), + ("approve", "merge", "push", "create_pr", "self_select_without_assignment"), + ), + ROLE_CONTROLLER: ( + ("comment", "diagnose", "allocate"), + ("approve", "merge", "push", "create_pr"), + ), +} + + +@dataclass +class WorkCandidate: + """One assignable Gitea issue or PR presented to the allocator.""" + + kind: str # issue | pr + number: int + state: str = "open" + labels: tuple[str, ...] = () + title: str = "" + priority: int = 0 + head_sha: str | None = None + # Routing signals (callers derive from Gitea / review feedback). + request_changes_current_head: bool = False + approval_on_current_head: bool = False + approval_stale: bool = False + approval_contaminated: bool = False + mergeable: bool = False + blocked: bool = False + dependency_unmet: bool = False + dependency_reason: str | None = None + already_claimed_elsewhere: bool = False + + def __post_init__(self) -> None: + self.kind = (self.kind or "").strip().lower() + self.state = (self.state or "open").strip().lower() + self.labels = tuple( + str(x).strip().lower() for x in (self.labels or ()) if str(x).strip() + ) + if self.kind not in WORK_KINDS: + raise InvalidWorkKindError( + f"candidate kind '{self.kind}' is not assignable; only " + f"{sorted(WORK_KINDS)} (never raw incidents)" + ) + + def as_dict(self) -> dict[str, Any]: + return { + "kind": self.kind, + "number": self.number, + "state": self.state, + "labels": list(self.labels), + "title": self.title, + "priority": self.priority, + "head_sha": self.head_sha, + "request_changes_current_head": self.request_changes_current_head, + "approval_on_current_head": self.approval_on_current_head, + "approval_stale": self.approval_stale, + "approval_contaminated": self.approval_contaminated, + "mergeable": self.mergeable, + "blocked": self.blocked, + "dependency_unmet": self.dependency_unmet, + "dependency_reason": self.dependency_reason, + } + + +@dataclass +class SkipRecord: + kind: str + number: int + reason: str + + def as_dict(self) -> dict[str, Any]: + return {"kind": self.kind, "number": self.number, "reason": self.reason} + + +def normalize_role(role: str | None, *, profile_name: str | None = None) -> str: + """Map profile/role strings to a canonical allocator role.""" + raw = (role or "").strip().lower() + if raw in VALID_ROLES: + return raw + prof = (profile_name or "").strip().lower() + for token in VALID_ROLES: + if token in prof or prof.endswith(f"-{token}"): + return token + if "author" in raw: + return ROLE_AUTHOR + if "review" in raw: + return ROLE_REVIEWER + if "merg" in raw: + return ROLE_MERGER + if "reconcil" in raw: + return ROLE_RECONCILER + if "control" in raw: + return ROLE_CONTROLLER + raise ControlPlaneError( + f"unknown allocator role '{role}' (profile={profile_name!r}); " + f"expected one of {sorted(VALID_ROLES)}" + ) + + +def expected_role_for_candidate(c: WorkCandidate) -> str: + """ADR §5.3 routing: which role should take this work next.""" + if c.kind == "pr": + if c.approval_contaminated: + return ROLE_RECONCILER + if c.request_changes_current_head: + return ROLE_AUTHOR + if c.approval_stale: + return ROLE_REVIEWER + if c.approval_on_current_head and c.mergeable: + return ROLE_MERGER + # Open PR without terminal verdict → reviewer + return ROLE_REVIEWER + # Issues: ready work → author by default; blocked stays controller/none + labels = set(c.labels) + if "status:blocked" in labels or c.blocked: + return ROLE_CONTROLLER + return ROLE_AUTHOR + + +def role_actions(role: str) -> tuple[tuple[str, ...], tuple[str, ...]]: + return ROLE_ACTIONS.get(role, ROLE_ACTIONS[ROLE_AUTHOR]) + + +def classify_skip( + c: WorkCandidate, + *, + role: str, + terminal_pr: int | None, +) -> str | None: + """Return skip reason, or None if candidate is selectable for *role*.""" + if c.state in ("merged", "closed"): + return f"{c.kind}#{c.number} is {c.state}; never assign" + if c.blocked or "status:blocked" in c.labels: + return f"{c.kind}#{c.number} is blocked" + if c.dependency_unmet: + return ( + c.dependency_reason + or f"{c.kind}#{c.number} has unmet dependencies" + ) + if c.already_claimed_elsewhere: + return f"{c.kind}#{c.number} already claimed elsewhere" + if c.kind == "pr" and not (c.head_sha or "").strip(): + return f"pr#{c.number} missing head_sha pin" + + # Terminal path first: when an active terminal PR exists, only that PR + # (or controller diagnosis) is assignable for review-path roles. + if terminal_pr is not None and c.kind == "pr" and c.number != terminal_pr: + if role in (ROLE_REVIEWER, ROLE_MERGER): + return ( + f"pr#{c.number} skipped: active terminal-review lock on " + f"PR #{terminal_pr} must be resolved first" + ) + + expected = expected_role_for_candidate(c) + if role == ROLE_CONTROLLER: + # Controller may inspect anything but only assigns diagnosis targets + # when contaminated / blocked. + if expected == ROLE_RECONCILER or c.blocked: + return None + return f"{c.kind}#{c.number} does not require controller (expected {expected})" + + if role != expected: + return ( + f"{c.kind}#{c.number} expects role '{expected}', active role is '{role}'" + ) + + # Ready-gate for issues: prefer status:ready when labels present. + if c.kind == "issue" and c.labels: + if "status:ready" not in c.labels and "status:in-progress" not in c.labels: + # Allow unlabeled open issues; only skip explicit non-ready states. + if any(l.startswith("status:") for l in c.labels): + return f"issue#{c.number} not status:ready ({','.join(c.labels)})" + return None + + +def sort_candidates(candidates: Sequence[WorkCandidate]) -> list[WorkCandidate]: + """Higher priority first; then lower number (older issues) for stability.""" + return sorted( + candidates, + key=lambda c: (-int(c.priority), c.kind != "pr", int(c.number)), + ) + + +def allocate_next_work( + db: ControlPlaneDB, + *, + session_id: str, + role: str, + remote: str, + org: str, + repo: str, + candidates: Sequence[WorkCandidate], + apply: bool = False, + profile_name: str | None = None, + username: str | None = None, + lease_ttl_seconds: int | None = None, +) -> dict[str, Any]: + """Select and optionally reserve the next work unit via control-plane DB. + + *apply=False* (default): dry-run selection only — no lease/assignment. + *apply=True*: atomic ``assign_and_lease`` for the selected candidate. + + Never uses file locks or comment-only leases as the assignment source. + """ + if db is None: + return { + "success": False, + "outcome": OUTCOME_NO_SAFE, + "reasons": [ + "control-plane DB substrate unavailable (fail closed, #600/#613)" + ], + "skipped": [], + "assignment": None, + "substrate": "control_plane_db", + "file_lock_only": False, + "comment_lease_only": False, + } + + try: + role_norm = normalize_role(role, profile_name=profile_name) + except ControlPlaneError as exc: + return { + "success": False, + "outcome": OUTCOME_ROLE_INELIGIBLE, + "reasons": [str(exc)], + "skipped": [], + "assignment": None, + "substrate": "control_plane_db", + } + + session_id = (session_id or "").strip() or f"alloc-{uuid.uuid4().hex[:12]}" + try: + db.upsert_session( + session_id=session_id, + role=role_norm, + profile=profile_name, + pid=os.getpid(), + ) + except Exception as exc: # noqa: BLE001 — surface structured + return { + "success": False, + "outcome": OUTCOME_NO_SAFE, + "reasons": [ + f"failed to register session in control-plane DB: {exc} " + "(fail closed, #613)" + ], + "skipped": [], + "assignment": None, + "substrate": "control_plane_db", + } + + # Expire stale leases globally before selection. + try: + db.expire_stale_leases() + except Exception as exc: # noqa: BLE001 + return { + "success": False, + "outcome": OUTCOME_NO_SAFE, + "reasons": [f"lease expiry failed: {exc} (fail closed)"], + "skipped": [], + "assignment": None, + "substrate": "control_plane_db", + } + + terminal = None + try: + terminal = db.get_active_terminal_lock(remote=remote, org=org, repo=repo) + except Exception as exc: # noqa: BLE001 + return { + "success": False, + "outcome": OUTCOME_NO_SAFE, + "reasons": [f"terminal lock lookup failed: {exc} (fail closed)"], + "skipped": [], + "assignment": None, + "substrate": "control_plane_db", + } + terminal_pr = int(terminal["terminal_pr"]) if terminal else None + + skipped: list[SkipRecord] = [] + ordered = sort_candidates(list(candidates)) + selected: WorkCandidate | None = None + for c in ordered: + reason = classify_skip(c, role=role_norm, terminal_pr=terminal_pr) + if reason: + skipped.append(SkipRecord(c.kind, c.number, reason)) + continue + selected = c + break + + if selected is None: + # If terminal lock blocks all review work, surface that explicitly. + if terminal_pr is not None and role_norm in (ROLE_REVIEWER, ROLE_MERGER): + outcome = OUTCOME_BLOCKED_TERMINAL + reasons = [ + f"no safe work for role '{role_norm}': active terminal-review " + f"lock on PR #{terminal_pr} (resolve terminal path first, #332/#600)" + ] + else: + outcome = OUTCOME_NO_SAFE + reasons = [ + f"no safe assignable work for role '{role_norm}' " + f"among {len(ordered)} candidates" + ] + return { + "success": True, + "outcome": outcome, + "apply": bool(apply), + "role": role_norm, + "profile_name": profile_name, + "username": username, + "session_id": session_id, + "remote": remote, + "org": org, + "repo": repo, + "selected": None, + "expected_role_next": None, + "reasons": reasons, + "skipped": [s.as_dict() for s in skipped], + "terminal_pr": terminal_pr, + "assignment": None, + "substrate": "control_plane_db", + "file_lock_only": False, + "comment_lease_only": False, + "downstream_note": ( + "#612 incident bridge remains downstream of #600; " + "allocator never assigns raw monitoring incidents" + ), + } + + expected_role = expected_role_for_candidate(selected) + allowed, forbidden = role_actions(role_norm) + selection = { + "kind": selected.kind, + "number": selected.number, + "title": selected.title, + "labels": list(selected.labels), + "head_sha": selected.head_sha, + "priority": selected.priority, + "expected_role_next": expected_role, + "reason_selected": ( + f"highest-priority candidate for role '{role_norm}' " + f"(expected_role={expected_role})" + ), + } + + if not apply: + return { + "success": True, + "outcome": OUTCOME_PREVIEW, + "apply": False, + "role": role_norm, + "profile_name": profile_name, + "username": username, + "session_id": session_id, + "remote": remote, + "org": org, + "repo": repo, + "selected": selection, + "expected_role_next": expected_role, + "reasons": [ + "dry-run only (apply=false); no assignment/lease created — " + "call again with apply=true to reserve via control-plane DB" + ], + "skipped": [s.as_dict() for s in skipped], + "terminal_pr": terminal_pr, + "assignment": None, + "substrate": "control_plane_db", + "file_lock_only": False, + "comment_lease_only": False, + "downstream_note": ( + "#612 incident bridge remains downstream of #600; " + "allocator never assigns raw monitoring incidents" + ), + } + + # Atomic reserve via #613 substrate. + ttl = lease_ttl_seconds if lease_ttl_seconds is not None else None + try: + kwargs: dict[str, Any] = { + "session_id": session_id, + "role": role_norm, + "remote": remote, + "org": org, + "repo": repo, + "kind": selected.kind, + "number": selected.number, + "expected_head_sha": selected.head_sha, + "allowed_actions": allowed, + "forbidden_actions": forbidden, + "phase": "allocated", + } + if ttl is not None: + kwargs["lease_ttl_seconds"] = int(ttl) + result = db.assign_and_lease(**kwargs) + except (InvalidWorkKindError, LeaseRequiredError, ControlPlaneError) as exc: + return { + "success": False, + "outcome": OUTCOME_NO_SAFE, + "apply": True, + "role": role_norm, + "session_id": session_id, + "remote": remote, + "org": org, + "repo": repo, + "selected": selection, + "expected_role_next": expected_role, + "reasons": [f"atomic assign+lease failed: {exc} (fail closed, #613)"], + "skipped": [s.as_dict() for s in skipped], + "terminal_pr": terminal_pr, + "assignment": None, + "substrate": "control_plane_db", + "file_lock_only": False, + "comment_lease_only": False, + } + + if result.outcome == "wait": + return { + "success": True, + "outcome": OUTCOME_WAIT, + "apply": True, + "role": role_norm, + "session_id": session_id, + "remote": remote, + "org": org, + "repo": repo, + "selected": selection, + "expected_role_next": expected_role, + "reasons": [result.reason or "foreign active lease"], + "skipped": [s.as_dict() for s in skipped], + "terminal_pr": terminal_pr, + "assignment": result.as_dict(), + "owner_session_id": result.owner_session_id, + "substrate": "control_plane_db", + "file_lock_only": False, + "comment_lease_only": False, + } + + if result.outcome == "no_safe_work": + return { + "success": True, + "outcome": OUTCOME_NO_SAFE, + "apply": True, + "role": role_norm, + "session_id": session_id, + "remote": remote, + "org": org, + "repo": repo, + "selected": selection, + "expected_role_next": expected_role, + "reasons": [result.reason or "no_safe_work"], + "skipped": [s.as_dict() for s in skipped], + "terminal_pr": terminal_pr, + "assignment": result.as_dict(), + "substrate": "control_plane_db", + "file_lock_only": False, + "comment_lease_only": False, + } + + # assigned + return { + "success": True, + "outcome": OUTCOME_ASSIGNED, + "apply": True, + "role": role_norm, + "profile_name": profile_name, + "username": username, + "session_id": session_id, + "remote": remote, + "org": org, + "repo": repo, + "selected": selection, + "expected_role_next": expected_role, + "reasons": [ + selection["reason_selected"], + result.reason or "atomic assign+lease created", + ], + "skipped": [s.as_dict() for s in skipped], + "terminal_pr": terminal_pr, + "assignment": result.as_dict(), + "lease_proof": { + "assignment_id": result.assignment_id, + "lease_id": result.lease_id, + "expires_at": result.expires_at, + "expected_head_sha": result.expected_head_sha, + "allowed_actions": list(result.allowed_actions), + "forbidden_actions": list(result.forbidden_actions), + "source": "control_plane_db.assign_and_lease", + }, + "next_valid_command": _next_command(role_norm, selected), + "substrate": "control_plane_db", + "file_lock_only": False, + "comment_lease_only": False, + "downstream_note": ( + "#612 incident bridge remains downstream of #600; " + "allocator never assigns raw monitoring incidents" + ), + } + + +def _next_command(role: str, c: WorkCandidate) -> str: + if role == ROLE_AUTHOR and c.kind == "issue": + return f"implement issue #{c.number} under a branches/ worktree; open PR when ready" + if role == ROLE_AUTHOR and c.kind == "pr": + return ( + f"address REQUEST_CHANGES on PR #{c.number} at head " + f"{(c.head_sha or '')[:12]} and push fixes" + ) + if role == ROLE_REVIEWER: + return ( + f"review PR #{c.number} pinned at head {(c.head_sha or '')[:12]} " + "via full reviewer workflow" + ) + if role == ROLE_MERGER: + return ( + f"merge PR #{c.number} only with explicit operator MERGE " + f"authorization at head {(c.head_sha or '')[:12]}" + ) + if role == ROLE_RECONCILER: + return f"diagnose contested state for {c.kind}#{c.number}" + return f"proceed on {c.kind}#{c.number} under role {role}" + + +def candidate_from_dict(data: dict[str, Any]) -> WorkCandidate: + """Build a WorkCandidate from a plain dict (tests / MCP inventory).""" + return WorkCandidate( + kind=str(data.get("kind") or "issue"), + number=int(data["number"]), + state=str(data.get("state") or "open"), + labels=tuple(data.get("labels") or ()), + title=str(data.get("title") or ""), + priority=int(data.get("priority") or 0), + head_sha=data.get("head_sha"), + request_changes_current_head=bool(data.get("request_changes_current_head")), + approval_on_current_head=bool(data.get("approval_on_current_head")), + approval_stale=bool(data.get("approval_stale")), + approval_contaminated=bool(data.get("approval_contaminated")), + mergeable=bool(data.get("mergeable")), + blocked=bool(data.get("blocked")), + dependency_unmet=bool(data.get("dependency_unmet")), + dependency_reason=data.get("dependency_reason"), + already_claimed_elsewhere=bool(data.get("already_claimed_elsewhere")), + ) diff --git a/docs/architecture/control-plane-db-substrate.md b/docs/architecture/control-plane-db-substrate.md index 3eefeb8..1947794 100644 --- a/docs/architecture/control-plane-db-substrate.md +++ b/docs/architecture/control-plane-db-substrate.md @@ -40,9 +40,19 @@ - **#600** must call this substrate (not file locks / comment-only leases alone) for completion. - **#612** must write `incident_links` here and create **Gitea issues**; the allocator assigns those issues, not raw incidents. +## Allocator API (#600) + +Module: `allocator_service.py` · MCP tool: `gitea_allocate_next_work` + +- Workers call `gitea_allocate_next_work(apply=false|true)` instead of self-selecting work. +- `apply=false` returns a dry-run selection (`outcome=preview`) with skip reasons. +- `apply=true` reserves via `ControlPlaneDB.assign_and_lease` (atomic assignment+lease). +- Coordination source is **always** the control-plane DB — not file locks or comment-only leases. +- Routing follows ADR §5.3 (REQUEST_CHANGES → author, terminal path first, foreign lease → wait). +- Raw monitoring incidents are never candidates (#612 remains downstream). + ## Non-goals (intentionally deferred) -- Full `gitea_allocate_next_work` routing policy (REQUEST_CHANGES → author, etc.) — **#600** - Sentry/GlitchTip provider adapters and auto-watchdog — **#612** - Gitea comment/label mirror writers (may be added by allocator tooling later) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 636eaea..a22faea 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -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__": diff --git a/task_capability_map.py b/task_capability_map.py index 3842c2b..1ee097d 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -139,6 +139,17 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.pr.create", "role": "author", }, + # #600: controller-owned allocator — any authenticated profile may call; + # routing enforces role match to selected work. Uses control-plane DB (#613). + "allocate_next_work": { + "permission": "gitea.read", + "role": "author", + }, + "gitea_allocate_next_work": { + "permission": "gitea.read", + "role": "author", + }, + "reconcile_landed_pr": { "permission": "gitea.read", "role": "author", diff --git a/tests/test_allocator_service.py b/tests/test_allocator_service.py new file mode 100644 index 0000000..3e252a1 --- /dev/null +++ b/tests/test_allocator_service.py @@ -0,0 +1,366 @@ +"""Tests for controller-owned allocator (#600) on control-plane DB (#613).""" + +from __future__ import annotations + +import os +import tempfile +import threading +import unittest +from concurrent.futures import ThreadPoolExecutor, as_completed + +from allocator_service import ( + OUTCOME_ASSIGNED, + OUTCOME_BLOCKED_TERMINAL, + OUTCOME_NO_SAFE, + OUTCOME_PREVIEW, + OUTCOME_WAIT, + WorkCandidate, + allocate_next_work, + candidate_from_dict, + classify_skip, + expected_role_for_candidate, +) +from control_plane_db import ControlPlaneDB, InvalidWorkKindError + + +class AllocatorServiceTest(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self._tmp.name, "cp.sqlite3") + self.db = ControlPlaneDB(self.db_path) + + def tearDown(self) -> None: + self._tmp.cleanup() + + def _alloc(self, **kwargs): + defaults = dict( + db=self.db, + session_id="s-test", + role="author", + remote="prgs", + org="org", + repo="repo", + candidates=[], + apply=False, + profile_name="prgs-author", + username="jcwalker3", + ) + defaults.update(kwargs) + return allocate_next_work(**defaults) + + def test_selects_ready_issue_for_author(self) -> None: + cands = [ + WorkCandidate( + kind="issue", + number=612, + labels=("status:ready",), + title="bridge", + dependency_unmet=True, + dependency_reason="downstream of #600", + ), + WorkCandidate( + kind="issue", + number=600, + labels=("status:ready", "type:feature"), + title="allocator", + priority=50, + ), + WorkCandidate( + kind="issue", + number=601, + labels=("status:blocked",), + title="blocked", + blocked=True, + ), + ] + res = self._alloc(candidates=cands, apply=False) + self.assertTrue(res["success"]) + self.assertEqual(res["outcome"], OUTCOME_PREVIEW) + self.assertEqual(res["selected"]["number"], 600) + self.assertEqual(res["selected"]["kind"], "issue") + # When 600 is highest priority valid work, lower-priority blocked/dep + # candidates are not visited. Prove they are skipped when they sort first. + res2 = self._alloc( + candidates=[ + WorkCandidate( + kind="issue", + number=612, + labels=("status:ready",), + priority=99, + dependency_unmet=True, + dependency_reason="downstream of #600", + ), + WorkCandidate( + kind="issue", + number=601, + labels=("status:blocked",), + priority=98, + blocked=True, + ), + WorkCandidate( + kind="issue", + number=600, + labels=("status:ready",), + priority=1, + ), + ], + apply=False, + ) + skipped_nums = {s["number"] for s in res2["skipped"]} + self.assertIn(612, skipped_nums) + self.assertIn(601, skipped_nums) + self.assertEqual(res2["selected"]["number"], 600) + self.assertIsNone(res["assignment"]) + self.assertFalse(res["file_lock_only"]) + self.assertFalse(res["comment_lease_only"]) + + def test_atomic_assign_and_lease_on_apply(self) -> None: + cands = [ + WorkCandidate( + kind="issue", + number=600, + labels=("status:ready",), + priority=10, + ) + ] + res = self._alloc(candidates=cands, apply=True, session_id="s-a") + self.assertEqual(res["outcome"], OUTCOME_ASSIGNED) + asn = res["assignment"] + self.assertEqual(asn["outcome"], "assigned") + self.assertIsNotNone(asn["assignment_id"]) + self.assertIsNotNone(asn["lease_id"]) + self.assertEqual(asn["work_number"], 600) + self.assertIn("implement", asn["allowed_actions"]) + self.assertIn("merge", asn["forbidden_actions"]) + proof = res["lease_proof"] + self.assertEqual(proof["source"], "control_plane_db.assign_and_lease") + + def test_concurrent_allocators_no_double_assign(self) -> None: + cand = WorkCandidate( + kind="pr", + number=100, + head_sha="a" * 40, + priority=10, + ) + # Seed work item so both race the same target + self.db.upsert_work_item( + remote="prgs", + org="org", + repo="repo", + kind="pr", + number=100, + current_head_sha="a" * 40, + ) + results = [] + lock = threading.Lock() + + def worker(sid: str): + r = allocate_next_work( + self.db, + session_id=sid, + role="reviewer", + remote="prgs", + org="org", + repo="repo", + candidates=[cand], + apply=True, + profile_name="prgs-reviewer", + ) + with lock: + results.append(r) + + with ThreadPoolExecutor(max_workers=2) as pool: + futs = [pool.submit(worker, f"s-{i}") for i in range(2)] + for f in as_completed(futs): + f.result() + + outcomes = [r["outcome"] for r in results] + self.assertEqual(outcomes.count(OUTCOME_ASSIGNED), 1, outcomes) + self.assertEqual(outcomes.count(OUTCOME_WAIT), 1, outcomes) + + def test_blocked_and_dependency_skipped(self) -> None: + cands = [ + WorkCandidate(kind="issue", number=1, blocked=True, labels=("status:blocked",)), + WorkCandidate( + kind="issue", + number=612, + labels=("status:ready",), + dependency_unmet=True, + dependency_reason="downstream of #600", + ), + ] + res = self._alloc(candidates=cands, apply=True, role="author") + self.assertEqual(res["outcome"], OUTCOME_NO_SAFE) + self.assertIsNone(res["selected"]) + reasons = " ".join(s["reason"] for s in res["skipped"]) + self.assertIn("blocked", reasons.lower()) + self.assertIn("600", reasons) + + def test_already_leased_returns_wait(self) -> None: + self.db.upsert_session(session_id="owner", role="author") + self.db.assign_and_lease( + session_id="owner", + role="author", + remote="prgs", + org="org", + repo="repo", + kind="issue", + number=50, + ) + res = self._alloc( + session_id="other", + candidates=[ + WorkCandidate(kind="issue", number=50, labels=("status:ready",)) + ], + apply=True, + ) + self.assertEqual(res["outcome"], OUTCOME_WAIT) + self.assertEqual(res["owner_session_id"], "owner") + + def test_role_ineligible_skipped(self) -> None: + # PR with current-head REQUEST_CHANGES expects author, not reviewer. + cand = WorkCandidate( + kind="pr", + number=9, + head_sha="b" * 40, + request_changes_current_head=True, + ) + self.assertEqual(expected_role_for_candidate(cand), "author") + res = self._alloc( + role="reviewer", + profile_name="prgs-reviewer", + candidates=[cand], + apply=False, + ) + self.assertEqual(res["outcome"], OUTCOME_NO_SAFE) + self.assertTrue(any("expects role" in s["reason"] for s in res["skipped"])) + + def test_terminal_lock_blocks_downstream_reviewer_prs(self) -> None: + self.db.set_terminal_lock( + remote="prgs", + org="org", + repo="repo", + terminal_pr=10, + decision="request_changes", + status="active", + ) + cands = [ + WorkCandidate(kind="pr", number=11, head_sha="c" * 40, priority=5), + WorkCandidate(kind="pr", number=10, head_sha="d" * 40, priority=1), + ] + res = self._alloc( + role="reviewer", + profile_name="prgs-reviewer", + candidates=cands, + apply=False, + ) + # Terminal PR #10 is selectable; #11 skipped for terminal path. + self.assertEqual(res["selected"]["number"], 10) + self.assertTrue( + any( + s["number"] == 11 and "terminal-review lock" in s["reason"] + for s in res["skipped"] + ) + ) + + def test_terminal_lock_blocks_all_when_only_downstream(self) -> None: + self.db.set_terminal_lock( + remote="prgs", + org="org", + repo="repo", + terminal_pr=10, + decision="request_changes", + status="active", + ) + res = self._alloc( + role="reviewer", + profile_name="prgs-reviewer", + candidates=[ + WorkCandidate(kind="pr", number=99, head_sha="e" * 40), + ], + apply=False, + ) + self.assertEqual(res["outcome"], OUTCOME_BLOCKED_TERMINAL) + + def test_no_work_structured(self) -> None: + res = self._alloc(candidates=[], apply=True) + self.assertEqual(res["outcome"], OUTCOME_NO_SAFE) + self.assertIsNone(res["selected"]) + self.assertTrue(res["reasons"]) + + def test_rejects_incident_kind(self) -> None: + with self.assertRaises(InvalidWorkKindError): + WorkCandidate(kind="sentry_incident", number=1) + + def test_612_downstream_marker_in_result(self) -> None: + res = self._alloc( + candidates=[ + WorkCandidate(kind="issue", number=600, labels=("status:ready",)) + ], + apply=True, + ) + self.assertIn("612", res.get("downstream_note", "")) + + def test_substrate_not_file_or_comment_lease(self) -> None: + res = self._alloc( + candidates=[ + WorkCandidate(kind="issue", number=1, labels=("status:ready",)) + ], + apply=True, + ) + self.assertEqual(res["substrate"], "control_plane_db") + self.assertFalse(res["file_lock_only"]) + self.assertFalse(res["comment_lease_only"]) + self.assertEqual( + res["lease_proof"]["source"], "control_plane_db.assign_and_lease" + ) + + def test_candidate_from_dict(self) -> None: + c = candidate_from_dict( + { + "kind": "pr", + "number": 7, + "head_sha": "f" * 40, + "approval_on_current_head": True, + "mergeable": True, + } + ) + self.assertEqual(expected_role_for_candidate(c), "merger") + + def test_merger_gets_clean_approval(self) -> None: + c = WorkCandidate( + kind="pr", + number=3, + head_sha="1" * 40, + approval_on_current_head=True, + mergeable=True, + priority=100, + ) + res = self._alloc( + role="merger", + profile_name="prgs-merger", + candidates=[c], + apply=True, + session_id="merger-1", + ) + self.assertEqual(res["outcome"], OUTCOME_ASSIGNED) + self.assertEqual(res["assignment"]["work_number"], 3) + self.assertIn("merge", res["assignment"]["allowed_actions"]) + + def test_db_unavailable_fails_closed(self) -> None: + res = allocate_next_work( + None, # type: ignore[arg-type] + session_id="x", + role="author", + remote="prgs", + org="o", + repo="r", + candidates=[], + apply=True, + ) + self.assertFalse(res["success"]) + self.assertIn("unavailable", res["reasons"][0].lower()) + + +if __name__ == "__main__": + unittest.main()