From ad13d872df3ac22630617c4805832d0875e258a3 Mon Sep 17 00:00:00 2001 From: jcwalker3 Date: Mon, 20 Jul 2026 13:52:39 -0500 Subject: [PATCH] fix: exclude foreign-claimed work from allocation instead of blockading the queue (Closes #765) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Recovered preserved candidate d06198b onto current master 0c2f45a via clean cherry-pick. Active foreign leases are excluded before allocator ranking; controller_instance_id ownership is persisted; dashboard matches allocator exclusion. Stable patch-id bacafc5f… preserved. Co-Authored-By: Claude Opus 4.8 (1M context) --- allocator_service.py | 192 ++++++++- control_plane_db.py | 125 +++++- gitea_mcp_server.py | 16 + .../test_allocator_foreign_lease_exclusion.py | 380 ++++++++++++++++++ workflow_dashboard.py | 44 +- 5 files changed, 735 insertions(+), 22 deletions(-) create mode 100644 tests/test_allocator_foreign_lease_exclusion.py diff --git a/allocator_service.py b/allocator_service.py index bd35b73..d6d6a62 100644 --- a/allocator_service.py +++ b/allocator_service.py @@ -21,7 +21,7 @@ from __future__ import annotations import os import uuid from dataclasses import dataclass, field -from typing import Any, Sequence +from typing import Any, Mapping, Sequence from control_plane_db import ( ControlPlaneDB, @@ -40,6 +40,16 @@ OUTCOME_NEEDS_CONTROLLER = "needs_controller" OUTCOME_NO_SAFE = "no_safe_work" OUTCOME_ROLE_INELIGIBLE = "role_ineligible" OUTCOME_PREVIEW = "preview" # dry-run only (apply=false) +# #765: ownership could not be established for every remaining candidate. +OUTCOME_OWNERSHIP_DEFECT = "allocator_ownership_defect" + +# #765 skip reason code for work already claimed by a different controller. +SKIP_CLAIMED_BY_OTHER_SESSION = "claimed_by_other_session" + +# Ownership verdicts for a live claim on a candidate (#765). +OWNERSHIP_OWN = "own" +OWNERSHIP_FOREIGN = "foreign" +OWNERSHIP_UNKNOWN = "unknown" # Human-readable statement of how a winner is chosen (#758 AC10). Reported # alongside allocator results so the flat status:ready tier and its @@ -144,9 +154,76 @@ class SkipRecord: kind: str number: int reason: str + reason_code: str | None = None def as_dict(self) -> dict[str, Any]: - return {"kind": self.kind, "number": self.number, "reason": self.reason} + return { + "kind": self.kind, + "number": self.number, + "reason": self.reason, + "reason_code": self.reason_code, + } + + +CONTROLLER_INSTANCE_ENV = "GITEA_CONTROLLER_INSTANCE_ID" + + +def resolve_controller_instance_id( + env: Mapping[str, str] | None = None, +) -> str | None: + """Return this controller's stable identity, or ``None`` if undeclared. + + Deliberately has no derived fallback. The obvious candidates are unsafe: + ``session_id`` is regenerated per invocation, and the MCP process pid is + shared by every controller attached to the same daemon — two independent + controllers really do report the same pid and profile. Guessing from either + would let one controller adopt another's lease, which is the failure #765 + exists to prevent. When this returns ``None``, live claims are treated as + unidentified: they are excluded from selection and reported as ownership + defects rather than adopted. + """ + source = env if env is not None else os.environ + return (source.get(CONTROLLER_INSTANCE_ENV) or "").strip() or None + + +def classify_claim_ownership( + claim: dict[str, Any] | None, + *, + session_id: str | None, + controller_instance_id: str | None, +) -> str | None: + """Classify a live claim as own / foreign / unknown ownership (#765). + + Returns ``None`` when the candidate carries no live claim. + + Session ids are regenerated per allocator invocation, so they only prove + ownership positively (an exact match is certainly this session). The + durable signal is ``controller_instance_id``. When either side lacks one, + ownership is *unknown*: the allocator must not assume that a lease sharing + the same profile belongs to this controller, so unknown is treated as + not-ours for selection purposes and reported as an ownership defect. + """ + if not claim: + return None + claim_session = str(claim.get("session_id") or "").strip() + claim_instance = str(claim.get("controller_instance_id") or "").strip() + own_session = str(session_id or "").strip() + own_instance = str(controller_instance_id or "").strip() + + if claim_session and own_session and claim_session == own_session: + return OWNERSHIP_OWN + if claim_instance and own_instance: + return ( + OWNERSHIP_OWN if claim_instance == own_instance else OWNERSHIP_FOREIGN + ) + if not claim_instance and not own_instance: + # Neither side declares a controller identity. The session ids differ + # (an exact match returned OWN above), so this is simply someone + # else's lease: foreign, and we wait rather than adopt. + return OWNERSHIP_FOREIGN + # Exactly one side is identified, so the two cannot be compared: this may + # or may not be our own task under a different session id. Never guess. + return OWNERSHIP_UNKNOWN def normalize_role(role: str | None, *, profile_name: str | None = None) -> str: @@ -203,8 +280,16 @@ def classify_skip( *, role: str, terminal_pr: int | None, + claim_ownership: str | None = None, ) -> str | None: - """Return skip reason, or None if candidate is selectable for *role*.""" + """Return skip reason, or None if candidate is selectable for *role*. + + *claim_ownership* (#765) is the verdict from + :func:`classify_claim_ownership` for this candidate's live claim. Foreign + and unknown claims are excluded so one session's in-progress task can never + blockade the queue for a different controller; ``own`` stays selectable so + a controller can resume its own work. + """ 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: @@ -214,6 +299,16 @@ def classify_skip( c.dependency_reason or f"{c.kind}#{c.number} has unmet dependencies" ) + if claim_ownership in (OWNERSHIP_FOREIGN, OWNERSHIP_UNKNOWN): + detail = ( + "owned by another controller instance" + if claim_ownership == OWNERSHIP_FOREIGN + else "owner could not be identified; never adopt on a guess" + ) + return ( + f"{c.kind}#{c.number} {SKIP_CLAIMED_BY_OTHER_SESSION}: " + f"active lease {detail}" + ) 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(): @@ -287,6 +382,8 @@ def allocate_next_work( profile_name: str | None = None, username: str | None = None, lease_ttl_seconds: int | None = None, + controller_instance_id: str | None = None, + claims: Mapping[tuple[str, int], dict[str, Any]] | None = None, ) -> dict[str, Any]: """Select and optionally reserve the next work unit via control-plane DB. @@ -328,6 +425,7 @@ def allocate_next_work( role=role_norm, profile=profile_name, pid=os.getpid(), + controller_instance_id=controller_instance_id, ) except Exception as exc: # noqa: BLE001 — surface structured return { @@ -369,25 +467,99 @@ def allocate_next_work( } terminal_pr = int(terminal["terminal_pr"]) if terminal else None + # #765: live claims exclude work owned by a *different* controller before + # ranking, so one session's in-progress task cannot blockade the queue. + if claims is None: + try: + claims = db.list_active_claims(remote=remote, org=org, repo=repo) + except Exception as exc: # noqa: BLE001 + return { + "success": False, + "outcome": OUTCOME_NO_SAFE, + "reasons": [ + f"active claim lookup failed: {exc} (fail closed, #765)" + ], + "skipped": [], + "assignment": None, + "substrate": "control_plane_db", + } + skipped: list[SkipRecord] = [] + claims_excluded: list[dict[str, Any]] = [] + ownership_defects: list[dict[str, Any]] = [] ordered = sort_candidates(list(candidates)) selected: WorkCandidate | None = None for c in ordered: - reason = classify_skip(c, role=role_norm, terminal_pr=terminal_pr) + claim = claims.get((c.kind, int(c.number))) if claims else None + ownership = classify_claim_ownership( + claim, + session_id=session_id, + controller_instance_id=controller_instance_id, + ) + reason = classify_skip( + c, + role=role_norm, + terminal_pr=terminal_pr, + claim_ownership=ownership, + ) if reason: - skipped.append(SkipRecord(c.kind, c.number, reason)) + is_claim_skip = SKIP_CLAIMED_BY_OTHER_SESSION in reason + skipped.append( + SkipRecord( + c.kind, + c.number, + reason, + SKIP_CLAIMED_BY_OTHER_SESSION if is_claim_skip else None, + ) + ) + if is_claim_skip and claim: + record = { + "kind": c.kind, + "number": c.number, + "ownership": ownership, + "lease_id": claim.get("lease_id"), + "owner_session_id": claim.get("session_id"), + "owner_controller_instance_id": claim.get( + "controller_instance_id" + ), + "expires_at": claim.get("expires_at"), + } + claims_excluded.append(record) + if ownership == OWNERSHIP_UNKNOWN: + ownership_defects.append(record) continue selected = c break if selected is None: # If terminal lock blocks all review work, surface that explicitly. + owner_session_id: str | None = None 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)" ] + elif ownership_defects: + # #765: every remaining candidate is claimed and at least one owner + # could not be identified. Report the defect; never adopt. + outcome = OUTCOME_OWNERSHIP_DEFECT + reasons = [ + f"no safe assignable work for role '{role_norm}': " + f"{len(ownership_defects)} candidate(s) carry an active lease " + "whose controller ownership could not be established. Record a " + "controller_instance_id on those sessions; the allocator will " + "not assume a shared profile means shared ownership (#765)." + ] + elif claims_excluded: + outcome = OUTCOME_WAIT + reasons = [ + f"no unclaimed work for role '{role_norm}': " + f"{len(claims_excluded)} candidate(s) are actively claimed by " + "another controller. Waiting; their leases are not adopted (#765)." + ] + # Preserve the pre-#765 wait contract: name the blocking owner. + owner_session_id = claims_excluded[0].get("owner_session_id") else: outcome = OUTCOME_NO_SAFE reasons = [ @@ -414,6 +586,10 @@ def allocate_next_work( "substrate": "control_plane_db", "file_lock_only": False, "comment_lease_only": False, + "controller_instance_id": controller_instance_id, + "claims_excluded": list(claims_excluded), + "ownership_defects": list(ownership_defects), + "owner_session_id": owner_session_id, "downstream_note": ( "#612 incident bridge remains downstream of #600; " "allocator never assigns raw monitoring incidents" @@ -460,6 +636,9 @@ def allocate_next_work( "substrate": "control_plane_db", "file_lock_only": False, "comment_lease_only": False, + "controller_instance_id": controller_instance_id, + "claims_excluded": list(claims_excluded), + "ownership_defects": list(ownership_defects), "downstream_note": ( "#612 incident bridge remains downstream of #600; " "allocator never assigns raw monitoring incidents" @@ -583,6 +762,9 @@ def allocate_next_work( "substrate": "control_plane_db", "file_lock_only": False, "comment_lease_only": False, + "controller_instance_id": controller_instance_id, + "claims_excluded": list(claims_excluded), + "ownership_defects": list(ownership_defects), "downstream_note": ( "#612 incident bridge remains downstream of #600; " "allocator never assigns raw monitoring incidents" diff --git a/control_plane_db.py b/control_plane_db.py index 0dd9e38..a4adb2e 100644 --- a/control_plane_db.py +++ b/control_plane_db.py @@ -302,6 +302,7 @@ class ControlPlaneDB: conn.executescript(_SCHEMA_SQL) self._migrate_incident_links_null_scope(conn) self._migrate_lease_lifecycle_columns(conn) + self._migrate_session_ownership_columns(conn) conn.execute( "INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)", ("schema_version", str(SCHEMA_VERSION)), @@ -492,32 +493,62 @@ class ControlPlaneDB: namespace: str | None = None, pid: int | None = None, status: str = "active", + controller_instance_id: str | None = None, ) -> dict[str, Any]: + """Register/refresh a session row. + + *controller_instance_id* (#765) is the stable identity of the + controller that owns this session. Session ids are regenerated per + invocation, so they cannot express "my own in-progress task"; the + controller instance can. It is never overwritten with ``None``, so a + heartbeat from a caller that does not supply one cannot erase + ownership. + """ now = _ts() + instance = (controller_instance_id or "").strip() or None with self._tx() as conn: existing = conn.execute( "SELECT session_id FROM sessions WHERE session_id = ?", (session_id,), ).fetchone() if existing: - conn.execute( - """ - UPDATE sessions - SET role = ?, profile = ?, namespace = ?, pid = ?, - last_heartbeat_at = ?, status = ? - WHERE session_id = ? - """, - (role, profile, namespace, pid, now, status, session_id), - ) + if instance is None: + conn.execute( + """ + UPDATE sessions + SET role = ?, profile = ?, namespace = ?, pid = ?, + last_heartbeat_at = ?, status = ? + WHERE session_id = ? + """, + (role, profile, namespace, pid, now, status, session_id), + ) + else: + conn.execute( + """ + UPDATE sessions + SET role = ?, profile = ?, namespace = ?, pid = ?, + last_heartbeat_at = ?, status = ?, + controller_instance_id = ? + WHERE session_id = ? + """, + ( + role, profile, namespace, pid, now, status, + instance, session_id, + ), + ) else: conn.execute( """ INSERT INTO sessions( session_id, role, profile, namespace, pid, - started_at, last_heartbeat_at, status - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + started_at, last_heartbeat_at, status, + controller_instance_id + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, - (session_id, role, profile, namespace, pid, now, now, status), + ( + session_id, role, profile, namespace, pid, now, now, + status, instance, + ), ) row = conn.execute( "SELECT * FROM sessions WHERE session_id = ?", @@ -1215,6 +1246,73 @@ class ControlPlaneDB: if name not in cols: conn.execute(f"ALTER TABLE leases ADD COLUMN {name} {decl}") + _SESSION_OWNERSHIP_COLUMNS: tuple[tuple[str, str], ...] = ( + ("controller_instance_id", "TEXT"), + ) + + def _migrate_session_ownership_columns(self, conn: sqlite3.Connection) -> None: + """Add the stable controller identity to sessions (#765). + + Pre-existing rows migrate with ``NULL``. A NULL instance is treated as + *unknown ownership* by the allocator and is never silently adopted. + """ + cols = { + row[1] + for row in conn.execute("PRAGMA table_info(sessions)").fetchall() + } + if not cols: + return + for name, decl in self._SESSION_OWNERSHIP_COLUMNS: + if name not in cols: + conn.execute(f"ALTER TABLE sessions ADD COLUMN {name} {decl}") + + def list_active_claims( + self, + *, + remote: str | None = None, + org: str | None = None, + repo: str | None = None, + role: str | None = None, + limit: int = 500, + ) -> dict[tuple[str, int], dict[str, Any]]: + """Map ``(work_kind, work_number)`` to its live claim (#765). + + Only ``active`` leases count as claims; released/expired rows never + withhold work. Callers compare the returned ``controller_instance_id`` + against their own to decide own-task vs foreign-task. + """ + claims: dict[tuple[str, int], dict[str, Any]] = {} + for row in self.list_leases( + remote=remote, + org=org, + repo=repo, + role=role, + statuses=("active",), + limit=limit, + ): + kind = str(row.get("work_kind") or "").strip().lower() + number = row.get("work_number") + if not kind or number is None: + continue + key = (kind, int(number)) + claim = { + "lease_id": row.get("lease_id"), + "session_id": row.get("session_id"), + "controller_instance_id": row.get("session_controller_instance_id"), + "role": row.get("role"), + "profile": row.get("session_profile"), + "expires_at": row.get("expires_at"), + "work_kind": kind, + "work_number": int(number), + } + # Keep the longest-lived claim when duplicates exist. + previous = claims.get(key) + if previous is None or str(claim["expires_at"] or "") > str( + previous["expires_at"] or "" + ): + claims[key] = claim + return claims + def _lease_columns(self, conn: sqlite3.Connection) -> set[str]: return { row[1] @@ -1256,7 +1354,8 @@ class ControlPlaneDB: w.number AS work_number, w.state AS work_state, w.current_head_sha AS work_head_sha, s.pid AS session_pid, s.profile AS session_profile, - s.status AS session_status + s.status AS session_status, + s.controller_instance_id AS session_controller_instance_id FROM leases l JOIN work_items w ON w.work_item_id = l.work_item_id LEFT JOIN sessions s ON s.session_id = l.session_id diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 7a44b0e..ecc3a68 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -19211,6 +19211,7 @@ def gitea_allocate_next_work( apply=bool(apply), profile_name=profile_name, username=username, + controller_instance_id=allocator_service.resolve_controller_instance_id(), ) if inv_reasons: result.setdefault("inventory_warnings", inv_reasons) @@ -19378,6 +19379,7 @@ def gitea_workflow_dashboard( leases: list[dict] = [] terminal_pr: int | None = None terminal_lock: dict | None = None + claims: dict = {} db, db_errs = _control_plane_db_or_error() if db is None: inv_reasons.extend(db_errs or ["control-plane DB unavailable"]) @@ -19416,6 +19418,18 @@ def gitea_workflow_dashboard( inv_reasons.append( f"terminal lock lookup failed: {_redact(str(exc))}" ) + try: + # #765: live claims so the dashboard never advertises another + # controller's active task as safe next work. + claims = db.list_active_claims( + remote=remote if remote in REMOTES else remote, + org=o, + repo=r, + ) + except Exception as exc: # noqa: BLE001 + inv_reasons.append( + f"active claim lookup failed: {_redact(str(exc))}" + ) snap = workflow_dashboard.build_workflow_dashboard( candidates=candidates, @@ -19427,6 +19441,8 @@ def gitea_workflow_dashboard( terminal_lock=terminal_lock, inventory_complete=inventory_complete, inventory_reasons=inv_reasons, + claims=claims, + controller_instance_id=allocator_service.resolve_controller_instance_id(), ) payload = snap.as_dict() payload["inventory_source"] = ( diff --git a/tests/test_allocator_foreign_lease_exclusion.py b/tests/test_allocator_foreign_lease_exclusion.py new file mode 100644 index 0000000..1c87829 --- /dev/null +++ b/tests/test_allocator_foreign_lease_exclusion.py @@ -0,0 +1,380 @@ +"""Allocator ownership exclusion tests (#765). + +One session's active lease must never blockade the author queue for a +different controller. Covers: foreign lease skipped, next unclaimed candidate +selected, own task resumable, task-local blocker quarantined, all-claimed -> +wait, same profile + different controller_instance_id -> different ownership, +and claimed candidates reported in skipped results. +""" + +from __future__ import annotations + +import os +import tempfile +import unittest + +from allocator_service import ( + OUTCOME_OWNERSHIP_DEFECT, + OUTCOME_PREVIEW, + OUTCOME_WAIT, + OWNERSHIP_FOREIGN, + OWNERSHIP_OWN, + OWNERSHIP_UNKNOWN, + SKIP_CLAIMED_BY_OTHER_SESSION, + WorkCandidate, + allocate_next_work, + classify_claim_ownership, + resolve_controller_instance_id, +) +from control_plane_db import ControlPlaneDB + +REMOTE = "prgs" +ORG = "Scaled-Tech-Consulting" +REPO = "Gitea-Tools" + +MINE = "ctl-mine-0001" +THEIRS = "ctl-theirs-0002" + + +def _issue(number: int, **kwargs) -> WorkCandidate: + base = dict( + kind="issue", + number=number, + state="open", + labels=("status:ready", "type:bug"), + title=f"issue {number}", + priority=20, + ) + base.update(kwargs) + return WorkCandidate(**base) + + +def _claim(number: int, *, session_id: str, instance: str | None, kind: str = "issue"): + return { + "lease_id": f"lease-{number}", + "session_id": session_id, + "controller_instance_id": instance, + "role": "author", + "profile": "prgs-author", + "expires_at": "2026-07-20T07:06:09Z", + "work_kind": kind, + "work_number": number, + } + + +class AllocatorOwnershipTestCase(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3")) + + def _allocate( + self, + candidates, + *, + claims, + session_id="sess-mine", + instance=MINE, + apply=False, + role="author", + ): + return allocate_next_work( + self.db, + session_id=session_id, + role=role, + remote=REMOTE, + org=ORG, + repo=REPO, + candidates=candidates, + apply=apply, + profile_name="prgs-author", + controller_instance_id=instance, + claims=claims, + ) + + +class TestOwnershipClassification(AllocatorOwnershipTestCase): + def test_no_claim_returns_none(self): + self.assertIsNone( + classify_claim_ownership( + None, session_id="s", controller_instance_id=MINE + ) + ) + + def test_same_controller_instance_is_own(self): + claim = _claim(1, session_id="other-session", instance=MINE) + self.assertEqual( + classify_claim_ownership( + claim, session_id="sess-mine", controller_instance_id=MINE + ), + OWNERSHIP_OWN, + ) + + def test_same_profile_different_instance_is_foreign(self): + """Shared profile must not imply shared ownership.""" + claim = _claim(1, session_id="other-session", instance=THEIRS) + self.assertEqual( + classify_claim_ownership( + claim, session_id="sess-mine", controller_instance_id=MINE + ), + OWNERSHIP_FOREIGN, + ) + + def test_exact_session_match_is_own(self): + claim = _claim(1, session_id="sess-mine", instance=None) + self.assertEqual( + classify_claim_ownership( + claim, session_id="sess-mine", controller_instance_id=None + ), + OWNERSHIP_OWN, + ) + + def test_legacy_claim_with_neither_side_identified_is_foreign(self): + """No identities anywhere: a different session id is simply not ours.""" + claim = _claim(1, session_id="someone-else", instance=None) + self.assertEqual( + classify_claim_ownership( + claim, session_id="sess-mine", controller_instance_id=None + ), + OWNERSHIP_FOREIGN, + ) + + def test_claim_identified_but_local_undeclared_is_unknown(self): + """Only one side identified: not comparable, so never adopt.""" + claim = _claim(1, session_id="someone-else", instance=THEIRS) + self.assertEqual( + classify_claim_ownership( + claim, session_id="sess-mine", controller_instance_id=None + ), + OWNERSHIP_UNKNOWN, + ) + + def test_local_identified_but_claim_undeclared_is_unknown(self): + """A legacy lease may be our own under an old session id; do not guess.""" + claim = _claim(1, session_id="someone-else", instance=None) + self.assertEqual( + classify_claim_ownership( + claim, session_id="sess-mine", controller_instance_id=MINE + ), + OWNERSHIP_UNKNOWN, + ) + + def test_resolve_controller_instance_id_reads_env(self): + self.assertEqual( + resolve_controller_instance_id({"GITEA_CONTROLLER_INSTANCE_ID": MINE}), + MINE, + ) + self.assertIsNone(resolve_controller_instance_id({})) + self.assertIsNone( + resolve_controller_instance_id({"GITEA_CONTROLLER_INSTANCE_ID": " "}) + ) + + +class TestForeignLeaseDoesNotBlockade(AllocatorOwnershipTestCase): + def test_foreign_claim_skipped_and_next_issue_selected(self): + """Skip the claimed issue, select the next unclaimed one.""" + candidates = [_issue(607), _issue(615), _issue(617)] + claims = { + ("issue", 607): _claim(607, session_id="sess-theirs", instance=THEIRS) + } + result = self._allocate(candidates, claims=claims) + + self.assertEqual(result["outcome"], OUTCOME_PREVIEW) + self.assertEqual(result["selected"]["number"], 615) + skipped_607 = [s for s in result["skipped"] if s["number"] == 607] + self.assertEqual(len(skipped_607), 1) + self.assertEqual( + skipped_607[0]["reason_code"], SKIP_CLAIMED_BY_OTHER_SESSION + ) + self.assertIn(SKIP_CLAIMED_BY_OTHER_SESSION, skipped_607[0]["reason"]) + + def test_claimed_candidate_appears_in_skipped_inventory(self): + """Skipped reporting must reflect claimed candidates.""" + candidates = [_issue(607), _issue(615)] + claims = { + ("issue", 607): _claim(607, session_id="sess-theirs", instance=THEIRS) + } + result = self._allocate(candidates, claims=claims) + self.assertEqual(len(result["skipped"]), 1) + self.assertEqual(len(result["claims_excluded"]), 1) + excluded = result["claims_excluded"][0] + self.assertEqual(excluded["number"], 607) + self.assertEqual(excluded["ownership"], OWNERSHIP_FOREIGN) + self.assertEqual(excluded["owner_controller_instance_id"], THEIRS) + + def test_task_local_blocker_does_not_freeze_unrelated_work(self): + """A quarantined task must not stop the rest of the queue.""" + candidates = [_issue(607), _issue(615), _issue(617)] + claims = { + ("issue", 607): _claim(607, session_id="sess-theirs", instance=THEIRS) + } + first = self._allocate(candidates, claims=claims) + self.assertEqual(first["selected"]["number"], 615) + + # 615 then gets claimed by yet another controller; queue still advances. + claims[("issue", 615)] = _claim( + 615, session_id="sess-third", instance="ctl-third-0003" + ) + second = self._allocate(candidates, claims=claims) + self.assertEqual(second["selected"]["number"], 617) + + def test_multiple_controllers_get_different_issues(self): + """Concurrent author sessions work on different issues.""" + candidates = [_issue(607), _issue(615)] + claims = { + ("issue", 607): _claim(607, session_id="sess-theirs", instance=THEIRS) + } + mine = self._allocate(candidates, claims=claims, instance=MINE) + theirs = self._allocate( + candidates, claims=claims, session_id="sess-theirs", instance=THEIRS + ) + self.assertEqual(mine["selected"]["number"], 615) + # The other controller may still be handed its own in-progress task. + self.assertEqual(theirs["selected"]["number"], 607) + + def test_unclaimed_queue_is_unaffected(self): + candidates = [_issue(607), _issue(615)] + result = self._allocate(candidates, claims={}) + self.assertEqual(result["selected"]["number"], 607) + self.assertEqual(result["skipped"], []) + self.assertEqual(result["claims_excluded"], []) + + +class TestOwnTaskResume(AllocatorOwnershipTestCase): + def test_controller_may_resume_its_own_active_task(self): + """Own claim stays selectable across a new session id.""" + candidates = [_issue(607), _issue(615)] + claims = { + ("issue", 607): _claim(607, session_id="sess-mine-old", instance=MINE) + } + result = self._allocate( + candidates, claims=claims, session_id="sess-mine-new", instance=MINE + ) + self.assertEqual(result["selected"]["number"], 607) + self.assertEqual(result["claims_excluded"], []) + + def test_own_claim_by_exact_session_is_selectable(self): + candidates = [_issue(607)] + claims = {("issue", 607): _claim(607, session_id="sess-mine", instance=None)} + result = self._allocate( + candidates, claims=claims, session_id="sess-mine", instance=None + ) + self.assertEqual(result["selected"]["number"], 607) + + +class TestAllCandidatesClaimed(AllocatorOwnershipTestCase): + def test_all_claimed_returns_wait_not_a_claimed_selection(self): + """Never hand back a claimed issue; report waiting instead.""" + candidates = [_issue(607), _issue(615)] + claims = { + ("issue", 607): _claim(607, session_id="sess-a", instance=THEIRS), + ("issue", 615): _claim(615, session_id="sess-b", instance="ctl-c-0003"), + } + result = self._allocate(candidates, claims=claims) + self.assertIsNone(result["selected"]) + self.assertEqual(result["outcome"], OUTCOME_WAIT) + self.assertEqual(len(result["claims_excluded"]), 2) + + def test_unidentifiable_owner_reports_ownership_defect(self): + """Refuse to adopt when ownership cannot be established.""" + candidates = [_issue(607)] + claims = {("issue", 607): _claim(607, session_id="sess-legacy", instance=None)} + result = self._allocate(candidates, claims=claims) + self.assertIsNone(result["selected"]) + self.assertEqual(result["outcome"], OUTCOME_OWNERSHIP_DEFECT) + self.assertEqual(len(result["ownership_defects"]), 1) + self.assertEqual( + result["ownership_defects"][0]["ownership"], OWNERSHIP_UNKNOWN + ) + + +class TestClaimsFromControlPlaneDb(AllocatorOwnershipTestCase): + """End-to-end against the real substrate, not injected claim dicts.""" + + def _seed_lease(self, number: int, *, session_id: str, instance: str | None): + self.db.upsert_session( + session_id=session_id, + role="author", + profile="prgs-author", + pid=4242, + controller_instance_id=instance, + ) + return self.db.assign_and_lease( + session_id=session_id, + role="author", + remote=REMOTE, + org=ORG, + repo=REPO, + kind="issue", + number=number, + ) + + def test_controller_instance_id_persists_on_session(self): + row = self.db.upsert_session( + session_id="sess-x", + role="author", + profile="prgs-author", + pid=1, + controller_instance_id=MINE, + ) + self.assertEqual(row["controller_instance_id"], MINE) + + def test_heartbeat_without_instance_does_not_erase_ownership(self): + self.db.upsert_session( + session_id="sess-x", + role="author", + profile="prgs-author", + pid=1, + controller_instance_id=MINE, + ) + row = self.db.upsert_session( + session_id="sess-x", role="author", profile="prgs-author", pid=1 + ) + self.assertEqual(row["controller_instance_id"], MINE) + + def test_list_active_claims_surfaces_owner_instance(self): + self._seed_lease(607, session_id="sess-theirs", instance=THEIRS) + claims = self.db.list_active_claims(remote=REMOTE, org=ORG, repo=REPO) + self.assertIn(("issue", 607), claims) + self.assertEqual(claims[("issue", 607)]["controller_instance_id"], THEIRS) + + def test_live_foreign_lease_is_excluded_without_injected_claims(self): + self._seed_lease(607, session_id="sess-theirs", instance=THEIRS) + result = allocate_next_work( + self.db, + session_id="sess-mine", + role="author", + remote=REMOTE, + org=ORG, + repo=REPO, + candidates=[_issue(607), _issue(615)], + apply=False, + profile_name="prgs-author", + controller_instance_id=MINE, + ) + self.assertEqual(result["selected"]["number"], 615) + self.assertEqual( + result["skipped"][0]["reason_code"], SKIP_CLAIMED_BY_OTHER_SESSION + ) + + def test_apply_reserves_the_unclaimed_issue(self): + self._seed_lease(607, session_id="sess-theirs", instance=THEIRS) + result = allocate_next_work( + self.db, + session_id="sess-mine", + role="author", + remote=REMOTE, + org=ORG, + repo=REPO, + candidates=[_issue(607), _issue(615)], + apply=True, + profile_name="prgs-author", + controller_instance_id=MINE, + ) + self.assertEqual(result["outcome"], "assigned_work") + self.assertEqual(result["selected"]["number"], 615) + self.assertEqual(result["assignment"]["work_number"], 615) + + +if __name__ == "__main__": + unittest.main() diff --git a/workflow_dashboard.py b/workflow_dashboard.py index 8b63449..d2b10ee 100644 --- a/workflow_dashboard.py +++ b/workflow_dashboard.py @@ -14,7 +14,7 @@ Design rules: from __future__ import annotations from dataclasses import dataclass, field -from typing import Any, Iterable, Sequence +from typing import Any, Iterable, Mapping, Sequence from allocator_service import ( ROLE_AUTHOR, @@ -22,7 +22,11 @@ from allocator_service import ( ROLE_MERGER, ROLE_RECONCILER, ROLE_REVIEWER, + OWNERSHIP_FOREIGN, + OWNERSHIP_UNKNOWN, + SKIP_CLAIMED_BY_OTHER_SESSION, WorkCandidate, + classify_claim_ownership, classify_skip, expected_role_for_candidate, sort_candidates, @@ -265,9 +269,13 @@ def _entry_for_candidate( c: WorkCandidate, *, terminal_pr: int | None, + claim_ownership: str | None = None, ) -> QueueEntry: expected = expected_role_for_candidate(c) badges = _badges_for_candidate(c, terminal_pr=terminal_pr) + claimed_by_other = claim_ownership in (OWNERSHIP_FOREIGN, OWNERSHIP_UNKNOWN) + if claimed_by_other and "claimed" not in badges: + badges = tuple(list(badges) + ["claimed"]) safe_roles: list[str] = [] block_reason: str | None = None @@ -276,6 +284,12 @@ def _entry_for_candidate( block_reason = "status blocked" elif c.dependency_unmet: block_reason = c.dependency_reason or "unmet dependency" + elif claimed_by_other: + # #765: never advertise another controller's active task as safe work. + block_reason = ( + f"{SKIP_CLAIMED_BY_OTHER_SESSION}: active lease held by another " + "controller" + ) elif c.already_claimed_elsewhere: block_reason = "already claimed elsewhere" elif c.kind == "pr" and not (c.head_sha or "").strip(): @@ -291,7 +305,12 @@ def _entry_for_candidate( if block_reason is None: # Safe only for the expected role, and only when classify_skip agrees. - skip = classify_skip(c, role=expected, terminal_pr=terminal_pr) + skip = classify_skip( + c, + role=expected, + terminal_pr=terminal_pr, + claim_ownership=claim_ownership, + ) if skip is None: safe_roles.append(expected) else: @@ -450,8 +469,16 @@ def build_workflow_dashboard( terminal_lock: dict[str, Any] | None = None, inventory_complete: bool = True, inventory_reasons: Sequence[str] | None = None, + claims: Mapping[tuple[str, int], dict[str, Any]] | None = None, + session_id: str | None = None, + controller_instance_id: str | None = None, ) -> DashboardSnapshot: - """Build a full dashboard snapshot from injected inventory (pure).""" + """Build a full dashboard snapshot from injected inventory (pure). + + *claims* (#765) maps ``(kind, number)`` to the live lease holding that work + item. Items claimed by a different controller are never presented as safe + next work for this one. + """ reasons = [str(r) for r in (inventory_reasons or ()) if str(r).strip()] if not inventory_complete: reasons.append( @@ -461,7 +488,16 @@ def build_workflow_dashboard( ranked = sort_candidates(list(candidates)) entries = [ - _entry_for_candidate(c, terminal_pr=terminal_pr) for c in ranked + _entry_for_candidate( + c, + terminal_pr=terminal_pr, + claim_ownership=classify_claim_ownership( + (claims or {}).get((c.kind, int(c.number))), + session_id=session_id, + controller_instance_id=controller_instance_id, + ), + ) + for c in ranked ] open_prs = [e for e in entries if e.kind == "pr"]