diff --git a/allocator_service.py b/allocator_service.py index d6d6a62..40787ce 100644 --- a/allocator_service.py +++ b/allocator_service.py @@ -18,6 +18,8 @@ after they exist as normal issues; this module never assigns incidents. from __future__ import annotations +import hashlib +import json import os import uuid from dataclasses import dataclass, field @@ -42,9 +44,15 @@ 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" +# #776: excluded issue still carries a live same-owner lease — resume or release. +OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE = "blocked_by_excluded_own_lease" +# #776: dry-run/apply candidate-set fingerprint mismatch (CAS drift). +OUTCOME_CANDIDATE_SET_DRIFT = "candidate_set_drift" # #765 skip reason code for work already claimed by a different controller. SKIP_CLAIMED_BY_OTHER_SESSION = "claimed_by_other_session" +# #776: controller-supplied pre-rank exclusion. +SKIP_EXCLUDED_BY_CONTROLLER = "excluded_by_controller" # Ownership verdicts for a live claim on a candidate (#765). OWNERSHIP_OWN = "own" @@ -369,6 +377,113 @@ def sort_candidates(candidates: Sequence[WorkCandidate]) -> list[WorkCandidate]: ) +def _require_strict_int(value: Any, *, field: str) -> int: + """Parse an issue number; reject bools and non-integers (#776).""" + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError( + f"{field} must be an integer (booleans and non-integers rejected; " + f"got {type(value).__name__})" + ) + return int(value) + + +def normalize_exclude_issue_numbers( + exclude_issue_numbers: Any = None, +) -> list[int]: + """Normalize controller-supplied exclusions to a sorted unique int list (#776). + + ``None`` / omitted → empty list (existing behavior). Accepts a list/tuple of + integers. Rejects scalars, bools-as-ints, nested structures, and strings. + """ + if exclude_issue_numbers is None: + return [] + if isinstance(exclude_issue_numbers, (str, bytes)) or not isinstance( + exclude_issue_numbers, (list, tuple) + ): + raise ValueError( + "exclude_issue_numbers must be a list of integers " + f"(got {type(exclude_issue_numbers).__name__})" + ) + out: list[int] = [] + seen: set[int] = set() + for idx, raw in enumerate(exclude_issue_numbers): + num = _require_strict_int(raw, field=f"exclude_issue_numbers[{idx}]") + if num not in seen: + seen.add(num) + out.append(num) + return sorted(out) + + +def candidate_set_fingerprint( + candidates: Sequence[WorkCandidate], + *, + exclude_issue_numbers: Sequence[int] | None = None, +) -> str: + """Stable CAS fingerprint of normalized candidate set + exclusions (#776 AC4).""" + payload = { + "candidates": sorted( + ({"kind": c.kind, "number": int(c.number)} for c in candidates), + key=lambda x: (x["kind"], x["number"]), + ), + "exclude_issue_numbers": list( + normalize_exclude_issue_numbers(exclude_issue_numbers) + ), + } + blob = json.dumps(payload, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(blob.encode("utf-8")).hexdigest() + + +def normalize_candidates_payload(raw: Any) -> list[WorkCandidate]: + """Decode MCP ``candidates_json`` from list or JSON string (#776 AC3). + + Accepts: + * an already-decoded ``list`` of candidate dicts (native MCP transport); + * a valid JSON string that decodes to such a list (backward compatible). + + Rejects malformed JSON, scalars, non-list containers, invalid records, + booleans-as-integers, and unsupported types with fail-closed ``ValueError``. + """ + if raw is None: + raise ValueError("candidates_json is empty") + if isinstance(raw, (bytes, bytearray)): + try: + raw = raw.decode("utf-8") + except Exception as exc: # noqa: BLE001 + raise ValueError( + f"candidates_json bytes are not valid utf-8: {exc}" + ) from exc + if isinstance(raw, str): + text = raw.strip() + if not text: + raise ValueError("candidates_json string is empty") + try: + decoded = json.loads(text) + except json.JSONDecodeError as exc: + raise ValueError( + f"malformed candidates_json JSON: {exc.msg} at pos {exc.pos}" + ) from exc + raw = decoded + if not isinstance(raw, list): + raise ValueError( + "candidates_json must be a JSON list (or already-decoded list); " + f"got {type(raw).__name__}" + ) + candidates: list[WorkCandidate] = [] + for idx, item in enumerate(raw): + if not isinstance(item, dict): + raise ValueError( + f"invalid candidate record at index {idx}: expected object, " + f"got {type(item).__name__}" + ) + try: + candidates.append(candidate_from_dict(item)) + except (KeyError, TypeError, ValueError, InvalidWorkKindError) as exc: + raise ValueError( + f"invalid candidate record at index {idx}: {exc}" + ) from exc + return candidates + + def allocate_next_work( db: ControlPlaneDB, *, @@ -384,12 +499,20 @@ def allocate_next_work( lease_ttl_seconds: int | None = None, controller_instance_id: str | None = None, claims: Mapping[tuple[str, int], dict[str, Any]] | None = None, + exclude_issue_numbers: Sequence[int] | None = None, + expected_candidate_set_fingerprint: str | 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. + *exclude_issue_numbers* (#776): numbers removed before ranking. Omitted / + empty preserves prior behavior. + + *expected_candidate_set_fingerprint* (#776 AC4): when set on apply, rejects + material candidate-set drift vs a prior dry-run. + Never uses file locks or comment-only leases as the assignment source. """ if db is None: @@ -469,6 +592,9 @@ def allocate_next_work( # #765: live claims exclude work owned by a *different* controller before # ranking, so one session's in-progress task cannot blockade the queue. + # #776 AC5: load live claims in this call path immediately before selection + # (and before apply reserve) so ownership is never stale within the + # allocation attempt. Test callers may inject *claims* explicitly. if claims is None: try: claims = db.list_active_claims(remote=remote, org=org, repo=repo) @@ -484,10 +610,131 @@ def allocate_next_work( "substrate": "control_plane_db", } + try: + exclude_nums = normalize_exclude_issue_numbers(exclude_issue_numbers) + except ValueError as exc: + return { + "success": False, + "outcome": OUTCOME_NO_SAFE, + "apply": bool(apply), + "reasons": [ + f"invalid exclude_issue_numbers: {exc} (fail closed, #776)" + ], + "skipped": [], + "assignment": None, + "substrate": "control_plane_db", + } + exclude_set = set(exclude_nums) + cas_fp = candidate_set_fingerprint( + candidates, exclude_issue_numbers=exclude_nums + ) + expected_fp = (expected_candidate_set_fingerprint or "").strip() or None + if expected_fp and expected_fp != cas_fp: + return { + "success": False, + "outcome": OUTCOME_CANDIDATE_SET_DRIFT, + "apply": bool(apply), + "reasons": [ + "candidate-set fingerprint drift: apply rejected rather than " + "silently leasing a different candidate (#776 AC4)" + ], + "candidate_set_fingerprint": cas_fp, + "expected_candidate_set_fingerprint": expected_fp, + "exclude_issue_numbers": list(exclude_nums), + "skipped": [], + "assignment": None, + "substrate": "control_plane_db", + "file_lock_only": False, + "comment_lease_only": False, + } + skipped: list[SkipRecord] = [] claims_excluded: list[dict[str, Any]] = [] ownership_defects: list[dict[str, Any]] = [] - ordered = sort_candidates(list(candidates)) + controller_excluded: list[dict[str, Any]] = [] + + # #776 AC2: remove excluded numbers *before* ranking / selection / lease. + rankable: list[WorkCandidate] = [] + for c in candidates: + if int(c.number) in exclude_set: + reason = ( + f"{c.kind}#{c.number} {SKIP_EXCLUDED_BY_CONTROLLER}: " + "controller pre-rank exclusion" + ) + skipped.append( + SkipRecord( + c.kind, + c.number, + reason, + SKIP_EXCLUDED_BY_CONTROLLER, + ) + ) + controller_excluded.append( + { + "kind": c.kind, + "number": c.number, + "reason_code": SKIP_EXCLUDED_BY_CONTROLLER, + } + ) + # #776 AC5: same-owner live lease on an excluded issue is a + # structured resume/release blocker, never a silent strand. + 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, + ) + if ownership == OWNERSHIP_OWN and claim: + return { + "success": True, + "outcome": OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE, + "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": [ + f"{c.kind}#{c.number} is excluded_by_controller but " + "carries a live same-owner lease; resume or release " + "that lease before allocating other work (#776 AC5)" + ], + "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, + "controller_instance_id": controller_instance_id, + "claims_excluded": list(claims_excluded), + "ownership_defects": list(ownership_defects), + "controller_excluded": list(controller_excluded), + "exclude_issue_numbers": list(exclude_nums), + "candidate_set_fingerprint": cas_fp, + "blocked_lease": { + "kind": c.kind, + "number": c.number, + "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"), + "safe_next_action": ( + "resume the same-owner lease or release it, then " + "re-run allocation without stranding the excluded " + "issue" + ), + }, + } + continue + rankable.append(c) + + ordered = sort_candidates(rankable) selected: WorkCandidate | None = None for c in ordered: claim = claims.get((c.kind, int(c.number))) if claims else None @@ -560,11 +807,21 @@ def allocate_next_work( ] # Preserve the pre-#765 wait contract: name the blocking owner. owner_session_id = claims_excluded[0].get("owner_session_id") + elif controller_excluded and not ordered: + # #776 AC7: every candidate was controller-excluded → wait / no lease. + outcome = OUTCOME_WAIT + reasons = [ + f"no assignable work for role '{role_norm}': all " + f"{len(controller_excluded)} candidate(s) were removed by " + f"{SKIP_EXCLUDED_BY_CONTROLLER} before ranking; no assignment " + "or lease created (#776 AC7)" + ] else: outcome = OUTCOME_NO_SAFE reasons = [ f"no safe assignable work for role '{role_norm}' " - f"among {len(ordered)} candidates" + f"among {len(ordered)} rankable candidates " + f"({len(controller_excluded)} controller-excluded)" ] return { "success": True, @@ -589,6 +846,9 @@ def allocate_next_work( "controller_instance_id": controller_instance_id, "claims_excluded": list(claims_excluded), "ownership_defects": list(ownership_defects), + "controller_excluded": list(controller_excluded), + "exclude_issue_numbers": list(exclude_nums), + "candidate_set_fingerprint": cas_fp, "owner_session_id": owner_session_id, "downstream_note": ( "#612 incident bridge remains downstream of #600; " @@ -639,6 +899,9 @@ def allocate_next_work( "controller_instance_id": controller_instance_id, "claims_excluded": list(claims_excluded), "ownership_defects": list(ownership_defects), + "controller_excluded": list(controller_excluded), + "exclude_issue_numbers": list(exclude_nums), + "candidate_set_fingerprint": cas_fp, "downstream_note": ( "#612 incident bridge remains downstream of #600; " "allocator never assigns raw monitoring incidents" @@ -765,6 +1028,9 @@ def allocate_next_work( "controller_instance_id": controller_instance_id, "claims_excluded": list(claims_excluded), "ownership_defects": list(ownership_defects), + "controller_excluded": list(controller_excluded), + "exclude_issue_numbers": list(exclude_nums), + "candidate_set_fingerprint": cas_fp, "downstream_note": ( "#612 incident bridge remains downstream of #600; " "allocator never assigns raw monitoring incidents" @@ -796,14 +1062,32 @@ def _next_command(role: str, c: WorkCandidate) -> str: def candidate_from_dict(data: dict[str, Any]) -> WorkCandidate: - """Build a WorkCandidate from a plain dict (tests / MCP inventory).""" + """Build a WorkCandidate from a plain dict (tests / MCP inventory). + + #776: reject booleans-as-integers and non-int numbers fail-closed. + """ + if "number" not in data: + raise KeyError("number") + number = _require_strict_int(data["number"], field="number") + priority_raw = data.get("priority") or 0 + if isinstance(priority_raw, bool) or not isinstance(priority_raw, (int, float)): + # Allow numeric strings only for priority? Keep strict for bools. + if isinstance(priority_raw, str) and priority_raw.strip().lstrip("-").isdigit(): + priority = int(priority_raw) + else: + raise ValueError( + f"priority must be numeric (booleans rejected; " + f"got {type(priority_raw).__name__})" + ) + else: + priority = int(priority_raw) return WorkCandidate( kind=str(data.get("kind") or "issue"), - number=int(data["number"]), + number=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), + priority=priority, 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")), diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 65f7413..5e2fc7e 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -19202,7 +19202,9 @@ def gitea_allocate_next_work( repo: str | None = None, include_issues: bool = True, include_prs: bool = True, - candidates_json: str | None = None, + candidates_json: Any = None, + exclude_issue_numbers: list[int] | None = None, + expected_candidate_set_fingerprint: str | None = None, limit: int = 50, ) -> dict: """Controller-owned next-work allocator using the #613 control-plane DB (#600). @@ -19216,10 +19218,18 @@ def gitea_allocate_next_work( leases as the coordination source). Outcomes include: ``assigned_work``, ``preview``, ``wait``, - ``blocked_by_terminal_path``, ``no_safe_work``, ``role_ineligible``. + ``blocked_by_terminal_path``, ``no_safe_work``, ``role_ineligible``, + ``blocked_by_excluded_own_lease``, ``candidate_set_drift``. - *candidates_json* may inject a JSON list of candidate dicts (tests / - controller overrides). When omitted, open issues/PRs are loaded from Gitea. + *candidates_json* may inject a candidate list as an already-decoded list + (native MCP transport) or a JSON string (backward compatible). When + omitted, open issues/PRs are loaded from Gitea. + + *exclude_issue_numbers* (#776): typed list of issue/PR numbers removed + before ranking. Omitted/empty retains prior behavior. + + *expected_candidate_set_fingerprint* (#776 AC4): optional CAS pin from a + prior dry-run; apply fails closed on material candidate-set drift. #612 remains downstream: raw monitoring incidents are never candidates. """ @@ -19273,15 +19283,12 @@ def gitea_allocate_next_work( inv_reasons: list[str] = [] candidates: list[Any] = [] - if candidates_json: + if candidates_json is not None and 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)) + # #776 AC3: accept decoded list or JSON string; reject malformed. + candidates = allocator_service.normalize_candidates_payload( + candidates_json + ) except Exception as exc: # noqa: BLE001 return { "success": False, @@ -19324,24 +19331,40 @@ def gitea_allocate_next_work( 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, - controller_instance_id=allocator_service.resolve_controller_instance_id(), - ) + try: + 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, + controller_instance_id=allocator_service.resolve_controller_instance_id(), + exclude_issue_numbers=exclude_issue_numbers, + expected_candidate_set_fingerprint=expected_candidate_set_fingerprint, + ) + except ValueError as exc: + return { + "success": False, + "outcome": allocator_service.OUTCOME_NO_SAFE, + "apply": bool(apply), + "reasons": [ + f"invalid allocator input: {_redact(str(exc))} (fail closed, #776)" + ], + "assignment": None, + "substrate": "control_plane_db", + } 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" + "candidates_json" + if candidates_json is not None and candidates_json != "" + else "gitea_live" ) result["inventory_complete"] = True result["selection_policy"] = allocator_service.SELECTION_POLICY @@ -19435,7 +19458,7 @@ def gitea_workflow_dashboard( include_issues: bool = True, include_prs: bool = True, include_non_active_leases: bool = True, - candidates_json: str | None = None, + candidates_json: Any = None, limit: int = 100, ) -> dict: """Read-only MCP workflow dashboard: queues, leases, blockers, next safe action (#605). @@ -19444,9 +19467,10 @@ def gitea_workflow_dashboard( still requires ``gitea_allocate_next_work``. Never presents blocked or terminal-locked items as safe next work. - *candidates_json* may inject a JSON list of candidate dicts (tests / - offline fixtures). When omitted, open issues/PRs are loaded from Gitea. - Incomplete live inventory fails closed (no safe suggestions). + *candidates_json* may inject a candidate list as an already-decoded list + or JSON string (#776 transport parity). When omitted, open issues/PRs are + loaded from Gitea. Incomplete live inventory fails closed (no safe + suggestions). """ read_block = _profile_operation_gate("gitea.read") if read_block: @@ -19471,15 +19495,11 @@ def gitea_workflow_dashboard( inv_reasons: list[str] = [] candidates: list[Any] = [] inventory_complete = True - if candidates_json: + if candidates_json is not None and 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)) + candidates = allocator_service.normalize_candidates_payload( + candidates_json + ) except Exception as exc: # noqa: BLE001 return { "success": False, @@ -19570,7 +19590,9 @@ def gitea_workflow_dashboard( ) payload = snap.as_dict() payload["inventory_source"] = ( - "candidates_json" if candidates_json else "gitea_live" + "candidates_json" + if candidates_json is not None and candidates_json != "" + else "gitea_live" ) payload["limit_applies_to"] = "lease_listing_only" return payload diff --git a/tests/test_allocator_pre_rank_exclusions.py b/tests/test_allocator_pre_rank_exclusions.py new file mode 100644 index 0000000..64d5d60 --- /dev/null +++ b/tests/test_allocator_pre_rank_exclusions.py @@ -0,0 +1,349 @@ +"""Allocator pre-rank exclusions and candidates_json transport (#776). + +Covers: +* #617 excluded before ranking (never leased when exclude_issue_numbers=[617]); +* excluded top candidate selects the next safe candidate; +* all candidates excluded → WAIT, no lease; +* decoded-list and JSON-string candidates_json; +* malformed / type-invalid fail-closed cases; +* dry-run/apply fingerprint match and drift rejection; +* foreign lease and same-owner lease on excluded issue; +* skipped-accounting reason parity (excluded_by_controller); +* public MCP entry-point coverage for exclude_issue_numbers. +""" + +from __future__ import annotations + +import json +import os +import tempfile +import unittest +from unittest.mock import patch + +import gitea_mcp_server as srv +from allocator_service import ( + OUTCOME_ASSIGNED, + OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE, + OUTCOME_CANDIDATE_SET_DRIFT, + OUTCOME_NO_SAFE, + OUTCOME_PREVIEW, + OUTCOME_WAIT, + SKIP_CLAIMED_BY_OTHER_SESSION, + SKIP_EXCLUDED_BY_CONTROLLER, + WorkCandidate, + allocate_next_work, + candidate_from_dict, + candidate_set_fingerprint, + normalize_candidates_payload, + normalize_exclude_issue_numbers, +) +from control_plane_db import ControlPlaneDB + +REMOTE = "prgs" +ORG = "Scaled-Tech-Consulting" +REPO = "Gitea-Tools" +MINE = "ctl-mine-776" +THEIRS = "ctl-theirs-776" + + +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-21T12:00:00Z", + "work_kind": kind, + "work_number": number, + } + + +def _cand_dict(number: int, **kwargs) -> dict: + d = { + "kind": "issue", + "number": number, + "state": "open", + "labels": ["status:ready", "type:bug"], + "title": f"issue {number}", + "priority": 20, + } + d.update(kwargs) + return d + + +class AllocatorExcludeServiceTest(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3")) + + def _alloc(self, candidates, **kwargs): + defaults = dict( + session_id="sess-776", + role="author", + remote=REMOTE, + org=ORG, + repo=REPO, + profile_name="prgs-author", + controller_instance_id=MINE, + claims={}, + apply=False, + ) + defaults.update(kwargs) + return allocate_next_work(self.db, candidates=candidates, **defaults) + + def test_exclude_617_before_ranking_never_selects(self) -> None: + """AC2/AC8: highest-ranked #617 is removed before ranking.""" + cands = [_issue(617), _issue(700), _issue(701)] + res = self._alloc(cands, exclude_issue_numbers=[617]) + self.assertEqual(res["outcome"], OUTCOME_PREVIEW) + self.assertEqual(res["selected"]["number"], 700) + skipped = {s["number"]: s for s in res["skipped"]} + self.assertIn(617, skipped) + self.assertEqual( + skipped[617]["reason_code"], SKIP_EXCLUDED_BY_CONTROLLER + ) + self.assertIn(SKIP_EXCLUDED_BY_CONTROLLER, skipped[617]["reason"]) + + def test_excluded_top_selects_next_safe(self) -> None: + """AC2: excluding the oldest ready issue promotes the next number.""" + cands = [_issue(600), _issue(601), _issue(602)] + res = self._alloc(cands, exclude_issue_numbers=[600]) + self.assertEqual(res["selected"]["number"], 601) + + def test_all_candidates_excluded_wait_no_lease(self) -> None: + """AC7: every candidate excluded → WAIT, no assignment.""" + cands = [_issue(617), _issue(700)] + res = self._alloc(cands, exclude_issue_numbers=[617, 700], apply=True) + self.assertEqual(res["outcome"], OUTCOME_WAIT) + self.assertIsNone(res["selected"]) + self.assertIsNone(res["assignment"]) + self.assertEqual(len(res["controller_excluded"]), 2) + + def test_omit_exclude_retains_existing_behavior(self) -> None: + """AC1/AC9: omit exclusions → #617 still wins when oldest ready.""" + cands = [_issue(617), _issue(700)] + res = self._alloc(cands) + self.assertEqual(res["selected"]["number"], 617) + self.assertEqual(res.get("exclude_issue_numbers"), []) + + def test_foreign_lease_still_skipped(self) -> None: + """AC6/AC9: foreign claims keep SKIP_CLAIMED_BY_OTHER_SESSION.""" + cands = [_issue(617), _issue(700)] + claims = { + ("issue", 700): _claim(700, session_id="other", instance=THEIRS), + } + res = self._alloc( + cands, exclude_issue_numbers=[617], claims=claims + ) + # 617 excluded, 700 foreign → wait, no selection + self.assertEqual(res["outcome"], OUTCOME_WAIT) + self.assertIsNone(res["selected"]) + codes = {s["reason_code"] for s in res["skipped"]} + self.assertIn(SKIP_EXCLUDED_BY_CONTROLLER, codes) + self.assertIn(SKIP_CLAIMED_BY_OTHER_SESSION, codes) + + def test_same_owner_lease_on_excluded_blocks_resume_release(self) -> None: + """AC5: excluded + live same-owner lease → structured blocker.""" + cands = [_issue(617), _issue(700)] + claims = { + ("issue", 617): _claim(617, session_id="sess-776", instance=MINE), + } + res = self._alloc( + cands, exclude_issue_numbers=[617], claims=claims, apply=True + ) + self.assertEqual(res["outcome"], OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE) + self.assertIsNone(res["assignment"]) + self.assertEqual(res["blocked_lease"]["number"], 617) + self.assertIn("resume", res["blocked_lease"]["safe_next_action"]) + + def test_dry_run_apply_fingerprint_match(self) -> None: + """AC4: dry-run and apply share the same fingerprint.""" + cands = [_issue(617), _issue(700)] + dry = self._alloc(cands, exclude_issue_numbers=[617], apply=False) + apply_res = self._alloc( + cands, + exclude_issue_numbers=[617], + apply=True, + expected_candidate_set_fingerprint=dry["candidate_set_fingerprint"], + ) + self.assertEqual( + dry["candidate_set_fingerprint"], + apply_res["candidate_set_fingerprint"], + ) + self.assertEqual(apply_res["outcome"], OUTCOME_ASSIGNED) + self.assertEqual(apply_res["selected"]["number"], 700) + + def test_apply_rejects_fingerprint_drift(self) -> None: + """AC4: material candidate-set drift fails closed on apply.""" + cands = [_issue(617), _issue(700)] + res = self._alloc( + cands, + exclude_issue_numbers=[617], + apply=True, + expected_candidate_set_fingerprint="0" * 64, + ) + self.assertFalse(res["success"]) + self.assertEqual(res["outcome"], OUTCOME_CANDIDATE_SET_DRIFT) + self.assertIsNone(res["assignment"]) + + def test_fingerprint_stable_helper(self) -> None: + cands = [_issue(700), _issue(617)] + a = candidate_set_fingerprint(cands, exclude_issue_numbers=[617]) + b = candidate_set_fingerprint( + list(reversed(cands)), exclude_issue_numbers=[617] + ) + self.assertEqual(a, b) + + def test_normalize_exclude_rejects_bool(self) -> None: + with self.assertRaises(ValueError): + normalize_exclude_issue_numbers([True]) + + def test_normalize_exclude_rejects_scalar(self) -> None: + with self.assertRaises(ValueError): + normalize_exclude_issue_numbers(617) + + +class CandidatesJsonNormalizeTest(unittest.TestCase): + def test_decoded_list(self) -> None: + """AC3: already-decoded list from MCP transport.""" + cands = normalize_candidates_payload([_cand_dict(617), _cand_dict(700)]) + self.assertEqual([c.number for c in cands], [617, 700]) + + def test_json_string(self) -> None: + """AC3: backward-compatible JSON string.""" + raw = json.dumps([_cand_dict(617)]) + cands = normalize_candidates_payload(raw) + self.assertEqual(cands[0].number, 617) + + def test_malformed_json_fail_closed(self) -> None: + with self.assertRaises(ValueError) as ctx: + normalize_candidates_payload("{not json") + self.assertIn("malformed", str(ctx.exception).lower()) + + def test_scalar_fail_closed(self) -> None: + with self.assertRaises(ValueError): + normalize_candidates_payload(42) + + def test_bool_number_fail_closed(self) -> None: + with self.assertRaises(ValueError): + normalize_candidates_payload([_cand_dict(True)]) # type: ignore[arg-type] + + def test_invalid_record_fail_closed(self) -> None: + with self.assertRaises(ValueError): + normalize_candidates_payload(["not-a-dict"]) + + def test_object_not_list_fail_closed(self) -> None: + with self.assertRaises(ValueError): + normalize_candidates_payload(json.dumps({"number": 1})) + + def test_candidate_from_dict_rejects_bool_number(self) -> None: + with self.assertRaises(ValueError): + candidate_from_dict({"kind": "issue", "number": True}) + + +class AllocateNextWorkMcpExcludeTest(unittest.TestCase): + """Public MCP entry-point coverage (#776 AC8).""" + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3")) + + def tearDown(self) -> None: + self._tmp.cleanup() + + def _call(self, **kwargs): + with patch("gitea_mcp_server._profile_operation_gate", return_value=None), patch( + "gitea_mcp_server._resolve", return_value=("h", ORG, REPO) + ), patch( + "gitea_mcp_server.get_profile", + return_value={"profile_name": "prgs-author", "role": "author"}, + ), patch( + "gitea_mcp_server._authenticated_username", return_value="jcwalker3" + ), patch( + "gitea_mcp_server._control_plane_db_or_error", return_value=(self.db, []) + ), patch( + "gitea_mcp_server.sentry_observability.monitor_checkin", return_value=None + ): + return srv.gitea_allocate_next_work( + remote="prgs", org=ORG, repo=REPO, role="author", **kwargs + ) + + def test_mcp_exclude_617_decoded_list_never_selects(self) -> None: + """AC8: public tool with decoded list + exclude_issue_numbers=[617].""" + candidates = [_cand_dict(617), _cand_dict(700)] + res = self._call( + candidates_json=candidates, + exclude_issue_numbers=[617], + apply=False, + ) + self.assertTrue(res.get("success"), res) + self.assertEqual(res["selected"]["number"], 700) + skipped = {s["number"]: s for s in res["skipped"]} + self.assertEqual( + skipped[617]["reason_code"], SKIP_EXCLUDED_BY_CONTROLLER + ) + self.assertNotEqual(res["selected"]["number"], 617) + + def test_mcp_exclude_617_json_string_apply(self) -> None: + """AC8: JSON-string transport + apply never leases #617.""" + raw = json.dumps([_cand_dict(617), _cand_dict(700)]) + res = self._call( + candidates_json=raw, + exclude_issue_numbers=[617], + apply=True, + ) + self.assertEqual(res["outcome"], OUTCOME_ASSIGNED) + self.assertEqual(res["assignment"]["work_number"], 700) + self.assertNotEqual(res["selected"]["number"], 617) + + def test_mcp_malformed_candidates_json_fail_closed(self) -> None: + res = self._call(candidates_json="{bad", apply=False) + self.assertFalse(res["success"]) + self.assertIsNone(res["assignment"]) + self.assertTrue(any("fail closed" in r for r in res["reasons"])) + + def test_mcp_bool_number_fail_closed(self) -> None: + res = self._call( + candidates_json=[{"kind": "issue", "number": True, "priority": 20}], + apply=False, + ) + self.assertFalse(res["success"]) + self.assertIsNone(res["assignment"]) + + def test_mcp_fingerprint_dry_run_apply_parity(self) -> None: + candidates = [_cand_dict(617), _cand_dict(700)] + dry = self._call( + candidates_json=candidates, + exclude_issue_numbers=[617], + apply=False, + ) + apply_res = self._call( + candidates_json=candidates, + exclude_issue_numbers=[617], + apply=True, + expected_candidate_set_fingerprint=dry["candidate_set_fingerprint"], + ) + self.assertEqual( + dry["candidate_set_fingerprint"], + apply_res["candidate_set_fingerprint"], + ) + self.assertEqual(apply_res["selected"]["number"], 700) + + +if __name__ == "__main__": + unittest.main()