facetted-interfaces/history/FIAM_v0.2_Draft6/tools/validate_fiam.py

#!/usr/bin/env python3
from __future__ import annotations
import argparse, glob, json, re
from pathlib import Path
import yaml
from jsonschema import Draft202012Validator
ROOT=Path(__file__).resolve().parents[1]
CORE=ROOT/'schemas/FIAM-Core-v0.2.schema.json'
CANON=ROOT/'canon/FIAM-Canon-v0.2.yaml'
SECTIONS=['participants','nodes','ports','contracts','endpoints','bindings','channels','information_resources','information_shapes','interactions','interaction_flows','artifacts','zones','security','assertions']
KINDS={'participants':'participant','nodes':'system-node','ports':'port','contracts':'interface-contract','endpoints':'endpoint','bindings':'binding','channels':'channel','information_resources':'information-resource','information_shapes':'information-shape','interactions':'interaction','interaction_flows':'interaction-flow','artifacts':'artifact','zones':'zone','security':'security-profile','assertions':'assertion'}
REF_FIELDS={'owner','contract','provider_port','channel','operation','binding','interaction','from','via','to','applies_to','describes','implements','subject','object_ref'}
REF_LIST_FIELDS={'consumer_ports','operations','endpoints','participants','supersedes','conflicts_with'}
IDENTITY_SAME_KIND={'fiam.relation.same-entity-as','fiam.relation.alias-of','fiam.relation.version-of','fiam.relation.observationally-equivalent-to'}
SHAPE_PART={'fiam.relation.accepts-shape','fiam.relation.produces-shape','fiam.relation.errors-with-shape'}

def collect(doc):
    objs={}; objmap={}; assertions=set()
    for section in SECTIONS:
        for o in doc.get(section,[]) or []:
            if isinstance(o,dict) and 'id' in o:
                kind=o.get('kind') or KINDS.get(section); objs[o['id']]=kind; objmap[o['id']]=o
                if section=='assertions': assertions.add(o['id'])
            if section=='contracts' and isinstance(o,dict):
                for op in o.get('operations',[]) or []:
                    if isinstance(op,dict) and 'id' in op: objs[op['id']]='operation'; objmap[op['id']]=op
    return objs,objmap,assertions

def iter_term_refs(x,path=()):
    if isinstance(x,dict):
        for k,v in x.items(): yield from iter_term_refs(v,path+(k,))
    elif isinstance(x,list):
        for i,v in enumerate(x): yield from iter_term_refs(v,path+(i,))
    elif isinstance(x,str) and x.startswith('fiam.'):
        yield path,x

def path_values(doc,path):
    vals=[doc]
    for raw in path.split('.'):
        arr=raw.endswith('[]'); key=raw[:-2] if arr else raw
        nxt=[]
        for v in vals:
            if not isinstance(v,dict) or key not in v: continue
            x=v[key]
            if arr:
                if isinstance(x,list): nxt.extend(x)
                else: nxt.append(x)
            else: nxt.append(x)
        vals=nxt
    return vals

def entity_refs(doc,objs,assertions):
    err=[]
    def walk(o,path=()):
        if isinstance(o,dict):
            for k,v in o.items():
                p=path+(k,)
                if k in REF_FIELDS:
                    vals=v if isinstance(v,list) else [v]
                    for r in vals:
                        if isinstance(r,str) and r not in objs: err.append(f"{'.'.join(map(str,p))}: unresolved entity ref {r}")
                elif k in REF_LIST_FIELDS and isinstance(v,list):
                    target=assertions if k in {'supersedes','conflicts_with'} else objs
                    for r in v:
                        if isinstance(r,str) and r not in target: err.append(f"{'.'.join(map(str,p))}: unresolved ref {r}")
                elif k=='participations' and isinstance(v,list):
                    for j,part in enumerate(v):
                        c=part.get('contract') if isinstance(part,dict) else None
                        if c and c not in objs: err.append(f"{'.'.join(map(str,p+(j,'contract')))}: unresolved contract {c}")
                elif k=='participants' and isinstance(v,dict):
                    for rk,rv in v.items():
                        vals=rv if isinstance(rv,list) else [rv]
                        for r in vals:
                            if isinstance(r,str) and r not in objs: err.append(f"{'.'.join(map(str,p+(rk,)))}: unresolved participant {r}")
                walk(v,p)
        elif isinstance(o,list):
            for i,v in enumerate(o): walk(v,path+(i,))
    walk(doc); return err

