#!/usr/bin/env python3 """Validate seeds/*.yaml against the Oikos meta-schema (Phase 0 gate). Checks the same invariants the Go ingest (internal/ontology) will enforce: ontology.yaml - every entity type's parent exists; hierarchy is acyclic - layer/cardinality values are legal - lifecycle references exist; default/terminal/transition states are declared - relationship endpoint types exist (may be abstract) - inverse names don't collide with forward names inventory.yaml - slugs unique and well-formed (:) - entity types exist and are NOT abstract - states are legal for the type's lifecycle (walking up the hierarchy for the lifecycle definition is not needed — lifecycle binds per type) - relationship endpoints exist; edge type exists; endpoint entity types are the declared source/target types or descendants (hierarchy walk) - cardinality: one-to-one / one-to-many / many-to-one uniqueness holds within the seed - attributes validate against attribute_schema (if jsonschema installed) policy.yaml - approval_rules reference existing risk classes / entity types / entities - autonomy values sane Exit 0 = clean (warnings allowed), 1 = errors. """ import sys from pathlib import Path try: import yaml except ImportError: print("ERROR: PyYAML required (pip install pyyaml)") sys.exit(1) try: import jsonschema HAVE_JSONSCHEMA = True except ImportError: HAVE_JSONSCHEMA = False SEEDS = Path(__file__).resolve().parent.parent / "seeds" LAYERS = {"meta", "infrastructure", "governance", "cognition"} CARDINALITIES = {"one-to-one", "one-to-many", "many-to-one", "many-to-many"} errors: list[str] = [] warnings: list[str] = [] def err(msg: str) -> None: errors.append(msg) def warn(msg: str) -> None: warnings.append(msg) def load(name: str) -> dict: path = SEEDS / name if not path.exists(): err(f"{name}: file missing") return {} with open(path) as f: try: return yaml.safe_load(f) or {} except yaml.YAMLError as e: err(f"{name}: YAML parse error: {e}") return {} # ─── ontology.yaml ──────────────────────────────────────────────────── onto = load("ontology.yaml") etypes: dict = onto.get("entity_types", {}) rtypes: dict = onto.get("relationship_types", {}) lifecycles: dict = onto.get("lifecycles", {}) for name, lc in lifecycles.items(): states = lc.get("states", []) if not states: err(f"lifecycle {name}: no states") continue if lc.get("default_state") not in states: err(f"lifecycle {name}: default_state {lc.get('default_state')!r} not in states") for t in lc.get("terminal_states", []): if t not in states: err(f"lifecycle {name}: terminal state {t!r} not in states") for frm, tos in (lc.get("transitions") or {}).items(): if frm not in states: err(f"lifecycle {name}: transition source {frm!r} not in states") for to, spec in (tos or {}).items(): if to not in states: err(f"lifecycle {name}: transition target {to!r} not in states") if spec is not None and not isinstance(spec.get("requires", []), list): err(f"lifecycle {name}: {frm}->{to} requires must be a list") for name, et in etypes.items(): parent = et.get("parent") if parent is not None and parent not in etypes: err(f"entity type {name}: parent {parent!r} not defined") layer = et.get("layer") if layer not in LAYERS: err(f"entity type {name}: layer {layer!r} invalid (want {sorted(LAYERS)})") lc = et.get("lifecycle") if lc is not None and lc not in lifecycles: err(f"entity type {name}: lifecycle {lc!r} not defined") schema = et.get("attributes") if schema is not None and HAVE_JSONSCHEMA: try: jsonschema.Draft202012Validator.check_schema(schema) except jsonschema.SchemaError as e: err(f"entity type {name}: attribute_schema is not valid JSON Schema: {e.message}") # hierarchy acyclicity + ancestor helper def ancestors(t: str) -> list[str]: chain, seen = [], set() cur = t while cur is not None: if cur in seen: err(f"entity type hierarchy cycle at {cur!r}") break seen.add(cur) chain.append(cur) cur = etypes.get(cur, {}).get("parent") return chain for name in etypes: ancestors(name) def is_a(t: str, target: str) -> bool: return target in ancestors(t) fwd_names = set(rtypes) for name, rt in rtypes.items(): for endpoint in ("source", "target"): v = rt.get(endpoint) if v not in etypes: err(f"relationship type {name}: {endpoint} {v!r} not a defined entity type") if rt.get("cardinality") not in CARDINALITIES: err(f"relationship type {name}: cardinality {rt.get('cardinality')!r} invalid") inv = rt.get("inverse") if inv and inv in fwd_names: err(f"relationship type {name}: inverse {inv!r} collides with a forward name") # ─── inventory.yaml ─────────────────────────────────────────────────── inv = load("inventory.yaml") entities: list = inv.get("entities", []) relationships: list = inv.get("relationships", []) slugs: dict = {} for e in entities: slug = e.get("slug") if not slug or ":" not in slug: err(f"entity {e}: slug missing or not ':'") continue if slug in slugs: err(f"duplicate slug {slug!r}") slugs[slug] = e t = e.get("type") if t not in etypes: err(f"{slug}: type {t!r} not in ontology") continue if etypes[t].get("abstract"): err(f"{slug}: type {t!r} is abstract — cannot be instantiated") if not e.get("name"): err(f"{slug}: name missing") state = e.get("state") lc_name = etypes[t].get("lifecycle") if state is not None: if lc_name is None: err(f"{slug}: has state {state!r} but type {t!r} has no lifecycle") elif state not in lifecycles.get(lc_name, {}).get("states", []): err(f"{slug}: state {state!r} not in lifecycle {lc_name!r}") schema = etypes[t].get("attributes") if schema and HAVE_JSONSCHEMA and e.get("attributes"): v = jsonschema.Draft202012Validator(schema) for ve in v.iter_errors(e["attributes"]): warn(f"{slug}: attributes: {ve.message}") # uniqueness of (type, name) seen_tn = set() for e in entities: tn = (e.get("type"), e.get("name")) if tn in seen_tn: err(f"duplicate (type, name): {tn}") seen_tn.add(tn) edge_keys = set() by_card_src: dict = {} by_card_tgt: dict = {} for r in relationships: src, tgt, rt_name = r.get("source"), r.get("target"), r.get("type") ctx = f"edge {src} -{rt_name}-> {tgt}" if rt_name not in rtypes: err(f"{ctx}: relationship type not in ontology") continue ok = True for label, slug in (("source", src), ("target", tgt)): if slug not in slugs: err(f"{ctx}: {label} entity {slug!r} not in inventory") ok = False if not ok: continue key = (src, tgt, rt_name) if key in edge_keys: err(f"{ctx}: duplicate edge") edge_keys.add(key) rt = rtypes[rt_name] for label, slug, want in (("source", src, rt["source"]), ("target", tgt, rt["target"])): actual = slugs[slug]["type"] if not is_a(actual, want): err(f"{ctx}: {label} type {actual!r} is not a {want!r} (or descendant)") # cardinality bookkeeping (source→target multiplicity) card = rt.get("cardinality") if card in ("one-to-one", "many-to-one"): # each source has at most one outgoing edge of this type k = (rt_name, src) if k in by_card_src: err(f"{ctx}: cardinality {card} — source {src!r} already has a {rt_name!r} edge") by_card_src[k] = tgt if card in ("one-to-one", "one-to-many"): # each target has at most one incoming edge of this type k = (rt_name, tgt) if k in by_card_tgt: err(f"{ctx}: cardinality {card} — target {tgt!r} already has a {rt_name!r} edge") by_card_tgt[k] = src # ─── policy.yaml ────────────────────────────────────────────────────── pol = load("policy.yaml") rclasses: dict = pol.get("risk_classes", {}) for name, rc in rclasses.items(): if rc.get("approval_required") not in ("none", "operator", "operator_confirmed"): err(f"risk class {name}: approval_required invalid") rules = pol.get("approval_rules", []) seen_rules = set() for rule in rules: ctx = f"rule ({rule.get('entity_type')}, {rule.get('action')}, {rule.get('scope_entity')})" if rule.get("risk_class") not in rclasses: err(f"{ctx}: risk_class {rule.get('risk_class')!r} not defined") if rule.get("autonomy_level") not in ("auto", "escalate", "never"): err(f"{ctx}: autonomy_level invalid") et = rule.get("entity_type") if et is not None and et not in etypes: err(f"{ctx}: entity_type {et!r} not in ontology") scope = rule.get("scope_entity") if scope is not None and scope not in slugs: err(f"{ctx}: scope_entity {scope!r} not in inventory") key = (et, rule.get("action"), scope) if key in seen_rules: err(f"{ctx}: duplicate rule") seen_rules.add(key) # rule sanity: autonomy 'auto' requires the class to allow autonomy rc = rclasses.get(rule.get("risk_class"), {}) if rule.get("autonomy_level") == "auto" and not rc.get("autonomy_allowed"): err(f"{ctx}: autonomy_level=auto but risk class forbids autonomy") auto = pol.get("autonomy_settings", {}) ga = auto.get("global.auto_act") if ga not in ("off", "reversible_low"): err(f"autonomy_settings: global.auto_act {ga!r} invalid (off | reversible_low)") for k in auto: if k.startswith("never_auto_act."): slug = k.split(".", 1)[1] if slug not in slugs: warn(f"autonomy_settings: {k} references unknown entity {slug!r}") # ─── report ─────────────────────────────────────────────────────────── print(f"entity types: {len(etypes)} ({sum(1 for e in etypes.values() if e.get('abstract'))} abstract)") print(f"relationship types: {len(rtypes)}") print(f"lifecycles: {len(lifecycles)}") print(f"entities: {len(entities)}") print(f"relationships: {len(relationships)}") print(f"risk classes: {len(rclasses)}, rules: {len(rules)}") if not HAVE_JSONSCHEMA: warn("jsonschema not installed — attribute schema validation skipped") for w in warnings: print(f"WARN {w}") for e in errors: print(f"ERROR {e}") print(f"\n{'FAIL' if errors else 'OK'} — {len(errors)} error(s), {len(warnings)} warning(s)") sys.exit(1 if errors else 0)