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 
Issues already represented by open PRs must not be selected for fresh work
unless the operator explicitly requests continuation. Adds classification,
continuation report proofs (old/new head SHA, PR author, branch), contradictory
no-PR claim detection, and edited-PR inventory coverage checks.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-06 11:40:20 -04:00
co-authored by Claude Opus 4.8
parent bab803ff3d
commit 63a7ba8287
3 changed files with 413 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
review_proofs.py
+284
View File
@@ -1042,6 +1042,16 @@ HANDOFF_ROLE_FIELDS = {
("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")),
),
}
@@ -1562,6 +1572,280 @@ def assess_empty_queue_report(
}
# ── 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
+123
View File
@@ -21,12 +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,
@@ -36,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,
@@ -1662,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()