def participation_checks(doc):
    errors=[]; ports={p['id']:p for p in doc.get('ports',[]) or [] if 'id' in p}
    def has(pid,c,mode):
        parts=ports.get(pid,{}).get('participations')
        if not parts: return None
        return any(x.get('contract')==c and x.get('mode')==mode for x in parts)
    for b in doc.get('bindings',[]) or []:
        c=b.get('contract'); pp=b.get('provider_port')
        if c and pp and has(pp,c,'offers') is False: errors.append(f"binding {b['id']}: provider {pp} does not offer {c}")
        for cp in b.get('consumer_ports',[]) or []:
            if c and has(cp,c,'requires') is False: errors.append(f"binding {b['id']}: consumer {cp} does not require {c}")
    return errors

def value_type_ok(value, typ):
    if typ=='boolean': return isinstance(value,bool)
    if typ=='string': return isinstance(value,str)
    if typ=='number': return isinstance(value,(int,float)) and not isinstance(value,bool)
    if typ=='integer': return isinstance(value,int) and not isinstance(value,bool)
    return True

def subset(expected,actual):
    if isinstance(expected,dict): return isinstance(actual,dict) and all(k in actual and subset(v,actual[k]) for k,v in expected.items())
    if isinstance(expected,list): return isinstance(actual,list) and all(any(subset(e,a) for a in actual) for e in expected)
    return expected==actual

def execute_check(check,doc,objs,objmap,terms,deprecated):
    k=check['kind']
    if k=='all-assertions-have-evidence':
        missing=[a.get('id') for a in doc.get('assertions',[]) or [] if not a.get('evidence')]
        return None if not missing else f"assertions without evidence: {missing}"
    if k=='no-deprecated-terms':
        used=[t for _,t in iter_term_refs(doc) if t in deprecated]
        return None if not used else f"deprecated terms used: {sorted(set(used))}"
    if k=='no-legacy-representations':
        refs=[]
        for c in doc.get('contracts',[]) or []:
            if 'representations' in c: refs.append(c.get('id'))
            for op in c.get('operations',[]) or []:
                if 'representations' in op: refs.append(op.get('id'))
        return None if not refs else f"legacy representations fields on {refs}"
    if k=='entity-exists':
        r=check.get('ref'); ek=check.get('entity_kind')
        if r not in objs: return f"entity {r} does not exist"
        if ek and objs[r]!=ek: return f"entity {r} is {objs[r]}, expected {ek}"
        return None
    if k in {'relation-exists','relation-absent'}:
        matches=[]
        for a in doc.get('assertions',[]) or []:
            if a.get('assertion_kind')=='relation' and a.get('polarity')=='asserted' and a.get('subject')==check.get('subject') and a.get('predicate')==check.get('predicate') and a.get('object_ref')==check.get('object_ref'):
                if check.get('facets') is None or subset(check['facets'],a.get('facets',{})): matches.append(a)
        exists=bool(matches)
        if k=='relation-exists' and not exists: return f"expected relation not found: {check.get('subject')} {check.get('predicate')} {check.get('object_ref')} facets={check.get('facets')}"
        if k=='relation-absent' and exists: return f"forbidden relation exists: {check.get('subject')} {check.get('predicate')} {check.get('object_ref')}"
        return None
    if k in {'field-equals','field-absent','field-contains'}:
        r=check.get('ref'); field=check.get('field'); obj=objmap.get(r)
        if obj is None: return f"entity {r} not found"
        if k=='field-absent': return None if field not in obj else f"field {field} unexpectedly present on {r}"
        if field not in obj: return f"field {field} missing on {r}"
        if k=='field-equals': return None if obj[field]==check.get('value') else f"{r}.{field}={obj[field]!r}, expected {check.get('value')!r}"
        v=obj[field]; wanted=check.get('value')
        ok=(wanted in v) if isinstance(v,list) else (isinstance(v,dict) and wanted in v) or v==wanted
        return None if ok else f"{r}.{field} does not contain {wanted!r}"
    return f"unknown semantic check kind {k}"



