dtoro
18cb79caf9
- seeds/ontology.yaml: 59 entity types (5 abstract, is-a hierarchy), 46 relationship types with cardinality, 6 lifecycles with terminal states and named precondition checks - seeds/inventory.yaml: 110 entities / 142 relationships translated from legacy inventory.yaml (fleet, services, ingress, storage, governance, archaeology); thin spots marked for backfill - seeds/policy.yaml: 4 risk classes, 27 approval rules (hierarchy-aware, per-entity overrides), autonomy kill-switch off (cold start) - api/openapi.yaml: full v1 REST contract (40 paths), RFC 9457 errors, cursor pagination, idempotency, ETag/If-Match, scopes; redocly-clean - docs/adr/0001-0010: initial architecture decision records - scripts/validate-seeds.py: Phase 0 gate — hierarchy, lifecycles, endpoints, cardinality, policy cross-refs (0 errors) - plan: layer CHECK gains 'meta' (root type), cardinality gains 'many-to-one' Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
285 lines
11 KiB
Python
285 lines
11 KiB
Python
#!/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 (<prefix>:<name>)
|
|
- 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 '<prefix>:<name>'")
|
|
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)
|