Merge remote-tracking branch 'refs/remotes/origin/develop' into develop

Author: wmayner

changelog.d/eq23-ii-cap.fix.md

@@ -0,0 +1 @@
+Implement Eq. 23 ii(s) cap: system-level partition evaluation now uses GID only (Eqs. 19-20), and phi is capped by ii(s) = min(ii_c, ii_e) where ii_d = min(i_diff_d, i_spec_d). Mechanism-level partition evaluation also uses GID only when `REPERTOIRE_DISTANCE=INTRINSIC_INFORMATION` (the composite metric is a system-level concept).

changelog.d/tied-specified-state.fix.md

@@ -0,0 +1 @@
+Fix permutation symmetry breaking caused by arbitrary tie-breaking among specified states. When multiple states tie for maximum intrinsic information, `integration_value()` now evaluates all tied states and selects the minimum (the "cruelest cut"). The resolved state is back-propagated to `system_state` on the SIA so that downstream consumers (e.g., `phi_structure` congruence filtering) see the correct specified state.

pyphi/new_big_phi/__init__.py

@@ -5,6 +5,7 @@
 
 from collections.abc import Iterable
 from dataclasses import dataclass
+from dataclasses import replace
 from enum import Enum
 from enum import auto
 from enum import unique
@@ -31,6 +32,7 @@
 from pyphi.models.cuts import NullCut
 from pyphi.models.cuts import SystemPartition
 from pyphi.models.mechanism import RepertoireIrreducibilityAnalysis
+from pyphi.models.mechanism import StateSpecification
 from pyphi.models.subsystem import CauseEffectStructure
 from pyphi.models.subsystem import SystemStateSpecification
 from pyphi.parallel import MapReduce
@@ -143,6 +145,31 @@ def ties(self):
     def set_ties(self, ties):
         self._ties = ties
 
+    def resolve_system_state(self) -> None:
+        """Update system_state to reflect the specified states resolved by the MIP.
+
+        When the system has tied specified states, the MIP resolves the tie by
+        selecting the state most vulnerable to the winning partition. This
+        back-propagates that resolution into system_state so that downstream
+        consumers (e.g., congruence filtering in phi_structure) see the correct
+        specified states.
+        """
+        if self.system_state is None:
+            return
+        new_cause = self.system_state.cause
+        new_effect = self.system_state.effect
+        if self.cause is not None and self.cause.specified_state is not None:
+            new_cause = self.cause.specified_state
+        if self.effect is not None and self.effect.specified_state is not None:
+            new_effect = self.effect.specified_state
+        if (
+            new_cause is not self.system_state.cause
+            or new_effect is not self.system_state.effect
+        ):
+            self.system_state = replace(
+                self.system_state, cause=new_cause, effect=new_effect
+            )
+
     def __eq__(self, other):
         return cmp.general_eq(self, other, self._sia_attributes)
 
@@ -288,16 +315,15 @@ def normalization_factor(partition: Cut | GeneralKCut) -> float:
     return 1.0
 
 
