"""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, }