.pouch/skills/ack/scripts/validate_tasks.py

#!/usr/bin/env python3
"""校验 tasks.yaml 是否符合 ack 任务板结构。

权威结构是同目录上层的 templates/tasks.schema.json（跨语言可用）。
本脚本是参考实现：
  - 始终执行内置语义校验；
  - 安装了 jsonschema 时，再叠加 schema 结构校验；
  - knowledge 字段会检查引用格式、候选结构和 Test 检查结果。
YAML 优先使用 PyYAML；未安装时使用 fail-closed 的 ACK YAML 子集。
JSON 任务板只使用标准库，两种格式都拒绝重复键。

用法:
    python3 validate_tasks.py [tasks.yaml]
    python3 validate_tasks.py --schema path/to/tasks.schema.json tasks.yaml

退出码: 0 通过 / 1 校验失败 / 2 环境或用法错误。
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path

from approval_payload import approval_payload_hash
from yaml_subset import (
    DuplicateKeyError,
    YamlSubsetError,
    load_json_unique,
    load_yaml_subset,
    make_unique_pyyaml_loader,
)
from worker_profiles import validate_routing_document

STATUS_ENUM = {
    "open",
    "dispatched",
    "fixed_by_dev",
    "retesting",
    "failed_retest",
    "verified",
    "blocked",
    "leftover",
}
MAX_ROUNDS = 3
KNOWLEDGE_KINDS = {"guardrail", "pitfall", "verification"}
KNOWLEDGE_CHECK_RESULTS = {"passed", "failed", "not_applicable"}
KNOWLEDGE_REF_RE = re.compile(r"^K-[A-Z0-9][A-Z0-9-]*@[1-9][0-9]*$")
TASK_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
ATTEMPT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*-A[1-9][0-9]*$")
DELIVERY_RUN_ID_RE = re.compile(r"^DR-[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
REGRESSION_RUN_ID_RE = re.compile(r"^RR-[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
REGRESSION_CASE_ID_RE = re.compile(r"^REG-[A-Za-z0-9][A-Za-z0-9-]*$")
DELIVERY_PROFILE_RE = re.compile(r"^[a-z][a-z0-9-]{0,63}$")
GIT_REVISION_RE = re.compile(r"^[0-9a-f]{7,64}$")
SEMVER_RE = re.compile(
    r"^(0|[1-9][0-9]*)\."
    r"(0|[1-9][0-9]*)\."
    r"(0|[1-9][0-9]*)"
    r"(?:-(?:0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*)"
    r"(?:\.(?:0|[1-9][0-9]*|[0-9]*[A-Za-z-][0-9A-Za-z-]*))*)?"
    r"(?:\+[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?$"
)
KNOWLEDGE_SCOPE_FIELDS = {
    "components",
    "paths",
    "dependencies",
    "versions",
    "tags",
    "symbols",
    "errorSignatures",
}
KNOWLEDGE_APPLICATION_FIELDS = {"ref", "result", "evidence"}
KNOWLEDGE_CANDIDATE_FIELDS = {
    "kind",
    "title",
    "claim",
    "scope",
    "appliesWhen",
    "directive",
    "rationale",
    "evidenceRefs",
    "proposedBy",
    "proposedAt",
}
KNOWLEDGE_CHECK_FIELDS = {
    "ref",
    "result",
    "evidence",
    "checkedBy",
    "checkedAt",
}
DELIVERY_RUN_FIELDS = {
    "id",
    "profile",
    "taskIds",
    "status",
    "sourceRevision",
    "configRevision",
    "pullRequest",
    "artifacts",
    "deployments",
    "evidence",
    "updatedAt",
}
DELIVERY_RUN_OPTIONAL_FIELDS = {"intent"}
DELIVERY_RUN_INTENTS = {"testEnvironment", "release"}
DELIVERY_STATUSES = {
    "planned",
    "running",
    "blocked",
    "failed",
    "validation_ready",
    "review_ready",
    "released",
    "skipped",
}
DELIVERY_ARTIFACT_FIELDS = {"id", "type", "reference", "digest"}
DELIVERY_DEPLOYMENT_FIELDS = {"environment", "result", "evidence"}
REGRESSION_RUN_FIELDS = {
    "id",
    "suite",
    "caseIds",
    "status",
    "sourceRevision",
    "configRevision",
    "baseUrl",
    "results",
    "evidence",
    "updatedAt",
}
REGRESSION_RUN_OPTIONAL_FIELDS = {"deliveryRunId", "taskIds"}
REGRESSION_RUN_SUITES = {"smoke", "full", "custom"}
REGRESSION_RUN_STATUSES = {
    "planned",
    "running",
    "passed",
    "failed",
    "blocked",
    "skipped",
}
REGRESSION_RESULT_FIELDS = {"caseId", "result", "evidence"}
REGRESSION_RESULTS = {"pass", "fail", "skipped"}
REGRESSION_CANDIDATE_FIELDS = {
    "id",
    "title",
    "surface",
    "suite",
    "setup",
    "steps",
    "expected",
    "sourceKind",
}
REGRESSION_CANDIDATE_REQUIRED_FIELDS = {"title", "surface", "steps", "expected"}
REGRESSION_SURFACES = {"browser", "api"}
REGRESSION_SUITES = {"smoke", "full"}
REGRESSION_SOURCE_KINDS = {"feature", "bug"}
REGRESSION_EXPECTED_KINDS = {
    "visible-text",
    "api-status",
    "api-field",
    "url",
    "interaction",
}
FEISHU_REQUIRED_FIELDS = {
    "title", "actual", "expected", "stepsToReproduce", "acceptance",
    "attachments", "updatedAt",
}
FEISHU_OPTIONAL_FIELDS = {"priority", "fixLogic"}
FEISHU_CLARIFIED_FIELDS = {
    "title", "details", "problemStatement", "expectedOutcome", "acceptance",
    "intakeStatus", "ackTaskId", "attachments", "updatedAt",
}
FEISHU_CONFIG_FIELDS = {"provider", "workflow", "profile", "baseToken", "tableId", "viewId", "fields"}
FEISHU_SOURCE_FIELDS = {
    "kind", "workflow", "ref", "recordId", "updatedAt", "approvedRevision",
    "approvedPayloadHash",
}
FEISHU_WORKFLOWS = {"read-only-v1", "reviewed-writeback-v1", "clarified-writeback-v1"}
FEISHU_PROFILE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
FEISHU_SOURCE_REF_RE = re.compile(r"^feishu-base:sha256:[0-9a-f]{64}$")
FEISHU_RECORD_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,255}$")
DISPATCH_FIELDS = {
    "taskId",
    "dispatchId",
    "worker",
    "developer",
    "test",
    "rounds",
    "environmentIncidents",
}
ENVIRONMENT_INCIDENT_FIELDS = {
    "id",
    "attemptId",
    "role",
    "phase",
    "status",
    "summary",
    "evidence",
    "impact",
    "recoveryAction",
    "userAction",
    "reportedAt",
    "resolvedAt",
}
ENVIRONMENT_INCIDENT_ROLES = {"coordinator", "developer", "test"}
ENVIRONMENT_INCIDENT_PHASES = {
    "launch",
    "orchestration",
    "service",
    "test_data",
    "browser",
    "tooling",
    "permissions",
    "other",
}
ENVIRONMENT_INCIDENT_STATUSES = {"open", "resolved"}
KNOWLEDGE_CANDIDATE_REQUIRED_FIELDS = {
    "kind",
    "title",
    "claim",
    "scope",
    "appliesWhen",
    "directive",
    "rationale",
    "evidenceRefs",
}
KNOWLEDGE_CANDIDATE_TEXT_FIELDS = {
    "title",
    "claim",
    "appliesWhen",
    "directive",
    "rationale",
}


def _nonempty_string(value: object) -> bool:
    return isinstance(value, str) and bool(value.strip())


def load_document(path: Path) -> dict:
    try:
        content = path.read_text(encoding="utf-8")
    except OSError as exc:
        sys.stderr.write(f"任务板读取失败: {exc}\n")
        raise SystemExit(1)

    if path.suffix.lower() == ".json":
        try:
            data = load_json_unique(content)
        except (json.JSONDecodeError, DuplicateKeyError) as exc:
            sys.stderr.write(f"JSON 解析失败: {exc}\n")
            raise SystemExit(1)
    else:
        try:
            import yaml  # type: ignore
        except ImportError:
            try:
                data = load_yaml_subset(content)
            except YamlSubsetError as exc:
                sys.stderr.write(f"YAML 子集解析失败: {exc}\n")
                raise SystemExit(1)
        else:
            try:
                data = yaml.load(
                    content,
                    Loader=make_unique_pyyaml_loader(yaml),
                )
            except yaml.YAMLError as exc:  # type: ignore
                sys.stderr.write(f"YAML 解析失败: {exc}\n")
                raise SystemExit(1)
    if not isinstance(data, dict):
        sys.stderr.write("任务板顶层必须是对象（mapping）\n")
        raise SystemExit(1)
    return data


def validate_knowledge_ref_list(
    value: object,
    where: str,
    errors: list[str],
) -> list[str]:
    if not isinstance(value, list):
        errors.append(f"{where}: 必须是列表")
        return []

    refs: list[str] = []
    seen: set[str] = set()
    for index, ref in enumerate(value):
        item_where = f"{where}[{index}]"
        if not isinstance(ref, str) or not KNOWLEDGE_REF_RE.fullmatch(ref):
            errors.append(f"{item_where}: 必须使用 K-<id>@<revision> 格式")
            continue
        if ref in seen:
            errors.append(f"{item_where}: 引用重复: {ref}")
        seen.add(ref)
        refs.append(ref)
    return refs


def reject_unknown_fields(
    value: dict,
    allowed: set[str],
    where: str,
    errors: list[str],
) -> None:
    """Mirror additionalProperties=false for builtin knowledge validation."""
    for field in sorted(set(value) - allowed):
        errors.append(f"{where}: 未知字段 {field!r}")


def validate_optional_string_fields(
    value: dict,
    fields: set[str],
    where: str,
    errors: list[str],
) -> None:
    for field in sorted(fields):
        if field in value and not isinstance(value[field], str):
            errors.append(f"{where}.{field}: 必须是字符串")


def validate_knowledge_fields(
    task: dict,
    where: str,
    status: object,
    errors: list[str],
) -> None:
    refs = (
        validate_knowledge_ref_list(
            task["knowledgeRefs"],
            f"{where}.knowledgeRefs",
            errors,
        )
        if "knowledgeRefs" in task
        else []
    )
    if "knowledgeApplied" in task:
        applications = task["knowledgeApplied"]
        if not isinstance(applications, list):
            errors.append(f"{where}.knowledgeApplied: 必须是列表")
        else:
            seen_applications: set[str] = set()
            for index, application in enumerate(applications):
                application_where = f"{where}.knowledgeApplied[{index}]"
                if not isinstance(application, dict):
                    errors.append(f"{application_where}: 必须是对象")
                    continue
                reject_unknown_fields(
                    application,
                    KNOWLEDGE_APPLICATION_FIELDS,
                    application_where,
                    errors,
                )
                ref = application.get("ref")
                if not isinstance(ref, str) or not KNOWLEDGE_REF_RE.fullmatch(ref):
                    errors.append(
                        f"{application_where}.ref: 必须使用 K-<id>@<revision> 格式"
                    )
                else:
                    if ref in seen_applications:
                        errors.append(f"{application_where}.ref: 应用结果重复: {ref}")
                    seen_applications.add(ref)
                    if ref not in refs:
                        errors.append(f"{application_where}.ref: {ref} 不在 knowledgeRefs 中")
                if application.get("result") not in {"applied", "not_applicable"}:
                    errors.append(
                        f"{application_where}.result: 必须是 applied/not_applicable"
                    )
                evidence = application.get("evidence")
                if not isinstance(evidence, str) or not evidence.strip():
                    errors.append(f"{application_where}.evidence: 必须提供非空证据")

    if "knowledgeCandidates" in task:
        candidates = task["knowledgeCandidates"]
        if not isinstance(candidates, list):
            errors.append(f"{where}.knowledgeCandidates: 必须是列表")
        else:
            for index, candidate in enumerate(candidates):
                candidate_where = f"{where}.knowledgeCandidates[{index}]"
                if not isinstance(candidate, dict):
                    errors.append(f"{candidate_where}: 必须是对象")
                    continue
                reject_unknown_fields(
                    candidate,
                    KNOWLEDGE_CANDIDATE_FIELDS,
                    candidate_where,
                    errors,
                )
                missing = sorted(
                    key
                    for key in KNOWLEDGE_CANDIDATE_REQUIRED_FIELDS
                    if key not in candidate
                )
                if missing:
                    errors.append(
                        f"{candidate_where}: 缺少必填字段 {', '.join(missing)}"
                    )
                if candidate.get("kind") not in KNOWLEDGE_KINDS:
                    errors.append(
                        f"{candidate_where}.kind: 必须是 {sorted(KNOWLEDGE_KINDS)}"
                    )
                for field in sorted(KNOWLEDGE_CANDIDATE_TEXT_FIELDS):
                    value = candidate.get(field)
                    if not isinstance(value, str) or not value.strip():
                        errors.append(
                            f"{candidate_where}.{field}: 必须是非空字符串"
                        )
                validate_optional_string_fields(
                    candidate,
                    {"proposedBy", "proposedAt"},
                    candidate_where,
                    errors,
                )
                scope = candidate.get("scope")
                if not isinstance(scope, dict):
                    errors.append(f"{candidate_where}.scope: 至少包含一个非空作用域")
                else:
                    unknown_scope_fields = sorted(
                        set(scope) - KNOWLEDGE_SCOPE_FIELDS - {"all"}
                    )
                    for field in unknown_scope_fields:
                        errors.append(
                            f"{candidate_where}.scope: 未知字段 {field!r}"
                        )
                    if "all" in scope and not isinstance(scope.get("all"), bool):
                        errors.append(f"{candidate_where}.scope.all: 必须是布尔值")
                    populated_scope_fields: list[str] = []
                    for field in sorted(KNOWLEDGE_SCOPE_FIELDS & set(scope)):
                        values = scope.get(field)
                        if not isinstance(values, list) or any(
                            not isinstance(item, str) or not item.strip()
                            for item in values
                        ):
                            errors.append(
                                f"{candidate_where}.scope.{field}: "
                                "必须是非空字符串列表"
                            )
                        elif len(values) != len(set(values)):
                            errors.append(
                                f"{candidate_where}.scope.{field}: 不能包含重复值"
                            )
                        elif values:
                            populated_scope_fields.append(field)
                    if scope.get("all") is True and populated_scope_fields:
                        errors.append(
                            f"{candidate_where}.scope: all=true 时不能同时填写作用域维度"
                        )
                    if scope.get("all") is not True and not populated_scope_fields:
                        errors.append(
                            f"{candidate_where}.scope: 至少包含一个非空作用域"
                        )
                evidence_refs = candidate.get("evidenceRefs")
                if (
                    not isinstance(evidence_refs, list)
                    or not evidence_refs
                    or not all(
                        isinstance(ref, str) and ref.strip()
                        for ref in evidence_refs
                    )
                ):
                    errors.append(
                        f"{candidate_where}.evidenceRefs: 必须是非空字符串列表"
                    )
                elif len(evidence_refs) != len(set(evidence_refs)):
                    errors.append(
                        f"{candidate_where}.evidenceRefs: 不能包含重复值"
                    )

    if "knowledgeChecks" not in task:
        return
    checks = task["knowledgeChecks"]
    if not isinstance(checks, list):
        errors.append(f"{where}.knowledgeChecks: 必须是列表")
        return

    seen_checks: set[str] = set()
    for index, check in enumerate(checks):
        check_where = f"{where}.knowledgeChecks[{index}]"
        if not isinstance(check, dict):
            errors.append(f"{check_where}: 必须是对象")
            continue
        reject_unknown_fields(
            check,
            KNOWLEDGE_CHECK_FIELDS,
            check_where,
            errors,
        )
        validate_optional_string_fields(
            check,
            {"checkedBy", "checkedAt"},
            check_where,
            errors,
        )
        ref = check.get("ref")
        result = check.get("result")
        evidence = check.get("evidence")
        if not isinstance(ref, str) or not KNOWLEDGE_REF_RE.fullmatch(ref):
            errors.append(f"{check_where}.ref: 必须使用 K-<id>@<revision> 格式")
        else:
            if ref in seen_checks:
                errors.append(f"{check_where}.ref: 检查结果重复: {ref}")
            seen_checks.add(ref)
            if ref not in refs:
                errors.append(f"{check_where}.ref: {ref} 不在 knowledgeRefs 中")
        if result not in KNOWLEDGE_CHECK_RESULTS:
            errors.append(
                f"{check_where}.result: 必须是 {sorted(KNOWLEDGE_CHECK_RESULTS)}"
            )
        if not isinstance(evidence, str) or not evidence.strip():
            errors.append(f"{check_where}.evidence: 必须提供非空证据")
        if status == "verified" and result == "failed":
            errors.append(f"{check_where}: verified 任务不能保留失败的知识检查")


def validate_delivery_runs(
    value: object,
    task_statuses: dict[str, object],
    errors: list[str],
) -> None:
    if not isinstance(value, list):
        errors.append("deliveryRuns 必须是列表")
        return

    seen_run_ids: set[str] = set()
    for index, run in enumerate(value):
        where = f"deliveryRuns[{index}]"
        if not isinstance(run, dict):
            errors.append(f"{where}: 必须是对象")
            continue
        reject_unknown_fields(
            run, DELIVERY_RUN_FIELDS | DELIVERY_RUN_OPTIONAL_FIELDS, where, errors
        )
        missing = sorted(DELIVERY_RUN_FIELDS - set(run))
        for field in missing:
            errors.append(f"{where}.{field}: 必填")

        run_id = run.get("id")
        if not isinstance(run_id, str) or DELIVERY_RUN_ID_RE.fullmatch(run_id) is None:
            errors.append(f"{where}.id: 必须使用 DR-<id> 格式")
        elif run_id in seen_run_ids:
            errors.append(f"{where}.id: 不能重复 {run_id!r}")
        else:
            seen_run_ids.add(run_id)

        profile = run.get("profile")
        if not isinstance(profile, str) or DELIVERY_PROFILE_RE.fullmatch(profile) is None:
            errors.append(f"{where}.profile: 必须使用小写连字符 ID")
        status = run.get("status")
        if status not in DELIVERY_STATUSES:
            errors.append(f"{where}.status: 必须是 {sorted(DELIVERY_STATUSES)}")

        intent = run.get("intent")
        if "intent" in run and intent not in DELIVERY_RUN_INTENTS:
            errors.append(
                f"{where}.intent: 必须是 {sorted(DELIVERY_RUN_INTENTS)}"
            )
        task_ids = run.get("taskIds")
        allow_empty_tasks = intent in DELIVERY_RUN_INTENTS
        if (
            not isinstance(task_ids, list)
            or (not task_ids and not allow_empty_tasks)
            or any(not _nonempty_string(task_id) for task_id in (task_ids or []))
        ):
            errors.append(
                f"{where}.taskIds: 必须是任务 ID 列表"
                if allow_empty_tasks
                else f"{where}.taskIds: 必须是非空任务 ID 列表"
            )
            task_ids = []
        elif len(task_ids) != len(set(task_ids)):
            errors.append(f"{where}.taskIds: 不能包含重复值")
        for task_id in task_ids:
            if task_id not in task_statuses:
                errors.append(f"{where}.taskIds: 未知任务 {task_id!r}")
            elif task_statuses[task_id] != "verified":
                errors.append(
                    f"{where}: delivery run 只能引用 verified 任务，"
                    f"{task_id!r} 当前是 {task_statuses[task_id]!r}"
                )

        for field in ("sourceRevision", "configRevision"):
            revision = run.get(field)
            if revision is not None and (
                not isinstance(revision, str) or GIT_REVISION_RE.fullmatch(revision) is None
            ):
                errors.append(f"{where}.{field}: 必须是 null 或 7..64 位小写十六进制 revision")
        pull_request = run.get("pullRequest")
        if pull_request is not None and not isinstance(pull_request, str):
            errors.append(f"{where}.pullRequest: 必须是字符串或 null")
        if status != "skipped":
            for field in ("sourceRevision", "configRevision"):
                if not _nonempty_string(run.get(field)):
                    errors.append(f"{where}.{field}: status={status!r} 时必须填写")
        if status in {"review_ready", "released"}:
            if not _nonempty_string(run.get("pullRequest")):
                errors.append(f"{where}.pullRequest: status={status!r} 时必须填写")

        artifacts = run.get("artifacts")
        if not isinstance(artifacts, list):
            errors.append(f"{where}.artifacts: 必须是列表")
        else:
            seen_artifacts: set[str] = set()
            for artifact_index, artifact in enumerate(artifacts):
                artifact_where = f"{where}.artifacts[{artifact_index}]"
                if not isinstance(artifact, dict):
                    errors.append(f"{artifact_where}: 必须是对象")
                    continue
                reject_unknown_fields(
                    artifact,
                    DELIVERY_ARTIFACT_FIELDS,
                    artifact_where,
                    errors,
                )
                artifact_id = artifact.get("id")
                if (
                    not isinstance(artifact_id, str)
                    or DELIVERY_PROFILE_RE.fullmatch(artifact_id) is None
                ):
                    errors.append(f"{artifact_where}.id: 必须使用小写连字符 ID")
                elif artifact_id in seen_artifacts:
                    errors.append(f"{artifact_where}.id: 不能重复 {artifact_id!r}")
                else:
                    seen_artifacts.add(artifact_id)
                if artifact.get("type") not in {"deb", "oci-image", "file"}:
                    errors.append(f"{artifact_where}.type: 必须是 deb/oci-image/file")
                if not _nonempty_string(artifact.get("reference")):
                    errors.append(f"{artifact_where}.reference: 必须是非空字符串")
                digest = artifact.get("digest")
                if digest is not None and (
                    not isinstance(digest, str)
                    or re.fullmatch(r"sha256:[0-9a-f]{64}", digest) is None
                ):
                    errors.append(f"{artifact_where}.digest: 必须是 sha256:<64 hex> 或 null")
                if status in {"validation_ready", "review_ready", "released"} and not _nonempty_string(digest):
                    errors.append(
                        f"{artifact_where}.digest: status={status!r} 时必须填写"
                    )

        deployments = run.get("deployments")
        if not isinstance(deployments, list):
            errors.append(f"{where}.deployments: 必须是列表")
        else:
            seen_environments: set[str] = set()
            for deployment_index, deployment in enumerate(deployments):
                deployment_where = f"{where}.deployments[{deployment_index}]"
                if not isinstance(deployment, dict):
                    errors.append(f"{deployment_where}: 必须是对象")
                    continue
                reject_unknown_fields(
                    deployment,
                    DELIVERY_DEPLOYMENT_FIELDS,
                    deployment_where,
                    errors,
                )
                environment = deployment.get("environment")
                if (
                    not isinstance(environment, str)
                    or DELIVERY_PROFILE_RE.fullmatch(environment) is None
                ):
                    errors.append(f"{deployment_where}.environment: 必须使用小写连字符 ID")
                elif environment in seen_environments:
                    errors.append(f"{deployment_where}.environment: 不能重复 {environment!r}")
                else:
                    seen_environments.add(environment)
                if deployment.get("result") not in {"succeeded", "failed", "rolled_back"}:
                    errors.append(
                        f"{deployment_where}.result: 必须是 succeeded/failed/rolled_back"
                    )
                if not _nonempty_string(deployment.get("evidence")):
                    errors.append(f"{deployment_where}.evidence: 必须是非空字符串")

        evidence = run.get("evidence")
        if not isinstance(evidence, list) or any(
            not _nonempty_string(item) for item in evidence
        ):
            errors.append(f"{where}.evidence: 必须是字符串列表")
        elif status in {"blocked", "failed", "validation_ready", "review_ready", "released", "skipped"} and not evidence:
            errors.append(f"{where}.evidence: status={status!r} 时不能为空")
        if not _nonempty_string(run.get("updatedAt")):
            errors.append(f"{where}.updatedAt: 必须是非空字符串")


def validate_regression_fields(task: dict, where: str, errors: list[str]) -> None:
    if "regressionRefs" in task:
        refs = task["regressionRefs"]
        if not isinstance(refs, list):
            errors.append(f"{where}.regressionRefs: 必须是列表")
        else:
            seen: set[str] = set()
            for index, ref in enumerate(refs):
                item_where = f"{where}.regressionRefs[{index}]"
                if not isinstance(ref, str) or REGRESSION_CASE_ID_RE.fullmatch(ref) is None:
                    errors.append(f"{item_where}: 必须使用 REG-<id> 格式")
                    continue
                if ref in seen:
                    errors.append(f"{item_where}: 不能重复 {ref!r}")
                seen.add(ref)

    if "regressionCandidates" not in task:
        return
    candidates = task["regressionCandidates"]
    if not isinstance(candidates, list):
        errors.append(f"{where}.regressionCandidates: 必须是列表")
        return
    for index, candidate in enumerate(candidates):
        item_where = f"{where}.regressionCandidates[{index}]"
        if not isinstance(candidate, dict):
            errors.append(f"{item_where}: 必须是对象")
            continue
        reject_unknown_fields(
            candidate, REGRESSION_CANDIDATE_FIELDS, item_where, errors
        )
        missing = sorted(REGRESSION_CANDIDATE_REQUIRED_FIELDS - set(candidate))
        for field in missing:
            errors.append(f"{item_where}.{field}: 必填")
        if "id" in candidate and (
            not isinstance(candidate.get("id"), str)
            or REGRESSION_CASE_ID_RE.fullmatch(candidate["id"]) is None
        ):
            errors.append(f"{item_where}.id: 必须使用 REG-<id> 格式")
        if not _nonempty_string(candidate.get("title")):
            errors.append(f"{item_where}.title: 必须是非空字符串")
        if candidate.get("surface") not in REGRESSION_SURFACES:
            errors.append(f"{item_where}.surface: 必须是 {sorted(REGRESSION_SURFACES)}")
        if "suite" in candidate and candidate.get("suite") not in REGRESSION_SUITES:
            errors.append(f"{item_where}.suite: 必须是 {sorted(REGRESSION_SUITES)}")
        if "sourceKind" in candidate and (
            candidate.get("sourceKind") not in REGRESSION_SOURCE_KINDS
        ):
            errors.append(
                f"{item_where}.sourceKind: 必须是 {sorted(REGRESSION_SOURCE_KINDS)}"
            )
        steps = candidate.get("steps")
        if (
            not isinstance(steps, list)
            or not steps
            or any(not _nonempty_string(step) for step in steps)
        ):
            errors.append(f"{item_where}.steps: 必须是非空字符串列表")
        expected = candidate.get("expected")
        if not isinstance(expected, list) or not expected:
            errors.append(f"{item_where}.expected: 必须是非空列表")
            continue
        for expected_index, item in enumerate(expected):
            expected_where = f"{item_where}.expected[{expected_index}]"
            if not isinstance(item, dict):
                errors.append(f"{expected_where}: 必须是对象")
                continue
            if item.get("kind") not in REGRESSION_EXPECTED_KINDS:
                errors.append(
                    f"{expected_where}.kind: 必须是 {sorted(REGRESSION_EXPECTED_KINDS)}"
                )
            if not _nonempty_string(item.get("value")):
                errors.append(f"{expected_where}.value: 必须是非空字符串")


def validate_regression_runs(value: object, errors: list[str]) -> None:
    if not isinstance(value, list):
        errors.append("regressionRuns 必须是列表")
        return

    seen_run_ids: set[str] = set()
    for index, run in enumerate(value):
        where = f"regressionRuns[{index}]"
        if not isinstance(run, dict):
            errors.append(f"{where}: 必须是对象")
            continue
        reject_unknown_fields(
            run,
            REGRESSION_RUN_FIELDS | REGRESSION_RUN_OPTIONAL_FIELDS,
            where,
            errors,
        )
        missing = sorted(REGRESSION_RUN_FIELDS - set(run))
        for field in missing:
            errors.append(f"{where}.{field}: 必填")

        run_id = run.get("id")
        if not isinstance(run_id, str) or REGRESSION_RUN_ID_RE.fullmatch(run_id) is None:
            errors.append(f"{where}.id: 必须使用 RR-<id> 格式")
        elif run_id in seen_run_ids:
            errors.append(f"{where}.id: 不能重复 {run_id!r}")
        else:
            seen_run_ids.add(run_id)

        if run.get("suite") not in REGRESSION_RUN_SUITES:
            errors.append(f"{where}.suite: 必须是 {sorted(REGRESSION_RUN_SUITES)}")
        status = run.get("status")
        if status not in REGRESSION_RUN_STATUSES:
            errors.append(f"{where}.status: 必须是 {sorted(REGRESSION_RUN_STATUSES)}")

        case_ids = run.get("caseIds")
        if (
            not isinstance(case_ids, list)
            or not case_ids
            or any(
                not isinstance(case_id, str)
                or REGRESSION_CASE_ID_RE.fullmatch(case_id) is None
                for case_id in case_ids
            )
        ):
            errors.append(f"{where}.caseIds: 必须是非空 REG-<id> 列表")
            case_ids = []
        elif len(case_ids) != len(set(case_ids)):
            errors.append(f"{where}.caseIds: 不能包含重复值")

        task_ids = run.get("taskIds")
        if task_ids is not None and (
            not isinstance(task_ids, list)
            or any(not _nonempty_string(task_id) for task_id in task_ids)
        ):
            errors.append(f"{where}.taskIds: 必须是任务 ID 列表")
        delivery_run_id = run.get("deliveryRunId")
        if delivery_run_id is not None and (
            not isinstance(delivery_run_id, str)
            or DELIVERY_RUN_ID_RE.fullmatch(delivery_run_id) is None
        ):
            errors.append(f"{where}.deliveryRunId: 必须使用 DR-<id> 格式")

        for field in ("sourceRevision", "configRevision"):
            revision = run.get(field)
            if revision is not None and (
                not isinstance(revision, str) or GIT_REVISION_RE.fullmatch(revision) is None
            ):
                errors.append(
                    f"{where}.{field}: 必须是 null 或 7..64 位小写十六进制 revision"
                )
        if status != "skipped":
            for field in ("sourceRevision", "configRevision", "baseUrl"):
                if not _nonempty_string(run.get(field)):
                    errors.append(f"{where}.{field}: status={status!r} 时必须填写")

        results = run.get("results")
        if not isinstance(results, list):
            errors.append(f"{where}.results: 必须是列表")
            results = []
        elif status in {"passed", "failed"} and not results:
            errors.append(f"{where}.results: status={status!r} 时不能为空")
        seen_results: set[str] = set()
        for result_index, result in enumerate(results):
            result_where = f"{where}.results[{result_index}]"
            if not isinstance(result, dict):
                errors.append(f"{result_where}: 必须是对象")
                continue
            reject_unknown_fields(result, REGRESSION_RESULT_FIELDS, result_where, errors)
            case_id = result.get("caseId")
            if (
                not isinstance(case_id, str)
                or REGRESSION_CASE_ID_RE.fullmatch(case_id) is None
            ):
                errors.append(f"{result_where}.caseId: 必须使用 REG-<id> 格式")
            elif case_id in seen_results:
                errors.append(f"{result_where}.caseId: 不能重复 {case_id!r}")
            else:
                seen_results.add(case_id)
                if case_id not in case_ids:
                    errors.append(f"{result_where}.caseId: 不在 caseIds 中")
            if result.get("result") not in REGRESSION_RESULTS:
                errors.append(
                    f"{result_where}.result: 必须是 {sorted(REGRESSION_RESULTS)}"
                )
            if not _nonempty_string(result.get("evidence")):
                errors.append(f"{result_where}.evidence: 必须是非空字符串")
        if status == "passed" and any(
            isinstance(item, dict) and item.get("result") == "fail" for item in results
        ):
            errors.append(f"{where}: passed 运行不能包含 fail 结果")
        if status == "failed" and not any(
            isinstance(item, dict) and item.get("result") == "fail" for item in results
        ):
            errors.append(f"{where}: failed 运行必须包含至少一条 fail 结果")

        evidence = run.get("evidence")
        if not isinstance(evidence, list) or any(
            not _nonempty_string(item) for item in evidence
        ):
            errors.append(f"{where}.evidence: 必须是字符串列表")
        if not _nonempty_string(run.get("updatedAt")):
            errors.append(f"{where}.updatedAt: 必须是非空字符串")


def validate_with_schema(data: dict, schema_path: Path) -> list[str]:
    import jsonschema  # type: ignore

    schema = json.loads(schema_path.read_text(encoding="utf-8"))
    validator = jsonschema.Draft7Validator(schema)
    errors = []
    for err in sorted(validator.iter_errors(data), key=lambda e: list(e.path)):
        loc = "/".join(str(p) for p in err.path) or "<root>"
        errors.append(f"[schema] {loc}: {err.message}")
    return errors


def validate_builtin(data: dict) -> list[str]:
    errors: list[str] = []
    project_intake_workflow = "read-only-v1"

    def validate_string_fields(
        value: dict,
        fields: set[str],
        where: str,
        *,
        nullable: bool = False,
    ) -> None:
        expected = "必须是字符串或 null" if nullable else "必须是字符串"
        for field in sorted(fields):
            if field not in value:
                continue
            field_value = value[field]
            if not isinstance(field_value, str) and not (
                nullable and field_value is None
            ):
                errors.append(f"{where}.{field}: {expected}")

    def validate_string_lists(
        value: dict,
        fields: set[str],
        where: str,
    ) -> None:
        for field in sorted(fields):
            if field not in value:
                continue
            items = value[field]
            if not isinstance(items, list):
                errors.append(f"{where}.{field}: 必须是列表")
            elif any(not isinstance(item, str) for item in items):
                errors.append(f"{where}.{field}: 列表项必须是字符串")

    def validate_object_fields(
        value: dict,
        fields: set[str],
        where: str,
    ) -> None:
        for field in sorted(fields):
            if field in value and not isinstance(value[field], dict):
                errors.append(f"{where}.{field}: 必须是对象")

    version = data.get("version")
    if (
        not isinstance(version, int)
        or isinstance(version, bool)
        or version < 1
    ):
        errors.append("version 必须是 >=1 的整数")
    validate_string_fields(
        data,
        {"updatedAt", "source", "ackVersion", "kitVersion"},
        "<root>",
    )

    project = data.get("project")
    if not isinstance(project, dict):
        errors.append("project 必须是对象")
    else:
        if not _nonempty_string(project.get("name")):
            errors.append("project.name 必须是非空字符串")
        validate_string_fields(
            project,
            {"repoPath", "baseUrl", "devWorktree", "overlayFile", "deliveryFile"},
            "project",
        )
        if "bugIntake" in project:
            intake = project["bugIntake"]
            if not isinstance(intake, dict):
                errors.append("project.bugIntake 必须是对象")
            else:
                reject_unknown_fields(intake, FEISHU_CONFIG_FIELDS, "project.bugIntake", errors)
                if intake.get("provider") != "feishu-base":
                    errors.append("project.bugIntake.provider 必须是 feishu-base")
                workflow = intake.get("workflow", "read-only-v1")
                if workflow in FEISHU_WORKFLOWS:
                    project_intake_workflow = workflow
                if workflow not in FEISHU_WORKFLOWS:
                    errors.append("project.bugIntake.workflow 非法")
                profile = intake.get("profile")
                if not isinstance(profile, str) or FEISHU_PROFILE_RE.fullmatch(profile) is None:
                    errors.append("project.bugIntake.profile 非法")
                for key in ("baseToken", "tableId", "viewId"):
                    value = intake.get(key)
                    if not isinstance(value, str) or not value.strip() or any(char.isspace() for char in value):
                        errors.append(f"project.bugIntake.{key} 必须是无空白非空字符串")
                fields = intake.get("fields")
                expected_fields = (
                    FEISHU_CLARIFIED_FIELDS
                    if workflow == "clarified-writeback-v1"
                    else FEISHU_REQUIRED_FIELDS
                )
                allowed_fields = expected_fields | (
                    set() if workflow == "clarified-writeback-v1" else FEISHU_OPTIONAL_FIELDS
                )
                if (
                    not isinstance(fields, dict)
                    or not expected_fields.issubset(fields)
                    or not set(fields).issubset(allowed_fields)
                ):
                    errors.append("project.bugIntake.fields 必须且只能映射所需逻辑字段")
                elif any(not isinstance(v, str) or not v.strip() or any(c.isspace() for c in v) for v in fields.values()):
                    errors.append("project.bugIntake.fields 字段值必须是无空白非空字符串")
                elif len(set(fields.values())) != len(fields):
                    errors.append("project.bugIntake.fields 字段值不能重复")
                elif workflow == "reviewed-writeback-v1" and not {
                    "fixLogic", "priority"
                }.issubset(fields):
                    errors.append(
                        "reviewed-writeback-v1 必须映射 project.bugIntake.fields.fixLogic 和 priority"
                    )
        if (
            "knowledgeFile" in project
            and project.get("knowledgeFile") != ".pouch/ack/knowledge.yaml"
        ):
            errors.append(
                "project.knowledgeFile 必须固定为 .pouch/ack/knowledge.yaml"
            )
        if (
            "deliveryFile" in project
            and project.get("deliveryFile") != ".pouch/ack/delivery.yaml"
        ):
            errors.append(
                "project.deliveryFile 必须固定为 .pouch/ack/delivery.yaml"
            )
        if "deliveryFile" in project and not isinstance(data.get("deliveryRuns"), list):
            errors.append("引用 deliveryFile 的任务板必须包含 deliveryRuns 列表")
        if "deliveryRuns" in data and "deliveryFile" not in project:
            errors.append("deliveryRuns 存在时 project.deliveryFile 必须存在")
        if (
            "regressionFile" in project
            and project.get("regressionFile") != ".pouch/ack/regression.yaml"
        ):
            errors.append(
                "project.regressionFile 必须固定为 .pouch/ack/regression.yaml"
            )
        if "regressionFile" in project and not isinstance(
            data.get("regressionRuns"), list
        ):
            errors.append("引用 regressionFile 的任务板必须包含 regressionRuns 列表")
        if "regressionRuns" in data and "regressionFile" not in project:
            errors.append("regressionRuns 存在时 project.regressionFile 必须存在")

    ack_version = data.get("ackVersion")
    version_match = SEMVER_RE.fullmatch(ack_version) if isinstance(ack_version, str) else None
    if "ackVersion" in data and version_match is None:
        errors.append("ackVersion 必须是合法 SemVer（例如 0.10.0）")
    routing_required = (
        (isinstance(project, dict) and "orchestration" in project)
        or "workerReceipts" in data
        or (
            version_match is not None
            and (int(version_match.group(1)), int(version_match.group(2)))
            >= (0, 10)
        )
    )
    if routing_required:
        errors.extend(validate_routing_document(data))

    if "summary" in data:
        summary = data["summary"]
        if not isinstance(summary, dict):
            errors.append("summary 必须是对象")
        else:
            validate_string_lists(
                summary,
                {"verified", "open", "failedRetest", "leftovers"},
                "summary",
            )
    if "statusReference" in data and not isinstance(
        data["statusReference"], dict
    ):
        errors.append("statusReference 必须是对象")

    tasks = data.get("tasks")
    if not isinstance(tasks, list):
        errors.append("tasks 必须是列表")
        return errors

    seen_ids: set[str] = set()
    seen_source_refs: set[str] = set()
    for i, task in enumerate(tasks):
        where = f"tasks[{i}]"
        if not isinstance(task, dict):
            errors.append(f"{where}: 必须是对象")
            continue
        tid = task.get("id")
        title = task.get("title")
        status = task.get("status")
        if not _nonempty_string(tid):
            errors.append(f"{where}: id 必须是非空字符串")
        else:
            where = f"tasks[{i}] {tid}"
            if routing_required and TASK_ID_RE.fullmatch(tid) is None:
                errors.append(
                    f"{where}: v0.10 自动路由 id 只允许字母、数字、点、下划线和连字符"
                )
            if tid in seen_ids:
                errors.append(f"{where}: id 重复")
            seen_ids.add(tid)
        if not _nonempty_string(title):
            errors.append(f"{where}: title 必须是非空字符串")
        if status not in STATUS_ENUM:
            errors.append(
                f"{where}: status={status!r} 非法，应为 {sorted(STATUS_ENUM)}"
            )

        validate_string_fields(
            task,
            {
                "type",
                "priority",
                "assignee",
                "component",
                "description",
                "fixLogic",
                "expected",
                "actual",
            },
            where,
        )
        validate_string_lists(
            task,
            {"specRefs", "testRefs", "stepsToReproduce", "acceptanceCriteria"},
            where,
        )
        validate_object_fields(task, {"evidence", "verification"}, where)

        if "source" in task:
            source = task["source"]
            # `source` was historically an open extension point.  Preserve
            # non-Feishu strings/objects and tighten only the namespaced shape.
            if isinstance(source, dict) and source.get("kind") == "feishu-base":
                reject_unknown_fields(source, FEISHU_SOURCE_FIELDS, f"{where}.source", errors)
                ref = source.get("ref")
                if not isinstance(ref, str) or FEISHU_SOURCE_REF_RE.fullmatch(ref) is None:
                    errors.append(f"{where}.source.ref: 必须是不透明 feishu-base SHA-256 引用")
                else:
                    if ref in seen_source_refs:
                        errors.append(f"{where}.source.ref: 来源引用重复")
                    seen_source_refs.add(ref)
                record_id = source.get("recordId")
                if not isinstance(record_id, str) or FEISHU_RECORD_ID_RE.fullmatch(record_id) is None:
                    errors.append(f"{where}.source.recordId: 必须是合法飞书记录 ID")
                if not _nonempty_string(source.get("updatedAt")):
                    errors.append(f"{where}.source.updatedAt: 必须是非空字符串")
                source_workflow = source.get("workflow", "read-only-v1")
                if source_workflow not in FEISHU_WORKFLOWS:
                    errors.append(f"{where}.source.workflow: 非法")
                if (
                    project_intake_workflow in {"reviewed-writeback-v1", "clarified-writeback-v1"}
                    and source_workflow != project_intake_workflow
                    and status not in {"verified", "leftover"}
                ):
                    errors.append(
                        f"{where}.source.workflow: reviewed 项目的可执行飞书任务必须先迁移审核"
                    )
                approved_revision = source.get("approvedRevision")
                stored_payload_hash = source.get("approvedPayloadHash")
                is_approved_workflow = source_workflow in {"reviewed-writeback-v1", "clarified-writeback-v1"}
                if is_approved_workflow and approved_revision is None:
                    errors.append(f"{where}.source.approvedRevision: writeback workflow 必填")
                elif approved_revision is not None and (
                    not isinstance(approved_revision, str)
                    or re.fullmatch(r"sha256:[0-9a-f]{64}", approved_revision) is None
                ):
                    errors.append(f"{where}.source.approvedRevision: 必须是 sha256 revision")
                if is_approved_workflow:
                    if (
                        not isinstance(stored_payload_hash, str)
                        or re.fullmatch(r"sha256:[0-9a-f]{64}", stored_payload_hash) is None
                    ):
                        errors.append(f"{where}.source.approvedPayloadHash: reviewed workflow 必填")
                    required_strings = (
                        ("title", "description", "actual", "expected")
                        if source_workflow == "clarified-writeback-v1"
                        else ("title", "priority", "actual", "expected", "fixLogic")
                    )
                    for field in required_strings:
                        if not _nonempty_string(task.get(field)):
                            errors.append(f"{where}.{field}: reviewed workflow 必须是非空字符串")
                    required_lists = (
                        ("acceptanceCriteria",)
                        if source_workflow == "clarified-writeback-v1"
                        else ("stepsToReproduce", "acceptanceCriteria")
                    )
                    for field in required_lists:
                        items = task.get(field)
                        if (
                            not isinstance(items, list)
                            or not items
                            or any(not _nonempty_string(item) for item in items)
                        ):
                            errors.append(f"{where}.{field}: reviewed workflow 必须是非空字符串列表")
                    if (
                        isinstance(stored_payload_hash, str)
                        and re.fullmatch(r"sha256:[0-9a-f]{64}", stored_payload_hash)
                        and stored_payload_hash != approval_payload_hash(task)
                    ):
                        errors.append(f"{where}.source.approvedPayloadHash: 与任务审核字段不匹配")
                elif stored_payload_hash is not None:
                    errors.append(f"{where}.source.approvedPayloadHash: 只允许 reviewed workflow")

        validate_knowledge_fields(task, where, status, errors)
        validate_regression_fields(task, where, errors)

        if "dispatch" not in task:
            dispatch = {}
        elif not isinstance(task["dispatch"], dict):
            errors.append(f"{where}.dispatch: 必须是对象")
            dispatch = {}
        else:
            dispatch = task["dispatch"]
            reject_unknown_fields(
                dispatch,
                DISPATCH_FIELDS,
                f"{where}.dispatch",
                errors,
            )
            validate_string_fields(
                dispatch,
                {"taskId", "dispatchId", "worker"},
                f"{where}.dispatch",
                nullable=True,
            )

        rounds = dispatch.get("rounds", [])
        if not isinstance(rounds, list):
            errors.append(f"{where}.dispatch.rounds: 必须是列表")
        else:
            if len(rounds) > MAX_ROUNDS:
                errors.append(
                    f"{where}: 派发轮次 {len(rounds)} 超过上限 {MAX_ROUNDS}"
                )
            seen_attempt_ids: set[str] = set()
            round_numbers: list[int] = []
            for round_index, round_item in enumerate(rounds):
                round_where = f"{where}.dispatch.rounds[{round_index}]"
                if not isinstance(round_item, dict):
                    errors.append(f"{round_where}: 必须是对象")
                    continue
                if round_item.get("result") not in {"passed", "failed"}:
                    errors.append(f"{round_where}.result: 必须是 passed/failed")
                if "evidence" in round_item and not isinstance(
                    round_item["evidence"], str
                ):
                    errors.append(f"{round_where}.evidence: 必须是字符串")
                round_number = round_item.get("round")
                round_number_is_valid = (
                    isinstance(round_number, int)
                    and not isinstance(round_number, bool)
                    and 1 <= round_number <= MAX_ROUNDS
                )
                if not round_number_is_valid:
                    errors.append(
                        f"{round_where}.round: 必须是 1..{MAX_ROUNDS} 的整数"
                    )
                else:
                    round_numbers.append(round_number)
                if "attemptId" in round_item:
                    attempt_id = round_item["attemptId"]
                    if (
                        not isinstance(attempt_id, str)
                        or not ATTEMPT_ID_RE.fullmatch(attempt_id)
                    ):
                        errors.append(
                            f"{round_where}.attemptId: "
                            "必须使用 <task-id>-A<round> 格式"
                        )
                    else:
                        if attempt_id in seen_attempt_ids:
                            errors.append(
                                f"{round_where}.attemptId: "
                                f"轮次内不能重复: {attempt_id}"
                            )
                        seen_attempt_ids.add(attempt_id)
                        if (
                            isinstance(tid, str)
                            and round_number_is_valid
                            and attempt_id != f"{tid}-A{round_number}"
                        ):
                            errors.append(
                                f"{round_where}.attemptId: 应为 "
                                f"{tid}-A{round_number}"
                            )
            expected_rounds = list(range(1, len(rounds) + 1))
            if round_numbers != expected_rounds:
                errors.append(
                    f"{where}.dispatch.rounds: round 必须从 1 连续递增且不重复"
                )

        incidents = dispatch.get("environmentIncidents", [])
        if not isinstance(incidents, list):
            errors.append(f"{where}.dispatch.environmentIncidents: 必须是列表")
        else:
            seen_incident_ids: set[str] = set()
            for incident_index, incident in enumerate(incidents):
                incident_where = (
                    f"{where}.dispatch.environmentIncidents[{incident_index}]"
                )
                if not isinstance(incident, dict):
                    errors.append(f"{incident_where}: 必须是对象")
                    continue
                reject_unknown_fields(
                    incident,
                    ENVIRONMENT_INCIDENT_FIELDS,
                    incident_where,
                    errors,
                )
                incident_id = incident.get("id")
                expected_id = (
                    f"{tid}-ENV-{incident_index + 1}"
                    if isinstance(tid, str)
                    else None
                )
                if not isinstance(incident_id, str) or incident_id != expected_id:
                    errors.append(f"{incident_where}.id: 应为 {expected_id}")
                elif incident_id in seen_incident_ids:
                    errors.append(f"{incident_where}.id: 不能重复 {incident_id}")
                else:
                    seen_incident_ids.add(incident_id)
                if incident.get("role") not in ENVIRONMENT_INCIDENT_ROLES:
                    errors.append(
                        f"{incident_where}.role: 必须是 coordinator/developer/test"
                    )
                if incident.get("phase") not in ENVIRONMENT_INCIDENT_PHASES:
                    errors.append(f"{incident_where}.phase: 非法环境阶段")
                incident_status = incident.get("status")
                if incident_status not in ENVIRONMENT_INCIDENT_STATUSES:
                    errors.append(f"{incident_where}.status: 必须是 open/resolved")
                for field in (
                    "summary",
                    "evidence",
                    "impact",
                    "recoveryAction",
                    "userAction",
                    "reportedAt",
                ):
                    if not _nonempty_string(incident.get(field)):
                        errors.append(f"{incident_where}.{field}: 必须是非空字符串")
                if "attemptId" in incident and not (
                    incident["attemptId"] is None
                    or _nonempty_string(incident["attemptId"])
                ):
                    errors.append(f"{incident_where}.attemptId: 必须是字符串或 null")
                if "resolvedAt" in incident and not (
                    incident["resolvedAt"] is None
                    or _nonempty_string(incident["resolvedAt"])
                ):
                    errors.append(f"{incident_where}.resolvedAt: 必须是字符串或 null")
                if incident_status == "resolved" and not _nonempty_string(
                    incident.get("resolvedAt")
                ):
                    errors.append(
                        f"{incident_where}: resolved 必须填写 resolvedAt"
                    )

        resolution = task.get("resolution")
        if "resolution" in task:
            if not isinstance(resolution, dict):
                errors.append(f"{where}.resolution: 必须是对象")
            else:
                validate_string_fields(
                    resolution,
                    {
                        "fixedBy",
                        "verifiedBy",
                        "verifiedAt",
                        "leftoverReason",
                    },
                    f"{where}.resolution",
                    nullable=True,
                )
                validate_object_fields(
                    resolution,
                    {"evidence"},
                    f"{where}.resolution",
                )

        if status == "leftover":
            if (
                not isinstance(resolution, dict)
                or not _nonempty_string(resolution.get("leftoverReason"))
            ):
                errors.append(f"{where}: leftover 必须填 resolution.leftoverReason")

    if "deliveryRuns" in data:
        task_statuses = {
            task.get("id"): task.get("status")
            for task in tasks
            if isinstance(task, dict) and _nonempty_string(task.get("id"))
        }
        validate_delivery_runs(data["deliveryRuns"], task_statuses, errors)
    if "regressionRuns" in data:
        validate_regression_runs(data["regressionRuns"], errors)

    return errors


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="校验 tasks.yaml 结构")
    parser.add_argument("tasks", nargs="?", default="tasks.yaml", help="任务板路径")
    parser.add_argument("--schema", help="tasks.schema.json 路径（默认自动探测）")
    args = parser.parse_args(argv)

    tasks_path = Path(args.tasks)
    if not tasks_path.is_file():
        sys.stderr.write(f"找不到任务板文件: {tasks_path}\n")
        return 2

    data = load_document(tasks_path)

    schema_path = Path(args.schema) if args.schema else (
        Path(__file__).resolve().parent.parent / "templates" / "tasks.schema.json"
    )
    if args.schema and not schema_path.is_file():
        sys.stderr.write(f"找不到指定的 schema 文件: {schema_path}\n")
        return 2

    errors = validate_builtin(data)
    mode = "内置语义规则"
    try:
        import jsonschema  # type: ignore  # noqa: F401

        if schema_path.is_file():
            errors = validate_with_schema(data, schema_path) + errors
            mode = f"schema ({schema_path.name}) + 内置语义规则"
        else:
            mode = "内置语义规则（未找到 schema 文件）"
    except ImportError:
        pass

    if errors:
        sys.stderr.write(f"任务板校验失败（{mode}），共 {len(errors)} 项：\n")
        for e in errors:
            sys.stderr.write(f"  - {e}\n")
        return 1

    sys.stdout.write(f"任务板校验通过（{mode}）：{tasks_path}\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())