THE KEY EMERGES survival_kernel.py
The reusable survival engine.
This is the central architectural claim of the demo:
the same kernel operates at multiple levels without modification.
```
Level 1: signal -> structure
Substrate: ExposureStream (noisy point clouds)
Candidates: TransformCandidates (lens parameters)
Fitness: invariance across exposures under transform
Level 2: structure -> composition
Substrate: PromotedStructure pair
Candidates: CompositionCandidates (relation parameters)
Fitness: structural compatibility under world admissibility
```
Same kernel. New substrate. No modifications at the level of kernel logic.
This module contains zero signal-specific or key-specific assumptions.
It operates on whatever substrate, candidate factory, and evaluator
are passed to it. The architectural separation is enforced by the
interface not by convention.
CODE COMMENT CONTRACT (from spec):
At kernel entry and at each reuse site, comments must explicitly state:
Same kernel, new substrate. No modifications for this level.
SURVIVAL LOOP:
1. Initialize population within admissible parameter bounds
2. Evaluate each candidate (admissibility first, then fitness)
3. Hard-veto inadmissible candidates
4. Rank admissible survivors by fitness
5. Select elite fraction to carry forward
6. Fill population via mutation and crossover
7. Track promotion criteria
8. Repeat for G generations or until promotion threshold met
9. Emit renderer snapshots throughout
FAILURE IS REAL:
The kernel produces genuine failures inadmissible candidates,
non-surviving candidates, populations that fail to converge.
These are not scripted. They are the system doing its job.
The renderer observes them. They must not be suppressed.
from **future** import annotations
import numpy as np
import time
from dataclasses import dataclass, field
from typing import Any, Callable, List, Optional, Tuple
from core_types import (
ArmID, StageLabel, SurvivalConfig, SurvivalResult,
TransformCandidate, EvaluationResult, RendererSnapshot
)
#
# Generic candidate protocol
#
# The survival kernel does not know whether it is operating on
# TransformCandidates or CompositionCandidates. It receives:
# - a list of candidates (any type with a candidate_id field)
# - a factory function that generates new candidates
# - a mutator function that produces mutated copies
# - a crossover function that recombines two candidates
# - an evaluator function that returns EvaluationResult
# - a promotion gate function that returns bool
#
# All domain-specific logic lives in the passed functions.
# The kernel owns only the loop structure.
CandidateFactory = Callable[[int], List[Any]] # gen -> [candidates]
Mutator = Callable[[Any], Any] # candidate -> candidate
Crossover = Callable[[Any, Any, str], Any] # (a, b, id) -> candidate
Evaluator = Callable[[Any], EvaluationResult] # candidate -> result
PromotionGate = Callable[[List[EvaluationResult]], bool] # history -> bool
#
# Promotion gate implementations
#
def make_threshold_gate(threshold: float,
stability_gens: int = 5) -> PromotionGate:
Standard promotion gate: promote when best score exceeds threshold
for stability_gens consecutive generations.
```
Promotion is not triggered by timer or generation count alone.
It requires sustained performance the structure must hold,
not just peak.
The gate is stateful it tracks consecutive generations above
threshold. The state is reset if score drops below threshold.
"""
consecutive = [0] # mutable counter in closure
def gate(eval_history: List[EvaluationResult]) -> bool:
"""
eval_history: all evaluations from the current generation.
Returns True when promotion criteria are met.
"""
admissible = [e for e in eval_history if e.is_admissible]
if not admissible:
consecutive[0] = 0
return False
best_score = max(e.invariance_score for e in admissible)
if best_score >= threshold:
consecutive[0] += 1
else:
consecutive[0] = 0
return consecutive[0] >= stability_gens
gate.reset = lambda: consecutive.__setitem__(0, 0)
return gate
```
#
# Renderer state extraction (passive)
#
def make_renderer_snapshot(frame_id: int,
stage: StageLabel,
generation: int,
population: List[Any],
eval_results: List[EvaluationResult],
promoted: bool,
overlay_text: Optional[str] = None,
extra: Optional[dict] = None) -> RendererSnapshot:
Extract a renderer snapshot from current kernel state.
```
This is passive observation it reads state, never drives it.
No timer-driven events. Every snapshot corresponds to a real
computational event: a generation completing, a promotion firing,
an admissibility veto occurring.
"""
n_admissible = sum(1 for e in eval_results if e.is_admissible)
n_inadmissible = sum(1 for e in eval_results if not e.is_admissible)
scores = [e.invariance_score for e in eval_results if e.is_admissible]
fitness_dist = np.array(scores) if scores else np.array([0.0])
candidate_views = []
for cand, result in zip(population, eval_results):
view = {
"candidate_id": result.candidate_id,
"is_admissible": result.is_admissible,
"score": result.invariance_score,
"failure_reason": result.failure_reason,
}
# Attach params if candidate has them (works for both transform and composition)
if hasattr(cand, 'params'):
view["params"] = cand.params.tolist()
candidate_views.append(view)
snapshot = RendererSnapshot(
frame_id=frame_id,
stage=stage,
timestamp_gen=generation,
fitness_distribution=fitness_dist,
admissibility_failures=n_inadmissible,
candidate_views=candidate_views,
overlay_text=overlay_text
)
if extra:
snapshot.signal_views.update(extra)
return snapshot
```
#
# Core survival loop
#
def run_survival_kernel(
population: List[Any],
evaluator: Evaluator,
mutator: Mutator,
crossover_fn: Crossover,
gate: PromotionGate,
config: SurvivalConfig,
stage: StageLabel,
arm_id: Optional[ArmID] = None,
snapshot_callback: Optional[Callable[[RendererSnapshot], None]] = None,
verbose: bool = True
) -> SurvivalResult:
Run the survival kernel on an initialized population.
```
Same kernel, new substrate. No modifications for this level.
This function is called identically for:
- arm 1 lens search (population = TransformCandidates)
- arm 2 lens search (population = TransformCandidates)
- composition search (population = CompositionCandidates)
The only differences are in the population contents and the
evaluator/mutator/crossover functions passed in. The loop
structure, selection logic, and promotion tracking are identical.
Parameters
----------
population : initial population (pre-initialized by caller)
evaluator : candidate -> EvaluationResult
mutator : candidate -> mutated candidate
crossover_fn : (candidate_a, candidate_b, child_id) -> candidate
gate : promotion gate (stateful)
config : SurvivalConfig
stage : StageLabel for renderer snapshots
arm_id : which arm (None for composition level)
snapshot_callback : optional function called with each RendererSnapshot
verbose : print generation summaries to stdout
Returns
-------
SurvivalResult with full history for renderer and downstream modules
"""
rng = np.random.default_rng(config.rng_seed)
# History tracking store summaries, not full objects
# Storing full candidate objects across 100 gens x 40 candidates
# accumulates significant memory. We store:
# - best score per generation (for convergence plot)
# - summary eval stats per generation (for renderer)
# - the best candidate per generation (for promotion tracking)
# Full population objects are NOT retained across generations.
population_history: List[List[Any]] = [] # best candidate per gen only
fitness_history: List[List[EvaluationResult]] = [] # summary evals only
best_per_generation: List[float] = []
frame_id = 0
total_evaluated = 0
total_inadmissible = 0
total_non_surviving = 0
promotion_generation: Optional[int] = None
promoted_candidate_id: Optional[str] = None
best_candidate_object: Optional[Any] = None # actual evolved best
best_score_ever: float = 0.0
n_elite = max(1, int(config.population_size * config.elite_fraction))
if verbose:
label = f"arm={arm_id.name}" if arm_id else "composition"
print(f"\n{'='*60}")
print(f"SURVIVAL KERNEL {stage.name} [{label}]")
print(f" population={config.population_size}, "
f"generations={config.max_generations}, "
f"threshold={config.promotion_threshold}, "
f"stability={config.promotion_stability_gens}")
print(f"{'='*60}")
start_time = time.time()
for generation in range(config.max_generations):
# --- Evaluate current population ---
eval_results: List[EvaluationResult] = []
for candidate in population:
result = evaluator(candidate)
eval_results.append(result)
total_evaluated += 1
if not result.is_admissible:
# Hard veto inadmissible, not non-surviving
# These are tracked separately per the code comment contract
total_inadmissible += 1
elif result.invariance_score <= config.promotion_threshold:
total_non_surviving += 1
# --- Compute generation statistics ---
admissible = [(c, e) for c, e in zip(population, eval_results)
if e.is_admissible]
inadmissible_count = len(population) - len(admissible)
if admissible:
scores = [e.invariance_score for _, e in admissible]
best_score = max(scores)
mean_score = float(np.mean(scores))
# Track actual best candidate object across all generations
best_this_gen = max(admissible, key=lambda ce: ce[1].invariance_score)
if best_score > best_score_ever:
best_score_ever = best_score
best_candidate_object = best_this_gen[0]
else:
best_score = 0.0
mean_score = 0.0
best_per_generation.append(best_score)
# --- Store history for renderer summary only, not full population ---
# Store only the best candidate and a compact eval summary.
# Full population objects are discarded after selection.
if admissible:
best_cand_gen, best_eval_gen = max(
zip(population, eval_results),
key=lambda ce: ce[1].invariance_score if ce[1].is_admissible else -1
)
population_history.append([best_cand_gen])
else:
population_history.append([population[0]])
# Store compact eval results (drop large diagnostics dicts to save memory)
compact_evals = [
EvaluationResult(
candidate_id=e.candidate_id,
is_admissible=e.is_admissible,
invariance_score=e.invariance_score,
failure_reason=e.failure_reason,
diagnostics={} # drop diagnostics from history
)
for e in eval_results
]
fitness_history.append(compact_evals)
# --- Renderer snapshot ---
overlay = _generation_overlay_text(
generation, best_score, inadmissible_count,
len(population), promotion_generation is not None
)
snapshot = make_renderer_snapshot(
frame_id=frame_id,
stage=stage,
generation=generation,
population=population,
eval_results=eval_results,
promoted=promotion_generation is not None,
overlay_text=overlay
)
frame_id += 1
if snapshot_callback:
snapshot_callback(snapshot)
if verbose and (generation % 10 == 0 or generation < 5):
print(f" gen {generation:3d}: best={best_score:.4f}, "
f"mean={mean_score:.4f}, "
f"admissible={len(admissible)}/{len(population)}, "
f"vetoed={inadmissible_count}")
# --- Check promotion gate ---
if promotion_generation is None:
if gate(eval_results):
promotion_generation = generation
if admissible:
best_cand, best_eval = max(admissible,
key=lambda ce: ce[1].invariance_score)
promoted_candidate_id = best_cand.candidate_id
# Lock the actual evolved best at promotion time
if best_candidate_object is None:
best_candidate_object = best_cand
if verbose:
print(f"\n >>> PROMOTION at generation {generation} "
f"(score={best_score:.4f}) <<<")
print(f" >>> Promoted: {promoted_candidate_id} <<<\n")
# Continue running to completion do not short-circuit
# The renderer needs to show the post-promotion state
# --- Selection and reproduction ---
if not admissible:
# Full population was inadmissible reinitialize with noise
# This is a real event, not an error. The search space may
# have been initialized poorly. The kernel recovers.
if verbose and generation % 10 == 0:
print(f" gen {generation:3d}: WARNING all candidates "
f"inadmissible, partial reinit")
# Keep existing population, apply strong mutation to escape
new_population = [mutator(c) for c in population]
new_population = [mutator(c) for c in new_population] # double mutation
population = new_population
continue
# Sort admissible by score (descending)
admissible_sorted = sorted(admissible,
key=lambda ce: ce[1].invariance_score,
reverse=True)
# Elite: carry forward top fraction unchanged
elite_candidates = [c for c, _ in admissible_sorted[:n_elite]]
# Fill rest of population
new_population = list(elite_candidates)
while len(new_population) < config.population_size:
roll = rng.random()
if roll < config.crossover_rate and len(admissible_sorted) >= 2:
# Tournament selection for parents
p1 = _tournament_select(admissible_sorted, rng)
p2 = _tournament_select(admissible_sorted, rng)
child_id = f"g{generation}_x{len(new_population)}"
child = crossover_fn(p1, p2, child_id)
child = mutator(child)
new_population.append(child)
else:
# Mutate from elite or admissible pool
parent_idx = rng.integers(0, len(admissible_sorted))
parent, _ = admissible_sorted[parent_idx]
child = mutator(parent)
new_population.append(child)
population = new_population[:config.population_size]
elapsed = time.time() - start_time
if verbose:
print(f"\n KERNEL COMPLETE in {elapsed:.2f}s")
print(f" Total evaluated: {total_evaluated}")
print(f" Inadmissible (hard-vetoed): {total_inadmissible} "
f"({100*total_inadmissible/max(total_evaluated,1):.1f}%)")
print(f" Admissible non-surviving: {total_non_surviving}")
print(f" Promoted: {promotion_generation is not None} "
f"(gen {promotion_generation})")
if promoted_candidate_id:
print(f" Best candidate: {promoted_candidate_id}")
return SurvivalResult(
arm_id=arm_id,
stage=stage,
config=config,
population_history=population_history,
fitness_history=fitness_history,
best_per_generation=best_per_generation,
promoted=promotion_generation is not None,
promotion_generation=promotion_generation,
promoted_candidate_id=promoted_candidate_id,
best_candidate_object=best_candidate_object,
total_evaluated=total_evaluated,
total_inadmissible=total_inadmissible,
total_non_surviving=total_non_surviving
)
```
#
# Tournament selection (internal utility)
#
def _tournament_select(admissible_sorted: List[Tuple[Any, EvaluationResult]],
rng: np.random.Generator,
tournament_size: int = 3) -> Any:
Tournament selection from the admissible population.
Returns the candidate object (not the eval result).
n = len(admissible_sorted)
indices = rng.integers(0, n, size=min(tournament_size, n))
# admissible_sorted is already sorted descending by score
# so the lowest index wins the tournament
winner_idx = min(indices)
candidate, _ = admissible_sorted[winner_idx]
return candidate
#
# Generation overlay text (passive describes what just happened)
#
def _generation_overlay_text(generation: int,
best_score: float,
inadmissible_count: int,
pop_size: int,
promoted: bool) -> str:
Sparse on-screen text that describes what is actually happening.
Text follows the process it does not lead it.
if generation == 0:
return Candidate lenses tested under survival.
if inadmissible_count == pop_size:
return All candidates inadmissible constraint enforced.
if inadmissible_count > pop_size // 2:
return Most candidates vetoed by world constraint.
if promoted:
return Invariance survived. Structure promoted.
if best_score > 0.8:
return High-invariance candidate emerging.
if best_score > 0.5:
return Partial invariance found. Survival pressure continues.
return Candidate lenses tested under survival.
#
# Population initializer (convenience used by main.py)
#
def initialize_population(candidate_factory: CandidateFactory,
population_size: int,
generation: int = 0) -> List[Any]:
Initialize a population using the provided factory function.
```
Initialization occurs within admissible bounds the factory
is responsible for generating lawful candidates.
The kernel does not know what kind of candidates are being created.
The factory encapsulates all domain-specific initialization logic.
This restricts search to lawful questions, not known answers.
"""
return candidate_factory(population_size)
```
#
# Sanity check
#
if **name** == **main**:
import sys
from signal_model import build_signal_arms
from transform_families import (
random_candidate, mutate_candidate, crossover_candidates,
apply_candidate, ARM1_PARAM_BOUNDS, ARM2_PARAM_BOUNDS
)
from invariance_metrics import evaluate_candidate
from world_constraint import DEFAULT_WORLD
from core_types import ArmID, StageLabel, SurvivalConfig
```
stream_1, stream_2, c1, c2 = build_signal_arms(seed=42)
world = DEFAULT_WORLD
print("THE KEY EMERGES survival_kernel.py sanity check")
print("=" * 60)
print()
# -----------------------------------------------------------------------
# ARM 1 lens search
# Same kernel, new substrate (arm 1 signal)
# -----------------------------------------------------------------------
print("LEVEL 1A: Arm 1 lens search")
print("Same kernel, new substrate. No modifications for this level.")
print()
rng_1 = np.random.default_rng(0)
def factory_arm1(n: int) -> List[Any]:
return [random_candidate(ArmID.ARM_1, f"a1_g0_{i}", rng=rng_1)
for i in range(n)]
def evaluator_arm1(cand: Any) -> EvaluationResult:
return evaluate_candidate(cand, stream_1, world)
def mutator_arm1(cand: Any) -> Any:
return mutate_candidate(cand, rng=rng_1)
def crossover_arm1(a: Any, b: Any, child_id: str) -> Any:
return crossover_candidates(a, b, child_id, rng=rng_1)
config_1 = SurvivalConfig(
population_size=25,
max_generations=40,
elite_fraction=0.2,
mutation_rate=0.35,
crossover_rate=0.5,
promotion_threshold=0.80,
promotion_stability_gens=3,
rng_seed=0
)
gate_1 = make_threshold_gate(
config_1.promotion_threshold,
config_1.promotion_stability_gens
)
pop_1 = initialize_population(factory_arm1, config_1.population_size)
result_1 = run_survival_kernel(
population=pop_1,
evaluator=evaluator_arm1,
mutator=mutator_arm1,
crossover_fn=crossover_arm1,
gate=gate_1,
config=config_1,
stage=StageLabel.LENS_SEARCH,
arm_id=ArmID.ARM_1,
verbose=True
)
print(f"\nArm 1 result: promoted={result_1.promoted}, "
f"gen={result_1.promotion_generation}, "
f"candidate={result_1.promoted_candidate_id}")
print(f" Best score trajectory (every 10 gens): "
f"{[round(result_1.best_per_generation[g], 3) for g in range(0, len(result_1.best_per_generation), 10)]}")
print()
print("-" * 60)
print()
# -----------------------------------------------------------------------
# ARM 2 lens search
# Same kernel, new substrate (arm 2 signal)
# -----------------------------------------------------------------------
print("LEVEL 1B: Arm 2 lens search")
print("Same kernel, new substrate. No modifications for this level.")
print()
rng_2 = np.random.default_rng(1)
def factory_arm2(n: int) -> List[Any]:
return [random_candidate(ArmID.ARM_2, f"a2_g0_{i}", rng=rng_2)
for i in range(n)]
def evaluator_arm2(cand: Any) -> EvaluationResult:
return evaluate_candidate(cand, stream_2, world)
def mutator_arm2(cand: Any) -> Any:
return mutate_candidate(cand, mutation_rate=0.4, rng=rng_2)
def crossover_arm2(a: Any, b: Any, child_id: str) -> Any:
return crossover_candidates(a, b, child_id, rng=rng_2)
config_2 = SurvivalConfig(
population_size=25,
max_generations=35,
elite_fraction=0.15,
mutation_rate=0.4,
crossover_rate=0.5,
promotion_threshold=0.70,
promotion_stability_gens=4,
rng_seed=1
)
gate_2 = make_threshold_gate(
config_2.promotion_threshold,
config_2.promotion_stability_gens
)
pop_2 = initialize_population(factory_arm2, config_2.population_size)
result_2 = run_survival_kernel(
population=pop_2,
evaluator=evaluator_arm2,
mutator=mutator_arm2,
crossover_fn=crossover_arm2,
gate=gate_2,
config=config_2,
stage=StageLabel.LENS_SEARCH,
arm_id=ArmID.ARM_2,
verbose=True
)
print(f"\nArm 2 result: promoted={result_2.promoted}, "
f"gen={result_2.promotion_generation}, "
f"candidate={result_2.promoted_candidate_id}")
print(f" Best score trajectory (every 10 gens): "
f"{[round(result_2.best_per_generation[g], 3) for g in range(0, len(result_2.best_per_generation), 10)]}")
print()
print("=" * 60)
print("KERNEL ARCHITECTURE VERIFIED:")
print(" - Same run_survival_kernel() called for both arms")
print(" - No arm-specific logic inside the kernel")
print(" - Domain logic lives in evaluator/mutator/crossover")
print(" - Promotion gate tracks stability, not generation count")
print(" - Inadmissible candidates tracked separately from non-survivors")
print("=" * 60)
```