feat(reviewer): add queue-target lock before PR inventory trust gate (#200)

Operators can supply an active PR backlog for one repo while agents
inventory another and falsely report trusted_empty. reconcile_queue_target
locks the target repo from operator context/backlog before list_prs runs;
pr_inventory_trust_gate blocks target_repo_mismatch and unresolved locks.
assess_reviewer_queue_inventory wires the lock into the canonical path.

Tests cover Gitea-Tools backlog vs mcp-control-plane inventory mismatch,
ambiguous dual-repo context, conflicting directive vs backlog, and final-
report field requirements.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

review_proofs.py

@@ -69,6 +69,251 @@ def resolve_repos_from_user_reference(
     return list(configured)
 
 
+_PR_NUMBER_RE = re.compile(r"(?:\bPR\s*#?|#)(\d+)\b", re.IGNORECASE)
+
+
+def _parse_pr_numbers(text):
+    """Extract PR numbers from operator context or backlog prose."""
+    if not text:
+        return []
+    seen = set()
+    ordered = []
+    for match in _PR_NUMBER_RE.finditer(text):
+        num = int(match.group(1))
+        if num not in seen:
+            seen.add(num)
+            ordered.append(num)
+    return ordered
+
+
+def _repo_hint_from_text(text, configured):
+    """Return a single configured repo named explicitly in *text*, if any."""
+    if not text:
+        return None
+    lower = text.lower()
+    for repo in configured:
+        if repo.lower() in lower:
+            return repo
+    resolved = resolve_repos_from_user_reference(text, configured)
+    if len(resolved) == 1:
+        return resolved[0]
+    return None
+
+
+def reconcile_queue_target(
+    *,
+    operator_context: str | None = None,
+    supplied_pr_backlog: list[dict] | None = None,
+    inventoried_repo: str | None = None,
+    project_context: str | None = None,
+    configured_repos: list[str] | None = None,
+) -> dict:
+    """Reconcile the inventory target repo before listing open PRs (#200).
+
+    Compares operator-supplied PR numbers/titles/backlog against the repo the
+    workflow is about to inventory. Returns a ``queue_target_lock`` dict whose
+    ``status`` must be ``resolved`` before an empty queue may stop cleanly.
+    """
+    if configured_repos is None:
+        configured_repos = [
+            "Scaled-Tech-Consulting/Gitea-Tools",
+            "Scaled-Tech-Consulting/mcp-control-plane",
+        ]
+
+    backlog_items = []
+    for item in supplied_pr_backlog or []:
+        number = item.get("number")
+        if number is None:
+            continue
+        repo = (item.get("repo") or "").strip() or None
+        backlog_items.append({
+            "number": int(number),
+            "repo": repo,
+            "title": (item.get("title") or "").strip() or None,
+        })
+
+    context_numbers = _parse_pr_numbers(operator_context or "")
+    backlog_numbers = [item["number"] for item in backlog_items]
+    supplied_pr_numbers = list(dict.fromkeys(backlog_numbers + context_numbers))
+
+    context_repo = _repo_hint_from_text(operator_context, configured_repos)
+    project_repo = _repo_hint_from_text(project_context, configured_repos)
+
+    resolution_source = None
+    resolved_repo = None
+    reasons = []
+
+    explicit_repos = {
+        item["repo"] for item in backlog_items if item.get("repo")
+    }
+    if len(explicit_repos) == 1:
+        resolved_repo = next(iter(explicit_repos))
+        resolution_source = "supplied_pr_backlog"
+    elif context_repo:
+        resolved_repo = context_repo
+        resolution_source = "operator_context"
+    elif project_repo and supplied_pr_numbers:
+        resolved_repo = project_repo
+        resolution_source = "project_context"
+
+    if (
+        context_repo
+        and explicit_repos
+        and context_repo not in explicit_repos
+    ):
+        return {
+            "status": "unresolved",
+            "resolved_repo": None,
+            "resolution_source": None,
+            "supplied_pr_numbers": supplied_pr_numbers,
+            "reconciliation": [],
+            "inventoried_repo": (inventoried_repo or "").strip() or None,
+            "reasons": [
+                "operator context repo conflicts with supplied PR backlog "
+                f"repos ({context_repo} vs {sorted(explicit_repos)})"
+            ],
+            "allow_clean_stop": False,
+            "allow_trusted_empty": False,
+        }
+
+    reconciliation = []
+    for number in supplied_pr_numbers:
+        expected_repo = None
+        for item in backlog_items:
+            if item["number"] == number and item.get("repo"):
+                expected_repo = item["repo"]
+                break
+        if expected_repo is None:
+            expected_repo = resolved_repo
+        reconciliation.append({
+            "pr_number": number,
+            "expected_repo": expected_repo,
+            "inventoried_repo": inventoried_repo,
+            "matches_inventoried_repo": (
+                expected_repo is not None
+                and inventoried_repo is not None
+                and expected_repo == inventoried_repo
+            ),
+        })
+
+    inventoried = (inventoried_repo or "").strip() or None
+
+    if supplied_pr_numbers and resolved_repo is None:
+        status = "unresolved"
+        reasons.append(
+            "operator supplied PR numbers but target repository could not "
+            "be resolved"
+        )
+    elif (
+        supplied_pr_numbers
+        and resolved_repo
+        and inventoried
+        and inventoried != resolved_repo
+    ):
+        status = "target_repo_mismatch"
+        reasons.append(
+            f"inventoried repository '{inventoried}' does not own the "
+            f"operator-supplied PR backlog (expected '{resolved_repo}')"
+        )
+    elif supplied_pr_numbers and resolved_repo and inventoried == resolved_repo:
+        status = "resolved"
+    elif not supplied_pr_numbers and inventoried:
+        status = "resolved"
+        resolution_source = resolution_source or "inventoried_repo_only"
+        resolved_repo = inventoried
+    elif not supplied_pr_numbers:
+        status = "unresolved"
+        reasons.append("no operator-supplied PR backlog to reconcile")
+    else:
+        status = "resolved"
+
+    allow_clean_stop = status == "resolved"
+    allow_trusted_empty = status == "resolved"
+
+    return {
+        "status": status,
+        "resolved_repo": resolved_repo,
+        "resolution_source": resolution_source,
+        "supplied_pr_numbers": supplied_pr_numbers,
+        "reconciliation": reconciliation,
+        "inventoried_repo": inventoried,
+        "reasons": reasons,
+        "allow_clean_stop": allow_clean_stop,
+        "allow_trusted_empty": allow_trusted_empty,
+    }
+
+
+resolve_pr_queue_target = reconcile_queue_target
+
+
+def assess_queue_target_final_report(report_text, queue_target_lock):
+    """Require final reports to document queue-target reconciliation."""
+    lock = queue_target_lock or {}
+    text = report_text or ""
+    lower = text.lower()
+    missing = []
+
+    if not lock.get("resolved_repo"):
+        missing.append("resolved repo")
+    elif lock["resolved_repo"].lower() not in lower:
+        missing.append("resolved repo")
+
+    source = lock.get("resolution_source")
+    if not source:
+        missing.append("resolution source")
+    else:
+        source_lower = str(source).lower()
+        if (
+            source_lower not in lower
+            and source_lower.replace("_", " ") not in lower
+        ):
+            missing.append("resolution source")
+
+    for number in lock.get("supplied_pr_numbers") or []:
+        if f"#{number}" not in lower and f"pr {number}" not in lower:
+            missing.append(f"supplied PR #{number} reconciliation")
+            break
+
+    if lock.get("status") and str(lock["status"]).lower() not in lower:
+        missing.append("queue_target_lock.status")
+
+    if missing:
+        return {
+            "complete": False,
+            "downgraded": True,
+            "missing_fields": missing,
+            "reasons": [
+                f"final report missing queue-target field: {field}"
+                for field in missing
+            ],
+        }
+    return {
+        "complete": True,
+        "downgraded": False,
+        "missing_fields": [],
+        "reasons": [],
+    }
+
+
+def format_queue_target_lock_report(lock: dict) -> list[str]:
+    """Render queue-target lock lines for inventory output."""
+    lines = [
+        f"queue_target_lock.status: {lock.get('status', 'unknown')}",
+    ]
+    if lock.get("resolved_repo"):
+        lines.append(f"queue_target_lock.resolved_repo: {lock['resolved_repo']}")
+    if lock.get("resolution_source"):
+        lines.append(
+            f"queue_target_lock.resolution_source: {lock['resolution_source']}"
+        )
+    if lock.get("supplied_pr_numbers"):
+        nums = ", ".join(f"#{n}" for n in lock["supplied_pr_numbers"])
+        lines.append(f"queue_target_lock.supplied_pr_numbers: {nums}")
+    for reason in lock.get("reasons") or []:
+        lines.append(f"queue_target_lock.reason: {reason}")
+    return lines
+
+
 SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION = (
     "evidence missing: report contamination as unknown and "
     "choose another PR or stop"
@@ -1234,11 +1479,31 @@ def pr_inventory_trust_gate(
     user_context: str | None = None,
     corroboration_open_pr_counter: int | None = None,
     has_finality_metadata: bool = False,
+    queue_target_lock: dict | None = None,
 ) -> dict:
     """Evaluate whether an empty PR list is trusted or untrusted.
 
     Returns a dict with 'status', 'reasons', and 'corroborated'.
     """
+    lock = queue_target_lock or {}
+    lock_status = lock.get("status")
+    if lock_status == "target_repo_mismatch":
+        return {
+            "status": "target_repo_mismatch",
+            "reasons": list(lock.get("reasons") or []),
+            "corroborated": False,
+            "queue_target_lock": lock_status,
+        }
+    if lock and lock_status != "resolved":
+        return {
+            "status": "untrusted_empty",
+            "reasons": [
+                f"queue_target_lock.status is '{lock_status}', not 'resolved'"
+            ] + list(lock.get("reasons") or []),
+            "corroborated": False,
+            "queue_target_lock": lock_status,
+        }
+
     if list_prs_response is None or not isinstance(list_prs_response, list):
         return {
             "status": "inventory_error",
@@ -1308,6 +1573,7 @@ def pr_inventory_trust_gate(
         "status": "trusted_empty",
         "reasons": [],
         "corroborated": corroborated,
+        "queue_target_lock": lock_status or "resolved",
     }
 
 
@@ -1323,6 +1589,9 @@ def assess_reviewer_queue_inventory(
     required_repos: list[str] | None = None,
     *,
     user_context: str | None = None,
+    operator_context: str | None = None,
+    supplied_pr_backlog: list[dict] | None = None,
+    project_context: str | None = None,
 ) -> dict:
     """Canonical reviewer queue path: completeness plus per-repo trust gates (#196).
 
@@ -1337,9 +1606,12 @@ def assess_reviewer_queue_inventory(
     completeness = assess_inventory_completeness(repo_reports, required)
 
     trust_gates: dict[str, dict] = {}
+    queue_target_locks: dict[str, dict] = {}
     blockers: list[str] = []
     can_claim_empty = bool(completeness.get("complete"))
 
+    context = operator_context or user_context
+
     for report in repo_reports or []:
         repo = (report.get("repo") or "").strip()
         count = report.get("open_pr_count")
@@ -1351,6 +1623,15 @@ def assess_reviewer_queue_inventory(
         if list_response is None:
             list_response = []
 
+        queue_target_lock = reconcile_queue_target(
+            operator_context=context,
+            supplied_pr_backlog=supplied_pr_backlog,
+            inventoried_repo=repo,
+            project_context=project_context,
+            configured_repos=required,
+        )
+        queue_target_locks[repo] = queue_target_lock
+
         gate = pr_inventory_trust_gate(
             list_response,
             remote=report.get("remote"),
@@ -1364,6 +1645,7 @@ def assess_reviewer_queue_inventory(
                 "corroboration_open_pr_counter"
             ),
             has_finality_metadata=report.get("pagination_complete") is True,
+            queue_target_lock=queue_target_lock,
         )
         trust_gates[repo] = gate
         status = gate.get("status")
@@ -1383,14 +1665,21 @@ def assess_reviewer_queue_inventory(
         ),
         "inventory_reasons": list(completeness.get("reasons") or []),
         "trust_gates": trust_gates,
+        "queue_target_locks": queue_target_locks,
         "blockers": blockers,
         "reasons": list(completeness.get("reasons") or []) + blockers,
     }
 
 
-def format_pr_inventory_trust_gate_report(gate: dict) -> list[str]:
+def format_pr_inventory_trust_gate_report(
+    gate: dict,
+    queue_target_lock: dict | None = None,
+) -> list[str]:
     """Render trust-gate lines for MCP inventory output."""
-    lines = [f"pr_inventory_trust_gate.status: {gate.get('status', 'unknown')}"]
+    lines = []
+    if queue_target_lock:
+        lines.extend(format_queue_target_lock_report(queue_target_lock))
+    lines.append(f"pr_inventory_trust_gate.status: {gate.get('status', 'unknown')}")
     if gate.get("corroborated"):
         lines.append("pr_inventory_trust_gate.corroborated: true")
     for reason in gate.get("reasons") or []:

skills/llm-project-workflow/SKILL.md

@@ -216,7 +216,10 @@ Worktree folder = branch with `/` replaced by `-`
    the other.
    Both configured repos must be reported with state filter, pagination proof,
    and open-PR count (`review_proofs.assess_inventory_completeness` and
-   `resolve_repos_from_user_reference`).
+   `resolve_repos_from_user_reference`). Before inventory, reconcile the
+   operator-supplied PR backlog against the target repo
+   (`review_proofs.reconcile_queue_target`); never report `trusted_empty`
+   for one repo while ignoring contradictory supplied PR numbers in another.
 7. **Role-boundary proof (#175):** a reviewer queue task must not silently
    become author implementation. If no eligible PR exists, stop with the
    queue report. Do not claim issues, create branches, commit, push, or open

tests/test_review_proofs.py

@@ -26,6 +26,7 @@ from review_proofs import (  # noqa: E402
     assess_capability_proof,
     assess_controller_handoff,
     assess_inventory_completeness,
+    assess_queue_target_final_report,
     assess_reviewer_queue_inventory,
     assess_live_state_recheck,
     assess_review_mutation_final_report,
@@ -36,6 +37,7 @@ from review_proofs import (  # noqa: E402
     assess_validation_report,
     build_final_report,
     pr_inventory_trust_gate,
+    reconcile_queue_target,
     resolve_repos_from_user_reference,
     verify_pinned_head_checkout,
 )
@@ -1112,6 +1114,144 @@ class TestReviewMutationFinalReport(unittest.TestCase):
         self.assertTrue(final["review_mutation_complete"])
 
 
+class TestQueueTargetReconciliation(unittest.TestCase):
+    """Queue target lock: reconcile operator-supplied backlog before inventory (#200)."""
+
+    CONFIGURED = [
+        "Scaled-Tech-Consulting/Gitea-Tools",
+        "Scaled-Tech-Consulting/mcp-control-plane",
+    ]
+    GITEA_TOOLS = "Scaled-Tech-Consulting/Gitea-Tools"
+    MCP = "Scaled-Tech-Consulting/mcp-control-plane"
+    OPERATOR_CONTEXT = (
+        "six open PRs in Scaled-Tech-Consulting/Gitea-Tools including "
+        "#195, #193, #192, #190, #187, and #181"
+    )
+    PROFILE = {
+        "profile_name": "prgs-reviewer",
+        "allowed_operations": ["read", "gitea.read"],
+    }
+
+    def test_supplied_gitea_tools_prs_but_inventoried_mcp_is_mismatch(self):
+        lock = reconcile_queue_target(
+            operator_context=self.OPERATOR_CONTEXT,
+            inventoried_repo=self.MCP,
+            configured_repos=self.CONFIGURED,
+        )
+        self.assertEqual(lock["status"], "target_repo_mismatch")
+        self.assertEqual(lock["resolved_repo"], self.GITEA_TOOLS)
+        self.assertEqual(lock["resolution_source"], "operator_context")
+        self.assertIn(195, lock["supplied_pr_numbers"])
+        self.assertFalse(lock["allow_clean_stop"])
+        self.assertFalse(lock["allow_trusted_empty"])
+
+    def test_wrong_repo_zero_open_cannot_stop_cleanly(self):
+        lock = reconcile_queue_target(
+            operator_context=self.OPERATOR_CONTEXT,
+            inventoried_repo=self.MCP,
+            configured_repos=self.CONFIGURED,
+        )
+        gate = pr_inventory_trust_gate(
+            [],
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="mcp-control-plane",
+            state="open",
+            authenticated_profile=self.PROFILE,
+            local_remote_url=(
+                "https://gitea.prgs.cc/Scaled-Tech-Consulting/"
+                "mcp-control-plane.git"
+            ),
+            corroboration_open_pr_counter=0,
+            queue_target_lock=lock,
+        )
+        self.assertEqual(gate["status"], "target_repo_mismatch")
+        self.assertNotEqual(gate["status"], "trusted_empty")
+
+    def test_supplied_pr_numbers_reconciled_before_empty_stop(self):
+        lock = reconcile_queue_target(
+            operator_context=(
+                "currently we have 6 open PRs: #195, #193, #192, "
+                "#190, #187, #181 in Gitea-Tools"
+            ),
+            inventoried_repo=self.GITEA_TOOLS,
+            configured_repos=self.CONFIGURED,
+        )
+        self.assertEqual(lock["status"], "resolved")
+        self.assertEqual(len(lock["supplied_pr_numbers"]), 6)
+        self.assertTrue(all(item["matches_inventoried_repo"]
+                            for item in lock["reconciliation"]))
+
+    def test_ambiguous_repo_context_fails_closed(self):
+        lock = reconcile_queue_target(
+            operator_context=(
+                "open PRs in mcp-control-plane and gitea-tools including PR #195"
+            ),
+            inventoried_repo=self.MCP,
+            configured_repos=self.CONFIGURED,
+        )
+        self.assertEqual(lock["status"], "unresolved")
+        self.assertFalse(lock["allow_trusted_empty"])
+        self.assertTrue(
+            any("could not be resolved" in r for r in lock["reasons"])
+        )
+
+    def test_conflicting_directive_vs_backlog_fails_closed(self):
+        lock = reconcile_queue_target(
+            operator_context="Repository: Scaled-Tech-Consulting/mcp-control-plane",
+            supplied_pr_backlog=[
+                {"number": 195, "repo": self.GITEA_TOOLS},
+            ],
+            inventoried_repo=self.MCP,
+            configured_repos=self.CONFIGURED,
+        )
+        self.assertEqual(lock["status"], "unresolved")
+        self.assertIn("conflicts", " ".join(lock["reasons"]).lower())
+
+    def test_final_report_must_document_reconciliation(self):
+        lock = reconcile_queue_target(
+            operator_context=self.OPERATOR_CONTEXT,
+            inventoried_repo=self.GITEA_TOOLS,
+            configured_repos=self.CONFIGURED,
+        )
+        incomplete = assess_queue_target_final_report(
+            "Open PRs: 0. Stopping.", lock
+        )
+        self.assertFalse(incomplete["complete"])
+        self.assertTrue(incomplete["downgraded"])
+
+        complete_report = "\n".join([
+            "Queue inventory complete.",
+            f"Resolved repo: {self.GITEA_TOOLS}",
+            "Resolution source: operator_context",
+            "queue_target_lock.status: resolved",
+            "Supplied PR reconciliation: #195, #193, #192, #190, #187, #181",
+        ])
+        complete = assess_queue_target_final_report(complete_report, lock)
+        self.assertTrue(complete["complete"])
+        self.assertFalse(complete["downgraded"])
+
+    def test_assess_reviewer_queue_inventory_blocks_mismatch(self):
+        result = assess_reviewer_queue_inventory(
+            [{
+                "repo": self.MCP,
+                "state_filter": "open",
+                "pagination_complete": True,
+                "open_pr_count": 0,
+                "list_prs_response": [],
+                "remote": "prgs",
+                "authenticated_profile": self.PROFILE,
+                "local_remote_url": (
+                    "https://gitea.prgs.cc/Scaled-Tech-Consulting/"
+                    "mcp-control-plane.git"
+                ),
+            }],
+            operator_context=self.OPERATOR_CONTEXT,
+        )
+        self.assertFalse(result["can_claim_empty_queue"])
+        self.assertIn("target_repo_mismatch", str(result["trust_gates"]))
+
+
 class TestPRInventoryTrustGate(unittest.TestCase):
     """Issue #194: unit tests for the PR inventory trust gate."""