AISA v2.0 / Technical documentation / validate_case_journeys.py
validate_case_journeys.py
Python · 268 lines · 15,116 bytes · wiki path 10 Architecture/contracts/validate_case_journeys.py · download the raw file · cited from Case journeys — testing the architecture with real work · Case protocol — requests, records, prompts and results · Contracts · Developer implementation guide
Same folder: Connectors.Abstractions.cs · configuration-change.schema.json · configuration-input.schema.json · connector-capability.schema.json · eval-set.schema.json · execution-obligations.schema.json · ledger-record.schema.json · message-envelope.schema.json · model-turn.schema.json · module-manifest.schema.json · openapi.yaml · realtime-events.schema.json · runtime-state.schema.json · validate_contracts.py · validate_runtime_contracts.py · write-shape.schema.json
"""Offline design conformance. Does not call models, APIs, databases or browsers.
Production must resolve references and the approved plan from authoritative storage.
These fixtures test structural/semantic joins, not provenance or deployed behaviour.
"""
from pathlib import Path
import copy
import json
import re
import hashlib
import yaml
from jsonschema import Draft202012Validator, FormatChecker
from referencing import Registry
from referencing.jsonschema import DRAFT202012
HERE = Path(__file__).resolve().parent
ARCH = HERE.parent
def read(name):
return json.loads((HERE / name).read_text(encoding="utf-8"))
def structural(schema, value):
return ["schema: " + e.message for e in Draft202012Validator(schema, format_checker=FormatChecker()).iter_errors(value)]
def execution_errors(value):
errors = structural(read("execution-obligations.schema.json"), value)
if errors:
return errors
steps = {s["id"]: s for s in value["steps"]}
assertions = {a["id"]: a for a in value["assertions"]}
scope = value["scope"]
terminal = value["case_state"] in ("resolved", "closed")
if len(steps) != len(value["steps"]) or len(assertions) != len(value["assertions"]):
errors.append("duplicate_identity")
if set(scope["members"]) != set(scope["required_assertions"]):
errors.append("incomplete_inventory")
required = {a for ids in scope["required_assertions"].values() for a in ids}
assigned = {a for s in steps.values() if s["required"] for a in s["assertion_ids"]}
if not required <= assigned:
errors.append("unassigned_required_assertion")
for member, ids in scope["required_assertions"].items():
if any(a in assertions and assertions[a]["member"] != member for a in ids):
errors.append("wrong_assertion_member")
if any(a["member"] not in scope["members"] for a in assertions.values()):
errors.append("unknown_member")
def passes(aid):
a = assertions.get(aid)
return bool(a and a["verdict"] == "pass" and a["evidence_ref"] and a["scope_revision"] == scope["revision"])
visiting, visited = set(), set()
def visit(sid):
if sid in visiting:
errors.append("dependency_cycle")
return
if sid in visited:
return
visiting.add(sid)
for dep in steps[sid]["depends_on"]:
if dep not in steps:
errors.append("missing_dependency")
else:
visit(dep)
visiting.remove(sid)
visited.add(sid)
for sid in steps:
visit(sid)
calls = []
for s in steps.values():
if s["kind"] == "external":
if not s.get("external"):
errors.append("missing_external_owner")
if s.get("call_ref") or s.get("grant_ref") or s.get("effect_state"):
errors.append("external_as_platform_effect")
if s["state"] == "verified" and not s.get("external", {}).get("response_ref"):
errors.append("missing_external_response")
if s["kind"] == "effect":
if "effect_state" not in s:
errors.append("missing_effect_disposition")
if s.get("effect_state") in ("applied", "partial", "unknown") and not (s.get("call_ref") and s.get("grant_ref")):
errors.append("missing_effect_authority")
if s.get("call_ref"):
calls.append(s["call_ref"])
if s["state"] == "verified" and s.get("effect_state") != "applied":
errors.append("unproved_effect")
elif s.get("call_ref") or s.get("grant_ref") or s.get("effect_state"):
errors.append("non_effect_has_call")
if s["state"] == "verified":
if not s["evidence_refs"] or not all(passes(a) for a in s["assertion_ids"]):
errors.append("unproved_step")
if any(dep not in steps or steps[dep]["state"] != "verified" for dep in s["depends_on"]):
errors.append("dependency_not_verified")
if s["state"] == "waived" and not s.get("waiver_ref"):
errors.append("unrecorded_waiver")
if s.get("restoration_required"):
restores = [r for r in steps.values() if r.get("restores_step") == s["id"]]
if not restores or any(not r.get("due_at") or not (r["required"] or r.get("successor_case_ref")) for r in restores):
errors.append("missing_restoration_obligation")
if s.get("restores_step") and s["restores_step"] not in steps:
errors.append("unknown_restore_target")
if terminal and s.get("effect_state") in ("unknown", "partial"):
errors.append("unreconciled_effect")
if terminal and s["required"] and s["state"] != "verified":
errors.append("required_work_open")
if len(set(calls)) != len(calls):
errors.append("duplicate_effect_call")
if terminal:
if not all(passes(a) for a in required):
errors.append("incomplete_scope_proof")
if not value["delivery_ref"]:
errors.append("delivery_not_recorded")
if value["post_result"]["state"] == "not_queued" or not value["post_result"]["task_ref"]:
errors.append("learning_not_durable")
if value["case_state"] == "closed" and not value["acceptance_ref"]:
errors.append("acceptance_missing")
return sorted(set(errors))
def delta_errors(value):
errors = structural(read("configuration-change.schema.json"), value)
if errors:
return errors
declared = {s["id"] for s in read("manifests/configuration.json")["stages"]}
selected = set(value["selected_stages"])
targets = {t["id"] for t in value["targets"]}
if not selected <= declared or "S1" not in selected:
errors.append("invalid_stage_selection")
if len(targets) != len(value["targets"]):
errors.append("duplicate_target")
if any(c["stage"] not in selected or c["target_id"] not in targets for c in value["changes"]):
errors.append("unbound_change")
skipped = {s["stage"] for s in value["skipped_stages"]}
if skipped & selected or not skipped <= declared:
errors.append("invalid_skipped_stage")
edges = [(e['from'], e['to']) for e in value['stage_edges']]
if len(set(edges)) != len(edges) or any(a not in selected or b not in selected for a,b in edges):
errors.append('invalid_stage_edge')
remaining = set(selected)
while remaining:
ready = {s for s in remaining if not any(b == s and a in remaining for a,b in edges)}
if not ready:
errors.append('stage_dependency_cycle'); break
remaining -= ready
reached = {'S1'}
for _ in selected:
reached |= {b for a,b in edges if a in reached}
if not selected <= reached:
errors.append('unreachable_selected_stage')
keys = [t["key"] for t in value["templates"]]
if len(set(keys)) != len(keys):
errors.append("duplicate_template")
if value["templates"] and "S2" not in selected:
errors.append("template_without_rating_stage")
if any(not set(t["target_ids"]) <= targets for t in value["templates"]):
errors.append("unbound_template_target")
return errors
def reusable(article, request):
"""The lookup prefilter; semantic applicability still needs current evidence."""
return (article["state"] == "active" and article["family"] == request["family"]
and article["observed_version"] == request["observed_version"]
and request["applicability_proved"] and not article["contradicted"])
def independent_successes(uses, proposing_origin):
return len({u["origin"] for u in uses if u["origin"] != proposing_origin
and u["verified"] and not u["contradicted"] and not u["same_case_family"]})
def main():
failures, checks = [], 0
def check(name, ok):
nonlocal checks
checks += 1
if not ok:
failures.append(name)
cases = json.loads((ARCH / "ui/case-journeys.json").read_text(encoding="utf-8"))["cases"]
expected = {f"{prefix}-{i:02}" for prefix, count in [("DC",24),("DV",19),("CF",25)] for i in range(1,count+1)}
ids = [c["id"] for c in cases]
check("all catalogue cases, exactly once", len(ids) == len(set(ids)) == 68 and set(ids) == expected)
profiles = {p["key"] for m in (HERE / "manifests").glob("*.json") for p in json.loads(m.read_text())["profiles"]}
for c in cases:
check(c["id"] + " source", "**"+c["id"]+" — " in (ARCH/c["source"]).read_text(encoding="utf-8"))
check(c["id"] + " known profiles", bool(c["profiles"]) and set(c["profiles"]) <= profiles)
check(c["id"] + " complete first/next/failure journey", all(len(c[k]) > 25 for k in ("investigate","execute","verify","learn","next","falsifier")))
api = yaml.safe_load((HERE/'openapi.yaml').read_text(encoding='utf-8'))
registry = Registry().with_resource('urn:aisa:openapi', DRAFT202012.create_resource(api))
report = {'request_ref':'paper:instruction','expected_version':'5','response_kind':'report_completed','step_id':'external-step','plan_revision':1,'text':'Reported; verify it.','evidence_refs':[]}
response_schema = api['components']['schemas']['CaseResponse']
check('external report has scoped response shape',not structural(response_schema,report))
missing_revision = {k:v for k,v in report.items() if k!='plan_revision'}
check('external report cannot omit plan revision',bool(structural(response_schema,missing_revision)))
check('client cannot choose an executor identity',bool(structural(response_schema,{**report,'actor_id':'someone-else'})))
check('ordinary answer identifies its question',bool(structural(response_schema,{**report,'response_kind':'answer'})))
traces = json.loads((ARCH/'ui/case-runtime-traces.json').read_text(encoding='utf-8'))
command_keys = []
for cid, trace in traces.items():
check(cid+' ten wire transitions',len(trace['transitions']) == 10)
for t in trace['transitions']:
for call in t['http']:
operation = api['paths'].get(call['path'],{}).get(call['method'].lower())
check(cid+' registered endpoint '+call['path'],operation is not None)
if call['method'] not in ('GET','HEAD'):
command_keys.append(call.get('headers',{}).get('Idempotency-Key'))
if operation and 'body' in call and 'requestBody' in operation:
schema = operation['requestBody']['content']['application/json']['schema']
if schema.get('$ref','').startswith('#/'):
schema = {**schema,'$ref':'urn:aisa:openapi'+schema['$ref']}
check(cid+' request body '+call['path'],not list(Draft202012Validator(schema,registry=registry).iter_errors(call['body'])))
for init in trace['agent_initializations']:
if init['execution_kind']=='llm':
check(cid+' request bytes '+init['profile_key'],hashlib.sha256(init['rendered_input'].encode()).hexdigest()==init['input_sha256'])
check(cid+' no self parent '+init['profile_key'],init['task_id'] != init['parent_task_id'])
else:
check(cid+' deterministic has no prompt '+init['profile_key'],init['rendered_input'] is None)
for response in trace['model_responses']:
check(cid+' model action shape',not structural(read('model-turn.schema.json'),response))
for message in trace['messages']:
check(cid+' '+message['type']+' envelope',not structural(read('message-envelope.schema.json'),message))
check('distinct logical mutations use distinct command keys',None not in command_keys and len(command_keys)==len(set(command_keys)))
for f in read("fixtures/execution-obligations.examples.json"):
errors = execution_errors(f["value"])
check(f["name"] + ": " + str(errors), (not errors) if f["expected"] == "valid" else f["expected"] in errors)
delta = read("fixtures/configuration-change.valid.json")
check("multi-template delta", not delta_errors(delta))
schemas = [read('configuration-change.schema.json'), json.loads((ARCH/'20 Configuration/contracts/product.schema.json').read_text())]
input_registry=Registry().with_resources((s['$id'],DRAFT202012.create_resource(s)) for s in schemas)
check('delta accepted at configuration-input seam',not list(Draft202012Validator(read('configuration-input.schema.json'),registry=input_registry).iter_errors(delta)))
for name, mutate, code in [
("duplicate template", lambda x: x["templates"].append(copy.deepcopy(x["templates"][0])), "duplicate_template"),
("missing target", lambda x: x["templates"][0].update(target_ids=["absent"]), "unbound_template_target"),
("rating stage skipped", lambda x: x["selected_stages"].remove("S2"), "template_without_rating_stage"),
("undeclared stage", lambda x: x["selected_stages"].append("S99"), "invalid_stage_selection"),
("cycle in delta graph", lambda x: x['stage_edges'].append({'from':'release','to':'S1'}), 'stage_dependency_cycle'),
("detached delta stage", lambda x: x['stage_edges'].pop(), 'unreachable_selected_stage'),
]:
v = copy.deepcopy(delta); mutate(v); check(name, code in delta_errors(v))
article = dict(state="active", family="motor-myr", observed_version="v1", contradicted=False)
request = dict(family="motor-myr", observed_version="v1", applicability_proved=True)
check("current verified precedent", reusable(article, request))
for field, value in [("state","proposed"),("state","retired"),("observed_version","v0"),("contradicted",True),("family","cargo")]:
changed = {**article, field:value}; check("reject stale/incompatible precedent " + field + str(value), not reusable(changed, request))
check("retrieval alone is not reuse", not reusable(article,{**request,"applicability_proved":False}))
uses = [dict(origin="request-b",verified=True,contradicted=False,same_case_family=False)]*2
check("mirror does not count twice", independent_successes(uses,"request-a") == 1)
check("proposing case does not validate itself", independent_successes(uses,"request-b") == 0)
check("subcase/fork does not count", independent_successes([{**uses[0],"same_case_family":True}],"request-a") == 0)
check("reported-only use does not count", independent_successes([{**uses[0],"verified":False}],"request-a") == 0)
check("two independent proved uses", independent_successes(uses+[{**uses[0],"origin":"request-c"}],"request-a") == 2)
html = (ARCH / "ui/case_journeys.html").read_text(encoding="utf-8")
for image in set(re.findall(r"wf-[a-z-]+\.svg",html)):
check("wireframe exists " + image, (ARCH/"ui"/image).is_file())
print(f"{checks} journey/contract conformance checks; {len(failures)} failures")
for failure in failures:
print("FAIL", failure)
return bool(failures)
if __name__ == "__main__":
raise SystemExit(main())