689 lines
28 KiB
Python
689 lines
28 KiB
Python
#!/usr/bin/env python3
|
||
"""Read-only audit of Harness Coding task coordination in Gitea."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import os
|
||
import re
|
||
import sys
|
||
import urllib.parse
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from setup_gitea_labels import ApiError, GiteaClient, LABELS, validate_config
|
||
from validate_harness_governance import (
|
||
TASK_ID,
|
||
is_safe_repo_path,
|
||
parse_frontmatter,
|
||
parse_frontmatter_text,
|
||
scopes_overlap,
|
||
)
|
||
|
||
|
||
STATUS_LABELS = {
|
||
"status/todo",
|
||
"status/doing",
|
||
"status/blocked",
|
||
"status/review",
|
||
"status/done",
|
||
}
|
||
ACTIVE_LABELS = {"status/doing", "status/blocked", "status/review"}
|
||
READY_OR_ACTIVE_LABELS = ACTIVE_LABELS | {"status/todo"}
|
||
TASK_IN_TITLE = re.compile(r"^\[(T-\d{3}[a-z]?)\]")
|
||
BODY_TASK_ID = re.compile(r"(?m)^\s*-\s*task_id:\s*`?(T-\d{3}[a-z]?)`?\s*$")
|
||
BODY_TASK_FILE = re.compile(
|
||
r"(?m)^\s*-\s*task_file:\s*`?(docs/tasks/T-\d{3}[a-z]?\.md)`?\s*$"
|
||
)
|
||
FIELD = re.compile(r"(?m)^\s*(?:-\s*)?([a-z_]+):\s*`?([^`\r\n]+?)`?\s*$")
|
||
MAX_LEASE = timedelta(hours=24)
|
||
CLOCK_SKEW = timedelta(minutes=5)
|
||
CLAIM_IDENTITY_FIELDS = (
|
||
"task",
|
||
"claimed_by",
|
||
"allocated_by",
|
||
"context_ref",
|
||
"claim_branch",
|
||
"work_branch",
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True, order=True)
|
||
class AuditFinding:
|
||
issue: int
|
||
rule: str
|
||
message: str
|
||
|
||
def render(self) -> str:
|
||
subject = "repository" if self.issue <= 0 else f"issue #{self.issue}"
|
||
return f"ERROR [{self.rule}] {subject}: {self.message}"
|
||
|
||
|
||
@dataclass
|
||
class RemoteTask:
|
||
number: int
|
||
task_id: str
|
||
state: str
|
||
status: str | None
|
||
labels: set[str]
|
||
body: str
|
||
work_branch: str | None = None
|
||
context_ref: str | None = None
|
||
claimed_by: str | None = None
|
||
claimed_at: datetime | None = None
|
||
lease_until: datetime | None = None
|
||
write_paths: list[str] | None = None
|
||
|
||
|
||
def label_names(item: dict[str, Any]) -> set[str]:
|
||
labels = item.get("labels")
|
||
if not isinstance(labels, list):
|
||
return set()
|
||
return {
|
||
label["name"]
|
||
for label in labels
|
||
if isinstance(label, dict) and isinstance(label.get("name"), str)
|
||
}
|
||
|
||
|
||
def paged(client: GiteaClient, path: str) -> list[dict[str, Any]]:
|
||
result: list[dict[str, Any]] = []
|
||
seen_pages: set[tuple[str, ...]] = set()
|
||
page = 1
|
||
separator = "&" if "?" in path else "?"
|
||
while True:
|
||
values = client.request("GET", f"{path}{separator}limit=50&page={page}")
|
||
if not isinstance(values, list):
|
||
raise RuntimeError("Gitea 分页响应格式异常。")
|
||
if not values:
|
||
return result
|
||
if not all(isinstance(value, dict) for value in values):
|
||
raise RuntimeError("Gitea 分页响应包含非对象条目。")
|
||
signature = tuple(
|
||
str(value.get("id") or value.get("number") or value.get("name"))
|
||
for value in values
|
||
)
|
||
if signature in seen_pages or page > 1000:
|
||
raise RuntimeError("Gitea 分页重复，已停止以避免无限读取。")
|
||
seen_pages.add(signature)
|
||
result.extend(value for value in values if isinstance(value, dict))
|
||
page += 1
|
||
|
||
|
||
def parse_fields(text: str) -> dict[str, str]:
|
||
return {match.group(1): match.group(2).strip() for match in FIELD.finditer(text)}
|
||
|
||
|
||
def parse_write_paths(text: str) -> list[str]:
|
||
lines = text.splitlines()
|
||
values: list[str] = []
|
||
collecting = False
|
||
for line in lines:
|
||
if re.match(r"^\s*(?:-\s*)?write_paths:\s*$", line):
|
||
collecting = True
|
||
continue
|
||
if collecting:
|
||
item = re.match(r"^\s+-\s+`?([^`\r\n]+?)`?\s*$", line)
|
||
if item:
|
||
value = item.group(1).strip()
|
||
if "【" not in value:
|
||
values.append(value)
|
||
continue
|
||
if line.strip():
|
||
break
|
||
return values
|
||
|
||
|
||
def parse_datetime(value: str | None) -> datetime | None:
|
||
if not value or "【" in value:
|
||
return None
|
||
if not re.fullmatch(
|
||
r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})",
|
||
value,
|
||
):
|
||
return None
|
||
try:
|
||
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||
except ValueError:
|
||
return None
|
||
if parsed.tzinfo is None:
|
||
parsed = parsed.replace(tzinfo=timezone.utc)
|
||
return parsed.astimezone(timezone.utc)
|
||
|
||
|
||
def select_latest_claim(
|
||
comments: list[dict[str, Any]], task_id: str, dispatcher_login: str | None
|
||
) -> tuple[str | None, list[str]]:
|
||
required = {
|
||
"task",
|
||
"claimed_by",
|
||
"allocated_by",
|
||
"context_ref",
|
||
"claim_branch",
|
||
"work_branch",
|
||
"claimed_at",
|
||
"lease_until",
|
||
}
|
||
if not dispatcher_login:
|
||
return None, ["未配置可信 dispatcher Gitea 登录名，无法验证 CLAIM 作者。"]
|
||
|
||
claims: list[tuple[int, str, dict[str, str], str, str]] = []
|
||
for index, comment in enumerate(comments):
|
||
body = comment.get("body")
|
||
if not isinstance(body, str):
|
||
continue
|
||
lines = body.strip().splitlines()
|
||
if not lines or lines[0].strip() not in {"CLAIM", "CLAIM RENEWAL"}:
|
||
continue
|
||
marker = lines[0].strip()
|
||
fields = parse_fields(body)
|
||
if fields.get("task") != task_id or not required.issubset(fields):
|
||
continue
|
||
if not parse_write_paths(body):
|
||
continue
|
||
user = comment.get("user")
|
||
author = str(user.get("login") or "") if isinstance(user, dict) else ""
|
||
comment_id = comment.get("id")
|
||
order = comment_id if isinstance(comment_id, int) else index
|
||
claims.append((order, marker, fields, author, body))
|
||
|
||
selected: str | None = None
|
||
identity: tuple[str, ...] | None = None
|
||
errors: list[str] = []
|
||
for _, marker, fields, author, body in sorted(claims, key=lambda item: item[0]):
|
||
if author != dispatcher_login or fields.get("allocated_by") != dispatcher_login:
|
||
errors.append("CLAIM 必须由配置的 dispatcher 账号发布，且 allocated_by 与作者一致。")
|
||
continue
|
||
candidate_identity = tuple(fields.get(key, "") for key in CLAIM_IDENTITY_FIELDS)
|
||
if marker == "CLAIM":
|
||
if identity is not None:
|
||
errors.append("同一任务存在重复的初始 CLAIM。")
|
||
continue
|
||
identity = candidate_identity
|
||
selected = body
|
||
continue
|
||
if identity is None:
|
||
errors.append("CLAIM RENEWAL 之前缺少有效初始 CLAIM。")
|
||
continue
|
||
if candidate_identity != identity:
|
||
errors.append("CLAIM RENEWAL 改变了任务、领取者、dispatcher 或分支身份字段。")
|
||
continue
|
||
selected = body
|
||
return selected, sorted(set(errors))
|
||
|
||
|
||
def latest_claim(
|
||
comments: list[dict[str, Any]], task_id: str, dispatcher_login: str | None
|
||
) -> str | None:
|
||
return select_latest_claim(comments, task_id, dispatcher_login)[0]
|
||
|
||
|
||
def pull_request_head(pr: dict[str, Any]) -> str:
|
||
head = pr.get("head")
|
||
return str(head.get("ref") or "") if isinstance(head, dict) else ""
|
||
|
||
|
||
def branch_commits(branches: list[dict[str, Any]]) -> dict[str, str]:
|
||
result: dict[str, str] = {}
|
||
for branch in branches:
|
||
name = branch.get("name")
|
||
commit = branch.get("commit")
|
||
if isinstance(name, str) and isinstance(commit, dict) and isinstance(
|
||
commit.get("id"), str
|
||
):
|
||
result[name] = commit["id"]
|
||
return result
|
||
|
||
|
||
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:
|
||
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"]
|
||
if task.context_ref:
|
||
required_values.append(task.context_ref)
|
||
required_values.extend(task.write_paths or [])
|
||
if not re.search(rf"(?i)\bCloses\s+#{task.number}\b", body):
|
||
findings.append(AuditFinding(task.number, "pull-request", "PR body 未链接对应 Issue。"))
|
||
if any(value not in body for value in required_values):
|
||
findings.append(
|
||
AuditFinding(task.number, "pull-request", "PR body 缺少任务、context_ref 或写路径。")
|
||
)
|
||
return findings
|
||
|
||
|
||
def remote_task_metadata(
|
||
client: GiteaClient, task_id: str, branch: str
|
||
) -> tuple[dict[str, Any] | None, list[str]]:
|
||
file_path = urllib.parse.quote(f"docs/tasks/{task_id}.md", safe="/")
|
||
ref = urllib.parse.quote(branch, safe="")
|
||
try:
|
||
response = client.request("GET", f"/contents/{file_path}?ref={ref}")
|
||
except ApiError as exc:
|
||
if exc.status == 404:
|
||
return None, ["工作分支缺少任务文件。"]
|
||
raise
|
||
if not isinstance(response, dict) or not isinstance(response.get("content"), str):
|
||
return None, ["工作分支任务文件响应格式异常。"]
|
||
try:
|
||
text = base64.b64decode(response["content"]).decode("utf-8")
|
||
except (ValueError, UnicodeDecodeError):
|
||
return None, ["工作分支任务文件不是有效 UTF-8 / base64。"]
|
||
metadata, _, errors = parse_frontmatter_text(text)
|
||
return metadata, errors
|
||
|
||
|
||
def local_tasks(root: Path) -> dict[str, dict[str, Any]]:
|
||
result: dict[str, dict[str, Any]] = {}
|
||
task_dir = root / "docs" / "tasks"
|
||
if not task_dir.is_dir():
|
||
return result
|
||
for path in sorted(task_dir.glob("T-*.md")):
|
||
metadata, _, _ = parse_frontmatter(path)
|
||
task_id = metadata.get("id")
|
||
if isinstance(task_id, str) and TASK_ID.fullmatch(task_id):
|
||
result[task_id] = metadata
|
||
return result
|
||
|
||
|
||
def audit_labels(client: GiteaClient) -> list[AuditFinding]:
|
||
findings: list[AuditFinding] = []
|
||
existing = client.list_labels()
|
||
for desired in LABELS:
|
||
current = existing.get(desired["name"])
|
||
if current is None:
|
||
findings.append(AuditFinding(0, "labels", f"缺少 {desired['name']}。"))
|
||
elif bool(current.get("exclusive")) != desired["exclusive"]:
|
||
findings.append(AuditFinding(0, "labels", f"{desired['name']} exclusive 属性不一致。"))
|
||
return findings
|
||
|
||
|
||
def audit_repository(
|
||
root: Path,
|
||
client: GiteaClient,
|
||
now: datetime,
|
||
dispatcher_login: str | None = None,
|
||
) -> tuple[list[AuditFinding], int]:
|
||
findings = audit_labels(client)
|
||
issues = [
|
||
issue
|
||
for issue in paged(client, "/issues?state=all&type=issues")
|
||
if "kind/task" in label_names(issue) and not issue.get("pull_request")
|
||
]
|
||
branches = branch_commits(paged(client, "/branches"))
|
||
pull_requests = paged(client, "/pulls?state=all")
|
||
local = local_tasks(root)
|
||
remote: dict[str, RemoteTask] = {}
|
||
|
||
for issue in issues:
|
||
number = issue.get("number")
|
||
title = issue.get("title")
|
||
body = issue.get("body")
|
||
state = issue.get("state")
|
||
if not isinstance(number, int) or not isinstance(title, str):
|
||
continue
|
||
body = body if isinstance(body, str) else ""
|
||
title_match = TASK_IN_TITLE.search(title)
|
||
body_match = BODY_TASK_ID.search(body)
|
||
task_id = title_match.group(1) if title_match else ""
|
||
if not task_id:
|
||
findings.append(AuditFinding(number, "mapping", "标题缺少 [T-编号]。"))
|
||
if body_match is None or body_match.group(1) != task_id:
|
||
findings.append(AuditFinding(number, "mapping", "正文 task_id 与标题不一致。"))
|
||
task_file_match = BODY_TASK_FILE.search(body)
|
||
if task_id and (
|
||
task_file_match is None
|
||
or task_file_match.group(1) != f"docs/tasks/{task_id}.md"
|
||
):
|
||
findings.append(AuditFinding(number, "mapping", "task_file 与任务 ID 不一致。"))
|
||
if task_id in remote:
|
||
findings.append(AuditFinding(number, "mapping", "任务 ID 映射到多个 Issue。"))
|
||
|
||
labels = label_names(issue)
|
||
statuses = sorted(labels & STATUS_LABELS)
|
||
status = statuses[0] if len(statuses) == 1 else None
|
||
local_issue = local.get(task_id, {}).get("issue") if task_id else None
|
||
if len(statuses) > 1:
|
||
findings.append(AuditFinding(number, "status", "存在多个 status/* 标签。"))
|
||
elif not statuses and local_issue == number:
|
||
findings.append(AuditFinding(number, "status", "已映射任务缺少 status/* 标签。"))
|
||
if status:
|
||
if task_id not in local or local_issue != number:
|
||
findings.append(AuditFinding(number, "mapping", "可领取 Issue 未映射默认分支任务文件。"))
|
||
else:
|
||
local_status = local[task_id].get("status")
|
||
expected_local = "DONE" if status == "status/done" else "TODO"
|
||
if local_status != expected_local:
|
||
findings.append(
|
||
AuditFinding(
|
||
number,
|
||
"status",
|
||
f"远端 {status} 要求默认分支任务为 {expected_local}。",
|
||
)
|
||
)
|
||
if status in READY_OR_ACTIVE_LABELS:
|
||
deps = local[task_id].get("deps")
|
||
if not isinstance(deps, list):
|
||
findings.append(
|
||
AuditFinding(number, "dependency", "默认分支任务 deps 不是列表。")
|
||
)
|
||
else:
|
||
unready = sorted(
|
||
str(dep)
|
||
for dep in deps
|
||
if not isinstance(dep, str)
|
||
or dep not in local
|
||
or local[dep].get("status") != "DONE"
|
||
)
|
||
if unready:
|
||
findings.append(
|
||
AuditFinding(
|
||
number,
|
||
"dependency",
|
||
"任务依赖尚未全部 DONE：" + ", ".join(unready) + "。",
|
||
)
|
||
)
|
||
for prefix in ("type/", "priority/"):
|
||
scoped = [name for name in labels if name.startswith(prefix)]
|
||
if len(scoped) != 1:
|
||
findings.append(
|
||
AuditFinding(number, "labels", f"可领取任务必须恰有一个 {prefix} 标签。")
|
||
)
|
||
if status == "status/done" and state != "closed":
|
||
findings.append(AuditFinding(number, "status", "status/done 的 Issue 必须关闭。"))
|
||
if status != "status/done" and state == "closed":
|
||
findings.append(AuditFinding(number, "status", "未完成 Issue 不应关闭。"))
|
||
|
||
task = RemoteTask(
|
||
number=number,
|
||
task_id=task_id,
|
||
state=state if isinstance(state, str) else "",
|
||
status=status,
|
||
labels=labels,
|
||
body=body,
|
||
write_paths=parse_write_paths(body),
|
||
)
|
||
if task_id:
|
||
remote[task_id] = task
|
||
|
||
local_metadata = local.get(task_id, {})
|
||
if status == "status/done":
|
||
work_branch = local_metadata.get("work_branch")
|
||
context_ref = local_metadata.get("context_ref")
|
||
write_paths = local_metadata.get("write_paths")
|
||
task.work_branch = work_branch if isinstance(work_branch, str) else None
|
||
task.context_ref = context_ref if isinstance(context_ref, str) else None
|
||
task.write_paths = (
|
||
[value for value in write_paths if isinstance(value, str)]
|
||
if isinstance(write_paths, list)
|
||
else []
|
||
)
|
||
if not task.work_branch or not task.context_ref or not task.write_paths:
|
||
findings.append(
|
||
AuditFinding(number, "task-file", "DONE 任务缺少长期分支、context_ref 或写路径。")
|
||
)
|
||
|
||
claim_branch = f"claims/{task_id}" if task_id else ""
|
||
if status in ACTIVE_LABELS:
|
||
if claim_branch not in branches:
|
||
findings.append(AuditFinding(number, "claim", "活跃任务缺少 claim 分支。"))
|
||
comments = paged(client, f"/issues/{number}/comments")
|
||
claim, claim_errors = select_latest_claim(
|
||
comments, task_id, dispatcher_login
|
||
)
|
||
for message in claim_errors:
|
||
findings.append(AuditFinding(number, "claim-author", message))
|
||
if claim is None:
|
||
findings.append(
|
||
AuditFinding(number, "claim", "活跃任务缺少由可信 dispatcher 发布的结构化 CLAIM。")
|
||
)
|
||
else:
|
||
fields = parse_fields(claim)
|
||
task.work_branch = fields.get("work_branch")
|
||
task.context_ref = fields.get("context_ref")
|
||
task.claimed_by = fields.get("claimed_by")
|
||
task.claimed_at = parse_datetime(fields.get("claimed_at"))
|
||
task.lease_until = parse_datetime(fields.get("lease_until"))
|
||
claim_paths = parse_write_paths(claim)
|
||
if claim_paths:
|
||
task.write_paths = claim_paths
|
||
required_claim = {
|
||
"task",
|
||
"claimed_by",
|
||
"allocated_by",
|
||
"context_ref",
|
||
"claim_branch",
|
||
"work_branch",
|
||
"claimed_at",
|
||
"lease_until",
|
||
}
|
||
missing_claim = sorted(required_claim - set(fields))
|
||
if missing_claim:
|
||
findings.append(
|
||
AuditFinding(
|
||
number,
|
||
"claim",
|
||
"CLAIM 缺少字段：" + ", ".join(missing_claim) + "。",
|
||
)
|
||
)
|
||
if fields.get("task") != task_id:
|
||
findings.append(AuditFinding(number, "claim", "CLAIM task 不一致。"))
|
||
if fields.get("claim_branch") != claim_branch:
|
||
findings.append(AuditFinding(number, "claim", "CLAIM claim_branch 不一致。"))
|
||
context_ref = task.context_ref
|
||
if not context_ref or not re.fullmatch(r"[0-9a-fA-F]{40}", context_ref):
|
||
findings.append(AuditFinding(number, "claim", "CLAIM context_ref 无效。"))
|
||
elif branches.get(claim_branch) != context_ref:
|
||
findings.append(AuditFinding(number, "claim", "claim 分支 SHA 与 context_ref 不一致。"))
|
||
if (
|
||
not task.claimed_by
|
||
or not re.fullmatch(r"[A-Za-z0-9._-]+", task.claimed_by)
|
||
or task.work_branch != f"agent/{task.claimed_by}/{task_id}"
|
||
):
|
||
findings.append(AuditFinding(number, "claim", "claimed_by 与工作分支命名不一致。"))
|
||
if (
|
||
not claim_paths
|
||
or len(claim_paths) != len(set(claim_paths))
|
||
or any(not is_safe_repo_path(path) for path in claim_paths)
|
||
or f"docs/tasks/{task_id}.md" not in claim_paths
|
||
):
|
||
findings.append(
|
||
AuditFinding(
|
||
number,
|
||
"claim",
|
||
"CLAIM write_paths 必须安全、唯一并包含任务文件。",
|
||
)
|
||
)
|
||
if not task.work_branch or task.work_branch not in branches:
|
||
findings.append(AuditFinding(number, "claim", "工作分支不存在。"))
|
||
elif task.work_branch:
|
||
metadata, metadata_errors = remote_task_metadata(
|
||
client, task_id, task.work_branch
|
||
)
|
||
for _ in metadata_errors:
|
||
findings.append(AuditFinding(number, "task-file", "工作分支任务文件无效。"))
|
||
if metadata is not None:
|
||
expected_status = {
|
||
"status/doing": "DOING",
|
||
"status/blocked": "BLOCKED",
|
||
"status/review": "DONE",
|
||
}.get(status)
|
||
comparisons = {
|
||
"id": task_id,
|
||
"issue": number,
|
||
"context_ref": context_ref,
|
||
"claim_branch": claim_branch,
|
||
"work_branch": task.work_branch,
|
||
"status": expected_status,
|
||
}
|
||
for key, expected in comparisons.items():
|
||
if metadata.get(key) != expected:
|
||
findings.append(
|
||
AuditFinding(
|
||
number,
|
||
"task-file",
|
||
f"工作分支任务字段 {key} 与协调状态不一致。",
|
||
)
|
||
)
|
||
metadata_paths = metadata.get("write_paths")
|
||
if not isinstance(metadata_paths, list) or set(metadata_paths) != set(
|
||
claim_paths
|
||
):
|
||
findings.append(
|
||
AuditFinding(
|
||
number,
|
||
"task-file",
|
||
"工作分支 write_paths 与 CLAIM 不一致。",
|
||
)
|
||
)
|
||
if task.claimed_at is None:
|
||
findings.append(AuditFinding(number, "stale", "CLAIM 缺少有效 claimed_at。"))
|
||
elif task.claimed_at > now + CLOCK_SKEW:
|
||
findings.append(AuditFinding(number, "stale", "claimed_at 超出允许时钟偏差。"))
|
||
if task.lease_until is None:
|
||
findings.append(AuditFinding(number, "stale", "CLAIM 缺少有效 lease_until。"))
|
||
elif task.claimed_at is not None:
|
||
if task.lease_until <= task.claimed_at:
|
||
findings.append(AuditFinding(number, "stale", "lease_until 必须晚于 claimed_at。"))
|
||
elif task.lease_until - task.claimed_at > MAX_LEASE:
|
||
findings.append(AuditFinding(number, "stale", "claim 租期不得超过 24 小时。"))
|
||
if task.lease_until <= now:
|
||
findings.append(AuditFinding(number, "stale", "claim 已过期，需人工审查回收。"))
|
||
elif status == "status/todo" and claim_branch in branches:
|
||
findings.append(AuditFinding(number, "claim", "TODO 仍存在 claim 分支。"))
|
||
|
||
matching_prs = [
|
||
pr
|
||
for pr in pull_requests
|
||
if task_id and TASK_IN_TITLE.search(str(pr.get("title") or ""))
|
||
and TASK_IN_TITLE.search(str(pr.get("title") or "")).group(1) == task_id
|
||
]
|
||
if status == "status/review":
|
||
open_prs = [pr for pr in matching_prs if pr.get("state") == "open"]
|
||
if len(open_prs) != 1:
|
||
findings.append(
|
||
AuditFinding(number, "pull-request", "status/review 必须恰有一个 open PR。")
|
||
)
|
||
else:
|
||
findings.extend(validate_pull_request(open_prs[0], task))
|
||
if status == "status/done":
|
||
merged_prs = [pr for pr in matching_prs if bool(pr.get("merged"))]
|
||
if len(merged_prs) != 1:
|
||
findings.append(
|
||
AuditFinding(number, "pull-request", "status/done 必须恰有一个 merged PR。")
|
||
)
|
||
else:
|
||
findings.extend(validate_pull_request(merged_prs[0], task))
|
||
|
||
for task_id, metadata in sorted(local.items()):
|
||
issue_number = metadata.get("issue")
|
||
if isinstance(issue_number, int):
|
||
mapped = remote.get(task_id)
|
||
if mapped is None or mapped.number != issue_number:
|
||
findings.append(AuditFinding(issue_number, "mapping", f"本地 {task_id} 没有唯一远端映射。"))
|
||
elif metadata.get("status") == "DONE" and mapped.status != "status/done":
|
||
findings.append(AuditFinding(issue_number, "status", "本地 DONE 与远端状态不一致。"))
|
||
|
||
active = sorted(
|
||
(task for task in remote.values() if task.status in ACTIVE_LABELS),
|
||
key=lambda task: task.task_id,
|
||
)
|
||
workers: dict[str, str] = {}
|
||
for task in active:
|
||
if not task.claimed_by:
|
||
continue
|
||
previous = workers.get(task.claimed_by)
|
||
if previous:
|
||
findings.append(
|
||
AuditFinding(
|
||
task.number,
|
||
"worker-overlap",
|
||
f"claimed_by 同时活跃于 {previous} 和 {task.task_id}。",
|
||
)
|
||
)
|
||
else:
|
||
workers[task.claimed_by] = task.task_id
|
||
for index, left in enumerate(active):
|
||
for right in active[index + 1 :]:
|
||
if any(
|
||
scopes_overlap(a, b)
|
||
for a in left.write_paths or []
|
||
for b in right.write_paths or []
|
||
):
|
||
findings.append(
|
||
AuditFinding(
|
||
right.number,
|
||
"scope-overlap",
|
||
f"活跃 write_paths 与 {left.task_id} 重叠。",
|
||
)
|
||
)
|
||
return sorted(set(findings)), len(issues)
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="只读审计 Gitea 任务协调状态。")
|
||
parser.add_argument("--repo", help="目标 owner/repo；也可设置 GITEA_REPOSITORY。")
|
||
parser.add_argument(
|
||
"--dispatcher",
|
||
default=os.environ.get("GITEA_DISPATCHER_LOGIN"),
|
||
help="可信 dispatcher 的 Gitea 登录名；也可设置 GITEA_DISPATCHER_LOGIN。",
|
||
)
|
||
parser.add_argument(
|
||
"--root",
|
||
type=Path,
|
||
default=Path(__file__).resolve().parents[1],
|
||
help="本地仓库根目录。",
|
||
)
|
||
return parser.parse_args()
|
||
|
||
|
||
def main() -> int:
|
||
args = parse_args()
|
||
root = args.root.resolve()
|
||
if not root.is_dir() or not (root / "docs" / "tasks").is_dir():
|
||
print("ERROR: --root 必须是包含 docs/tasks 的仓库目录。", file=sys.stderr)
|
||
return 2
|
||
repo = args.repo or os.environ.get("GITEA_REPOSITORY")
|
||
if not repo:
|
||
print("ERROR: 需要 --repo owner/repo 或 GITEA_REPOSITORY。", file=sys.stderr)
|
||
return 2
|
||
try:
|
||
root_url, owner, name = validate_config(
|
||
os.environ.get("GITEA_URL", ""),
|
||
os.environ.get("GITEA_TOKEN", ""),
|
||
repo,
|
||
)
|
||
client = GiteaClient(root_url, owner, name, os.environ["GITEA_TOKEN"])
|
||
findings, count = audit_repository(
|
||
root, client, datetime.now(timezone.utc), args.dispatcher
|
||
)
|
||
except ValueError as exc:
|
||
print(f"ERROR: {exc}", file=sys.stderr)
|
||
return 2
|
||
except ApiError as exc:
|
||
print(f"ERROR: Gitea 只读审计失败（HTTP {exc.status}）。", file=sys.stderr)
|
||
return 2
|
||
except RuntimeError:
|
||
print("ERROR: Gitea 只读审计失败（网络、代理或响应格式异常）。", file=sys.stderr)
|
||
return 2
|
||
if findings:
|
||
for finding in findings:
|
||
print(finding.render(), file=sys.stderr)
|
||
print(f"Gitea 协调审计失败：{len(findings)} 项不一致。", file=sys.stderr)
|
||
return 1
|
||
print(f"Gitea 协调审计通过：检查 {count} 个 kind/task Issue，未执行远端写入。")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|