Files
cmbuyer/scripts/validate_agent_context.py
T

227 lines
7.9 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
"""Validate the agent context manifest with the Python standard library."""
from __future__ import annotations
import argparse
import json
import re
import sys
from pathlib import Path, PurePosixPath
from typing import Any
EXPECTED_SCHEMA = "docs/agent-context.schema.json"
REQUIRED_TOP_LEVEL = {
"schema",
"schema_version",
"authority",
"bootstrap",
"routes",
"tasks",
"refresh",
"degraded_mode",
}
REQUIRED_BOOTSTRAP = {
"AGENTS.md",
"docs/00-ai-start-here.md",
"docs/05-coding-rules.md",
"docs/current-state.md",
}
TASK_PATH_KEYS = {"roadmap", "directory", "template"}
SENSITIVE_KEY = re.compile(r"(?:token|password|secret|credential)", re.IGNORECASE)
URI_SCHEME = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*:")
def load_json(path: Path, root: Path, errors: list[str]) -> Any:
try:
return json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError:
errors.append(f"文件不存在:{display_path(path, root)}")
except json.JSONDecodeError as exc:
errors.append(
f"JSON 语法错误:{display_path(path, root)}:{exc.lineno}:{exc.colno}"
)
return None
def display_path(path: Path, root: Path) -> str:
try:
return path.relative_to(root).as_posix()
except ValueError:
return path.as_posix()
def require_mapping(value: Any, name: str, errors: list[str]) -> dict[str, Any]:
if not isinstance(value, dict):
errors.append(f"{name} 必须是对象。")
return {}
return value
def require_string_list(value: Any, name: str, errors: list[str]) -> list[str]:
if not isinstance(value, list) or not value or not all(
isinstance(item, str) and item for item in value
):
errors.append(f"{name} 必须是非空字符串数组。")
return []
if len(value) != len(set(value)):
errors.append(f"{name} 不得包含重复路径。")
return value
def validate_repo_path(root: Path, value: str, name: str, errors: list[str]) -> None:
path = PurePosixPath(value)
if (
path.is_absolute()
or ".." in path.parts
or "\\" in value
or URI_SCHEME.match(value)
):
errors.append(f"{name} 必须是安全的仓库相对路径:{value}")
return
target = root.joinpath(*path.parts)
if not target.exists():
errors.append(f"{name} 引用路径不存在:{value}")
def find_sensitive_keys(value: Any, location: str, errors: list[str]) -> None:
if isinstance(value, dict):
for key, child in value.items():
child_location = f"{location}.{key}"
if SENSITIVE_KEY.search(key):
errors.append(f"清单不得保存敏感配置字段:{child_location}")
find_sensitive_keys(child, child_location, errors)
elif isinstance(value, list):
for index, child in enumerate(value):
find_sensitive_keys(child, f"{location}[{index}]", errors)
def validate_manifest(root: Path) -> list[str]:
root = root.resolve()
manifest_path = root / "docs" / "agent-context.json"
errors: list[str] = []
manifest = load_json(manifest_path, root, errors)
schema = load_json(root / EXPECTED_SCHEMA, root, errors)
if manifest is None or schema is None:
return errors
if not isinstance(schema, dict) or schema.get("type") != "object":
errors.append("agent-context.schema.json 不是有效的对象 Schema。")
root_object = require_mapping(manifest, "manifest", errors)
actual_keys = set(root_object)
missing = sorted(REQUIRED_TOP_LEVEL - actual_keys)
unexpected = sorted(actual_keys - REQUIRED_TOP_LEVEL)
if missing:
errors.append("缺少顶层字段:" + ", ".join(missing))
if unexpected:
errors.append("存在未知顶层字段:" + ", ".join(unexpected))
if root_object.get("schema") != EXPECTED_SCHEMA:
errors.append(f"schema 必须是 {EXPECTED_SCHEMA}。")
if root_object.get("schema_version") != 1:
errors.append("schema_version 必须为 1。")
authority = require_mapping(root_object.get("authority"), "authority", errors)
for key in ("bootstrap", "framework_templates", "project_facts", "coordination"):
if not isinstance(authority.get(key), str) or not authority[key]:
errors.append(f"authority.{key} 必须是非空字符串。")
bootstrap = require_mapping(root_object.get("bootstrap"), "bootstrap", errors)
always_read = require_string_list(
bootstrap.get("always_read"), "bootstrap.always_read", errors
)
missing_bootstrap = sorted(REQUIRED_BOOTSTRAP - set(always_read))
if missing_bootstrap:
errors.append("bootstrap.always_read 缺少:" + ", ".join(missing_bootstrap))
routes = require_mapping(root_object.get("routes"), "routes", errors)
if not routes:
errors.append("routes 至少需要一个任务类型。")
path_values: list[tuple[str, str]] = [(EXPECTED_SCHEMA, "schema")]
path_values.extend((path, "bootstrap.always_read") for path in always_read)
for route, value in routes.items():
paths = require_string_list(value, f"routes.{route}", errors)
path_values.extend((path, f"routes.{route}") for path in paths)
tasks = require_mapping(root_object.get("tasks"), "tasks", errors)
if set(tasks) != TASK_PATH_KEYS:
errors.append("tasks 必须且只能包含 roadmap、directory、template。")
for key in sorted(TASK_PATH_KEYS):
value = tasks.get(key)
if isinstance(value, str) and value:
path_values.append((value, f"tasks.{key}"))
else:
errors.append(f"tasks.{key} 必须是非空字符串。")
refresh = require_mapping(root_object.get("refresh"), "refresh", errors)
expected_refresh = {
"context_ref": "default_branch_head_sha",
"cache_key": "file_sha",
"unchanged_file": "reuse_within_current_session",
"changed_ref": "reread_manifest_and_routed_documents",
}
if refresh != expected_refresh:
errors.append("refresh 必须使用约定的提交 SHA 与文件 SHA 刷新策略。")
degraded = require_mapping(root_object.get("degraded_mode"), "degraded_mode", errors)
expected_degraded = {
"continue_claimed_task": True,
"claim_new_task": False,
"write_remote_state": False,
}
if degraded != expected_degraded:
errors.append("degraded_mode 必须禁止领取新任务和写入远端状态。")
for value, name in path_values:
validate_repo_path(root, value, name, errors)
find_sensitive_keys(root_object, "manifest", errors)
return errors
def manifest_summary(root: Path) -> tuple[int, int]:
manifest = json.loads(
(root / "docs" / "agent-context.json").read_text(encoding="utf-8")
)
paths = {manifest["schema"]}
paths.update(manifest["bootstrap"]["always_read"])
for values in manifest["routes"].values():
paths.update(values)
paths.update(manifest["tasks"].values())
return len(manifest["routes"]), len(paths)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="校验 Agent 上下文清单。")
parser.add_argument(
"--root",
type=Path,
default=Path(__file__).resolve().parents[1],
help="仓库根目录;默认取脚本上一级。",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
root = args.root.resolve()
if not root.is_dir():
print("ERROR: 仓库根目录不存在。", file=sys.stderr)
return 2
errors = validate_manifest(root)
if errors:
for error in errors:
print(f"ERROR: {error}", file=sys.stderr)
return 1
route_count, path_count = manifest_summary(root)
print(
"agent-context 校验通过:"
f"{route_count} 个任务路由,{path_count} 个有效仓库路径。"
)
return 0
if __name__ == "__main__":
raise SystemExit(main())