-def integration_value(
+def _integration_value_for_state(
     direction: Direction,
     subsystem: Subsystem,
+    cut_subsystem: Subsystem,
     partition: Cut,
-    system_state: SystemStateSpecification,
-    repertoire_distance: str | None = None,
+    specified: StateSpecification,
+    repertoire_distance: str,
 ) -> RepertoireIrreducibilityAnalysis:
-    repertoire_distance = fallback(repertoire_distance, config.REPERTOIRE_DISTANCE)
-    cut_subsystem = subsystem.apply_cut(partition)
-    # TODO(4.0) deal with proliferation of special cases for GID
+    """Compute the integration value for a single specified state."""
     mechanism = purview = subsystem.node_indices
     if repertoire_distance in [
         "GENERALIZED_INTRINSIC_DIFFERENCE",
@@ -307,22 +333,45 @@ def integration_value(
             direction,
             mechanism,
             purview,
-            system_state[direction].state,
-        ).squeeze()[system_state[direction].state]
+            specified.state,
+        ).squeeze()[specified.state]
     else:
         partitioned_repertoire = cut_subsystem.repertoire(
             direction, subsystem.node_indices, subsystem.node_indices
         )
-    ria = subsystem.evaluate_partition(
+    return subsystem.evaluate_partition(
         direction,
         subsystem.node_indices,
         subsystem.node_indices,
         partition,  # pyright: ignore[reportArgumentType] - Cut passed to Bipartition param in IIT 4.0
         partitioned_repertoire=partitioned_repertoire,
         repertoire_distance=repertoire_distance,
-        state=system_state[direction],
+        state=specified,
     )
-    return ria
+
+
+def integration_value(
+    direction: Direction,
+    subsystem: Subsystem,
+    partition: Cut,
+    system_state: SystemStateSpecification,
+    repertoire_distance: str | None = None,
+) -> RepertoireIrreducibilityAnalysis:
+    repertoire_distance = fallback(repertoire_distance, config.REPERTOIRE_DISTANCE)
+    cut_subsystem = subsystem.apply_cut(partition)
+    specified = system_state[direction]
+    tied_specs = specified.ties if specified.ties else (specified,)
+    # When there are tied specified states, evaluate all of them and take the
+    # minimum integration (the "cruelest cut"): among equally-specified states,
+    # the partition should be evaluated against the one it hurts most.
+    best_ria = None
+    for spec in tied_specs:
+        ria = _integration_value_for_state(
+            direction, subsystem, cut_subsystem, partition, spec, repertoire_distance
+        )
+        if best_ria is None or ria.phi < best_ria.phi:
+            best_ria = ria
+    return best_ria
 
 
 def intrinsic_differentiation_value(
@@ -365,19 +414,26 @@ def evaluate_partition(
         directions = Direction.both()
     directions = tuple(directions)
     validate.directions(directions)
+
+    # Eqs. 19-20: system-level partition integration uses GID only.
+    # The ii(s) cap (Eq. 23) is applied separately below.
+    effective_distance = fallback(repertoire_distance, config.REPERTOIRE_DISTANCE)
+    partition_distance = (
+        "GENERALIZED_INTRINSIC_DIFFERENCE"
+        if effective_distance == "INTRINSIC_INFORMATION"
+        else effective_distance
+    )
+
     integration = {
         direction: integration_value(
             direction,
             subsystem,
             partition,
             system_state,
-            repertoire_distance=repertoire_distance,
+            repertoire_distance=partition_distance,
         )
         for direction in directions
     }
-    phi = min(integration[direction].phi for direction in directions)
-    norm = normalization_factor(partition)
-    normalized_phi = phi * norm
 
     intrinsic_differentiation = {
         direction: intrinsic_differentiation_value(
@@ -388,6 +444,19 @@ def evaluate_partition(
         for direction in directions
     }
 
+    phi = min(integration[direction].phi for direction in directions)
+
+    # Eq. 23: φ_s(s) = min{φ_c(s), φ_e(s), ii(s)}
+    # where ii(s) = min_d{min(i_diff_d, i_spec_d)}
+    if effective_distance == "INTRINSIC_INFORMATION":
+        for direction in directions:
+            i_spec = float(system_state[direction].intrinsic_information)
+            i_diff = float(intrinsic_differentiation[direction])
+            phi = min(phi, i_spec, i_diff)
+
+    norm = normalization_factor(partition)
+    normalized_phi = phi * norm
+
     result = SystemIrreducibilityAnalysis(
         phi=phi,
         normalized_phi=normalized_phi,
@@ -548,6 +617,7 @@ def is_disconnecting_partition(partition):
         elif candidate_key == mip_key:
             ties.append(candidate_mip_sia)
     for tied_mip in ties:
+        tied_mip.resolve_system_state()
         tied_mip.set_ties(ties)
 
     if config.CLEAR_SUBSYSTEM_CACHES_AFTER_COMPUTING_SIA:

pyphi/subsystem.py

@@ -811,15 +811,16 @@ def evaluate_partition(
             partitioned repertoires, and the partitioned repertoire.
         """
         repertoire_distance = fallback(repertoire_distance, config.REPERTOIRE_DISTANCE)
+        # Mechanism-level partition evaluation uses GID only.
+        # INTRINSIC_INFORMATION is a system-level composite (Eq. 23).
+        if repertoire_distance == "INTRINSIC_INFORMATION":
+            repertoire_distance = "GENERALIZED_INTRINSIC_DIFFERENCE"
         # TODO(4.0) refactor
         # TODO(4.0) consolidate logic with system level partitions
         if repertoire is None:
             repertoire = self.repertoire(direction, mechanism, purview)
         # TODO(4.0) use same partitioned_repertoire func
-        if repertoire_distance in [
-            "GENERALIZED_INTRINSIC_DIFFERENCE",
-            "INTRINSIC_INFORMATION",
-        ]:
+        if repertoire_distance == "GENERALIZED_INTRINSIC_DIFFERENCE":
             func = metrics.distribution.measures[repertoire_distance]
             assert not isinstance(repertoire, (int, float)), (
                 "GID requires full repertoire"

test/example_networks.py

@@ -2,6 +2,7 @@
 
 import pyphi
 from pyphi import utils
+from pyphi.labels import NodeLabels
 from pyphi.macro import Blackbox
 from pyphi.macro import MacroSubsystem
 from pyphi.network import Network
@@ -754,3 +755,45 @@ def propagation_delay():
     return MacroSubsystem(
         network, cs, network.node_indices, time_scale=time_scale, blackbox=blackbox
     )
+
+
+# Permutation-equivalent pair for symmetry tests
+# ================================================
+
+
+def and_xor_network():
+    """AND-XOR 2-node network. Node 0: AND(0,1), Node 1: XOR(0,1).
+
+    Both nodes receive input from both nodes (all-ones CM).
+    Deterministic transitions:
+        (0,0) -> (0,0), (1,0) -> (0,1), (0,1) -> (0,1), (1,1) -> (1,0)
+    """
+    # fmt: off
+    tpm = np.array([
+        [0, 0],  # (0,0) -> (0,0)
+        [0, 1],  # (1,0) -> (0,1)
+        [0, 1],  # (0,1) -> (0,1)
+        [1, 0],  # (1,1) -> (1,0)
+    ])
+    # fmt: on
+    cm = np.ones((2, 2))
+    return Network(tpm, cm=cm, node_labels=NodeLabels(("AND", "XOR"), tuple(range(2))))
+
+
+def xor_and_network():
+    """XOR-AND 2-node network (AND-XOR with nodes 0 and 1 permuted).
+
+    Both nodes receive input from both nodes (all-ones CM).
+    Deterministic transitions:
+        (0,0) -> (0,0), (1,0) -> (1,0), (0,1) -> (1,0), (1,1) -> (0,1)
+    """
+    # fmt: off
+    tpm = np.array([
+        [0, 0],  # (0,0) -> (0,0)
+        [1, 0],  # (1,0) -> (1,0)
+        [1, 0],  # (0,1) -> (1,0)
+        [0, 1],  # (1,1) -> (0,1)
+    ])
+    # fmt: on
+    cm = np.ones((2, 2))
+    return Network(tpm, cm=cm, node_labels=NodeLabels(("XOR", "AND"), tuple(range(2))))