def evidence_has_mode(evidence, mode):
    records=evidence if isinstance(evidence,list) else [evidence]
    return any(isinstance(r,dict) and r.get('mode')==mode for r in records)

def asserted_relations(doc, predicate=None):
    out=[]
    for a in doc.get('assertions',[]) or []:
        if a.get('assertion_kind')!='relation' or a.get('polarity')!='asserted': continue
        if predicate and a.get('predicate')!=predicate: continue
        out.append(a)
    return out

def equivalence_map(doc, predicate='fiam.relation.same-entity-as'):
    parent={}
    def find(x):
        parent.setdefault(x,x)
        if parent[x]!=x: parent[x]=find(parent[x])
        return parent[x]
    def union(a,b):
        ra,rb=find(a),find(b)
        if ra!=rb: parent[rb]=ra
    for a in asserted_relations(doc,predicate):
        if a.get('subject') and a.get('object_ref'): union(a['subject'],a['object_ref'])
    return find

def relation_reaches(doc, child, parent, predicates):
    if child==parent: return True
    same=equivalence_map(doc)
    if same(child)==same(parent): return True
    graph={}
    for a in asserted_relations(doc):
        if a.get('predicate') in predicates:
            graph.setdefault(a.get('subject'),set()).add(a.get('object_ref'))
    seen=set(); stack=[child]
    while stack:
        x=stack.pop()
        if x in seen: continue
        seen.add(x)
        for y in graph.get(x,set()):
            if y==parent or same(y)==same(parent): return True
            stack.append(y)
    return False

def shape_coherence_checks(doc, objs, objmap, canon):
    err=[]; pol=canon.get('semantic_policies',{}).get('resource_shape',{})
    collection_kind=pol.get('collection_shape_kind','fiam.information-shape.collection')
    collection_resources=set(pol.get('collection_resource_kinds',[]))
    collection_scopes={'fiam.resource-scope.concrete-collection','fiam.resource-scope.logical-selection'}
    for a in asserted_relations(doc):
        pred=a.get('predicate'); subj=a.get('subject'); obj=a.get('object_ref')
        if pred=='fiam.relation.conforms-to-shape' and objs.get(subj)=='information-resource' and objs.get(obj)=='information-shape':
            rk=objmap.get(subj,{}).get('resource_kind')
            if rk in collection_resources and objmap.get(obj,{}).get('shape_kind')!=collection_kind:
                err.append(f"assertion {a.get('id')}: collection-like resource {subj} ({rk}) conforms-to non-collection shape {obj} ({objmap.get(obj,{}).get('shape_kind')})")
        if pred=='fiam.relation.element-shape':
            sk=objs.get(subj)
            if sk=='information-shape' and objmap.get(subj,{}).get('shape_kind')!=collection_kind:
                err.append(f"assertion {a.get('id')}: element-shape subject {subj} must have shape_kind {collection_kind}")
            elif sk=='information-resource':
                so=objmap.get(subj,{})
                if so.get('resource_kind') not in collection_resources and so.get('resource_scope') not in collection_scopes:
                    err.append(f"assertion {a.get('id')}: element-shape resource subject {subj} is not collection-like")
    return err

