Compare commits

...
Author SHA1 Message Date
sysadmin 054628c50d Merge pull request 'feat(continuation): formalize continuation proofs for claim and push (closes #189)' (#253) from feat/issue-189-continuation-mode-proofs into feat/issue-188-continuation-selection-wall 2026-07-06 11:27:25 -05:00
sysadmin 1158594a20 feat(continuation): formalize continuation proofs for claim and push (closes #189)
Extend #188 continuation mode with issue claim status evidence,
force-with-lease push reporting, canonical secret sweep helper, and
composite final-report checks wired into assess_issue_selection_final_report.
2026-07-06 11:54:40 -04:00
3 changed files with 264 additions and 6 deletions
+117 -1
View File
@@ -1046,6 +1046,8 @@ HANDOFF_ROLE_FIELDS = {
("Continuation mode", ("continuation mode", "continuation")), ("Continuation mode", ("continuation mode", "continuation")),
("Existing PR", ("existing pr", "pr number")), ("Existing PR", ("existing pr", "pr number")),
("PR author", ("pr author", "existing pr author")), ("PR author", ("pr author", "existing pr author")),
("Issue claim status", ("issue claim", "claim status",
"status:in-progress")),
("Branch", ("branch", "existing branch")), ("Branch", ("branch", "existing branch")),
("Old PR head", ("old pr head", "old head")), ("Old PR head", ("old pr head", "old head")),
("New PR head", ("new pr head", "new head")), ("New PR head", ("new pr head", "new head")),
@@ -1054,6 +1056,13 @@ HANDOFF_ROLE_FIELDS = {
), ),
} }
# Canonical secret/provenance sweep for comparable continuation runs (#189).
CANONICAL_SECRET_SWEEP_COMMAND = (
"git diff prgs/master...HEAD | rg -i "
"'(token|password|secret|api[_-]?key|authorization:)'"
)
CANONICAL_SECRET_SWEEP_SCOPE = "full feature-branch diff against prgs/master"
def _handoff_section_lines(report_text): def _handoff_section_lines(report_text):
"""Return the lines of the Controller Handoff section, or None.""" """Return the lines of the Controller Handoff section, or None."""
@@ -1657,18 +1666,89 @@ def assess_fresh_issue_selection(classifications: list[dict] | None) -> dict:
} }
def canonical_secret_sweep_report(*, clean: bool) -> dict:
"""Return a sweep report using the canonical command/scope (#189)."""
return {
"method": CANONICAL_SECRET_SWEEP_COMMAND,
"scope": CANONICAL_SECRET_SWEEP_SCOPE,
"clean": clean,
}
def assess_force_with_lease_push_report(
report_text: str,
*,
old_remote_head: str | None = None,
expected_lease_head: str | None = None,
new_pushed_head: str | None = None,
branch_pushed: str | None = None,
used_force_with_lease: bool | None = None,
) -> dict:
"""Issue #189: force-with-lease pushes must disclose lease evidence."""
if not used_force_with_lease:
return {"complete": True, "downgraded": False, "reasons": []}
text = report_text or ""
lower = text.lower()
reasons = []
if "force-with-lease" not in lower and "force with lease" not in lower:
reasons.append(
"continuation push used force-with-lease but report does not say so"
)
for label, sha in (
("old remote", old_remote_head),
("lease", expected_lease_head),
("pushed", new_pushed_head),
):
if not sha:
reasons.append(f"force-with-lease proof missing {label} head SHA")
elif not _FULL_SHA.match(sha.lower()):
reasons.append(
f"force-with-lease {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 branch_pushed:
if branch_pushed.lower() not in lower:
reasons.append("continuation report missing pushed branch name")
only_branch_tokens = (
"only feature branch",
"push branch only",
"only the feature branch",
"only pushed feature branch",
)
if not any(t in lower for t in only_branch_tokens):
reasons.append(
"continuation report missing confirmation that only the "
"feature branch was pushed"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def assess_continuation_mode_report( def assess_continuation_mode_report(
report_text: str, report_text: str,
*, *,
pr_number: int | None = None, pr_number: int | None = None,
pr_author: str | None = None, pr_author: str | None = None,
issue_number: int | None = None,
issue_claim_status: str | None = None,
branch: str | None = None, branch: str | None = None,
old_head_sha: str | None = None, old_head_sha: str | None = None,
new_head_sha: str | None = None, new_head_sha: str | None = None,
session_authored_pr: bool | None = None, session_authored_pr: bool | None = None,
continuation_allowed_reason: str | None = None, continuation_allowed_reason: str | None = None,
) -> dict: ) -> dict:
"""Issue #188: continuation mode must disclose full PR evidence.""" """Issue #188/#189: continuation mode must disclose full PR evidence."""
text = report_text or "" text = report_text or ""
lower = text.lower() lower = text.lower()
reasons = [] reasons = []
@@ -1681,6 +1761,23 @@ def assess_continuation_mode_report(
reasons.append(f"continuation report missing PR #{pr_number}") reasons.append(f"continuation report missing PR #{pr_number}")
if pr_author and pr_author.lower() not in lower: if pr_author and pr_author.lower() not in lower:
reasons.append("continuation report missing PR author") reasons.append("continuation report missing PR author")
if issue_number is not None:
if (
f"#{issue_number}" not in lower
and f"issue #{issue_number}" not in lower
and f"issue {issue_number}" not in lower
):
reasons.append(f"continuation report missing issue #{issue_number}")
if issue_claim_status:
claim_lower = issue_claim_status.lower()
claim_tokens = (
claim_lower,
"claim status",
"issue claim",
"status:in-progress",
)
if not any(t in lower for t in claim_tokens):
reasons.append("continuation report missing issue claim status")
if branch and branch.lower() not in lower: if branch and branch.lower() not in lower:
reasons.append("continuation report missing branch name") reasons.append("continuation report missing branch name")
@@ -1817,12 +1914,26 @@ def assess_issue_selection_final_report(
report_text, report_text,
pr_number=proof.get("pr_number"), pr_number=proof.get("pr_number"),
pr_author=proof.get("pr_author"), pr_author=proof.get("pr_author"),
issue_number=proof.get("issue_number"),
issue_claim_status=proof.get("issue_claim_status"),
branch=proof.get("branch"), branch=proof.get("branch"),
old_head_sha=proof.get("old_head_sha"), old_head_sha=proof.get("old_head_sha"),
new_head_sha=proof.get("new_head_sha"), new_head_sha=proof.get("new_head_sha"),
session_authored_pr=proof.get("session_authored_pr"), session_authored_pr=proof.get("session_authored_pr"),
continuation_allowed_reason=proof.get("continuation_allowed_reason"), continuation_allowed_reason=proof.get("continuation_allowed_reason"),
) )
push_proof = proof.get("push_proof") or {}
checks["force_with_lease_push"] = assess_force_with_lease_push_report(
report_text,
old_remote_head=push_proof.get("old_remote_head"),
expected_lease_head=push_proof.get("expected_lease_head"),
new_pushed_head=push_proof.get("new_pushed_head"),
branch_pushed=push_proof.get("branch_pushed"),
used_force_with_lease=push_proof.get("used_force_with_lease"),
)
sweep = proof.get("secret_sweep")
if sweep is not None:
checks["secret_sweep"] = assess_secret_sweep(sweep)
reasons = [] reasons = []
downgraded = False downgraded = False
@@ -1831,6 +1942,11 @@ def assess_issue_selection_final_report(
if verdict in ("missing", "incomplete"): if verdict in ("missing", "incomplete"):
downgraded = True downgraded = True
reasons.extend(result.get("reasons") or []) reasons.extend(result.get("reasons") or [])
elif result.get("proven") is False:
downgraded = True
reasons.extend(
f"{name}: {r}" for r in (result.get("reasons") or [])
)
elif result.get("downgraded") or not result.get("complete", True): elif result.get("downgraded") or not result.get("complete", True):
downgraded = True downgraded = True
reasons.extend( reasons.extend(
+10 -5
View File
@@ -379,11 +379,16 @@ Role-specific fields (append to the compact block):
`Linked issue status:`, `Cleanup status:` `Linked issue status:`, `Cleanup status:`
- author tasks: `Selected issue:`, `Claim/comment status:`, - author tasks: `Selected issue:`, `Claim/comment status:`,
`PR number opened:`, `No review/merge:` (explicit confirmation) `PR number opened:`, `No review/merge:` (explicit confirmation)
- continuation tasks (#188): `Continuation mode:`, `Existing PR:`, - continuation tasks (#188/#189): `Continuation mode:`, `Existing PR:`,
`PR author:`, `Branch:`, `Old PR head:`, `New PR head:`, `PR author:`, `Issue claim status:`, `Branch:`, `Old PR head:`,
`Session authored PR:`, `Why continuation allowed:` issues with open `New PR head:`, `Session authored PR:`, `Why continuation allowed:`
PRs are excluded from fresh selection unless operator explicitly requests when rebasing with `git push --force-with-lease`, also record
continuation (`review_proofs.classify_issue_for_selection`, `Old remote head:`, `Lease head:`, `Pushed head:`, and confirm
`Push branch only:` (feature branch only). Use the canonical secret sweep
from `review_proofs.CANONICAL_SECRET_SWEEP_COMMAND` so runs are
comparable. Issues with open PRs are excluded from fresh selection unless
the operator explicitly requests continuation
(`review_proofs.classify_issue_for_selection`,
`assess_issue_selection_final_report`) `assess_issue_selection_final_report`)
- queue/inventory tasks: `Repositories checked:`, `Open PR counts:`, - queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
`Selected PR or reason none selected:`, `Inventory completeness:` `Selected PR or reason none selected:`, `Inventory completeness:`
+137
View File
@@ -28,6 +28,9 @@ from review_proofs import ( # noqa: E402
assess_capability_proof, assess_capability_proof,
assess_contradictory_no_pr_claim, assess_contradictory_no_pr_claim,
assess_continuation_mode_report, assess_continuation_mode_report,
assess_force_with_lease_push_report,
canonical_secret_sweep_report,
CANONICAL_SECRET_SWEEP_COMMAND,
assess_controller_handoff, assess_controller_handoff,
assess_edited_pr_inventory_coverage, assess_edited_pr_inventory_coverage,
assess_empty_queue_report, assess_empty_queue_report,
@@ -1761,6 +1764,7 @@ class TestIssueSelectionContinuation(unittest.TestCase):
"- Continuation mode: issue #182 continuation", "- Continuation mode: issue #182 continuation",
"- Existing PR: #186", "- Existing PR: #186",
"- PR author: jcwalker3", "- PR author: jcwalker3",
"- Issue claim status: status:in-progress",
"- Branch: feat/issue-182-controller-handoff", "- Branch: feat/issue-182-controller-handoff",
f"- Old PR head: {self.OLD_SHA}", f"- Old PR head: {self.OLD_SHA}",
f"- New PR head: {self.NEW_SHA}", f"- New PR head: {self.NEW_SHA}",
@@ -1773,6 +1777,8 @@ class TestIssueSelectionContinuation(unittest.TestCase):
continuation_proof={ continuation_proof={
"pr_number": 186, "pr_number": 186,
"pr_author": "jcwalker3", "pr_author": "jcwalker3",
"issue_number": 182,
"issue_claim_status": "status:in-progress",
"branch": "feat/issue-182-controller-handoff", "branch": "feat/issue-182-controller-handoff",
"old_head_sha": self.OLD_SHA, "old_head_sha": self.OLD_SHA,
"new_head_sha": self.NEW_SHA, "new_head_sha": self.NEW_SHA,
@@ -1785,5 +1791,136 @@ class TestIssueSelectionContinuation(unittest.TestCase):
self.assertEqual(result["grade"], "A") self.assertEqual(result["grade"], "A")
class TestContinuationModeProofs(unittest.TestCase):
"""Issue #189: formal continuation proofs beyond #188 selection wall."""
OLD_SHA = PINNED
NEW_SHA = OTHER
REMOTE_OLD = "601c608c00000000000000000000000000000000"
LEASE_SHA = "45c5cac2bc49dd112766ea218b18a71e9c5f8e99"
PUSHED_SHA = "45c5cac2bc49dd112766ea218b18a71e9c5f8e99"
def test_force_with_lease_requires_lease_evidence(self):
report = (
"Issue #182 continuation. force-with-lease push to "
"feat/issue-182-controller-handoff-enforcement."
)
result = assess_force_with_lease_push_report(
report,
old_remote_head=self.REMOTE_OLD,
expected_lease_head=self.LEASE_SHA,
new_pushed_head=self.PUSHED_SHA,
branch_pushed="feat/issue-182-controller-handoff-enforcement",
used_force_with_lease=True,
)
self.assertTrue(result["downgraded"])
def test_force_with_lease_complete_with_full_evidence(self):
report = (
"Issue #182 continuation with git push --force-with-lease. "
f"Old remote head {self.REMOTE_OLD}. "
f"Lease head {self.LEASE_SHA}. "
f"Pushed head {self.PUSHED_SHA}. "
"Branch feat/issue-182-controller-handoff-enforcement. "
"Push branch only: only the feature branch was pushed."
)
result = assess_force_with_lease_push_report(
report,
old_remote_head=self.REMOTE_OLD,
expected_lease_head=self.LEASE_SHA,
new_pushed_head=self.PUSHED_SHA,
branch_pushed="feat/issue-182-controller-handoff-enforcement",
used_force_with_lease=True,
)
self.assertTrue(result["complete"])
def test_continuation_requires_issue_claim_status(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-test."
)
result = assess_continuation_mode_report(
report,
pr_number=186,
pr_author="jcwalker3",
issue_number=182,
issue_claim_status="status:in-progress",
branch="feat/issue-182-test",
old_head_sha=self.OLD_SHA,
new_head_sha=self.NEW_SHA,
)
self.assertTrue(result["downgraded"])
def test_canonical_secret_sweep_helper(self):
sweep = canonical_secret_sweep_report(clean=True)
self.assertEqual(sweep["method"], CANONICAL_SECRET_SWEEP_COMMAND)
self.assertTrue(sweep["clean"])
def test_continuation_final_report_with_push_and_sweep_earns_a(self):
branch = "feat/issue-182-controller-handoff-enforcement"
report = "\n".join([
"Issue #182 continuation mode — status:in-progress already set.",
f"PR #186 updated via git push --force-with-lease.",
f"Old PR head {self.OLD_SHA}; new PR head {self.NEW_SHA}.",
f"Old remote head {self.REMOTE_OLD}.",
f"Lease head {self.LEASE_SHA}. Pushed head {self.PUSHED_SHA}.",
f"Branch {branch}. Push branch only: only the feature branch.",
"PR author: jcwalker3. 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",
f"- Branch/SHA: {branch}",
"- 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",
"- Issue claim status: status:in-progress",
f"- Branch: {branch}",
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",
"issue_number": 182,
"issue_claim_status": "status:in-progress",
"branch": branch,
"old_head_sha": self.OLD_SHA,
"new_head_sha": self.NEW_SHA,
"session_authored_pr": True,
"continuation_allowed_reason": "operator requested rebase",
"push_proof": {
"old_remote_head": self.REMOTE_OLD,
"expected_lease_head": self.LEASE_SHA,
"new_pushed_head": self.PUSHED_SHA,
"branch_pushed": branch,
"used_force_with_lease": True,
},
"secret_sweep": canonical_secret_sweep_report(clean=True),
},
edited_pr_numbers=[186],
inventoried_pr_numbers=[186],
)
self.assertEqual(result["grade"], "A")
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()