#!/usr/bin/env python3 """校验 ACK 项目知识护栏库及其任务引用。 权威结构位于 templates/knowledge.schema.json。jsonschema 是可选依赖；无论是否 安装，引用、作用域、复核、时效和冲突等语义规则都会执行。本脚本只解析数据， 不会执行 directive、verification 或其它自由文本中的命令。 用法: python3 validate_knowledge.py [knowledge.yaml] python3 validate_knowledge.py knowledge.yaml --tasks tasks.yaml python3 validate_knowledge.py --schema path/to/knowledge.schema.json knowledge.yaml 退出码: 0 通过 / 1 校验失败 / 2 环境或用法错误。 """ from __future__ import annotations import argparse import json import re import stat import sys from datetime import datetime from pathlib import Path from typing import Any from yaml_subset import ( DuplicateKeyError, YamlSubsetError, load_json_unique, load_yaml_subset, make_unique_pyyaml_loader, ) KINDS = {"guardrail", "pitfall", "verification"} STATUSES = {"active", "stale", "superseded", "archived"} CHECK_RESULTS = {"passed", "failed", "not_applicable"} APPLIED_RESULTS = {"applied", "not_applicable"} TERMINAL_TASK_STATUSES = {"verified", "leftover"} MAX_ROUNDS = 3 ID_RE = re.compile(r"^K-[A-Z0-9][A-Z0-9-]*$") REF_RE = re.compile(r"^(K-[A-Z0-9][A-Z0-9-]*)@([1-9][0-9]*)$") SUBJECT_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$") VERIFICATION_REF_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") SAFE_RELATIVE_PATH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]*$") SCOPE_FIELDS = ( "components", "paths", "dependencies", "versions", "tags", "symbols", "errorSignatures", ) FORBIDDEN_EXECUTION_KEYS = { "argv", "command", "commands", "executable", "script", "shell", } SECRET_PATTERNS = ( ( "private key", re.compile(r"-----BEGIN [A-Z0-9 ]*PRIVATE KEY-----"), ), ( "GitHub token", re.compile(r"\bgh[pousr]_[A-Za-z0-9]{20,}\b"), ), ( "OpenAI-style token", re.compile(r"\bsk-[A-Za-z0-9_-]{20,}\b"), ), ( "AWS access key", re.compile(r"\b(?:AKIA|ASIA)[A-Z0-9]{16}\b"), ), ( "URL credentials", re.compile(r"https?://[^/\s:@]+:[^/\s@]+@"), ), ( "inline secret assignment", re.compile( r"(?i)\b(?:api[_-]?key|access[_-]?token|password|secret|token)" r"\s*[:=]\s*[\"']?[^\s,\"']{8,}" ), ), ) ENTRY_REQUIRED = { "id", "revision", "kind", "status", "title", "subject", "scope", "appliesWhen", "directive", "rationale", "verification", "provenance", "owner", "author", "reviewer", "approval", "createdAt", "lastValidatedAt", "reviewAfter", "temporary", "removalCondition", "statusReason", "supersedes", "conflictsWith", } def load_yaml_text(content: str, label: str = "YAML") -> dict[str, Any]: try: import yaml # type: ignore except ImportError: try: data = load_yaml_subset(content) except YamlSubsetError as exc: sys.stderr.write(f"{label} 子集解析失败: {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"{label} 解析失败: {exc}\n") raise SystemExit(1) if not isinstance(data, dict): sys.stderr.write(f"{label} 顶层必须是对象（mapping）\n") raise SystemExit(1) return data def load_yaml(path: Path, label: str = "YAML") -> dict[str, Any]: try: content = path.read_text(encoding="utf-8") except OSError as exc: sys.stderr.write(f"{label} 读取失败: {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"{label} JSON 解析失败: {exc}\n") raise SystemExit(1) if not isinstance(data, dict): sys.stderr.write(f"{label} 顶层必须是对象（mapping）\n") raise SystemExit(1) return data return load_yaml_text(content, label) def stable_ref(entry: dict[str, Any]) -> str | None: entry_id = entry.get("id") revision = entry.get("revision") if ( isinstance(entry_id, str) and ID_RE.fullmatch(entry_id) and isinstance(revision, int) and not isinstance(revision, bool) and revision >= 1 ): return f"{entry_id}@{revision}" return None def _nonempty(value: Any) -> bool: return isinstance(value, str) and bool(value.strip()) def _timestamp(value: Any) -> datetime | None: if not isinstance(value, str): return None try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: return None return parsed if parsed.tzinfo is not None else None def infer_project_root(document_path: Path) -> Path | None: """从标准 .pouch/ack 布局或 Git marker 推断项目根，不解析文档 symlink。""" lexical = document_path.expanduser().absolute() parent = lexical.parent if parent.name == "ack" and parent.parent.name in {".pouch", "docs"}: return parent.parent.parent.resolve() for candidate in (parent, *parent.parents): if (candidate / ".git").exists(): return candidate.resolve() return None def _is_within(candidate: Path, root: Path) -> bool: try: candidate.relative_to(root) except ValueError: return False return True def _validate_registry_target( project_root: Path, relative_path: str, where: str, ) -> list[str]: """Mirror runner path policy without executing or following symlinks.""" errors: list[str] = [] current = project_root parts = relative_path.split("/") for index, segment in enumerate(parts): current = current / segment try: metadata = current.lstat() except FileNotFoundError: return [f"{where}: 检查目标不存在: {relative_path!r}"] except OSError as exc: return [f"{where}: 检查目标不可访问: {relative_path!r}: {exc}"] if stat.S_ISLNK(metadata.st_mode): return [f"{where}: 检查目标路径不能包含 symlink: {relative_path!r}"] if index < len(parts) - 1 and not stat.S_ISDIR(metadata.st_mode): return [f"{where}: 中间路径不是目录: {relative_path!r}"] if not stat.S_ISREG(metadata.st_mode): errors.append(f"{where}: 检查目标必须是普通文件: {relative_path!r}") elif metadata.st_mode & 0o111 == 0: errors.append(f"{where}: 检查目标不可执行: {relative_path!r}") return errors def _tasks_project_root( tasks_data: dict[str, Any], tasks_path: Path | None ) -> Path | None: inferred = infer_project_root(tasks_path) if tasks_path else None if inferred is not None and inferred.is_dir(): return inferred # Legacy task boards may still declare repoPath. It is only a fallback for # non-standard layouts; .pouch/ack location is authoritative when available. project = tasks_data.get("project") repo_path = project.get("repoPath") if isinstance(project, dict) else None if _nonempty(repo_path): candidate = Path(repo_path).expanduser() if not candidate.is_absolute(): if inferred is None: return None candidate = inferred / candidate if not candidate.is_dir(): return None return candidate.resolve(strict=True) return None def _unknown_keys(value: dict[str, Any], allowed: set[str], where: str) -> list[str]: return [f"{where}: 未知字段 {key!r}" for key in sorted(set(value) - allowed)] def validate_builtin_structure(data: dict[str, Any]) -> list[str]: """jsonschema 不可用时覆盖关键结构规则。""" errors: list[str] = [] errors.extend( _unknown_keys( data, {"version", "updatedAt", "project", "verificationRegistry", "entries"}, "", ) ) version = data.get("version") if not isinstance(version, int) or isinstance(version, bool) or version < 1: errors.append("version 必须是 >=1 的整数") if _timestamp(data.get("updatedAt")) is None: errors.append("updatedAt 必须是带时区的 ISO 8601 时间") project = data.get("project") if not isinstance(project, dict): errors.append("project 必须是对象") else: errors.extend(_unknown_keys(project, {"name"}, "project")) if not _nonempty(project.get("name")): errors.append("project.name 必填") registry = data.get("verificationRegistry") if not isinstance(registry, dict): errors.append("verificationRegistry 必须是对象") else: for ref, target in registry.items(): where = f"verificationRegistry.{ref}" if not isinstance(ref, str) or not VERIFICATION_REF_RE.fullmatch(ref): errors.append(f"{where}: 检查 ID 格式非法") if not isinstance(target, dict): errors.append(f"{where}: 必须是对象") continue errors.extend(_unknown_keys(target, {"path", "args"}, where)) if not _nonempty(target.get("path")): errors.append(f"{where}.path: 必须是非空字符串") args = target.get("args") if not isinstance(args, list) or any( not isinstance(arg, str) for arg in args ): errors.append(f"{where}.args: 必须是结构化字符串数组") entries = data.get("entries") if not isinstance(entries, list): errors.append("entries 必须是列表") return errors for index, entry in enumerate(entries): where = f"entries[{index}]" if not isinstance(entry, dict): errors.append(f"{where}: 必须是对象") continue errors.extend(_unknown_keys(entry, ENTRY_REQUIRED, where)) for field in sorted(ENTRY_REQUIRED - set(entry)): errors.append(f"{where}: {field} 必填") entry_id = entry.get("id") if not isinstance(entry_id, str) or not ID_RE.fullmatch(entry_id): errors.append(f"{where}.id: 必须匹配 K-[A-Z0-9-]+") revision = entry.get("revision") if ( not isinstance(revision, int) or isinstance(revision, bool) or revision < 1 ): errors.append(f"{where}.revision: 必须是 >=1 的整数") if entry.get("kind") not in KINDS: errors.append(f"{where}.kind: 应为 {sorted(KINDS)}") if entry.get("status") not in STATUSES: errors.append(f"{where}.status: 应为 {sorted(STATUSES)}") if not isinstance(entry.get("temporary"), bool): errors.append(f"{where}.temporary: 必须是布尔值") for field in ( "title", "appliesWhen", "directive", "rationale", "owner", "author", "reviewer", ): if not _nonempty(entry.get(field)): errors.append(f"{where}.{field}: 必须是非空字符串") subject = entry.get("subject") if not isinstance(subject, str) or not SUBJECT_RE.fullmatch(subject): errors.append(f"{where}.subject: 必须是小写连字符标识") scope = entry.get("scope") if not isinstance(scope, dict): errors.append(f"{where}.scope: 必须是对象") else: allowed = {"all", *SCOPE_FIELDS} errors.extend(_unknown_keys(scope, allowed, f"{where}.scope")) if not isinstance(scope.get("all"), bool): errors.append(f"{where}.scope.all: 必须是布尔值") for field in SCOPE_FIELDS: values = scope.get(field) if not isinstance(values, list): errors.append(f"{where}.scope.{field}: 必须是列表") elif ( any(not _nonempty(item) for item in values) or len(values) != len(set(values)) ): errors.append( f"{where}.scope.{field}: 必须是非空且不重复的字符串列表" ) verification = entry.get("verification") if not isinstance(verification, dict): errors.append(f"{where}.verification: 必须是对象") else: errors.extend( _unknown_keys( verification, {"ref", "expected"}, f"{where}.verification" ) ) verification_ref = verification.get("ref") if ( not isinstance(verification_ref, str) or not VERIFICATION_REF_RE.fullmatch(verification_ref) ): errors.append( f"{where}.verification.ref: 必须是检查 ID，不能是命令" ) if not _nonempty(verification.get("expected")): errors.append(f"{where}.verification.expected: 必填") provenance = entry.get("provenance") provenance_fields = {"taskId", "attemptId", "codeRef", "evidenceRef"} if not isinstance(provenance, dict): errors.append(f"{where}.provenance: 必须是对象") else: errors.extend( _unknown_keys(provenance, provenance_fields, f"{where}.provenance") ) for field in sorted(provenance_fields): if not _nonempty(provenance.get(field)): errors.append(f"{where}.provenance.{field}: 必填") approval = entry.get("approval") approval_fields = {"approvedBy", "approvedAt", "evidenceRef"} if approval is not None: if not isinstance(approval, dict): errors.append(f"{where}.approval: 必须为 null 或对象") else: errors.extend( _unknown_keys(approval, approval_fields, f"{where}.approval") ) for field in ("approvedBy", "evidenceRef"): if not _nonempty(approval.get(field)): errors.append(f"{where}.approval.{field}: 必填") if _timestamp(approval.get("approvedAt")) is None: errors.append( f"{where}.approval.approvedAt: 必须是带时区的 ISO 8601 时间" ) for field in ("createdAt", "lastValidatedAt"): if _timestamp(entry.get(field)) is None: errors.append(f"{where}.{field}: 必须是带时区的 ISO 8601 时间") if entry.get("reviewAfter") is not None and _timestamp( entry.get("reviewAfter") ) is None: errors.append( f"{where}.reviewAfter: 必须为 null 或带时区的 ISO 8601 时间" ) for field in ("removalCondition", "statusReason"): if entry.get(field) is not None and not _nonempty(entry.get(field)): errors.append(f"{where}.{field}: 必须为 null 或非空字符串") for field in ("supersedes", "conflictsWith"): refs = entry.get(field) if not isinstance(refs, list): errors.append(f"{where}.{field}: 必须是列表") elif ( any(not isinstance(ref, str) or not REF_RE.fullmatch(ref) for ref in refs) or len(refs) != len(set(refs)) ): errors.append( f"{where}.{field}: 必须是不重复的 K-...@revision 引用列表" ) return errors def _find_forbidden_keys(value: Any, path: str = "") -> list[str]: errors: list[str] = [] if isinstance(value, dict): for key, child in value.items(): child_path = f"{path}.{key}" if str(key).casefold() in FORBIDDEN_EXECUTION_KEYS: errors.append( f"{child_path}: knowledge.yaml 不保存或执行自由命令；请使用 verification.ref" ) errors.extend(_find_forbidden_keys(child, child_path)) elif isinstance(value, list): for index, child in enumerate(value): errors.extend(_find_forbidden_keys(child, f"{path}[{index}]")) return errors def _find_secrets(value: Any, path: str = "") -> list[str]: errors: list[str] = [] if isinstance(value, dict): for key, child in value.items(): errors.extend(_find_secrets(child, f"{path}.{key}")) elif isinstance(value, list): for index, child in enumerate(value): errors.extend(_find_secrets(child, f"{path}[{index}]")) elif isinstance(value, str): for label, pattern in SECRET_PATTERNS: if pattern.search(value): errors.append( f"{path}: 疑似包含 {label}，知识库只能保存脱敏摘要和证据引用" ) return errors def _canonical_scope(scope: Any) -> tuple[Any, ...] | None: if not isinstance(scope, dict): return None values: list[Any] = [scope.get("all")] for field in SCOPE_FIELDS: field_values = scope.get(field) if not isinstance(field_values, list): return None values.append(tuple(sorted(item for item in field_values if isinstance(item, str)))) return tuple(values) def validate_semantics( data: dict[str, Any], *, project_root: Path | None = None ) -> list[str]: """始终执行的跨字段和跨条目语义校验。""" errors = _find_forbidden_keys(data) errors.extend(_find_secrets(data)) project = data.get("project") if isinstance(project, dict) and not _nonempty(project.get("name")): errors.append("project.name 必填") entries = data.get("entries") if not isinstance(entries, list): return errors registry = data.get("verificationRegistry") resolved_root: Path | None = None if project_root is not None and project_root.is_dir(): try: resolved_root = project_root.resolve(strict=True) except OSError as exc: errors.append(f"项目根目录无法解析: {exc}") if isinstance(registry, dict): for ref, target in registry.items(): if not isinstance(target, dict): continue path = target.get("path") if not isinstance(path, str): continue segments = path.replace("\\", "/").split("/") lexically_invalid = ( path.startswith(("/", "\\")) or re.match(r"^[A-Za-z]:", path) or "\\" in path or ".." in segments or not SAFE_RELATIVE_PATH_RE.fullmatch(path) ) if lexically_invalid: errors.append( f"verificationRegistry.{ref}.path: {path!r} 必须是仓库内相对路径" ) elif not _nonempty(path): errors.append(f"verificationRegistry.{ref}.path: 必须是非空字符串") elif resolved_root is not None: errors.extend( _validate_registry_target( resolved_root, path, f"verificationRegistry.{ref}.path", ) ) by_ref: dict[str, dict[str, Any]] = {} locations: dict[str, str] = {} active_by_id: dict[str, str] = {} active_subject_scopes: dict[tuple[str, tuple[Any, ...]], str] = {} superseded_targets: set[str] = set() for index, entry in enumerate(entries): if not isinstance(entry, dict): continue where = f"entries[{index}]" for field in ( "title", "appliesWhen", "directive", "rationale", "owner", "author", "reviewer", ): if not _nonempty(entry.get(field)): errors.append(f"{where}.{field}: 必须是非空字符串") verification = entry.get("verification") if isinstance(verification, dict) and not _nonempty( verification.get("expected") ): errors.append(f"{where}.verification.expected: 必填") provenance = entry.get("provenance") if isinstance(provenance, dict): for field in ("taskId", "attemptId", "codeRef", "evidenceRef"): if not _nonempty(provenance.get(field)): errors.append(f"{where}.provenance.{field}: 必填") approval = entry.get("approval") if isinstance(approval, dict): for field in ("approvedBy", "evidenceRef"): if not _nonempty(approval.get(field)): errors.append(f"{where}.approval.{field}: 必填") ref = stable_ref(entry) if ref: if ref in by_ref: errors.append(f"{where}: 稳定引用 {ref} 重复（首次位于 {locations[ref]}）") else: by_ref[ref] = entry locations[ref] = where if entry.get("status") == "active" and isinstance(entry.get("id"), str): previous = active_by_id.get(entry["id"]) if previous: errors.append( f"{where}: {entry['id']} 同时存在多个 active revision（{previous}, {ref}）" ) elif ref: active_by_id[entry["id"]] = ref scope = entry.get("scope") if isinstance(scope, dict): populated = [ field for field in SCOPE_FIELDS if isinstance(scope.get(field), list) and scope.get(field) ] if scope.get("all") is True and populated: errors.append(f"{where}.scope: all=true 时不能同时填写作用域维度") if scope.get("all") is not True and not populated: errors.append(f"{where}.scope: 必须设置 all=true 或至少一个作用域维度") if ( scope.get("all") is True and entry.get("status") == "active" and not isinstance(entry.get("approval"), dict) ): errors.append( f"{where}: 全项目 active 知识必须记录 Decision Owner approval" ) paths = scope.get("paths") if isinstance(paths, list): for pattern in paths: if not isinstance(pattern, str): continue segments = pattern.replace("\\", "/").split("/") if ( pattern.startswith(("/", "\\")) or re.match(r"^[A-Za-z]:", pattern) or "\\" in pattern or ".." in segments ): errors.append( f"{where}.scope.paths: {pattern!r} 必须是仓库内相对 glob" ) created = _timestamp(entry.get("createdAt")) validated = _timestamp(entry.get("lastValidatedAt")) review_after = _timestamp(entry.get("reviewAfter")) if created and validated and validated < created: errors.append(f"{where}: lastValidatedAt 不能早于 createdAt") if validated and review_after and review_after <= validated: errors.append(f"{where}: reviewAfter 必须晚于 lastValidatedAt") if ( entry.get("status") == "active" and review_after is not None and review_after <= datetime.now().astimezone() ): errors.append( f"{where}: active 条目已超过 reviewAfter，必须重新验证或标记 stale" ) if ( entry.get("status") == "active" and _nonempty(entry.get("author")) and entry.get("author") == entry.get("reviewer") ): errors.append(f"{where}: active 条目的 reviewer 必须独立于 author") if entry.get("temporary") is True: if not _nonempty(entry.get("removalCondition")): errors.append(f"{where}: temporary 条目必须填写 removalCondition") if review_after is None: errors.append(f"{where}: temporary 条目必须填写 reviewAfter") if entry.get("status") in {"stale", "superseded", "archived"} and not _nonempty( entry.get("statusReason") ): errors.append(f"{where}: 非 active 条目必须填写 statusReason") verification = entry.get("verification") if isinstance(verification, dict) and _nonempty(verification.get("ref")): verification_ref = verification["ref"] if not isinstance(registry, dict) or verification_ref not in registry: errors.append( f"{where}.verification.ref: {verification_ref!r} 未在 verificationRegistry 注册" ) if ( entry.get("status") == "active" and isinstance(entry.get("subject"), str) and (scope_key := _canonical_scope(scope)) is not None ): key = (entry["subject"], scope_key) previous = active_subject_scopes.get(key) if previous: errors.append( f"{where}: 相同 subject 和 scope 已有 active 条目 {previous}" ) elif ref: active_subject_scopes[key] = ref refs = entry.get("supersedes") if isinstance(refs, list): superseded_targets.update(ref for ref in refs if isinstance(ref, str)) for ref, entry in by_ref.items(): for field in ("supersedes", "conflictsWith"): related = entry.get(field) if not isinstance(related, list): continue for target_ref in related: if not isinstance(target_ref, str) or not REF_RE.fullmatch(target_ref): continue if target_ref == ref: errors.append(f"{locations[ref]}.{field}: 不能引用自身 {ref}") continue target = by_ref.get(target_ref) if target is None: errors.append( f"{locations[ref]}.{field}: 找不到精确版本 {target_ref}" ) continue if field == "supersedes" and target.get("status") != "superseded": errors.append( f"{locations[ref]}.supersedes: {target_ref} 必须标记为 superseded" ) if ( field == "conflictsWith" and entry.get("status") == "active" and target.get("status") == "active" ): errors.append( f"{locations[ref]}.conflictsWith: active 条目不能与 active {target_ref} 冲突" ) for ref, entry in by_ref.items(): if entry.get("status") == "superseded" and ref not in superseded_targets: errors.append(f"{locations[ref]}: superseded 条目必须被其它 revision 显式引用") return errors def validate_with_schema( data: dict[str, Any], schema_path: Path ) -> list[str]: import jsonschema # type: ignore schema = json.loads(schema_path.read_text(encoding="utf-8")) validator = jsonschema.Draft7Validator( schema, format_checker=jsonschema.FormatChecker() ) errors: list[str] = [] for error in sorted(validator.iter_errors(data), key=lambda item: list(item.path)): location = "/".join(str(part) for part in error.path) or "" errors.append(f"[schema] {location}: {error.message}") return errors def validate_all( data: dict[str, Any], schema_path: Path, *, use_schema: bool | None = None, project_root: Path | None = None, ) -> tuple[list[str], str]: """运行 schema/fallback，并且无条件叠加语义校验。""" if use_schema is None: try: import jsonschema # type: ignore # noqa: F401 except ImportError: use_schema = False else: use_schema = schema_path.is_file() errors = validate_builtin_structure(data) if use_schema and schema_path.is_file(): errors = validate_with_schema(data, schema_path) + errors mode = f"schema ({schema_path.name}) + 内置结构 + 内置语义" else: mode = "内置结构 + 内置语义" errors.extend(validate_semantics(data, project_root=project_root)) return list(dict.fromkeys(errors)), mode def _validate_knowledge_file_binding( tasks_data: dict[str, Any], *, knowledge_path: Path | None, tasks_path: Path | None, project_root: Path | None, project_root_is_explicit: bool, ) -> list[str]: errors: list[str] = [] project = tasks_data.get("project") if not isinstance(project, dict): return ["[tasks] project 必须是对象，无法核对 knowledgeFile"] knowledge_file = project.get("knowledgeFile") if not _nonempty(knowledge_file): return ["[tasks] project.knowledgeFile 必填"] if knowledge_file != ".pouch/ack/knowledge.yaml": errors.append( "[tasks] project.knowledgeFile 必须固定为 " "'.pouch/ack/knowledge.yaml'" ) relative = Path(knowledge_file) segments = knowledge_file.replace("\\", "/").split("/") if ( relative.is_absolute() or re.match(r"^[A-Za-z]:", knowledge_file) or "\\" in knowledge_file or ".." in segments ): errors.append("[tasks] project.knowledgeFile 必须是项目内相对路径") return errors if knowledge_path is None or tasks_path is None: return errors declared_repo_path = project.get("repoPath") declared_root = _tasks_project_root(tasks_data, tasks_path) if ( _nonempty(declared_repo_path) and declared_root is None and not project_root_is_explicit ): errors.append( f"[tasks] project.repoPath={declared_repo_path!r} 不存在或不是目录" ) binding_root = project_root or declared_root if binding_root is None or not binding_root.is_dir(): errors.append( "[tasks] 无法从 .pouch/ack 布局确定现有项目根目录；" "请传入 --project-root" ) return errors try: resolved_root = binding_root.resolve(strict=True) expected = (resolved_root / relative).resolve(strict=False) except (OSError, RuntimeError) as exc: errors.append(f"[tasks] project.knowledgeFile 无法安全解析: {exc}") return errors if not _is_within(expected, resolved_root): errors.append("[tasks] project.knowledgeFile 经 symlink 解析后逃逸项目根目录") return errors try: actual = knowledge_path.resolve(strict=True) except (OSError, RuntimeError) as exc: errors.append(f"[tasks] 当前知识文件无法安全解析: {exc}") return errors if actual != expected: errors.append( f"[tasks] project.knowledgeFile 指向 {expected}，" f"与当前知识文件 {actual} 不一致" ) return errors def _collect_task_attempts( tasks: list[Any], errors: list[str] ) -> tuple[dict[str, dict[str, Any]], dict[str, set[str]]]: tasks_by_id: dict[str, dict[str, Any]] = {} attempts_by_task: dict[str, set[str]] = {} all_attempts: set[str] = set() for index, task in enumerate(tasks): if not isinstance(task, dict) or not _nonempty(task.get("id")): continue task_id = task["id"] if task_id in tasks_by_id: continue tasks_by_id[task_id] = task attempts: set[str] = set() dispatch = task.get("dispatch") rounds = dispatch.get("rounds") if isinstance(dispatch, dict) else None if isinstance(rounds, list): round_numbers = [ round_item.get("round") if isinstance(round_item, dict) else None for round_item in rounds ] expected_rounds = list(range(1, len(rounds) + 1)) if round_numbers != expected_rounds: errors.append( f"[tasks] {task_id}.dispatch.rounds: round 必须按 " f"1..N 连续且不超过 {MAX_ROUNDS}" ) for round_index, round_item in enumerate(rounds): if not isinstance(round_item, dict): continue attempt_id = round_item.get("attemptId") if attempt_id is None: continue where = ( f"[tasks] {task_id}.dispatch.rounds[{round_index}].attemptId" ) round_number = round_item.get("round") if ( not isinstance(round_number, int) or isinstance(round_number, bool) or not 1 <= round_number <= MAX_ROUNDS ): errors.append( f"{where}: round 必须先是 1..{MAX_ROUNDS} 的整数" ) continue expected = f"{task_id}-A{round_number}" if attempt_id != expected: errors.append( f"{where}: 必须精确等于 {expected!r}，不能使用 dispatchId" ) continue if attempt_id in attempts or attempt_id in all_attempts: errors.append(f"{where}: attemptId 重复 {attempt_id!r}") continue attempts.add(attempt_id) all_attempts.add(attempt_id) attempts_by_task[task_id] = attempts return tasks_by_id, attempts_by_task def validate_task_references( knowledge: dict[str, Any], tasks_data: dict[str, Any], *, knowledge_path: Path | None = None, tasks_path: Path | None = None, project_root: Path | None = None, project_root_is_explicit: bool = False, ) -> list[str]: """校验 tasks.yaml 对固定知识版本的引用和 verified gate。""" errors = _validate_knowledge_file_binding( tasks_data, knowledge_path=knowledge_path, tasks_path=tasks_path, project_root=project_root, project_root_is_explicit=project_root_is_explicit, ) knowledge_project = knowledge.get("project") tasks_project = tasks_data.get("project") if isinstance(knowledge_project, dict) and isinstance(tasks_project, dict): knowledge_name = knowledge_project.get("name") tasks_name = tasks_project.get("name") if ( isinstance(knowledge_name, str) and isinstance(tasks_name, str) and knowledge_name != tasks_name ): errors.append( f"[tasks] project.name={tasks_name!r} 与知识库项目 " f"{knowledge_name!r} 不一致" ) entries = knowledge.get("entries") if not isinstance(entries, list): return errors by_ref = { ref: entry for entry in entries if isinstance(entry, dict) and (ref := stable_ref(entry)) is not None } tasks = tasks_data.get("tasks") if not isinstance(tasks, list): return ["[tasks] tasks 必须是列表"] tasks_by_id, attempts_by_task = _collect_task_attempts(tasks, errors) for index, entry in enumerate(entries): if not isinstance(entry, dict): continue provenance = entry.get("provenance") if not isinstance(provenance, dict): continue task_id = provenance.get("taskId") attempt_id = provenance.get("attemptId") where = f"[knowledge] entries[{index}].provenance" if not _nonempty(task_id) or task_id not in tasks_by_id: errors.append(f"{where}.taskId: 在 tasks.yaml 中找不到 {task_id!r}") continue if ( not _nonempty(attempt_id) or attempt_id not in attempts_by_task.get(task_id, set()) ): errors.append( f"{where}.attemptId: {attempt_id!r} 未命中任务 {task_id!r} " "dispatch.rounds[].attemptId" ) for index, task in enumerate(tasks): if not isinstance(task, dict): continue task_id = task.get("id") or index where = f"[tasks] {task_id}" requires_active = task.get("status") not in TERMINAL_TASK_STATUSES raw_refs = task.get("knowledgeRefs", []) if not isinstance(raw_refs, list): errors.append(f"{where}.knowledgeRefs: 必须是列表") continue selected_refs: set[str] = set() for ref in raw_refs: if not isinstance(ref, str) or not REF_RE.fullmatch(ref): errors.append(f"{where}.knowledgeRefs: {ref!r} 不是 K-...@revision") continue if ref in selected_refs: errors.append(f"{where}.knowledgeRefs: 重复引用 {ref}") continue selected_refs.add(ref) entry = by_ref.get(ref) if entry is None: errors.append(f"{where}.knowledgeRefs: 找不到精确版本 {ref}") elif requires_active and entry.get("status") != "active": errors.append( f"{where}.knowledgeRefs: {ref} 状态为 {entry.get('status')!r}，必须 active" ) applied = task.get("knowledgeApplied", []) if not isinstance(applied, list): errors.append(f"{where}.knowledgeApplied: 必须是对象列表") else: seen_applied: set[str] = set() for item in applied: if not isinstance(item, dict): errors.append(f"{where}.knowledgeApplied: 每项必须是对象") continue ref = item.get("ref") if not isinstance(ref, str) or not REF_RE.fullmatch(ref): errors.append(f"{where}.knowledgeApplied: ref 必须是 K-...@revision") continue if ref in seen_applied: errors.append(f"{where}.knowledgeApplied: 重复结果 {ref}") seen_applied.add(ref) if ref not in selected_refs: errors.append(f"{where}.knowledgeApplied: {ref} 未列入 knowledgeRefs") if ref not in by_ref: errors.append(f"{where}.knowledgeApplied: 找不到 {ref}") elif requires_active and by_ref[ref].get("status") != "active": errors.append(f"{where}.knowledgeApplied: {ref} 不是 active") if item.get("result") not in APPLIED_RESULTS: errors.append( f"{where}.knowledgeApplied[{ref}].result: 应为 {sorted(APPLIED_RESULTS)}" ) checks = task.get("knowledgeChecks", []) passed_checks: set[str] = set() if not isinstance(checks, list): errors.append(f"{where}.knowledgeChecks: 必须是对象列表") else: seen_checks: set[str] = set() for check in checks: if not isinstance(check, dict): errors.append(f"{where}.knowledgeChecks: 每项必须是对象") continue ref = check.get("ref") if not isinstance(ref, str) or not REF_RE.fullmatch(ref): errors.append(f"{where}.knowledgeChecks: ref 必须是 K-...@revision") continue if ref in seen_checks: errors.append(f"{where}.knowledgeChecks: 重复结果 {ref}") seen_checks.add(ref) if ref not in selected_refs: errors.append(f"{where}.knowledgeChecks: {ref} 未列入 knowledgeRefs") if ref not in by_ref: errors.append(f"{where}.knowledgeChecks: 找不到 {ref}") elif requires_active and by_ref[ref].get("status") != "active": errors.append(f"{where}.knowledgeChecks: {ref} 不是 active") result = check.get("result") if result not in CHECK_RESULTS: errors.append( f"{where}.knowledgeChecks[{ref}].result: 应为 {sorted(CHECK_RESULTS)}" ) elif result == "passed": passed_checks.add(ref) if task.get("status") == "verified": for ref in selected_refs: entry = by_ref.get(ref) if ( entry is not None and ref not in passed_checks ): errors.append( f"{where}: verified 任务的 knowledgeRef {ref} 缺少 passed knowledgeCheck" ) return errors def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="校验 ACK knowledge.yaml") parser.add_argument( "knowledge", nargs="?", default="knowledge.yaml", help="知识库路径" ) parser.add_argument("--schema", help="knowledge.schema.json 路径") parser.add_argument("--tasks", help="可选 tasks.yaml，用于跨文件引用校验") parser.add_argument( "--project-root", help="可选项目根目录；默认从 tasks.yaml 的 .pouch/ack 布局推断", ) args = parser.parse_args(argv) knowledge_path = Path(args.knowledge) if not knowledge_path.is_file(): sys.stderr.write(f"找不到知识库文件: {knowledge_path}\n") return 2 schema_path = ( Path(args.schema) if args.schema else Path(__file__).resolve().parent.parent / "templates" / "knowledge.schema.json" ) if args.schema and not schema_path.is_file(): sys.stderr.write(f"找不到 schema 文件: {schema_path}\n") return 2 tasks_path: Path | None = None tasks_data: dict[str, Any] | None = None if args.tasks: tasks_path = Path(args.tasks) if not tasks_path.is_file(): sys.stderr.write(f"找不到任务板文件: {tasks_path}\n") return 2 tasks_data = load_yaml(tasks_path, "任务板") project_root_is_explicit = bool(args.project_root) if project_root_is_explicit: project_root = Path(args.project_root).expanduser() if not project_root.is_dir(): sys.stderr.write(f"项目根目录不存在: {project_root}\n") return 2 project_root = project_root.resolve() else: project_root = ( _tasks_project_root(tasks_data, tasks_path) if tasks_data is not None else None ) if project_root is None: project_root = infer_project_root(knowledge_path) data = load_yaml(knowledge_path, "知识库") errors, mode = validate_all( data, schema_path, project_root=project_root ) registry = data.get("verificationRegistry") if ( project_root is None and isinstance(registry, dict) and registry ): errors.append( "verificationRegistry 非空但无法确定项目根目录；请传入 --project-root" ) if tasks_data is not None: errors.extend( validate_task_references( data, tasks_data, knowledge_path=knowledge_path, tasks_path=tasks_path, project_root=project_root, project_root_is_explicit=project_root_is_explicit, ) ) mode += " + tasks 引用" errors = list(dict.fromkeys(errors)) if errors: sys.stderr.write(f"知识库校验失败（{mode}），共 {len(errors)} 项：\n") for error in errors: sys.stderr.write(f" - {error}\n") return 1 sys.stdout.write(f"知识库校验通过（{mode}）：{knowledge_path}\n") return 0 if __name__ == "__main__": raise SystemExit(main())