From 27203fd1812042222b4900ff9e50b2cfac53fd29 Mon Sep 17 00:00:00 2001 From: agent-b883b4bc188823a2 Date: Sat, 25 Apr 2026 21:29:13 +0200 Subject: [PATCH] build(agent): c3po#b883b4 iteration --- AGENTS.md | 20 ++-- README.md | 59 +++++---- hopemesh/__init__.py | 30 ++++- hopemesh/cli.py | 145 ++++++++++++++++++++-- hopemesh/contracts.py | 273 ++++++++++++++++++++++++++++++++++++++++++ hopemesh/core.py | 89 ++++++++++++-- hopemesh/sdk.py | 77 ++++++++++-- test.sh | 16 ++- tests/test_cli.py | 34 ++++++ tests/test_core.py | 65 ++++++++++ 10 files changed, 743 insertions(+), 65 deletions(-) create mode 100644 hopemesh/contracts.py create mode 100644 tests/test_cli.py create mode 100644 tests/test_core.py diff --git a/AGENTS.md b/AGENTS.md index 0a0b2cc..acf2334 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -2,15 +2,15 @@ Architectural rules and contribution guidelines for HopeMesh 2.0 bootstrap. -- Language: Python. Packaging via pyproject.toml with setuptools. -- Key modules: - - hopemesh.contracts: Graph-of-Contracts registry skeleton and primitive types. - - hopemesh.sdk: Tiny DSL-like helpers for LocalProblem, SharedVariables, PlanDelta, Policy, AttestationHint, AuditLog, PrivacyBudget. - - hopemesh.cli: Lightweight command-line interface stub for adapters and governance operations. - -- Tests and build: - - test.sh should run: python3 -m build and a quick import/instantiate smoke test. +- Language: Python. Packaging via `pyproject.toml` with setuptools. +- Core modules: + - `hopemesh.core`: canonical primitives and fingerprinting helpers. + - `hopemesh.contracts`: Graph-of-Contracts registry plus tamper-evident governance ledger. + - `hopemesh.sdk`: DSL-style constructors and delta merge helpers. + - `hopemesh.cli`: local registry and ledger CLI. +- Testing: + - `test.sh` runs `python3 -m build`, a smoke import/instantiate check, and `python3 -m unittest discover -s tests -p 'test_*.py'`. - Publishing: - - Create READY_TO_PUBLISH when all checks pass. + - Create `READY_TO_PUBLISH` only when the repo is fully green and the implementation scope matches the published description. -This document is a placeholder for the initial bootstrap rules. As the repo evolves, update with concrete testing commands and CI expectations. +Keep changes small, deterministic, and standard-library only unless a new dependency is explicitly justified. diff --git a/README.md b/README.md index 3a7c6b7..f38cc20 100644 --- a/README.md +++ b/README.md @@ -1,29 +1,40 @@ -# HopeMesh 2.0 (Skeleton) +# HopeMesh 2.0 -This repository contains a production-ready skeleton for a federated, privacy-preserving humanitarian resource allocation platform as described in the HopeMesh 2.0 concept. The initial code focuses on: -- Defining canonical primitives: LocalProblem, SharedVariables, PlanDelta, Policy, AttestationHint, AuditLog, PrivacyBudget. -- A Graph-of-Contracts registry scaffold to manage data schemas and adapters. -- A minimal, deterministic SDK for building adapters that interoperate with the primitives. -- Packaging metadata and a test script that validates import/instantiation and packaging. +HopeMesh 2.0 is a Python package for federated humanitarian allocation workflows in disrupted networks. -How to run -- Ensure Python 3.8+ is installed. -- Install build tooling and build the package: - - python3 -m pip install --upgrade build - - python3 -m build -- Smoke test: - - python3 - << 'PY' - from hopemesh.sdk import LocalProblem, SharedVariables - lp = LocalProblem(name="RegionA Allocation") - sv = SharedVariables(aggregate_type="sum") - print(lp, sv) - PY +It provides: +- Canonical primitives for `LocalProblem`, `SharedVariables`, `PlanDelta`, `Policy`, `AttestationHint`, `AuditLog`, and `PrivacyBudget`. +- A Graph-of-Contracts registry for versioned schemas and adapters. +- A tamper-evident governance ledger. +- Small SDK helpers for deterministic construction and delta merging. +- A CLI for local registry and ledger snapshots. -This repo is intentionally minimal to bootstrap the architecture and testing workflow. A more complete implementation will grow the registry, adapters, and cryptographic governance features in subsequent commits. +## Layout -Usage notes: -- Run tests via ./test.sh -- The test script builds the package and runs a smoke test importing and instantiating primitives. -- After successful tests, create a READY_TO_PUBLISH marker in the repo root. +- `hopemesh/core.py`: primitive dataclasses and fingerprinting helpers. +- `hopemesh/contracts.py`: Graph-of-Contracts and governance ledger. +- `hopemesh/sdk.py`: DSL-style constructors and merge helpers. +- `hopemesh/cli.py`: local CLI for registry and ledger operations. +- `tests/`: unit tests for primitives, registry, ledger, and helpers. -READY_TO_PUBLISH is created when all checks pass. +## Quick Start + +```bash +python3 -m pip install --upgrade build +python3 -m build +./test.sh +``` + +## CLI + +```bash +python -m hopemesh.cli --version +python -m hopemesh.cli registry init --store .hopemesh/state.json +python -m hopemesh.cli registry add-schema --name allocation --schema-version 1.0 --definition-json '{"fields":["region","qty"]}' +python -m hopemesh.cli ledger verify +``` + +## Notes + +- The project is intentionally dependency-light and uses the Python standard library for validation, hashing, and snapshotting. +- `test.sh` runs `python3 -m build`, a smoke import/instantiate check, and the unit test suite. diff --git a/hopemesh/__init__.py b/hopemesh/__init__.py index 519c98d..60874bd 100644 --- a/hopemesh/__init__.py +++ b/hopemesh/__init__.py @@ -1,8 +1,18 @@ -"""hopemesh package root (skeleton). -Minimal primitives for HopeMesh 2.0 prototype. -""" +"""HopeMesh 2.0 primitives and registry scaffolding.""" from .core import LocalProblem, SharedVariables, PlanDelta, Policy, AttestationHint, AuditLog, PrivacyBudget +from .contracts import ContractRecord, ContractLink, GraphOfContracts, GovernanceLedger, LedgerEntry +from .sdk import ( + make_local_problem, + make_shared_variables, + make_plan_delta, + make_policy, + make_attestation_hint, + make_audit_log, + make_privacy_budget, + merge_plan_deltas, + validate_deltas, +) __all__ = [ "LocalProblem", @@ -12,4 +22,18 @@ __all__ = [ "AttestationHint", "AuditLog", "PrivacyBudget", + "ContractRecord", + "ContractLink", + "GraphOfContracts", + "GovernanceLedger", + "LedgerEntry", + "make_local_problem", + "make_shared_variables", + "make_plan_delta", + "make_policy", + "make_attestation_hint", + "make_audit_log", + "make_privacy_budget", + "merge_plan_deltas", + "validate_deltas", ] diff --git a/hopemesh/cli.py b/hopemesh/cli.py index e74d032..72f1cda 100644 --- a/hopemesh/cli.py +++ b/hopemesh/cli.py @@ -1,17 +1,144 @@ -"""Lightweight CLI stub for adapters and governance ops.""" +"""CLI for HopeMesh registry and governance snapshots.""" + +from __future__ import annotations + import argparse +import json +from pathlib import Path +from typing import Any, Dict -def main(): - parser = argparse.ArgumentParser(prog="hopemesh", description="HopeMesh 2.0 skeleton CLI") - subparsers = parser.add_subparsers(dest="cmd", required=False) +from .contracts import GraphOfContracts, GovernanceLedger - parser.add_argument("--version", action="store_true", help="Show version") - args = parser.parse_args() - if args.version: +DEFAULT_STORE = Path(".hopemesh/state.json") + + +def _load_state(store: Path) -> Dict[str, Any]: + if not store.exists(): + return {"registry": {"records": [], "links": []}, "ledger": {"entries": []}} + return json.loads(store.read_text(encoding="utf-8")) + + +def _save_state(store: Path, state: Dict[str, Any]) -> None: + store.parent.mkdir(parents=True, exist_ok=True) + store.write_text(json.dumps(state, indent=2, sort_keys=True), encoding="utf-8") + + +def _graph_from_state(state: Dict[str, Any]) -> GraphOfContracts: + return GraphOfContracts.from_snapshot(state.get("registry", {})) + + +def _ledger_from_state(state: Dict[str, Any]) -> GovernanceLedger: + return GovernanceLedger.from_snapshot(state.get("ledger", {})) + + +def _store_snapshot(store: Path, graph: GraphOfContracts, ledger: GovernanceLedger) -> None: + _save_state(store, {"registry": graph.snapshot(), "ledger": ledger.snapshot()}) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="hopemesh", description="HopeMesh 2.0 registry and governance CLI") + parser.add_argument("--version", dest="show_version", action="store_true", help="Show version") + subparsers = parser.add_subparsers(dest="cmd") + + registry = subparsers.add_parser("registry", help="Manage the Graph-of-Contracts") + registry_sub = registry.add_subparsers(dest="registry_cmd", required=True) + + registry_init = registry_sub.add_parser("init", help="Create an empty registry store") + registry_init.add_argument("--store", type=Path, default=DEFAULT_STORE) + + registry_schema = registry_sub.add_parser("add-schema", help="Register a schema contract") + registry_schema.add_argument("--store", type=Path, default=DEFAULT_STORE) + registry_schema.add_argument("--name", required=True) + registry_schema.add_argument("--schema-version", required=True) + registry_schema.add_argument("--definition-json", required=True) + + registry_adapter = registry_sub.add_parser("add-adapter", help="Register an adapter contract") + registry_adapter.add_argument("--store", type=Path, default=DEFAULT_STORE) + registry_adapter.add_argument("--name", required=True) + registry_adapter.add_argument("--adapter-version", required=True) + registry_adapter.add_argument("--definition-json", required=True) + + registry_link = registry_sub.add_parser("link", help="Link two contracts") + registry_link.add_argument("--store", type=Path, default=DEFAULT_STORE) + registry_link.add_argument("--source-id", required=True) + registry_link.add_argument("--target-id", required=True) + registry_link.add_argument("--relation", default="implements") + + registry_show = registry_sub.add_parser("snapshot", help="Print the registry snapshot") + registry_show.add_argument("--store", type=Path, default=DEFAULT_STORE) + + ledger = subparsers.add_parser("ledger", help="Manage the governance ledger") + ledger_sub = ledger.add_subparsers(dest="ledger_cmd", required=True) + + ledger_append = ledger_sub.add_parser("append", help="Append a governance event") + ledger_append.add_argument("--store", type=Path, default=DEFAULT_STORE) + ledger_append.add_argument("--event-type", required=True) + ledger_append.add_argument("--payload-json", required=True) + ledger_append.add_argument("--actor", default="system") + + ledger_verify = ledger_sub.add_parser("verify", help="Verify ledger integrity") + ledger_verify.add_argument("--store", type=Path, default=DEFAULT_STORE) + + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + + if args.show_version: print("hopemesh-sdk 0.1.0") - elif not args.cmd: + return 0 + + if not args.cmd: parser.print_help() + return 0 + + state = _load_state(args.store) + graph = _graph_from_state(state) + ledger = _ledger_from_state(state) + + if args.cmd == "registry": + if args.registry_cmd == "init": + _store_snapshot(args.store, graph, ledger) + print(str(args.store)) + return 0 + if args.registry_cmd in {"add-schema", "add-adapter"}: + definition = json.loads(args.definition_json) + if args.registry_cmd == "add-schema": + record = graph.register_schema(args.name, args.schema_version, definition) + else: + record = graph.register_adapter(args.name, args.adapter_version, definition) + ledger.append("registry.register", {"contract_id": record.contract_id, "kind": record.kind}) + _store_snapshot(args.store, graph, ledger) + print(record.contract_id) + return 0 + if args.registry_cmd == "link": + link = graph.link(args.source_id, args.target_id, args.relation) + ledger.append("registry.link", link.to_dict()) + _store_snapshot(args.store, graph, ledger) + print(json.dumps(link.to_dict(), sort_keys=True)) + return 0 + if args.registry_cmd == "snapshot": + print(json.dumps(graph.snapshot(), indent=2, sort_keys=True)) + return 0 + + if args.cmd == "ledger": + if args.ledger_cmd == "append": + payload = json.loads(args.payload_json) + entry = ledger.append(args.event_type, payload, actor=args.actor) + _store_snapshot(args.store, graph, ledger) + print(json.dumps(entry.to_dict(), indent=2, sort_keys=True)) + return 0 + if args.ledger_cmd == "verify": + issues = ledger.verify() + print(json.dumps({"ok": not issues, "issues": issues}, indent=2, sort_keys=True)) + return 0 + + parser.error("unknown command") + return 2 + if __name__ == "__main__": - main() + raise SystemExit(main()) diff --git a/hopemesh/contracts.py b/hopemesh/contracts.py new file mode 100644 index 0000000..aefd8df --- /dev/null +++ b/hopemesh/contracts.py @@ -0,0 +1,273 @@ +from __future__ import annotations + +from dataclasses import dataclass, asdict, field +from typing import Any, Dict, Iterable, List, Optional +import hashlib +import json +import time + +from .core import AttestationHint, AuditLog, PlanDelta, Policy, _canonical_json + + +def _hash(payload: Any) -> str: + return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest() + + +@dataclass +class ContractRecord: + contract_id: str + kind: str + name: str + version: str + definition: Dict[str, Any] = field(default_factory=dict) + attestation: Optional[AttestationHint] = None + created_at: float = field(default_factory=lambda: time.time()) + + def to_dict(self) -> Dict[str, Any]: + data = asdict(self) + if self.attestation is not None: + data["attestation"] = self.attestation.to_dict() + return data + + +@dataclass +class ContractLink: + source_id: str + target_id: str + relation: str + created_at: float = field(default_factory=lambda: time.time()) + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + + +class GraphOfContracts: + def __init__(self) -> None: + self._records: Dict[str, ContractRecord] = {} + self._links: List[ContractLink] = [] + + def register_schema( + self, + name: str, + version: str, + definition: Dict[str, Any], + attestation: Optional[AttestationHint] = None, + ) -> ContractRecord: + contract_id = f"schema:{name}:{version}" + record = ContractRecord( + contract_id=contract_id, + kind="schema", + name=name, + version=version, + definition=dict(definition), + attestation=attestation, + ) + self._records[contract_id] = record + return record + + def register_adapter( + self, + name: str, + version: str, + definition: Dict[str, Any], + attestation: Optional[AttestationHint] = None, + ) -> ContractRecord: + contract_id = f"adapter:{name}:{version}" + record = ContractRecord( + contract_id=contract_id, + kind="adapter", + name=name, + version=version, + definition=dict(definition), + attestation=attestation, + ) + self._records[contract_id] = record + return record + + def link(self, source_id: str, target_id: str, relation: str = "implements") -> ContractLink: + if source_id not in self._records: + raise KeyError(f"Unknown source contract: {source_id}") + if target_id not in self._records: + raise KeyError(f"Unknown target contract: {target_id}") + link = ContractLink(source_id=source_id, target_id=target_id, relation=relation) + self._links.append(link) + return link + + def resolve(self, contract_id: str) -> ContractRecord: + return self._records[contract_id] + + def records(self) -> List[ContractRecord]: + return list(self._records.values()) + + def links(self) -> List[ContractLink]: + return list(self._links) + + def validate(self) -> List[str]: + issues: List[str] = [] + for link in self._links: + if link.source_id not in self._records: + issues.append(f"missing source {link.source_id}") + if link.target_id not in self._records: + issues.append(f"missing target {link.target_id}") + return issues + + def snapshot(self) -> Dict[str, Any]: + return { + "records": [record.to_dict() for record in self.records()], + "links": [link.to_dict() for link in self.links()], + "fingerprint": self.fingerprint(), + } + + @classmethod + def from_snapshot(cls, snapshot: Dict[str, Any]) -> "GraphOfContracts": + graph = cls() + for record_data in snapshot.get("records", []): + attestation = record_data.get("attestation") + hint = AttestationHint(**attestation) if attestation else None + record = ContractRecord( + contract_id=record_data["contract_id"], + kind=record_data["kind"], + name=record_data["name"], + version=record_data["version"], + definition=dict(record_data.get("definition", {})), + attestation=hint, + created_at=record_data.get("created_at", time.time()), + ) + graph._records[record.contract_id] = record + for link_data in snapshot.get("links", []): + graph._links.append( + ContractLink( + source_id=link_data["source_id"], + target_id=link_data["target_id"], + relation=link_data["relation"], + created_at=link_data.get("created_at", time.time()), + ) + ) + return graph + + def fingerprint(self) -> str: + payload = { + "records": [record.to_dict() for record in sorted(self._records.values(), key=lambda item: item.contract_id)], + "links": [link.to_dict() for link in self._links], + } + return _hash(payload) + + +@dataclass +class LedgerEntry: + entry_id: str + event_type: str + payload: Dict[str, Any] + actor: str + timestamp: float + previous_hash: str + entry_hash: str + signature: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + + +class GovernanceLedger: + def __init__(self) -> None: + self._entries: List[LedgerEntry] = [] + + def append( + self, + event_type: str, + payload: Dict[str, Any], + actor: str = "system", + signature: Optional[str] = None, + timestamp: Optional[float] = None, + ) -> LedgerEntry: + ts = time.time() if timestamp is None else timestamp + previous_hash = self._entries[-1].entry_hash if self._entries else "genesis" + entry_id = payload.get("entry_id") or f"{event_type}-{len(self._entries) + 1}" + basis = { + "entry_id": entry_id, + "event_type": event_type, + "payload": payload, + "actor": actor, + "timestamp": ts, + "previous_hash": previous_hash, + } + entry_hash = _hash(basis) + entry = LedgerEntry( + entry_id=entry_id, + event_type=event_type, + payload=dict(payload), + actor=actor, + timestamp=ts, + previous_hash=previous_hash, + entry_hash=entry_hash, + signature=signature or entry_hash, + ) + self._entries.append(entry) + return entry + + def entries(self) -> List[LedgerEntry]: + return list(self._entries) + + def verify(self) -> List[str]: + issues: List[str] = [] + previous_hash = "genesis" + for entry in self._entries: + basis = { + "entry_id": entry.entry_id, + "event_type": entry.event_type, + "payload": entry.payload, + "actor": entry.actor, + "timestamp": entry.timestamp, + "previous_hash": previous_hash, + } + expected_hash = _hash(basis) + if expected_hash != entry.entry_hash: + issues.append(f"entry {entry.entry_id} hash mismatch") + if entry.previous_hash != previous_hash: + issues.append(f"entry {entry.entry_id} chain mismatch") + previous_hash = entry.entry_hash + return issues + + def snapshot(self) -> Dict[str, Any]: + return {"entries": [entry.to_dict() for entry in self._entries], "fingerprint": self.fingerprint()} + + @classmethod + def from_snapshot(cls, snapshot: Dict[str, Any]) -> "GovernanceLedger": + ledger = cls() + for entry_data in snapshot.get("entries", []): + ledger._entries.append( + LedgerEntry( + entry_id=entry_data["entry_id"], + event_type=entry_data["event_type"], + payload=dict(entry_data.get("payload", {})), + actor=entry_data["actor"], + timestamp=entry_data["timestamp"], + previous_hash=entry_data["previous_hash"], + entry_hash=entry_data["entry_hash"], + signature=entry_data.get("signature"), + ) + ) + return ledger + + def fingerprint(self) -> str: + return _hash({"entries": [entry.to_dict() for entry in self._entries]}) + + +def validate_plan_deltas(deltas: Iterable[PlanDelta]) -> List[str]: + issues: List[str] = [] + for delta in deltas: + if not delta.actions: + issues.append(f"{delta.delta_id}: empty actions") + if delta.signature is not None and not delta.signature.strip(): + issues.append(f"{delta.delta_id}: blank signature") + return issues + + +def policy_summary(policy: Policy) -> str: + return json.dumps(policy.to_dict(), sort_keys=True) + + +def audit_plan_delta(delta: PlanDelta, actor: str, message: str) -> AuditLog: + payload = {"delta_id": delta.delta_id, "actor": actor, "message": message, "fingerprint": delta.fingerprint()} + signature = _hash(payload) + return AuditLog(entry_id=f"audit-{delta.delta_id}", message=message, signature=signature, anchor=payload["fingerprint"]) diff --git a/hopemesh/core.py b/hopemesh/core.py index 015bf87..33c1fd2 100644 --- a/hopemesh/core.py +++ b/hopemesh/core.py @@ -1,29 +1,60 @@ from __future__ import annotations -from dataclasses import dataclass, field + +from dataclasses import dataclass, field, asdict from typing import Any, Dict, List, Optional +import hashlib +import json import time +def _canonical_json(payload: Any) -> str: + return json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + + +def _fingerprint(payload: Any) -> str: + return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest() + + @dataclass class LocalProblem: name: str region: Optional[str] = None + objective: str = "" + constraints: Dict[str, Any] = field(default_factory=dict) metadata: Dict[str, Any] = field(default_factory=dict) created_at: float = field(default_factory=lambda: time.time()) + def __post_init__(self) -> None: + if not self.name.strip(): + raise ValueError("LocalProblem.name must not be empty") + def __repr__(self) -> str: - return f"LocalProblem(name={self.name!r}, region={self.region!r})" + return f"LocalProblem(name={self.name!r}, region={self.region!r}, objective={self.objective!r})" + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + + def fingerprint(self) -> str: + return _fingerprint(self.to_dict()) @dataclass class SharedVariables: description: str = "" data: Dict[str, Any] = field(default_factory=dict) - aggregate_type: str = "sum" # placeholder for privacy-preserving aggregation + aggregate_type: str = "sum" + privacy_budget_tag: Optional[str] = None + source_regions: List[str] = field(default_factory=list) created_at: float = field(default_factory=lambda: time.time()) def __repr__(self) -> str: - return f"SharedVariables(desc={self.description!r}, keys={list(self.data.keys())})" + return f"SharedVariables(desc={self.description!r}, keys={list(self.data.keys())}, aggregate={self.aggregate_type!r})" + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + + def fingerprint(self) -> str: + return _fingerprint(self.to_dict()) @dataclass @@ -31,10 +62,27 @@ class PlanDelta: delta_id: str actions: List[Dict[str, Any]] = field(default_factory=list) timestamp: float = field(default_factory=lambda: time.time()) + actor_did: Optional[str] = None + causality: Dict[str, int] = field(default_factory=dict) + policy_tags: List[str] = field(default_factory=list) + privacy_budget_tag: Optional[str] = None annotations: Dict[str, Any] = field(default_factory=dict) + signature: Optional[str] = None + + def __post_init__(self) -> None: + if not self.delta_id.strip(): + raise ValueError("PlanDelta.delta_id must not be empty") + if not isinstance(self.actions, list): + raise TypeError("PlanDelta.actions must be a list") def __repr__(self) -> str: - return f"PlanDelta(delta_id={self.delta_id!r}, actions={len(self.actions)} actions)" + return f"PlanDelta(delta_id={self.delta_id!r}, actions={len(self.actions)} actions, actor={self.actor_did!r})" + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + + def fingerprint(self) -> str: + return _fingerprint(self.to_dict()) @dataclass @@ -42,9 +90,14 @@ class Policy: policy_id: str rules: Dict[str, Any] = field(default_factory=dict) effective_at: float = field(default_factory=lambda: time.time()) + scope: str = "global" + approved_by: Optional[str] = None def __repr__(self) -> str: - return f"Policy(id={self.policy_id!r}, rules={len(self.rules)})" + return f"Policy(id={self.policy_id!r}, rules={len(self.rules)}, scope={self.scope!r})" + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) @dataclass @@ -52,10 +105,14 @@ class AttestationHint: hint_id: str scheme: str = "dummy" payload: Dict[str, Any] = field(default_factory=dict) + expires_at: Optional[float] = None def __repr__(self) -> str: return f"AttestationHint(id={self.hint_id!r}, scheme={self.scheme!r})" + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + @dataclass class AuditLog: @@ -63,21 +120,37 @@ class AuditLog: message: str timestamp: float = field(default_factory=lambda: time.time()) signature: Optional[str] = None + anchor: Optional[str] = None def __repr__(self) -> str: return f"AuditLog(id={self.entry_id!r}, ts={self.timestamp})" + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + @dataclass class PrivacyBudget: budget_id: str limit: float used: float = 0.0 - def consume(self, amount: float) -> None: - self.used = min(self.used + amount, self.limit) + + def consume(self, amount: float) -> bool: + if amount < 0: + raise ValueError("PrivacyBudget.consume requires a non-negative amount") + if self.used + amount > self.limit: + return False + self.used += amount + return True + + def can_consume(self, amount: float) -> bool: + return amount >= 0 and self.used + amount <= self.limit def remaining(self) -> float: return self.limit - self.used + def to_dict(self) -> Dict[str, Any]: + return asdict(self) + def __repr__(self) -> str: return f"PrivacyBudget(id={self.budget_id!r}, remaining={self.remaining()})" diff --git a/hopemesh/sdk.py b/hopemesh/sdk.py index 9e87aa0..6d862bd 100644 --- a/hopemesh/sdk.py +++ b/hopemesh/sdk.py @@ -1,18 +1,55 @@ from __future__ import annotations +"""Tiny DSL-like helpers for HopeMesh primitives and contract tooling.""" + +from typing import Iterable + from .core import LocalProblem, SharedVariables, PlanDelta, Policy, AttestationHint, AuditLog, PrivacyBudget +from .contracts import validate_plan_deltas -"""Tiny DSL-like helpers for HopeMesh primitives. -This module provides lightweight constructors to enable adapters to plug into the canonical primitives. -""" +def make_local_problem( + name: str, + region: str | None = None, + objective: str = "", + constraints: dict | None = None, + **metadata, +) -> LocalProblem: + return LocalProblem(name=name, region=region, objective=objective, constraints=constraints or {}, metadata=metadata) -def make_local_problem(name: str, region: str | None = None, **kwargs) -> LocalProblem: - return LocalProblem(name=name, region=region, metadata=kwargs) +def make_shared_variables( + description: str = "", + data: dict | None = None, + aggregate_type: str = "sum", + privacy_budget_tag: str | None = None, + source_regions: list[str] | None = None, +) -> SharedVariables: + return SharedVariables( + description=description, + data=data or {}, + aggregate_type=aggregate_type, + privacy_budget_tag=privacy_budget_tag, + source_regions=source_regions or [], + ) -def make_shared_variables(description: str = "", data: dict | None = None, aggregate_type: str = "sum") -> SharedVariables: - return SharedVariables(description=description, data=data or {}, aggregate_type=aggregate_type) - -def make_plan_delta(delta_id: str, actions: list[dict], **kwargs) -> PlanDelta: - return PlanDelta(delta_id=delta_id, actions=actions, annotations=kwargs) +def make_plan_delta( + delta_id: str, + actions: list[dict], + actor_did: str | None = None, + causality: dict | None = None, + policy_tags: list[str] | None = None, + privacy_budget_tag: str | None = None, + signature: str | None = None, + **annotations, +) -> PlanDelta: + return PlanDelta( + delta_id=delta_id, + actions=actions, + actor_did=actor_did, + causality=causality or {}, + policy_tags=policy_tags or [], + privacy_budget_tag=privacy_budget_tag, + signature=signature, + annotations=annotations, + ) def make_policy(policy_id: str, **rules) -> Policy: return Policy(policy_id=policy_id, rules=rules) @@ -25,3 +62,23 @@ def make_audit_log(entry_id: str, message: str, signature: str | None = None) -> def make_privacy_budget(budget_id: str, limit: float) -> PrivacyBudget: return PrivacyBudget(budget_id=budget_id, limit=limit) + + +def merge_plan_deltas(deltas: Iterable[PlanDelta]) -> PlanDelta: + ordered = sorted(deltas, key=lambda delta: (delta.timestamp, delta.delta_id)) + actions: list[dict] = [] + contributors: list[str] = [] + for delta in ordered: + contributors.append(delta.delta_id) + actions.extend(delta.actions) + delta_id = "merged-" + "-".join(contributors) if contributors else "merged-empty" + return PlanDelta( + delta_id=delta_id, + actions=actions, + causality={"merged_from": len(contributors)}, + annotations={"contributors": contributors}, + ) + + +def validate_deltas(deltas: Iterable[PlanDelta]) -> list[str]: + return validate_plan_deltas(deltas) diff --git a/test.sh b/test.sh index aeaf27c..0eed93c 100644 --- a/test.sh +++ b/test.sh @@ -7,14 +7,28 @@ python3 -m build >/tmp/build.log 2>&1 || { echo "Smoke test: import modules and create objects..." python3 - << 'PY' -from hopemesh.sdk import make_local_problem, make_shared_variables, make_plan_delta +from hopemesh import GraphOfContracts, GovernanceLedger +from hopemesh.sdk import make_local_problem, make_shared_variables, make_plan_delta, merge_plan_deltas lp = make_local_problem("RegionA Allocation", region="RegionA") sv = make_shared_variables("example", {"stock": 100}) pd = make_plan_delta("pd-001", [{"allocate": {"region": "RegionA", "qty": 10}}]) +graph = GraphOfContracts() +schema = graph.register_schema("allocation", "1.0", {"fields": ["region", "qty"]}) +adapter = graph.register_adapter("regional-hub", "1.0", {"transport": "tls"}) +graph.link(adapter.contract_id, schema.contract_id) +ledger = GovernanceLedger() +ledger.append("demo", {"delta": pd.delta_id}) +merged = merge_plan_deltas([pd]) print(lp) print(sv) print(pd) +print(graph.fingerprint()) +print(ledger.verify()) +print(merged) print("OK") PY +echo "Running unit tests..." +python3 -m unittest discover -s tests -p 'test_*.py' + echo "All tests passed." diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..a998fdb --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,34 @@ +import tempfile +import unittest +from pathlib import Path + +from hopemesh.cli import main + + +class HopeMeshCLITests(unittest.TestCase): + def test_registry_and_ledger_commands(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + store = Path(tmpdir) / "state.json" + self.assertEqual(main(["registry", "init", "--store", str(store)]), 0) + self.assertEqual( + main( + [ + "registry", + "add-schema", + "--store", + str(store), + "--name", + "allocation", + "--schema-version", + "1.0", + "--definition-json", + '{"fields":["region","qty"]}', + ] + ), + 0, + ) + self.assertEqual(main(["ledger", "verify", "--store", str(store)]), 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..d1f8073 --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,65 @@ +import unittest + +from hopemesh import ( + GraphOfContracts, + GovernanceLedger, + LocalProblem, + PrivacyBudget, + make_local_problem, + make_plan_delta, + make_shared_variables, + merge_plan_deltas, + validate_deltas, +) + + +class HopeMeshCoreTests(unittest.TestCase): + def test_primitives_and_budget(self) -> None: + problem = LocalProblem(name="Water allocation", region="North", objective="minimize unmet demand") + shared = make_shared_variables("stock", {"water": 10}, source_regions=["North"]) + budget = PrivacyBudget("pb-1", 3.0) + + self.assertTrue(problem.fingerprint()) + self.assertTrue(shared.fingerprint()) + self.assertTrue(budget.consume(1.5)) + self.assertFalse(budget.consume(2.0)) + self.assertEqual(round(budget.remaining(), 2), 1.5) + + def test_graph_of_contracts(self) -> None: + graph = GraphOfContracts() + schema = graph.register_schema("resource-plan", "1.0", {"fields": ["id", "qty"]}) + adapter = graph.register_adapter("field-hub", "1.0", {"transport": "tls"}) + link = graph.link(adapter.contract_id, schema.contract_id) + + self.assertEqual(link.relation, "implements") + self.assertEqual(graph.validate(), []) + snapshot = graph.snapshot() + restored = GraphOfContracts.from_snapshot(snapshot) + self.assertEqual(restored.fingerprint(), graph.fingerprint()) + + def test_ledger_chain(self) -> None: + ledger = GovernanceLedger() + first = ledger.append("audit", {"message": "created"}) + second = ledger.append("audit", {"message": "updated"}, actor="operator") + + self.assertNotEqual(first.entry_hash, second.entry_hash) + self.assertEqual(ledger.verify(), []) + restored = GovernanceLedger.from_snapshot(ledger.snapshot()) + self.assertEqual(restored.verify(), []) + + def test_delta_helpers(self) -> None: + delta_a = make_plan_delta("a", [{"allocate": {"region": "North", "qty": 2}}]) + delta_b = make_plan_delta("b", [{"allocate": {"region": "South", "qty": 3}}]) + merged = merge_plan_deltas([delta_b, delta_a]) + + self.assertEqual(len(merged.actions), 2) + self.assertEqual(validate_deltas([delta_a, delta_b]), []) + + def test_factory_alias(self) -> None: + problem = make_local_problem("Hospital supply", region="East", urgent=True) + self.assertEqual(problem.region, "East") + self.assertEqual(problem.metadata["urgent"], True) + + +if __name__ == "__main__": + unittest.main()