validate_runtime_contracts.py

Python · 216 lines · 9,411 bytes · wiki path 10 Architecture/contracts/validate_runtime_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_contracts.py · validate_tool_packages.py · write-shape.schema.json

#!/usr/bin/env python3
"""Offline contract/conformance checks. No model, estate, database or network calls.

These validate the specification and its examples, not a platform implementation.
Run: python validate_runtime_contracts.py
"""
from __future__ import annotations

import copy
import json
from pathlib import Path

from jsonschema import Draft202012Validator

HERE = Path(__file__).parent


def load(name):
    return json.loads((HERE / name).read_text(encoding="utf-8"))


def schema_errors(name, document):
    schema = load(name + ".schema.json")
    Draft202012Validator.check_schema(schema)
    return [e.message for e in Draft202012Validator(schema).iter_errors(document)]


def manifest_errors(doc):
    errors = schema_errors("module-manifest", doc)
    if errors:
        return errors
    profiles = {p["key"] for p in doc["profiles"]}
    stages = {s["id"] for s in doc["stages"]}
    if len(profiles) != len(doc["profiles"]):
        errors.append("duplicate profile key")
    if len(stages) != len(doc["stages"]):
        errors.append("duplicate stage id")
    if doc["root_profile"] not in profiles or doc["entry_stage"] not in stages:
        errors.append("entry or root does not resolve")
    if doc["kind"] == "embedded" and (doc["case_module"] != "support" or doc["root_profile"] != "support.root"):
        errors.append("the embedded data graph retains the Support case and root")
    for stage in doc["stages"]:
        selected = {stage["profile"]} | set(stage.get("profile_variants", {}).values())
        if not selected <= profiles:
            errors.append("stage profile does not resolve")
        for contract in stage["consumes"] + stage["produces"]:
            if contract.startswith("paper:"):
                continue
            candidates = [HERE / (contract + ".schema.json"), HERE / "drafts" / (contract + ".schema.json")]
            if contract == "product":
                candidates.append(HERE.parent / "20 Configuration/contracts/product.schema.json")
            if not any(p.exists() for p in candidates):
                errors.append("unresolved stage contract: " + contract)
    graph = {key: [] for key in stages}
    for edge in doc["edges"]:
        if edge["from"] not in stages or edge["to"] not in stages:
            errors.append("edge endpoint does not resolve")
        elif edge["kind"] == "dependency":
            graph[edge["from"]].append(edge["to"])
    visiting, seen = set(), set()

    def visit(node):
        if node in visiting:
            errors.append("dependency cycle must be an explicit rework edge")
            return
        if node in seen:
            return
        visiting.add(node)
        for child in graph[node]:
            visit(child)
        visiting.remove(node)
        seen.add(node)

    for node in graph:
        visit(node)
    return errors


def model_errors(doc):
    errors = schema_errors("model-turn", doc)
    if errors:
        return errors
    if doc["kind"] == "tool_calls":
        if not doc["tool_calls"] or doc["message"] is not None:
            errors.append("tool_calls needs calls and no message")
        for call in doc["tool_calls"]:
            try:
                args = json.loads(call["arguments_json"])
                if not isinstance(args, dict):
                    errors.append("tool arguments must be an object")
            except ValueError:
                errors.append("tool arguments are malformed JSON")
    else:
        if doc["tool_calls"] or doc["message"] is None:
            return errors + ["message needs a message and no calls"]
        msg = doc["message"]
        try:
            payload = json.loads(msg["payload_json"])
            definitions = load("message-envelope.schema.json")["$defs"]
            schema = {"$ref": "#/$defs/" + msg["type"], "$defs": definitions}
            errors += [e.message for e in Draft202012Validator(schema).iter_errors(payload)]
        except ValueError:
            errors.append("message payload is malformed JSON")
        if msg["type"] in ("Plan", "Result") and not msg["paper_ref"]:
            errors.append("Plan and Result require a stored paper")
    return errors


def state_errors(doc):
    errors = schema_errors("runtime-state", doc)
    if errors:
        return errors
    pending = doc["pending_call_ids"] + doc["pending_gate_ids"] + doc["pending_message_ids"]
    if doc["phase"] == "complete" and pending:
        errors.append("completed task cannot retain pending work")
    if doc["phase"] == "waiting" and not pending:
        errors.append("waiting state must name its wake-up dependency")
    for key in ("manifest_policy_id", "manifest_hash", "case_type_id", "profile_id", "framework_version", "runtime_version"):
        if not doc["pins"][key]:
            errors.append("missing pin: " + key)
    return errors


