File size: 4,816 Bytes
d997548
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
#!/usr/bin/env python3
"""
R5-compliant cell splitting.

Rule (R5 Hypothesis 1):
  If S(c, t) >= S* for T_cons >= 3 consecutive steps:
      split c into two new cells.

Split mechanics:
  - Parent cell's neurons are partitioned into two halves.
  - Each child inherits:
      * half the neurons
      * mutated DNA (via dna_kernel.mutate_dna)
      * K reset to [0]*6 (re-measure needed)
  - Parent state becomes 'split'.
  - Children registered in the cell registry.
"""
import json
from dataclasses import asdict
from pathlib import Path
from typing import Optional
from datetime import datetime, timezone

from .saturation import Cell, S_STAR, T_CONSECUTIVE
from .dna_kernel import DNA, mutate_dna, random_dna


SPLIT_COOLDOWN_STEPS = 300

def is_ready_to_split(cell: Cell, t_consecutive: int = T_CONSECUTIVE,
                      s_star: float = S_STAR, current_step: int = None) -> bool:
    """
    Check if cell has S >= S* for at least t_consecutive steps.
    Plus: enforce a cooldown between splits.
    """
    if len(cell.S_history) < t_consecutive:
        return False
    recent = cell.S_history[-t_consecutive:]
    if not all(s >= s_star for s in recent):
        return False
    # Cooldown check
    if current_step is not None:
        last = getattr(cell, "last_split_step", 0)
        if (current_step - last) < SPLIT_COOLDOWN_STEPS:
            return False
    return True


def can_split(cell: Cell, min_neurons: int = 256) -> bool:
    """Cell must have at least min_neurons to split."""
    return cell.num_neurons() >= min_neurons and cell.state == "active"


def split_cell(cell: Cell, dna: DNA, split_id: str) -> tuple:
    """
    Split a cell into two.
    Returns (child_a, child_b, dna_a, dna_b).
    """
    mid = (cell.neuron_start + cell.neuron_end) // 2

    # Children inherit halves of the parent's neuron range
    child_a = Cell(
        cell_id=f"{cell.cell_id}-a",
        layer_idx=cell.layer_idx,
        neuron_start=cell.neuron_start,
        neuron_end=mid,
        state="active",
        parent_id=cell.cell_id,
        created_at=datetime.now(timezone.utc).isoformat(),
    )
    child_b = Cell(
        cell_id=f"{cell.cell_id}-b",
        layer_idx=cell.layer_idx,
        neuron_start=mid,
        neuron_end=cell.neuron_end,
        state="active",
        parent_id=cell.cell_id,
        created_at=datetime.now(timezone.utc).isoformat(),
    )
    # Children inherit cooldown from parent's last split
    child_a.last_split_step = getattr(cell, "last_split_step", 0)
    child_b.last_split_step = getattr(cell, "last_split_step", 0)

    # Mutate DNA for each child
    dna_a = mutate_dna(dna, rate=0.02)
    dna_a.cell_id = child_a.cell_id
    dna_b = mutate_dna(dna, rate=0.02)
    dna_b.cell_id = child_b.cell_id

    return child_a, child_b, dna_a, dna_b


def apply_split(cell: Cell, dna: DNA, registry_path: Optional[Path] = None) -> dict:
    """
    Check + split if ready.
    Returns dict with:
      - split: bool
      - reason: str
      - children: list of cell_ids (if split)
    """
    if not can_split(cell):
        return {"split": False, "reason": "cannot_split", "children": []}

    if not is_ready_to_split(cell):
        return {"split": False, "reason": "S_not_sustained", "children": []}

    child_a, child_b, dna_a, dna_b = split_cell(cell, dna, split_id=cell.cell_id)

    # Parent state -> 'split'
    cell.state = "split"
    cell.splits += 1
    cell.last_split_step = getattr(cell, "last_split_step", 0)

    # Save children + DNA if registry path is provided
    if registry_path:
        registry_path = Path(registry_path)
        registry_path.mkdir(parents=True, exist_ok=True)
        from .saturation import save_cells
        save_cells([child_a, child_b], registry_path / f"{cell.cell_id}_children.json")
        from .dna_kernel import save_dna
        save_dna(dna_a, registry_path / f"{dna_a.cell_id}.dna.json")
        save_dna(dna_b, registry_path / f"{dna_b.cell_id}.dna.json")

    return {
        "split": True,
        "reason": "S_sustained_above_star",
        "children": [child_a.cell_id, child_b.cell_id],
    }


if __name__ == "__main__":
    from .saturation import Cell
    from .dna_kernel import random_dna

    # Create a test cell with S history above S*
    c = Cell(cell_id="cell-test", layer_idx=0, neuron_start=0, neuron_end=1536)
    c.S_history = [0.5, 0.6, 0.87, 0.90, 0.92]  # last 3 above S*=0.85
    c.S = 0.92

    dna = random_dna(c.cell_id, seed=42)

    print(f"Cell: {c.cell_id}, neurons {c.neuron_start}-{c.neuron_end}")
    print(f"S_history (last 3): {c.S_history[-3:]}")
    print(f"Ready to split? {is_ready_to_split(c)}")
    print()

    result = apply_split(c, dna, registry_path=None)
    print(f"Result: {result}")
    print(f"Parent state after split: {c.state}")