135 lines
3.4 KiB
Python
135 lines
3.4 KiB
Python
"""Governance primitives for MercuryMesh MVP.
|
|
|
|
This module provides a small, production-friendly skeleton for cross-venue
|
|
governance components described in the long-term improvement plan. The goal
|
|
is to offer a minimal, well-typed representation that can be wired into the
|
|
existing FederatedCoordinator and the Graph-of-Contracts registry without
|
|
pulling in heavy dependencies.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, asdict
|
|
from typing import Dict, Any, Optional
|
|
import json
|
|
|
|
|
|
@dataclass
|
|
class FoundationModelBlock:
|
|
"""High-level block describing a foundation model dependency or contract."""
|
|
name: str
|
|
version: str
|
|
metadata: Optional[Dict[str, Any]] = None
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.to_dict())
|
|
|
|
|
|
@dataclass
|
|
class PlanningPolicyBlock:
|
|
"""Block describing planning policies (e.g., constraints, objectives)."""
|
|
policy_id: str
|
|
description: str
|
|
constraints: Dict[str, Any]
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.to_dict())
|
|
|
|
|
|
@dataclass
|
|
class PlanDeltaBlock:
|
|
"""Canonical PlanDelta representation for governance payloads."""
|
|
plan_delta: PlanDelta
|
|
note: str = "" # optional human-readable note
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {"plan_delta": self.plan_delta.to_dict(), "note": self.note}
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.to_dict())
|
|
|
|
|
|
@dataclass
|
|
class DualVariablesBlock:
|
|
"""Represents a pair of dual variables for privacy/regularization budgets."""
|
|
duals: Dict[str, float]
|
|
total_budget: float
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.to_dict())
|
|
|
|
|
|
@dataclass
|
|
class PrivacyBudgetBlock:
|
|
"""Block representing an application-level privacy/DP budget."""
|
|
budgets: Dict[str, float]
|
|
global_budget: float
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.to_dict())
|
|
|
|
|
|
@dataclass
|
|
class AuditLogBlock:
|
|
"""Tamper-evident provenance micro-log for governance actions."""
|
|
entry: str
|
|
signer: str
|
|
timestamp: str
|
|
contract_id: str
|
|
version: int
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.to_dict())
|
|
|
|
|
|
@dataclass
|
|
class GoCEntry:
|
|
"""GoC (Governance-on-Chain) compatible per-message metadata envelope."""
|
|
name: str
|
|
version: str
|
|
timestamp: str
|
|
nonce: str
|
|
signer: str
|
|
payload: Dict[str, Any]
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps(self.to_dict())
|
|
|
|
|
|
class GoCRegistry:
|
|
"""Tiny in-process registry for adapters/modules with governance metadata."""
|
|
|
|
def __init__(self) -> None:
|
|
self.adapters: Dict[str, GoCEntry] = {}
|
|
|
|
def register_adapter(self, name: str, entry: GoCEntry) -> None:
|
|
self.adapters[name] = entry
|
|
|
|
def get_adapter(self, name: str) -> Optional[GoCEntry]:
|
|
return self.adapters.get(name)
|
|
|
|
def list_adapters(self) -> Dict[str, GoCEntry]:
|
|
return dict(self.adapters)
|
|
|
|
|
|
# Lightweight default registry instance for quick MVP wiring
|
|
GoC_REGISTRY = GoCRegistry()
|