From 434601d06d8da65bdc226612c1c1db7077f4ee74 Mon Sep 17 00:00:00 2001 From: agent-54de0bcc6a17828b Date: Fri, 24 Apr 2026 18:39:37 +0200 Subject: [PATCH] build(agent): semicolon#54de0b iteration --- .gitignore | 21 +++++ AGENTS.md | 25 ++++++ README.md | 59 +++++++++++++- pyproject.toml | 34 ++++++++ src/solfuse/__init__.py | 39 ++++++++++ src/solfuse/cli.py | 24 ++++++ src/solfuse/delta_sync.py | 61 +++++++++++++++ src/solfuse/governance.py | 124 +++++++++++++++++++++++++++++ src/solfuse/identity.py | 44 +++++++++++ src/solfuse/models.py | 132 +++++++++++++++++++++++++++++++ src/solfuse/registry.py | 160 ++++++++++++++++++++++++++++++++++++++ src/solfuse/solver.py | 82 +++++++++++++++++++ src/solfuse/transport.py | 31 ++++++++ test.sh | 6 ++ tests/test_solfuse.py | 98 +++++++++++++++++++++++ 15 files changed, 938 insertions(+), 2 deletions(-) create mode 100644 .gitignore create mode 100644 AGENTS.md create mode 100644 pyproject.toml create mode 100644 src/solfuse/__init__.py create mode 100644 src/solfuse/cli.py create mode 100644 src/solfuse/delta_sync.py create mode 100644 src/solfuse/governance.py create mode 100644 src/solfuse/identity.py create mode 100644 src/solfuse/models.py create mode 100644 src/solfuse/registry.py create mode 100644 src/solfuse/solver.py create mode 100644 src/solfuse/transport.py create mode 100644 test.sh create mode 100644 tests/test_solfuse.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bd5590b --- /dev/null +++ b/.gitignore @@ -0,0 +1,21 @@ +node_modules/ +.npmrc +.env +.env.* +__tests__/ +coverage/ +.nyc_output/ +dist/ +build/ +.cache/ +*.log +.DS_Store +tmp/ +.tmp/ +__pycache__/ +*.pyc +.venv/ +venv/ +*.egg-info/ +.pytest_cache/ +READY_TO_PUBLISH diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..ad5e0cb --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,25 @@ +# SolFuse Repository Rules + +## Architecture +- Language: Python 3.11+ +- Layout: `src/solfuse` package with `pytest` tests in `tests/` +- Core modules: + - `models.py`: canonical protocol primitives + - `identity.py`: DID-style identities and signed envelopes + - `registry.py`: graph-of-contracts registry backed by SQLite + - `solver.py`: ADMM-lite federated planner + - `governance.py`: audit ledger and privacy-budget tracking + - `delta_sync.py`: islanding-friendly incremental plan journal + - `transport.py`: canonical serialization and TLS context helpers + +## Testing +- Install editable dependencies: `python3 -m pip install -e .[dev]` +- Run tests: `pytest` +- Build package: `python3 -m build` +- Full validation: `bash test.sh` + +## Contribution Rules +- Keep changes minimal and production-oriented. +- Prefer validated models and deterministic behavior. +- Do not add new dependencies unless they are used directly. +- Keep public APIs documented in `README.md`. diff --git a/README.md b/README.md index dd0356a..8dc3b40 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,58 @@ -# solfuse-privacy-preserving-federated-orc +# SolFuse -Gap: A missing, practical platform to federate optimization across microgrids that combine solar, wind, and thermal storage with cross-utility assets, while preserving data locality, enabling offline islanding, and providing auditable governance. Sol \ No newline at end of file +SolFuse is a privacy-preserving federation layer for hybrid microgrids. +It coordinates solar, wind, and thermal storage sites through summarized signals, incremental plan deltas, and auditable governance instead of raw data exchange. + +## What It Provides + +- Canonical protocol primitives for local optimization, shared signals, dual variables, and plan deltas +- A SQLite-backed graph-of-contracts registry for adapter versioning and schema conformance +- An ADMM-lite solver for federated dispatch planning +- Signed message envelopes and DID-style identities for short-lived trust +- Audit logging and privacy-budget accounting +- Delta-sync journaling for islanded/offline operation + +## Package Layout + +- `solfuse.models`: protocol objects and canonical payload schemas +- `solfuse.identity`: identity generation, signing, and verification +- `solfuse.registry`: adapter registry and schema conformance checks +- `solfuse.solver`: federated consensus solver +- `solfuse.delta_sync`: incremental plan journal and replay helpers +- `solfuse.governance`: audit log and privacy budget ledger +- `solfuse.transport`: canonical envelope serialization and TLS context helpers +- `solfuse.cli`: small demo entry point + +## Quick Start + +```bash +python3 -m pip install -e .[dev] +pytest +solfuse demo +``` + +## Minimal Example + +```python +from solfuse.models import LocalProblem +from solfuse.solver import ADMMLiteSolver + +solver = ADMMLiteSolver(rho=1.0, tolerance=1e-6, max_iterations=100) +result = solver.solve([ + LocalProblem(site_id="solar-a", preferred_dispatch=[4.0, 4.5], quadratic_weight=2.0), + LocalProblem(site_id="storage-b", preferred_dispatch=[1.0, 1.5], quadratic_weight=1.5), +]) + +print(result.consensus) +``` + +## Governance Model + +Each message can be wrapped in a signed envelope, recorded in the audit ledger, and charged against a privacy budget. +Contract adapters are versioned and validated against JSON Schema so controllers can be swapped without changing the federation core. + +## Testing And Build + +- `bash test.sh` +- `pytest` +- `python3 -m build` diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..a08544d --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,34 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "solfuse" +version = "0.1.0" +description = "Privacy-preserving federated orchestration for hybrid energy microgrids" +readme = "README.md" +requires-python = ">=3.11" +license = "MIT" +authors = [{name = "OpenCode Swarm"}] +dependencies = [ + "cryptography>=43.0.0", + "jsonschema>=4.23.0", + "packaging>=24.0", + "numpy>=1.26.4", + "pydantic>=2.8.2", +] + +[project.optional-dependencies] +dev = [ + "build>=1.2.2", + "pytest>=8.3.2", +] + +[project.scripts] +solfuse = "solfuse.cli:main" + +[tool.setuptools] +package-dir = {"" = "src"} + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/src/solfuse/__init__.py b/src/solfuse/__init__.py new file mode 100644 index 0000000..0eb5bc1 --- /dev/null +++ b/src/solfuse/__init__.py @@ -0,0 +1,39 @@ +"""SolFuse canonical federation primitives.""" + +from .delta_sync import DeltaSyncJournal +from .governance import AuditLedger, PrivacyBudgetLedger +from .identity import AgentIdentity, generate_identity, sign_message, verify_message +from .models import ( + AdapterContract, + AuditLogEntry, + DualVariable, + LocalProblem, + PlanAction, + PlanDelta, + PrivacyBudget, + SharedVariable, + SolverResult, +) +from .registry import GraphOfContractsRegistry +from .solver import ADMMLiteSolver + +__all__ = [ + "ADMMLiteSolver", + "AdapterContract", + "AgentIdentity", + "AuditLedger", + "AuditLogEntry", + "DeltaSyncJournal", + "DualVariable", + "GraphOfContractsRegistry", + "LocalProblem", + "PlanAction", + "PlanDelta", + "PrivacyBudget", + "PrivacyBudgetLedger", + "SharedVariable", + "SolverResult", + "generate_identity", + "sign_message", + "verify_message", +] diff --git a/src/solfuse/cli.py b/src/solfuse/cli.py new file mode 100644 index 0000000..463c7b2 --- /dev/null +++ b/src/solfuse/cli.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +import argparse +import json + +from .models import LocalProblem +from .solver import ADMMLiteSolver + + +def main() -> None: + parser = argparse.ArgumentParser(prog="solfuse") + subcommands = parser.add_subparsers(dest="command", required=True) + subcommands.add_parser("demo") + args = parser.parse_args() + + if args.command == "demo": + solver = ADMMLiteSolver() + result = solver.solve( + [ + LocalProblem(site_id="solar-a", preferred_dispatch=[4.0, 4.5], quadratic_weight=2.0), + LocalProblem(site_id="storage-b", preferred_dispatch=[1.0, 1.5], quadratic_weight=1.5), + ] + ) + print(json.dumps(result.model_dump(mode="json"), indent=2)) diff --git a/src/solfuse/delta_sync.py b/src/solfuse/delta_sync.py new file mode 100644 index 0000000..8529813 --- /dev/null +++ b/src/solfuse/delta_sync.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path +from typing import Iterable + +from .models import PlanDelta + + +class DeltaSyncJournal: + def __init__(self, path: str | Path = ":memory:") -> None: + self.path = str(path) + self._conn = sqlite3.connect(self.path) + self._conn.row_factory = sqlite3.Row + self._ensure_schema() + + def _ensure_schema(self) -> None: + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS plan_deltas ( + delta_id TEXT PRIMARY KEY, + site_id TEXT NOT NULL, + adapter_id TEXT NOT NULL, + revision INTEGER NOT NULL, + round_id INTEGER NOT NULL, + parent_revision INTEGER, + payload TEXT NOT NULL + ) + """ + ) + self._conn.commit() + + def append(self, delta: PlanDelta) -> None: + payload = json.dumps(delta.model_dump(mode="json"), sort_keys=True) + self._conn.execute( + """ + INSERT OR REPLACE INTO plan_deltas + (delta_id, site_id, adapter_id, revision, round_id, parent_revision, payload) + VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + delta.delta_id, + delta.site_id, + delta.adapter_id, + delta.revision, + delta.round_id, + delta.parent_revision, + payload, + ), + ) + self._conn.commit() + + def load(self) -> list[PlanDelta]: + rows = self._conn.execute("SELECT payload FROM plan_deltas ORDER BY revision, delta_id").fetchall() + return [PlanDelta.model_validate(json.loads(row["payload"])) for row in rows] + + def merge(self, deltas: Iterable[PlanDelta]) -> list[PlanDelta]: + for delta in deltas: + self.append(delta) + return self.load() diff --git a/src/solfuse/governance.py b/src/solfuse/governance.py new file mode 100644 index 0000000..bd69ff2 --- /dev/null +++ b/src/solfuse/governance.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path +from typing import Iterable + +from .models import AuditLogEntry, PrivacyBudget + + +class PrivacyBudgetLedger: + def __init__(self, path: str | Path = ":memory:") -> None: + self.path = str(path) + self._conn = sqlite3.connect(self.path) + self._conn.row_factory = sqlite3.Row + self._ensure_schema() + + def _ensure_schema(self) -> None: + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS privacy_budgets ( + actor_id TEXT PRIMARY KEY, + epsilon REAL NOT NULL, + delta REAL NOT NULL, + spent_epsilon REAL NOT NULL, + spent_delta REAL NOT NULL + ) + """ + ) + self._conn.commit() + + def set_budget(self, actor_id: str, budget: PrivacyBudget) -> None: + self._conn.execute( + "REPLACE INTO privacy_budgets VALUES (?, ?, ?, ?, ?)", + (actor_id, budget.epsilon, budget.delta, budget.spent_epsilon, budget.spent_delta), + ) + self._conn.commit() + + def charge(self, actor_id: str, epsilon: float, delta: float) -> PrivacyBudget: + current = self.get_budget(actor_id) + if current is None: + raise KeyError(actor_id) + updated = current.consume(epsilon, delta) + self.set_budget(actor_id, updated) + return updated + + def get_budget(self, actor_id: str) -> PrivacyBudget | None: + row = self._conn.execute("SELECT * FROM privacy_budgets WHERE actor_id = ?", (actor_id,)).fetchone() + if row is None: + return None + return PrivacyBudget( + epsilon=row["epsilon"], + delta=row["delta"], + spent_epsilon=row["spent_epsilon"], + spent_delta=row["spent_delta"], + ) + + +class AuditLedger: + def __init__(self, path: str | Path = ":memory:") -> None: + self.path = str(path) + self._conn = sqlite3.connect(self.path) + self._conn.row_factory = sqlite3.Row + self._ensure_schema() + + def _ensure_schema(self) -> None: + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS audit_log ( + event_id TEXT PRIMARY KEY, + actor_id TEXT NOT NULL, + event_type TEXT NOT NULL, + payload TEXT NOT NULL, + created_at TEXT NOT NULL, + privacy_budget TEXT, + signature TEXT + ) + """ + ) + self._conn.commit() + + def append(self, entry: AuditLogEntry) -> None: + self._conn.execute( + """ + INSERT OR REPLACE INTO audit_log + (event_id, actor_id, event_type, payload, created_at, privacy_budget, signature) + VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + entry.event_id, + entry.actor_id, + entry.event_type, + json.dumps(entry.payload, sort_keys=True), + entry.created_at.isoformat(), + None if entry.privacy_budget is None else entry.privacy_budget.model_dump_json(), + entry.signature, + ), + ) + self._conn.commit() + + def recent(self, limit: int = 20) -> list[AuditLogEntry]: + rows = self._conn.execute("SELECT * FROM audit_log ORDER BY created_at DESC LIMIT ?", (limit,)).fetchall() + entries: list[AuditLogEntry] = [] + for row in rows: + entries.append( + AuditLogEntry( + event_id=row["event_id"], + actor_id=row["actor_id"], + event_type=row["event_type"], + payload=json.loads(row["payload"]), + created_at=row["created_at"], + privacy_budget=None if row["privacy_budget"] is None else PrivacyBudget.model_validate_json(row["privacy_budget"]), + signature=row["signature"], + ) + ) + return entries + + def coverage(self) -> float: + total = self._conn.execute("SELECT COUNT(*) AS count FROM audit_log").fetchone()["count"] + signed = self._conn.execute("SELECT COUNT(*) AS count FROM audit_log WHERE signature IS NOT NULL").fetchone()["count"] + return 0.0 if total == 0 else signed / total + + def export(self) -> list[dict[str, object]]: + return [entry.model_dump(mode="json") for entry in self.recent(limit=10_000)] diff --git a/src/solfuse/identity.py b/src/solfuse/identity.py new file mode 100644 index 0000000..0359daa --- /dev/null +++ b/src/solfuse/identity.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from dataclasses import dataclass +from hashlib import sha256 +import base64 +import json +from typing import Any + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey +from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat + + +def _canonical_json(payload: Any) -> bytes: + return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") + + +@dataclass(frozen=True) +class AgentIdentity: + did: str + private_key: Ed25519PrivateKey + + @property + def public_key_bytes(self) -> bytes: + return self.private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw) + + +def generate_identity(label: str) -> AgentIdentity: + private_key = Ed25519PrivateKey.generate() + digest = sha256(label.encode("utf-8") + private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)).hexdigest() + return AgentIdentity(did=f"did:solfuse:{digest}", private_key=private_key) + + +def sign_message(identity: AgentIdentity, payload: Any) -> str: + signature = identity.private_key.sign(_canonical_json(payload)) + return base64.urlsafe_b64encode(signature).decode("ascii") + + +def verify_message(public_key: Ed25519PublicKey | bytes, payload: Any, signature: str) -> bool: + key = public_key if isinstance(public_key, Ed25519PublicKey) else Ed25519PublicKey.from_public_bytes(public_key) + try: + key.verify(base64.urlsafe_b64decode(signature.encode("ascii")), _canonical_json(payload)) + return True + except Exception: + return False diff --git a/src/solfuse/models.py b/src/solfuse/models.py new file mode 100644 index 0000000..f6ce9ce --- /dev/null +++ b/src/solfuse/models.py @@ -0,0 +1,132 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from hashlib import sha256 +from typing import Any +from uuid import uuid4 + +from pydantic import BaseModel, ConfigDict, Field, field_validator + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +class SolfuseModel(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +class LocalProblem(SolfuseModel): + site_id: str + preferred_dispatch: list[float] + quadratic_weight: float = Field(gt=0) + lower_bounds: list[float] | None = None + upper_bounds: list[float] | None = None + round_id: int = Field(default=0, ge=0) + metadata: dict[str, Any] = Field(default_factory=dict) + + @field_validator("preferred_dispatch") + @classmethod + def _dispatch_non_empty(cls, value: list[float]) -> list[float]: + if not value: + raise ValueError("preferred_dispatch must not be empty") + return value + + @field_validator("lower_bounds", "upper_bounds") + @classmethod + def _bounds_match_dispatch(cls, value: list[float] | None, info) -> list[float] | None: + if value is None: + return value + if "preferred_dispatch" in info.data and len(value) != len(info.data["preferred_dispatch"]): + raise ValueError("bounds must match preferred_dispatch length") + return value + + +class SharedVariable(SolfuseModel): + name: str + version: str = "1.0" + values: list[float] + round_id: int = Field(default=0, ge=0) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class DualVariable(SolfuseModel): + name: str + version: str = "1.0" + values: list[float] + round_id: int = Field(default=0, ge=0) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class PlanAction(SolfuseModel): + resource_id: str + action: str + value: float + unit: str = "kW" + metadata: dict[str, Any] = Field(default_factory=dict) + + +class PlanDelta(SolfuseModel): + delta_id: str = Field(default_factory=lambda: uuid4().hex) + site_id: str + adapter_id: str + revision: int = Field(ge=0) + round_id: int = Field(ge=0) + parent_revision: int | None = Field(default=None, ge=0) + actions: list[PlanAction] + metadata: dict[str, Any] = Field(default_factory=dict) + + @field_validator("actions") + @classmethod + def _actions_non_empty(cls, value: list[PlanAction]) -> list[PlanAction]: + if not value: + raise ValueError("actions must not be empty") + return value + + +class PrivacyBudget(SolfuseModel): + epsilon: float = Field(ge=0) + delta: float = Field(ge=0) + spent_epsilon: float = Field(default=0, ge=0) + spent_delta: float = Field(default=0, ge=0) + + def consume(self, epsilon: float, delta: float) -> "PrivacyBudget": + new_epsilon = self.spent_epsilon + epsilon + new_delta = self.spent_delta + delta + if new_epsilon > self.epsilon or new_delta > self.delta: + raise ValueError("privacy budget exceeded") + return self.model_copy(update={"spent_epsilon": new_epsilon, "spent_delta": new_delta}) + + +class AuditLogEntry(SolfuseModel): + event_id: str = Field(default_factory=lambda: uuid4().hex) + actor_id: str + event_type: str + payload: dict[str, Any] + created_at: datetime = Field(default_factory=_utcnow) + privacy_budget: PrivacyBudget | None = None + signature: str | None = None + + +class AdapterContract(SolfuseModel): + name: str + version: str + adapter_type: str + input_schema: dict[str, Any] + output_schema: dict[str, Any] + capabilities: list[str] = Field(default_factory=list) + metadata: dict[str, Any] = Field(default_factory=dict) + + @property + def fingerprint(self) -> str: + material = self.model_dump(mode="json", exclude={"metadata"}) + return sha256(str(material).encode("utf-8")).hexdigest() + + +class SolverResult(SolfuseModel): + consensus: list[float] + iterations: int + primal_residual: float + dual_residual: float + deltas: list[PlanDelta] = Field(default_factory=list) + metadata: dict[str, Any] = Field(default_factory=dict) diff --git a/src/solfuse/registry.py b/src/solfuse/registry.py new file mode 100644 index 0000000..bb168d7 --- /dev/null +++ b/src/solfuse/registry.py @@ -0,0 +1,160 @@ +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path +from typing import Any + +from jsonschema import ValidationError, validate +from packaging.version import InvalidVersion, Version + +from .models import AdapterContract + + +class GraphOfContractsRegistry: + def __init__(self, path: str | Path = ":memory:") -> None: + self.path = str(path) + self._conn = sqlite3.connect(self.path) + self._conn.row_factory = sqlite3.Row + self._ensure_schema() + + def _ensure_schema(self) -> None: + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS contracts ( + name TEXT NOT NULL, + version TEXT NOT NULL, + adapter_type TEXT NOT NULL, + input_schema TEXT NOT NULL, + output_schema TEXT NOT NULL, + capabilities TEXT NOT NULL, + metadata TEXT NOT NULL, + fingerprint TEXT NOT NULL, + PRIMARY KEY (name, version) + ) + """ + ) + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS edges ( + source_name TEXT NOT NULL, + source_version TEXT NOT NULL, + target_name TEXT NOT NULL, + target_version TEXT NOT NULL, + relation TEXT NOT NULL, + PRIMARY KEY (source_name, source_version, target_name, target_version, relation) + ) + """ + ) + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS conformance_checks ( + contract_name TEXT NOT NULL, + contract_version TEXT NOT NULL, + payload_kind TEXT NOT NULL, + passed INTEGER NOT NULL, + error TEXT, + payload TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP + ) + """ + ) + self._conn.commit() + + def register(self, contract: AdapterContract) -> AdapterContract: + self._conn.execute( + """ + INSERT OR REPLACE INTO contracts + (name, version, adapter_type, input_schema, output_schema, capabilities, metadata, fingerprint) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + contract.name, + contract.version, + contract.adapter_type, + json.dumps(contract.input_schema, sort_keys=True), + json.dumps(contract.output_schema, sort_keys=True), + json.dumps(contract.capabilities, sort_keys=True), + json.dumps(contract.metadata, sort_keys=True), + contract.fingerprint, + ), + ) + self._conn.commit() + return contract + + def link(self, source: tuple[str, str], target: tuple[str, str], relation: str = "compatible_with") -> None: + self._conn.execute( + "INSERT OR REPLACE INTO edges VALUES (?, ?, ?, ?, ?)", + (source[0], source[1], target[0], target[1], relation), + ) + self._conn.commit() + + def latest(self, name: str) -> AdapterContract: + versions = self.versions(name) + if not versions: + raise KeyError(name) + latest_version = versions[-1] + return self.get(name, latest_version) + + def get(self, name: str, version: str) -> AdapterContract: + row = self._conn.execute( + "SELECT * FROM contracts WHERE name = ? AND version = ?", + (name, version), + ).fetchone() + if row is None: + raise KeyError(f"{name}@{version}") + return self._row_to_contract(row) + + def versions(self, name: str) -> list[str]: + rows = self._conn.execute("SELECT version FROM contracts WHERE name = ? ORDER BY version", (name,)).fetchall() + versions = [row["version"] for row in rows] + try: + return sorted(versions, key=Version) + except InvalidVersion: + return sorted(versions) + + def validate_payload(self, name: str, version: str, payload: dict[str, Any], *, kind: str = "input") -> None: + contract = self.get(name, version) + schema = contract.input_schema if kind == "input" else contract.output_schema + try: + validate(instance=payload, schema=schema) + passed = True + error = None + except ValidationError as exc: + passed = False + error = str(exc) + self._record_conformance(contract, kind, payload, passed, error) + raise + self._record_conformance(contract, kind, payload, passed, error) + + def _record_conformance(self, contract: AdapterContract, kind: str, payload: dict[str, Any], passed: bool, error: str | None) -> None: + self._conn.execute( + """ + INSERT INTO conformance_checks + (contract_name, contract_version, payload_kind, passed, error, payload) + VALUES (?, ?, ?, ?, ?, ?) + """, + (contract.name, contract.version, kind, int(passed), error, json.dumps(payload, sort_keys=True)), + ) + self._conn.commit() + + def conformance_report(self, name: str) -> dict[str, Any]: + rows = self._conn.execute( + "SELECT contract_version, payload_kind, passed, COUNT(*) AS count FROM conformance_checks WHERE contract_name = ? GROUP BY contract_version, payload_kind, passed", + (name,), + ).fetchall() + return { + "contract": name, + "checks": [dict(row) for row in rows], + } + + def _row_to_contract(self, row: sqlite3.Row) -> AdapterContract: + return AdapterContract( + name=row["name"], + version=row["version"], + adapter_type=row["adapter_type"], + input_schema=json.loads(row["input_schema"]), + output_schema=json.loads(row["output_schema"]), + capabilities=json.loads(row["capabilities"]), + metadata=json.loads(row["metadata"]), + ) diff --git a/src/solfuse/solver.py b/src/solfuse/solver.py new file mode 100644 index 0000000..29097bf --- /dev/null +++ b/src/solfuse/solver.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +from dataclasses import dataclass +import math + +import numpy as np + +from .models import LocalProblem, PlanAction, PlanDelta, SolverResult + + +@dataclass(frozen=True) +class _State: + x: np.ndarray + z: np.ndarray + u: np.ndarray + + +class ADMMLiteSolver: + def __init__(self, *, rho: float = 1.0, tolerance: float = 1e-6, max_iterations: int = 100) -> None: + if rho <= 0: + raise ValueError("rho must be positive") + self.rho = rho + self.tolerance = tolerance + self.max_iterations = max_iterations + + def solve(self, problems: list[LocalProblem]) -> SolverResult: + if not problems: + raise ValueError("at least one local problem is required") + + horizons = {len(problem.preferred_dispatch) for problem in problems} + if len(horizons) != 1: + raise ValueError("all problems must share the same dispatch horizon") + + horizon = horizons.pop() + lower = np.array([problem.lower_bounds or [-math.inf] * horizon for problem in problems], dtype=float) + upper = np.array([problem.upper_bounds or [math.inf] * horizon for problem in problems], dtype=float) + preferred = np.array([problem.preferred_dispatch for problem in problems], dtype=float) + weights = np.array([problem.quadratic_weight for problem in problems], dtype=float).reshape(-1, 1) + + x = preferred.copy() + z = preferred.mean(axis=0) + u = np.zeros_like(x) + primal_residual = float("inf") + dual_residual = float("inf") + + for iteration in range(1, self.max_iterations + 1): + previous_z = z.copy() + x = np.clip((weights * preferred + self.rho * (z - u)) / (weights + self.rho), lower, upper) + z = np.mean(x + u, axis=0) + u = u + x - z + + primal_residual = float(np.max(np.linalg.norm(x - z, axis=1))) + dual_residual = float(np.linalg.norm(z - previous_z) * math.sqrt(len(problems))) + if primal_residual <= self.tolerance and dual_residual <= self.tolerance: + break + + deltas: list[PlanDelta] = [] + for index, problem in enumerate(problems): + actions = [ + PlanAction(resource_id=f"{problem.site_id}:{slot}", action="dispatch", value=float(value), metadata={"slot": slot}) + for slot, value in enumerate(x[index]) + ] + deltas.append( + PlanDelta( + site_id=problem.site_id, + adapter_id=f"adapter:{problem.site_id}", + revision=iteration, + round_id=problem.round_id, + parent_revision=max(0, iteration - 1), + actions=actions, + metadata={"solver": "admm-lite", "round_id": problem.round_id}, + ) + ) + + return SolverResult( + consensus=[float(value) for value in z], + iterations=iteration, + primal_residual=primal_residual, + dual_residual=dual_residual, + deltas=deltas, + metadata={"rho": self.rho, "tolerance": self.tolerance, "max_iterations": self.max_iterations}, + ) diff --git a/src/solfuse/transport.py b/src/solfuse/transport.py new file mode 100644 index 0000000..a4f9940 --- /dev/null +++ b/src/solfuse/transport.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from ssl import CERT_REQUIRED, SSLContext, Purpose, TLSVersion, create_default_context +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + + +class MessageEnvelope(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + envelope_id: str + created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + source_did: str + schema_name: str + schema_version: str + payload: dict[str, Any] + signature: str + requires_tls: bool = True + + +def build_tls_context(*, server_side: bool, certfile: str | None = None, keyfile: str | None = None, cafile: str | None = None) -> SSLContext: + context = create_default_context(Purpose.CLIENT_AUTH if server_side else Purpose.SERVER_AUTH, cafile=cafile) + context.minimum_version = TLSVersion.TLSv1_3 + context.verify_mode = CERT_REQUIRED + if server_side: + if not certfile or not keyfile: + raise ValueError("server TLS context requires certfile and keyfile") + context.load_cert_chain(certfile=certfile, keyfile=keyfile) + return context diff --git a/test.sh b/test.sh new file mode 100644 index 0000000..fbc1778 --- /dev/null +++ b/test.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +set -euo pipefail + +python3 -m pip install -e ".[dev]" +pytest +python3 -m build diff --git a/tests/test_solfuse.py b/tests/test_solfuse.py new file mode 100644 index 0000000..21c5a5b --- /dev/null +++ b/tests/test_solfuse.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +import json + +import pytest + +from solfuse import ( + ADMMLiteSolver, + AdapterContract, + AuditLedger, + AuditLogEntry, + DeltaSyncJournal, + GraphOfContractsRegistry, + LocalProblem, + PrivacyBudget, + PrivacyBudgetLedger, + generate_identity, + sign_message, + verify_message, +) + + +def test_solver_converges_and_emits_deltas(): + solver = ADMMLiteSolver(rho=1.0, tolerance=1e-8, max_iterations=50) + result = solver.solve( + [ + LocalProblem(site_id="solar-a", preferred_dispatch=[4.0, 5.0], quadratic_weight=2.0), + LocalProblem(site_id="storage-b", preferred_dispatch=[2.0, 1.0], quadratic_weight=1.5), + ] + ) + + assert result.iterations <= 50 + assert len(result.consensus) == 2 + assert len(result.deltas) == 2 + assert result.primal_residual <= 1e-8 + + +def test_registry_validates_contract_payloads(): + registry = GraphOfContractsRegistry() + contract = AdapterContract( + name="solar-inverter", + version="1.0.0", + adapter_type="inverter", + input_schema={ + "type": "object", + "properties": {"site_id": {"type": "string"}, "dispatch_kw": {"type": "number"}}, + "required": ["site_id", "dispatch_kw"], + "additionalProperties": False, + }, + output_schema={"type": "object"}, + capabilities=["dispatch", "telemetry"], + ) + registry.register(contract) + registry.validate_payload("solar-inverter", "1.0.0", {"site_id": "a", "dispatch_kw": 3.5}) + with pytest.raises(Exception): + registry.validate_payload("solar-inverter", "1.0.0", {"site_id": "a"}) + report = registry.conformance_report("solar-inverter") + assert report["contract"] == "solar-inverter" + assert report["checks"] + + +def test_identity_signing_round_trip(): + identity = generate_identity("site-alpha") + payload = {"hello": "world", "n": 1} + signature = sign_message(identity, payload) + assert verify_message(identity.private_key.public_key(), payload, signature) + assert not verify_message(identity.private_key.public_key(), {"hello": "world", "n": 2}, signature) + + +def test_audit_and_privacy_budget_ledgers(): + budgets = PrivacyBudgetLedger() + budgets.set_budget("actor-1", PrivacyBudget(epsilon=1.0, delta=1e-5)) + updated = budgets.charge("actor-1", 0.25, 1e-6) + assert updated.spent_epsilon == pytest.approx(0.25) + + audit = AuditLedger() + entry = AuditLogEntry(actor_id="actor-1", event_type="plan_delta", payload={"a": 1}, signature="sig") + audit.append(entry) + recent = audit.recent() + assert recent[0].actor_id == "actor-1" + assert audit.coverage() == 1.0 + + +def test_delta_sync_journal_replays_sorted_deltas(): + journal = DeltaSyncJournal() + solver = ADMMLiteSolver() + deltas = solver.solve( + [LocalProblem(site_id="x", preferred_dispatch=[1.0], quadratic_weight=1.0)] + ).deltas + journal.merge(deltas) + loaded = journal.load() + assert loaded[0].site_id == "x" + + +def test_cli_payload_is_json_serializable(): + solver = ADMMLiteSolver() + result = solver.solve([LocalProblem(site_id="x", preferred_dispatch=[1.0], quadratic_weight=1.0)]) + json.dumps(result.model_dump(mode="json"))