#!/usr/bin/env python3 """智能体/技能登记脚本——把 Git 侧的 agent/skill 元数据影子进库，供看板只读。 落库设计 §2.10 的 A 方案： - Git 侧（角色合同文档、.agent/agents/*.md、.agent/skills/*/SKILL.md）仍是配置权威； - 本脚本从中心角色合同读取责任、模型策略和工具政策，角色 frontmatter 只提供 name/description； upsert 进 example_agent_role / example_skill； - 看板只读登记表、不读 Git。幂等可重跑（upsert），配置变更后重跑即同步。 跑法（仓库根目录）： .venv/bin/python .agent/skills/access-database/scripts/sync_agent_registry.py --check .venv/bin/python .agent/skills/access-database/scripts/sync_agent_registry.py --check-db .venv/bin/python .agent/skills/access-database/scripts/sync_agent_registry.py """ import argparse import json import re from pathlib import Path from muse_db import connect from muse_role_contract import ROLE_CONTRACT_RELATIVE_PATH, load_role_contract_catalog ROOT = Path(__file__).resolve().parents[4] # .agent/skills/access-database/scripts → 仓库根 AGENTS_DIR = ROOT / ".agent" / "agents" SKILLS_DIR = ROOT / ".agent" / "skills" TABLE_RE = re.compile(r"\b(muse_[a-z_]+|example_[a-z_]+)\b") SKILL_NAME_RE = re.compile(r"^[a-z][a-z0-9]*(?:-[a-z0-9]+)+$") SKILL_FRONTMATTER_KEYS = frozenset({"name", "description", "disable-model-invocation"}) ROLE_FRONTMATTER_KEYS = frozenset({"name", "description", "skills", "tools"}) EXPECTED_ROLES = frozenset({"writer", "planner", "extractor", "detector", "judge"}) FRONTMATTER_KEY_RE = re.compile(r"^([A-Za-z0-9_-]+)\s*:") def parse_frontmatter(text: str) -> dict: """解析开头 --- 围栏内的 frontmatter 为 key->字符串（值可含冒号/逗号）。 只有顶格且键名合法的行才算键：block scalar（`description: |`）的续行一律缩进， 其正文里的冒号（如"要求严格 1:1 复刻"）不得被误读成字段名。 """ lines = text.splitlines() if not lines or lines[0].strip() != "---": return {} fm = {} for line in lines[1:]: if line.strip() == "---": break match = FRONTMATTER_KEY_RE.match(line) if match: _, _, value = line.partition(":") fm[match.group(1)] = value.strip() return fm def validate_role_catalog(agents_dir: Path = AGENTS_DIR) -> list[tuple[Path, dict]]: """校验五个角色文件、frontmatter 与宿主无关的身份字段。""" entries = [] seen = set() for md in sorted(agents_dir.glob("*.md")): fm = parse_frontmatter(md.read_text(encoding="utf-8")) unexpected = sorted(set(fm) - ROLE_FRONTMATTER_KEYS) if unexpected: raise ValueError(f"{md}: 角色 frontmatter 含未登记字段: {unexpected}") role = fm.get("name", "") if role != md.stem: raise ValueError(f"{md}: 文件名 {md.stem!r} 与 name {role!r} 不一致") if role in seen: raise ValueError(f"{md}: 角色 name 重复: {role}") if not fm.get("description"): raise ValueError(f"{md}: frontmatter 缺少 description") for key in ("skills", "tools"): if key not in fm or not fm[key].strip(): raise ValueError(f"{md}: frontmatter 缺少 {key}") seen.add(role) entries.append((md, fm)) if seen != EXPECTED_ROLES: raise ValueError( f"角色目录必须精确包含 {sorted(EXPECTED_ROLES)}，实际 {sorted(seen)}" ) if agents_dir.resolve() == AGENTS_DIR.resolve(): catalog = load_role_contract_catalog(ROOT) if set(catalog.roles) != EXPECTED_ROLES: raise ValueError("角色合同文档与角色目录不一致") return entries def validate_skill_catalog(skills_dir: Path = SKILLS_DIR) -> list[tuple[Path, dict]]: """校验 Skill 目录、frontmatter 与动作式名称，返回稳定排序的目录项。""" entries = [] seen = set() for sk in sorted(skills_dir.glob("*/SKILL.md")): fm = parse_frontmatter(sk.read_text(encoding="utf-8")) name = fm.get("name", "") directory = sk.parent.name unexpected = sorted(set(fm) - SKILL_FRONTMATTER_KEYS) if unexpected: raise ValueError(f"{sk}: frontmatter 含未登记字段: {unexpected}") if not name: raise ValueError(f"{sk}: frontmatter 缺少 name") if directory != name: raise ValueError(f"{sk}: 目录名 {directory!r} 与 name {name!r} 不一致") if not SKILL_NAME_RE.fullmatch(name): raise ValueError(f"{sk}: name 必须使用小写 动作-对象") if name in seen: raise ValueError(f"{sk}: Skill name 重复: {name}") if not fm.get("description"): raise ValueError(f"{sk}: frontmatter 缺少 description") invocation_flag = fm.get("disable-model-invocation") if invocation_flag is not None and invocation_flag not in {"true", "false"}: raise ValueError( f"{sk}: disable-model-invocation 必须是 true 或 false" ) seen.add(name) entries.append((sk, fm)) if not entries: raise ValueError(f"{skills_dir}: 未发现任何 SKILL.md") return entries def sync_roles(conn, catalog: list[tuple[Path, dict]] | None = None) -> int: entries = catalog or validate_role_catalog() contracts = load_role_contract_catalog(ROOT) names = [] for md, fm in entries: role = fm["name"] contract = contracts.for_role(role) names.append(role) display = contract.display_name responsibility = contract.contract_prompt conn.execute( """INSERT INTO example_agent_role (role, display_name, model, tools, responsibility, source_ref, synced_at, creator, updater) VALUES (%s,%s,NULL,NULL,%s,%s,CURRENT_TIMESTAMP,'sync_agent_registry','sync_agent_registry') ON CONFLICT (tenant_id, role) DO UPDATE SET display_name=EXCLUDED.display_name, model=NULL, tools=NULL, responsibility=EXCLUDED.responsibility, source_ref=EXCLUDED.source_ref, synced_at=CURRENT_TIMESTAMP, updater='sync_agent_registry', deleted=FALSE""", (role, display, responsibility, ROLE_CONTRACT_RELATIVE_PATH.as_posix())) conn.execute( """UPDATE example_agent_role SET deleted=TRUE, synced_at=CURRENT_TIMESTAMP, updater='sync_agent_registry' WHERE tenant_id=0 AND deleted=FALSE AND NOT (role=ANY(%s))""", (names,), ) return len(entries) def sync_skills(conn, catalog: list[tuple[Path, dict]] | None = None) -> int: entries = catalog or validate_skill_catalog() names = [] for sk, fm in entries: text = sk.read_text(encoding="utf-8") name = fm["name"] names.append(name) tables = sorted(set(TABLE_RE.findall(text))) # 尽力抽取涉及的表名（不区分读写，待人工核） model_used = None if "MiniMax" in text: model_used = "MiniMax-M3" elif re.search(r"\bopus\b", text): model_used = "opus" conn.execute( """INSERT INTO example_skill (skill_name, purpose, reads, model_used, source_ref, synced_at, creator, updater) VALUES (%s,%s,%s::jsonb,%s,%s,CURRENT_TIMESTAMP,'sync_agent_registry','sync_agent_registry') ON CONFLICT (tenant_id, skill_name) DO UPDATE SET purpose=EXCLUDED.purpose, reads=EXCLUDED.reads, model_used=EXCLUDED.model_used, source_ref=EXCLUDED.source_ref, synced_at=CURRENT_TIMESTAMP, updater='sync_agent_registry', deleted=FALSE""", (name, fm.get("description"), json.dumps(tables, ensure_ascii=False) if tables else None, model_used, str(sk.relative_to(ROOT)))) conn.execute( """UPDATE example_skill SET deleted=TRUE, synced_at=CURRENT_TIMESTAMP, updater='sync_agent_registry' WHERE tenant_id=0 AND deleted=FALSE AND NOT (skill_name=ANY(%s))""", (names,), ) return len(entries) def check_database(conn, roles, skills) -> None: """把数据库当前活跃影子与 Git 目录逐项对账。""" contracts = load_role_contract_catalog(ROOT) expected_roles = { fm["name"]: { "source_ref": ROLE_CONTRACT_RELATIVE_PATH.as_posix(), "responsibility": contracts.for_role(fm["name"]).contract_prompt, "model": None, } for _path, fm in roles } actual_roles = { row[0]: {"source_ref": row[1], "responsibility": row[2], "model": row[3]} for row in conn.execute( "SELECT role, source_ref, responsibility, model FROM example_agent_role " "WHERE tenant_id=0 AND deleted=FALSE" ).fetchall() } expected_skills = { fm["name"]: { "source_ref": str(path.relative_to(ROOT)), "purpose": fm["description"], } for path, fm in skills } actual_skills = { row[0]: {"source_ref": row[1], "purpose": row[2]} for row in conn.execute( "SELECT skill_name, source_ref, purpose FROM example_skill " "WHERE tenant_id=0 AND deleted=FALSE" ).fetchall() } if actual_roles != expected_roles: raise ValueError("example_agent_role 与 Git 角色目录不一致，请执行同步") if actual_skills != expected_skills: raise ValueError("example_skill 与 Git Skill 目录不一致，请执行同步") def main(): parser = argparse.ArgumentParser(description="校验或同步 Agent/Skill 登记") mode = parser.add_mutually_exclusive_group() mode.add_argument("--check", action="store_true", help="只校验 Git 侧角色与 Skill 目录") mode.add_argument("--check-db", action="store_true", help="只读对账 Git 与数据库活跃影子") args = parser.parse_args() roles = validate_role_catalog() skills = validate_skill_catalog() if args.check: print(f"目录校验通过：{len(roles)} 个角色、{len(skills)} 个 Skill") return if args.check_db: with connect(readonly=True) as conn: check_database(conn, roles, skills) print(f"数据库影子一致：{len(roles)} 个角色、{len(skills)} 个 Skill") return with connect() as conn: nr = sync_roles(conn, roles) ns = sync_skills(conn, skills) conn.commit() print(f"已登记 {nr} 个角色、{ns} 个技能 → example_agent_role / example_skill") if __name__ == "__main__": main()