68 lines
3.0 KiB
Python
68 lines
3.0 KiB
Python
"""EnergiBridge: a minimal, production-friendly bridge for NovaPlan interoperability.
|
|
|
|
This module provides a small, production-oriented scaffold that maps NovaPlan
|
|
primitives to a CatOpt-like canonical representation and exposes a simple
|
|
Contract-as-Code (CaC) workflow for interoperability across adapters.
|
|
|
|
- to_object(local_problem) and to_morphism(plan_delta) reuse the existing
|
|
catoptBridge logic already in nova_plan.catopt_bridge.
|
|
- EnergiBridge.register_contract(contract_id, content, key) signs and stores a
|
|
contract artifact in the in-repo CaC registry, enabling provenance for adapters.
|
|
|
|
The implementation is intentionally small and non-breaking so existing tests
|
|
continue to pass. It serves as a deliberate MVP wiring point for future
|
|
feature growth.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, cast
|
|
|
|
from nova_plan.catopt_bridge import to_object, delta_to_morphism
|
|
from nova_plan.contracts import PlanDelta, CaCContract, CaCRegistry, sign_ca_contract, GoCRegistry
|
|
|
|
|
|
class EnergiBridge:
|
|
"""Canonical EnergiBridge facade for cross-domain interoperability.
|
|
|
|
Provides helpers to map core primitives to canonical representations and
|
|
to publish a signed contract artifact for governance and replay-safe
|
|
interoperability.
|
|
"""
|
|
|
|
def __init__(self, signer_key: str | None = None) -> None:
|
|
self.signer_key = signer_key
|
|
|
|
# Canonical mapping helpers (thin wrappers around existing bridge logic)
|
|
def to_object(self, local_problem) -> object:
|
|
"""Convert a LocalProblem to a canonical ObjectI via existing bridge."""
|
|
return to_object(local_problem)
|
|
|
|
def to_morphism(self, plan_delta: PlanDelta, contract_id: str = "default") -> object:
|
|
"""Convert a PlanDelta to a canonical Morphism, with optional contract_id."""
|
|
return delta_to_morphism(plan_delta, contract_id=contract_id)
|
|
|
|
# CaC publishing interface
|
|
def register_contract(self, contract_id: str, content: Dict[str, Any], key: str) -> CaCContract:
|
|
"""Sign and register a contract artifact in the CaC registry.
|
|
|
|
- contract_id: unique id for the contract artifact
|
|
- content: the contract content payload (big or small, domain-specific)
|
|
- key: signing key (private material); used to derive a signature deterministically
|
|
Returns the signed CaCContract object and registers it in CaCRegistry.
|
|
"""
|
|
contract = CaCContract(contract_id=contract_id, version=1, content=content)
|
|
signed = sign_ca_contract(contract, key)
|
|
# Persist in CaC registry (store the unsigned contract metadata for MVP)
|
|
CaCRegistry.register(contract)
|
|
# Also publish provenance to the GoC registry for cross-adapter provenance tracking
|
|
try:
|
|
GoCRegistry.register_signed_contract(cast(CaCContract, contract), signer="EnergiBridge", signature=signed.signature)
|
|
except Exception:
|
|
# Best-effort provenance extension; do not fail publishing if registry is not ready
|
|
pass
|
|
return contract
|
|
|
|
|
|
__all__ = ["EnergiBridge"]
|