diff --git a/scripts/audit_gitea_coordination.py b/scripts/audit_gitea_coordination.py index b62689e..653435c 100644 --- a/scripts/audit_gitea_coordination.py +++ b/scripts/audit_gitea_coordination.py @@ -243,13 +243,25 @@ def validate_pull_request( pr: dict[str, Any], task: RemoteTask ) -> list[AuditFinding]: findings: list[AuditFinding] = [] - if not task.work_branch or pull_request_head(pr) != task.work_branch: + body = pr.get("body") + body = body if isinstance(body, str) else "" + head = pull_request_head(pr) + head_matches = bool(task.work_branch) and head == task.work_branch + if ( + not head_matches + and bool(pr.get("merged")) + and re.fullmatch(r"refs/pull/\d+/head", head) + and bool(task.work_branch) + and task.work_branch in body + ): + # Gitea rewrites head.ref after a merged source branch is deleted. The + # immutable PR body still records the claimed work branch. + head_matches = True + if not head_matches: findings.append( AuditFinding(task.number, "pull-request", "PR head 与 claim 工作分支不一致。") ) - body = pr.get("body") - body = body if isinstance(body, str) else "" - required_values = [f"docs/tasks/{task.task_id}.md"] + required_values = [f"docs/tasks/{task.task_id}.md", task.work_branch or ""] if task.context_ref: required_values.append(task.context_ref) required_values.extend(task.write_paths or []) diff --git a/tests/test_governance.py b/tests/test_governance.py index 3cdb9fe..e5141e8 100644 --- a/tests/test_governance.py +++ b/tests/test_governance.py @@ -23,6 +23,8 @@ from audit_gitea_coordination import ( parse_datetime, parse_write_paths, select_latest_claim, + RemoteTask, + validate_pull_request, ) from setup_gitea_labels import LABELS, NoRedirect, build_plan from test_gitea_claim_race import PROBE_PREFIX, new_probe_branch @@ -180,6 +182,31 @@ class OfflineRuleTests(unittest.TestCase): class GiteaHelperTests(unittest.TestCase): + def test_merged_pr_accepts_gitea_deleted_branch_ref(self) -> None: + sha = "a" * 40 + task = RemoteTask( + number=1, + task_id="T-123", + state="closed", + status="status/done", + labels={"status/done"}, + body="", + work_branch="agent/worker-1/T-123", + context_ref=sha, + write_paths=["docs/tasks/T-123.md"], + ) + pr = { + "merged": True, + "head": {"ref": "refs/pull/1/head"}, + "body": ( + "Closes #1\n" + "task_file: docs/tasks/T-123.md\n" + f"context_ref: {sha}\n" + "work_branch: agent/worker-1/T-123\n" + ), + } + self.assertEqual([], validate_pull_request(pr, task)) + def test_waiting_status_is_not_claimable_or_active(self) -> None: self.assertIn("status/waiting", STATUS_LABELS) self.assertNotIn("status/waiting", READY_OR_ACTIVE_LABELS)