Files
.pouch/tests/test_ack_tasks_validation.py
T
laily ad6695245b 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

1281 lines
46 KiB
Python

from __future__ import annotations
import copy
import importlib.util
import json
import re
import subprocess
import sys
import tempfile
import textwrap
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
VALIDATOR = REPO_ROOT / "skills" / "ack" / "scripts" / "validate_tasks.py"
EXAMPLE = REPO_ROOT / "skills" / "ack" / "examples" / "tasks.example.yaml"
SCHEMA = REPO_ROOT / "skills" / "ack" / "templates" / "tasks.schema.json"
sys.path.insert(0, str(VALIDATOR.parent))
from approval_payload import approval_payload_hash # noqa: E402
def valid_knowledge_board() -> dict:
return {
"version": 1,
"project": {"name": "demo"},
"tasks": [
{
"id": "T-1",
"title": "validate knowledge fields",
"status": "open",
"knowledgeRefs": ["K-001@1"],
"knowledgeApplied": [
{
"ref": "K-001@1",
"result": "applied",
"evidence": "followed the guardrail",
}
],
"knowledgeCandidates": [
{
"kind": "pitfall",
"title": "candidate",
"claim": "the failure is reproducible",
"scope": {"components": ["web"]},
"appliesWhen": "the web component changes",
"directive": "run the reviewed check",
"rationale": "avoid the repeated failure",
"evidenceRefs": ["tasks.yaml#T-1"],
"proposedBy": "developer",
"proposedAt": "2026-07-31T10:00:00+08:00",
}
],
"knowledgeChecks": [
{
"ref": "K-001@1",
"result": "passed",
"evidence": "independently verified",
"checkedBy": "test",
"checkedAt": "2026-07-31T10:05:00+08:00",
}
],
}
],
}
def valid_manual_routing_board() -> dict:
board = valid_knowledge_board()
board["ackVersion"] = "0.10.0"
board["project"]["orchestration"] = {
"profileVersion": 1,
"mode": "manual",
"allowedWorktrees": [],
"modelAllowlist": {},
"profiles": {},
"defaults": {},
}
board["workerReceipts"] = []
return board
class AckTaskValidationTests(unittest.TestCase):
def run_validator(
self,
content: str | None = None,
*extra_args: str,
no_site_packages: bool = False,
suffix: str = ".yaml",
) -> subprocess.CompletedProcess[str]:
command = [sys.executable]
if no_site_packages:
command.append("-S")
command.append(str(VALIDATOR))
if content is None:
return subprocess.run(
[*command, *extra_args, str(EXAMPLE)],
cwd=REPO_ROOT,
text=True,
capture_output=True,
check=False,
)
with tempfile.TemporaryDirectory() as temp_dir:
task_file = Path(temp_dir) / f"tasks{suffix}"
task_file.write_text(textwrap.dedent(content), encoding="utf-8")
return subprocess.run(
[*command, *extra_args, str(task_file)],
cwd=REPO_ROOT,
text=True,
capture_output=True,
check=False,
)
def assert_board_rejected_in_all_modes(
self,
board: dict,
*expected_messages: str,
) -> None:
for no_site_packages in (False, True):
with self.subTest(no_site_packages=no_site_packages):
result = self.run_validator(
json.dumps(board),
no_site_packages=no_site_packages,
suffix=".json",
)
self.assertEqual(result.returncode, 1, result.stdout)
for message in expected_messages:
self.assertIn(message, result.stderr)
if no_site_packages:
self.assertIn("内置语义规则", result.stderr)
self.assertNotIn("[schema]", result.stderr)
def assert_board_accepted_in_all_modes(self, board: dict) -> None:
for no_site_packages in (False, True):
with self.subTest(no_site_packages=no_site_packages):
result = self.run_validator(
json.dumps(board),
no_site_packages=no_site_packages,
suffix=".json",
)
self.assertEqual(result.returncode, 0, result.stderr)
def test_example_with_knowledge_fields_is_valid(self) -> None:
result = self.run_validator()
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("任务板校验通过", result.stdout)
def test_approved_feishu_plan_fields_are_validated(self) -> None:
board = valid_knowledge_board()
board["tasks"][0]["fixLogic"] = "change the parser and preserve legacy input"
board["tasks"][0]["acceptanceCriteria"] = [
"the submitted value is saved",
"the original crash no longer occurs",
]
self.assert_board_accepted_in_all_modes(board)
invalid = copy.deepcopy(board)
invalid["tasks"][0]["acceptanceCriteria"] = ["valid", 7]
self.assert_board_rejected_in_all_modes(invalid, "列表项必须是字符串")
def test_reviewed_feishu_workflow_requires_fix_logic_and_approval_revision(self) -> None:
fields = {
"title": "fTitle",
"actual": "fActual",
"expected": "fExpected",
"stepsToReproduce": "fSteps",
"acceptance": "fAcceptance",
"priority": "fPriority",
"attachments": "fAttachments",
"updatedAt": "fUpdated",
}
board = valid_knowledge_board()
board["project"]["bugIntake"] = {
"provider": "feishu-base",
"workflow": "reviewed-writeback-v1",
"profile": "tenant-b",
"baseToken": "baseToken",
"tableId": "tblBugs",
"viewId": "vewReview",
"fields": fields,
}
self.assert_board_rejected_in_all_modes(
board, "reviewed-writeback-v1 必须映射",
)
board["project"]["bugIntake"]["fields"]["fixLogic"] = "fFixLogic"
board["tasks"][0].update({
"priority": "P1",
"actual": "save crashes",
"expected": "save succeeds",
"stepsToReproduce": ["open", "save"],
"fixLogic": "preserve input while fixing the parser",
"acceptanceCriteria": ["save succeeds", "the crash no longer occurs"],
})
board["tasks"][0]["source"] = {
"kind": "feishu-base",
"workflow": "reviewed-writeback-v1",
"ref": "feishu-base:sha256:" + "a" * 64,
"recordId": "recA",
"updatedAt": "2026-08-03T12:00:00+08:00",
"approvedRevision": "sha256:" + "b" * 64,
}
board["tasks"][0]["source"]["approvedPayloadHash"] = approval_payload_hash(
board["tasks"][0]
)
self.assert_board_accepted_in_all_modes(board)
board["tasks"][0]["source"]["approvedRevision"] = "latest"
self.assert_board_rejected_in_all_modes(board, "必须是 sha256 revision")
board["tasks"][0]["source"]["approvedRevision"] = "sha256:" + "b" * 64
board["tasks"][0]["fixLogic"] = "silently changed after approval"
self.assert_board_rejected_in_all_modes(board, "与任务审核字段不匹配")
board["tasks"][0]["fixLogic"] = "preserve input while fixing the parser"
board["tasks"][0]["source"]["approvedPayloadHash"] = approval_payload_hash(
board["tasks"][0]
)
board["tasks"][0]["description"] = "modify unrelated modules"
self.assert_board_rejected_in_all_modes(board, "与任务审核字段不匹配")
legacy = copy.deepcopy(board)
legacy["tasks"][0]["description"] = "validate knowledge fields"
legacy["tasks"][0]["source"] = {
"kind": "feishu-base",
"ref": "feishu-base:sha256:" + "c" * 64,
"recordId": "recLegacy",
"updatedAt": "2026-08-01T12:00:00+08:00",
}
self.assert_board_rejected_in_all_modes(legacy, "必须先迁移审核")
legacy["tasks"][0]["status"] = "verified"
self.assert_board_accepted_in_all_modes(legacy)
def test_legacy_sources_remain_open_while_feishu_sources_are_strict(self) -> None:
for legacy_source in (
"manual",
{"kind": "jira", "ref": "JIRA-123", "project": "OPS"},
):
with self.subTest(legacy_source=legacy_source):
board = valid_knowledge_board()
board["tasks"][0]["source"] = legacy_source
self.assert_board_accepted_in_all_modes(board)
valid_ref = "feishu-base:sha256:" + "a" * 64
valid = valid_knowledge_board()
valid["tasks"][0]["source"] = {
"kind": "feishu-base",
"ref": valid_ref,
"recordId": "recA",
"updatedAt": "2026-08-01T12:00:00Z",
}
self.assert_board_accepted_in_all_modes(valid)
raw = copy.deepcopy(valid)
raw["tasks"][0]["source"]["ref"] = "feishu-base:tenant:base-secret:recA"
self.assert_board_rejected_in_all_modes(
raw,
"必须是不透明 feishu-base SHA-256 引用",
)
duplicate = copy.deepcopy(valid)
second = copy.deepcopy(duplicate["tasks"][0])
second["id"] = "T-2"
duplicate["tasks"].append(second)
self.assert_board_rejected_in_all_modes(duplicate, "来源引用重复")
def test_v010_requires_structured_routing_but_v009_remains_readable(self) -> None:
current = valid_knowledge_board()
current["ackVersion"] = "0.10.0"
self.assert_board_rejected_in_all_modes(
current,
"project.orchestration: is required",
)
legacy = valid_knowledge_board()
legacy["ackVersion"] = "0.9.0"
self.assert_board_accepted_in_all_modes(legacy)
def test_builtin_rejects_invalid_ack_semver_in_all_modes(self) -> None:
for invalid in ("0.10", "00.10.0", "0.10.01", "v0.10.0", "0.10.0-"):
with self.subTest(ack_version=invalid):
board = valid_knowledge_board()
board["ackVersion"] = invalid
self.assert_board_rejected_in_all_modes(
board,
"ackVersion 必须是合法 SemVer",
)
def test_schema_declares_semver_routing_and_attempt_contracts(self) -> None:
schema = json.loads(SCHEMA.read_text(encoding="utf-8"))
ack_pattern = re.compile(schema["properties"]["ackVersion"]["pattern"])
for valid in ("0.9.0", "0.10.0", "0.11.2-alpha.1+build.7", "1.0.0"):
with self.subTest(valid_semver=valid):
self.assertIsNotNone(ack_pattern.fullmatch(valid))
for invalid in ("0.10", "00.10.0", "0.10.01", "v0.10.0", "0.10.0-"):
with self.subTest(invalid_semver=invalid):
self.assertIsNone(ack_pattern.fullmatch(invalid))
(
current_gate,
orchestration_gate,
receipts_gate,
delivery_gate,
regression_gate,
) = schema["allOf"]
current_pattern = re.compile(
current_gate["if"]["properties"]["ackVersion"]["pattern"]
)
for current in ("0.10.0", "0.99.1", "1.0.0", "12.34.56+build"):
self.assertIsNotNone(current_pattern.search(current))
self.assertIsNone(current_pattern.search("0.9.99"))
self.assertIn("workerReceipts", current_gate["then"]["required"])
self.assertIn(
"workerReceipts",
current_gate["then"]["properties"],
)
self.assertIn(
"orchestration",
current_gate["then"]["properties"]["project"]["required"],
)
self.assertIn(
"orchestration",
current_gate["then"]["properties"]["project"]["properties"],
)
self.assertIn("workerReceipts", orchestration_gate["then"]["required"])
self.assertIn(
"workerReceipts",
orchestration_gate["then"]["properties"],
)
self.assertIn(
"orchestration",
receipts_gate["then"]["properties"]["project"]["required"],
)
self.assertIn(
"orchestration",
receipts_gate["then"]["properties"]["project"]["properties"],
)
for gate in (current_gate, orchestration_gate, receipts_gate):
self.assertEqual(
gate["then"]["properties"]["tasks"]["$ref"],
"#/definitions/launchableTasks",
)
self.assertIn("deliveryRuns", delivery_gate["then"]["required"])
self.assertIn("regressionRuns", regression_gate["then"]["required"])
delivery_run = schema["definitions"]["deliveryRun"]
revision_gate = delivery_run["allOf"][0]
self.assertIn("planned", revision_gate["if"]["properties"]["status"]["enum"])
self.assertEqual(
revision_gate["then"]["properties"]["sourceRevision"]["type"],
"string",
)
role_dispatch = schema["definitions"]["roleDispatch"]
self.assertIn("attemptId", role_dispatch["required"])
receipt_rule = role_dispatch["allOf"][0]
self.assertEqual(
receipt_rule["then"]["properties"]["attemptId"]["type"],
"null",
)
self.assertEqual(
receipt_rule["else"]["properties"]["attemptId"]["type"],
"string",
)
def test_delivery_run_is_separate_and_requires_verified_tasks(self) -> None:
board = valid_manual_routing_board()
board["project"]["deliveryFile"] = "docs/ack/delivery.yaml"
board["tasks"][0]["status"] = "verified"
board["deliveryRuns"] = [
{
"id": "DR-demo-1",
"profile": "review",
"taskIds": ["T-1"],
"status": "review_ready",
"sourceRevision": "a" * 40,
"configRevision": "b" * 40,
"pullRequest": "https://forge.example/demo/pulls/1",
"artifacts": [
{
"id": "service-deb",
"type": "deb",
"reference": "demo_1.0.0_amd64.deb",
"digest": "sha256:" + "c" * 64,
}
],
"deployments": [
{
"environment": "test-server",
"result": "succeeded",
"evidence": "health endpoint returned 200",
}
],
"evidence": ["CI run 42 passed"],
"updatedAt": "2026-08-01T10:00:00+08:00",
}
]
self.assert_board_accepted_in_all_modes(board)
board["tasks"][0]["status"] = "open"
self.assert_board_rejected_in_all_modes(
board,
"delivery run 只能引用 verified 任务",
)
def test_validation_ready_delivery_run_does_not_require_pull_request(self) -> None:
board = valid_manual_routing_board()
board["project"]["deliveryFile"] = "docs/ack/delivery.yaml"
board["tasks"][0]["status"] = "verified"
board["deliveryRuns"] = [
{
"id": "DR-local-1",
"profile": "local-validation",
"taskIds": ["T-1"],
"status": "validation_ready",
"sourceRevision": "a" * 64,
"configRevision": "b" * 64,
"pullRequest": None,
"artifacts": [
{
"id": "local-service",
"type": "file",
"reference": "music-pilot",
"digest": "sha256:" + "c" * 64,
}
],
"deployments": [
{
"environment": "local-8080",
"result": "succeeded",
"evidence": "HTTP 200 and preflight passed",
}
],
"evidence": ["http://127.0.0.1:8080 ready for user validation"],
"updatedAt": "2026-08-03T23:10:00+08:00",
}
]
self.assert_board_accepted_in_all_modes(board)
def test_intent_delivery_run_allows_empty_task_ids(self) -> None:
board = valid_manual_routing_board()
board["project"]["deliveryFile"] = "docs/ack/delivery.yaml"
board["deliveryRuns"] = [
{
"id": "DR-test-env-1",
"profile": "test-local",
"intent": "testEnvironment",
"taskIds": [],
"status": "validation_ready",
"sourceRevision": "a" * 40,
"configRevision": "b" * 40,
"pullRequest": None,
"artifacts": [
{
"id": "garden-bin",
"type": "file",
"reference": "garden",
"digest": "sha256:" + "c" * 64,
}
],
"deployments": [
{
"environment": "local-write",
"result": "succeeded",
"evidence": "GET /login returned 200",
}
],
"evidence": ["http://write.localhost:8080/ ready"],
"updatedAt": "2026-08-23T00:57:00+08:00",
}
]
self.assert_board_accepted_in_all_modes(board)
del board["deliveryRuns"][0]["intent"]
self.assert_board_rejected_in_all_modes(
board,
"taskIds: 必须是非空任务 ID 列表",
)
def test_delivery_runs_and_delivery_file_must_appear_together(self) -> None:
board = valid_manual_routing_board()
board["deliveryRuns"] = []
self.assert_board_rejected_in_all_modes(
board,
"deliveryRuns 存在时 project.deliveryFile 必须存在",
)
board = valid_manual_routing_board()
board["project"]["deliveryFile"] = "docs/ack/delivery.yaml"
self.assert_board_rejected_in_all_modes(
board,
"引用 deliveryFile 的任务板必须包含 deliveryRuns 列表",
)
def test_delivery_run_binds_revisions_and_final_artifact_digest(self) -> None:
board = valid_manual_routing_board()
board["project"]["deliveryFile"] = "docs/ack/delivery.yaml"
board["deliveryRuns"] = [
{
"id": "DR-demo-2",
"profile": "review",
"taskIds": ["T-1"],
"status": "planned",
"sourceRevision": None,
"configRevision": None,
"pullRequest": None,
"artifacts": [],
"deployments": [],
"evidence": [],
"updatedAt": "2026-08-01T10:00:00+08:00",
}
]
self.assert_board_rejected_in_all_modes(
board,
"delivery run 只能引用 verified 任务",
"sourceRevision: status='planned' 时必须填写",
"configRevision: status='planned' 时必须填写",
)
board["tasks"][0]["status"] = "verified"
run = board["deliveryRuns"][0]
run.update(
{
"status": "review_ready",
"sourceRevision": "a" * 40,
"configRevision": "b" * 40,
"pullRequest": "https://forge.example/demo/pulls/2",
"artifacts": [
{
"id": "service-deb",
"type": "deb",
"reference": "demo_1.0.0_amd64.deb",
"digest": None,
}
],
"evidence": ["CI run 43 passed"],
}
)
self.assert_board_rejected_in_all_modes(
board,
"digest: status='review_ready' 时必须填写",
)
run.update(
{
"status": "skipped",
"sourceRevision": None,
"configRevision": None,
"pullRequest": None,
"artifacts": [],
"evidence": [],
}
)
self.assert_board_rejected_in_all_modes(
board,
"evidence: status='skipped' 时不能为空",
)
@unittest.skipUnless(
importlib.util.find_spec("jsonschema") is not None,
"jsonschema is required for the schema-only contract test",
)
def test_schema_alone_enforces_v010_and_mutual_routing_presence(self) -> None:
import jsonschema # type: ignore
schema = json.loads(SCHEMA.read_text(encoding="utf-8"))
validator = jsonschema.Draft7Validator(schema)
def messages(board: dict) -> list[str]:
return [error.message for error in validator.iter_errors(board)]
base = {
"version": 1,
"ackVersion": "0.9.0",
"project": {"name": "demo"},
"tasks": [],
}
self.assertEqual(messages(base), [])
for current in ("0.10.0", "0.11.2-alpha.1+build.7", "1.0.0"):
with self.subTest(ack_version=current):
board = copy.deepcopy(base)
board["ackVersion"] = current
errors = messages(board)
self.assertIn("'workerReceipts' is a required property", errors)
self.assertIn("'orchestration' is a required property", errors)
manual = {
"profileVersion": 1,
"mode": "manual",
"allowedWorktrees": [],
"modelAllowlist": {},
"profiles": {},
"defaults": {},
}
orchestration_only = copy.deepcopy(base)
orchestration_only["project"]["orchestration"] = manual
self.assertIn(
"'workerReceipts' is a required property",
messages(orchestration_only),
)
receipts_only = copy.deepcopy(base)
receipts_only["workerReceipts"] = []
self.assertIn(
"'orchestration' is a required property",
messages(receipts_only),
)
complete = copy.deepcopy(orchestration_only)
complete["ackVersion"] = "0.10.0+routing.1"
complete["workerReceipts"] = []
self.assertEqual(messages(complete), [])
def test_v010_task_ids_match_launcher_while_legacy_ids_remain_readable(
self,
) -> None:
current = valid_manual_routing_board()
current["tasks"][0]["id"] = "BUG/001"
self.assert_board_rejected_in_all_modes(
current,
"v0.10 自动路由 id 只允许字母、数字、点、下划线和连字符",
)
legacy = valid_knowledge_board()
legacy["ackVersion"] = "0.9.0"
legacy["tasks"][0]["id"] = "BUG/001"
self.assert_board_accepted_in_all_modes(legacy)
def test_knowledge_applied_and_checks_must_reference_selected_knowledge(self) -> None:
result = self.run_validator(
"""
version: 1
project:
name: demo
tasks:
- id: T-1
title: invalid refs
status: verified
knowledgeRefs: ["K-001@1"]
knowledgeApplied:
- ref: "K-002@1"
result: applied
evidence: "used the rule"
knowledgeChecks:
- ref: "K-003@1"
result: failed
evidence: "still broken"
"""
)
self.assertEqual(result.returncode, 1)
self.assertIn("K-002@1 不在 knowledgeRefs 中", result.stderr)
self.assertIn("K-003@1 不在 knowledgeRefs 中", result.stderr)
self.assertIn("verified 任务不能保留失败", result.stderr)
def test_knowledge_refs_require_revision(self) -> None:
result = self.run_validator(
"""
version: 1
project:
name: demo
tasks:
- id: T-1
title: invalid ref
status: open
knowledgeRefs: ["K-001"]
"""
)
self.assertEqual(result.returncode, 1)
self.assertIn("K-<id>@<revision>", result.stderr)
def test_candidate_requires_actionable_scope_and_evidence(self) -> None:
result = self.run_validator(
"""
version: 1
project:
name: demo
tasks:
- id: T-1
title: invalid candidate
status: open
knowledgeCandidates:
- kind: guess
title: maybe
claim: uncertain
scope: {}
appliesWhen: sometimes
directive: retry
rationale: unknown
evidenceRefs: []
"""
)
self.assertEqual(result.returncode, 1)
self.assertIn(".kind: 必须是", result.stderr)
self.assertIn(".scope: 至少包含一个非空作用域", result.stderr)
self.assertIn(".evidenceRefs: 必须是非空字符串列表", result.stderr)
def test_forbidden_knowledge_properties_fail_with_and_without_jsonschema(self) -> None:
board = valid_knowledge_board()
task = board["tasks"][0]
task["knowledgeApplied"][0]["unexpectedApplication"] = True
task["knowledgeCandidates"][0]["unexpectedCandidate"] = True
task["knowledgeChecks"][0]["unexpectedCheck"] = True
self.assert_board_rejected_in_all_modes(
board,
"knowledgeApplied[0]: 未知字段 'unexpectedApplication'",
"knowledgeCandidates[0]: 未知字段 'unexpectedCandidate'",
"knowledgeChecks[0]: 未知字段 'unexpectedCheck'",
)
def test_valid_optional_knowledge_fields_pass_in_all_modes(self) -> None:
self.assert_board_accepted_in_all_modes(valid_knowledge_board())
def test_explicit_null_knowledge_collections_fail_in_all_modes(self) -> None:
board = {
"version": 1,
"project": {"name": "demo"},
"tasks": [
{
"id": "T-REFS",
"title": "null refs",
"status": "open",
"knowledgeRefs": None,
},
{
"id": "T-APPLIED",
"title": "null applications",
"status": "open",
"knowledgeApplied": None,
},
{
"id": "T-CANDIDATES",
"title": "null candidates",
"status": "open",
"knowledgeCandidates": None,
},
{
"id": "T-CHECKS",
"title": "null checks",
"status": "open",
"knowledgeChecks": None,
},
],
}
self.assert_board_rejected_in_all_modes(
board,
"knowledgeRefs: 必须是列表",
"knowledgeApplied: 必须是列表",
"knowledgeCandidates: 必须是列表",
"knowledgeChecks: 必须是列表",
)
def test_knowledge_item_types_and_blank_evidence_fail_in_all_modes(self) -> None:
board = valid_knowledge_board()
task = board["tasks"][0]
task["knowledgeApplied"][0].update(
{"ref": 1, "result": "unknown", "evidence": " "}
)
task["knowledgeChecks"][0].update(
{"ref": False, "result": "unknown", "evidence": "\t"}
)
self.assert_board_rejected_in_all_modes(
board,
"knowledgeApplied[0].ref: 必须使用 K-<id>@<revision> 格式",
"knowledgeApplied[0].result: 必须是 applied/not_applicable",
"knowledgeApplied[0].evidence: 必须提供非空证据",
"knowledgeChecks[0].ref: 必须使用 K-<id>@<revision> 格式",
"knowledgeChecks[0].result: 必须是",
"knowledgeChecks[0].evidence: 必须提供非空证据",
)
def test_candidate_text_scope_and_evidence_refs_fail_in_all_modes(self) -> None:
board = valid_knowledge_board()
candidate = board["tasks"][0]["knowledgeCandidates"][0]
candidate.update(
{
"title": 1,
"claim": " ",
"appliesWhen": [],
"directive": "",
"rationale": None,
"scope": {
"components": ["web", "web"],
"paths": [" "],
},
"evidenceRefs": ["tasks.yaml#T-1", "tasks.yaml#T-1"],
}
)
self.assert_board_rejected_in_all_modes(
board,
"knowledgeCandidates[0].title: 必须是非空字符串",
"knowledgeCandidates[0].claim: 必须是非空字符串",
"knowledgeCandidates[0].appliesWhen: 必须是非空字符串",
"knowledgeCandidates[0].directive: 必须是非空字符串",
"knowledgeCandidates[0].rationale: 必须是非空字符串",
"knowledgeCandidates[0].scope.components: 不能包含重复值",
"knowledgeCandidates[0].scope.paths: 必须是非空字符串列表",
"knowledgeCandidates[0].evidenceRefs: 不能包含重复值",
)
def test_candidate_evidence_refs_reject_blank_strings_in_all_modes(self) -> None:
board = valid_knowledge_board()
board["tasks"][0]["knowledgeCandidates"][0]["evidenceRefs"] = [" "]
self.assert_board_rejected_in_all_modes(
board,
"knowledgeCandidates[0].evidenceRefs: 必须是非空字符串列表",
)
def test_optional_knowledge_field_types_fail_in_all_modes(self) -> None:
board = valid_knowledge_board()
candidate = board["tasks"][0]["knowledgeCandidates"][0]
candidate["proposedBy"] = 1
candidate["proposedAt"] = []
check = board["tasks"][0]["knowledgeChecks"][0]
check["checkedBy"] = False
check["checkedAt"] = {}
self.assert_board_rejected_in_all_modes(
board,
"knowledgeCandidates[0].proposedBy: 必须是字符串",
"knowledgeCandidates[0].proposedAt: 必须是字符串",
"knowledgeChecks[0].checkedBy: 必须是字符串",
"knowledgeChecks[0].checkedAt: 必须是字符串",
)
def test_attempt_id_must_match_task_and_round_and_be_unique(self) -> None:
board = {
"version": 1,
"project": {"name": "demo"},
"tasks": [
{
"id": "BUG-017",
"title": "invalid attempt ids",
"status": "failed_retest",
"dispatch": {
"rounds": [
{
"round": 1,
"attemptId": "OTHER-A1",
"result": "failed",
},
{
"round": 1,
"attemptId": "OTHER-A1",
"result": "failed",
},
{
"round": 3,
"attemptId": "invalid/attempt",
"result": "failed",
},
]
},
}
],
}
self.assert_board_rejected_in_all_modes(
board,
"dispatch.rounds[0].attemptId: 应为 BUG-017-A1",
"dispatch.rounds[1].attemptId: 轮次内不能重复: OTHER-A1",
"dispatch.rounds[2].attemptId: 必须使用 <task-id>-A<round> 格式",
)
def test_valid_optional_attempt_ids_pass_in_all_modes(self) -> None:
board = {
"version": 1,
"project": {"name": "demo"},
"tasks": [
{
"id": "BUG-017",
"title": "valid attempt ids",
"status": "failed_retest",
"dispatch": {
"rounds": [
{
"round": 1,
"attemptId": "BUG-017-A1",
"result": "failed",
},
{
"round": 2,
"attemptId": "BUG-017-A2",
"result": "failed",
},
]
},
}
],
}
self.assert_board_accepted_in_all_modes(board)
def test_boolean_version_fails_in_all_modes(self) -> None:
board = valid_knowledge_board()
board["version"] = True
self.assert_board_rejected_in_all_modes(
board,
"version 必须是 >=1 的整数",
)
def test_knowledge_file_is_fixed_in_all_modes(self) -> None:
board = valid_knowledge_board()
board["project"]["knowledgeFile"] = "docs/ack/alternate.yaml"
self.assert_board_rejected_in_all_modes(
board,
"project.knowledgeFile 必须固定为 docs/ack/knowledge.yaml",
)
def test_basic_identifiers_must_be_nonempty_strings_in_all_modes(self) -> None:
board = valid_knowledge_board()
board["project"]["name"] = 7
board["tasks"][0]["id"] = 9
board["tasks"][0]["title"] = " "
self.assert_board_rejected_in_all_modes(
board,
"project.name 必须是非空字符串",
"id 必须是非空字符串",
"title 必须是非空字符串",
)
def test_root_project_and_summary_types_match_schema_in_all_modes(self) -> None:
board = valid_knowledge_board()
board.update(
{
"updatedAt": [],
"source": {},
"ackVersion": 1,
"kitVersion": False,
"summary": {
"verified": [1],
"open": {},
"failedRetest": [False],
"leftovers": None,
},
"statusReference": [],
}
)
board["project"].update(
{
"repoPath": [],
"baseUrl": {},
"devWorktree": 1,
"overlayFile": False,
}
)
self.assert_board_rejected_in_all_modes(
board,
"<root>.updatedAt: 必须是字符串",
"<root>.source: 必须是字符串",
"<root>.ackVersion: 必须是字符串",
"<root>.kitVersion: 必须是字符串",
"project.repoPath: 必须是字符串",
"project.baseUrl: 必须是字符串",
"project.devWorktree: 必须是字符串",
"project.overlayFile: 必须是字符串",
"summary.verified: 列表项必须是字符串",
"summary.open: 必须是列表",
"summary.failedRetest: 列表项必须是字符串",
"summary.leftovers: 必须是列表",
"statusReference 必须是对象",
)
def test_task_optional_types_match_schema_in_all_modes(self) -> None:
board = valid_knowledge_board()
board["tasks"][0].update(
{
"type": [],
"priority": {},
"assignee": False,
"component": 1,
"specRefs": {},
"testRefs": [1],
"description": [],
"stepsToReproduce": [{}],
"expected": False,
"actual": None,
"evidence": [],
"verification": [],
}
)
self.assert_board_rejected_in_all_modes(
board,
".type: 必须是字符串",
".priority: 必须是字符串",
".assignee: 必须是字符串",
".component: 必须是字符串",
".specRefs: 必须是列表",
".testRefs: 列表项必须是字符串",
".description: 必须是字符串",
".stepsToReproduce: 列表项必须是字符串",
".expected: 必须是字符串",
".actual: 必须是字符串",
".evidence: 必须是对象",
".verification: 必须是对象",
)
def test_dispatch_and_resolution_types_match_schema_in_all_modes(self) -> None:
board = valid_knowledge_board()
board["tasks"][0].update(
{
"dispatch": {
"taskId": [],
"dispatchId": {},
"worker": False,
"rounds": [
{
"round": 1,
"result": "failed",
"evidence": [],
}
],
},
"resolution": {
"fixedBy": [],
"verifiedBy": {},
"verifiedAt": False,
"leftoverReason": 1,
"evidence": [],
},
}
)
self.assert_board_rejected_in_all_modes(
board,
".dispatch.taskId: 必须是字符串或 null",
".dispatch.dispatchId: 必须是字符串或 null",
".dispatch.worker: 必须是字符串或 null",
".dispatch.rounds[0].evidence: 必须是字符串",
".resolution.fixedBy: 必须是字符串或 null",
".resolution.verifiedBy: 必须是字符串或 null",
".resolution.verifiedAt: 必须是字符串或 null",
".resolution.leftoverReason: 必须是字符串或 null",
".resolution.evidence: 必须是对象",
)
def test_explicit_null_structures_fail_in_all_modes(self) -> None:
board = {
"version": 1,
"project": {"name": "demo"},
"tasks": [
{
"id": "T-DISPATCH",
"title": "invalid dispatch",
"status": "open",
"dispatch": None,
},
{
"id": "T-ROUNDS",
"title": "invalid rounds",
"status": "open",
"dispatch": {"rounds": None},
},
{
"id": "T-RESOLUTION",
"title": "invalid resolution",
"status": "open",
"resolution": None,
},
],
}
self.assert_board_rejected_in_all_modes(
board,
"T-DISPATCH.dispatch: 必须是对象",
"T-ROUNDS.dispatch.rounds: 必须是列表",
"T-RESOLUTION.resolution: 必须是对象",
)
def test_dispatch_rejects_free_command_field_without_jsonschema(self) -> None:
board = valid_knowledge_board()
board["tasks"][0]["dispatch"] = {
"command": "cursor-agent --yolo",
"rounds": [],
}
self.assert_board_rejected_in_all_modes(
board,
"dispatch: 未知字段 'command'",
)
def test_valid_optional_schema_fields_pass_in_all_modes(self) -> None:
board = valid_knowledge_board()
board.update(
{
"updatedAt": "2026-07-31T10:00:00+08:00",
"source": "manual",
"ackVersion": "0.9.0",
"kitVersion": "0.8.0",
"summary": {
"verified": ["T-1"],
"open": [],
"failedRetest": [],
"leftovers": [],
},
"statusReference": {},
}
)
board["project"].update(
{
"repoPath": "/repo",
"baseUrl": "http://127.0.0.1:3000",
"devWorktree": "/repo-dev",
"overlayFile": "docs/ack/project.md",
"knowledgeFile": "docs/ack/knowledge.yaml",
}
)
board["tasks"][0].update(
{
"type": "bug",
"priority": "P1",
"assignee": "developer",
"component": "web",
"specRefs": ["spec.md"],
"testRefs": ["tests/test_web.py"],
"description": "description",
"stepsToReproduce": ["open page"],
"expected": "works",
"actual": "fails",
"evidence": {},
"verification": {},
"dispatch": {
"taskId": "orca-task",
"dispatchId": None,
"worker": "worker-1",
"rounds": [
{
"round": 1,
"attemptId": "T-1-A1",
"result": "failed",
"evidence": "test output",
}
],
},
"resolution": {
"fixedBy": "developer",
"verifiedBy": None,
"verifiedAt": None,
"leftoverReason": None,
"evidence": {},
},
}
)
self.assert_board_accepted_in_all_modes(board)
def test_round_numbers_must_be_contiguous_and_within_budget(self) -> None:
board = {
"version": 1,
"project": {"name": "demo"},
"tasks": [
{
"id": "BUG-017",
"title": "invalid round number",
"status": "failed_retest",
"dispatch": {
"rounds": [
{
"round": 999,
"attemptId": "BUG-017-A999",
"result": "failed",
}
]
},
}
],
}
self.assert_board_rejected_in_all_modes(
board,
"dispatch.rounds[0].round: 必须是 1..3 的整数",
"dispatch.rounds: round 必须从 1 连续递增且不重复",
)
def test_environment_incidents_do_not_consume_round_budget(self) -> None:
board = valid_knowledge_board()
board["tasks"][0]["status"] = "fixed_by_dev"
board["tasks"][0]["dispatch"] = {
"rounds": [],
"environmentIncidents": [
{
"id": f"T-1-ENV-{index}",
"attemptId": f"T-1-A{index}",
"role": "test",
"phase": "browser",
"status": "resolved",
"summary": "browser runtime was unavailable",
"evidence": "browser executable lookup returned no result",
"impact": "interactive acceptance signals were not evaluated",
"recoveryAction": "launch a network-enabled fresh Test worker",
"userAction": "none; Coordinator continues the recovery",
"reportedAt": "2026-08-03T20:00:00+08:00",
"resolvedAt": "2026-08-03T20:05:00+08:00",
}
for index in range(1, 5)
],
}
self.assert_board_accepted_in_all_modes(board)
def test_environment_incidents_require_actionable_reporting(self) -> None:
board = valid_knowledge_board()
board["tasks"][0]["dispatch"] = {
"rounds": [],
"environmentIncidents": [
{
"id": "WRONG-ENV-9",
"role": "observer",
"phase": "unknown",
"status": "resolved",
"summary": "",
"evidence": "",
"impact": "",
"recoveryAction": "",
"userAction": "",
"reportedAt": "",
}
],
}
self.assert_board_rejected_in_all_modes(
board,
"environmentIncidents[0].id: 应为 T-1-ENV-1",
"environmentIncidents[0].role: 必须是 coordinator/developer/test",
"environmentIncidents[0].phase: 非法环境阶段",
"environmentIncidents[0].summary: 必须是非空字符串",
"environmentIncidents[0].userAction: 必须是非空字符串",
"environmentIncidents[0]: resolved 必须填写 resolvedAt",
)
def test_leftover_reason_must_be_nonempty_string_in_all_modes(self) -> None:
board = {
"version": 1,
"project": {"name": "demo"},
"tasks": [
{
"id": "T-1",
"title": "invalid leftover reason",
"status": "leftover",
"resolution": {"leftoverReason": True},
}
],
}
self.assert_board_rejected_in_all_modes(
board,
"leftover 必须填 resolution.leftoverReason",
)
def test_explicit_missing_schema_is_an_environment_error(self) -> None:
result = self.run_validator(None, "--schema", "/definitely/missing/schema.json")
self.assertEqual(result.returncode, 2)
self.assertIn("找不到指定的 schema 文件", result.stderr)
if __name__ == "__main__":
unittest.main()