#!/usr/bin/env python3 """ Validate the five platform seams (Delivery § 1.4.4 item 5: "contracts written and validated against a deliberately broken fixture"). Two layers, as in validate_product.py: 1. structural — the JSON Schema (jsonschema, Draft 2020-12); 2. semantic — rules a schema cannot express, RECOMPUTED from the document. python validate_contracts.py # runs every fixture: valid ones must pass, broken ones must fail python validate_contracts.py # validates one document against the seam it declares A document declares its seam with "$schema_ref" (or is matched by fixture name). Exit 0 = every expectation met. Draft (inter-stage) contracts under drafts/ are validated the same way but with additionalProperties allowed — the platform validates only their required core (D139). """ from __future__ import annotations import hashlib, json, sys, io from pathlib import Path HERE = Path(__file__).parent sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") try: import jsonschema from jsonschema import Draft202012Validator from referencing import Registry from referencing.jsonschema import DRAFT202012 except ImportError: # pragma: no cover jsonschema = None SEAMS = { "route-packet": "drafts/route-packet.schema.json", "message-envelope": "message-envelope.schema.json", "connector-capability": "connector-capability.schema.json", "eval-set": "eval-set.schema.json", "ledger-record": "ledger-record.schema.json", "write-shape": "write-shape.schema.json", # drafts (D139) — required core only, extensions allowed "configuration-summary": "drafts/configuration-summary.schema.json", "sign-out-document": "drafts/sign-out-document.schema.json", "stage-contract": "drafts/stage-contract.schema.json", "normalised-arrival": "drafts/normalised-arrival.schema.json", "solution-packet": "drafts/solution-packet.schema.json", "experience-entry": "drafts/experience-entry.schema.json", "brief": "drafts/brief.schema.json", "branch-set": "drafts/branch-set.schema.json", "cross-repo-shape": "drafts/cross-repo-shape.schema.json", "handover-packet": "drafts/handover-packet.schema.json", "skill-manifest": "drafts/skill-manifest.schema.json", } GENESIS = "0" * 64 def sha256_hex(s: str) -> str: return hashlib.sha256(s.encode("utf-8")).hexdigest() def canonical(obj) -> str: """A stand-in for SERDICA-JCS-1 (RFC 8785 JCS): sorted keys, no whitespace, UTF-8. The C# AI.Contracts canonicaliser is authoritative. These fixtures use the compatible ASCII/integer subset; this helper is not a complete RFC 8785 implementation.""" return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) # ---------------------------------------------------------------- semantic rules per seam def sem_message_envelope(d, out): t, p = d.get("type"), d.get("payload") or {} if t == "Consult" and p.get("depth", 1) != 1: out.append("consult depth must be 1 (Agents § 5 rule 5)") if t == "Consult" and "." in str(p.get("domain", "")) and "/" not in str(p.get("domain", "")): out.append("a consult is addressed to a knowledge domain (a tree path), never to an agent profile (Agents § 5 rule 1)") if t == "ConsultReply" and p.get("outcome") == "no_answer" and d.get("trust_class") != "absent": out.append("a no_answer ConsultReply carries trust_class 'absent'") if t in ("Plan", "Result") and "artefact" not in d: out.append(f"{t} must reference an artefact — the body never travels in the message") if t == "Spawn" and p.get("budget_minutes", 1) <= 0 and p.get("kind") != "consult": out.append("a spawned task needs a positive time budget (D135)") if any(k not in ("type", "version", "message_id", "case_id", "task_path", "recipient_task_id", "sender", "occurred_on", "deadline", "priority", "trust_class", "artefact", "payload") for k in d): out.append("free text outside the typed payload is not an output (Agent Runtime § 2 rule 5)") def sem_connector_capability(d, out): if d.get("kind") == "operation": if d.get("effect_class") == "irreversible" and d.get("recovery") == "retryable": out.append("an irreversible operation cannot be retryable") if d.get("effect_of_timeout") == "unknown" and d.get("recovery") != "reconcile_first": out.append("effect_of_timeout 'unknown' requires recovery 'reconcile_first'") ev = d.get("evidence_returned") or [] if d.get("effect_class") not in (None, "read") and "row_count" not in ev and "returned_ids" not in ev: out.append("a write operation must return row_count or returned_ids as evidence (Trust and Data § 5)") route = d.get("route") or {} if d.get("kind") == "reach" and route.get("kind") == "direct" and "verified_on" not in route and d.get("state") == "published": out.append("a published direct reach must be verified (CK_SCOPES_WRITE_VERIFIED)") if route.get("kind") not in (None, "direct", "unreachable"): out.append(f"route kind '{route.get('kind')}' is not a route: direct connections only (D132)") def sem_eval_set(d, out): cases = d.get("cases") or [] keys = [c.get("case_key") for c in cases] if len(keys) != len(set(keys)): out.append("case_key must be unique within a set") if d.get("grader_profile_key") == d.get("owner_profile_key"): out.append("the grader is the isolated platform.grader, never the author's own profile (Agent Framework § 6.5)") if d.get("corpus_kind") == "merged_changes": for c in cases: lp = c.get("lineage_pair") or {} if not (lp.get("master_port") or lp.get("no_counterpart_reason")): out.append(f"case {c.get('case_key')}: a Development case is a lineage pair — master_port or no_counterpart_reason (G36)") def sem_ledger_record(d, out): body = {k: v for k, v in d.items() if k not in ("prev_hash", "record_hash", "_comment")} ph, rh = d.get("prev_hash", ""), d.get("record_hash", "") expected = sha256_hex(ph + canonical(body)) if rh != expected: out.append(f"record_hash does not equal SHA-256(prev_hash || canonical(record)); expected {expected[:12]}... — the record was changed after hashing") if ph and ph == rh: out.append("prev_hash equals record_hash") if d.get("kind") == "WRITE_EXECUTED": if str(d.get("actual_action", "")).strip().lower().startswith("select"): out.append("a WRITE_EXECUTED record whose action is a SELECT is misclassified") if not (d.get("lineage") or {}).get("grant_id"): out.append("a WRITE_EXECUTED record names the grant it ran under (Trust and Data § 5)") for k in ("requested_action", "actual_action"): v = str(d.get(k, "")) if any(t in v for t in ("\n{", "[{", "ROWS:")): out.append(f"{k} looks like it carries result rows — never (Trust and Data § 5)") def sem_write_shape(d, out): import re if d.get("object") == "write_shape": declared = {p.get("name") for p in d.get("parameters") or []} tpl, td = str(d.get("template", "")), str(d.get("teardown_template", "")) used = set(re.findall(r"{{\s*([a-z][a-z0-9_]*)\s*}}", tpl)) used_td = set(re.findall(r"{{\s*([a-z][a-z0-9_]*)\s*}}", td)) if used - declared: out.append(f"template uses undeclared parameters: {sorted(used - declared)} (auditor check 1)") if used_td - declared - used: out.append(f"teardown uses parameters the template never binds: {sorted(used_td - declared - used)}") if re.search(r"\blike\s+'%", tpl, re.I) or "and similar" in tpl.lower(): out.append("template carries an open predicate (unbounded LIKE) — refused by exactness") if not re.search(r"^\s*\d+\s*$|\.length\b|\bcount\b", str(d.get("expected_counts", "")), re.I): out.append("expected_counts must be a number or an expression over the plan's numbers, never an adjective (auditor check 1)") if d.get("write_class") in ("W1", "W2") and d.get("target_class") != "working": out.append("W1/W2 are working-environment classes only (Gating § 3)") if d.get("state") == "approved" and "expires_on" not in d: out.append("an approved shape carries expires_on (D133)") if re.search(r"^\s*DELETE\b", td, re.I) and re.search(r"^\s*INSERT\b", tpl, re.I): tpl_keys, td_keys = used, used_td if not tpl_keys <= td_keys and td_keys < tpl_keys: out.append("teardown of an INSERT is keyed by fewer parameters than the insert — it would remove more than the write added (auditor check 3)") else: volatile = {"artefact_hash", "grant_id", "auditor_verdict_required", "preflight"} body = {k: v for k, v in d.items() if k not in volatile} if d.get("artefact_hash") != sha256_hex(canonical(body)): out.append("write packet semantic hash mismatch") seen_steps = set() for step in d.get("steps") or []: if step.get("step_id") in seen_steps: out.append("duplicate write step id") for binding in (step.get("returned_bindings") or {}).values(): if binding.get("source_step") not in seen_steps: out.append("returned binding must reference an earlier step") seen_steps.add(step.get("step_id")) for s in d.get("steps") or []: if not isinstance(s.get("expected_count"), int): out.append(f"step {s.get('step_id')}: expected_count must be a number") for k, v in (s.get("bound_parameters") or {}).items(): if isinstance(v, str) and v.strip() in ("", "?", "TBD"): out.append(f"step {s.get('step_id')}: parameter {k} is unbound") if not (d.get("teardown") or {}).get("derived_from_shape"): out.append("teardown must be derived from the shape, not authored later (auditor check 3)") def sem_draft(d, out): # drafts: only the required core is enforced; a document may extend itself per case (D139) if d.get("x_status") not in (None, "draft"): out.append("inter-stage contracts are drafts (D139)") SEMANTIC = { "message-envelope": sem_message_envelope, "connector-capability": sem_connector_capability, "eval-set": sem_eval_set, "ledger-record": sem_ledger_record, "write-shape": sem_write_shape, } def load(p: Path): return json.loads(p.read_text(encoding="utf-8")) def validate(doc_path: Path, seam: str) -> list[str]: doc = load(doc_path) schema = load(HERE / SEAMS[seam]) findings: list[str] = [] if jsonschema is None: findings.append("jsonschema not installed — structural layer skipped (pip install jsonschema)") else: local = [load(p) for p in HERE.rglob('*.schema.json')] registry = Registry().with_resources((s['$id'], DRAFT202012.create_resource(s)) for s in local if '$id' in s) v = Draft202012Validator(schema, registry=registry) for e in sorted(v.iter_errors(doc), key=lambda e: list(e.path)): findings.append("schema: " + "/".join(str(x) for x in e.path) + ": " + e.message[:160]) if seam == "skill-manifest": from validate_tool_packages import skill_errors findings.extend(skill_errors(doc)) else: SEMANTIC.get(seam, sem_draft)(doc, findings) return findings def seam_of(p: Path) -> str | None: name = p.name for s in SEAMS: if name.startswith(s): return s try: return load(p).get("$schema_ref") except Exception: return None def main(argv): if len(argv) > 1: p = Path(argv[1]) seam = seam_of(p) if not seam: print(f"cannot tell which seam {p.name} belongs to; name it .*.json or set $schema_ref"); return 2 f = validate(p, seam) print("\n".join(" BLOCK " + x for x in f) if f else f" OK {p.name} ({seam})") return 1 if f else 0 # Runtime and case-journey examples have their own semantic conformance runners. fixtures = sorted(p for p in (HERE / "fixtures").glob("*.json") if not p.name.startswith(("model-turn.", "runtime-state.", "execution-obligations.", "configuration-change.", "tool-package."))) bad = 0 for p in fixtures: seam = seam_of(p) if not seam: print(f" ? {p.name}: unknown seam"); bad += 1; continue f = validate(p, seam) expect_broken = ".broken." in p.name ok = bool(f) == expect_broken tag = "OK " if ok else "WRONG " kind = "broken->refused" if expect_broken and f else ("valid->accepted" if not expect_broken and not f else ("broken->ACCEPTED" if expect_broken else "valid->REFUSED")) print(f" {tag} {p.name:48s} {kind} ({len(f)} finding(s))") for x in f: print(" - " + x) bad += 0 if ok else 1 print(f"\n{len(fixtures)} fixtures, {bad} unexpected outcome(s)") return 1 if bad else 0 if __name__ == "__main__": sys.exit(main(sys.argv))