"""Configured neighbor-pressure updates for Dissonance events. Phase, capacity contrast and edge weights enter the selected propagation policy. Its coefficients and threshold alarms are supplied contracts, not laws selected by the nodal identity. A propagation record or a cascade label does not prove a trajectory bifurcation or spontaneous support formation. See ``theory/STRUCTURAL_OPERATORS.md`` and ``docs/contracts/OPERATOR_EVENTS.md`` for the live event, admission and observation boundaries. """ from __future__ import annotations import math from collections.abc import Mapping, MutableMapping from dataclasses import dataclass from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from ..types import NodeId, TNFRGraph from .._exact_time import finite_represented_real from ..alias import get_attr, set_attr from ..constants.aliases import ALIAS_D2EPI, ALIAS_DNFR, ALIAS_THETA, ALIAS_VF from ..constants.canonical import DELTA_PHI_MAX from ..constants.operational import EMERGENT_FREQ_BALANCE_CANONICAL from ..errors import TNFRValueError from ..utils import angle_diff __all__ = [ "DissonancePropagationPolicy", "resolve_dissonance_propagation_policy", "propagate_dissonance", "compute_network_dissonance_field", "detect_bifurcation_cascade", ] _PROPAGATION_MODES = frozenset({"phase_weighted", "uniform", "frequency_weighted"}) _PROPAGATION_EVENTS_KEY = "_oz_propagation" _MISSING = object() @dataclass(frozen=True, slots=True) class DissonancePropagationPolicy: """Admitted phase and magnitude cutoffs for neighbor propagation.""" phase_threshold: float min_propagation: float @dataclass(frozen=True, slots=True) class _DissonanceProposal: """One fully validated neighbour update and its rollback snapshot.""" neighbor: Any data: MutableMapping[str, Any] new_dnfr: float event: dict[str, Any] had_primary_dnfr: bool primary_dnfr_before: Any events_before: list[Any] | None events_length_before: int @dataclass(frozen=True, slots=True) class _DissonancePropagationPlan: """Fully validated OZ neighbor transaction awaiting its commit.""" proposals: tuple[_DissonanceProposal, ...] affected: frozenset[Any] def _finite_scalar(value: Any, label: str) -> float: """Materialize one finite real scalar for the propagation transaction.""" try: return finite_represented_real(value, label)[0] except (TypeError, ValueError) as exc: raise TNFRValueError(str(exc)) from exc def _finite_nonnegative(value: Any, label: str) -> float: result = _finite_scalar(value, label) if result < 0.0: raise TNFRValueError(f"{label} must be nonnegative") return result def resolve_dissonance_propagation_policy( graph_data: Mapping[str, Any], ) -> DissonancePropagationPolicy: """Admit propagation thresholds without preparing or applying an event. These graph-level values are supplied policies, not glyph factors. The phase threshold must be positive because it divides the phase separation. A zero minimum propagation is permitted; zero contributions still do not propagate. Runtime resolution remains conditional on a nonempty outgoing neighborhood. """ phase_threshold = _finite_scalar( graph_data.get("OZ_PHASE_THRESHOLD", DELTA_PHI_MAX), "OZ phase threshold", ) if phase_threshold <= 0.0: raise TNFRValueError("OZ phase threshold must be positive") min_propagation = _finite_nonnegative( graph_data.get("OZ_MIN_PROPAGATION", 0.05), "OZ minimum propagation", ) return DissonancePropagationPolicy(phase_threshold, min_propagation) def _read_node_scalar( data: MutableMapping[str, Any], aliases: tuple[str, ...], default: float, label: str, ) -> float: """Read the first declared alias strictly, without non-finite fallback.""" try: raw = get_attr( data, aliases, default, strict=True, conv=lambda item: item, ) except (OverflowError, TypeError, ValueError) as exc: raise TNFRValueError(f"{label} must be a finite real scalar") from exc return _finite_scalar(raw, label) def _coupling_weight(G: TNFRGraph, source: NodeId, target: NodeId) -> float: """Read outgoing conductance, summing parallel edges when present.""" edge_data = G.get_edge_data(source, target) if edge_data is None: raise TNFRValueError( f"OZ neighbor {target!r} has no outgoing edge from {source!r}" ) if not isinstance(edge_data, Mapping): raise TNFRValueError("OZ edge data must be a mapping") if bool(G.is_multigraph()): weights = [] for key, attributes in edge_data.items(): if not isinstance(attributes, Mapping): raise TNFRValueError( f"OZ parallel edge {key!r} attributes must be a mapping" ) weights.append( _finite_nonnegative( attributes.get("weight", 1.0), f"OZ edge weight {source!r}->{target!r}[{key!r}]", ) ) try: combined = math.fsum(weights) except OverflowError as exc: raise TNFRValueError( f"OZ aggregate edge weight {source!r}->{target!r} must be finite" ) from exc return _finite_nonnegative( combined, f"OZ aggregate edge weight {source!r}->{target!r}" ) return _finite_nonnegative( edge_data.get("weight", 1.0), f"OZ edge weight {source!r}->{target!r}", ) def _prepare_dissonance_propagation( G: TNFRGraph, source_node: NodeId, dissonance_magnitude: float, propagation_mode: str = "phase_weighted", *, dnfr_overrides: Mapping[Any, float] | None = None, ) -> _DissonancePropagationPlan: """Validate and materialize a complete OZ propagation transaction.""" if dnfr_overrides is None: dnfr_overrides = {} elif not isinstance(dnfr_overrides, Mapping): raise TNFRValueError("OZ pressure overrides must be a mapping") magnitude = _finite_nonnegative(dissonance_magnitude, "OZ dissonance magnitude") if ( not isinstance(propagation_mode, str) or propagation_mode not in _PROPAGATION_MODES ): supported = ", ".join(sorted(_PROPAGATION_MODES)) raise TNFRValueError(f"OZ propagation_mode must be one of: {supported}") neighbors = list(G.neighbors(source_node)) if not neighbors: return _DissonancePropagationPlan((), frozenset()) source_data = G.nodes[source_node] if not isinstance(source_data, MutableMapping): raise TNFRValueError("OZ source node data must be mutable mapping") source_theta = _read_node_scalar( source_data, ALIAS_THETA, 0.0, f"OZ source phase {source_node!r}", ) source_vf = _finite_nonnegative( _read_node_scalar( source_data, ALIAS_VF, 1.0, f"OZ source structural frequency {source_node!r}", ), f"OZ source structural frequency {source_node!r}", ) policy = resolve_dissonance_propagation_policy(G.graph) phase_threshold, min_propagation = policy.phase_threshold, policy.min_propagation # Phase one validates the complete outgoing neighborhood and materializes # every proposal without mutating nodal state or propagation telemetry. proposals: list[_DissonanceProposal] = [] seen: set[NodeId] = set() for neighbor in neighbors: if neighbor in seen: raise TNFRValueError(f"OZ outgoing neighborhood repeats node {neighbor!r}") seen.add(neighbor) neighbor_data = G.nodes[neighbor] if not isinstance(neighbor_data, MutableMapping): raise TNFRValueError( f"OZ neighbor data {neighbor!r} must be mutable mapping" ) neighbor_theta = _read_node_scalar( neighbor_data, ALIAS_THETA, 0.0, f"OZ neighbor phase {neighbor!r}", ) neighbor_vf = _finite_nonnegative( _read_node_scalar( neighbor_data, ALIAS_VF, 1.0, f"OZ neighbor structural frequency {neighbor!r}", ), f"OZ neighbor structural frequency {neighbor!r}", ) has_dnfr_override = neighbor in dnfr_overrides if has_dnfr_override: neighbor_dnfr = _finite_scalar( dnfr_overrides[neighbor], f"OZ planned neighbor pressure {neighbor!r}", ) else: neighbor_dnfr = _read_node_scalar( neighbor_data, ALIAS_DNFR, 0.0, f"OZ neighbor pressure {neighbor!r}", ) coupling_weight = _coupling_weight(G, source_node, neighbor) delta_theta = _finite_nonnegative( abs(angle_diff(source_theta, neighbor_theta)), f"OZ phase separation {source_node!r}->{neighbor!r}", ) if delta_theta > phase_threshold: continue phase_weight = _finite_nonnegative( 1.0 - (delta_theta / phase_threshold), f"OZ phase weight {source_node!r}->{neighbor!r}", ) if propagation_mode == "frequency_weighted": frequency_denominator = _finite_scalar( max(neighbor_vf, source_vf, 1e-10), f"OZ frequency denominator {source_node!r}->{neighbor!r}", ) freq_weight = _finite_nonnegative( min(neighbor_vf, source_vf) / frequency_denominator, f"OZ frequency weight {source_node!r}->{neighbor!r}", ) else: freq_weight = 1.0 weighted_magnitude = _finite_nonnegative( magnitude * coupling_weight, f"OZ weighted magnitude {source_node!r}->{neighbor!r}", ) phase_magnitude = _finite_nonnegative( weighted_magnitude * phase_weight, f"OZ phase-weighted magnitude {source_node!r}->{neighbor!r}", ) propagated_dnfr = _finite_nonnegative( phase_magnitude * freq_weight, f"OZ propagated pressure {source_node!r}->{neighbor!r}", ) if propagated_dnfr == 0.0 or propagated_dnfr < min_propagation: continue new_dnfr = _finite_scalar( neighbor_dnfr + propagated_dnfr, f"OZ proposed pressure {neighbor!r}", ) events = neighbor_data.get(_PROPAGATION_EVENTS_KEY, _MISSING) if events is _MISSING: events_before = None events_length_before = 0 elif isinstance(events, list): events_before = events events_length_before = len(events) else: raise TNFRValueError( f"{_PROPAGATION_EVENTS_KEY} for neighbor {neighbor!r} " "must be a list" ) had_primary_dnfr = ALIAS_DNFR[0] in neighbor_data if has_dnfr_override and had_primary_dnfr: primary_dnfr_before = neighbor_dnfr else: primary_dnfr_before = neighbor_data.get(ALIAS_DNFR[0], _MISSING) proposals.append( _DissonanceProposal( neighbor=neighbor, data=neighbor_data, new_dnfr=new_dnfr, event={ "from_node": source_node, "magnitude": propagated_dnfr, "phase_weight": phase_weight, "coupling_weight": coupling_weight, }, had_primary_dnfr=had_primary_dnfr, primary_dnfr_before=primary_dnfr_before, events_before=events_before, events_length_before=events_length_before, ) ) return _DissonancePropagationPlan( tuple(proposals), frozenset(proposal.neighbor for proposal in proposals) ) def _restore_dissonance_proposals( proposals: tuple[_DissonanceProposal, ...] | list[_DissonanceProposal], ) -> None: """Restore proposal targets and telemetry containers in reverse order.""" for proposal in reversed(proposals): if proposal.had_primary_dnfr: proposal.data[ALIAS_DNFR[0]] = proposal.primary_dnfr_before else: proposal.data.pop(ALIAS_DNFR[0], None) if proposal.events_before is None: proposal.data.pop(_PROPAGATION_EVENTS_KEY, None) else: del proposal.events_before[proposal.events_length_before :] proposal.data[_PROPAGATION_EVENTS_KEY] = proposal.events_before def _rollback_dissonance_propagation(plan: _DissonancePropagationPlan) -> None: """Undo a successfully committed internal propagation plan.""" _restore_dissonance_proposals(plan.proposals) def _commit_dissonance_propagation( plan: _DissonancePropagationPlan, ) -> set[Any]: """Commit one validated plan, rolling back any unexpected write failure.""" # The public Dissonance path prepares before applying local OZ. Reject a # stale plan before the first neighbor write if lifecycle hooks changed one # of the fields that the propagation transaction owns. for proposal in plan.proposals: if proposal.had_primary_dnfr: if ( ALIAS_DNFR[0] not in proposal.data or proposal.data[ALIAS_DNFR[0]] != proposal.primary_dnfr_before ): raise TNFRValueError( f"OZ propagation plan for {proposal.neighbor!r} became stale" ) elif ALIAS_DNFR[0] in proposal.data: raise TNFRValueError( f"OZ propagation plan for {proposal.neighbor!r} became stale" ) current_events = proposal.data.get(_PROPAGATION_EVENTS_KEY, _MISSING) if proposal.events_before is None: events_unchanged = current_events is _MISSING else: events_unchanged = ( current_events is proposal.events_before and len(proposal.events_before) == proposal.events_length_before ) if not events_unchanged: raise TNFRValueError( f"OZ propagation telemetry for {proposal.neighbor!r} became stale" ) attempted: list[_DissonanceProposal] = [] try: for proposal in plan.proposals: attempted.append(proposal) proposal.data[ALIAS_DNFR[0]] = proposal.new_dnfr if proposal.events_before is None: proposal.data[_PROPAGATION_EVENTS_KEY] = [proposal.event] else: proposal.events_before.append(proposal.event) except BaseException: _restore_dissonance_proposals(attempted) raise return set(plan.affected) def propagate_dissonance( G: TNFRGraph, source_node: NodeId, dissonance_magnitude: float, propagation_mode: str = "phase_weighted", ) -> set[NodeId]: """Atomically propagate OZ pressure to phase-compatible outgoing neighbors. The finite proposal is ``magnitude * conductance * phase_weight * frequency_weight``. Directed graphs use outgoing arcs and parallel conductances are summed. Validation and proposal materialization precede every nodal or telemetry write. """ plan = _prepare_dissonance_propagation( G, source_node, dissonance_magnitude, propagation_mode ) return _commit_dissonance_propagation(plan) def compute_network_dissonance_field( G: TNFRGraph, source_node: NodeId, radius: int = 2, ) -> dict[NodeId, float]: """Compute dissonance field propagation up to radius hops. Returns dict mapping node -> dissonance_level for all nodes within radius hops of source. Parameters ---------- G : TNFRGraph Network source_node : NodeId OZ application point radius : int Maximum propagation distance (default 2) Returns ------- dict[NodeId, float] Mapping of affected nodes to dissonance level Notes ----- Uses exponential decay: dissonance_level = source_dnfr * (0.5 ** distance) Only nodes reachable via paths (connected) are included in the field. Examples -------- >>> from tnfr.structural import create_nfr >>> from tnfr.operators.definitions import Dissonance >>> from tnfr.dynamics.propagation import compute_network_dissonance_field >>> >>> G, node0 = create_nfr("source") >>> # Create path topology: 0-1-2-3 >>> for i in range(1, 4): ... G.add_node(i) ... G.add_edge(i-1, i) >>> >>> Dissonance()(G, node0) >>> field = compute_network_dissonance_field(G, node0, radius=2) >>> # Returns: {1: high, 2: medium} (node 3 beyond radius) """ import networkx as nx field = {} source_dnfr = abs(float(get_attr(G.nodes[source_node], ALIAS_DNFR, 0.0))) # Get decay factor from graph config decay_factor = float( G.graph.get("OZ_DECAY_FACTOR", EMERGENT_FREQ_BALANCE_CANONICAL) ) # BFS to propagate with distance decay # Optimized implementation using single_source_shortest_path_length # This is O(N+E) instead of O(R*N*(N+E)) lengths = nx.single_source_shortest_path_length(G, source_node, cutoff=radius) for node, distance in lengths.items(): if distance == 0: continue decay = decay_factor**distance field[node] = source_dnfr * decay return field def detect_bifurcation_cascade( G: TNFRGraph, source_node: NodeId, threshold: float = EMERGENT_FREQ_BALANCE_CANONICAL, ) -> list[NodeId]: """Report acceleration crossings with recorded OZ input from one source. This advisory diagnostic joins a historical propagation record to a current three-sample acceleration observation. The propagation records have no clock or before/after acceleration; the join therefore establishes neither temporal proximity nor a causal bifurcation cascade. Parameters ---------- G : TNFRGraph Network containing nodes source_node : NodeId Node where OZ was applied threshold : float Finite nonnegative acceleration threshold in the selected history basis. Returns ------- list[NodeId] Neighbors with matching source records and observed threshold crossings. Notes ----- A candidate must have a positive propagation record naming ``source_node`` and an available acceleration whose magnitude strictly exceeds the threshold. Missing history is not a measured zero and cannot establish a crossing. All observations are validated before telemetry is written. This function does not establish grammar admission or an executable THOL birth proposal. The historical ``_bifurcation_cascade`` and ``_bifurcation_ready`` keys are retained for compatibility. ``triggered_by`` is a legacy source label, not a causal claim; ``causal_attribution`` is explicitly false. Stored markers are snapshots, not continuously refreshed admission evidence. Examples -------- >>> from tnfr.structural import create_nfr >>> from tnfr.operators.definitions import Emission, Dissonance >>> from tnfr.dynamics.propagation import detect_bifurcation_cascade >>> >>> G, node0 = create_nfr("source", epi=0.5, vf=1.2) >>> # Add neighbors with EPI history >>> for i in range(3): ... G.add_node(f"n{i}") ... G.add_edge(node0, f"n{i}") ... G.nodes[f"n{i}"]["_epi_history"] = [0.3, 0.45, 0.55] >>> >>> Dissonance()(G, node0, propagate_to_network=True) >>> cascade = detect_bifurcation_cascade(G, node0) >>> print(f"Cascade size: {len(cascade)}") See Also -------- tnfr.operators.nodal_equation.observe_structural_acceleration : History evidence tnfr.dynamics.bifurcation.get_bifurcation_paths : Identify viable paths """ from ..operators.nodal_equation import observe_structural_acceleration threshold = _finite_nonnegative(threshold, "Cascade acceleration threshold") observations = [] # Get neighbors affected by propagation neighbors = list(G.neighbors(source_node)) for neighbor in neighbors: data = G.nodes[neighbor] events = data.get(_PROPAGATION_EVENTS_KEY, []) if not isinstance(events, list): raise TNFRValueError(f"{_PROPAGATION_EVENTS_KEY} must be a list") matching = [ event for event in events if isinstance(event, Mapping) and event.get("from_node", _MISSING) == source_node ] magnitudes = [ _finite_nonnegative(event.get("magnitude"), "Recorded OZ magnitude") for event in matching ] if not any(value > 0.0 for value in magnitudes): continue observation = observe_structural_acceleration(G, neighbor) if observation.available: observations.append((neighbor, observation)) cascade_nodes = [] for neighbor, observation in observations: d2epi_neighbor = observation.value set_attr(G.nodes[neighbor], ALIAS_D2EPI, d2epi_neighbor) if abs(d2epi_neighbor) > threshold: cascade_nodes.append(neighbor) G.nodes[neighbor]["_bifurcation_cascade"] = { "triggered_by": source_node, "causal_attribution": False, "evidence_scope": "recorded_source_and_current_acceleration", "history_source": observation.source, "time_basis": observation.time_basis, "d2epi": d2epi_neighbor, "threshold": threshold, } G.nodes[neighbor]["_bifurcation_ready"] = True return cascade_nodes