diff --git a/README.md b/README.md index afdacd7..52b3df4 100644 --- a/README.md +++ b/README.md @@ -1,23 +1,26 @@ # CatOpt-Swarm -Safe, verifiable distributed optimization for robotic swarms. +CatOpt-Swarm is a validated Python package for safe distributed swarm optimization. +It models robot tasks, contract-tagged deltas, and mission policies in a canonical IR, +then runs a deterministic ADMM-lite coordinator with verification hooks. -CatOpt-Swarm models swarm coordination as a small validated IR: +## Core Concepts -- `LocalProblem` captures each robot's local objective and state. -- `PlanDelta` carries deterministic, contract-tagged updates. -- `SharedVariables` stores consensus state and versioning. -- `SafetyPolicy` enforces travel, separation, and energy limits. -- `ContractRegistry` records adapter and mission contracts as a graph. +- `LocalProblem` captures each robot's objective, state, and local limits. +- `PlanDelta` carries contract-tagged updates with sequence and base-version replay guards. +- `SharedVariables` tracks consensus, versioning, and applied sequence state. +- `SafetyPolicy` enforces travel distance, separation, and energy limits. +- `ContractRegistry` stores adapter contracts in a graph and supports conformance checks. -The solver is a deterministic ADMM-lite loop with: +## Behavior -- bounded-step local updates -- delta reconciliation with bounded staleness -- separation projection for collision avoidance -- audit logs and convergence certificates +- Deterministic delta ordering and bounded-staleness replay control +- Separation projection to keep robots apart during updates +- Audit logs and convergence certificates per solve +- Verification of consensus consistency, energy budgets, and trace continuity +- Adapter shims for aerial and ground controllers -## Package Layout +## Layout - `catopt_swarm.models` - `catopt_swarm.registry` @@ -38,10 +41,11 @@ python3 -m pip install -e . python3 -m catopt_swarm.cli ``` -Or use the compatibility script: +Compatibility entrypoints: ```bash python3 admm_solver.py +python3 solver.py ``` ## Test @@ -50,13 +54,15 @@ python3 admm_solver.py bash test.sh ``` -## What the demo covers +## Demo Coverage - Two-robot consensus planning - Adapter conformance checks for aerial and ground controllers - Graph-based contract registration - Safety verification after solve +- Delta replay and version tracking -## Notes +## Status -This repository currently focuses on the core planning and verification substrate. The next extension point is integrating ROS/Gazebo-backed mission replay and richer platform adapters. +This repository now covers the core planning, contract, and verification substrate. +ROS/Gazebo integration and richer mission templates remain natural next steps. diff --git a/catopt_swarm/adapters.py b/catopt_swarm/adapters.py index d9c7fd6..8b0b92a 100644 --- a/catopt_swarm/adapters.py +++ b/catopt_swarm/adapters.py @@ -18,13 +18,15 @@ class DroneControllerAdapter: defaults = _AdapterDefaults(max_step=12.0, energy_budget=80.0) def to_local_problem(self, robot_id: str, mission: dict[str, Any]) -> LocalProblem: + metadata = dict(mission.get("metadata", {})) + metadata["domain"] = self.domain return LocalProblem( robot_id=robot_id, target_pos=float(mission["target_pos"]), initial_pos=float(mission.get("initial_pos", 0.0)), max_step=float(mission.get("max_step", self.defaults.max_step)), energy_budget=float(mission.get("energy_budget", self.defaults.energy_budget)), - metadata={"domain": self.domain, **mission.get("metadata", {})}, + metadata=metadata, ) def to_plan_delta( @@ -50,13 +52,15 @@ class GroundRoverControllerAdapter: defaults = _AdapterDefaults(max_step=8.0, energy_budget=120.0) def to_local_problem(self, robot_id: str, mission: dict[str, Any]) -> LocalProblem: + metadata = dict(mission.get("metadata", {})) + metadata["domain"] = self.domain return LocalProblem( robot_id=robot_id, target_pos=float(mission["target_pos"]), initial_pos=float(mission.get("initial_pos", 0.0)), max_step=float(mission.get("max_step", self.defaults.max_step)), energy_budget=float(mission.get("energy_budget", self.defaults.energy_budget)), - metadata={"domain": self.domain, **mission.get("metadata", {})}, + metadata=metadata, ) def to_plan_delta( diff --git a/catopt_swarm/models.py b/catopt_swarm/models.py index 94fc879..95445bb 100644 --- a/catopt_swarm/models.py +++ b/catopt_swarm/models.py @@ -46,6 +46,7 @@ class SharedVariables(BaseModel): version: int = 0 global_consensus: float = 0.0 delta_clock: int = 0 + last_applied_sequences: dict[str, int] = Field(default_factory=dict) class DualVariables(BaseModel): @@ -56,6 +57,7 @@ class PlanDelta(BaseModel): author: str contract_id: str sequence: int + base_version: int = 0 timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) target_robot_id: str | None = None updates: dict[str, Any] = Field(default_factory=dict) @@ -67,6 +69,13 @@ class PlanDelta(BaseModel): raise ValueError("sequence must be non-negative") return value + @field_validator("base_version") + @classmethod + def _base_version_non_negative(cls, value: int) -> int: + if value < 0: + raise ValueError("base_version must be non-negative") + return value + class SafetyPolicy(BaseModel): max_travel_distance: float = 15.0 diff --git a/catopt_swarm/registry.py b/catopt_swarm/registry.py index d551b20..13da7e6 100644 --- a/catopt_swarm/registry.py +++ b/catopt_swarm/registry.py @@ -29,6 +29,8 @@ class ContractRegistry: self._graph = nx.DiGraph() def register(self, spec: ContractSpec) -> None: + if self.has_contract(spec.contract_id): + raise ValueError(f"contract already registered: {spec.contract_id}") self._graph.add_node(spec.contract_id, spec=spec) def link(self, parent_contract_id: str, child_contract_id: str, relation: str = "extends") -> None: @@ -42,7 +44,7 @@ class ContractRegistry: def summary(self) -> dict[str, Any]: return { - "contracts": list(self._graph.nodes), + "contracts": [self.get(contract_id).model_dump() for contract_id in self._graph.nodes], "links": [ {"from": source, "to": target, **data} for source, target, data in self._graph.edges(data=True) @@ -56,6 +58,9 @@ def check_adapter_conformance(adapter: Any, spec: ContractSpec) -> AdapterConfor issues.append(f"adapter name mismatch: expected {spec.adapter_name}") if getattr(adapter, "domain", None) != spec.domain: issues.append(f"domain mismatch: expected {spec.domain}") + for method_name in ("to_local_problem", "to_plan_delta"): + if not callable(getattr(adapter, method_name, None)): + issues.append(f"missing adapter method: {method_name}") return AdapterConformanceReport( adapter_name=getattr(adapter, "name", adapter.__class__.__name__), contract_id=spec.contract_id, diff --git a/catopt_swarm/solver.py b/catopt_swarm/solver.py index 7f02425..bd2321f 100644 --- a/catopt_swarm/solver.py +++ b/catopt_swarm/solver.py @@ -13,6 +13,8 @@ from .models import ( SwarmSolution, ) from .verification import verify_swarm_solution + + class ADMMSolver: def __init__(self, rho: float = 1.0, max_iter: int = 50, epsilon: float = 1e-3, staleness_bound: int = 2): self.rho = rho @@ -35,12 +37,19 @@ class ADMMSolver: applied: list[str] = [] indexed = {robot.robot_id: robot for robot in robots} for delta in self._ordered_deltas(deltas): - if shared.version - delta.sequence > self.staleness_bound: + if delta.base_version > shared.version: + continue + if shared.version - delta.base_version > self.staleness_bound: continue if delta.target_robot_id is None or delta.target_robot_id not in indexed: continue + last_sequence = shared.last_applied_sequences.get(delta.target_robot_id, -1) + if delta.sequence <= last_sequence: + continue indexed[delta.target_robot_id].apply_updates(delta.updates) indexed[delta.target_robot_id].version = max(indexed[delta.target_robot_id].version, delta.sequence) + shared.last_applied_sequences[delta.target_robot_id] = delta.sequence + shared.version = max(shared.version, delta.base_version, delta.sequence) applied.append(f"{delta.contract_id}:{delta.author}:{delta.sequence}") if applied: shared.delta_clock += len(applied) @@ -53,14 +62,26 @@ class ADMMSolver: half_span = minimum * (len(candidates) - 1) / 2.0 return [center - half_span + index * minimum for index in range(len(candidates))] + def _current_position(self, robot: LocalProblem) -> float: + current_pos = robot.current_pos + if current_pos is None: + current_pos = robot.initial_pos + return float(current_pos) + def solve( self, robots: Sequence[LocalProblem], safety_policy: SafetyPolicy, incoming_deltas: Iterable[PlanDelta] = (), ) -> SwarmSolution: + if not robots: + raise ValueError("at least one robot is required") + working = [robot.model_copy(deep=True) for robot in robots] - shared = SharedVariables(version=0, global_consensus=fsum(robot.target_pos for robot in working) / len(working)) + shared = SharedVariables( + version=max(robot.version for robot in working), + global_consensus=fsum(self._current_position(robot) for robot in working) / len(working), + ) audit_log: list[AuditLogEntry] = [] applied_deltas = self._apply_deltas(working, shared, incoming_deltas) @@ -71,18 +92,22 @@ class ADMMSolver: for iteration in range(self.max_iter): old_consensus = shared.global_consensus candidate_positions: list[float] = [] - previous_positions = [robot.current_pos for robot in working] + previous_positions: list[float] = [] for robot in working: - raw_position = (2.0 * robot.target_pos + self.rho * (shared.global_consensus - robot.dual_var)) / (2.0 + self.rho) - step = max(-robot.max_step, min(robot.max_step, raw_position - robot.current_pos)) - candidate_positions.append(robot.current_pos + step) + current_pos = self._current_position(robot) + previous_positions.append(current_pos) + raw_position = ( + 2.0 * robot.target_pos + self.rho * (shared.global_consensus - robot.dual_var) + ) / (2.0 + self.rho) + step = max(-robot.max_step, min(robot.max_step, raw_position - current_pos)) + candidate_positions.append(current_pos + step) projected_positions = self._project_for_separation(candidate_positions, safety_policy.min_separation) for robot, new_position in zip(working, projected_positions): robot.current_pos = new_position - if not safety_policy.verify(robot.initial_pos, robot.current_pos): + if not safety_policy.verify(robot.initial_pos, new_position): raise ValueError(f"Safety violation for {robot.robot_id}: travel bound exceeded") shared.global_consensus = fsum(projected_positions) / len(projected_positions) @@ -90,8 +115,9 @@ class ADMMSolver: final_primal = 0.0 for robot, previous_position in zip(working, previous_positions): - robot.dual_var += 0.5 * (robot.current_pos - shared.global_consensus) - final_primal = max(final_primal, abs(robot.current_pos - previous_position)) + current_pos = self._current_position(robot) + robot.dual_var += 0.5 * (current_pos - shared.global_consensus) + final_primal = max(final_primal, abs(current_pos - previous_position)) final_dual = abs(shared.global_consensus - old_consensus) * self.rho audit_log.append( diff --git a/catopt_swarm/verification.py b/catopt_swarm/verification.py index 5789c99..8d3477d 100644 --- a/catopt_swarm/verification.py +++ b/catopt_swarm/verification.py @@ -1,6 +1,7 @@ from __future__ import annotations from itertools import combinations +from typing import Any from pydantic import BaseModel, Field @@ -12,18 +13,45 @@ class VerificationReport(BaseModel): violations: list[str] = Field(default_factory=list) +def _current_position(robot: Any) -> float: + current_pos = robot.current_pos + if current_pos is None: + current_pos = robot.initial_pos + assert current_pos is not None + return float(current_pos) + + def verify_swarm_solution(solution: SwarmSolution, policy: SafetyPolicy) -> VerificationReport: violations: list[str] = [] robots = solution.robots + if solution.certificate.iterations != len(solution.audit_log): + violations.append("certificate iteration count does not match audit log") + + expected_version = solution.shared.version + if not solution.audit_log: + violations.append("audit log is empty") + else: + for expected_iteration, entry in enumerate(solution.audit_log): + if entry.iteration != expected_iteration: + violations.append("audit log iterations are not contiguous") + break + if solution.audit_log[-1].shared_version != expected_version: + violations.append("final shared version does not match audit log") + + average_position = sum(_current_position(robot) for robot in robots) / len(robots) if robots else 0.0 + if abs(solution.consensus - average_position) > policy.max_travel_distance: + violations.append("consensus diverged from robot state") + for robot in robots: - if not policy.verify(robot.initial_pos, robot.current_pos): + current_pos = _current_position(robot) + if not policy.verify(robot.initial_pos, current_pos): violations.append(f"{robot.robot_id}: travel bound exceeded") if robot.travel_distance > policy.energy_budget: violations.append(f"{robot.robot_id}: energy budget exceeded") for left, right in combinations(robots, 2): - if abs(left.current_pos - right.current_pos) < policy.min_separation: + if abs(_current_position(left) - _current_position(right)) < policy.min_separation: violations.append( f"{left.robot_id}/{right.robot_id}: separation below {policy.min_separation}" ) @@ -31,4 +59,7 @@ def verify_swarm_solution(solution: SwarmSolution, policy: SafetyPolicy) -> Veri if not solution.certificate.converged: violations.append("solver did not converge") + if solution.audit_log and solution.shared.delta_clock < len(solution.audit_log[0].applied_deltas): + violations.append("delta clock regressed") + return VerificationReport(passed=not violations, violations=violations) diff --git a/tests/test_catopt_swarm.py b/tests/test_catopt_swarm.py index d90b1d5..40b4b67 100644 --- a/tests/test_catopt_swarm.py +++ b/tests/test_catopt_swarm.py @@ -5,6 +5,7 @@ from catopt_swarm import ( DroneControllerAdapter, GroundRoverControllerAdapter, LocalProblem, + PlanDelta, SafetyPolicy, verify_swarm_solution, ) @@ -49,3 +50,64 @@ def test_rover_adapter_maps_mission_to_problem(): assert problem.robot_id == "rover-1" assert problem.target_pos == 3.5 assert problem.initial_pos == 1.0 + + +def test_solver_applies_delta_and_tracks_replay_state(): + robots = [ + LocalProblem(robot_id="drone-1", target_pos=0.0, current_pos=0.0), + LocalProblem(robot_id="drone-2", target_pos=10.0, current_pos=10.0), + ] + policy = SafetyPolicy(max_travel_distance=15.0, min_separation=1.0, energy_budget=20.0) + delta = PlanDelta( + author="mission-orchestrator", + contract_id="patrol-update", + sequence=1, + base_version=0, + target_robot_id="drone-2", + updates={"target_pos": 8.0}, + ) + + solution = ADMMSolver(rho=1.5, max_iter=40, epsilon=1e-4).solve(robots, policy, [delta]) + + assert solution.audit_log[0].applied_deltas == ["patrol-update:mission-orchestrator:1"] + assert solution.shared.last_applied_sequences["drone-2"] == 1 + + +def test_registry_rejects_duplicate_contracts(): + registry = ContractRegistry() + spec = ContractSpec( + contract_id="drone-patrol-v1", + adapter_name="drone-controller", + domain="aerial", + version="1.0.0", + ) + + registry.register(spec) + + try: + registry.register(spec) + except ValueError as exc: + assert "already registered" in str(exc) + else: + raise AssertionError("duplicate contract registration should fail") + + +def test_adapter_conformance_flags_missing_methods(): + spec = ContractSpec( + contract_id="drone-patrol-v1", + adapter_name="drone-controller", + domain="aerial", + version="1.0.0", + ) + + class IncompleteAdapter: + name = "drone-controller" + domain = "aerial" + + def to_local_problem(self, *args, **kwargs): + raise NotImplementedError + + report = check_adapter_conformance(IncompleteAdapter(), spec) + + assert not report.passed + assert any("missing adapter method" in issue for issue in report.issues)