refactor(agent-collaboration-kit): split stable core from project overrides

- Reorganize flat files into core/ templates/ examples/ scripts/ with VERSION
- Deduplicate: single-source state machine, three-round policy, acceptance signals
- Abstract orchestration in closed-loop.md with manual mode; move Orca commands to orca-adapter.md
- Add tasks.schema.json + validate_tasks.py (jsonschema or builtin rules, no hard deps)
- Add filled examples, concurrency write rule, kitVersion tracking; unify to Chinese

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
2026-07-06 10:01:09 +08:00
parent 58de2650c2
commit d7a74578dc
18 changed files with 1177 additions and 806 deletions
+167
View File
@@ -0,0 +1,167 @@
#!/usr/bin/env python3
"""校验 tasks.yaml 是否符合 agent-collaboration-kit 任务板结构。
权威结构是同目录上层的 templates/tasks.schema.json(跨语言可用)。
本脚本是参考实现:
- 若安装了 jsonschema,则用 schema 做完整校验;
- 否则回退到内置的关键规则校验(必填字段、状态枚举、三轮上限、leftover 留档)。
YAML 解析优先用 pyyaml;未安装时给出提示而非崩溃。
用法:
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 sys
from pathlib import Path
STATUS_ENUM = {
"open",
"dispatched",
"fixed_by_dev",
"retesting",
"failed_retest",
"verified",
"blocked",
"leftover",
}
MAX_ROUNDS = 3
def load_yaml(path: Path) -> dict:
try:
import yaml # type: ignore
except ImportError:
sys.stderr.write(
"需要 PyYAML 才能解析 YAMLpip install pyyaml\n"
"(或把任务板导出为 JSON 后再校验)\n"
)
raise SystemExit(2)
try:
with path.open(encoding="utf-8") as fh:
data = yaml.safe_load(fh)
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_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] = []
if not isinstance(data.get("version"), int) or data.get("version", 0) < 1:
errors.append("version 必须是 >=1 的整数")
project = data.get("project")
if not isinstance(project, dict) or not project.get("name"):
errors.append("project.name 必填")
tasks = data.get("tasks")
if not isinstance(tasks, list):
errors.append("tasks 必须是列表")
return errors
seen_ids: 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 tid:
errors.append(f"{where}: id 必填")
else:
where = f"tasks[{i}] {tid}"
if tid in seen_ids:
errors.append(f"{where}: id 重复")
seen_ids.add(tid)
if not title:
errors.append(f"{where}: title 必填")
if status not in STATUS_ENUM:
errors.append(
f"{where}: status={status!r} 非法,应为 {sorted(STATUS_ENUM)}"
)
dispatch = task.get("dispatch") or {}
rounds = dispatch.get("rounds") or []
if isinstance(rounds, list):
if len(rounds) > MAX_ROUNDS:
errors.append(
f"{where}: 派发轮次 {len(rounds)} 超过上限 {MAX_ROUNDS}"
)
for r in rounds:
if isinstance(r, dict) and r.get("result") not in {"passed", "failed"}:
errors.append(f"{where}: round.result 必须是 passed/failed")
if status == "leftover":
resolution = task.get("resolution") or {}
if not resolution.get("leftoverReason"):
errors.append(f"{where}: leftover 必须填 resolution.leftoverReason")
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_yaml(tasks_path)
schema_path = Path(args.schema) if args.schema else (
Path(__file__).resolve().parent.parent / "templates" / "tasks.schema.json"
)
mode = "内置规则"
try:
import jsonschema # type: ignore # noqa: F401
if schema_path.is_file():
errors = validate_with_schema(data, schema_path)
mode = f"schema ({schema_path.name})"
else:
errors = validate_builtin(data)
mode = "内置规则(未找到 schema 文件)"
except ImportError:
errors = validate_builtin(data)
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())