diff --git a/AGENTS.md b/AGENTS.md index efba6c5..8d77b56 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,6 +4,7 @@ Architecture - Language: Python 3.8+ - Minimal package: gravityweave - Core modules: registry, plan_delta (CRDT), ledger, adapters, solver +- Additional coordination modules: delta_synopsis, dtn, mission simulator Tech stack - Standard library (hashlib, hmac, json, time) diff --git a/README.md b/README.md index 71d419b..43ab637 100644 --- a/README.md +++ b/README.md @@ -1,15 +1,33 @@ -# GravityWeave (idea182-implementation) +# GravityWeave -Minimal GravityWeave skeleton implementing core primitives for a federated, -delay-tolerant mission optimization stack. This repo provides: +GravityWeave is a small Python package for federated orbital mission planning. +It focuses on deterministic coordination across delayed links, with compact +synopses, tamper-evident logging, and replay-friendly plan deltas. -- A Graph-of-Contracts registry (gravityweave.registry) -- A small CRDT-style PlanDelta implementation (ORSet, LWWRegister) -- Toy adapters: sat_planner and relay_module -- A minimal governance ledger with tamper-evident chaining (HMAC-signed) -- An ADMM-lite solver stub for local updates +## What Is Included -This is intentionally a small, well-tested starting point for the full -GravityWeave architecture described in the original idea. +- `gravityweave.registry`: Graph-of-Contracts registry for adapters and schemas +- `gravityweave.plan_delta`: CRDT-style `ORSet`, `LWWRegister`, and `PlanDelta` +- `gravityweave.delta_synopsis`: compact synopses, priority keys, and signature checks +- `gravityweave.dtn`: custody headers and a DTN acceptance heuristic +- `gravityweave.ledger`: append-only governance ledger with chained HMACs +- `gravityweave.adapters`: starter `sat_planner` and `relay_module` adapters +- `gravityweave.solver`: an ADMM-lite local update stub +- `gravityweave.mission`: deterministic mission simulator that ties everything together -Run tests: `bash test.sh` +## Example + +```python +from gravityweave import ContactWindow, MissionNode, MissionSimulator, MissionTask + +nodes = [MissionNode("sat1", "satellite"), MissionNode("relay1", "relay")] +tasks = [MissionTask("task-a", "sat1", "downlink", utility=9.0, deadline_step=0, safety_critical=True)] +windows = [ContactWindow("sat1", "relay1", step=0, bandwidth_bytes=1024)] + +sim = MissionSimulator(nodes, tasks, windows, ledger_key=b"ledger", synopsis_key=b"syn") +metrics = sim.run() +``` + +## Test And Build + +Run `bash test.sh` to build the package and execute the test suite. diff --git a/gravityweave/__init__.py b/gravityweave/__init__.py index 528d868..0417b04 100644 --- a/gravityweave/__init__.py +++ b/gravityweave/__init__.py @@ -11,7 +11,8 @@ from .plan_delta import PlanDelta, ORSet, LWWRegister from .ledger import GovernanceLedger from .adapters import sat_planner, relay_module from .solver import ADMMNode -from .delta_synopsis import build_synopsis, build_priority_key, serialize_synopsis, deserialize_synopsis +from .delta_synopsis import build_synopsis, build_priority_key, serialize_synopsis, deserialize_synopsis, verify_synopsis +from .mission import MissionTask, ContactWindow, MissionNode, MissionSimulator __all__ = [ "Registry", @@ -26,4 +27,9 @@ __all__ = [ "build_priority_key", "serialize_synopsis", "deserialize_synopsis", + "verify_synopsis", + "MissionTask", + "ContactWindow", + "MissionNode", + "MissionSimulator", ] diff --git a/gravityweave/delta_synopsis.py b/gravityweave/delta_synopsis.py index 8891585..e3396ba 100644 --- a/gravityweave/delta_synopsis.py +++ b/gravityweave/delta_synopsis.py @@ -4,7 +4,7 @@ Creates compact synopses for PlanDelta objects used by DTN custody holders to decide whether to fetch full payloads during short windows. The synopsis is kept intentionally small and includes a short HMAC-based signature. """ -from typing import Dict, Any, List +from typing import Dict, Any, List, Optional import json import hashlib import hmac @@ -31,7 +31,8 @@ def build_priority_key(urgency_score: int, impact_score: int) -> int: def build_synopsis(delta, parent_version: str = "", urgency_score: int = 0, impact_score: int = 0, - affected_assets: List[str] = None, signer: str = "", key: bytes = b"") -> Dict[str, Any]: + affected_assets: Optional[List[str]] = None, signer: str = "", key: bytes = b"", + delta_id: Optional[str] = None) -> Dict[str, Any]: """Build a compact synopsis for a PlanDelta. - delta: PlanDelta instance @@ -45,7 +46,7 @@ def build_synopsis(delta, parent_version: str = "", urgency_score: int = 0, impa summary_hash = hashlib.sha256(payload).hexdigest() size_estimate = len(payload) priority_key = build_priority_key(urgency_score, impact_score) - delta_id = uuid.uuid4().hex + delta_id = delta_id or uuid.uuid4().hex synopsis = { "delta_id": delta_id, @@ -73,3 +74,13 @@ def serialize_synopsis(syn: Dict[str, Any]) -> bytes: def deserialize_synopsis(b: bytes) -> Dict[str, Any]: return json.loads(b.decode("utf-8")) + + +def verify_synopsis(syn: Dict[str, Any], key: bytes = b"") -> bool: + summary_hash = syn.get("summary_hash", "") + if not summary_hash: + return False + if key: + expected = hmac.new(key, summary_hash.encode("utf-8"), hashlib.sha256).hexdigest() + return hmac.compare_digest(expected, syn.get("short_sig", "")) + return syn.get("short_sig", "") == "" diff --git a/gravityweave/dtn.py b/gravityweave/dtn.py index 55621c7..121dca0 100644 --- a/gravityweave/dtn.py +++ b/gravityweave/dtn.py @@ -8,7 +8,7 @@ on priority, remaining lifetime, and a node reputation score. The implementation is intentionally small and deterministic for unit testing. """ from dataclasses import dataclass, field -from typing import List +from typing import List, Optional import time @@ -20,7 +20,7 @@ class CustodyHeader: custody_chain: List[str] = field(default_factory=list) priority_key: int = 0 # combined urgency/impact key (0..127) - def is_expired(self, now: float = None) -> bool: + def is_expired(self, now: Optional[float] = None) -> bool: if now is None: now = time.time() return now > (self.created_at + self.lifetime_secs) @@ -30,7 +30,8 @@ class CustodyHeader: self.custody_chain.append(node_id) -def should_accept_custody(header: CustodyHeader, node_reputation: float, min_reputation: float = 0.2) -> bool: +def should_accept_custody(header: CustodyHeader, node_reputation: float, min_reputation: float = 0.2, + now: Optional[float] = None) -> bool: """Decide whether a node should accept custody for a delta. Decision factors (simple heuristic): @@ -40,7 +41,7 @@ def should_accept_custody(header: CustodyHeader, node_reputation: float, min_rep Returns True if the node should accept custody. """ - if header.is_expired(): + if header.is_expired(now=now): return False # Enforce a minimum reputation gate: nodes below min_reputation refuse custody diff --git a/gravityweave/ledger.py b/gravityweave/ledger.py index 2b385e8..8f15e19 100644 --- a/gravityweave/ledger.py +++ b/gravityweave/ledger.py @@ -8,7 +8,7 @@ import hashlib import hmac import json import time -from typing import List, Dict, Any +from typing import List, Dict, Any, Optional class GovernanceLedger: @@ -23,13 +23,13 @@ class GovernanceLedger: def _sign(self, payload: str) -> str: return hmac.new(self.node_key, payload.encode("utf-8"), hashlib.sha256).hexdigest() - def append(self, author: str, payload: Dict[str, Any]) -> Dict[str, Any]: + def append(self, author: str, payload: Dict[str, Any], ts: Optional[float] = None) -> Dict[str, Any]: prev_hash = self._chain[-1]["hash"] if self._chain else "" entry = { "author": author, "payload": payload, "prev_hash": prev_hash, - "ts": time.time(), + "ts": time.time() if ts is None else ts, } entry_hash = self._entry_hash(entry) signature = self._sign(entry_hash) diff --git a/gravityweave/mission.py b/gravityweave/mission.py new file mode 100644 index 0000000..9c78e6e --- /dev/null +++ b/gravityweave/mission.py @@ -0,0 +1,240 @@ +"""Deterministic mission coordination simulator. + +This module wires the package primitives together into a small but useful +coordination loop: +- local mission tasks become PlanDelta commitments +- contact windows carry synopses before full payloads +- custody acceptance is gated by the DTN policy helper +- every accepted transfer is anchored into the governance ledger + +The implementation is intentionally deterministic so tests can replay the same +scenario and compare outcomes byte-for-byte. +""" + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional +import hashlib +import json + +from .delta_synopsis import build_synopsis, verify_synopsis +from .dtn import CustodyHeader, should_accept_custody +from .ledger import GovernanceLedger +from .plan_delta import PlanDelta +from .adapters.sat_planner import create as create_sat_planner +from .adapters.relay_module import create as create_relay_module + + +@dataclass(frozen=True) +class MissionTask: + task_id: str + asset_id: str + kind: str + utility: float + deadline_step: int + safety_critical: bool = False + fuel_cost: float = 0.0 + power_cost: float = 0.0 + payload_bytes: int = 256 + + +@dataclass(frozen=True) +class ContactWindow: + sender_id: str + receiver_id: str + step: int + bandwidth_bytes: int + + +@dataclass +class MissionNode: + node_id: str + role: str + reputation: float = 0.9 + fuel: float = 100.0 + power: float = 100.0 + adapter: Optional[Any] = None + completed_tasks: List[str] = field(default_factory=list) + + def __post_init__(self) -> None: + if self.adapter is None: + if self.role == "relay": + self.adapter = create_relay_module(self.node_id) + else: + self.adapter = create_sat_planner(self.node_id) + + def build_delta(self, task: MissionTask, step: int) -> PlanDelta: + delta = PlanDelta(origin=self.node_id) + commitment = { + "task_id": task.task_id, + "asset_id": task.asset_id, + "type": task.kind, + "fuel": task.fuel_cost, + "power": task.power_cost, + "utility": task.utility, + "payload_bytes": task.payload_bytes, + "deadline_step": task.deadline_step, + "safety_critical": task.safety_critical, + "step": step, + } + delta.commitments.add(commitment, ts=float(step)) + delta.priority.write(task.utility, ts=float(step)) + return delta + + def apply_delta(self, delta: PlanDelta) -> bool: + accepted = True + if self.adapter is not None and hasattr(self.adapter, "apply_delta"): + accepted = bool(self.adapter.apply_delta(delta)) + if not accepted: + return False + + for commitment in delta.commitments.value(): + if not isinstance(commitment, dict): + continue + self.fuel = max(0.0, self.fuel - float(commitment.get("fuel", 0.0))) + self.power = max(0.0, self.power - float(commitment.get("power", 0.0))) + task_id = commitment.get("task_id") + if task_id and task_id not in self.completed_tasks: + self.completed_tasks.append(task_id) + return True + + +def _priority_scores(task: MissionTask, step: int) -> Dict[str, int]: + urgency = 7 if task.safety_critical else 0 + if task.deadline_step <= step: + urgency = 7 + elif task.deadline_step - step <= 1: + urgency = max(urgency, 6) + elif task.deadline_step - step <= 3: + urgency = max(urgency, 4) + + impact = min(15, max(0, int(round(task.utility)))) + return {"urgency": urgency, "impact": impact} + + +def _payload_size(delta: PlanDelta) -> int: + raw = json.dumps(delta.snapshot(), sort_keys=True, separators=(",", ":")).encode("utf-8") + return len(raw) + + +class MissionSimulator: + def __init__(self, nodes: List[MissionNode], tasks: List[MissionTask], windows: List[ContactWindow], + ledger_key: bytes, synopsis_key: bytes = b""): + self.nodes = {node.node_id: node for node in nodes} + self.tasks = list(tasks) + self.windows = sorted(windows, key=lambda window: (window.step, window.sender_id, window.receiver_id)) + self.synopsis_key = synopsis_key + self.ledger = GovernanceLedger(ledger_key) + self._task_queue: Dict[str, List[MissionTask]] = {node.node_id: [] for node in nodes} + for task in tasks: + self._task_queue.setdefault(task.asset_id, []).append(task) + for queue in self._task_queue.values(): + queue.sort(key=lambda task: (task.deadline_step, -task.utility, task.task_id)) + + def _next_task(self, node_id: str, step: int, exclude: Optional[set] = None) -> Optional[MissionTask]: + queue = self._task_queue.get(node_id, []) + excluded = exclude or set() + candidates = [task for task in queue if task.task_id not in self.nodes[node_id].completed_tasks and task.task_id not in excluded and task.deadline_step >= step] + if not candidates: + return None + return max(candidates, key=lambda task: (task.safety_critical, -task.deadline_step, task.utility, task.task_id)) + + def run(self) -> Dict[str, Any]: + if not self.tasks: + return { + "tasks_total": 0, + "tasks_completed": 0, + "safety_critical_completed": 0, + "synopsis_bytes": 0, + "payload_bytes": 0, + "average_latency": 0.0, + "replay_digest": hashlib.sha256(b"empty").hexdigest(), + } + + summary: List[Dict[str, Any]] = [] + completed_at: Dict[str, int] = {} + synopsis_bytes = 0 + payload_bytes = 0 + + max_step = max([task.deadline_step for task in self.tasks] + [window.step for window in self.windows]) + windows_by_step: Dict[int, List[ContactWindow]] = {} + for window in self.windows: + windows_by_step.setdefault(window.step, []).append(window) + + for step in range(max_step + 1): + step_assigned = set() + for window in windows_by_step.get(step, []): + sender = self.nodes[window.sender_id] + receiver = self.nodes[window.receiver_id] + task = self._next_task(sender.node_id, step, exclude=step_assigned) + if task is None: + continue + step_assigned.add(task.task_id) + + delta = sender.build_delta(task, step) + scores = _priority_scores(task, step) + synopsis = build_synopsis( + delta, + parent_version=f"step-{step}", + urgency_score=scores["urgency"], + impact_score=scores["impact"], + affected_assets=[task.asset_id], + signer=sender.node_id, + key=self.synopsis_key, + delta_id=task.task_id, + ) + synopsis_bytes += len(json.dumps(synopsis, sort_keys=True, separators=(",", ":")).encode("utf-8")) + + header = CustodyHeader( + delta_id=synopsis["delta_id"], + created_at=float(step), + lifetime_secs=max(1, task.deadline_step - step + 5), + priority_key=synopsis["priority_key"], + ) + if not should_accept_custody(header, node_reputation=receiver.reputation, now=float(step)): + continue + if not verify_synopsis(synopsis, key=self.synopsis_key): + continue + + payload_size = _payload_size(delta) + if payload_size > window.bandwidth_bytes: + continue + + if not receiver.apply_delta(delta): + continue + + payload_bytes += payload_size + sender.completed_tasks.append(task.task_id) + completed_at[task.task_id] = step + entry = self.ledger.append( + sender.node_id, + { + "task_id": task.task_id, + "sender": sender.node_id, + "receiver": receiver.node_id, + "priority_key": synopsis["priority_key"], + "payload_bytes": payload_size, + }, + ts=float(step), + ) + summary.append(entry) + + tasks_completed = len(completed_at) + safety_critical_completed = sum(1 for task in self.tasks if task.safety_critical and task.task_id in completed_at) + average_latency = 0.0 + if completed_at: + total_latency = sum(max(0, completed_at[task.task_id] - task.deadline_step) for task in self.tasks if task.task_id in completed_at) + average_latency = total_latency / float(tasks_completed) + + replay_digest = hashlib.sha256( + json.dumps(summary, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + + return { + "tasks_total": len(self.tasks), + "tasks_completed": tasks_completed, + "safety_critical_completed": safety_critical_completed, + "synopsis_bytes": synopsis_bytes, + "payload_bytes": payload_bytes, + "average_latency": average_latency, + "replay_digest": replay_digest, + } diff --git a/gravityweave/plan_delta.py b/gravityweave/plan_delta.py index 5ec4d16..e7ed4bf 100644 --- a/gravityweave/plan_delta.py +++ b/gravityweave/plan_delta.py @@ -5,7 +5,7 @@ merges deterministic on reconnection. They are intentionally small but include merge operations and vector-clock-like timestamps. """ from dataclasses import dataclass, field -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional import time @@ -20,13 +20,13 @@ class ORSet: return _json.dumps(element, sort_keys=True, separators=(',', ':')) - def add(self, element: Any): + def add(self, element: Any, ts: Optional[float] = None): k = self._key(element) - self.add_set[k] = {"val": element, "ts": time.time()} + self.add_set[k] = {"val": element, "ts": time.time() if ts is None else ts} - def remove(self, element: Any): + def remove(self, element: Any, ts: Optional[float] = None): k = self._key(element) - self.rem_set[k] = {"val": element, "ts": time.time()} + self.rem_set[k] = {"val": element, "ts": time.time() if ts is None else ts} def value(self) -> List[Any]: res: List[Any] = [] @@ -65,9 +65,9 @@ class LWWRegister: value: Any = None ts: float = 0.0 - def write(self, value: Any): + def write(self, value: Any, ts: Optional[float] = None): self.value = value - self.ts = time.time() + self.ts = time.time() if ts is None else ts def read(self): return self.value @@ -92,3 +92,19 @@ class PlanDelta: # origin is not merged; keep latest priority's origin heuristically p.origin = other.origin if other.priority.ts >= self.priority.ts else self.origin return p + + def snapshot(self) -> Dict[str, Any]: + return { + "commitments": self.commitments.value(), + "priority": self.priority.read(), + "priority_ts": self.priority.ts, + "origin": self.origin, + } + + @classmethod + def from_snapshot(cls, snapshot: Dict[str, Any]) -> "PlanDelta": + delta = cls(origin=snapshot.get("origin", "")) + for commitment in snapshot.get("commitments", []): + delta.commitments.add(commitment, ts=0.0) + delta.priority = LWWRegister(value=snapshot.get("priority"), ts=snapshot.get("priority_ts", 0.0)) + return delta diff --git a/tests/test_mission.py b/tests/test_mission.py new file mode 100644 index 0000000..e6bbe28 --- /dev/null +++ b/tests/test_mission.py @@ -0,0 +1,60 @@ +from gravityweave import ContactWindow, MissionNode, MissionSimulator, MissionTask +from gravityweave.delta_synopsis import build_synopsis, verify_synopsis +from gravityweave.plan_delta import PlanDelta + + +def test_plan_delta_snapshot_round_trip(): + delta = PlanDelta(origin="sat1") + delta.commitments.add({"task_id": "t1", "type": "downlink"}, ts=2.0) + delta.priority.write(11, ts=3.0) + + restored = PlanDelta.from_snapshot(delta.snapshot()) + + assert restored.origin == "sat1" + assert restored.priority.read() == 11 + assert restored.priority.ts == 3.0 + assert restored.commitments.value() == [{"task_id": "t1", "type": "downlink"}] + + +def test_synopsis_verification_with_fixed_id(): + delta = PlanDelta(origin="sat1") + delta.commitments.add({"task_id": "t2", "type": "science"}, ts=1.0) + delta.priority.write(8, ts=1.0) + + syn = build_synopsis( + delta, + parent_version="v1", + urgency_score=3, + impact_score=8, + affected_assets=["sat1"], + signer="sat1", + key=b"secret", + delta_id="fixed-delta", + ) + + assert syn["delta_id"] == "fixed-delta" + assert verify_synopsis(syn, key=b"secret") + + +def test_mission_simulator_is_deterministic(): + nodes = [MissionNode("sat1", "satellite"), MissionNode("relay1", "relay")] + tasks = [ + MissionTask("task-a", "sat1", "downlink", utility=9.0, deadline_step=0, safety_critical=True, payload_bytes=256), + MissionTask("task-b", "sat1", "science", utility=4.0, deadline_step=1, payload_bytes=256), + ] + windows = [ + ContactWindow("sat1", "relay1", step=0, bandwidth_bytes=1024), + ContactWindow("sat1", "relay1", step=1, bandwidth_bytes=1024), + ] + + sim_one = MissionSimulator(nodes=nodes, tasks=tasks, windows=windows, ledger_key=b"ledger", synopsis_key=b"syn") + metrics_one = sim_one.run() + + nodes_again = [MissionNode("sat1", "satellite"), MissionNode("relay1", "relay")] + sim_two = MissionSimulator(nodes=nodes_again, tasks=tasks, windows=windows, ledger_key=b"ledger", synopsis_key=b"syn") + metrics_two = sim_two.run() + + assert metrics_one == metrics_two + assert metrics_one["tasks_completed"] == 2 + assert metrics_one["safety_critical_completed"] == 1 + assert sim_one.ledger.verify_chain()