def shape_refinement_checks(doc, objs, objmap, canon, contract_ops):
    err=[]; policy=canon.get('semantic_policies',{}).get('shape_participation',{})
    refine=set(policy.get('refinement_relations',['fiam.relation.projection-of','fiam.relation.version-of']))
    # Shape participations indexed by subject/predicate.
    rels=[a for a in asserted_relations(doc) if a.get('predicate') in SHAPE_PART]
    by_subject_pred={}
    for a in rels: by_subject_pred.setdefault((a.get('subject'),a.get('predicate')),[]).append(a)
    bindings={b.get('id'):b for b in doc.get('bindings',[]) or []}
    interactions={i.get('id'):i for i in doc.get('interactions',[]) or []}
    # Binding -> Operation default.
    for a in rels:
        if objs.get(a.get('subject'))!='binding': continue
        b=bindings.get(a['subject'],{}); ops=b.get('operations') or contract_ops.get(b.get('contract'),[]) or []
        opref=(a.get('facets') or {}).get('operation_ref')
        if not opref and len(ops)==1: opref=ops[0]
        if not opref: continue
        parents=[x.get('object_ref') for x in by_subject_pred.get((opref,a['predicate']),[])]
        if parents and not any(relation_reaches(doc,a.get('object_ref'),p,refine) for p in parents):
            err.append(f"assertion {a.get('id')}: binding shape {a.get('object_ref')} does not refine Operation {opref} {a['predicate']} shape(s) {parents}")
    # Interaction -> most specific Binding default, otherwise Operation default.
    for a in rels:
        iid=a.get('subject')
        if objs.get(iid)!='interaction': continue
        inter=interactions.get(iid,{}); opref=inter.get('operation'); bid=inter.get('binding')
        parents=[]
        if bid:
            for br in by_subject_pred.get((bid,a['predicate']),[]):
                bref=(br.get('facets') or {}).get('operation_ref')
                if not bref or bref==opref: parents.append(br.get('object_ref'))
        if not parents and opref: parents=[x.get('object_ref') for x in by_subject_pred.get((opref,a['predicate']),[])]
        if parents and not any(relation_reaches(doc,a.get('object_ref'),p,refine) for p in parents):
            err.append(f"assertion {a.get('id')}: interaction shape {a.get('object_ref')} does not refine parent {a['predicate']} shape(s) {parents}")
    return err

def family_cache_checks(doc):
    err=[]
    assertions=doc.get('assertions',[]) or []
    for c in doc.get('contracts',[]) or []:
        for label in c.get('family_labels',[]) or []:
            ok=False
            for a in assertions:
                if a.get('assertion_kind')=='property' and a.get('polarity')=='asserted' and a.get('subject')==c.get('id') and a.get('predicate')=='fiam.property.family-label' and a.get('value')==label and evidence_has_mode(a.get('evidence'),'fiam.evidence.inferred'):
                    ok=True; break
            if not ok: err.append(f"contract {c.get('id')}: cached family label {label} lacks backing inferred fiam.property.family-label assertion")
    return err

