"""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 route_errors(value): errors=structural(read('drafts/route-packet.schema.json'),value) registered={json.loads(p.read_text())['module_id'] for p in (HERE/'manifests').glob('*.json') if json.loads(p.read_text())['kind']!='root'} if value.get('route') not in registered|{'waiting'}:errors.append('unregistered_route') return errors 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'] command_schema=api['components']['schemas']['Command'] context_command={'command':'steer','text':'Recheck the selected sub-plan','context':{'task_id':'fixture-child','plan_id':'fixture-plan','plan_revision':1}} check('composer names exact task and plan revision',not structural(command_schema,context_command)) missing_context_revision=copy.deepcopy(context_command);del missing_context_revision['context']['plan_revision'] check('composer plan reference needs its revision',bool(structural(command_schema,missing_context_revision))) forged_context=copy.deepcopy(context_command);forged_context['context']['grant_id']='invented' check('composer context cannot supply target authority',bool(structural(command_schema,forged_context))) 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')) specs = json.loads((ARCH/'ui/case-ui-specs.json').read_text(encoding='utf-8'))['cases'] manifests={p.stem:json.loads(p.read_text()) for p in (HERE/'manifests').glob('*.json')} check('UX fixtures and protocol traces cover the catalogues',set(specs)==set(traces)==expected) command_keys = [] for cid, trace in traces.items(): j=trace['journey'] source_case=next(c for c in cases if c['id']==cid) catalogue=(ARCH/source_case['source']).read_text(encoding='utf-8') paragraph=re.search(r'\*\*'+re.escape(cid)+r' — [^\n]+',catalogue).group(0) current_path=paragraph.split('*Current illustrated path [P]:*',1)[-1].split('*Journey:*',1)[0] cursor=0;ordered=True for stage in j['paths']['first']: token=stage['stage']+' '+stage['label'];position=current_path.find(token,cursor) if position<0:ordered=False;break cursor=position+len(token) check(cid+' current catalogue walk matches selected stage order','*Current illustrated path [P]:*' in paragraph and ordered) check(cid+' five shared patterns',set(j['screens'])=={'queue','workspace','decision','result','learning'}) check(cid+' real case content',bool(specs[cid]['request']) and bool(specs[cid]['outcome']) and len(specs[cid]['rows'])==3 and all(set(r)=={'label','before','after'} and all(r.values()) for r in specs[cid]['rows'])) check(cid+' no universal ten-step workflow','transitions' not in trace) check(cid+' owning Support closure',j['closure']['module']=='support' and j['closure']['stage']=='closure' and j['closure']['protocol']=='delivery') expected_states={s['uid'] for states in j['stage_states'].values() for s in states} check(cid+' every stage state has one trace',set(trace['stage_traces'])==expected_states) for stage_uid,states in j['stage_states'].items(): owners=[next(a['task_id'] for a in trace['stage_traces'][s['uid']]['agent_initializations'] if a['profile_key']==s['profile']) for s in states] check(cid+' card states resume the owning stage task '+stage_uid,len(set(owners))==1) for scenario,path in j['paths'].items(): projected=[uid for g in j['groups'][scenario] for uid in (g['steps'] if g['kind']=='sequence' else [uid for lane in g['lanes'] for uid in lane])] check(cid+' complete grouped '+scenario,set(projected)=={s['uid'] for s in path} and len(projected)==len(path)) check(cid+' recovery/learning outside normal path',not any(s['screen']=='learning' or (s['protocol']=='reversal' and not j['ui']['restoration_required']) for s in path)) for uid,t in trace['stage_traces'].items(): s=t['stage_context'] check(cid+' allowed screen '+uid,s['screen'] in j['screens']) if s['stage']: stage=next((x for x in manifests[s['module']]['stages'] if x['id']==s['stage']),None) check(cid+' declared stage '+uid,stage is not None) if stage: check(cid+' stage contracts '+uid,s['input_contracts']==stage['consumes'] and s['output_contracts']==stage['produces'] and s['done']==stage['done']) check(cid+' selected profiles '+uid,set(s['profiles'])<=profiles and s['profile'] in s['profiles']) check(cid+' case identity and parent join '+uid, bool(t['case_context']['join']) and bool(t['case_context']['projection'])) check(cid+' customer API projection '+uid,all(x['path'].startswith('/client/') for x in t['surface_requests']['customer'])) check(cid+' internal API projection '+uid,all(not x['path'].startswith('/client/') for x in t['surface_requests']['internal'])) if s['decision_actor']!='customer': check(cid+' no technical customer decision '+uid,not any(x['path'].endswith(('/accept','/decide')) and x['method']=='POST' for x in t['surface_requests']['customer'])) templates={a['task_id']:a for a in trace['agent_initializations']} for init in t['agent_initializations']: template=templates[init['template_task_id']] check(cid+' stage task binding '+uid,init['task_text']==t['stage_task'] and init['profile_key']==template['profile_key']) if init['execution_kind']=='llm': assembled=template['rendered_input']+init['rendered_suffix'] check(cid+' exact assembled stage bytes '+uid,hashlib.sha256(assembled.encode()).hexdigest()==init['input_sha256']) check(cid+' stage-local prompt '+uid,t['stage_task'] in assembled and 'Working discipline' in assembled and '[EXECUTABLE KNOWLEDGE]' in assembled) else: check(cid+' deterministic stage has no prompt '+uid,init['rendered_suffix'] is None and init['input_sha256'] is None) owner=next(a for a in t['agent_initializations'] if a['profile_key']==s['profile']) check(cid+' chat context binds stage owner '+uid,t['chat_protocol']['body']['context']['task_id']==owner['task_id']) check(cid+' sub-plans retain their parent '+uid,all(x['parent_task_id']==owner['task_id'] and x['task_id']!=owner['task_id'] for x in t['chat_protocol']['subplans'])) for call in t['http']+[x for calls in t['surface_requests'].values() for x in calls]+[t['chat_protocol']]: 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))) def path(cid,scenario='first'): return traces[cid]['journey']['paths'][scenario] check('Data looks before investigating',[s['stage'] for s in path('DC-02')]==['S3','S2','S3','S4','S5','S6']) check('Compatible method removes investigation but preserves proof',[s['stage'] for s in path('DC-02','similar')]==['S3','S4','S5','S6']) check('Known access skill skips unnecessary investigation',not any(s['stage']=='S2' for s in path('DC-03'))) check('Access fallback selects access investigator',next(s for s in traces['DC-03']['journey']['auxiliary'] if s['stage']=='S2')['profile']=='support.investigator.access') check('UI source selects UI implementer',next(s for s in path('DV-10') if s['stage']=='S3')['profiles']==['source.implementer.serdica-ui']) check('Document source joins both repositories',set(next(s for s in path('DV-02') if s['stage']=='S3')['profiles'])=={'source.implementer.serdica-backend','source.implementer.db-plsql'}) for cid in ['DC-19','CF-01']: check(cid+' required restoration precedes closure',path(cid)[-1]['stage']=='S7' and traces[cid]['journey']['ui']['restoration_required']) check(cid+' external action and platform effect share S5',any(s['protocol']=='external' for s in path(cid)) and any(s['protocol']=='execution' for s in traces[cid]['stage_traces'].values() for s in [s['stage_context']] if s['stage']=='S5')) check('CF-22 master-data remains Data',traces['CF-22']['journey']['ui']['working_module']=='data') check('CF-05 subset and parallel lanes',[s['stage'] for s in path('CF-05')]==['S1','S2','S4','S5','joint','release'] and any(g['kind']=='parallel' for g in traces['CF-05']['journey']['groups']['first'])) check('Wording does not invent rating work',[s['stage'] for s in path('CF-11')]==['S1','S5','joint','release']) check('Report-only reconciliation never applies',len(path('CF-25'))==1 and not any(s['protocol']=='release' for s in path('CF-25'))) check('Development real seven-stage path: the producer tests inside Implementation',[s['stage'] for s in path('DV-10')]==['S'+str(i) for i in range(1,8)]) check('Core offers assess and communicate without Source stages',all(all(s['module']=='commercial' for s in path(cid)) and [s['stage'] for s in path(cid)]==['O1','O2','O3','O4'] and traces[cid]['journey']['ui']['core_terms_pending'] for cid in ['DV-09','DV-16'])) offer_route={'case_id':'fixture-offer','request_type':'change_request','route':'commercial','offer_type':'core_development','reason':'Shared work requires assessed terms','evidence_refs':['paper:brief'],'open_questions':[]} check('Registered commercial route accepted',not route_errors(offer_route)) check('Unregistered route refused',bool(route_errors({**offer_route,'route':'invented-module'}))) check('Development recovery is S8',any(s['stage']=='S8' and s['branch']=='exception' for s in traces['DV-02']['journey']['auxiliary'])) check('Linked document Configuration includes actual preview/release',[(s['stage'],s['protocol']) for s in traces['DV-02']['journey']['linked']]==[('S1','scope'),('S5','external'),('joint','preview'),('release','release')]) check('Programmes remain scoped referrals',all(len(path(cid))==1 and path(cid)[0]['protocol']=='referral' for cid in ['DV-03','DV-08'])) 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") check('Adaptive HTML replaces repeated static case drawings','