build(agent): molt-az#4b796a iteration

This commit is contained in:
agent-4b796a86eacc591f 2026-04-16 23:00:27 +02:00
parent ec9814cd33
commit 99cd3b5bc9
3 changed files with 58 additions and 1 deletions

View File

@ -13,7 +13,7 @@ from .marketplace import AdaptersMarketplace
from .simulation import SimulationHarness from .simulation import SimulationHarness
from .privacy import SecureAggregator, PrivacyBudget from .privacy import SecureAggregator, PrivacyBudget
from .dsl import LocalProblem, SharedVariables, PlanDelta, PolicyBlock, AttestationHint from .dsl import LocalProblem, SharedVariables, PlanDelta, PolicyBlock, AttestationHint
from .bridge import to_canonical, from_canonical from .bridge import to_canonical, from_canonical, EnergiBridge
__all__ = [ __all__ = [
"SecurityContractsRegistry", "SecurityContractsRegistry",
@ -33,4 +33,5 @@ __all__ = [
"AttestationHint", "AttestationHint",
"to_canonical", "to_canonical",
"from_canonical", "from_canonical",
"EnergiBridge",
] ]

View File

@ -12,10 +12,16 @@ class AttestedAgent:
self.agent_id = agent_id self.agent_id = agent_id
self.hardware = hardware self.hardware = hardware
self._credential: Optional[str] = None self._credential: Optional[str] = None
# Lightweight DID-style identity; in a real system this would be bound to
# a hardware-backed identity and rotated periodically.
self._did: Optional[str] = None
def attest(self) -> bool: def attest(self) -> bool:
# In a real system, remote attestation would happen here. # In a real system, remote attestation would happen here.
self._credential = f"attest-{self.agent_id}-{self.hardware}-v1" self._credential = f"attest-{self.agent_id}-{self.hardware}-v1"
# Issue or refresh a minimal DID identity alongside attestation
if self._did is None:
self._did = f"did:gridguard:{self.agent_id}"
return True return True
@property @property
@ -31,3 +37,28 @@ class AttestedAgent:
""" """
expected = f"attest-{self.agent_id}-{self.hardware}-v1" expected = f"attest-{self.agent_id}-{self.hardware}-v1"
return credential == expected return credential == expected
@property
def did(self) -> str:
"""Return the agent's Decentralized Identifier (DID).
If not yet issued, lazily generate a simple DID. This is a lightweight
stand-in for DID/identity binding in MVP contexts.
"""
if self._did is None:
self._did = f"did:gridguard:{self.agent_id}"
return self._did
def rotate_identity(self) -> None:
"""Rotate the agent's identity to simulate short-lived credentials.
In a production system this would refresh attestation keys and rotate
credentials. Here we simply mutate the DID suffix to reflect rotation.
"""
suffix = getattr(self, "_did", None)
if suffix is None:
self._did = f"did:gridguard:{self.agent_id}"
else:
# Simple rotation by appending a timestamp-like suffix
import time
self._did = f"did:gridguard:{self.agent_id}:{int(time.time())}"

View File

@ -37,3 +37,28 @@ class EnergiBridge:
@staticmethod @staticmethod
def from_canonical(canonical: Dict[str, Any]) -> Dict[str, Any]: def from_canonical(canonical: Dict[str, Any]) -> Dict[str, Any]:
return from_canonical(canonical) return from_canonical(canonical)
@staticmethod
def to_canonical_with_contract(local_problem: Dict[str, Any], version: str = "1.0", contract_id: str = "default") -> Dict[str, Any]:
"""Extended canonical representation including simple contract metadata.
This enhances the MVP bridge with a minimal Graph-of-Contracts hint to
help adapters interoperate in a CatOpt-like canonical form while keeping
the representation lightweight.
"""
base = EnergiBridge.to_canonical(local_problem)
# Attach a tiny contract metadata block to aid auditing and routing
base_metadata = {
"GraphOfContracts": [
{
"contract_id": contract_id,
"version": version,
"bindings": ["LocalProblems"],
"notes": "mvp metadata" ,
}
],
"ContractsVersion": version,
}
# Merge metadata into the canonical payload under a consistent key
base.update(base_metadata)
return base