feat(validation): Add structural validation aggregator, health assessment, example & tests

Introduces unified ValidationReport (grammar + field tetrad) and compute_structural_health with actionable recommendations. Adds CLI script structural_health_report.py and example structural_health_demo.py. Updates grammar error factory to accept glyph sequences (maps to canonical names). Provides unit tests for aggregator, health, telemetry emitter, grammar factory. Maintains read-only telemetry; preserves invariants; U4 deferred (no bifurcation call).

Author: fer

examples/structural_health_demo.py

@@ -0,0 +1,118 @@
+"""Minimal structural health + telemetry demo (Phase 3).
+
+Shows integration of TelemetryEmitter with the structural validation
+aggregator and health summary utilities.
+
+Run:
+    python examples/structural_health_demo.py
+
+Outputs:
+ - Human-readable health summary
+ - Telemetry JSONL lines (in-memory example)
+
+Physics Alignment:
+Sequence chosen: [AL, UM, IL, SHA]
+ - AL (Emission)   : Generator (U1a)
+ - UM (Coupling)   : Requires phase compatibility (U3)
+ - IL (Coherence)  : Stabilizer (U2)
+ - SHA (Silence)   : Closure (U1b)
+This satisfies U1a initiation and U1b closure; includes stabilizer
+after coupling; safe canonical bootstrap variant.
+"""
+
+from __future__ import annotations
+
+import random
+from typing import List
+
+try:
+    import networkx as nx  # type: ignore
+except ImportError:  # pragma: no cover
+    raise SystemExit("networkx required for demo")
+
+from tnfr.metrics.telemetry import TelemetryEmitter
+from tnfr.validation.health import compute_structural_health
+from tnfr.validation.aggregator import run_structural_validation
+
+
+def _make_graph(n: int = 16, p: float = 0.15, seed: int = 42):
+    random.seed(seed)
+    G = nx.erdos_renyi_graph(n, p)  # type: ignore
+    # Populate minimal phase & ΔNFR attributes for field computations
+    for node in G.nodes:
+        G.nodes[node]["phase"] = random.random() * 2.0 * 3.141592653589793
+        G.nodes[node]["delta_nfr"] = random.random() * 0.05  # low pressure
+    return G
+
+
+def main() -> None:
+    sequence: List[str] = ["AL", "UM", "IL", "SHA"]
+    G = _make_graph()
+
+    # Baseline structural potential snapshot
+    from tnfr.physics.fields import compute_structural_potential
+
+    baseline_phi_s = compute_structural_potential(G)
+
+    # Telemetry emitter demonstration
+    telemetry_path = "results/telemetry/structural_health_demo.jsonl"
+    with TelemetryEmitter(telemetry_path) as emitter:
+        emitter.record(
+            G,
+            operator="start",
+            extra={"nodes": G.number_of_nodes()},
+        )
+        report = run_structural_validation(
+            G,
+            sequence=sequence,
+            baseline_structural_potential=baseline_phi_s,
+        )
+        emitter.record(
+            G,
+            operator="validation",
+            extra={
+                "risk_level": report.risk_level,
+                "status": report.status,
+                "max_phase_gradient": report.field_metrics[
+                    "max_phase_gradient"
+                ],
+            },
+        )
+        health = compute_structural_health(
+            G, sequence=sequence, baseline_phi_s=baseline_phi_s
+        )
+        emitter.record(
+            G,
+            operator="health",
+            extra={
+                "risk_level": health["risk_level"],
+                "recommended": health["recommended_actions"],
+            },
+        )
+        emitter.flush()
+    print("Telemetry Events (last run):")
+    try:
+        for ln in open(
+            telemetry_path, "r", encoding="utf-8"
+        ).read().splitlines()[-3:]:
+            print("  ", ln)
+    except FileNotFoundError:
+        print("  (no telemetry file found)")
+
+    # Human health summary
+    print("\nStructural Health Summary:")
+    print(f"Status       : {health['status']}")
+    print(f"Risk Level   : {health['risk_level']}")
+    print("Thresholds   :")
+    for k, v in health["thresholds_exceeded"].items():
+        print(f"  - {k}: {'EXCEEDED' if v else 'ok'}")
+    if health["recommended_actions"]:
+        print("Recommended  :", ", ".join(health["recommended_actions"]))
+    if health["notes"]:
+        print("Notes:")
+        for n in health["notes"]:
+            print("  -", n)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()

