nova-platform/core/contract_resolver.py

"""Nova Platform — Contract Resolver.

Resolves a validated consumer contract to a Stack instance (a flat dict
conforming to schemas/stack.schema.json).

Flow:
  1. Validate the contract dict against schemas/contract.schema.json.
  2. Load the environment via core.environment_check.check().
  3. Build an interpolation context {'env': env, 'contract': contract}.
  4. For each infrastructure entry: look up the module + version in the
     registry, interpolate ${env.*} / ${contract.*} tokens in inputs,
     and emit a flat stack resource {module, version, inputs}.
  5. Return the stack dict.

Engine-agnostic: no aws_*, no Terraform terms, no module paths. The stack
carries NO 'source' field (D-037/C-1 grill fix) — the adapter loads the
registry to map module -> terraform_dir. L2 is opaque (D-012): a single
stack resource, no children/wires expansion.
"""

import json
import re
from pathlib import Path

import jsonschema
import yaml

_TOKEN_RE = re.compile(r"\$\{([a-zA-Z_][a-zA-Z0-9_.]*)\}")


class ModuleNotFoundError(KeyError):
    """Raised when a contract references a module not in the registry."""


class VersionNotFoundError(KeyError):
    """Raised when a contract references a version not in the registry."""


def _lookup_dotted(context, dotted):
    parts = dotted.split(".")
    cur = context
    for part in parts:
        if isinstance(cur, dict) and part in cur:
            cur = cur[part]
        else:
            raise KeyError(dotted)
    return cur


def _expand_vars(value, context):
    if isinstance(value, str):
        def _replace(match):
            token = match.group(1)
            try:
                resolved = _lookup_dotted(context, token)
            except KeyError:
                raise ValueError(f"unresolved interpolation token: ${{{token}}}")
            if isinstance(resolved, (dict, list)):
                return json.dumps(resolved)
            return str(resolved)
        return _TOKEN_RE.sub(_replace, value)
    if isinstance(value, dict):
        return {k: _expand_vars(v, context) for k, v in value.items()}
    if isinstance(value, list):
        return [_expand_vars(v, context) for v in value]
    return value


def _latest_version(registry, module_name):
    versions = registry[module_name]
    non_deprecated = [(v, e) for v, e in versions.items()
                      if not e.get("deprecated", False)]
    if not non_deprecated:
        non_deprecated = list(versions.items())
    non_deprecated.sort(key=lambda x: [int(p) for p in x[0].split(".")],
                        reverse=True)
    return non_deprecated[0][0]


def _load_schema(path):
    with open(path) as fh:
        return json.load(fh)


def resolve(contract, registry, modules_dir, environments_dir=None,
            repo_root=None):
    """Resolve a validated contract dict to a flat Stack dict.

    Args:
        contract: validated contract dict (must conform to
            schemas/contract.schema.json).
        registry: modules/registry.json loaded as a dict.
        modules_dir: Path to the modules/ directory (unused for L2-opaque
            resolution but kept per D-011 for future interface.json reads).
        environments_dir: Path to core/environments/. If None, derived from
            repo_root / 'core' / 'environments'.
        repo_root: Path to the repo root. If None, derived from modules_dir
            parent's parent (modules_dir is <root>/modules).

    Returns:
        A flat stack dict conforming to schemas/stack.schema.json:
        {contract_id, contract_name, environment, resources: [{module,
        version, inputs}]}.

    Raises:
        ModuleNotFoundError: contract references an unknown module.
        VersionNotFoundError: contract references an unknown version.
        jsonschema.ValidationError: contract does not conform to schema.
        ValueError: unresolved interpolation token.
    """
    if repo_root is None:
        repo_root = Path(modules_dir).parent.parent
    if environments_dir is None:
        environments_dir = Path(repo_root) / "core" / "environments"

    contract_schema_path = Path(repo_root) / "schemas" / "contract.schema.json"
    contract_schema = _load_schema(contract_schema_path)
    jsonschema.validate(contract, contract_schema)

    from core import environment_check
    env = environment_check.check(contract["environment"], environments_dir)
    # Expose 'environment' as an alias for the env's 'name' field so
    # ${env.environment} resolves (the env JSON uses 'name', but contracts
    # reference the environment by ${env.environment}).
    env["environment"] = env.get("name", contract["environment"])

    context = {"env": env, "contract": contract}

    resources = []
    for item in contract["infrastructure"]:
        module_name = item["module"]
        if module_name not in registry:
            raise ModuleNotFoundError(module_name)
        version = item.get("version")
        if version is None:
            version = _latest_version(registry, module_name)
        elif version not in registry[module_name]:
            raise VersionNotFoundError(f"{module_name}@{version}")
        inputs = _expand_vars(item.get("inputs", {}), context)
        resources.append({
            "module": module_name,
            "version": version,
            "inputs": inputs,
        })

    return {
        "contract_id": contract["id"],
        "contract_name": contract["name"],
        "environment": contract["environment"],
        "resources": resources,
    }


def main(argv=None):
    import sys
    sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
    argv = argv or sys.argv[1:]
    if len(argv) < 2:
        print("usage: contract_resolver.py <contract.yaml> [out.json]",
              file=sys.stderr)
        return 2
    contract_path = argv[0]
    out_path = argv[1] if len(argv) > 1 else None
    repo_root = Path(__file__).resolve().parent.parent
    with open(contract_path) as fh:
        contract = yaml.safe_load(fh)
    with open(repo_root / "modules" / "registry.json") as fh:
        registry = json.load(fh)
    stack = resolve(contract, registry, repo_root / "modules",
                    repo_root=repo_root)
    if out_path:
        with open(out_path, "w") as fh:
            json.dump(stack, fh, indent=2)
    else:
        print(json.dumps(stack, indent=2))
    return 0


if __name__ == "__main__":
    import sys
    sys.exit(main())