diff --git a/src/idea15_edgemind_verifiable_onboard/energi_bridge.py b/src/idea15_edgemind_verifiable_onboard/energi_bridge.py index 02b5ca3..3346aba 100644 --- a/src/idea15_edgemind_verifiable_onboard/energi_bridge.py +++ b/src/idea15_edgemind_verifiable_onboard/energi_bridge.py @@ -82,6 +82,39 @@ class EnergiBridge: def get_adapter(self, name: str) -> AdapterContract | None: return self.adapters.get(name) + def map_audit_log(self, log: AuditLog) -> Dict[str, Any]: + """Map AuditLog to a canonical representation. + + Provides tamper-evident-ish governance trail in a simple canonical form. + """ + return { + "canonical": "AuditLog", + "entries": list(log.entries), + } + + def export_envelope( + self, + lp: LocalProblem | None = None, + delta: PlanDelta | None = None, + signals: SharedSignals | None = None, + log: AuditLog | None = None, + ) -> Dict[str, Any]: + """Export a combined canonical envelope of provided pieces. + + This enables cross-adapter interoperability by bundling multiple + canonical representations into a single exchange payload. + """ + payload: Dict[str, Any] = {"canonical": "Envelope"} + if lp is not None: + payload["LocalProblem"] = self.map_local_problem(lp) + if delta is not None: + payload["PlanDelta"] = self.map_plan_delta(delta) + if signals is not None: + payload["SharedSignals"] = self.map_shared_signals(signals) + if log is not None: + payload["AuditLog"] = self.map_audit_log(log) + return payload + __all__ = [ "LocalProblem", diff --git a/tests/test_energi_bridge.py b/tests/test_energi_bridge.py index 3c59e55..9349032 100644 --- a/tests/test_energi_bridge.py +++ b/tests/test_energi_bridge.py @@ -8,6 +8,7 @@ from idea15_edgemind_verifiable_onboard.energi_bridge import ( SharedSignals, PlanDelta, AdapterContract, + AuditLog, ) @@ -42,3 +43,22 @@ def test_map_plan_delta_basic(): assert canon["timestamp"] == 123.456 assert canon["delta_actions"] == ["move_forward"] assert canon["safety_tags"] == ["ok"] + + +def test_map_audit_log_and_export_envelope(): + eb = EnergiBridge() + log = AuditLog(entries=["started", "validated", "completed"]) + audit_can = eb.map_audit_log(log) + assert audit_can["canonical"] == "AuditLog" + assert audit_can["entries"] == ["started", "validated", "completed"] + + # Envelope export with all components + lp = LocalProblem(asset_id="robot-02", tasks=["navigate"], budgets={"energy": 5.0}) + delta = PlanDelta(timestamp=2.5, delta_actions=["turn"], safety_tags=["ok"]) + signals = SharedSignals(version=2, data={"latency": 5}) + envelope = eb.export_envelope(lp=lp, delta=delta, signals=signals, log=log) + assert envelope["canonical"] == "Envelope" + assert envelope["LocalProblem"]["asset_id"] == "robot-02" + assert envelope["PlanDelta"]["timestamp"] == 2.5 + assert envelope["SharedSignals"]["version"] == 2 + assert envelope["AuditLog"]["entries"] == ["started", "validated", "completed"]