interactions.py

"""Canonical interaction sequences composed from TNFR operators (centralized).

Overview
--------
This module provides physics-first helpers that compose structural operators
into interaction-like sequences while enforcing unified grammar awareness
(U1–U6) and instrumenting read-only telemetry via the Structural Field Tetrad
(Φ_s, |∇φ|, K_φ, ξ_C). These helpers are thin orchestrators and never mutate
EPI directly; they invoke canonical operators from
``tnfr.operators.definitions``.

Operator–Grammar Mapping (informative):
- Coupling (UM): requires phase verification (U3) inside operator
- Resonance (RA): propagates EPI coherently; phase-aware (U3)
- Coherence (IL): stabilizer enforcing boundedness (U2)
- Dissonance (OZ): controlled destabilizer; must be followed by handlers (U4a)
- Mutation (ZHIR): transformer at threshold; requires recent OZ and
    prior IL (U4b)
- SelfOrganization (THOL): creates sub-EPIs; stabilizer and
    transformer (U2, U4)
- Silence (SHA): closure/observation window (U1b)

Telemetry (read-only):
- |∇φ|: phase gradient; early stress indicator (threshold ≈ 0.38)
- K_φ: phase curvature (part of unified Ψ); confinement hotspots (|K_φ| ≥ 2.8274)
- Φ_s: structural potential; mean absolute drift as passive safety (ΔΦ_s < π/2)

Contracts (summary):
- EM-like: [UM → RA → IL]. Preserve identity; phase-verified coupling (U3).
- Weak-like: [IL? → OZ → ZHIR → IL]. U4b-compliant (prior IL + recent OZ).
- Strong-like: [UM → IL → THOL]. Curvature-driven confinement;
    post-coupling IL.
- Gravity-like: [IL → SHA?]. Telemetry-oriented stabilization (Φ_s drift only).

Usage pattern
-------------
Given a graph G with phase (``theta``/``phase``) and ΔNFR
(``delta_nfr``/``dnfr``) attributes set (e.g., via
``tnfr.physics.patterns``), call one of the helpers with an iterable of
nodes. Results contain operator names applied and key telemetry means
before/after, plus optional warnings when thresholds are exceeded.

Notes
-----
- English-only documentation for canonicity and single
    source-of-truth alignment.
- Extend carefully with physics-first justification and tests. Keep parameters
    minimal and respect U-rules and structural invariants at all times.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Iterable

from ..mathematics.unified_numerical import np

try:
    import networkx as nx  # type: ignore
except Exception:  # pragma: no cover
    nx = None  # type: ignore

from ..constants.canonical import U6_STRUCTURAL_POTENTIAL_LIMIT
from ..constants.canonical import (
    PHYSICS_CURVATURE_HOTSPOT_CANONICAL,
    PHYSICS_GRAD_THRESHOLD_CANONICAL,
    PHYSICS_HOTSPOT_FRACTION_CANONICAL,
)
from ..operators.definitions import (
    Coherence,
    Coupling,
    Dissonance,
    Mutation,
    Resonance,
    SelfOrganization,
    Silence,
)
from .fields import (
    compute_phase_curvature,
    compute_phase_gradient,
    compute_structural_potential,
)


@dataclass
class InteractionResult:
    applied: list[str]
    warnings: list[str]
    grad_before_mean: float | None = None
    grad_after_mean: float | None = None
    kphi_before_abs_mean: float | None = None
    kphi_after_abs_mean: float | None = None
    phi_s_drift_mean: float | None = None


def _mean(d: dict[Any, float]) -> float:
    return float(np.mean(list(d.values()))) if d else 0.0


def _apply_to_nodes(G: Any, nodes: Iterable[Any], ops: Iterable[Any]) -> list[str]:
    """Apply operator instances to each node in order; return names applied."""
    applied: list[str] = []
    for node in nodes:
        for op in ops:
            op(G, node)
            applied.append(op.name)
    return applied


def _telemetry_before_after(G: Any, *, compute_phi_s: bool = False) -> dict:
    grad_b = compute_phase_gradient(G)
    kphi_b = compute_phase_curvature(G)
    phi_b = compute_structural_potential(G) if compute_phi_s else None
    return {
        "grad_b": grad_b,
        "kphi_b": kphi_b,
        "phi_b": phi_b,
    }


def _telemetry_after(G: Any, snap: dict, *, compute_phi_s: bool = False) -> dict:
    grad_a = compute_phase_gradient(G)
    kphi_a = compute_phase_curvature(G)
    phi_a = compute_structural_potential(G) if compute_phi_s else None
    drift = None
    if compute_phi_s and snap.get("phi_b") is not None and phi_a is not None:
        b = snap["phi_b"]
        # Mean absolute drift across nodes present in both
        keys = list({*b.keys(), *phi_a.keys()})
        if keys:
            diffs = [abs(phi_a.get(k, 0.0) - b.get(k, 0.0)) for k in keys]
            drift = float(np.mean(diffs))
    return {
        "grad_a": grad_a,
        "kphi_a": kphi_a,
        "phi_a": phi_a,
        "phi_drift": drift,
    }


def em_like(
    G: Any,
    nodes: Iterable[Any],
    *,
    compute_phi_s: bool = False,
    grad_threshold: float = PHYSICS_GRAD_THRESHOLD_CANONICAL,
) -> InteractionResult:
    """EM-like sequence: [Coupling → Resonance → Coherence].

    Purpose
    -------
    Reinforce/coherently propagate existing patterns under phase-compatible
    coupling, then stabilize. Identity-preserving and phase-aware.

    Contracts
    ---------
    - U3: Coupling/Resonance require phase compatibility (verified internally).
    - U2: Coherence stabilizes to maintain boundedness.
    - Read-only telemetry: |∇φ|, K_φ, optional Φ_s drift.

    Parameters
    ----------
    G : Any
        NetworkX-like graph with node attributes ``theta``/``phase`` and
        ``delta_nfr``/``dnfr``.
    nodes : Iterable
        Nodes to which the operator sequence will be applied.
    compute_phi_s : bool, default False
        If True, compute Φ_s before/after and report mean drift.
    grad_threshold : float, default PHYSICS_GRAD_THRESHOLD_CANONICAL
        Heuristic early-warning threshold for mean |∇φ| (≈ 0.196, π/16;
        calibrated, audit 2026: NOT a derived bound — the kinematic bound is
        |∇φ| ≤ π and the sync-onset is σ-dependent ≈ 0.29).

    Returns
    -------
    InteractionResult
        Names of operators applied, warnings, and telemetry means.
    """
    snap = _telemetry_before_after(G, compute_phi_s=compute_phi_s)

    ops = [Coupling(), Resonance(), Coherence()]
    applied = _apply_to_nodes(G, nodes, ops)

    aft = _telemetry_after(G, snap, compute_phi_s=compute_phi_s)
    grad_mean_b = _mean(snap["grad_b"])  # type: ignore[index]
    grad_mean_a = _mean(aft["grad_a"])  # type: ignore[index]
    kphi_abs_b = _mean(
        {k: abs(v) for k, v in snap["kphi_b"].items()}
    )  # type: ignore[index]
    kphi_abs_a = _mean(
        {k: abs(v) for k, v in aft["kphi_a"].items()}
    )  # type: ignore[index]

    warnings: list[str] = []
    if grad_mean_a >= grad_threshold:
        warnings.append(
            (
                "phase gradient high after EM-like: "
                f"{grad_mean_a:.3f} ≥ {grad_threshold}"
            )
        )
    if aft.get("phi_drift") is not None and float(aft["phi_drift"]) >= U6_STRUCTURAL_POTENTIAL_LIMIT:
        warnings.append(
            (
                "structural potential drift exceeded threshold: "
                f"{float(aft['phi_drift']):.3f} ≥ {U6_STRUCTURAL_POTENTIAL_LIMIT:.3f}"
            )
        )

    return InteractionResult(
        applied=applied,
        warnings=warnings,
        grad_before_mean=grad_mean_b,
        grad_after_mean=grad_mean_a,
        kphi_before_abs_mean=kphi_abs_b,
        kphi_after_abs_mean=kphi_abs_a,
        phi_s_drift_mean=(
            float(aft["phi_drift"]) if aft.get("phi_drift") is not None else None
        ),
    )


def weak_like(
    G: Any,
    nodes: Iterable[Any],
    *,
    compute_phi_s: bool = False,
    ensure_stable_base: bool = True,
    grad_threshold: float = PHYSICS_GRAD_THRESHOLD_CANONICAL,
) -> InteractionResult:
    """Weak-like sequence: [IL (optional) → Dissonance → Mutation → Coherence].

    Purpose
    -------
    Trigger controlled qualitative transformations (ZHIR) with stabilized
    pre/post conditions.

    Contracts
    ---------
    - U4b: Mutation (ZHIR) requires a stable base (prior IL) and recent
      destabilizer (OZ) within ~3 operations.
    - U2: Final Coherence to contain elevated ΔNFR.
    - Read-only telemetry: |∇φ|, K_φ, optional Φ_s drift.

    Parameters
    ----------
    ensure_stable_base : bool, default True
        Insert IL before OZ→ZHIR to satisfy U4b stable base requirement.
    grad_threshold : float, default PHYSICS_GRAD_THRESHOLD_CANONICAL
        Heuristic early-warning threshold for mean |∇φ| (≈ 0.196, π/16;
        calibrated, audit 2026: NOT a derived bound — the kinematic bound is
        |∇φ| ≤ π and the sync-onset is σ-dependent ≈ 0.29).

    Returns
    -------
    InteractionResult
        Names of operators applied, warnings, and telemetry means.
    """
    snap = _telemetry_before_after(G, compute_phi_s=compute_phi_s)

    ops: list[Any] = []
    if ensure_stable_base:
        ops.append(Coherence())
    ops.extend([Dissonance(), Mutation(), Coherence()])
    applied = _apply_to_nodes(G, nodes, ops)

    aft = _telemetry_after(G, snap, compute_phi_s=compute_phi_s)
    grad_mean_b = _mean(snap["grad_b"])  # type: ignore[index]
    grad_mean_a = _mean(aft["grad_a"])  # type: ignore[index]
    kphi_abs_b = _mean(
        {k: abs(v) for k, v in snap["kphi_b"].items()}
    )  # type: ignore[index]
    kphi_abs_a = _mean(
        {k: abs(v) for k, v in aft["kphi_a"].items()}
    )  # type: ignore[index]

    warnings: list[str] = []
    if grad_mean_a >= grad_threshold:
        warnings.append(
            (
                "phase gradient high after Weak-like: "
                f"{grad_mean_a:.3f} ≥ {grad_threshold}"
            )
        )
    if aft.get("phi_drift") is not None and float(aft["phi_drift"]) >= U6_STRUCTURAL_POTENTIAL_LIMIT:
        warnings.append(
            (
                "structural potential drift exceeded threshold: "
                f"{float(aft['phi_drift']):.3f} ≥ {U6_STRUCTURAL_POTENTIAL_LIMIT:.3f}"
            )
        )

    return InteractionResult(
        applied=applied,
        warnings=warnings,
        grad_before_mean=grad_mean_b,
        grad_after_mean=grad_mean_a,
        kphi_before_abs_mean=kphi_abs_b,
        kphi_after_abs_mean=kphi_abs_a,
        phi_s_drift_mean=(
            float(aft["phi_drift"]) if aft.get("phi_drift") is not None else None
        ),
    )


def strong_like(
    G: Any,
    nodes: Iterable[Any],
    *,
    compute_phi_s: bool = False,
    curvature_hotspot_threshold: float = PHYSICS_CURVATURE_HOTSPOT_CANONICAL,
) -> InteractionResult:
    """Strong-like sequence: [Coupling → Coherence → SelfOrganization].

    Purpose
    -------
    Promote local confinement and sub-EPI formation in regions of high
    curvature, with stabilization immediately after coupling.

    Contracts
    ---------
    - U3: Coupling remains phase-aware.
    - U2: Coherence right after coupling (boundedness).
    - U5: SelfOrganization creates sub-EPIs while preserving parent coherence.
    - Telemetry: flags |K_φ| hotspots; optional Φ_s drift.

    Parameters
    ----------
    curvature_hotspot_threshold : float, default PHYSICS_CURVATURE_HOTSPOT_CANONICAL
        Canonical |K_φ| threshold for hotspot flagging (0.9×π ≈ 2.8274).

    Returns
    -------
    InteractionResult
        Names of operators applied, warnings, and telemetry means.
    """
    snap = _telemetry_before_after(G, compute_phi_s=compute_phi_s)

    ops = [Coupling(), Coherence(), SelfOrganization()]
    applied = _apply_to_nodes(G, nodes, ops)

    aft = _telemetry_after(G, snap, compute_phi_s=compute_phi_s)
    kphi_abs_a = {k: abs(v) for k, v in aft["kphi_a"].items()}  # type: ignore[index]
    hotspot_frac = 0.0
    if kphi_abs_a:
        vals = list(kphi_abs_a.values())
        hotspot_frac = float(
            sum(v >= curvature_hotspot_threshold for v in vals) / len(vals)
        )

    warnings: list[str] = []
    if hotspot_frac > PHYSICS_HOTSPOT_FRACTION_CANONICAL:  # heuristic
        warnings.append(
            (
                "curvature hotspots after Strong-like: "
                f"{hotspot_frac * 100:.1f}% ≥ {PHYSICS_HOTSPOT_FRACTION_CANONICAL * 100:.1f}%"
            )
        )
    if aft.get("phi_drift") is not None and float(aft["phi_drift"]) >= U6_STRUCTURAL_POTENTIAL_LIMIT:
        warnings.append(
            (
                "structural potential drift exceeded threshold: "
                f"{float(aft['phi_drift']):.3f} ≥ {U6_STRUCTURAL_POTENTIAL_LIMIT:.3f}"
            )
        )

    return InteractionResult(
        applied=applied,
        warnings=warnings,
        grad_before_mean=_mean(snap["grad_b"]),  # type: ignore[index]
        grad_after_mean=_mean(aft["grad_a"]),  # type: ignore[index]
        kphi_before_abs_mean=_mean(
            {k: abs(v) for k, v in snap["kphi_b"].items()}
        ),  # type: ignore[index]
        kphi_after_abs_mean=_mean(kphi_abs_a),
        phi_s_drift_mean=(
            float(aft["phi_drift"]) if aft.get("phi_drift") is not None else None
        ),
    )


def gravity_like(
    G: Any,
    nodes: Iterable[Any],
    *,
    compute_phi_s: bool = True,
    quiet: bool = True,
) -> InteractionResult:
    """Gravity-like: telemetry-oriented stabilization [Coherence → Silence].

    Purpose
    -------
    Stabilize and observe while interpreting Φ_s drift as passive confinement
    (no explicit attraction operator).

    Contracts
    ---------
    - U1b: Silence serves as closure/observation window when ``quiet=True``.
    - U2: Coherence preserves boundedness.
    - Telemetry: Φ_s drift is read-only and used as safety signal.

    Parameters
    ----------
    compute_phi_s : bool, default True
        Compute Φ_s before/after and report mean absolute drift.
    quiet : bool, default True
        Append Silence after Coherence for an observation window.

    Returns
    -------
    InteractionResult
        Names of operators applied, warnings, and telemetry means.
    """
    snap = _telemetry_before_after(G, compute_phi_s=compute_phi_s)

    ops: list[Any] = [Coherence()]
    if quiet:
        ops.append(Silence())
    applied = _apply_to_nodes(G, nodes, ops)

    aft = _telemetry_after(G, snap, compute_phi_s=compute_phi_s)
    warnings: list[str] = []
    if aft.get("phi_drift") is not None and float(aft["phi_drift"]) >= U6_STRUCTURAL_POTENTIAL_LIMIT:
        warnings.append(
            (
                "structural potential drift exceeded threshold: "
                f"{float(aft['phi_drift']):.3f} ≥ {U6_STRUCTURAL_POTENTIAL_LIMIT:.3f}"
            )
        )

    return InteractionResult(
        applied=applied,
        warnings=warnings,
        grad_before_mean=_mean(snap["grad_b"]),  # type: ignore[index]
        grad_after_mean=_mean(aft["grad_a"]),  # type: ignore[index]
        kphi_before_abs_mean=_mean(
            {k: abs(v) for k, v in snap["kphi_b"].items()}
        ),  # type: ignore[index]
        kphi_after_abs_mean=_mean(
            {k: abs(v) for k, v in aft["kphi_a"].items()}
        ),  # type: ignore[index]
        phi_s_drift_mean=(
            float(aft["phi_drift"]) if aft.get("phi_drift") is not None else None
        ),
    )


__all__ = [
    "InteractionResult",
    "em_like",
    "weak_like",
    "strong_like",
    "gravity_like",
]