def validate(path,schema,canon):
    doc=yaml.safe_load(open(path)); err=[]; warn=[]
    for e in Draft202012Validator(schema).iter_errors(doc): err.append(f"schema {list(e.absolute_path)}: {e.message}")
    terms={t['id']:t for t in canon['terms']}; schemes={s['id']:s for s in canon['schemes']}; deprecated={i for i,t in terms.items() if t.get('status')=='deprecated'}
    objs,objmap,assertion_ids=collect(doc); err+=entity_refs(doc,objs,assertion_ids); err+=participation_checks(doc); err+=family_cache_checks(doc)
    # all fiam.* references must exist
    for p,t in iter_term_refs(doc):
        if t not in terms: err.append(f"{'.'.join(map(str,p))}: unknown Canon term {t}")
        elif t in deprecated: warn.append(f"{'.'.join(map(str,p))}: deprecated Canon term {t}")
    # Canon path/scheme constraints
    for c in canon.get('term_usage_constraints',[]):
        allowed=set(c['schemes'])
        for v in path_values(doc,c['path']):
            vals=v if isinstance(v,list) else [v]
            for t in vals:
                if isinstance(t,str) and t.startswith('fiam.') and t in terms and terms[t].get('scheme') not in allowed:
                    err.append(f"{c['path']}: term {t} from {terms[t].get('scheme')} not allowed; expected {sorted(allowed)}")
    # Operation/Binding consistency. Direct operations should belong to the Binding contract.
    # Embedded contracts are allowed when an evidence-backed embedded-in assertion links the nested contract
    # to an operation of the Binding's contract (e.g. runtime MCP tools carried through tools/call).
    op_contract={}
    contract_ops={}
    for cobj in doc.get('contracts',[]) or []:
        ops=[op.get('id') for op in cobj.get('operations',[]) or [] if isinstance(op,dict) and op.get('id')]
        contract_ops[cobj.get('id')]=ops
        for oid in ops: op_contract[oid]=cobj.get('id')
    embedded_contract_targets=set()
    for a0 in doc.get('assertions',[]) or []:
        if a0.get('assertion_kind')=='relation' and a0.get('polarity')=='asserted' and a0.get('predicate')=='fiam.relation.embedded-in':
            subj=a0.get('subject'); target=a0.get('object_ref')
            if objs.get(subj)=='interface-contract' and objs.get(target)=='operation': embedded_contract_targets.add((subj,target))
    bindings={b.get('id'):b for b in doc.get('bindings',[]) or []}
    for inter in doc.get('interactions',[]) or []:
        oid=inter.get('operation'); bid=inter.get('binding')
        if not oid or not bid or bid not in bindings: continue
        b=bindings[bid]; bc=b.get('contract'); oc=op_contract.get(oid)
        if bc and oc and bc!=oc:
            allowed=any(ec==oc and target in contract_ops.get(bc,[]) for ec,target in embedded_contract_targets)
            if not allowed: err.append(f"interaction {inter.get('id')}: operation {oid} belongs to {oc}, but binding {bid} realizes {bc} and no embedded-in relation explains the indirection")

    # Canon property and relation semantics
    pred_constraints={x['predicate']:x for x in canon.get('assertion_predicate_constraints',[])}
    facet_rules={p:r for r in canon.get('relation_facet_constraints',[]) for p in r.get('predicates',[])}
    for i,a in enumerate(doc.get('assertions',[]) or []):
        pred=a.get('predicate'); kind=a.get('assertion_kind'); pol=a.get('polarity'); pt=terms.get(pred,{})
        expected='fiam.scheme.relation' if kind=='relation' else 'fiam.scheme.property'
        if pt and pt.get('scheme')!=expected: err.append(f"assertions.{i}.predicate: {kind} requires {expected}, got {pt.get('scheme')}")
        if not a.get('evidence'): err.append(f"assertions.{i}: evidence required")
        if a.get('id') in (a.get('supersedes') or []): err.append(f"assertion {a.get('id')}: cannot supersede itself")
        if a.get('id') in (a.get('conflicts_with') or []): err.append(f"assertion {a.get('id')}: cannot conflict with itself")
        if kind=='property' and pol=='asserted' and pred in pred_constraints:
            c=pred_constraints[pred]; val=a.get('value')
            if c.get('value_type') and not value_type_ok(val,c['value_type']): err.append(f"assertions.{i}: {pred} requires {c['value_type']}")
            if c.get('value_schemes') and isinstance(val,str) and val in terms and terms[val].get('scheme') not in c['value_schemes']: err.append(f"assertions.{i}: {pred} value {val} not from {c['value_schemes']}")
        if kind=='relation' and pol=='asserted':
            sk=objs.get(a.get('subject')); ok=objs.get(a.get('object_ref')); dom=pt.get('domain_kinds'); ran=pt.get('range_kinds')
            if dom and sk not in dom: err.append(f"assertions.{i}: {pred} subject kind {sk} not in {dom}")
            if ran and ok not in ran: err.append(f"assertions.{i}: {pred} object kind {ok} not in {ran}")
            if pred in IDENTITY_SAME_KIND:
                if a.get('subject')==a.get('object_ref'): err.append(f"assertions.{i}: {pred} cannot relate entity to itself")
                elif sk and ok and sk!=ok: err.append(f"assertions.{i}: {pred} requires same entity kind, got {sk}->{ok}")
            # Operational identity promotion policy: same-entity-as is query equivalence, not a merge.
            # A declared/governed assertion may establish it directly. Purely inferred promotion
            # requires the Canon threshold and multiple independent evidence signals.
            if pred=='fiam.relation.same-entity-as':
                policy=canon.get('semantic_policies',{}).get('identity_relations',{})
                records=a.get('evidence') if isinstance(a.get('evidence'),list) else [a.get('evidence')]
                records=[r for r in records if isinstance(r,dict)]
                if not any(r.get('mode')=='fiam.evidence.declared' for r in records):
                    threshold=float(policy.get('default_reconciliation_threshold',0.98))
                    need=int(policy.get('required_independent_signals_for_inferred_promotion',2))
                    inferred=[r for r in records if r.get('mode')=='fiam.evidence.inferred' and float(r.get('confidence',0))>=threshold]
                    sigs=set()
                    for r in inferred:
                        src=r.get('source')
                        if isinstance(src,dict): sigs.add((src.get('type'),src.get('ref'),src.get('locator')))
                        else: sigs.add(('source',str(src),None))
                    if len(inferred)<need or len(sigs)<need:
                        err.append(f"assertions.{i}: inferred same-entity-as requires >= {need} independent evidence signals at confidence >= {threshold}; use corresponds-to until governed promotion")
            # Relation facet constraints
            rule=facet_rules.get(pred); facets=a.get('facets') or {}
            if rule:
                allowed=rule.get('allowed_facets',{})
                for fk,fv in facets.items():
                    if fk not in allowed: err.append(f"assertions.{i}: facet {fk} not allowed on {pred}"); continue
                    spec=allowed[fk]
                    if spec.get('type') and not value_type_ok(fv,spec['type']): err.append(f"assertions.{i}.facets.{fk}: expected {spec['type']}")
                    if spec.get('scheme') and isinstance(fv,str) and fv in terms and terms[fv].get('scheme')!=spec['scheme']: err.append(f"assertions.{i}.facets.{fk}: {fv} not from {spec['scheme']}")
                    if spec.get('entity_kind'):
                        if not isinstance(fv,str) or objs.get(fv)!=spec['entity_kind']: err.append(f"assertions.{i}.facets.{fk}: expected ref to {spec['entity_kind']}")
            # Multi-operation Binding shape relation must disambiguate.
            if pred in SHAPE_PART and sk=='binding':
                b=objmap.get(a['subject'],{}); ops=b.get('operations') or contract_ops.get(b.get('contract'),[]) or []
                if len(ops)>1 and not (a.get('facets') or {}).get('operation_ref'): err.append(f"assertions.{i}: Binding shape participation with multiple operations requires facets.operation_ref")
                opref=(a.get('facets') or {}).get('operation_ref')
                if opref and ops and opref not in ops: err.append(f"assertions.{i}: operation_ref {opref} not realized by binding {a['subject']}")
    # Cross-assertion semantic invariants from Canon policies.
    err+=shape_coherence_checks(doc,objs,objmap,canon)
    err+=shape_refinement_checks(doc,objs,objmap,canon,contract_ops)

    # Legacy non-authoritative representations should warn (green corpus uses none).
    for c in doc.get('contracts',[]) or []:
        if 'representations' in c: warn.append(f"contract {c.get('id')}: legacy representations field is non-authoritative")
        for op in c.get('operations',[]) or []:
            if 'representations' in op: warn.append(f"operation {op.get('id')}: legacy representations field is non-authoritative")
    # Executable expectations are mandatory for specimen fixtures, not for ordinary ingested models.
    tests=doc.get('tests') or {}
    checks=tests.get('expected_properties') or []
    if doc.get('instance',{}).get('kind')=='specimen' and not checks:
        err.append('specimen tests.expected_properties must contain executable semantic checks')
    for j,c in enumerate(checks):
        msg=execute_check(c,doc,objs,objmap,terms,deprecated)
        if msg: err.append(f"tests.expected_properties.{j} ({c.get('id')}): {msg}")
    return err,warn

def main():
    ap=argparse.ArgumentParser(); ap.add_argument('paths',nargs='*'); args=ap.parse_args()
    schema=json.load(open(CORE)); Draft202012Validator.check_schema(schema); canon=yaml.safe_load(open(CANON))
    paths=args.paths or sorted(glob.glob(str(ROOT/'specimens/**/*.yaml'),recursive=True))
    failures=0; total_e=total_w=0
    for p in paths:
        e,w=validate(p,schema,canon); total_e+=len(e); total_w+=len(w); failures+=bool(e)
        pp=Path(p); label=pp.relative_to(ROOT) if ROOT in pp.parents else pp
        print(f"{label}: {'PASS' if not e else 'FAIL'} ({len(e)} errors, {len(w)} warnings)")
        for x in e: print('  ERROR:',x)
        for x in w: print('  WARN: ',x)
    print(f"SUMMARY: {len(paths)} files, {failures} failing, {total_e} errors, {total_w} warnings")
    return 1 if failures else 0
if __name__=='__main__': raise SystemExit(main())