File size: 5,521 Bytes
f109bf8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 | #!/usr/bin/env python3
"""Harness: registry/seed_agents.yaml → router/registry_facts.pl (stdlib + PyYAML).
Not product code. Regenerates Prolog agent facts so symbolic_rules.pl stays
aligned with the seed registry. Prefer: scripts/check_prolog_registry_sync.sh
"""
from __future__ import annotations
import sys
from pathlib import Path
try:
import yaml
except ImportError:
yaml = None
ROOT = Path(__file__).resolve().parents[1]
SEED = ROOT / "registry" / "seed_agents.yaml"
OUT = ROOT / "router" / "registry_facts.pl"
def to_atom(agent_id: str) -> str:
"""Registry PascalCase id → Prolog atom (downcase first letter)."""
if not agent_id:
raise ValueError("empty agent_id")
return agent_id[0].lower() + agent_id[1:]
def parse_seed_stdlib(text: str) -> list[dict]:
"""Minimal YAML subset parser for seed_agents.yaml (no PyYAML required)."""
agents: list[dict] = []
cur: dict | None = None
list_key: str | None = None
def flush():
nonlocal cur
if cur is not None:
agents.append(cur)
cur = None
for raw in text.splitlines():
if not raw.strip() or raw.lstrip().startswith("#"):
continue
if raw.startswith("- agent_id:"):
flush()
cur = {
"agent_id": raw.split(":", 1)[1].strip(),
"capabilities": [],
"permissions": [],
"dependencies": [],
"control_plane": False,
}
list_key = None
continue
if cur is None:
continue
indent = len(raw) - len(raw.lstrip(" "))
s = raw.strip()
if indent == 2 and ":" in s and not s.startswith("-"):
key, _, val = s.partition(":")
key = key.strip()
val = val.strip()
list_key = None
if key in ("capabilities", "permissions", "dependencies", "languages",
"input_types", "output_types", "required_evidence"):
if val.startswith("[") and val.endswith("]"):
inner = val[1:-1].strip()
cur[key] = [x.strip() for x in inner.split(",") if x.strip()] if inner else []
elif val == "":
list_key = key
cur[key] = []
else:
cur[key] = [val]
elif key == "control_plane":
cur[key] = val.lower() in ("true", "yes", "1")
elif key == "risk_class":
cur[key] = val
elif key == "domain":
cur[key] = val
else:
cur[key] = val
elif indent >= 4 and s.startswith("- ") and list_key:
cur.setdefault(list_key, []).append(s[2:].strip())
flush()
return agents
def load_agents() -> list[dict]:
text = SEED.read_text(encoding="utf-8")
if yaml is not None:
data = yaml.safe_load(text)
if not isinstance(data, list):
raise SystemExit("seed_agents.yaml: expected list")
return data
return parse_seed_stdlib(text)
def emit(agents: list[dict]) -> str:
lines: list[str] = [
"% SPDX-License-Identifier: AGPL-3.0-only",
"% Copyright (C) 2026 David contributors",
"% GENERATED by scripts/gen_registry_facts.py from registry/seed_agents.yaml",
"% Do not edit by hand — regenerate via scripts/check_prolog_registry_sync.sh",
"%",
"% Agent atom map: registry agent_id → Prolog atom = downcase first letter",
"% LegacyCobol → legacyCobol, Risk → risk, Quant → quant",
"% Predicates: agent_id/2, agent/1, control_plane/1, capability/2,",
"% permission/2, risk_class/2, depends/2",
"",
]
# agent_id/2 map first
for a in agents:
rid = a["agent_id"]
atom = to_atom(rid)
lines.append(f"agent_id('{rid}', {atom}).")
lines.append("")
for a in agents:
atom = to_atom(a["agent_id"])
lines.append(f"agent({atom}).")
lines.append("")
for a in agents:
if a.get("control_plane"):
lines.append(f"control_plane({to_atom(a['agent_id'])}).")
lines.append("")
for a in agents:
atom = to_atom(a["agent_id"])
for cap in a.get("capabilities") or []:
lines.append(f"capability({atom}, {cap}).")
lines.append("")
for a in agents:
atom = to_atom(a["agent_id"])
for perm in a.get("permissions") or []:
lines.append(f"permission({atom}, {perm}).")
lines.append("")
for a in agents:
atom = to_atom(a["agent_id"])
rc = a.get("risk_class")
if not rc:
raise SystemExit(f"missing risk_class for {a['agent_id']}")
lines.append(f"risk_class({atom}, {rc}).")
lines.append("")
for a in agents:
atom = to_atom(a["agent_id"])
for dep in a.get("dependencies") or []:
lines.append(f"depends({atom}, {to_atom(dep)}).")
lines.append("")
return "\n".join(lines)
def main() -> int:
out = Path(sys.argv[1]) if len(sys.argv) > 1 else OUT
agents = load_agents()
if not agents:
print("no agents loaded", file=sys.stderr)
return 1
text = emit(agents)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(text, encoding="utf-8")
print(f"wrote {out} ({len(agents)} agents)")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|