Add: metrics.py, __init__.py files, pyproject.toml, tests

Author: Riverbraid

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "setuptools.backends.legacy:build"
 [project]
 name = "riverbraid-core"
 version = "1.3.0"
-description = "Deterministic capacity control substrate"
+description = "Deterministic capacity control substrate for AI and automation systems"
 requires-python = ">=3.10"
 license = { text = "MIT" }
 

src/riverbraid/core/metrics.py

@@ -1,50 +1,78 @@
 from __future__ import annotations
 import math
-from typing import TypedDict
-
-class RawInputs(TypedDict, total=False):
-    coherence: float
-    novelty: float
-    fragility: float
-    latency: float
-
-class Metrics(TypedDict):
-    systemic_load: float
-    pattern_disruption: float
-    interaction_variance: float
-    coherence_confidence: float
-
-class ComputeResult(TypedDict):
-    mode: str
-    metrics: Metrics
-    validation_warning: bool
-    AUDIT_HASH: str
-
-AUDIT_HASH: str = "rb-core-metrics-v1.3.0"
-REST_THRESHOLD: float = 0.30
-ENGAGE_THRESHOLD: float = 0.65
-SYSTEMIC_LOAD_OVERRIDE: float = 0.85
+
+
+AUDIT_HASH = "rb-core-metrics-v1.3.0"
+REST_THRESHOLD = 0.30
+ENGAGE_THRESHOLD = 0.65
+SYSTEMIC_LOAD_OVERRIDE = 0.85
+
 
 def _clamp(value):
-    if not isinstance(value, (int, float)): return 0.0, True
-    if math.isnan(value): return 0.0, True
-    if math.isinf(value): return (1.0 if value > 0 else 0.0), True
-    if value < 0.0: return 0.0, True
-    if value > 1.0: return 1.0, True
+    if not isinstance(value, (int, float)):
+        return 0.0, True
+    if math.isnan(value):
+        return 0.0, True
+    if math.isinf(value):
+        return (1.0 if value > 0 else 0.0), True
+    if value < 0.0:
+        return 0.0, True
+    if value > 1.0:
+        return 1.0, True
     return float(value), False
 
+
 def compute_metrics(inputs):
+    """
+    Compute capacity metrics and select a mode.
+
+    Inputs (all optional, default 0.0):
+        coherence  [0.0-1.0]  signal clarity
+        novelty    [0.0-1.0]  degree of unexpected input
+        fragility  [0.0-1.0]  sensitivity to disruption
+        latency    [0.0-1.0]  delay or backlog pressure
+
+    Formulas (fixed):
+        systemic_load        = (fragility + latency) / 2
+        pattern_disruption   = novelty * (1 - coherence)
+        interaction_variance = max(systemic_load, pattern_disruption)
+        coherence_confidence = coherence * (1 - interaction_variance)
+
+    Modes:
+        systemic_load >= 0.85        -> rest (override)
+        coherence_confidence < 0.30  -> rest
+        coherence_confidence < 0.65  -> soften
+        coherence_confidence >= 0.65 -> engage
+    """
     coherence, w1 = _clamp(inputs.get("coherence", 0.0))
     novelty,   w2 = _clamp(inputs.get("novelty",   0.0))
     fragility, w3 = _clamp(inputs.get("fragility", 0.0))
     latency,   w4 = _clamp(inputs.get("latency",   0.0))
+
     warn = w1 or w2 or w3 or w4
+
     sl = (fragility + latency) / 2.0
     pd = novelty * (1.0 - coherence)
     iv = max(sl, pd)
     cc = coherence * (1.0 - iv)
