AISA v2.0 / Technical documentation / validate_contracts.py
validate_contracts.py
Python · 257 lines · 13,293 bytes · wiki path 10 Architecture/contracts/validate_contracts.py · download the raw file · cited from Contracts
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 · tool-package.schema.json · validate_case_journeys.py · validate_runtime_contracts.py · validate_tool_packages.py · write-shape.schema.json
#!/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 <file.json> # 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 <seam>.*.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))