From 4be4d536a8fdc9eedd1c622ed23e2b5a7329b235 Mon Sep 17 00:00:00 2001 From: agent-54de0bcc6a17828b Date: Fri, 24 Apr 2026 20:43:42 +0200 Subject: [PATCH] build(agent): semicolon#54de0b iteration --- AGENTS.md | 2 + README.md | 9 ++- crisisops/__init__.py | 5 ++ crisisops/cli.py | 15 +++-- crisisops/governance.py | 19 ++++-- crisisops/planner.py | 71 +++++++++++++++------- crisisops/replay.py | 129 ++++++++++++++++++++++++++++++++++++++++ tests/test_planner.py | 43 +++++++++++++- 8 files changed, 259 insertions(+), 34 deletions(-) create mode 100644 crisisops/replay.py diff --git a/AGENTS.md b/AGENTS.md index 999d279..d9442b6 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -3,6 +3,7 @@ Architecture overview - Core: crisisops/core.py defines the Graph-of-Contracts primitives: Object, Morphism, PlanDelta, DualVariables. - Planner: crisisops/planner.py implements a lightweight ADMM-like planner for MVP orchestration between domains. +- Replay: crisisops/replay.py provides deterministic snapshots, canonical JSON, and restore/replay helpers. - Adapters: crisisops/adapters_registry.py provides a registry; toy adapters live in crisisops/adapters/. - Governance: crisisops/governance.py offers a simple SQLite-backed governance ledger for auditability. - CLI: crisisops/cli.py exposes a tiny CLI for quick experiments. @@ -16,6 +17,7 @@ Tech stack Testing commands - Run tests: ./test.sh - Run planner demo: python -m crisisops.cli plan (or pytest tests/test_planner.py) +- Inspect a snapshot: python -m crisisops.cli snapshot Contribution rules - Implement small, composable features with clear interfaces. diff --git a/README.md b/README.md index 85f0135..b2e6d00 100644 --- a/README.md +++ b/README.md @@ -2,23 +2,26 @@ Overview - CrisisOps provides a Graph-of-Contracts (GoC) primitive model to coordinate crisis-response operations across domains (e.g., field logistics, shelters) with offline-first delta-sync and auditability. -- MVP demonstrates two domains with toy adapters and a minimal ADMM-like planner. It is designed to be production-ready in architecture, not just a proof-of-concept. +- The current implementation adds deterministic snapshots, replay helpers, and audit-ready plan hashes so field decisions can be replayed after an event. +- The planner now emits ordered dispatch actions with a lightweight utility score to make plan comparisons easier. What’s inside - Python package crisisops with core primitives, planner, adapters registry, and governance ledger. - Toy adapters: inventory_portal and gis_dispatch (under crisisops/adapters). -- CLI scaffold for quick experiments. -- Tests validating planner behavior. +- CLI scaffold for quick experiments and snapshot inspection. +- Tests validating planner behavior and deterministic replay. - Packaging metadata and test script for CI compliance. Getting started - Install: python3 -m build (from pyproject.toml) - Run tests: ./test.sh - Explore planner: python3 -m crisisops.cli plan +- Inspect a snapshot: python3 -m crisisops.cli snapshot Architecture links - Core primitives: crisisops.core - Planner: crisisops.planner +- Replay and snapshot utilities: crisisops.replay - Adapters registry and toy adapters: crisisops.adapters_registry, crisisops.adapters.* - Governance ledger: crisisops.governance diff --git a/crisisops/__init__.py b/crisisops/__init__.py index f12b0bf..49fbb7f 100644 --- a/crisisops/__init__.py +++ b/crisisops/__init__.py @@ -4,6 +4,7 @@ from .core import Object, Morphism, PlanDelta, DualVariables from .planner import ADMMPlanner from .adapters_registry import AdaptersRegistry from .governance import GovernanceLedger +from .replay import ReplayJournal, graph_snapshot, restore_graph, canonical_json __all__ = [ "Object", @@ -13,4 +14,8 @@ __all__ = [ "ADMMPlanner", "AdaptersRegistry", "GovernanceLedger", + "ReplayJournal", + "graph_snapshot", + "restore_graph", + "canonical_json", ] diff --git a/crisisops/cli.py b/crisisops/cli.py index aed308a..e3e9421 100644 --- a/crisisops/cli.py +++ b/crisisops/cli.py @@ -1,13 +1,13 @@ """Minimal CLI to exercise CrisisOps MVP.""" import argparse import json -import time from crisisops.planner import ADMMPlanner from crisisops.core import GraphOfContracts, Object, Morphism +from crisisops.replay import graph_snapshot -def build_sample_goC(): +def build_sample_goc(): # Simple two-object domains: field resources and shelters o1 = Object(id="fuel-truck-1", type="vehicle", attributes={"capacity": 1000}) o2 = Object(id="shelter-a", type="facility", attributes={"capacity": 500}) @@ -21,13 +21,20 @@ def main(): parser = argparse.ArgumentParser() sub = parser.add_subparsers(dest="cmd") sub.add_parser("plan", help="Run a toy planning pass on a sample GoC") + sub.add_parser("snapshot", help="Print a deterministic snapshot of the sample GoC") args = parser.parse_args() if args.cmd == "plan": - gof = build_sample_goC() + gof = build_sample_goc() planner = ADMMPlanner() plan = planner.optimize(gof) - print(json.dumps({"plan": plan.actions}, default=str, indent=2)) + snapshot = graph_snapshot(gof, plan=plan) + print(json.dumps({"plan": plan.actions, "snapshot_hash": snapshot["snapshot_hash"]}, default=str, indent=2)) + return 0 + if args.cmd == "snapshot": + gof = build_sample_goc() + snapshot = graph_snapshot(gof) + print(json.dumps(snapshot, default=str, indent=2)) return 0 print("No command provided. Use 'plan'.") return 1 diff --git a/crisisops/governance.py b/crisisops/governance.py index 75d754f..b5745ad 100644 --- a/crisisops/governance.py +++ b/crisisops/governance.py @@ -1,7 +1,9 @@ """A simple governance ledger using SQLite for auditability.""" import sqlite3 import time -from typing import List, Dict, Any +from typing import List, Dict, Any, cast + +from .replay import canonical_json class GovernanceLedger: @@ -25,15 +27,24 @@ class GovernanceLedger: ) self.conn.commit() - def log(self, action: str, actor: str, status: str, details: str | None = None) -> int: + def log(self, action: str, actor: str, status: str, details: Any | None = None) -> int: ts = time.time() + normalized_details = details + if isinstance(details, (dict, list)): + normalized_details = canonical_json(details) cur = self.conn.cursor() cur.execute( "INSERT INTO governance_logs (action, actor, timestamp, status, details) VALUES (?, ?, ?, ?, ?)", - (action, actor, ts, status, details), + (action, actor, ts, status, normalized_details), ) self.conn.commit() - return cur.lastrowid + return cast(int, cur.lastrowid) + + def log_plan(self, actor: str, status: str, plan_hash: str, details: Dict[str, Any] | None = None) -> int: + payload = {"plan_hash": plan_hash} + if details: + payload.update(details) + return self.log(action="plan", actor=actor, status=status, details=payload) def list_logs(self) -> List[Dict[str, Any]]: cur = self.conn.cursor() diff --git a/crisisops/planner.py b/crisisops/planner.py index ccb02a2..77b9ca7 100644 --- a/crisisops/planner.py +++ b/crisisops/planner.py @@ -7,7 +7,7 @@ reliability of critical deliveries while minimizing idle trips in a toy setting. from __future__ import annotations import time -from typing import Dict, List +from typing import Any, Dict, List from .core import GraphOfContracts, PlanDelta @@ -24,35 +24,66 @@ class ADMMPlanner: """Run a toy optimization pass over the GoC to produce a PlanDelta. We allocate available objects to matching morphisms (requests) by a - simplistic rule: if a source object and target object share a compatible - type and the requested quantity is available, we create an action. + deterministic rule: dispatch-style morphisms create actions when the + source and target exist, and the action payload includes a utility score + that favors balanced capacity and higher-priority signals. """ actions: List[Dict[str, object]] = [] now = time.time() - # Example heuristic: pair objects by type and create dispatch actions - # Primitive: for each morphism, try to assign a matching object as resource. - for mid, morph in gof.morphisms.items(): + for mid in sorted(gof.morphisms): + morph = gof.morphisms[mid] src = gof.objects.get(morph.source_id) dst = gof.objects.get(morph.target_id) if not src or not dst: continue - # Simple compatibility rule: if both share a category/role (type) - if src.type == dst.type: - action = { - "morphism_id": mid, - "action": "dispatch", - "resource_id": src.id, - "destination_id": dst.id, - "timestamp": now, - "details": { - "src_type": src.type, - "dst_type": dst.type, - }, - } - actions.append(action) + + if morph.type not in {"dispatch", "transfer"} and src.type != dst.type: + continue + + action = { + "morphism_id": mid, + "action": "dispatch", + "resource_id": src.id, + "destination_id": dst.id, + "timestamp": now, + "details": { + "src_type": src.type, + "dst_type": dst.type, + "priority": morph.signals.get("priority", "normal"), + "utility_score": self._utility_score(src.attributes, dst.attributes, morph.signals), + }, + } + actions.append(action) + + actions.sort(key=lambda action: (action["morphism_id"], action["resource_id"], action["destination_id"])) # Record plan delta plan = PlanDelta(timestamp=now, actions=actions) gof.plan_delta = plan return plan + + @staticmethod + def _numeric_attribute(attributes: Dict[str, Any], keys: List[str], default: float = 0.0) -> float: + for key in keys: + value = attributes.get(key) + if value is None: + continue + try: + return float(value) + except (TypeError, ValueError): + continue + return default + + def _utility_score( + self, + src_attributes: Dict[str, object], + dst_attributes: Dict[str, object], + signals: Dict[str, object], + ) -> float: + source_capacity = self._numeric_attribute(src_attributes, ["capacity", "stock", "qty", "available"], default=1.0) + target_need = self._numeric_attribute(dst_attributes, ["demand", "capacity", "need", "qty"], default=1.0) + balance = min(source_capacity, target_need) / max(source_capacity, target_need, 1.0) + priority = str(signals.get("priority", "normal")) + priority_boost = {"low": 0.9, "normal": 1.0, "high": 1.1, "critical": 1.25}.get(priority, 1.0) + return round(min(1.0, balance * priority_boost), 3) diff --git a/crisisops/replay.py b/crisisops/replay.py new file mode 100644 index 0000000..5c730d2 --- /dev/null +++ b/crisisops/replay.py @@ -0,0 +1,129 @@ +"""Deterministic snapshots and replay helpers for CrisisOps.""" +from __future__ import annotations + +import hashlib +import json +from typing import Any, Dict, Iterable, List + +from .core import DualVariables, GraphOfContracts, Morphism, Object, PlanDelta + + +SCHEMA_VERSION = "1.0" + + +def canonical_json(payload: Any) -> str: + return json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + + +def _numeric(value: Any, default: float = 0.0) -> float: + try: + return float(value) + except (TypeError, ValueError): + return default + + +def _action_sort_key(action: Dict[str, Any]) -> tuple: + return ( + str(action.get("morphism_id", "")), + str(action.get("action", "")), + str(action.get("resource_id", "")), + str(action.get("destination_id", "")), + ) + + +def _object_to_dict(item: Object) -> Dict[str, Any]: + return {"id": item.id, "type": item.type, "attributes": dict(item.attributes)} + + +def _morphism_to_dict(item: Morphism) -> Dict[str, Any]: + return { + "id": item.id, + "source_id": item.source_id, + "target_id": item.target_id, + "type": item.type, + "signals": dict(item.signals), + } + + +def _dual_to_dict(item: DualVariables) -> Dict[str, Any]: + return {"name": item.name, "values": dict(item.values)} + + +def _plan_to_dict(plan: PlanDelta | None) -> Dict[str, Any] | None: + if plan is None: + return None + actions = [dict(action) for action in plan.actions] + actions.sort(key=_action_sort_key) + return {"timestamp": plan.timestamp, "actions": actions} + + +def graph_snapshot(graph: GraphOfContracts, plan: PlanDelta | None = None) -> Dict[str, Any]: + payload = { + "schema_version": SCHEMA_VERSION, + "objects": [_object_to_dict(graph.objects[key]) for key in sorted(graph.objects)], + "morphisms": [_morphism_to_dict(graph.morphisms[key]) for key in sorted(graph.morphisms)], + "duals": [_dual_to_dict(graph.duals[key]) for key in sorted(graph.duals)], + "plan_delta": _plan_to_dict(plan or graph.plan_delta), + } + payload["snapshot_hash"] = snapshot_hash(payload) + return payload + + +def snapshot_hash(snapshot: Dict[str, Any]) -> str: + normalized = dict(snapshot) + normalized.pop("snapshot_hash", None) + return hashlib.sha256(canonical_json(normalized).encode("utf-8")).hexdigest() + + +def restore_graph(snapshot: Dict[str, Any]) -> GraphOfContracts: + objects = { + item["id"]: Object(id=item["id"], type=item["type"], attributes=dict(item.get("attributes", {}))) + for item in snapshot.get("objects", []) + } + morphisms = { + item["id"]: Morphism( + id=item["id"], + source_id=item["source_id"], + target_id=item["target_id"], + type=item["type"], + signals=dict(item.get("signals", {})), + ) + for item in snapshot.get("morphisms", []) + } + duals = { + item["name"]: DualVariables(name=item["name"], values=dict(item.get("values", {}))) + for item in snapshot.get("duals", []) + } + + plan_data = snapshot.get("plan_delta") + plan_delta = None + if plan_data is not None: + plan_delta = PlanDelta( + timestamp=_numeric(plan_data.get("timestamp")), + actions=[dict(action) for action in plan_data.get("actions", [])], + ) + + graph = GraphOfContracts(objects=objects, morphisms=morphisms, plan_delta=plan_delta, duals=duals) + return graph + + +def replay_graphs(snapshots: Iterable[Dict[str, Any]]) -> List[GraphOfContracts]: + return [restore_graph(snapshot) for snapshot in snapshots] + + +class ReplayJournal: + """A minimal append-only journal of canonical graph snapshots.""" + + def __init__(self) -> None: + self._snapshots: List[Dict[str, Any]] = [] + + def record(self, graph: GraphOfContracts, plan: PlanDelta | None = None) -> Dict[str, Any]: + snapshot = graph_snapshot(graph, plan=plan) + self._snapshots.append(snapshot) + return snapshot + + def list_snapshots(self) -> List[Dict[str, Any]]: + return [dict(snapshot) for snapshot in self._snapshots] + + def replay(self) -> List[GraphOfContracts]: + return replay_graphs(self._snapshots) diff --git a/tests/test_planner.py b/tests/test_planner.py index 4f7de2b..4f64f9b 100644 --- a/tests/test_planner.py +++ b/tests/test_planner.py @@ -1,6 +1,9 @@ import json + from crisisops.planner import ADMMPlanner from crisisops.core import GraphOfContracts, Object, Morphism +from crisisops.governance import GovernanceLedger +from crisisops.replay import graph_snapshot, restore_graph def build_simple_goC(): @@ -14,7 +17,41 @@ def test_planner_allocates_dispatch_actions(): gof = build_simple_goC() planner = ADMMPlanner() plan = planner.optimize(gof) - # Expect at least one action and that it's a dispatch action assert isinstance(plan.actions, list) - if plan.actions: - assert plan.actions[0]["action"] == "dispatch" + assert plan.actions[0]["action"] == "dispatch" + assert plan.actions[0]["details"]["utility_score"] > 0 + + +def test_graph_snapshot_is_deterministic_and_restorable(): + gof_a = build_simple_goC() + gof_b = GraphOfContracts( + objects={"b1": gof_a.objects["b1"], "a1": gof_a.objects["a1"]}, + morphisms={"m1": gof_a.morphisms["m1"]}, + ) + + snapshot_a = graph_snapshot(gof_a) + snapshot_b = graph_snapshot(gof_b) + + assert snapshot_a["snapshot_hash"] == snapshot_b["snapshot_hash"] + + restored = restore_graph(snapshot_a) + assert restored.objects["a1"].attributes["qty"] == 10 + assert restored.morphisms["m1"].target_id == "b1" + + +def test_snapshot_hash_tracks_plan_changes(): + gof = build_simple_goC() + planner = ADMMPlanner() + plan = planner.optimize(gof) + + snapshot = graph_snapshot(gof, plan=plan) + assert snapshot["plan_delta"]["actions"][0]["details"]["utility_score"] > 0 + assert snapshot["snapshot_hash"] + + +def test_governance_ledger_canonicalizes_structured_details(): + ledger = GovernanceLedger() + ledger.log_plan(actor="planner-1", status="approved", plan_hash="abc123", details={"agency": "ngo-a"}) + + record = ledger.list_logs()[0] + assert json.loads(record["details"]) == {"agency": "ngo-a", "plan_hash": "abc123"}