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

feat(author): add continuation-mode selection wall for open-PR issues (#188)

Browse Source 
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-06 11:40:35 -04:00
co-authored by Claude Opus 4.8
parent 056a232ef8
commit dc41b685d0
3 changed files with 700 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+448
View File
@@ -664,6 +664,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
report_text, review_decision_lock
)
empty_queue_report = assess_empty_queue_report(report_text)
contamination_status = contamination.get("status", "unknown")
checkout_proven = bool(checkout_proof.get("proven"))
validation_claimable = bool(validation.get("claimable"))
@@ -787,6 +789,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"reviewer worktree safety proof missing or failed (#233)"
)
downgrade_reasons.extend(worktree.get("reasons", []))
if empty_queue_report.get("claimed") and not empty_queue_report.get("proven"):
downgrade_reasons.append(
"empty-queue report missing or failed trust-gate proof (#198)"
)
downgrade_reasons.extend(empty_queue_report.get("reasons", []))
merge_allowed = (
identity_eligible
@@ -846,6 +853,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"unrelated_mutations_avoided": bool(
worktree.get("unrelated_mutations_avoided")
),
"empty_queue_trust_gate_proven": (
empty_queue_report.get("proven")
if empty_queue_report.get("claimed")
else True
),
"empty_queue_trust_gate_status": empty_queue_report.get("status"),
}
@@ -1005,6 +1018,7 @@ HANDOFF_ROLE_FIELDS = {
),
"author": (
("Selected issue", ("selected issue",)),
("Issue lock proof", ("issue lock proof", "lock before diff")),
("Claim/comment status", ("claim/comment status", "claim status",
"claim")),
("PR number opened", ("pr number opened", "pr opened", "pr number")),
@@ -1015,11 +1029,29 @@ HANDOFF_ROLE_FIELDS = {
("Repositories checked", ("repositories checked", "repos checked")),
("Open PR counts", ("open pr counts", "open pr count",
"open prs per repo")),
("PR inventory trust gate", ("pr inventory trust gate",
"pr_inventory_trust_gate.status",
"trust gate status")),
("Trust gate reasons", ("trust gate reasons",
"pr_inventory_trust_gate.reason")),
("Trust gate corroborated", ("trust gate corroborated",
"pr_inventory_trust_gate.corroborated")),
("Inventory profile", ("inventory profile", "inventory mcp profile")),
("Selected PR or reason", ("selected pr", "none selected",
"reason none selected")),
("Inventory completeness", ("inventory complete", "inventory scoped",
"inventory completeness")),
),
"continuation": (
("Continuation mode", ("continuation mode", "continuation")),
("Existing PR", ("existing pr", "pr number")),
("PR author", ("pr author", "existing pr author")),
("Branch", ("branch", "existing branch")),
("Old PR head", ("old pr head", "old head")),
("New PR head", ("new pr head", "new head")),
("Session authored PR", ("session authored pr", "authored pr")),
("Why continuation allowed", ("why continuation", "continuation allowed")),
),
}
@@ -1398,6 +1430,422 @@ def format_pr_inventory_trust_gate_report(gate: dict) -> list[str]:
return lines
_EMPTY_QUEUE_CLAIM = re.compile(
r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|"
r"nothing to review|queue cleared|inventory empty|"
r"open pr count:\s*0|workflow correctly stops with nothing",
re.I,
)
_WEAK_EMPTY_QUEUE_CORROBORATION = re.compile(
r"latest commit.*(?:merge|pr #)|merge of pr #|"
r"master latest commit|recent merge proves",
re.I,
)
_TRUST_GATE_STATUS_LINE = re.compile(
r"pr_inventory_trust_gate\.status:\s*(\S+)",
re.I,
)
def parse_trust_gate_status_from_report(report_text: str | None) -> str | None:
"""Extract ``pr_inventory_trust_gate.status`` from report text, if present."""
match = _TRUST_GATE_STATUS_LINE.search(report_text or "")
return match.group(1).strip().lower() if match else None
def assess_empty_queue_report(
report_text: str | None,
*,
trust_gate: dict | None = None,
task_role: str | None = None,
inventory_profile: str | None = None,
) -> dict:
"""Issue #198: empty-queue reports must cite the formal trust-gate result.
Blocks reports that claim an empty queue without
``pr_inventory_trust_gate.status == trusted_empty``, required inventory
metadata, or that rely on weak corroboration (e.g. a recent merge commit).
"""
text = report_text or ""
lower = text.lower()
reasons: list[str] = []
missing: list[str] = []
if not _EMPTY_QUEUE_CLAIM.search(text):
return {
"claimed": False,
"proven": True,
"block": False,
"status": None,
"missing_fields": [],
"reasons": [],
}
status = (
(trust_gate or {}).get("status")
or parse_trust_gate_status_from_report(text)
)
status_norm = (status or "").strip().lower() or None
if not status_norm:
missing.append("pr_inventory_trust_gate.status")
reasons.append(
"empty-queue report missing pr_inventory_trust_gate.status; "
"fail closed"
)
elif status_norm != "trusted_empty":
reasons.append(
f"empty-queue report has trust-gate status '{status_norm}', "
"not trusted_empty"
)
has_gate_reasons = (
"pr_inventory_trust_gate.reason" in lower
or bool((trust_gate or {}).get("reasons"))
)
if status_norm and status_norm != "trusted_empty" and not has_gate_reasons:
missing.append("pr_inventory_trust_gate.reasons")
if status_norm == "trusted_empty":
if "pr_inventory_trust_gate.corroborated" not in lower and (
trust_gate or {}
).get("corroborated") is not True:
missing.append("pr_inventory_trust_gate.corroborated")
inventory_markers = (
"repository:",
"remote:",
"owner:",
"state filter:",
"state_filter:",
)
if not any(marker in lower for marker in inventory_markers):
missing.append("inventory remote/owner/repo/state filter")
profile_markers = (
"mcp profile:",
"mcp-profile:",
"inventory profile:",
"active profile:",
)
has_profile = (
any(marker in lower for marker in profile_markers)
or bool((inventory_profile or "").strip())
)
if not has_profile:
missing.append("inventory MCP profile")
if _WEAK_EMPTY_QUEUE_CORROBORATION.search(text):
if "pr_inventory_trust_gate.status: trusted_empty" not in lower:
reasons.append(
"weak corroboration (recent merge commit) cannot substitute "
"for pr_inventory_trust_gate.status == trusted_empty"
)
role = (task_role or "").strip().lower()
if role == "author" and re.search(
r"reviewer queue|nothing to review|review backlog empty",
text,
re.I,
):
reasons.append(
"author-bound session presented reviewer queue inventory as a "
"reviewer decision"
)
if missing:
reasons.extend(
f"empty-queue report missing required field: {field}"
for field in missing
)
proven = not reasons and not missing
return {
"claimed": True,
"proven": proven,
"block": not proven,
"status": status_norm,
"missing_fields": missing,
"reasons": reasons,
}
# ── Issue selection / continuation mode (#188) ───────────────────────────────
ISSUE_SELECTION_UNCLAIMED_NO_PR = "unclaimed_no_pr"
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR = "represented_by_open_pr"
ISSUE_SELECTION_IN_PROGRESS = "in_progress"
ISSUE_SELECTION_CONTINUATION_EXPLICIT = "continuation_explicit"
ISSUE_SELECTION_EXCLUDED = "excluded"
_NO_OPEN_PR_CLAIM = re.compile(
r"no duplicate pr|no open pr|no pr open|no eligible pr|"
r"no existing pr|without an open pr",
re.IGNORECASE,
)
def classify_issue_for_selection(
issue_number: int,
*,
labels: list[str] | None = None,
open_prs: list[dict] | None = None,
operator_continuation_requested: bool = False,
continuation_issue_numbers: list[int] | None = None,
excluded: bool = False,
) -> dict:
"""Classify one issue for author queue selection (#188)."""
label_set = {str(l).lower() for l in (labels or [])}
prs = list(open_prs or [])
continuation_issues = set(continuation_issue_numbers or [])
if excluded:
status = ISSUE_SELECTION_EXCLUDED
selectable_for_fresh_work = False
reasons = ["issue explicitly excluded from selection"]
elif "status:in-progress" in label_set:
status = ISSUE_SELECTION_IN_PROGRESS
selectable_for_fresh_work = False
reasons = ["issue already marked status:in-progress"]
elif prs and (
operator_continuation_requested
or issue_number in continuation_issues
):
status = ISSUE_SELECTION_CONTINUATION_EXPLICIT
selectable_for_fresh_work = False
reasons = ["operator requested continuation for issue with open PR"]
elif prs:
status = ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR
selectable_for_fresh_work = False
reasons = [
f"issue #{issue_number} already represented by open PR "
f"#{prs[0].get('number')}"
]
else:
status = ISSUE_SELECTION_UNCLAIMED_NO_PR
selectable_for_fresh_work = True
reasons = []
return {
"issue_number": issue_number,
"status": status,
"selectable_for_fresh_work": selectable_for_fresh_work,
"open_prs": prs,
"reasons": reasons,
}
def assess_fresh_issue_selection(classifications: list[dict] | None) -> dict:
"""Fail closed when fresh selection picks an issue with an open PR."""
reasons = []
for item in classifications or []:
if item.get("selectable_for_fresh_work"):
continue
if item.get("status") == ISSUE_SELECTION_CONTINUATION_EXPLICIT:
continue
if item.get("status") == ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR:
reasons.append(
f"issue #{item.get('issue_number')} has open PR and was "
"selected for fresh work without continuation mode"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def assess_continuation_mode_report(
report_text: str,
*,
pr_number: int | None = None,
pr_author: str | None = None,
branch: str | None = None,
old_head_sha: str | None = None,
new_head_sha: str | None = None,
session_authored_pr: bool | None = None,
continuation_allowed_reason: str | None = None,
) -> dict:
"""Issue #188: continuation mode must disclose full PR evidence."""
text = report_text or ""
lower = text.lower()
reasons = []
if not any(p in lower for p in ("continuation", "continue pr", "continue issue")):
reasons.append("report does not declare continuation mode")
if pr_number is not None:
if f"#{pr_number}" not in lower and f"pr {pr_number}" not in lower:
reasons.append(f"continuation report missing PR #{pr_number}")
if pr_author and pr_author.lower() not in lower:
reasons.append("continuation report missing PR author")
if branch and branch.lower() not in lower:
reasons.append("continuation report missing branch name")
for label, sha in (("old", old_head_sha), ("new", new_head_sha)):
if not sha:
reasons.append(f"continuation proof missing {label} head SHA")
elif not _FULL_SHA.match(sha.lower()):
reasons.append(
f"continuation {label} head SHA is not a full 40-hex SHA"
)
elif sha.lower() not in lower:
reasons.append(
f"continuation report missing {label} head SHA in evidence"
)
if session_authored_pr is not None:
authored_tokens = ("session authored", "authored pr", "own pr", "my pr")
if not any(t in lower for t in authored_tokens):
reasons.append(
"continuation report missing whether session authored the PR"
)
if continuation_allowed_reason:
reason_lower = continuation_allowed_reason.lower()
if (
reason_lower not in lower
and not any(w in lower for w in reason_lower.split()[:3])
):
reasons.append("continuation report missing why continuation is allowed")
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def assess_contradictory_no_pr_claim(
report_text: str,
*,
edited_pr_numbers: list[int] | None = None,
issue_open_pr_map: dict[int, int] | None = None,
) -> dict:
"""Downgrade when report claims no open PR but later edits one."""
text = report_text or ""
lower = text.lower()
reasons = []
if not _NO_OPEN_PR_CLAIM.search(lower):
return {"complete": True, "downgraded": False, "reasons": []}
edited = list(edited_pr_numbers or [])
for pr_num in edited:
if f"#{pr_num}" in lower or f"pr {pr_num}" in lower:
reasons.append(
f"report claims no open PR but edited PR #{pr_num}"
)
for issue_num, pr_num in (issue_open_pr_map or {}).items():
if f"#{issue_num}" in lower or f"issue #{issue_num}" in lower:
if f"#{pr_num}" in lower or f"pr {pr_num}" in lower:
reasons.append(
f"report claims no open PR for issue #{issue_num} but "
f"PR #{pr_num} exists and was referenced"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def assess_edited_pr_inventory_coverage(
report_text: str,
*,
edited_pr_numbers: list[int] | None = None,
inventoried_pr_numbers: list[int] | None = None,
) -> dict:
"""Open PR inventory must include PRs the run later edits (#188)."""
text = report_text or ""
lower = text.lower()
reasons = []
inventoried = set(inventoried_pr_numbers or [])
for pr_num in edited_pr_numbers or []:
if pr_num in inventoried:
continue
if f"#{pr_num}" not in lower and f"pr {pr_num}" not in lower:
reasons.append(
f"edited PR #{pr_num} missing from open PR inventory"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def assess_issue_selection_final_report(
report_text: str,
*,
mode: str = "fresh",
classifications: list[dict] | None = None,
continuation_proof: dict | None = None,
edited_pr_numbers: list[int] | None = None,
inventoried_pr_numbers: list[int] | None = None,
issue_open_pr_map: dict[int, int] | None = None,
) -> dict:
"""Issue #188: composite A-bar for author issue-selection runs."""
handoff_role = "continuation" if mode == "continuation" else "author"
checks = {
"controller_handoff": assess_controller_handoff(
report_text, role=handoff_role
),
"contradictory_no_pr": assess_contradictory_no_pr_claim(
report_text,
edited_pr_numbers=edited_pr_numbers,
issue_open_pr_map=issue_open_pr_map,
),
"edited_pr_inventory": assess_edited_pr_inventory_coverage(
report_text,
edited_pr_numbers=edited_pr_numbers,
inventoried_pr_numbers=inventoried_pr_numbers,
),
}
if mode == "fresh":
checks["fresh_selection"] = assess_fresh_issue_selection(classifications)
else:
proof = continuation_proof or {}
checks["continuation_mode"] = assess_continuation_mode_report(
report_text,
pr_number=proof.get("pr_number"),
pr_author=proof.get("pr_author"),
branch=proof.get("branch"),
old_head_sha=proof.get("old_head_sha"),
new_head_sha=proof.get("new_head_sha"),
session_authored_pr=proof.get("session_authored_pr"),
continuation_allowed_reason=proof.get("continuation_allowed_reason"),
)
reasons = []
downgraded = False
for name, result in checks.items():
verdict = result.get("verdict")
if verdict in ("missing", "incomplete"):
downgraded = True
reasons.extend(result.get("reasons") or [])
elif result.get("downgraded") or not result.get("complete", True):
downgraded = True
reasons.extend(
f"{name}: {r}" for r in (result.get("reasons") or [])
)
return {
"grade": "A" if not downgraded else "downgraded",
"downgraded": downgraded,
"checks": checks,
"reasons": reasons,
"complete": not downgraded,
}
def assess_duplicate_search_proof(report_text, matches):
"""#207: reject LLM duplicate summaries that omit known title matches."""
return issue_duplicate_gate.assess_duplicate_search_proof(
skills/llm-project-workflow/SKILL.md
+6
View File
@@ -379,6 +379,12 @@ Role-specific fields (append to the compact block):
`Linked issue status:`, `Cleanup status:`
- author tasks: `Selected issue:`, `Claim/comment status:`,
`PR number opened:`, `No review/merge:` (explicit confirmation)
- continuation tasks (#188): `Continuation mode:`, `Existing PR:`,
`PR author:`, `Branch:`, `Old PR head:`, `New PR head:`,
`Session authored PR:`, `Why continuation allowed:` — issues with open
PRs are excluded from fresh selection unless operator explicitly requests
continuation (`review_proofs.classify_issue_for_selection`,
`assess_issue_selection_final_report`)
- queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
`Selected PR or reason none selected:`, `Inventory completeness:`
tests/test_review_proofs.py
+246 -17
View File
@@ -21,11 +21,19 @@ import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from review_proofs import ( # noqa: E402
ISSUE_SELECTION_CONTINUATION_EXPLICIT,
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
assess_author_pr_report,
assess_capability_evidence,
assess_capability_proof,
assess_contradictory_no_pr_claim,
assess_continuation_mode_report,
assess_controller_handoff,
assess_edited_pr_inventory_coverage,
assess_empty_queue_report,
assess_fresh_issue_selection,
assess_inventory_completeness,
assess_issue_selection_final_report,
assess_reviewer_queue_inventory,
assess_live_state_recheck,
assess_review_mutation_final_report,
@@ -35,6 +43,7 @@ from review_proofs import ( # noqa: E402
assess_sweep_evidence,
assess_validation_report,
build_final_report,
classify_issue_for_selection,
pr_inventory_trust_gate,
resolve_repos_from_user_reference,
verify_pinned_head_checkout,
@@ -944,24 +953,46 @@ class TestControllerHandoff(unittest.TestCase):
result = assess_controller_handoff(complete, role="review")
self.assertEqual(result["verdict"], "complete")
def test_author_role_requires_author_fields(self):
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #182",
def _author_role_fields(self, issue_number=182, pr_number=999):
return [
f"- Selected issue: #{issue_number}",
"- Issue lock proof: lock before diff on feat/x @ master",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
f"- PR number opened: #{pr_number}",
"- No review/merge: confirmed",
])
]
def test_handoff_role_fields_author_includes_issue_lock_proof(self):
from review_proofs import HANDOFF_ROLE_FIELDS
names = [name for name, _ in HANDOFF_ROLE_FIELDS["author"]]
self.assertIn("Issue lock proof", names)
def test_author_role_requires_author_fields(self):
complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields())
result = assess_controller_handoff(complete, role="author")
self.assertEqual(result["verdict"], "complete")
result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Issue lock proof", result["missing_fields"])
self.assertIn("No review/merge confirmation", result["missing_fields"])
def test_author_role_requires_issue_lock_proof(self):
without_lock = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #182",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
])
result = assess_controller_handoff(without_lock, role="author")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Issue lock proof", result["missing_fields"])
def test_author_role_rejects_equivalent_or_multiple_issues(self):
# 1. equivalent reference blocked
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: Issue #194 / #196 equivalent",
"- Issue lock proof: lock before diff on feat/x @ master",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
@@ -973,6 +1004,7 @@ class TestControllerHandoff(unittest.TestCase):
# 2. multiple issues blocked
incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #194, #196",
"- Issue lock proof: lock before diff on feat/x @ master",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
@@ -984,6 +1016,7 @@ class TestControllerHandoff(unittest.TestCase):
def test_author_role_rejects_fuzzy_pr_number(self):
incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #196",
"- Issue lock proof: lock before diff on feat/issue-196 @ master",
"- Claim/comment status: comment-claimed",
"- PR number opened: PR #203 / #204 equivalent",
"- No review/merge: confirmed",
@@ -996,6 +1029,10 @@ class TestControllerHandoff(unittest.TestCase):
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Repositories checked: Gitea-Tools, mcp-control-plane",
"- Open PR counts: 2 / 0",
"- PR inventory trust gate: trusted_nonempty / trusted_empty",
"- Trust gate reasons: none",
"- Trust gate corroborated: true",
"- Inventory profile: prgs-reviewer",
"- Selected PR or reason: none eligible (self-authored)",
"- Inventory completeness: complete, no pagination needed",
])
@@ -1015,23 +1052,17 @@ class TestControllerHandoff(unittest.TestCase):
def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
# 1. Workspace mutations: none is rejected when local_edits is True
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: #203",
"- No review/merge: confirmed",
])
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join(
self._author_role_fields(issue_number=196, pr_number=203))
res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
self.assertEqual(res["verdict"], "incomplete")
self.assertIn("Workspace mutations", res["missing_fields"])
# 2. Workspace mutations: edited files is allowed when local_edits is True
complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
"- Selected issue: #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: #203",
"- No review/merge: confirmed",
])
complete_eq = self.BASE_HANDOFF.replace(
"- Workspace mutations: none",
"- Workspace mutations: edited review_proofs.py",
) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203))
res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
self.assertEqual(res2["verdict"], "complete")
@@ -1262,6 +1293,89 @@ class TestAssessReviewerQueueInventory(unittest.TestCase):
self.assertEqual(result["trust_gates"], {})
class TestAssessEmptyQueueReport(unittest.TestCase):
"""Issue #198: empty-queue reports require formal trust-gate proof."""
def _trusted_report(self, **extra):
lines = [
"Queue inventory complete.",
"Repository: Scaled-Tech-Consulting/Gitea-Tools",
"Open PR count: 0",
"pr_inventory_trust_gate.status: trusted_empty",
"pr_inventory_trust_gate.corroborated: true",
"Inventory profile: prgs-reviewer",
"Workflow correctly stops with nothing to review.",
]
lines.extend(extra)
return "\n".join(lines)
def test_non_empty_report_not_claimed(self):
result = assess_empty_queue_report("Reviewed PR #236 and merged.")
self.assertFalse(result["claimed"])
self.assertTrue(result["proven"])
def test_empty_claim_without_trust_gate_blocked(self):
report = (
"Open PR count: 0\n"
"Pagination complete: yes\n"
"Queue cleared."
)
result = assess_empty_queue_report(report)
self.assertTrue(result["claimed"])
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_trusted_empty_report_with_required_fields_passes(self):
result = assess_empty_queue_report(self._trusted_report())
self.assertTrue(result["proven"])
def test_weak_merge_commit_corroboration_blocked(self):
report = (
"Open PR count: 0\n"
"Master latest commit is merge of PR #79 so queue is empty."
)
result = assess_empty_queue_report(report)
self.assertFalse(result["proven"])
self.assertTrue(
any("weak corroboration" in r for r in result["reasons"])
)
def test_author_session_reviewer_queue_wording_blocked(self):
result = assess_empty_queue_report(
self._trusted_report(),
task_role="author",
)
self.assertFalse(result["proven"])
def test_build_final_report_downgrades_weak_empty_queue(self):
final = build_final_report(
checkout_proof=_good_checkout(),
inventory=_good_inventory(),
validation=_good_validation(),
contamination=_good_contamination(),
identity_eligible=True,
merge_performed=False,
issue_status_verified=True,
capability_evidence=_good_capability_evidence(),
sweep=_good_sweep(),
live_state=_good_live_state(),
role_boundary=_good_role_boundary(),
review_mutation=_good_review_mutation(),
controller_handoff=_good_handoff(),
capability_proof=_good_capability_proof(),
sweep_proof=_good_secret_sweep(),
worktree_proof={
"worktree_path": "/repo/branches/review-pr-1",
"porcelain_status": "",
"scratch_used": True,
"scratch_path": "/repo/branches/review-pr-1",
},
report_text="Open PR count: 0. Queue cleared.",
)
self.assertNotEqual(final["grade"], "A")
self.assertFalse(final["empty_queue_trust_gate_proven"])
class TestCapabilityEvidence(unittest.TestCase):
"""#179 gap 1: capability claims need exact evidence."""
@@ -1556,5 +1670,120 @@ class TestAuthorReporting(unittest.TestCase):
self.assertFalse(result["complete"])
class TestIssueSelectionContinuation(unittest.TestCase):
"""Issue #188: continuation mode wall for issues with open PRs."""
OLD_SHA = PINNED
NEW_SHA = OTHER
OPEN_PR = [{"number": 187, "head": {"ref": "feat/issue-183-harden-author-run-reporting"}}]
def test_open_pr_issue_excluded_from_fresh_selection(self):
classified = classify_issue_for_selection(
183, open_prs=self.OPEN_PR,
)
self.assertEqual(classified["status"], ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR)
self.assertFalse(classified["selectable_for_fresh_work"])
blocked = assess_fresh_issue_selection([classified])
self.assertTrue(blocked["downgraded"])
def test_explicit_continuation_allows_represented_issue(self):
classified = classify_issue_for_selection(
183,
open_prs=self.OPEN_PR,
operator_continuation_requested=True,
)
self.assertEqual(classified["status"], ISSUE_SELECTION_CONTINUATION_EXPLICIT)
blocked = assess_fresh_issue_selection([classified])
self.assertFalse(blocked["downgraded"])
def test_contradictory_no_pr_claim_downgrades(self):
report = (
"Selected issue #183; no duplicate PR open. "
"Updated PR #187 on branch feat/issue-183-harden-author-run-reporting."
)
result = assess_contradictory_no_pr_claim(
report, edited_pr_numbers=[187], issue_open_pr_map={183: 187},
)
self.assertTrue(result["downgraded"])
def test_edited_pr_must_appear_in_inventory(self):
report = "Open PR inventory: PR #195 only."
result = assess_edited_pr_inventory_coverage(
report,
edited_pr_numbers=[187],
inventoried_pr_numbers=[195],
)
self.assertTrue(result["downgraded"])
def test_continuation_report_requires_old_and_new_head(self):
report = (
"Issue #182 continuation mode. PR #186. "
f"old head {self.OLD_SHA} -> new head {self.NEW_SHA}. "
"PR author: jcwalker3. Branch: feat/issue-182-controller-handoff. "
"Session authored PR: yes. Continuation allowed: operator requested."
)
result = assess_continuation_mode_report(
report,
pr_number=186,
pr_author="jcwalker3",
branch="feat/issue-182-controller-handoff",
old_head_sha=self.OLD_SHA,
new_head_sha=self.NEW_SHA,
session_authored_pr=True,
continuation_allowed_reason="operator requested continuation",
)
self.assertTrue(result["complete"])
def test_issue_selection_final_report_continuation_earns_a(self):
report = "\n".join([
"Issue #182 continuation mode — no new issue claimed.",
f"PR #186 updated: old head {self.OLD_SHA}, "
f"new head {self.NEW_SHA}.",
"PR author: jcwalker3. Branch: feat/issue-182-controller-handoff.",
"Session authored PR: yes.",
"Continuation allowed: operator requested rebase.",
"Open PR inventory included PR #186.",
"## Controller Handoff",
"- Task: continuation",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: author",
"- Identity: prgs-author",
"- Issue/PR: #182 / PR #186",
"- Branch/SHA: feat/issue-182-controller-handoff",
"- Files changed: review_proofs.py",
"- Validation: tests passed",
"- Mutations: push_branch",
"- Workspace mutations: none",
"- Current status: PR mergeable",
"- Blockers: none",
"- Next: review",
"- Safety: no review/merge",
"- Continuation mode: issue #182 continuation",
"- Existing PR: #186",
"- PR author: jcwalker3",
"- Branch: feat/issue-182-controller-handoff",
f"- Old PR head: {self.OLD_SHA}",
f"- New PR head: {self.NEW_SHA}",
"- Session authored PR: yes",
"- Why continuation allowed: operator requested rebase",
])
result = assess_issue_selection_final_report(
report,
mode="continuation",
continuation_proof={
"pr_number": 186,
"pr_author": "jcwalker3",
"branch": "feat/issue-182-controller-handoff",
"old_head_sha": self.OLD_SHA,
"new_head_sha": self.NEW_SHA,
"session_authored_pr": True,
"continuation_allowed_reason": "operator requested rebase",
},
edited_pr_numbers=[186],
inventoried_pr_numbers=[186],
)
self.assertEqual(result["grade"], "A")
if __name__ == "__main__":
unittest.main()