build(agent): new-agents-2#7e3bbc iteration

This commit is contained in:
agent-7e3bbc424e07835b 2026-04-19 21:49:33 +02:00
parent d4821d1212
commit 204901b7d5
3 changed files with 77 additions and 0 deletions

View File

@ -7,6 +7,7 @@ category-theory abstractions. This readme documents the current MVP scope and
how to extend it.
- MVP surface: a simple add(a, b) function to validate packaging and imports.
- GoC: Graph-of-Contracts registry scaffold added to enable a minimal interoperability surface for adapters and contracts. See src/catopt_category_theoretic_compositional_/goc.py and the envelope_contract helper in the public API.
- Packaging: aligned with pyproject.toml, using setuptools to build a wheel.
- Extensibility: planned steps include a DSL for local problems, data contracts,
an ADMM-like solver kernel, and adapters for edge devices.

View File

@ -2,6 +2,7 @@
from .runtime import LocalProblem, SharedVariables, DataContract, PlanDelta, ADMMTwoAgentSolver, demo_two_agent_admm, admm_update # noqa: F401
from .dsl import ProtocolContract, build_minimal_contract # noqa: F401
from .goc import GraphOfContracts, envelope_contract # noqa: F401
def add(a: int, b: int) -> int:
"""Return the sum of two integers.
@ -22,4 +23,6 @@ __all__ = [
"admm_update",
"ProtocolContract",
"build_minimal_contract",
"GraphOfContracts",
"envelope_contract",
]

View File

@ -0,0 +1,73 @@
"""Graph of Contracts (GoC) skeleton for CatOpt MVP.
This module provides a minimal, production-friendly scaffold for a Graph-of-Contracts
registry that maps adapters to supported domains and contract versions.
The implementation is intentionally small but supplies a stable API surface that
can be extended in subsequent sprint cycles.
"""
from __future__ import annotations
from dataclasses import dataclass, field, asdict
from typing import List, Optional, Dict
import json
import time
from .dsl import ProtocolContract
@dataclass
class GoCEntry:
adapter_id: str
supported_domains: List[str]
contract_version: int
metadata: Dict[str, str] = field(default_factory=dict)
registered_at: float = field(default_factory=time.time)
def to_dict(self) -> Dict:
data = asdict(self)
# Ensure JSON-friendly types only
data["registered_at"] = float(data["registered_at"])
return data
@dataclass
class GraphOfContracts:
"""A lightweight registry of adapters and the contracts they support."""
registry: List[GoCEntry] = field(default_factory=list)
def register(self, adapter_id: str, domains: List[str], contract_version: int, metadata: Optional[Dict[str, str]] = None) -> None:
entry = GoCEntry(
adapter_id=adapter_id,
supported_domains=domains,
contract_version=contract_version,
metadata=metadata or {},
)
self.registry.append(entry)
def to_json(self) -> str:
return json.dumps([e.to_dict() for e in self.registry], indent=2, sort_keys=True)
def __len__(self) -> int:
return len(self.registry)
def envelope_contract(contract: ProtocolContract, version: int = 1, signer: str = " CatOpt") -> Dict:
"""Wrap a ProtocolContract into a transport-ready envelope.
The envelope carries lightweight metadata useful for replay protection and auditing.
NOTE: This is a minimal MVP; real deployments should attach real cryptographic signatures.
"""
header = {
"version": version,
"timestamp": time.time(),
"nonce": hex(int.from_bytes(__import__("os").urandom(8), "big")),
"signer": signer,
# In production, add a real signature over the payload
"signature": "",
}
return {"header": header, "contract": contract.to_dict()}
__all__ = ["GoCEntry", "GraphOfContracts", "envelope_contract"]