def required_effect_gates(effect, traits):
    """Reference truth table for Gating 3.1; not production authorisation code."""
    known_effects = {"transactional", "compensable", "irreversible", "ddl"}
    known_traits = {"working", "shared", "customer_facing", "external", "customer_visible"}
    if effect not in known_effects or not traits or not set(traits) <= known_traits:
        raise ValueError("unclassified effect/target")
    if effect in ("transactional", "compensable") and set(traits) == {"external"}:
        raise ValueError("external alone does not establish a bounded target")
    gates = set()
    if effect == "ddl":
        gates.add("HW-ddl")
    if effect == "irreversible":
        gates.add("HW-irreversible")
    for trait, gate in (("shared", "HW-shared"), ("customer_facing", "H5"), ("customer_visible", "H5-send")):
        if trait in traits:
            gates.add(gate)
    if not gates:
        gates.add("HW-instance")
    return sorted(gates)


def recovery_action(state, landed=None):
    """Agent Runtime 7: a lost checkpoint never implies a repeat of an effect."""
    if state == "succeeded":
        return "reuse_result"
    if state == "prepared":
        return "recheck_then_dispatch"
    if state == "refused":
        return "return_refusal"
    if state in ("dispatched", "unknown", "failed"):
        return {"applied": "record_and_reuse", "not_applied": "capability_retry_rule"}.get(landed, "reconcile_or_park")
    raise ValueError("unknown journal state")


def main():
    checked, failed = 0, []

    def check(name, passed):
        nonlocal checked
        checked += 1
        if not passed:
            failed.append(name)

    manifests = [load("manifests/" + n + ".json") for n in ("support", "data", "configuration", "source")]
    owners = {}
    for doc in manifests:
        errors = manifest_errors(doc)
        check("manifest " + doc["module_id"] + ": " + "; ".join(errors), not errors)
        for profile in doc["profiles"]:
            for node in profile["owns"]:
                check("one owner for " + node, owners.get(node, profile["key"]) == profile["key"])
                owners[node] = profile["key"]
    broken = copy.deepcopy(manifests[0])
    broken["edges"].append({"from": "closure", "to": "S1", "on": "ready", "kind": "dependency"})
    check("reject implicit cycle", any("cycle" in e for e in manifest_errors(broken)))
    broken = copy.deepcopy(manifests[0])
    broken["stages"][0]["profile"] = "unregistered.profile"
    check("reject missing profile", any("profile" in e for e in manifest_errors(broken)))

    for family, validator in (("model-turn", model_errors), ("runtime-state", state_errors)):
        paths = sorted((HERE / "fixtures").glob(family + ".*.json"))
        check(family + " has valid and broken examples", len(paths) >= 2 and any(".valid." in p.name for p in paths) and any(".broken." in p.name for p in paths))
        for path in paths:
            errors = validator(json.loads(path.read_text(encoding="utf-8")))
            check(path.name, bool(errors) == (".broken." in path.name))

    check("shared PROD DDL composes all decisions", required_effect_gates("ddl", ["shared", "customer_facing"]) == ["H5", "HW-ddl", "HW-shared"])
    check("irreversible customer send composes", required_effect_gates("irreversible", ["customer_visible"]) == ["H5-send", "HW-irreversible"])
    check("working write has instance decision", required_effect_gates("transactional", ["working"]) == ["HW-instance"])
    for bad in (("read", ["working"]), ("ddl", ["guessed_target"]), ("compensable", ["external"])):
        try:
            required_effect_gates(*bad)
            check("unknown classification refused", False)
        except ValueError:
            check("unknown classification refused", True)
    for state, landed, expected in [
        ("succeeded", None, "reuse_result"),
        ("dispatched", "applied", "record_and_reuse"),
        ("dispatched", "not_applied", "capability_retry_rule"),
        ("unknown", "partly_applied", "reconcile_or_park"),
        ("unknown", "unknown", "reconcile_or_park"),
        ("failed", None, "reconcile_or_park"),
        ("prepared", None, "recheck_then_dispatch"),
    ]:
        check("recovery " + state + "/" + str(landed), recovery_action(state, landed) == expected)
    print(str(checked) + " runtime contract/conformance checks; " + str(len(failed)) + " failures")
    for name in failed:
        print("  FAIL " + name)
    return bool(failed)


if __name__ == "__main__":
    raise SystemExit(main())