build(agent): jabba#56a767 iteration

This commit is contained in:
agent-56a7678c6cd71659 2026-04-29 22:23:53 +02:00
parent 65fe7ea6de
commit f41ceec324
5 changed files with 64 additions and 1 deletions

View File

@ -22,3 +22,9 @@ Roadmap (high level):
- Add a tiny solver placeholder and an adapter interface
- Introduce a basic conformance test suite and registry stubs
- Add a minimal README for onboarding users and developers
New additions:
- ServiceGenome manifest: a lightweight runtime manifest for adapters/solvers
(runtime_role, capability_vector, resource_envelope, safety_invariants,
compliance_hook). This is exposed as catopt_studio.ServiceGenome and the
in-memory registry can store and retrieve genomes for contract entries.

View File

@ -1,6 +1,6 @@
"""Public API for the CatOpt Studio MVP."""
from .core import LocalProblem, SharedVariables, PlanDelta, PrivacyBudget, AuditLog, PolicyBlock, GraphOfContractsEntry
from .core import LocalProblem, SharedVariables, PlanDelta, PrivacyBudget, AuditLog, PolicyBlock, GraphOfContractsEntry, ServiceGenome
from .solver import SolverConfig, compile_to_solver, simulate_solver_step
from .adapters import RoverAdapter, HabitatAdapter
from .registry import default_registry, InMemoryRegistry
@ -13,6 +13,7 @@ __all__ = [
"AuditLog",
"PolicyBlock",
"GraphOfContractsEntry",
"ServiceGenome",
"SolverConfig",
"compile_to_solver",
"simulate_solver_step",

View File

@ -63,3 +63,19 @@ class PolicyBlock:
class GraphOfContractsEntry:
adapter_name: str
canonical_schema: str
@dataclass
class ServiceGenome:
"""Lightweight ServiceGenome manifest for adapters/solvers.
This mirrors the proposal in the community discussion and captures a
minimal set of fields useful for runtime selection, canarying and
admission control in edge environments.
"""
runtime_role: str # e.g., 'adapter' | 'local-solver' | 'aggregator'
capability_vector: Dict[str, Any] = field(default_factory=dict)
resource_envelope: Dict[str, Any] = field(default_factory=dict) # cpu, memory, energy estimates
safety_invariants: Dict[str, Any] = field(default_factory=dict)
compliance_hook: Optional[str] = None # URL for sandbox/compliance checks
version: Optional[str] = None

View File

@ -24,6 +24,7 @@ class InMemoryRegistry:
self._contracts: Dict[str, RegistryEntry] = {}
self._signals: List[Dict[str, Any]] = []
self._deltas: List[Dict[str, Any]] = []
self._service_genomes: Dict[str, Dict[str, Any]] = {}
def register_contract(self, contract_id: str, adapter: str, schema: Dict[str, Any]) -> RegistryEntry:
entry = RegistryEntry(contract_id=contract_id, adapter=adapter, schema=schema)
@ -50,6 +51,19 @@ class InMemoryRegistry:
"deltas_count": len(deltas),
}
def register_service_genome(self, contract_id: str, genome: Dict[str, Any]) -> None:
"""Register a ServiceGenome (manifest) for a contract.
This stores a small, JSON-serializable manifest describing runtime
role, capability vectors, resource envelope and declared safety
invariants. It is intentionally lightweight for tests and demos.
"""
self._service_genomes[contract_id] = {"genome": genome, "timestamp": datetime.utcnow()}
def get_service_genome(self, contract_id: str) -> Optional[Dict[str, Any]]:
record = self._service_genomes.get(contract_id)
return record
# Module-level default registry for convenience in tests and examples
default_registry = InMemoryRegistry()

View File

@ -0,0 +1,26 @@
# Simple tests for the ServiceGenome manifest and registry support
from catopt_studio import LocalProblem, ServiceGenome
from catopt_studio.registry import InMemoryRegistry
def test_service_genome_dataclass_and_registry():
sg = ServiceGenome(
runtime_role="local-solver",
capability_vector={"cpu": 2, "mem_mb": 256},
resource_envelope={"cpu_cycles_per_iter": 1000000},
safety_invariants={"max_actuation_rate": 5},
compliance_hook="https://example.com/check",
version="v0.1",
)
# ensure fields exist and are accessible
assert sg.runtime_role == "local-solver"
assert sg.capability_vector["cpu"] == 2
# registry roundtrip
reg = InMemoryRegistry()
reg.register_contract("svc1", "local-solver", {"schema": "sg-v1"})
reg.register_service_genome("svc1", sg.__dict__)
got = reg.get_service_genome("svc1")
assert got is not None
assert got["genome"]["version"] == "v0.1"