oikos phase 0: ontology + inventory + policy seeds, OpenAPI contract, ADRs
- 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>
This commit is contained in:
284
scripts/validate-seeds.py
Normal file
284
scripts/validate-seeds.py
Normal file
@@ -0,0 +1,284 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user