fix: rank complete allocator inventory and resolve declared dependencies (Closes #758)

The author allocator could select an ineligible issue for two independent
reasons, both fixed here.

Defect 1 — candidate truncation before ranking. The loader fetched the
complete open inventory via api_get_all, then sliced it to `limit` items
before constructing any WorkCandidate. Because every status:ready issue
ties at priority 20 and the tie breaks on lowest number, the slice — not
the ranking — decided the winner, so the same query returned different
answers at different `limit` values. Candidate construction now consumes
the full listing; `limit` bounds the reported skip list only, and any
such truncation is reported explicitly rather than silently.

Defect 2 — dependency state inferred from body substrings. Only
"blocked on #" and "downstream of #" were recognized, so the repository's
canonical `Depends: #N, #N` field never set dependency_unmet and
dependency-blocked issues were emitted as eligible. The new
allocator_dependencies module parses the canonical declaration into
structured references and resolves each against live issue state: open
means unmet, closed means met, and unavailable evidence fails closed. The
complete open listing proves openness without extra calls; anything
absent from it is confirmed by a cached targeted lookup instead of being
assumed closed. The legacy "blocked on #" marker still parses.

A failed listing now marks the inventory incomplete and the tool fails
closed instead of ranking a partial set.

Selection already advanced past a skipped candidate; with dependencies
resolved correctly that fall-through now actually engages, and is covered
by a regression.

Tests: 35 new cases across parser, resolver, loader, and an MCP-level
gitea_allocate_next_work regression, including limit-invariant selection
over a 73-candidate inventory and a guard proving pre-ranking truncation
would change the winner. No issue number is special-cased.

gitea_mcp_server.py

@@ -1765,6 +1765,7 @@ 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 allocator_dependencies  # noqa: E402
 import control_plane_db  # noqa: E402
 import lease_lifecycle  # noqa: E402
 import incident_bridge  # noqa: E402
@@ -17794,16 +17795,24 @@ def _allocator_candidates_from_gitea(
     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."""
+) -> tuple[list[Any], list[str], bool]:
+    """Build allocator candidates from live Gitea open issues/PRs.
+
+    Returns ``(candidates, reasons, inventory_complete)``. The inventory is
+    assembled in full and is never truncated before ranking (#758 AC1/AC3):
+    any result/display bound belongs to the caller's reporting, not to
+    candidate construction. ``inventory_complete`` is False when a required
+    listing failed, so the caller can fail closed instead of ranking a
+    silently short candidate set.
+    """
     reasons: list[str] = []
     candidates: list[Any] = []
+    inventory_complete = True
     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))}"]
+        return [], [f"failed to resolve Gitea target: {_redact(str(exc))}"], False
 
     if include_prs:
         try:
@@ -17813,7 +17822,8 @@ def _allocator_candidates_from_gitea(
         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))]:
+            inventory_complete = False
+        for pr in prs:
             if not isinstance(pr, dict):
                 continue
             number = pr.get("number")
@@ -17885,7 +17895,39 @@ def _allocator_candidates_from_gitea(
                     f"failed to list open issues: {_redact(str(exc2))}"
                 )
                 issues = []
-        for issue in issues[: max(1, int(limit))]:
+                inventory_complete = False
+
+        # Live dependency evidence (#758 AC5). The complete open-issue listing
+        # already proves which references are still open; anything absent from
+        # it is confirmed by a targeted lookup rather than assumed closed, so a
+        # deleted or unreachable reference fails closed instead of passing.
+        open_issue_numbers = {
+            int(i["number"])
+            for i in issues
+            if isinstance(i, dict)
+            and i.get("number") is not None
+            and i.get("pull_request") is None
+        }
+        dep_state_cache: dict[int, str | None] = {}
+
+        def _issue_state(number: int) -> str | None:
+            if number in open_issue_numbers:
+                return allocator_dependencies.DEP_STATE_OPEN
+            if number in dep_state_cache:
+                return dep_state_cache[number]
+            state: str | None = None
+            try:
+                data = api_request(
+                    "GET", f"{repo_api_url(h, o, r)}/issues/{number}", auth
+                )
+                if isinstance(data, dict):
+                    state = str(data.get("state") or "").strip().lower() or None
+            except Exception:  # noqa: BLE001 — unavailable evidence fails closed
+                state = None
+            dep_state_cache[number] = state
+            return state
+
+        for issue in issues:
             if not isinstance(issue, dict):
                 continue
             # Pull requests also appear in /issues on Gitea — skip them.
@@ -17903,21 +17945,15 @@ def _allocator_candidates_from_gitea(
             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
-            # Generic body markers for blocked-on unfinished deps.
-            # (Hard-coded #612→#600 block removed after #600 merged — #612.)
-            lower_body = body.lower()
-            if "blocked on #" in lower_body or "downstream of #" in lower_body:
-                # Only treat as unmet when body still marks a live open dependency
-                # pattern; callers may clear labels when deps complete.
-                dep_unmet = "blocked on #" in lower_body
-                dep_reason = (
-                    f"issue#{number} body marks an unmet dependency"
-                    if dep_unmet
-                    else None
-                )
+            # #758: parse the canonical "Depends: #N, #N" declaration into
+            # structured references and resolve each against live issue state.
+            # Body substrings no longer decide dependency status.
+            dep_refs = allocator_dependencies.parse_dependency_refs(body)
+            dep_result = allocator_dependencies.resolve_dependency_state(
+                dep_refs, _issue_state, subject=f"issue#{number}"
+            )
+            dep_unmet = dep_result["dependency_unmet"]
+            dep_reason = dep_result["reason"]
             try:
                 candidates.append(
                     allocator_service.WorkCandidate(
@@ -17937,7 +17973,7 @@ def _allocator_candidates_from_gitea(
                     f"skipped invalid issue candidate #{number}: {_redact(str(exc))}"
                 )
 
-    return candidates, reasons
+    return candidates, reasons, inventory_complete
 
 
 @mcp.tool()
@@ -18253,15 +18289,32 @@ def gitea_allocate_next_work(
                 "substrate": "control_plane_db",
             }
     else:
-        candidates, inv_reasons = _allocator_candidates_from_gitea(
+        candidates, inv_reasons, inventory_complete = _allocator_candidates_from_gitea(
             remote=remote,
             host=host,
             org=o,
             repo=r,
             include_issues=include_issues,
             include_prs=include_prs,
-            limit=limit,
         )
+        # #758 AC3: ranking a silently short inventory could select the wrong
+        # candidate, so an incomplete listing is a fail-closed stop.
+        if not inventory_complete:
+            return {
+                "success": False,
+                "outcome": allocator_service.OUTCOME_NO_SAFE,
+                "apply": bool(apply),
+                "reasons": inv_reasons
+                + [
+                    "candidate inventory is incomplete; refusing to rank a "
+                    "partial candidate set (fail closed, #758)"
+                ],
+                "inventory_complete": False,
+                "assignment": None,
+                "substrate": "control_plane_db",
+                "file_lock_only": False,
+                "comment_lease_only": False,
+            }
 
     sid = (session_id or "").strip() or (
         f"{profile_name or 'session'}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
@@ -18284,6 +18337,24 @@ def gitea_allocate_next_work(
     result["inventory_source"] = (
         "candidates_json" if candidates_json else "gitea_live"
     )
+    result["inventory_complete"] = True
+    result["selection_policy"] = allocator_service.SELECTION_POLICY
+    # #758 AC2/AC10: `limit` bounds the reported skip list only — selection has
+    # already happened over the complete inventory above. Any truncation here is
+    # stated explicitly so a shortened report is never read as full coverage.
+    reported_limit = max(1, int(limit))
+    skipped_all = result.get("skipped") or []
+    result["skipped_total"] = len(skipped_all)
+    result["limit_applies_to"] = "reported_skip_list_only"
+    if len(skipped_all) > reported_limit:
+        result["skipped"] = skipped_all[:reported_limit]
+        result["skipped_report_truncated"] = True
+        result.setdefault("reasons", []).append(
+            f"skip list truncated for reporting: showing {reported_limit} of "
+            f"{len(skipped_all)} skipped candidates (selection unaffected)"
+        )
+    else:
+        result["skipped_report_truncated"] = False
     # #606: watchdog check-ins for the recurring jobs this allocator run
     # performs — global stale-lease expiry, terminal-lock lookup, and the
     # allocator itself. Best-effort; a failed selection reports "error".