build(agent): jabba#56a767 iteration

gravityweave/validator.py

@@ -0,0 +1,84 @@
+"""Lightweight witness validator.
+
+Provides quick local validation of a feasibility witness against a node's
+resource envelope. Intended to be cheap and conservative so it can run on
+constrained nodes and during short custody windows.
+
+Verdicts:
+- "OK": witness comfortably within resource margins
+- "needs-review": within a configurable margin (conservative)
+- "unstable": witness exceeds available resources (infeasible)
+
+If a secret_key is provided, make_signed_verdict will attach an HMAC-SHA256
+signature over the verdict (short hex).
+"""
+from typing import Dict, Any, Tuple
+import json
+import hmac
+import hashlib
+
+
+def validate_witness(witness: Dict[str, Any], resources: Dict[str, Any], *, fuel_margin: float = 0.1, power_margin: float = 0.1) -> Tuple[str, Dict[str, Any]]:
+    """Validate a witness against resources.
+
+    Parameters:
+    - witness: output of generate_feasibility_witness
+    - resources: dict with keys 'available_fuel', 'available_power', 'max_payload_bytes'
+    - fuel_margin/power_margin: fractional margin; if witness consumption is
+      above available*(1 - margin) but <= available it's 'needs-review'.
+
+    Returns (verdict, details)
+    """
+    details: Dict[str, Any] = {"checks": []}
+
+    # helpers
+    def _check(name: str, required: float, available: float, margin: float):
+        if required <= available * (1.0 - margin):
+            return "ok", f"{name} within comfortable margin ({required} <= {available*(1.0-margin)})"
+        if required <= available:
+            return "review", f"{name} within hard limit but near margin ({required} <= {available})"
+        return "unstable", f"{name} exceeds available ({required} > {available})"
+
+    total_fuel = float(witness.get("total_fuel", 0.0))
+    total_power = float(witness.get("total_power", 0.0))
+    max_payload = int(witness.get("max_payload_bytes", 0))
+
+    avail_fuel = float(resources.get("available_fuel", 0.0))
+    avail_power = float(resources.get("available_power", 0.0))
+    avail_payload = int(resources.get("max_payload_bytes", 0))
+
+    f_status, f_msg = _check("fuel", total_fuel, avail_fuel, fuel_margin)
+    p_status, p_msg = _check("power", total_power, avail_power, power_margin)
+    pl_status, pl_msg = _check("payload", float(max_payload), float(avail_payload), 0.0)
+
+    details["checks"].append({"fuel": {"status": f_status, "msg": f_msg, "required": total_fuel, "available": avail_fuel}})
+    details["checks"].append({"power": {"status": p_status, "msg": p_msg, "required": total_power, "available": avail_power}})
+    details["checks"].append({"payload": {"status": pl_status, "msg": pl_msg, "required": max_payload, "available": avail_payload}})
+
+    # Decide overall verdict: unstable > needs-review > OK
+    statuses = {f_status, p_status, pl_status}
+    if "unstable" in statuses:
+        verdict = "unstable"
+    elif "review" in statuses:
+        verdict = "needs-review"
+    else:
+        verdict = "OK"
+
+    details["verdict_reason"] = f"derived from checks: {sorted(list(statuses))}"
+    return verdict, details
+
+
+def make_signed_verdict(witness: Dict[str, Any], resources: Dict[str, Any], secret_key: bytes = None, **validate_kwargs) -> Dict[str, Any]:
+    """Run validation and optionally attach an HMAC signature.
+
+    The returned dict contains: verdict, details, and if secret_key provided,
+    a short signature hex string (first 32 chars of HMAC-SHA256).
+    """
+    verdict, details = validate_witness(witness, resources, **validate_kwargs)
+    out = {"verdict": verdict, "details": details}
+    # canonicalize and sign
+    if secret_key is not None:
+        payload = json.dumps({"verdict": verdict, "details": details}, sort_keys=True, separators=(",", ":")).encode("utf-8")
+        sig = hmac.new(secret_key, payload, hashlib.sha256).hexdigest()[:32]
+        out["signature"] = sig
+    return out

tests/test_validator.py

@@ -0,0 +1,46 @@
+from gravityweave.plan_delta import PlanDelta
+from gravityweave.witness import generate_feasibility_witness
+from gravityweave.validator import validate_witness, make_signed_verdict
+
+
+def test_validate_witness_ok():
+    pd = PlanDelta(origin="node1")
+    pd.commitments.add({"type": "burn", "fuel": 5.0, "power": 1.0, "payload_bytes": 100})
+    w = generate_feasibility_witness(pd)
+
+    resources = {"available_fuel": 10.0, "available_power": 5.0, "max_payload_bytes": 200}
+    verdict, details = validate_witness(w, resources, fuel_margin=0.1, power_margin=0.1)
+    assert verdict == "OK"
+    assert "checks" in details
+
+
+def test_validate_witness_needs_review():
+    pd = PlanDelta(origin="node2")
+    pd.commitments.add({"type": "burn", "fuel": 9.0, "power": 1.0})
+    w = generate_feasibility_witness(pd)
+
+    resources = {"available_fuel": 10.0, "available_power": 5.0, "max_payload_bytes": 0}
+    verdict, _ = validate_witness(w, resources, fuel_margin=0.2, power_margin=0.1)
+    assert verdict == "needs-review"
+
+
+def test_validate_witness_unstable():
+    pd = PlanDelta(origin="node3")
+    pd.commitments.add({"type": "burn", "fuel": 11.0, "power": 1.0})
+    w = generate_feasibility_witness(pd)
+
+    resources = {"available_fuel": 10.0, "available_power": 5.0, "max_payload_bytes": 0}
+    verdict, _ = validate_witness(w, resources, fuel_margin=0.1, power_margin=0.1)
+    assert verdict == "unstable"
+
+
+def test_make_signed_verdict():
+    pd = PlanDelta(origin="node4")
+    pd.commitments.add({"type": "burn", "fuel": 1.0, "power": 0.5})
+    w = generate_feasibility_witness(pd)
+
+    resources = {"available_fuel": 5.0, "available_power": 5.0, "max_payload_bytes": 0}
+    secret = b"testkey"
+    out = make_signed_verdict(w, resources, secret_key=secret)
+    assert out["verdict"] == "OK"
+    assert "signature" in out and isinstance(out["signature"], str) and len(out["signature"]) == 32