scripts/structural_health_report.py

@@ -0,0 +1,173 @@
+"""CLI script: Generate TNFR structural health report.
+
+Reads a TNFR graph produced by a simulation (optional) and an operator
+sequence (optional) then prints a concise structural health summary.
+
+Usage examples:
+---------------
+python scripts/structural_health_report.py \
+    --graph examples/output/graph.pkl \
+    --sequence examples/output/sequence.txt \
+    --json results/reports/health_report.json
+
+python scripts/structural_health_report.py --random 32 --edge-prob 0.15
+
+Inputs
+------
+Graph formats supported:
+ - Pickle (NetworkX graph)
+ - Edge list (.edgelist) simple whitespace separated pairs
+
+Sequence file: one operator mnemonic per line (e.g. AL, UM, IL, SHA).
+
+Outputs
+-------
+STDOUT: Human-readable summary
+JSON (optional): Machine-readable payload
+
+All computations are telemetry-only; graph is never mutated.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import List
+
+try:  # networkx dependency
+    import networkx as nx  # type: ignore
+except ImportError:  # pragma: no cover
+    nx = None  # type: ignore
+
+from tnfr.validation.health import compute_structural_health
+
+
+def _load_graph(path: Path):
+    if nx is None:  # pragma: no cover
+        raise RuntimeError("networkx not available")
+    if path.suffix == ".pkl":
+        import pickle
+
+        with path.open("rb") as f:
+            return pickle.load(f)
+    if path.suffix == ".edgelist":
+        G = nx.read_edgelist(path)  # type: ignore
+        return G
+    raise ValueError(f"Unsupported graph format: {path.suffix}")
+
+
+def _load_sequence(path: Path) -> List[str]:
+    return [ln.strip() for ln in path.read_text().splitlines() if ln.strip()]
+
+
+def parse_args(argv: List[str]) -> argparse.Namespace:
+    p = argparse.ArgumentParser(description="TNFR structural health report")
+    p.add_argument(
+        "--graph",
+        type=Path,
+        help="Graph pickle (.pkl) or edge list (.edgelist)",
+        required=False,
+    )
+    p.add_argument(
+        "--sequence",
+        type=Path,
+        help="Operator sequence file (one mnemonic per line)",
+        required=False,
+    )
+    p.add_argument(
+        "--json",
+        type=Path,
+        help="Optional JSON output path",
+        required=False,
+    )
+    p.add_argument(
+        "--random",
+        type=int,
+        help="Generate random Erdos-Renyi graph with N nodes",
+    )
+    p.add_argument(
+        "--edge-prob",
+        type=float,
+        default=0.1,
+        help="Probability for random graph edges",
+    )
+    p.add_argument(
+        "--seed",
+        type=int,
+        default=42,
+        help="Random seed for reproducibility",
+    )
+    return p.parse_args(argv)
+
+
+def main(argv: List[str]) -> int:
+    args = parse_args(argv)
+    if nx is None:  # pragma: no cover
+        print("networkx required for health report", file=sys.stderr)
+        return 2
+
+    if args.graph and args.random:
+        print("Specify either --graph or --random, not both", file=sys.stderr)
+        return 2
+
+    if args.random:
+        nx.random.seed(args.seed)  # type: ignore[attr-defined]
+        G = nx.erdos_renyi_graph(args.random, args.edge_prob)  # type: ignore
+    elif args.graph:
+        G = _load_graph(args.graph)
+    else:
+        print("Must supply --graph or --random", file=sys.stderr)
+        return 2
+
+    sequence = _load_sequence(args.sequence) if args.sequence else None
+
+    health = compute_structural_health(G, sequence=sequence)
+
+    # Human summary
+    print("TNFR Structural Health Report")
+    print("--------------------------------")
+    print(f"Status            : {health['status']}")
+    print(f"Risk Level        : {health['risk_level']}")
+    subset = health["field_metrics_subset"]
+    if subset["mean_phi_s"] is not None:
+        print(f"Mean Φ_s          : {subset['mean_phi_s']:.4f}")
+    else:
+        print("Mean Φ_s          : NA")
+    if subset["max_phase_gradient"] is not None:
+        print(f"Max |∇φ|          : {subset['max_phase_gradient']:.4f}")
+    else:
+        print("Max |∇φ|          : NA")
+    if subset["max_k_phi"] is not None:
+        print(f"Max |K_φ|         : {subset['max_k_phi']:.4f}")
+    else:
+        print("Max |K_φ|         : NA")
+    if subset["xi_c"] is not None:
+        print(f"ξ_C               : {subset['xi_c']:.2f}")
+    else:
+        print("ξ_C               : NA")
+    if subset["delta_phi_s"] is not None:
+        print(f"ΔΦ_s drift        : {subset['delta_phi_s']:.4f}")
+    print("Threshold Flags   :")
+    for k, v in health["thresholds_exceeded"].items():
+        print(f"  - {k}: {'EXCEEDED' if v else 'ok'}")
+    if health["recommended_actions"]:
+        print("Recommended Actions:")
+        for act in health["recommended_actions"]:
+            print(f"  * {act}")
+    if health["notes"]:
+        print("Notes:")
+        for n in health["notes"]:
+            print(f"  - {n}")
+
+    if args.json:
+        args.json.parent.mkdir(parents=True, exist_ok=True)
+        args.json.write_text(json.dumps(health, indent=2))
+        print(f"JSON report written to {args.json}")
+
+    return 0
+
+
+if __name__ == "__main__":  # pragma: no cover
+    raise SystemExit(main(sys.argv[1:]))

src/tnfr/operators/grammar_error_factory.py

@@ -166,14 +166,42 @@ def collect_grammar_errors(
     validator = GrammarValidator()
     errors: List[ExtendedGrammarError] = []
 
+    # Accept glyph strings by wrapping them in lightweight stubs
+    # expected by GrammarValidator (which accesses .name / .canonical_name).
+    GLYPH_TO_NAME = {
+        "AL": "emission",
+        "EN": "reception",
+        "IL": "coherence",
+        "OZ": "dissonance",
+        "UM": "coupling",
+        "RA": "resonance",
+        "SHA": "silence",
+        "VAL": "expansion",
+        "NUL": "contraction",
+        "THOL": "self_organization",
+        "ZHIR": "mutation",
+        "NAV": "transition",
+        "REMESH": "recursivity",
+    }
+    
+    class _OpStub:  # local minimal stub
+        def __init__(self, glyph: str):
+            canonical = GLYPH_TO_NAME.get(glyph.upper(), glyph.lower())
+            self.canonical_name = canonical
+            self.name = canonical
+
+    normalized: List[Any] = [
+        (_OpStub(op) if isinstance(op, str) else op) for op in sequence
+    ]
+
     # Canonical operator names for reporting
     canonical = [
         getattr(op, "canonical_name", getattr(op, "name", "?"))
-        for op in sequence
+        for op in normalized
     ]
 
     # U1a
-    ok, msg = validator.validate_initiation(list(sequence), epi_initial)
+    ok, msg = validator.validate_initiation(list(normalized), epi_initial)
     if not ok:
         errors.append(
             make_grammar_error(
@@ -185,7 +213,7 @@ def collect_grammar_errors(
             )
         )
     # U1b
-    ok, msg = validator.validate_closure(list(sequence))
+    ok, msg = validator.validate_closure(list(normalized))
     if not ok:
         errors.append(
             make_grammar_error(
@@ -197,7 +225,7 @@ def collect_grammar_errors(
             )
         )
     # U2
-    ok, msg = validator.validate_convergence(list(sequence))
+    ok, msg = validator.validate_convergence(list(normalized))
     if not ok:
         errors.append(
             make_grammar_error(
@@ -208,7 +236,7 @@ def collect_grammar_errors(
             )
         )
     # U3
-    ok, msg = validator.validate_resonant_coupling(list(sequence))
+    ok, msg = validator.validate_resonant_coupling(list(normalized))
     if not ok:
         # Find first coupling/resonance candidate if available
         idx = next(
@@ -229,17 +257,4 @@ def collect_grammar_errors(
                 index=idx,
             )
         )
-    # U4 (validator returns both in one pass)
-    ok, msg = validator.validate_bifurcation(list(sequence))
-    if not ok:
-        # Distinguish U4a vs U4b heuristically using message content
-        rule_key = "U4b" if "transform" in msg.lower() else "U4a"
-        errors.append(
-            make_grammar_error(
-                rule=rule_key,
-                candidate="sequence",
-                message=msg,
-                sequence=canonical,
-            )
-        )
     return errors