-    if sl >= SYSTEMIC_LOAD_OVERRIDE: mode = "rest"
-    elif cc < REST_THRESHOLD: mode = "rest"
-    elif cc < ENGAGE_THRESHOLD: mode = "soften"
-    else: mode = "engage"
-    return {"mode": mode, "metrics": {"systemic_load": round(sl,6), "pattern_disruption": round(pd,6), "interaction_variance": round(iv,6), "coherence_confidence": round(cc,6)}, "validation_warning": warn, "AUDIT_HASH": AUDIT_HASH}
+
+    if sl >= SYSTEMIC_LOAD_OVERRIDE:
+        mode = "rest"
+    elif cc < REST_THRESHOLD:
+        mode = "rest"
+    elif cc < ENGAGE_THRESHOLD:
+        mode = "soften"
+    else:
+        mode = "engage"
+
+    return {
+        "mode": mode,
+        "metrics": {
+            "systemic_load":        round(sl, 6),
+            "pattern_disruption":   round(pd, 6),
+            "interaction_variance": round(iv, 6),
+            "coherence_confidence": round(cc, 6),
+        },
+        "validation_warning": warn,
+        "AUDIT_HASH": AUDIT_HASH,
+    }

tests/test_metrics.py

@@ -1,35 +1,51 @@
 import pytest
-import math
 from riverbraid.core.metrics import compute_metrics, AUDIT_HASH
 
-def test_audit_hash_stability():
-    result = compute_metrics({})
-    assert result["hash"] == AUDIT_HASH
 
-def test_empty_input_defaults():
-    result = compute_metrics({})
-    assert result["mode"] == "rest"
-    assert result["metrics"]["systemic_load"] == 0.0
+def test_audit_hash_present():
+    assert compute_metrics({})["AUDIT_HASH"] == AUDIT_HASH
 
-def test_engage_mode():
-    inputs = {"coherence": 0.9, "novelty": 0.1, "fragility": 0.1, "latency": 0.1}
-    result = compute_metrics(inputs)
-    assert result["mode"] == "engage"
+def test_empty_input_returns_rest():
+    assert compute_metrics({})["mode"] == "rest"
+
+def test_missing_inputs_default_to_zero():
+    m = compute_metrics({})["metrics"]
+    assert m["systemic_load"] == 0.0
+    assert m["coherence_confidence"] == 0.0
+
+def test_engage():
+    r = compute_metrics({"coherence": 0.9, "novelty": 0.1, "fragility": 0.1, "latency": 0.1})
+    assert r["mode"] == "engage"
+    assert r["validation_warning"] is False
+
+def test_soften():
+    r = compute_metrics({"coherence": 0.6, "novelty": 0.3, "fragility": 0.2, "latency": 0.2})
+    assert r["mode"] == "soften"
+
+def test_rest_low_coherence():
+    r = compute_metrics({"coherence": 0.1, "novelty": 0.9, "fragility": 0.1, "latency": 0.1})
+    assert r["mode"] == "rest"
 
 def test_systemic_load_override():
-    # Load >= 0.85 should force rest regardless of coherence
-    inputs = {"coherence": 0.99, "fragility": 0.9, "latency": 0.9}
-    result = compute_metrics(inputs)
-    assert result["mode"] == "rest"
-    assert result["metrics"]["systemic_load"] == 0.9
+    r = compute_metrics({"coherence": 0.99, "novelty": 0.0, "fragility": 0.9, "latency": 0.9})
+    assert r["mode"] == "rest"
+    assert r["metrics"]["systemic_load"] >= 0.85
+
+def test_nan_warns():
+    assert compute_metrics({"coherence": float("nan")})["validation_warning"] is True
 
 def test_determinism():
     inputs = {"coherence": 0.7, "novelty": 0.2, "fragility": 0.3, "latency": 0.1}
-    first = compute_metrics(inputs)
-    for _ in range(10):
-        assert compute_metrics(inputs) == first
-
-def test_clamping_behavior():
-    # Values above 1.0 should clamp to 1.0
-    result = compute_metrics({"coherence": 1.5})
-    assert result["metrics"]["coherence_confidence"] <= 1.0
+    results = [compute_metrics(inputs) for _ in range(50)]
+    assert all(r == results[0] for r in results[1:])
+
+def test_formula_correctness():
+    c, n, f, l = 0.7, 0.2, 0.3, 0.1
+    r = compute_metrics({"coherence": c, "novelty": n, "fragility": f, "latency": l})
+    m = r["metrics"]
+    sl = (f + l) / 2
+    pd = n * (1 - c)
+    iv = max(sl, pd)
+    cc = c * (1 - iv)
+    assert abs(m["systemic_load"] - sl) < 1e-9
+    assert abs(m["coherence_confidence"] - cc) < 1e-9