From 65fe7ea6de91c20463f604abb9ebd05ec28d5c6a Mon Sep 17 00:00:00 2001 From: agent-14fd4b738639d573 Date: Fri, 24 Apr 2026 19:02:30 +0200 Subject: [PATCH] build(agent): melter#14fd4b iteration --- catopt_studio/__init__.py | 6 +++ catopt_studio/adapters/__init__.py | 12 ++++++ catopt_studio/adapters/habitat_module.py | 38 ++++++++++++++++ catopt_studio/adapters/rover_planner.py | 46 ++++++++++++++++++++ catopt_studio/registry.py | 55 ++++++++++++++++++++++++ tests/test_adapters.py | 27 ++++++++++++ tests/test_registry.py | 12 ++++++ 7 files changed, 196 insertions(+) create mode 100644 catopt_studio/adapters/__init__.py create mode 100644 catopt_studio/adapters/habitat_module.py create mode 100644 catopt_studio/adapters/rover_planner.py create mode 100644 catopt_studio/registry.py create mode 100644 tests/test_adapters.py create mode 100644 tests/test_registry.py diff --git a/catopt_studio/__init__.py b/catopt_studio/__init__.py index 72c901e..44579a2 100644 --- a/catopt_studio/__init__.py +++ b/catopt_studio/__init__.py @@ -2,6 +2,8 @@ from .core import LocalProblem, SharedVariables, PlanDelta, PrivacyBudget, AuditLog, PolicyBlock, GraphOfContractsEntry from .solver import SolverConfig, compile_to_solver, simulate_solver_step +from .adapters import RoverAdapter, HabitatAdapter +from .registry import default_registry, InMemoryRegistry __all__ = [ "LocalProblem", @@ -14,4 +16,8 @@ __all__ = [ "SolverConfig", "compile_to_solver", "simulate_solver_step", + "RoverAdapter", + "HabitatAdapter", + "default_registry", + "InMemoryRegistry", ] diff --git a/catopt_studio/adapters/__init__.py b/catopt_studio/adapters/__init__.py new file mode 100644 index 0000000..710a6e1 --- /dev/null +++ b/catopt_studio/adapters/__init__.py @@ -0,0 +1,12 @@ +"""Adapters package for CatOpt Studio MVP. + +This package contains toy adapter implementations that translate device- +specific LocalProblem instances into a small canonical mapping and emit +PlanDelta objects. These are intentionally small, deterministic, and +well-tested to serve as examples for adapter development. +""" + +from .rover_planner import RoverAdapter +from .habitat_module import HabitatAdapter + +__all__ = ["RoverAdapter", "HabitatAdapter"] diff --git a/catopt_studio/adapters/habitat_module.py b/catopt_studio/adapters/habitat_module.py new file mode 100644 index 0000000..cc31fcb --- /dev/null +++ b/catopt_studio/adapters/habitat_module.py @@ -0,0 +1,38 @@ +"""Toy habitat module adapter. + +The HabitatAdapter provides a simple mapping for habitat-style LocalProblem +instances and produces small PlanDelta objects. It serves as a complementary +example adapter to the rover planner. +""" +from dataclasses import dataclass +from typing import Dict, Any + +from ..core import LocalProblem, PlanDelta + + +@dataclass +class HabitatAdapter: + name: str = "habitat_module" + + def to_canonical(self, lp: LocalProblem) -> Dict[str, Any]: + return { + "adapter": self.name, + "id": lp.id, + "domain": lp.domain, + "assets_count": len(lp.assets or {}), + "constraints": lp.constraints, + } + + def propose_delta(self, lp: LocalProblem, author: str = "habitat") -> PlanDelta: + # Simple policy: if any safety constraint mentions 'no_move', emit + # a delta that pauses movement; otherwise, propose a nominal plan. + delta = {} + for c in lp.constraints: + if "no_move" in c: + delta["action"] = "pause" + delta["reason"] = "safety_constraint" + break + else: + delta["action"] = "nominal" + + return PlanDelta(delta=delta, author=author, contract_id=lp.id) diff --git a/catopt_studio/adapters/rover_planner.py b/catopt_studio/adapters/rover_planner.py new file mode 100644 index 0000000..91352c5 --- /dev/null +++ b/catopt_studio/adapters/rover_planner.py @@ -0,0 +1,46 @@ +"""Toy rover planner adapter. + +The RoverAdapter demonstrates a minimal adapter that maps a LocalProblem +from a rover domain into a canonical representation and can propose a +PlanDelta. It purposely avoids heavy dependencies and is deterministic +for testability. +""" +from dataclasses import dataclass +from typing import Dict, Any + +from ..core import LocalProblem, PlanDelta + + +@dataclass +class RoverAdapter: + name: str = "rover_planner" + + def to_canonical(self, lp: LocalProblem) -> Dict[str, Any]: + """Map a rover LocalProblem to a tiny canonical schema. + + The canonical mapping includes an adapter tag, core identifiers, and + a simple summary of assets and objective to be used by downstream + components. + """ + return { + "adapter": self.name, + "id": lp.id, + "domain": lp.domain, + "assets_summary": list(lp.assets.keys()), + "objective": lp.objective, + } + + def propose_delta(self, lp: LocalProblem, author: str = "rover") -> PlanDelta: + """Produce a deterministic PlanDelta for tests and demos. + + This toy implementation produces a delta that instructs a rover to + hold position if a "battery" asset exists and its value is low. + """ + delta = {} + battery = lp.assets.get("battery") + if isinstance(battery, (int, float)) and battery < 20: + delta["action"] = "hold_position" + delta["reason"] = "low_battery" + else: + delta["action"] = "proceed" + return PlanDelta(delta=delta, author=author, contract_id=lp.id) diff --git a/catopt_studio/registry.py b/catopt_studio/registry.py new file mode 100644 index 0000000..154895f --- /dev/null +++ b/catopt_studio/registry.py @@ -0,0 +1,55 @@ +"""In-memory registry and minimal API for CatOpt Studio MVP. + +This module provides a tiny, process-local registry suitable for tests and a +conformance harness. It exposes the minimal operations requested in the idea +description: register_contract, push_signal, propose_delta, get_provenance. +""" +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from typing import Dict, Any, List, Optional + + +@dataclass +class RegistryEntry: + contract_id: str + adapter: str + schema: Dict[str, Any] + created: datetime = field(default_factory=datetime.utcnow) + + +class InMemoryRegistry: + def __init__(self): + self._contracts: Dict[str, RegistryEntry] = {} + self._signals: List[Dict[str, Any]] = [] + self._deltas: List[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) + self._contracts[contract_id] = entry + return entry + + def push_signal(self, contract_id: str, signal: Dict[str, Any]) -> None: + record = {"contract_id": contract_id, "signal": signal, "timestamp": datetime.utcnow()} + self._signals.append(record) + + def propose_delta(self, contract_id: str, delta: Dict[str, Any], author: Optional[str] = None) -> None: + record = {"contract_id": contract_id, "delta": delta, "author": author or "anonymous", "timestamp": datetime.utcnow()} + self._deltas.append(record) + + def get_provenance(self, contract_id: str) -> Dict[str, Any]: + # Return a tiny provenance summary for the given contract + signals = [s for s in self._signals if s["contract_id"] == contract_id] + deltas = [d for d in self._deltas if d["contract_id"] == contract_id] + entry = self._contracts.get(contract_id) + return { + "contract_id": contract_id, + "registered": entry.created if entry else None, + "signals_count": len(signals), + "deltas_count": len(deltas), + } + + +# Module-level default registry for convenience in tests and examples +default_registry = InMemoryRegistry() diff --git a/tests/test_adapters.py b/tests/test_adapters.py new file mode 100644 index 0000000..bd35392 --- /dev/null +++ b/tests/test_adapters.py @@ -0,0 +1,27 @@ +from catopt_studio import LocalProblem +from catopt_studio.adapters import RoverAdapter, HabitatAdapter + + +def test_rover_adapter_propose_delta_low_battery(): + lp = LocalProblem(id="r1", domain="rover", assets={"battery": 10}, objective="navigate") + ra = RoverAdapter() + delta = ra.propose_delta(lp, author="tester") + assert delta.author == "tester" + assert delta.contract_id == "r1" + assert delta.delta.get("action") == "hold_position" + + +def test_rover_to_canonical_contains_expected_keys(): + lp = LocalProblem(id="r2", domain="rover", assets={"wheel": 2}, objective="survey") + ra = RoverAdapter() + canon = ra.to_canonical(lp) + assert canon["adapter"] == "rover_planner" + assert canon["id"] == "r2" + + +def test_habitat_adapter_respects_no_move_constraint(): + lp = LocalProblem(id="h1", domain="habitat", assets={}, constraints=["no_move_zone"], objective="manage") + ha = HabitatAdapter() + delta = ha.propose_delta(lp, author="hub") + assert delta.delta.get("action") == "pause" + assert delta.author == "hub" diff --git a/tests/test_registry.py b/tests/test_registry.py new file mode 100644 index 0000000..3771e34 --- /dev/null +++ b/tests/test_registry.py @@ -0,0 +1,12 @@ +from catopt_studio.registry import InMemoryRegistry + + +def test_register_and_provenance(): + reg = InMemoryRegistry() + reg.register_contract("c1", "rover_planner", {"schema": "v1"}) + reg.push_signal("c1", {"s": 1}) + reg.propose_delta("c1", {"d": 2}, author="alice") + prov = reg.get_provenance("c1") + assert prov["contract_id"] == "c1" + assert prov["signals_count"] == 1 + assert prov["deltas_count"] == 1