acdl/acdl_platform/contract_resolver.py

"""ACDL Contract Resolver — resolve a contract to a Target Stack IR instance.

ARCHITECTURE.md §12.8: the contract declares intent in IR-typed terms;
the resolver resolves the contract to a target stack (list of L1
instances + inputs + relationships); the adapter compiles the target
stack to a plan.

Steps:
  1. Load the contract (YAML -> dict).
  2. Validate the contract against schemas/contract.schema.json.
  3. Look up the L2 in modules-ir/registry.json.
  4. Load the L2's composition.json (the thin-composition tree).
  5. Map the contract's inputs through the composition's wires to the
     child L1's inputs.
  6. Emit an IR instance {version, stack:{name, kind:l2, depth},
     resources:[<L1 instances with concrete inputs>], relationships:[...]}.
  7. Validate the IR instance against schemas/ir.schema.json.

CLI: contract_resolver.py <contract.yaml> <out_ir.json>
"""

import json
import os
import sys

import yaml
import jsonschema


REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def _load_json(path):
    with open(path, "r") as fh:
        return json.load(fh)


def resolve(contract_path, repo_root=None):
    """Resolve a contract YAML to an IR instance dict."""
    rr = repo_root or REPO_ROOT

    # 1. Load the contract YAML.
    with open(contract_path, "r") as fh:
        contract = yaml.safe_load(fh)

    # 2. Validate the contract against the contract schema.
    contract_schema = _load_json(os.path.join(rr, "schemas/contract.schema.json"))
    jsonschema.validate(contract, contract_schema)

    # 3. Look up the L2 in the registry.
    stack_name = contract["stack"]
    registry = _load_json(os.path.join(rr, "modules-ir/registry.json"))
    if stack_name not in registry:
        raise ValueError(f"stack {stack_name!r} not in registry")
    versions = registry[stack_name]
    # Pick the highest 1.x.x (spike: just take the first non-deprecated).
    entry = next(v for v in versions.values() if not v.get("deprecated", False))

    # 4. Load the L2's composition.json.
    composition_key = entry.get("composition") or entry.get("interface")
    composition = _load_json(os.path.join(rr, composition_key))

    # 5. Map the contract's inputs through the wires to the child L1's inputs.
    wires = composition.get("wires", {})
    contract_inputs = contract.get("inputs", {})
    children = composition.get("children", [])

    resources = []
    relationships = []
    for child in children:
        child_id = child["id"]
        child_module = child["module"]  # e.g. l1-s3@1.0.0
        # Map inputs via wires whose target is this child.
        child_inputs = {}
        for wire_name, wire in wires.items():
            if wire.get("target") == child_id and wire_name in contract_inputs:
                child_inputs[wire["input"]] = contract_inputs[wire_name]
        # Load the L1 interface to get the IR type + outputs.
        l1_name, l1_version = child_module.split("@", 1)
        l1_entry = registry.get(l1_name, {}).get(l1_version)
        if not l1_entry:
            raise ValueError(f"L1 {child_module!r} not in registry")
        l1_iface = _load_json(os.path.join(rr, l1_entry["interface"]))
        resources.append({
            "id": child_id,
            "type": l1_iface["type"],
            "module": child_module,
            "inputs": child_inputs,
            "outputs": l1_iface.get("outputs", {}),
        })
        relationships.append({"from": "root", "to": child_id, "kind": "parent"})

    # 6. Emit the IR instance.
    ir_instance = {
        "version": "1.0.0",
        "stack": {
            "name": composition["name"],
            "kind": composition["kind"],
            "depth": composition["depth"],
        },
        "resources": resources,
        "relationships": relationships,
    }

    # 7. Validate the IR instance against the IR schema.
    ir_schema = _load_json(os.path.join(rr, "schemas/ir.schema.json"))
    jsonschema.validate(ir_instance, ir_schema)

    return ir_instance


if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("usage: contract_resolver.py <contract.yaml> <out_ir.json>", file=sys.stderr)
        sys.exit(2)
    ir = resolve(sys.argv[1])
    with open(sys.argv[2], "w") as fh:
        json.dump(ir, fh, indent=2)
    print(f"resolver: emitted IR to {sys.argv[2]}", file=sys.stderr)