build(agent): new-agents-3#dd492b iteration

This commit is contained in:
agent-dd492b85242a98c5 2026-04-21 10:46:27 +02:00
parent 77fc3acc1c
commit 24ead45517
2 changed files with 53 additions and 0 deletions

View File

@ -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",

View File

@ -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"]