build(agent): weasel-1#856f80 iteration

gravityweave/__init__.py

@@ -11,6 +11,7 @@ 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
 
 __all__ = [
     "Registry",
@@ -21,4 +22,8 @@ __all__ = [
     "sat_planner",
     "relay_module",
     "ADMMNode",
+    "build_synopsis",
+    "build_priority_key",
+    "serialize_synopsis",
+    "deserialize_synopsis",
 ]

gravityweave/delta_synopsis.py

@@ -0,0 +1,75 @@
+"""PlanDelta synopsis and priority utils.
+
+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
+import json
+import hashlib
+import hmac
+import uuid
+
+
+def _serialize_delta(delta) -> bytes:
+    # serialize commitments and priority into a deterministic JSON blob
+    payload = {
+        "commitments": [c for c in delta.commitments.value()],
+        "priority": delta.priority.read(),
+        "origin": getattr(delta, "origin", ""),
+    }
+    return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+
+def build_priority_key(urgency_score: int, impact_score: int) -> int:
+    """Combine urgency (0-7) and impact (0-15) into a 0-127 key as (urgency<<4)|impact."""
+    if not (0 <= urgency_score <= 7):
+        raise ValueError("urgency_score must be 0..7")
+    if not (0 <= impact_score <= 15):
+        raise ValueError("impact_score must be 0..15")
+    return (urgency_score << 4) | (impact_score & 0x0F)
+
+
+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]:
+    """Build a compact synopsis for a PlanDelta.
+
+    - delta: PlanDelta instance
+    - parent_version: reference to parent PlanDelta/version
+    - urgency_score, impact_score: small integers used to compute priority_key
+    - affected_assets: optional list of affected node ids
+    - signer: identifier of signer
+    - key: optional bytes key used to produce a short HMAC signature
+    """
+    payload = _serialize_delta(delta)
+    summary_hash = hashlib.sha256(payload).hexdigest()
+    size_estimate = len(payload)
+    priority_key = build_priority_key(urgency_score, impact_score)
+    delta_id = uuid.uuid4().hex
+
+    synopsis = {
+        "delta_id": delta_id,
+        "parent_version": parent_version,
+        "priority_key": priority_key,
+        "affected_assets": affected_assets or [],
+        "summary_hash": summary_hash,
+        "size_estimate": size_estimate,
+        "signer": signer,
+    }
+
+    # short signature: HMAC of summary_hash with provided key if key present
+    if key:
+        short_sig = hmac.new(key, summary_hash.encode("utf-8"), hashlib.sha256).hexdigest()
+    else:
+        short_sig = ""
+
+    synopsis["short_sig"] = short_sig
+    return synopsis
+
+
+def serialize_synopsis(syn: Dict[str, Any]) -> bytes:
+    return json.dumps(syn, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+
+def deserialize_synopsis(b: bytes) -> Dict[str, Any]:
+    return json.loads(b.decode("utf-8"))

tests/test_synopsis.py

@@ -0,0 +1,29 @@
+import secrets
+from gravityweave.plan_delta import PlanDelta
+from gravityweave.delta_synopsis import build_synopsis, build_priority_key, serialize_synopsis, deserialize_synopsis
+
+
+def test_priority_key_bounds():
+    k = build_priority_key(3, 12)
+    assert isinstance(k, int)
+    assert k == (3 << 4) | 12
+
+
+def test_build_and_serialize_synopsis():
+    key = secrets.token_bytes(16)
+    p = PlanDelta(origin="nodeX")
+    p.commitments.add({"type": "burn", "fuel": 2.0})
+    p.priority.write(7)
+
+    syn = build_synopsis(p, parent_version="v0", urgency_score=2, impact_score=8,
+                         affected_assets=["sat1", "relayA"], signer="nodeX", key=key)
+
+    assert syn["parent_version"] == "v0"
+    assert syn["priority_key"] == (2 << 4) | 8
+    assert syn["size_estimate"] > 0
+    assert syn["summary_hash"]
+    assert syn["short_sig"]
+
+    b = serialize_synopsis(syn)
+    d = deserialize_synopsis(b)
+    assert d["delta_id"] == syn["delta_id"]