Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 77 Pull Requests 10 Actions Packages Projects Releases Wiki Activity

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

Browse Source 
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]>
This commit is contained in:
sysadmin
2026-07-06 12:29:16 -04:00
co-authored by Claude Opus 4.8
parent 08ed5a82d2
commit 70c868962a
3 changed files with 435 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+291 -2
View File
@@ -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"
@@ -1266,11 +1511,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",
@@ -1340,6 +1605,7 @@ def pr_inventory_trust_gate(
"status": "trusted_empty",
"reasons": [],
"corroborated": corroborated,
"queue_target_lock": lock_status or "resolved",
}
@@ -1355,6 +1621,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).
@@ -1369,9 +1638,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")
@@ -1383,6 +1655,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"),
@@ -1396,6 +1677,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")
@@ -1415,14 +1697,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 []: