#!/usr/bin/env python3 """ ACF conformance-fixture validator and v0.2 projector. Third parties can run this against standards/fixtures/ with only the Python standard library plus `jsonschema` installed: python3 standards/validate_fixtures.py What it checks: 1. The v0.2 JSON Schema (standards/acf-v0.2.schema.json) is itself a valid Draft 2020-12 schema. 2. Every engine fixture (f01-f07) satisfies the v0.1 verdict invariants of agent-correction-format-v0.1.md: verdict in {MATCH, REVIEW, BLOCK}; MATCH carries no corrections; every correction carries dimension/action/reason; the non-correctable fixture's failing dimension is marked correctable: false and no correction is emitted for it. 3. Every fixture projects to the reshaped v0.2 form defined by agent-correction-format-v0.2.md: failing dimensions become messages[] entries in the host's own shape (type, code, severity, path, content) with the code drawn from the com.provecart.conformity vocabulary and path rendered as a JSONPath; the correction's expected/actual/action ride as extension fields on the message; and one namespaced conformity object carries the verdict, the record id, the mandate digest, the subject digest and the expiry. requires_buyer_approval is false on every engine projection, because every engine-emitted correction brings the subject INTO the buyer's existing authorization. Projections are written to standards/fixtures/projections/. 4. The hand-authored consumer-behavior fixture (f08) validates against the schema's message definition directly. """ import json import re import sys from pathlib import Path HERE = Path(__file__).resolve().parent FIX = HERE / "fixtures" PROJ = FIX / "projections" # Path projection rules (spec v0.2): the v0.1 payload locates a # failing dimension by dimension name (and, for bookings, a subject # field name); on the host's messages[] the same field is addressed # by a JSONPath into the checkout object. The paths below address the # reference implementation's subject shapes; a host checkout object # addresses the equivalent field of its own shape by the same rule. POINTER_BY_DIMENSION = { ("cart", "price"): "/total_price", ("cart", "quantity"): "/items", ("cart", "item_match"): "/items/0/name", ("cart", "variant"): "/items/0/variant", ("cart", "currency"): "/currency", ("booking", "price"): "/price/total", ("booking", "currency"): "/currency", ("reservation", "price"): "/total_amount", ("reservation", "reservation_time"): "/time", ("reservation", "reservation_date"): "/date", ("reservation", "party_size"): "/party_size", } POINTER_BY_FIELD = { "departure_date": "/legs/0/depart_at", "return_date": "/legs/1/depart_at", "time": "/time", "date": "/date", } KNOWN_ACTIONS = { "remove_item", "replace_item", "change_variant", "reduce_quantity", "reduce_total", "change_currency", "correct_traveler_name", "change_airport", "change_travel_date", "change_departure_time", "add_itinerary", "rebuild_itinerary", "change_cabin", "change_fare_terms", "add_hotel_stay", "change_room", "change_provider", "change_location", "change_date", "change_time", "change_party_size", "correct_reservation_name", "reduce_deposit", "change_cancellation_terms", } failures = [] def check(name, cond, detail=""): print(f"{'PASS' if cond else 'FAIL'} {name}" + (f" {detail}" if detail and not cond else "")) if not cond: failures.append(name) def to_jsonpath(pointer): out = "$" for part in pointer.split("/"): if not part: continue if part.isdigit(): out += f"[{part}]" else: out += f".{part}" return out def project(fixture): resp = fixture["response"] kind = fixture["kind"] corrections = {c["dimension"]: c for c in resp.get("corrections") or []} messages = [] for r in resp["results"]: if r["status"] == "pass": continue dim = r["dimension"] c = corrections.get(dim) if r.get("correctable") is False: severity = "unrecoverable" elif resp["verdict"] == "REVIEW": severity = "requires_buyer_review" else: severity = "recoverable" field = (c or {}).get("field") if field and field in POINTER_BY_FIELD: pointer = POINTER_BY_FIELD[field] else: pointer = POINTER_BY_DIMENSION.get((kind, dim), "") msg = { "type": "error", "code": "com.provecart.conformity." + r.get("code", "unspecified").lower(), "severity": severity, "path": to_jsonpath(pointer) if pointer else "$", "content": (r.get("reasons") or [""])[0], } if c: msg["expected"] = c.get("expected") msg["actual"] = c.get("actual") msg["action"] = c["action"] msg["uncertain"] = r["status"] == "soft_fail" msg["requires_buyer_approval"] = False messages.append(msg) conformity = { "verdict": resp["verdict"], "record_id": resp["record_id"], "mandate_ref": {"digest": "sha256:fixture-mandate", "scheme": "hmac-sha256"}, "subject_digest": "sha256:" + resp["cart_hash"] if resp.get("cart_hash") else "sha256:fixture", "issued_at": "2026-11-05T00:00:00Z", "expires_at": "2026-11-05T00:15:00Z", "previous_record_id": None, } if corrections: conformity["max_attempts"] = resp.get("max_attempts", 3) return {"messages": messages, "com.provecart.conformity": conformity} def main(): try: import jsonschema except ImportError: print("jsonschema is required: pip install jsonschema") return 2 schema = json.loads((HERE / "acf-v0.2.schema.json").read_text()) jsonschema.Draft202012Validator.check_schema(schema) print("PASS schema is a valid Draft 2020-12 schema") message_validator = jsonschema.Draft202012Validator( {**schema["$defs"]["conformityMessage"], "$schema": "https://json-schema.org/draft/2020-12/schema"}) object_validator = jsonschema.Draft202012Validator( {**schema["$defs"]["conformityObject"], "$schema": "https://json-schema.org/draft/2020-12/schema"}) PROJ.mkdir(exist_ok=True) for path in sorted(FIX.glob("f0*.json")): fixture = json.loads(path.read_text()) fid = fixture["fixture_id"] resp = fixture.get("response", {}) if fixture["kind"] == "consumer-behavior": for msg in resp.get("messages", []): errs = list(message_validator.iter_errors(msg)) check(f"{fid}: message validates against v0.2 " f"message definition", not errs, str(errs[:1])) first = resp["messages"][0] check(f"{fid}: unknown action is not a known action", first["action"] not in KNOWN_ACTIONS) check(f"{fid}: code is in the conformity vocabulary " f"namespace", re.match(r"^com\.provecart\.conformity\.", first["code"]) is not None) continue check(f"{fid}: verdict is a valid verdict", resp["verdict"] in {"MATCH", "REVIEW", "BLOCK"}) corrections = resp.get("corrections") or [] if resp["verdict"] == "MATCH": check(f"{fid}: MATCH carries no corrections", not corrections) for c in corrections: check(f"{fid}: correction {c.get('action')} carries " f"dimension/action/reason", all(k in c for k in ("dimension", "action", "reason"))) failing_uncorrectable = [ r for r in resp["results"] if r.get("correctable") is False and r["status"] != "pass"] if failing_uncorrectable: dims = {r["dimension"] for r in failing_uncorrectable} check(f"{fid}: no correction emitted for " f"non-correctable dimension(s) {sorted(dims)}", not ({c["dimension"] for c in corrections} & dims)) projection = project(fixture) for msg in projection["messages"]: errs = list(message_validator.iter_errors(msg)) check(f"{fid}: projected message " f"{msg['code'].rsplit('.', 1)[-1]} validates", not errs, str(errs[:1])) errs = list(object_validator.iter_errors( projection["com.provecart.conformity"])) check(f"{fid}: projected conformity object validates", not errs, str(errs[:1])) (PROJ / f"{fid}.v0.2.json").write_text( json.dumps(projection, indent=2) + "\n") print() if failures: print(f"{len(failures)} FAILURE(S)") return 1 print("ALL FIXTURE CHECKS PASS") return 0 if __name__ == "__main__": sys.exit(main())