feat(ack): bind test env to deployer and add regression mode

ACK 0.19.0 hands test-environment deploys to the deployer skill, documents bug-fix as a first-class scenario, and adds docs/ack/regression.yaml harvest plus a /ack regression run.
2026-08-25 14:41:05 +08:00
parent ee31278947
commit ad6695245b
39 changed files with 1919 additions and 206 deletions
@@ -51,6 +51,8 @@ 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(
@@ -117,6 +119,51 @@ DELIVERY_STATUSES = {
 }
 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",
@@ -636,6 +683,205 @@ def validate_delivery_runs(
            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

@@ -779,6 +1025,19 @@ def validate_builtin(data: dict) -> list[str]:
            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") != "docs/ack/regression.yaml"
+        ):
+            errors.append(
+                "project.regressionFile 必须固定为 docs/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
@@ -941,6 +1200,7 @@ def validate_builtin(data: dict) -> list[str]:
                    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 = {}
@@ -1129,6 +1389,8 @@ def validate_builtin(data: dict) -> list[str]:
            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