Fix mutate_add_node innovation deduplication (issue #291)

The dedup key included the newly-allocated node ID, which is always
unique, so the tracker lookup never matched across genomes. Key by
the original connection endpoints instead via a new get_node_split()
method that also shares the node ID across genomes splitting the same
connection in one generation.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: CodeReclaimers

neat/genome.py

@@ -403,7 +403,24 @@ def mutate_add_node(self, config):
 
         # Choose a random connection to split
         conn_to_split = choice(list(self.connections.values()))
-        new_node_id = config.get_new_node_key(self.nodes)
+
+        i, o = conn_to_split.key
+
+        # Get the node ID and innovation numbers for this split from the tracker.
+        # If another genome already split this same connection in this generation,
+        # we get back the same node ID and innovation numbers — this is a core NEAT
+        # requirement for proper crossover alignment (Stanley & Miikkulainen, 2002).
+        new_node_id, in_innovation, out_innovation = (
+            config.innovation_tracker.get_node_split(
+                i, o, lambda: config.get_new_node_key(self.nodes)
+            )
+        )
+
+        # If this genome already has this node (e.g. from crossover with a genome
+        # that previously split this same connection), skip the mutation.
+        if new_node_id in self.nodes:
+            return
+
         ng = self.create_node(config, new_node_id)
 
         # Make the new node as neutral as possible with respect to the
@@ -419,18 +436,6 @@ def mutate_add_node(self, config):
         # the original connection (depending on the activation function of the new node).
         conn_to_split.enabled = False
 
-        i, o = conn_to_split.key
-        
-        # Get innovation numbers for the two new connections
-        # These are keyed by the connection being split, so multiple genomes splitting
-        # the same connection get matching innovation numbers
-        in_innovation = config.innovation_tracker.get_innovation_number(
-            i, new_node_id, 'add_node_in'
-        )
-        out_innovation = config.innovation_tracker.get_innovation_number(
-            new_node_id, o, 'add_node_out'
-        )
-        
         # Add the two new connections with their innovation numbers
         self.add_connection(config, i, new_node_id, 1.0, True, innovation=in_innovation)
         self.add_connection(config, new_node_id, o, conn_to_split.weight, True, innovation=out_innovation)

neat/innovation.py

@@ -96,6 +96,44 @@ def get_innovation_number(self, input_node, output_node, mutation_type='add_conn
 
         return innovation_number
 
+    def get_node_split(self, in_node, out_node, allocate_node_key):
+        """
+        Get or assign innovation numbers and a node ID for splitting a connection.
+
+        If this same connection (in_node -> out_node) has already been split in the
+        current generation, returns the previously assigned (node_id, in_innovation,
+        out_innovation). Otherwise, calls allocate_node_key() to obtain a new node ID
+        and assigns two new innovation numbers.
+
+        This ensures that when multiple genomes independently split the same connection
+        in one generation, they all receive the same node ID and matching innovation
+        numbers — a core requirement of the NEAT algorithm for proper crossover alignment.
+
+        Args:
+            in_node: The input node of the connection being split
+            out_node: The output node of the connection being split
+            allocate_node_key: A callable that returns a new unique node ID.
+                               Only called the first time this split is seen in
+                               the current generation.
+
+        Returns:
+            tuple: (node_id, in_innovation, out_innovation)
+        """
+        key = (in_node, out_node, 'split_node')
+
+        if key in self.generation_innovations:
+            return self.generation_innovations[key]
+
+        node_id = allocate_node_key()
+        self.global_counter += 1
+        in_innovation = self.global_counter
+        self.global_counter += 1
+        out_innovation = self.global_counter
+        result = (node_id, in_innovation, out_innovation)
+        self.generation_innovations[key] = result
+
+        return result
+
     def reset_generation(self):
         """
         Clear generation-specific tracking at the start of a new generation.

tests/test_genome_mutations.py

@@ -402,20 +402,23 @@ def test_add_connection_with_innovation_tracking(self):
     def test_remove_node_deletes_node(self):
         """
         Test that remove_node mutation deletes a node.
-        
+
         Should remove exactly one node from the genome.
         """
         genome = self.create_minimal_genome()
-        
-        # Add some nodes first
+
+        # Add nodes, also adding connections between splits so that
+        # each split targets a distinct connection (avoiding the
+        # within-generation dedup in the innovation tracker).
         for _ in range(3):
             if genome.connections:
                 genome.mutate_add_node(self.config.genome_config)
-        
+                genome.mutate_add_connection(self.config.genome_config)
+
         initial_count = len(genome.nodes)
         if initial_count > len(self.config.genome_config.output_keys):
             genome.mutate_delete_node(self.config.genome_config)
-            
+
             self.assertEqual(len(genome.nodes), initial_count - 1,
                            "Should remove one node")
 

tests/test_innovation.py

@@ -100,24 +100,91 @@ def test_reset_generation_preserves_global_counter(self):
         inn = tracker.get_innovation_number(2, 3, 'add_connection')
         self.assertEqual(inn, 4)
 
-    def test_add_node_mutations_tracked_separately(self):
-        """Test that add_node mutations track both connections separately."""
+    def test_node_split_deduplication(self):
+        """Test that splitting the same connection in one generation produces
+        the same node ID and innovation numbers, regardless of which genome
+        performs the split."""
         tracker = neat.InnovationTracker()
-        
-        # Split connection 0->1 by adding node 2
-        inn_in = tracker.get_innovation_number(0, 2, 'add_node_in')
-        inn_out = tracker.get_innovation_number(2, 1, 'add_node_out')
-        
-        self.assertEqual(inn_in, 1)
-        self.assertEqual(inn_out, 2)
-        
-        # Another genome splits same connection
-        inn_in2 = tracker.get_innovation_number(0, 2, 'add_node_in')
-        inn_out2 = tracker.get_innovation_number(2, 1, 'add_node_out')
-        
-        # Should get same innovation numbers
-        self.assertEqual(inn_in2, inn_in)
-        self.assertEqual(inn_out2, inn_out)
+        next_node = iter(range(10, 100))
+
+        # Genome A splits connection 0->1
+        node_a, inn_in_a, inn_out_a = tracker.get_node_split(
+            0, 1, lambda: next(next_node)
+        )
+        print(f"  Genome A: node={node_a}, in_inn={inn_in_a}, out_inn={inn_out_a}")
+
+        self.assertEqual(node_a, 10)    # first call allocates from iterator
+        self.assertEqual(inn_in_a, 1)
+        self.assertEqual(inn_out_a, 2)
+
+        # Genome B splits the SAME connection 0->1 in the same generation
+        node_b, inn_in_b, inn_out_b = tracker.get_node_split(
+            0, 1, lambda: next(next_node)
+        )
+        print(f"  Genome B: node={node_b}, in_inn={inn_in_b}, out_inn={inn_out_b}")
+
+        # Must get the same node ID and innovation numbers
+        self.assertEqual(node_b, node_a)
+        self.assertEqual(inn_in_b, inn_in_a)
+        self.assertEqual(inn_out_b, inn_out_a)
+
+    def test_node_split_different_connections_get_different_ids(self):
+        """Test that splitting different connections produces different node IDs
+        and innovation numbers."""
+        tracker = neat.InnovationTracker()
+        next_node = iter(range(10, 100))
+
+        node_a, inn_in_a, inn_out_a = tracker.get_node_split(
+            0, 1, lambda: next(next_node)
+        )
+        node_b, inn_in_b, inn_out_b = tracker.get_node_split(
+            2, 3, lambda: next(next_node)
+        )
+        print(f"  Split (0->1): node={node_a}, innovations=({inn_in_a}, {inn_out_a})")
+        print(f"  Split (2->3): node={node_b}, innovations=({inn_in_b}, {inn_out_b})")
+
+        self.assertNotEqual(node_a, node_b)
+        self.assertNotEqual(inn_in_a, inn_in_b)
+        self.assertNotEqual(inn_out_a, inn_out_b)
+
+    def test_node_split_allocator_called_only_once(self):
+        """Test that the allocate_node_key callable is only called the first time
+        a connection is split in a generation (not on dedup hits)."""
+        tracker = neat.InnovationTracker()
+        call_count = [0]
+
+        def allocator():
+            call_count[0] += 1
+            return 42
+
+        tracker.get_node_split(0, 1, allocator)
+        self.assertEqual(call_count[0], 1)
+
+        # Second call for same split should NOT call allocator
+        tracker.get_node_split(0, 1, allocator)
+        self.assertEqual(call_count[0], 1)
+
+    def test_node_split_reset_between_generations(self):
+        """Test that the same connection split in different generations gets
+        different node IDs and innovation numbers."""
+        tracker = neat.InnovationTracker()
+        next_node = iter(range(10, 100))
+
+        node_g1, inn_in_g1, inn_out_g1 = tracker.get_node_split(
+            0, 1, lambda: next(next_node)
+        )
+
+        tracker.reset_generation()
+
+        node_g2, inn_in_g2, inn_out_g2 = tracker.get_node_split(
+            0, 1, lambda: next(next_node)
+        )
+        print(f"  Gen 1: node={node_g1}, innovations=({inn_in_g1}, {inn_out_g1})")
+        print(f"  Gen 2: node={node_g2}, innovations=({inn_in_g2}, {inn_out_g2})")
+
+        self.assertNotEqual(node_g1, node_g2)
+        self.assertNotEqual(inn_in_g1, inn_in_g2)
+        self.assertNotEqual(inn_out_g1, inn_out_g2)
 
     def test_pickle_serialization(self):
         """Test that InnovationTracker can be pickled and unpickled."""