"""Atomic public implementation of SelfOrganization (THOL). THOL materializes operational fractality only after its complete U5 proposal is valid. The public operator separates read-only planning from commit and restores the graph if an external monitor or a later validation rejects the operation. """ from __future__ import annotations import logging import math from collections.abc import Mapping from dataclasses import dataclass, replace from typing import Any, ClassVar from ..config.operator_names import SELF_ORGANIZATION from ..constants.aliases import ALIAS_D2EPI, ALIAS_DNFR, ALIAS_THETA, ALIAS_VF from ..constants.canonical import COUPLING_GENTLE, COUPLING_MODERATE from ..glyph_history import next_operator_step from ..types import Glyph, TNFRGraph from ._argument_validation import ( finite_node_real, finite_real, nonnegative_integer, reject_operator_argument, require_list_sink, strict_bool, validate_common_execution_arguments, ) from ._thol_config import resolve_thol_bifurcation_threshold from ._thol_constants import THOL_CHILD_VF_DAMPING, THOL_SUB_EPI_SCALING from ._thol_pressure import propose_thol_pressure from .definitions_base import Operator from .network_stage import GraphTransactionSnapshot from .preconditions import self_organization as _thol_preconditions from .preconditions.self_organization import ( _finite_node_epi, validate_self_organization_strict, ) # Preserve the earlier private reader paths without duplicating their kernels. _active_acceleration_history_length = ( _thol_preconditions._active_acceleration_history_length ) _finite_epi_value = _thol_preconditions._finite_epi_value _OPERATOR = "Self-organization" @dataclass(frozen=True, slots=True) class _BifurcationProposal: sub_node_id: str sub_node_data: Mapping[str, Any] sub_nodes: tuple[Any, ...] hierarchy_children: tuple[Any, ...] sub_epis: tuple[Mapping[str, Any], ...] @dataclass(frozen=True, slots=True) class _ExecutionProposal: d2_epi: float tau: float dnfr_after: float bifurcation: _BifurcationProposal | None subepi_amplitude_alignment: float | None precondition_context: Mapping[str, Any] | None no_bifurcation_expected: bool | None depth_limit_reached: Mapping[str, Any] | None final_parent_epi: float _GraphSnapshot = GraphTransactionSnapshot def _checked_sum(left: float, right: float, label: str) -> float: return finite_real(left + right, operator=_OPERATOR, label=label) def _checked_product(left: float, right: float, label: str) -> float: return finite_real(left * right, operator=_OPERATOR, label=label) def _configured_tau(graph_data: Mapping[str, Any], kwargs: Mapping[str, Any]) -> float: """Compatibility entry point for the shared THOL threshold resolver.""" return resolve_thol_bifurcation_threshold(graph_data, kwargs.get("tau")) def _configured_epi_bounds(graph_data: Mapping[str, Any]) -> tuple[float, float]: """Return the signed scalar EPI interval used by THOL proposals.""" epi_min = finite_real( graph_data.get("EPI_MIN", -1.0), operator=_OPERATOR, label="EPI_MIN" ) epi_max = finite_real( graph_data.get("EPI_MAX", 1.0), operator=_OPERATOR, label="EPI_MAX" ) if epi_min > epi_max: reject_operator_argument(_OPERATOR, "EPI_MIN must not exceed EPI_MAX") return epi_min, epi_max def _existing_list(mapping: Mapping[str, Any], key: str) -> list[Any]: value = mapping.get(key, []) if not isinstance(value, list): reject_operator_argument(_OPERATOR, f"{key} must be a list") return value def _validate_signal_record(signals: Any, *, label: str) -> None: if signals is None: return if not isinstance(signals, Mapping): reject_operator_argument(_OPERATOR, f"{label} must be a mapping or None") for key in ("epi_gradient", "mean_neighbor_epi"): if key in signals: finite_real(signals[key], operator=_OPERATOR, label=f"{label}.{key}") if "phase_variance" in signals: finite_real( signals["phase_variance"], operator=_OPERATOR, label=f"{label}.phase_variance", lower=0.0, ) if "coupling_strength_mean" in signals: finite_real( signals["coupling_strength_mean"], operator=_OPERATOR, label=f"{label}.coupling_strength_mean", lower=0.0, upper=1.0, ) if "neighbor_count" in signals: nonnegative_integer( signals["neighbor_count"], operator=_OPERATOR, label=f"{label}.neighbor_count", ) def _validate_sub_epi_records(records: list[Any]) -> tuple[Mapping[str, Any], ...]: validated: list[Mapping[str, Any]] = [] for index, record in enumerate(records): label = f"sub_epis[{index}]" if not isinstance(record, Mapping): reject_operator_argument(_OPERATOR, f"{label} must be a mapping") if "epi" not in record: reject_operator_argument(_OPERATOR, f"{label}.epi is required") finite_real( record["epi"], operator=_OPERATOR, label=f"{label}.epi", lower=0.0, upper=1.0, ) for key in ("vf", "tau"): if key in record: finite_real( record[key], operator=_OPERATOR, label=f"{label}.{key}", lower=0.0, ) if "d2_epi" in record: finite_real( record["d2_epi"], operator=_OPERATOR, label=f"{label}.d2_epi", ) for key in ("timestamp", "bifurcation_level", "cascade_depth"): if key in record: nonnegative_integer( record[key], operator=_OPERATOR, label=f"{label}.{key}" ) if "metabolized" in record: strict_bool( record["metabolized"], operator=_OPERATOR, label=f"{label}.metabolized", ) if "hierarchy_path" in record and not isinstance( record["hierarchy_path"], list ): reject_operator_argument( _OPERATOR, f"{label}.hierarchy_path must be a list" ) _validate_signal_record(record.get("network_signals"), label=label) validated.append(record) return tuple(validated) def _retarget_bifurcation( proposal: _BifurcationProposal, sub_node_id: str ) -> _BifurcationProposal: """Return one proposal with every child-identity reference updated.""" if ( not proposal.sub_nodes or not proposal.hierarchy_children or not proposal.sub_epis ): raise RuntimeError("THOL bifurcation proposal has incomplete child references") record = dict(proposal.sub_epis[-1]) record["node_id"] = sub_node_id return replace( proposal, sub_node_id=sub_node_id, sub_nodes=(*proposal.sub_nodes[:-1], sub_node_id), hierarchy_children=(*proposal.hierarchy_children[:-1], sub_node_id), sub_epis=(*proposal.sub_epis[:-1], record), ) def _merge_stage_execution_proposals( snapshot: TNFRGraph, proposals: tuple[tuple[Any, _ExecutionProposal], ...], ) -> tuple[tuple[Any, _ExecutionProposal], ...]: """Allocate collision-free THOL children in immutable snapshot-node rank. A parent's ordinary direct-call identifier remains unchanged whenever it does not collide. Cross-target collisions are resolved by advancing that parent's local suffix. Snapshot rank, rather than requested target order, decides which proposal retains a colliding identifier. """ rank = {node: index for index, node in enumerate(snapshot.nodes)} if any(node not in rank for node, _proposal in proposals): raise RuntimeError("THOL stage proposal contains a non-snapshot target") ordered = sorted(proposals, key=lambda item: rank[item[0]]) reserved = set(snapshot.nodes) merged: dict[Any, _ExecutionProposal] = {} for node, proposal in ordered: bifurcation = proposal.bifurcation if bifurcation is None: merged[node] = proposal continue sub_index = len(bifurcation.sub_nodes) - 1 sub_node_id = f"{node}_sub_{sub_index}" while sub_node_id in reserved: sub_index += 1 sub_node_id = f"{node}_sub_{sub_index}" reserved.add(sub_node_id) if sub_node_id != bifurcation.sub_node_id: bifurcation = _retarget_bifurcation(bifurcation, sub_node_id) proposal = replace(proposal, bifurcation=bifurcation) merged[node] = proposal return tuple((node, merged[node]) for node, _proposal in proposals) class SelfOrganization(Operator): """Spawn a coherent sub-EPI through one atomic U5 transaction.""" __slots__ = () name: ClassVar[str] = SELF_ORGANIZATION glyph: ClassVar[Glyph] = Glyph.THOL def _validate_application_preconditions( self, G: TNFRGraph, node: Any, **kw: Any ) -> None: """Validate the optional public gate without writing its telemetry.""" validate_common_execution_arguments(G.graph, kw, operator=_OPERATOR) collect_metrics = bool(kw.get("collect_metrics", False)) or bool( G.graph.get("COLLECT_OPERATOR_METRICS", False) ) if collect_metrics: require_list_sink(G.graph, "operator_metrics", operator=_OPERATOR) monitor = G.graph.get("integrity_monitor") if monitor is not None and not all( callable(getattr(monitor, method, None)) for method in ("before_operator", "after_operator") ): reject_operator_argument( _OPERATOR, "integrity_monitor must provide callable before_operator and " "after_operator methods", ) requested = kw.get("validate_preconditions", True) enabled = G.graph.get("VALIDATE_OPERATOR_PRECONDITIONS", False) if requested and enabled: self._validate_preconditions(G, node, **kw) def _validate_preconditions(self, G: TNFRGraph, node: Any, **kw: Any) -> None: """Delegate to the shared read-only optional THOL gate.""" validate_self_organization_strict(G, node, tau=kw.get("tau")) def _execute(self, G: TNFRGraph, node: Any, **kw: Any) -> None: """Validate, apply and observe THOL as one graph transaction.""" proposal = self._prepare_execution(G, node, kw) snapshot = _GraphSnapshot(G) try: self._execute_transaction(G, node, proposal, kw) except BaseException as failure: monitor = G.graph.get("integrity_monitor") discard_pending = getattr(monitor, "discard_pending_operator", None) if callable(discard_pending): try: discard_pending() except Exception: pass snapshot.restore_after_failure(G, failure) raise def _execute_transaction( self, G: TNFRGraph, node: Any, proposal: _ExecutionProposal, kw: Mapping[str, Any], ) -> None: collect_metrics = bool(kw.get("collect_metrics", False)) or bool( G.graph.get("COLLECT_OPERATOR_METRICS", False) ) validate_equation = bool(kw.get("validate_nodal_equation", False)) or bool( G.graph.get("VALIDATE_NODAL_EQUATION", False) ) state_before = None if collect_metrics or validate_equation: state_before = self._capture_state(G, node) monitor = G.graph.get("integrity_monitor") if monitor is not None: monitor.before_operator(G, node) from . import _validated_execution_window from .network_stage import ( PointwiseStageProposal, _record_histories_and_patterns, ) # Planning is the single source of the direct and staged structural # action. Canonical setters preserve alias precedence and DeltaNFR cache # invalidation; the shared lifecycle helper preserves glyph provenance. self._commit_primary_channels(G, node, proposal) _record_histories_and_patterns( G, ( PointwiseStageProposal( node=node, glyph=self.glyph, payload=proposal, ), ), window=_validated_execution_window(G, kw.get("window")), ) self._commit_proposal(G, node, proposal) if monitor is not None: monitor.after_operator(G, node, self.name) if validate_equation and state_before is not None: from .nodal_equation import validate_nodal_equation validate_nodal_equation( G, node, epi_before=state_before["epi"], epi_after=proposal.final_parent_epi, dt=float(kw.get("dt", 1.0)), operator_name=self.name, strict=bool(G.graph.get("NODAL_EQUATION_STRICT", False)), ) if collect_metrics and state_before is not None: metrics = self._collect_metrics(G, node, state_before) G.graph.setdefault("operator_metrics", []).append(metrics) def _prepare_execution( self, G: TNFRGraph, node: Any, kw: Mapping[str, Any] ) -> _ExecutionProposal: validate_common_execution_arguments(G.graph, kw, operator=_OPERATOR) data = G.nodes[node] parent_epi = _finite_node_epi(data, label="EPI") parent_vf = finite_node_real( data, ALIAS_VF, 0.0, operator=_OPERATOR, label="nu_f", lower=0.0, ) parent_theta = finite_node_real( data, ALIAS_THETA, 0.0, operator=_OPERATOR, label="theta" ) dnfr = finite_node_real( data, ALIAS_DNFR, 0.0, operator=_OPERATOR, label="DeltaNFR" ) # The derivative reconstructed from active EPI history is # authoritative. ALIAS_D2EPI is cached telemetry and may be stale. d2_epi = self._compute_epi_acceleration(G, node) tau = _configured_tau(G.graph, kw) from .factor_contracts import resolve_runtime_operator_factors factors = resolve_runtime_operator_factors( G.graph.get("GLYPH_FACTORS"), self.glyph, G.graph ) acceleration = finite_real( factors["THOL_accel"], operator=_OPERATOR, label="THOL_accel", lower=math.nextafter(0.0, math.inf), ) dnfr_after = propose_thol_pressure(dnfr, d2_epi, acceleration).dnfr_after existing = _existing_list(data, "sub_epis") existing_records = _validate_sub_epi_records(existing) self._validate_subtree(G, node, frozenset()) bifurcation = None depth_limit_reached = None final_records = existing_records final_parent_epi = parent_epi if abs(d2_epi) > tau: depth_limit_reached = self._depth_limit_diagnostic(G, node) if depth_limit_reached is None: bifurcation = self._prepare_bifurcation( G, node, parent_epi=parent_epi, parent_vf=parent_vf, parent_theta=parent_theta, d2_epi=d2_epi, tau=tau, existing_records=existing_records, ) final_records = bifurcation.sub_epis amplitude_alignment = None if final_records: from .metabolism import _subepi_amplitude_alignment_from_records amplitude_alignment = _subepi_amplitude_alignment_from_records( final_records ) preconditions_active = bool(kw.get("validate_preconditions", True)) and bool( G.graph.get("VALIDATE_OPERATOR_PRECONDITIONS", False) ) context = None no_bifurcation = None if preconditions_active: from .preconditions.mutation import record_destabilizer_context context = record_destabilizer_context(G, node, record=False) no_bifurcation = bifurcation is None collect_metrics = bool(kw.get("collect_metrics", False)) or bool( G.graph.get("COLLECT_OPERATOR_METRICS", False) ) if collect_metrics: require_list_sink(G.graph, "operator_metrics", operator=_OPERATOR) self._validate_metric_inputs(G, node, bifurcation, final_records) monitor = G.graph.get("integrity_monitor") if monitor is not None: for method in ("before_operator", "after_operator"): if not callable(getattr(monitor, method, None)): reject_operator_argument( _OPERATOR, f"integrity_monitor.{method} must be callable", ) self._validate_nodal_proposal( G, kw, epi_before=parent_epi, epi_after=final_parent_epi, vf=parent_vf, dnfr=dnfr_after, ) return _ExecutionProposal( d2_epi=d2_epi, tau=tau, dnfr_after=dnfr_after, bifurcation=bifurcation, subepi_amplitude_alignment=amplitude_alignment, precondition_context=context, no_bifurcation_expected=no_bifurcation, depth_limit_reached=depth_limit_reached, final_parent_epi=final_parent_epi, ) def _depth_limit_diagnostic( self, G: TNFRGraph, node: Any ) -> Mapping[str, Any] | None: """Return a committed diagnostic when the configured child depth is full.""" parent_level = nonnegative_integer( G.nodes[node].get("_bifurcation_level", 0), operator=_OPERATOR, label="_bifurcation_level", ) max_depth = nonnegative_integer( G.graph.get("THOL_MAX_BIFURCATION_DEPTH", 5), operator=_OPERATOR, label="THOL_MAX_BIFURCATION_DEPTH", ) if parent_level < max_depth: return None require_list_sink(G.graph, "thol_depth_limits", operator=_OPERATOR) return { "node": node, "depth": parent_level, "max_depth": max_depth, } def _prepare_bifurcation( self, G: TNFRGraph, node: Any, *, parent_epi: float, parent_vf: float, parent_theta: float, d2_epi: float, tau: float, existing_records: tuple[Mapping[str, Any], ...], ) -> _BifurcationProposal: data = G.nodes[node] legacy_propagation = strict_bool( G.graph.get("THOL_PROPAGATION_ENABLED", False), operator=_OPERATOR, label="THOL_PROPAGATION_ENABLED", ) if legacy_propagation: reject_operator_argument( _OPERATOR, "THOL network propagation is outside the DeltaNFR channel; " "use Resonance in a grammar-valid operator word", ) metabolic_enabled = strict_bool( G.graph.get("THOL_METABOLIC_ENABLED", True), operator=_OPERATOR, label="THOL_METABOLIC_ENABLED", ) signals = None if metabolic_enabled: for neighbor in G.neighbors(node): _finite_node_epi(G.nodes[neighbor], label=f"neighbor {neighbor!r} EPI") finite_node_real( G.nodes[neighbor], ALIAS_THETA, 0.0, operator=_OPERATOR, label=f"neighbor {neighbor!r} theta", ) from .metabolism import capture_network_signals signals = capture_network_signals(G, node) _validate_signal_record(signals, label="captured network signals") gradient_weight = COUPLING_MODERATE complexity_weight = COUPLING_GENTLE if signals is not None: gradient_weight = finite_real( G.graph.get("THOL_METABOLIC_GRADIENT_WEIGHT", COUPLING_MODERATE), operator=_OPERATOR, label="THOL_METABOLIC_GRADIENT_WEIGHT", lower=0.0, upper=1.0, ) complexity_weight = finite_real( G.graph.get("THOL_METABOLIC_COMPLEXITY_WEIGHT", COUPLING_GENTLE), operator=_OPERATOR, label="THOL_METABOLIC_COMPLEXITY_WEIGHT", lower=0.0, upper=1.0, ) from .metabolism import compose_subepi_amplitude raw_sub_epi = finite_real( compose_subepi_amplitude( parent_epi, signals, scaling_factor=THOL_SUB_EPI_SCALING, gradient_weight=gradient_weight, complexity_weight=complexity_weight, ), operator=_OPERATOR, label="sub-EPI amplitude proposal", lower=0.0, upper=1.0, ) epi_min, epi_max = _configured_epi_bounds(G.graph) child_min = max(0.0, epi_min) child_max = min(1.0, epi_max) if child_min > child_max: reject_operator_argument( _OPERATOR, "configured EPI interval cannot represent a nonnegative sub-EPI", ) sub_epi = finite_real( max(child_min, min(child_max, raw_sub_epi)), operator=_OPERATOR, label="bounded sub-EPI proposal", lower=child_min, upper=child_max, ) parent_level = nonnegative_integer( data.get("_bifurcation_level", 0), operator=_OPERATOR, label="_bifurcation_level", ) hierarchy_level = nonnegative_integer( data.get("hierarchy_level", 0), operator=_OPERATOR, label="hierarchy_level", ) parent_path = data.get("_hierarchy_path", []) if not isinstance(parent_path, list): reject_operator_argument(_OPERATOR, "_hierarchy_path must be a list") if node in parent_path: reject_operator_argument(_OPERATOR, "_hierarchy_path contains a cycle") child_path = [*parent_path, node] child_level = parent_level + 1 sub_nodes = _existing_list(data, "sub_nodes") for child in sub_nodes: if child not in G: reject_operator_argument( _OPERATOR, f"sub-node reference {child!r} is missing" ) if G.nodes[child].get("parent_node") != node: reject_operator_argument( _OPERATOR, f"sub-node {child!r} has inconsistent parent identity" ) sub_index = len(sub_nodes) sub_node_id = f"{node}_sub_{sub_index}" while sub_node_id in G: sub_index += 1 sub_node_id = f"{node}_sub_{sub_index}" hierarchy = G.graph.get("hierarchy", {}) if not isinstance(hierarchy, dict): reject_operator_argument(_OPERATOR, "hierarchy must be a dictionary") if node in hierarchy: hierarchy_children = hierarchy[node] if not isinstance(hierarchy_children, list): reject_operator_argument( _OPERATOR, "hierarchy children must be stored in a list" ) if tuple(hierarchy_children) != tuple(sub_nodes): reject_operator_argument( _OPERATOR, "graph hierarchy children must match parent sub_nodes", ) else: # sub_nodes is the node-local U5 source of truth for legacy graphs # that predate the redundant graph-level hierarchy index. hierarchy_children = sub_nodes timestamp = next_operator_step(data) child_vf = _checked_product( parent_vf, THOL_CHILD_VF_DAMPING, "sub-node nu_f proposal" ) child_theta = finite_real( parent_theta % math.tau, operator=_OPERATOR, label="sub-node theta proposal", lower=0.0, upper=math.tau, ) from ..constants import DNFR_PRIMARY, EPI_PRIMARY, THETA_PRIMARY, VF_PRIMARY sub_node_data = { EPI_PRIMARY: sub_epi, VF_PRIMARY: child_vf, THETA_PRIMARY: child_theta, DNFR_PRIMARY: 0.0, "parent_node": node, "hierarchy_level": hierarchy_level + 1, "_bifurcation_level": child_level, "_hierarchy_path": child_path, "epi_history": [sub_epi], "glyph_history": [], } record = { "epi": sub_epi, "vf": child_vf, "timestamp": timestamp, "d2_epi": d2_epi, "tau": tau, "node_id": sub_node_id, "metabolized": signals is not None, "network_signals": signals, "bifurcation_level": child_level, "hierarchy_path": child_path, } return _BifurcationProposal( sub_node_id=sub_node_id, sub_node_data=sub_node_data, sub_nodes=(*sub_nodes, sub_node_id), hierarchy_children=(*hierarchy_children, sub_node_id), sub_epis=(*existing_records, record), ) def _validate_metric_inputs( self, G: TNFRGraph, node: Any, bifurcation: _BifurcationProposal | None, final_records: tuple[Mapping[str, Any], ...], ) -> None: if not final_records: _existing_list(G.graph, "sub_epi") self._validate_subtree(G, node, frozenset()) def _validate_subtree( self, G: TNFRGraph, node: Any, ancestors: frozenset[Any] ) -> None: if node in ancestors: reject_operator_argument(_OPERATOR, "sub-EPI hierarchy contains a cycle") if node not in G: reject_operator_argument(_OPERATOR, f"sub-node {node!r} is missing") next_ancestors = ancestors | {node} data = G.nodes[node] _finite_node_epi(data, label=f"sub-node {node!r} EPI") finite_node_real( data, ALIAS_VF, 0.0, operator=_OPERATOR, label=f"sub-node {node!r} nu_f", lower=0.0, ) finite_node_real( data, ALIAS_DNFR, 0.0, operator=_OPERATOR, label=f"sub-node {node!r} DeltaNFR", ) finite_node_real( data, ALIAS_THETA, 0.0, operator=_OPERATOR, label=f"sub-node {node!r} theta", ) records = _existing_list(data, "sub_epis") validated_records = _validate_sub_epi_records(records) children = _existing_list(data, "sub_nodes") hierarchy = G.graph.get("hierarchy", {}) if not isinstance(hierarchy, dict): reject_operator_argument(_OPERATOR, "hierarchy must be a dictionary") if node in hierarchy: hierarchy_children = hierarchy[node] if not isinstance(hierarchy_children, list): reject_operator_argument( _OPERATOR, "hierarchy children must be stored in a list" ) if tuple(hierarchy_children) != tuple(children): reject_operator_argument( _OPERATOR, "graph hierarchy children must match parent sub_nodes", ) if children: for child in children: if child not in G: reject_operator_argument( _OPERATOR, f"sub-node reference {child!r} is missing" ) if G.nodes[child].get("parent_node") != node: reject_operator_argument( _OPERATOR, f"sub-node {child!r} has inconsistent parent identity", ) self._validate_subtree(G, child, next_ancestors) for record in validated_records: child = record.get("node_id") if child is None: continue try: child_exists = child in G listed_child = child in children except TypeError: reject_operator_argument( _OPERATOR, "sub-EPI node_id must be a hashable node identifier" ) if not child_exists: reject_operator_argument( _OPERATOR, f"sub-EPI node_id reference {child!r} is missing" ) # Listed children were already traversed above. Record-only legacy # references remain supported, but receive the same full validation. if not listed_child: self._validate_subtree(G, child, next_ancestors) def _validate_nodal_proposal( self, G: TNFRGraph, kw: Mapping[str, Any], *, epi_before: float, epi_after: float, vf: float, dnfr: float, ) -> None: active = bool(kw.get("validate_nodal_equation", False)) or bool( G.graph.get("VALIDATE_NODAL_EQUATION", False) ) if not active: return from ..errors import TNFRValueError from .nodal_equation import _validate_held_nodal_step try: _validate_held_nodal_step( G.graph, epi_before=epi_before, epi_after=epi_after, dt=kw.get("dt", 1.0), vf=vf, dnfr=dnfr, operator_name=self.name, strict=G.graph.get("NODAL_EQUATION_STRICT", False), ) except TNFRValueError as exc: # Preserve the public operator's preflight error type while sharing # all arithmetic, defaults and clipping semantics with the observer. reject_operator_argument(_OPERATOR, str(exc)) def _commit_primary_channels( self, G: TNFRGraph, node: Any, proposal: _ExecutionProposal ) -> None: """Commit proposed nodal channels through canonical alias boundaries.""" from ..alias import set_attr, set_dnfr set_attr(G.nodes[node], ALIAS_D2EPI, proposal.d2_epi) set_dnfr(G, node, proposal.dnfr_after) def _commit_support_and_hierarchy( self, G: TNFRGraph, node: Any, proposal: _ExecutionProposal ) -> None: """Commit the already merged structural support for one target.""" bifurcation = proposal.bifurcation if bifurcation is None: return G.add_node(bifurcation.sub_node_id, **dict(bifurcation.sub_node_data)) G.nodes[node]["sub_nodes"] = list(bifurcation.sub_nodes) hierarchy = G.graph.setdefault("hierarchy", {}) hierarchy[node] = list(bifurcation.hierarchy_children) G.nodes[node]["sub_epis"] = list(bifurcation.sub_epis) def _validate_merged_stage_support( self, candidate: TNFRGraph, proposals: tuple[tuple[Any, _ExecutionProposal], ...], ) -> None: """Materialize and validate all merged support on a detached candidate.""" generated: set[Any] = set() for node, proposal in proposals: bifurcation = proposal.bifurcation if bifurcation is not None: if ( bifurcation.sub_node_id in candidate or bifurcation.sub_node_id in generated ): raise RuntimeError( "THOL stage child identifier merge is not collision-free" ) generated.add(bifurcation.sub_node_id) self._commit_support_and_hierarchy(candidate, node, proposal) hierarchy = candidate.graph.get("hierarchy", {}) if not isinstance(hierarchy, dict): raise RuntimeError("THOL stage hierarchy merge changed container type") for node, proposal in proposals: self._validate_subtree(candidate, node, frozenset()) bifurcation = proposal.bifurcation if bifurcation is None: continue child = bifurcation.sub_node_id if candidate.nodes[child].get("parent_node") != node: raise RuntimeError("THOL stage child has inconsistent parent identity") if tuple(_existing_list(candidate.nodes[node], "sub_nodes")) != ( bifurcation.sub_nodes ): raise RuntimeError("THOL stage sub-node merge changed") if tuple(hierarchy.get(node, ())) != bifurcation.hierarchy_children: raise RuntimeError("THOL stage hierarchy merge changed") if tuple(_existing_list(candidate.nodes[node], "sub_epis")) != ( bifurcation.sub_epis ): raise RuntimeError("THOL stage sub-EPI merge changed") def _commit_lifecycle_proposal( self, G: TNFRGraph, node: Any, proposal: _ExecutionProposal, *, emit_depth_warning: bool = True, ) -> None: """Commit target-local diagnostics and ordered graph telemetry.""" if proposal.depth_limit_reached is not None: diagnostic = dict(proposal.depth_limit_reached) G.nodes[node]["_thol_depth_limit_reached"] = True G.graph.setdefault("thol_depth_limits", []).append(diagnostic) if emit_depth_warning: self._emit_depth_limit_warning(node, proposal) if proposal.subepi_amplitude_alignment is not None: G.nodes[node][ "_thol_subepi_amplitude_alignment" ] = proposal.subepi_amplitude_alignment if proposal.precondition_context is not None: G.nodes[node]["_mutation_context"] = dict(proposal.precondition_context) if proposal.no_bifurcation_expected is not None: G.nodes[node][ "_thol_no_bifurcation_expected" ] = proposal.no_bifurcation_expected def _emit_depth_limit_warning( self, node: Any, proposal: _ExecutionProposal ) -> None: """Publish one accepted depth-limit warning.""" diagnostic = proposal.depth_limit_reached if diagnostic is None: return logging.getLogger(__name__).warning( "Node %r: THOL child depth %d reached configured maximum %d; " "pressure was reorganized without creating another child.", node, diagnostic["depth"], diagnostic["max_depth"], ) def _commit_proposal( self, G: TNFRGraph, node: Any, proposal: _ExecutionProposal ) -> None: """Commit support and lifecycle data for the direct public call.""" self._commit_support_and_hierarchy(G, node, proposal) self._commit_lifecycle_proposal(G, node, proposal) def _compute_epi_acceleration(self, G: TNFRGraph, node: Any) -> float: """Read the signed value supplied by the shared structural derivative.""" from .nodal_equation import compute_d2epi_dt2 return finite_real( compute_d2epi_dt2(G, node, store=False), operator=_OPERATOR, label="signed EPI acceleration", ) def _collect_metrics( self, G: TNFRGraph, node: Any, state_before: dict[str, Any] ) -> dict[str, Any]: from .metrics import self_organization_metrics return self_organization_metrics( G, node, state_before["epi"], state_before["vf"] ) _CANONICAL_SELF_ORGANIZATION_EXECUTE = SelfOrganization._execute