Add harness coding docs for SoftBox (Go + Gio dual-build)

Initialize the full harness coding document set from the harness_coding_docs template, customized for the SoftBox project: - Vision, requirements, tech stack (modern Go 1.25 + Gio v0.10.1; Win7 legacy Go 1.20.14 + Gio v0.6.0), architecture, coding rules - Protocol contracts (signed catalog, package protocol v1, Ed25519 license, events, CLI) and Gio view structure - Roadmap Phase 0-6 with 20 suggested tasks; T-001 (monorepo skeleton) filed and ready to claim - Agent entry points (AGENTS.md, docs/00-ai-start-here.md), context manifest, governance scripts and tests - Merge Go gitignore with harness rules; keep go.work tracked Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-16 14:33:54 +08:00
parent 03bd4ab62a
commit 0184595ccc
43 changed files with 5010 additions and 10 deletions
@@ -0,0 +1,688 @@
+#!/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())