diff --git a/docs/lean/life-components-schema.md b/docs/lean/life-components-schema.md index d78570bea2..a41b9ee351 100644 --- a/docs/lean/life-components-schema.md +++ b/docs/lean/life-components-schema.md @@ -1,4 +1,4 @@ -# Schéma de motifs et de réactions de Game of Life — tranche 1+2 de #15635 +# Schéma de motifs et de réactions de Game of Life — tranches 1-4 de #15635 Le moteur cellulaire de #15571 cherche un motif en énumérant des cellules dans une boîte, puis en encodant son évolution en SAT. Son espace de recherche croît @@ -22,8 +22,8 @@ le catalogue curé est [`scripts/lean/life_components_fixture.json`](../../scrip |---|---| | §1 schéma minimal versionné (motifs + réactions) | **livré** | | §2 catalogue reproductible réduit, rejoué indépendamment | **livré** | -| §3 générateur de contraintes compositionnelles | hors tranche | -| §4 démonstrateur multi-composants + ablations de pruning | hors tranche | +| §3 générateur de contraintes compositionnelles | **livré** ([`life_compose.py`](../../scripts/lean/life_compose.py)) | +| §4 démonstrateur multi-composants + ablations de pruning | **livré** (objectifs, verdicts, ablations ci-dessous) | Le schéma porte les champs que #15635 nomme pour §1, mais **tous ne sont pas encore validés mécaniquement** — la distinction est explicite ci-dessous. @@ -90,8 +90,8 @@ sa nature et sa provenance. Sont **recalculés par replay de l'évolution jointe Les **ports** sont validés structurellement (nom, direction non nulle, position dans la région occupée). Leur *sémantique de compatibilité* — quel port de quel -composant peut en rencontrer un autre, à quelle phase — appartient à §3 et n'est -pas prétendue ici. +composant peut en rencontrer un autre, à quelle phase — est livrée par la +couche compositionnelle de §3 (section « Ports » ci-dessous). ## Le catalogue curé @@ -110,13 +110,201 @@ Six motifs et trois réactions, tous **rejoués** à chaque validation : |---|---|---:|---| | `glider_pair_annihilation` | consumable | 12 | vide (annihilation complète) | | `glider_pair_two_blocks` | consumable | 5 | deux `block` | -| `block_catalyses_glider` | catalytic | 43 | 12 cellules, le `block` survit parmi les débris | +| `block_catalyses_glider` | catalytic | 8 | un `block` (le réactif poussé de `(4,2)` à `(2,3)`, zéro débris) | La troisième réaction est délibérément **non triviale** : le produit n'est pas la forme du réactif, donc « le bloc survit » est une mesure qui discrimine. Un cas où le produit serait exactement la forme du catalyseur serait vrai par construction et ne prouverait rien. +Elle a été **re-mesurée en tranche 3** à partir du dernier état pur de +l'approche naturelle du glider (phase 3, bloc à `(4,2)`) : le placement de la +tranche 1+2 (phase 0, bloc `(3,0)`) démarrait au milieu de l'interaction et +était inatteignable par dérive pure — sa precondition porte désormais +l'exigence d'atteignabilité, et le détail est dans la section « Deux leçons +mesurées » ci-dessous. + +## La couche compositionnelle — tranche 3 + +Le module [`scripts/lean/life_compose.py`](../../scripts/lean/life_compose.py) +cherche des **compositions d'événements** du catalogue : un objectif borné +(`Goal`) déclare des comptages finaux minimaux, des événements requis, des +plafonds (événements, composants, gliders), un horizon et une surface ; le +moteur énumère les placements d'événements qui satisfont toutes les contraintes +simultanément, puis **certifie** chaque témoin candidat par replay indépendant +dans le moteur du dépôt. Le modèle de recherche est propositionnel, exact et +borné : + +- un **spawn** `(motif, ancre, phase)` vit sur une trajectoire fermée + `cells_at(t)`, vérifiée cellule à cellule contre `step()` pour chaque motif, + chaque phase, trois ancres, 13 pas — zéro écart. Le compteur de périodes + dépend de la phase de départ (`(p0 + t) // P`, pas `t // P`) : un glider + `p0=3` à `t=1` a déjà une période complète écoulée ; +- un **événement** `(réaction, offset, t_fire, bindings)` occupe la fenêtre + temporelle `[t_fire, t_fire + stabilisation]` et la région spatiale de la + réaction translatée ; ses réactifs consommés disparaissent de l'état, ses + produits naissent avec leur propre trajectoire ; +- la scène composée n'est **pas simulée pendant la recherche** : elle est + reconstruite et rejouée intégralement à la certification seulement. C'est ce + qui rend `FOUND` fiable alors que la recherche raisonne sur une abstraction. + +### Ports : la sémantique de compatibilité (R2) + +Le port d'un réactif **mobile** porte les signes de la translation par période +(le glider : `(1,-1)`) — la direction le long de laquelle il peut *atteindre* +son placement par dérive pure. Le port d'un réactif **statique** porte l'axe du +connecteur (le bloc : `(-1,0)` vers la ligne d'approche). La famille R2 rejette +tout binding dont la direction de spawn n'est pas celle du port déclaré : +direction opposée (le mobile s'éloigne), axe statique sur un réactif mobile, +translation orthogonale à la ligne du motif. + +### Congruence de phase à l'arrivée (R4) + +Un réactif mobile doit arriver dans la **phase déclarée** de la réaction, sinon +la fenêtre ne correspond pas au replay mesuré : `p0 = (phase_déclarée − +t_fire) mod P`. Sur les quatre phases de départ possibles d'un glider, +exactement une satisfait la congruence — le test de non-régression l'exige. + +### Fenêtres spatiales (R3) + +Deux événements aux fenêtres spatiales chevauchantes, ou dont un objet vivant +passe à moins de 3 (Chebyshev) des cellules d'une fenêtre étrangère, sont +rejetés — avec une exception structurelle : un produit est *par construction* +la dernière image de la fenêtre de son propre événement, il n'est jamais +confronté à elle (le replay aux bornes exactes couvre déjà cette transition). +Le seuil 3 n'est pas un réglage, c'est la mesure ci-dessous. + +### Dédup d'états (R1) + +La clé d'état est le **multiset des spawns et des événements complets** +(réaction, offset, `t_fire`, réactifs liés décrits par leur géométrie et leur +temps de naissance) : les objets vivants en découlent, et **l'ordre de tir des +épisodes indépendants n'y figure pas** — il n'est pas sémantique (`certify` +calcule la scène pas à pas comme une union de fenêtres indexées par le temps, +et un réactif n'est lié qu'une fois ; `goal_reached` compte des identifiants de +réaction). + +Deux clés plus grossières ont produit des **faux négatifs de complétude**, +mesurés : une clé indépendante de la position avalait le produit d'un second +épisode né ailleurs ; une clé réduite à la *signature des vivants + nombre +d'événements* confondait deux scènes aux mêmes produits à des **temps de tir +différents** — sous ablation de R4, un état non certifiant partageant la clé du +témoin était déplié d'abord, le témoin était dédupliqué, et la recherche +concluait `IMPOSSIBLE_BOUNDED` alors qu'un témoin existait. Un verdict +d'impossibilité qui peut être faux n'est pas un verdict. + +La clé est donc le **garde-fou de complétude** du module. Honnêtement mesuré : +sur les trois instances du démonstrateur elle ne mord **jamais** (`r1_state_dedup` += 0) — il n'y a pas d'état régénéré à collapser dans cet ordre de parcours —, +et l'ablation de R1 est par conséquent **neutre**. La clé grossière affichait +274 nœuds là où la clé saine en explore 10 901 : sa réduction était achetée par +des confusions non fondées, pas par du travail évité. La sensibilité de la clé +à l'ancre, au temps de tir et à l'ordre est figée par un test dédié. + +### Deux leçons mesurées, figées en tests de non-régression + +1. **Non-interaction.** Deux ensembles de cellules à distance de Chebyshev ≥ 3 + n'interagissent jamais ; à distance 2, tout dépend du contenu. Deux blocs + gap-1 sont stables, mais un glider à distance 2 d'un bloc engendre des + naissances croisées — une cellule morte voit 2 voisins d'un objet plus 1 de + l'autre. Le seuil 3 de R3 est cette mesure, pas une heuristique. +2. **Atteignabilité du placement déclaré.** Le placement tranche 1+2 de + `block_catalyses_glider` démarrait au milieu de l'interaction : l'approche + naturelle du glider annihile tout (population 0). La réaction a été + re-mesurée au **dernier état pur** de l'approche (glider phase 3, bloc + `(4,2)`), et une famille catalytique propre a été vérifiée le long de la + diagonale d'arrivée (offsets `x+y=6` du coin : `(4,2)`, `(5,1)`, `(6,0)`, + `(7,-1)` — tous donnent « bloc seul survit, glider absorbé, zéro débris »). + La déclaration curée est désormais atteignable ; le placement historique est + le contrôle négatif du test d'atteignabilité. + +### Impossibilité de la réutilisation stricte — cause racine + +L'objectif `two_blocks_catalyse` (réutiliser **strictement** les blocs produits +par la paire comme catalyseurs d'une seconde réaction) est `IMPOSSIBLE_BOUNDED` +pour une raison physique mesurée, pas un échec de recherche : le produit de +`glider_pair_two_blocks` est une paire de blocs gap-1 ; tout épisode catalytique +qui lie l'un des blocs met l'autre à Chebyshev 2, et le recul du bloc pendant +la transmutation fait naître des cellules dans la colonne du gap (naissances +croisées mesurées à `t=19`). Le replay exact rejette donc chaque candidat — +l'épuisement **est** la bonne réponse, et le compteur `replay_rejected ≥ 1` +l'atteste. La variante `two_blocks_catalyse_free` (le second épisode apporte +son propre bloc de catalyste) est `FOUND`. + +## Démonstrateur — tranche 4 + +### Verdicts mesurés (base, 3 répétitions, passe séquentielle) + +| Objectif | Verdict | Nœuds | Candidats | Replays rejetés | Temps | Pic mémoire | +|---|---|---:|---:|---:|---|---:| +| `two_blocks` | **FOUND** — 2 blocs en 25 pas (témoin identique ×3) | 2 | 2 | 0 | 0.004 s | 66 Kio | +| `two_blocks_catalyse` | **IMPOSSIBLE_BOUNDED** | 1 227 | 5 208 | 1 176 | 187-204 s | 1.5 Mio | +| `two_blocks_catalyse_free` | **FOUND** — 3 blocs en 61 pas, 19 cellules initiales | 10 901 | 101 | 9 | 12.6-13.4 s | 16.5 Mio | + +### Ablations des familles d'élagage (3 répétitions ; la certification n'est jamais ablatable) + +Budget de nœuds : 20 000 (`two_blocks`, variante libre), 4 000 (stricte — la +base y épuise à 1 227 nœuds). + +| Objectif | Famille ablatée | Verdict | Nœuds | Candidats | Temps | +|---|---|---|---:|---:|---| +| `two_blocks` | — (base) | FOUND | 2 | 2 | 0.004 s | +| | R1 dédup | FOUND | 2 | 2 | 0.004 s | +| | R2 ports | FOUND | 3 | 15 | 0.02 s | +| | R3 fenêtres | FOUND | 2 | 2 | 0.004 s | +| | R4 congruence | FOUND | 2 | 1 | 0.004 s | +| `two_blocks_catalyse` | — (base) | IMPOSSIBLE_BOUNDED | 1 227 | 5 208 | 187-204 s | +| | R1 dédup | IMPOSSIBLE_BOUNDED | 1 227 | 5 208 | 181-209 s | +| | R2 ports | IMPOSSIBLE_BOUNDED | 3 588 | 31 057 | 353-357 s | +| | R3 fenêtres | IMPOSSIBLE_BOUNDED | 2 403 | 5 208 | 212-241 s | +| | R4 congruence | TIMEOUT | 4 001 | 3 111 | 385-442 s | +| `two_blocks_catalyse_free` | — (base) | FOUND | 10 901 | 101 | 12.6-13.4 s | +| | R1 dédup | FOUND | 10 901 | 101 | 12.5-12.6 s | +| | R2 ports | TIMEOUT | 20 001 | 263 | 25.7-27.0 s | +| | R3 fenêtres | FOUND | 14 720 | 101 | 4.5-4.9 s | +| | R4 congruence | TIMEOUT | 20 001 | 41 | 14.6-15.5 s | + +**Lecture honnête des ablations :** + +- **R2 (ports) et R4 (congruence de phase) sont décisives** : les désactiver + fait perdre le témoin de la variante libre au plafond de nœuds (20 001 nœuds + sans certification) et double le coût de la stricte. +- **R3 (fenêtres) ne change aucun verdict**, et son ablation est même *plus + rapide* sur la variante libre (4.5 s contre 12.6 s) : son calcul de fenêtres + coûte plus qu'il ne rapporte sur ces instances bornées. Sa raison d'être est + la sûreté du modèle (rejeter les compositions aux fenêtres chevauchantes + avant le replay) ; la certification reste le filet, couvert par un contrôle + négatif dédié. +- **R1 (dédup) est neutre ici** : la clé saine ne collide jamais (0 collapse + mesuré). Elle n'en est pas moins le garde-fou de complétude — sa version + grossière produisait des faux `IMPOSSIBLE_BOUNDED` (mesuré, testé). +- Aucune ablation ne produit un verdict `IMPOSSIBLE_BOUNDED` sur un objectif + `FOUND` : retirer un filtre élargit l'espace, le pire cas est le plafond + (`TIMEOUT`). La propriété est structurelle — et le faux négatif mesuré sous + ablation de R4 venait précisément d'une clé de dédup qui la violait en + avalant un état du chemin-témoin. + +### Comparaison avec la baseline cellulaire (#15571) + +Le même dépôt sait chercher des still lifes par énumération de cellules : +`synthesize(1, (0,0), 5, 8)` énumère **1 807 780 candidats** en **58 s** par +répétition sur la même cible (81 still lifes, cible trouvée, contenu identique +sur les trois répétitions ; 70-77 s sous la charge des trois rapports complets +lancés en parallèle). La couche compositionnelle répond à ses objectifs en +millièmes à dizaines de secondes, avec des espaces de recherche comptés en +centaines de candidats : elle n'énumère pas des cellules, elle énumère des +**placements d'événements connus** — c'est le levier attendu de la couche +symbolique. + +La comparaison est honnête seulement si l'on distingue les deux questions : la +baseline établit une **existence** (parmi tous les still lifes de ≤ 8 cellules +d'une boîte 5×5, la cible est là) et paie l'énumération complète de cet espace, +alors que le compositionnel **construit** la cible (deux gliders dont on sait +qu'ils s'annihilent en deux blocs) sans être confiné à une boîte. Le gain n'est +donc pas « le même problème résolu plus vite », c'est le prix de la +connaissance : quand le catalogue connaît les briques, il ne cherche plus des +cellules ; quand il ne les connaît pas, la baseline reste l'arbitre. + ## Provenance : trois natures d'information, distinguées Le critère 8 de #15635 demande de ne pas confondre trois choses. Le catalogue @@ -142,7 +330,11 @@ les sépare champ par champ : ``` python scripts/lean/life_components.py --fixture scripts/lean/life_components_fixture.json python scripts/lean/life_components.py --fixture scripts/lean/life_components_fixture.json --json +python scripts/lean/life_compose.py --objective two_blocks_catalyse_free +python scripts/lean/life_compose.py --objective two_blocks --full-report --json +python scripts/lean/life_compose.py --objective two_blocks_catalyse --full-report --node-budget 20000 --json python -m pytest scripts/lean/tests/test_life_components.py -q +python -m pytest scripts/lean/tests/test_life_compose.py -q ``` Le validateur sort `0` si toutes les métadonnées déclarées correspondent au diff --git a/scripts/lean/README.md b/scripts/lean/README.md index 6805f5fbe6..f72119eea4 100644 --- a/scripts/lean/README.md +++ b/scripts/lean/README.md @@ -350,3 +350,33 @@ ses réactifs est refusée. Document complet (périmètre, ce qui est mesuré vs déclaré, provenance des faits, limites assumées) : [`docs/lean/life-components-schema.md`](../../docs/lean/life-components-schema.md). + +## `life_compose.py` — le générateur de contraintes compositionnelles (tranches 3+4 de #15635) + +Au-dessus du catalogue symbolique (tranche 1+2), ce module **compose** : étant +donné un objectif borné (comptages finaux minimaux, événements requis, budgets +de composants/gliders/événements, horizon, surface), il énumère les placements +d'événements du catalogue qui satisfont conjointement les contraintes +spatiales (fenêtres disjointes), temporelles (congruence de phase à +l'arrivée), d'interface (ports compatibles) et de clearance, puis **certifie** +chaque témoin par replay indépendant dans le moteur du dépôt. Les quatre +familles d'élagage (R1 dédup d'états, R2 compatibilité de ports, R3 fenêtres +spatiales, R4 congruence de phase) sont instrumentées par des compteurs et +ablatables — la certification, elle, n'est jamais ablatable. + +``` +python scripts/lean/life_compose.py --objective two_blocks_catalyse_free +python scripts/lean/life_compose.py --objective two_blocks --full-report --json +python -m pytest scripts/lean/tests/test_life_compose.py -q +``` + +Trois verdicts distincts : `FOUND` (témoin rejoué exactement par le moteur +indépendant), `IMPOSSIBLE_BOUNDED` (épuisement explicite de l'espace borné), +`TIMEOUT` (plafond de nœuds atteint avant épuisement). Deux leçons mesurées en +tranche 3 sont figées par des tests de non-régression : la non-interaction +universelle est Chebyshev >= 3 (à distance 2, des naissances croisées +apparaissent selon le contenu), et un placement réactif déclaré doit être +atteignable par dérive pure le long du port du réactif mobile. + +Détail complet (modèle propositionnel, familles de contraintes, ablations, +démonstrateur, comparaison baseline) : [`docs/lean/life-components-schema.md`](../../docs/lean/life-components-schema.md). diff --git a/scripts/lean/life_components_fixture.json b/scripts/lean/life_components_fixture.json index e6bd916d65..675de49d25 100644 --- a/scripts/lean/life_components_fixture.json +++ b/scripts/lean/life_components_fixture.json @@ -856,8 +856,8 @@ 0 ], "direction": [ - 0, - 1 + 1, + -1 ], "phase": 0 }, @@ -868,8 +868,8 @@ 2 ], "direction": [ - 0, - 1 + -1, + -1 ], "phase": 0 } @@ -952,8 +952,8 @@ 0 ], "direction": [ - 0, - 1 + 1, + -1 ], "phase": 0 }, @@ -964,8 +964,8 @@ 0 ], "direction": [ - 0, - 1 + -1, + -1 ], "phase": 0 } @@ -1010,68 +1010,28 @@ 0, 0 ], - "phase": 0 + "phase": 3 }, { "motif_id": "block", "offset": [ - 3, - 0 + 4, + 2 ], "phase": 0 } ], - "products": [], - "product_cells": [ - [ - -5, - -4 - ], - [ - -5, - -3 - ], - [ - -4, - -5 - ], - [ - -4, - -2 - ], - [ - -3, - -5 - ], - [ - -3, - -2 - ], - [ - -2, - -4 - ], - [ - -2, - -3 - ], - [ - 7, - -4 - ], - [ - 7, - -3 - ], - [ - 8, - -4 - ], - [ - 8, - -3 - ] + "products": [ + { + "motif_id": "block", + "offset": [ + 2, + 3 + ], + "phase": 0 + } ], + "product_cells": null, "ports": [ { "name": "in0", @@ -1080,44 +1040,44 @@ 0 ], "direction": [ - 0, - 1 + 1, + -1 ], - "phase": 0 + "phase": 3 }, { "name": "in1", "offset": [ - 3, - 0 + 4, + 2 ], "direction": [ - 0, - 1 + -1, + 0 ], "phase": 0 } ], "temporal_offset": 0, - "stabilization_time": 43, + "stabilization_time": 8, "occupied_region": [ - -6, - -9, - 15, - 13 + 0, + -1, + 6, + 6 ], "clearance": [ [ - -6, - -14, - 2, - 2 + -3, + -1, + 3, + 6 ] ], "nature": "catalytic", "spatial_transform": null, "preconditions": [ - "cellules des reactifs placees telles que declarees" + "reactif glider atteignant son placement declare par derivee pure le long de (1,-1)" ], "effects": [ "catalytic" @@ -1125,8 +1085,8 @@ "provenance": { "source": "mesure firsthand par le moteur Life du depot", "license": "domaine public (objet mathematique du jeu de la vie)", - "retrieved": "2026-09-11", - "note": "stabilise a t=43 ; region occupee mesuree" + "retrieved": "2026-09-12", + "note": "re-mesuree tranche 3 : depart au dernier etat pur de l approche naturelle (glider phase 3, block (4,2)) ; le placement tranche 1-2 (phase 0, block (3,0)) demarrait au milieu de l interaction, inatteignable par derivee pure" } } ] diff --git a/scripts/lean/life_compose.py b/scripts/lean/life_compose.py new file mode 100644 index 0000000000..39b8dac4c0 --- /dev/null +++ b/scripts/lean/life_compose.py @@ -0,0 +1,1378 @@ +#!/usr/bin/env python3 +"""Générateur de contraintes compositionnelles + démonstrateur (tranche 3+4 de #15635). + +Ce module ajoute la couche de recherche par composants au-dessus du schéma +symbolique de `life_components` (tranche 1+2, PR #15653) et du moteur cellulaire +de #15571 (`life_synthesize`). La chaîne visée par l'issue : + + objectif génératif + -> sélection et assemblage de composants/réactions + -> contraintes temporelles, spatiales et d'interface + -> replay indépendant dans le moteur Life + -> témoin ou frontière d'impossibilité bornée + +Le modèle à contraintes PROPOSE, le replay du moteur dispose : toute scène +jugée satisfaisante au niveau symbolique est re-certifiée pas à pas contre le +vrai moteur avant d'être publiée. Un écart modèle/replay est compté comme +``replay_rejected``, jamais maquillé. + +Sémantique des ports (définie par cette tranche — la tranche 1 ne spécifiait +que la structure) : + + * pour un réactif MOBILE, ``direction`` = les signes, par axe, de la + translation par période du motif qui emprunte le port (le glider + ``t=(1,-1)`` entre par un port ``[1,-1]``) ; + * pour un réactif STATIQUE, ``direction`` = l'axe du connecteur : la + direction, depuis le placement déclaré, vers le lieu de l'interaction + (le bloc de ``block_catalyses_glider`` porte ``[-1,0]``). + +Quatre familles de pruning, toutes instrumentées (compteurs, ablations une à +une, le replay de certification n'étant JAMAIS ablé) : + + * **R1 bris de symétrie** : ancrage du premier événement à l'origine + (invariance par translation de la scène entière) et déduplication des + états de recherche équivalents ; + * **R2 filtrage type/port** : un réactif ne se lie qu'à un composant du + motif déclaré dont la translation est compatible avec le port ; les + réactions sans contribution possible à l'objectif sont écartées ; + * **R3 pruning spatial** : enveloppes gonflées (+1, seuil de + non-interaction de Conway) disjointes pour toute paire non réagissante, + respect des zones de clearance et des cellules de fenêtre des événements ; + * **R4 pruning temporel** : congruence de phase (un ``t`` hors de la classe + requise élimine le candidat) et fenêtres de tir dans l'horizon borné. + +Usage : + python scripts/lean/life_compose.py --objective two_blocks_catalyse + python scripts/lean/life_compose.py --objective two_blocks --json + python scripts/lean/life_compose.py --objective two_blocks_catalyse --ablate r3 + python scripts/lean/life_compose.py --objective two_blocks --full-report --json +""" + +from __future__ import annotations + +import argparse +import itertools +import json +import time +import tracemalloc +from dataclasses import dataclass +from math import comb +from pathlib import Path +from typing import Any, Iterator, Sequence + +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from life_synthesize import Cell, Grid, evolve, normalize, step # noqa: E402 +from life_components import ( # noqa: E402 + Catalog, + Reaction, + load_catalog, + measure_motif, +) + +# Seuil de non-interaction de Conway : deux groupes de cellules à distance +# de Chebysyshev >= 2 ne se voient pas. Gonfler chaque enveloppe de 1 +# matérialise exactement cette frontière. +GROW = 1 + +DEFAULT_NODE_BUDGET = 200_000 + +ABLATION_FAMILIES = ("r1", "r2", "r3", "r4") + + +class BudgetHit(RuntimeError): + """Le budget de noeuds est épuisé : verdict TIMEOUT, pas un échec.""" + + +# --------------------------------------------------------------------------- +# Modèle : composants (variables de choix) et événements +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class Spawn: + """Composant initial : un motif posé à t=0 (variable de choix).""" + + motif_id: str + anchor: tuple[int, int] + phase0: int + + +@dataclass(frozen=True) +class ProductBirth: + """Composant né d'un événement (placement produit déclaré).""" + + event_idx: int + placement_idx: int + motif_id: str + anchor: tuple[int, int] + phase0: int + t0: int + + +@dataclass(frozen=True) +class StaticProduct: + """Produit statique brut (champ product_cells d'une réaction).""" + + event_idx: int + cells: tuple[Cell, ...] + t0: int + + +Handle = Spawn | ProductBirth | StaticProduct + + +@dataclass(frozen=True) +class EventInst: + """Une réaction tirée : cadre translaté, instant, liaisons par réactif.""" + + idx: int + reaction_id: str + offset: tuple[int, int] + t_fire: int + bindings: tuple[Handle, ...] + + +def handle_key(h: Handle) -> tuple: + if isinstance(h, Spawn): + return ("spawn", h.motif_id, h.anchor, h.phase0) + if isinstance(h, ProductBirth): + return ("product", h.event_idx, h.placement_idx) + return ("static", h.event_idx) + + +def t_start(h: Handle) -> int: + if isinstance(h, Spawn): + return 0 + return h.t0 + + +def is_spawn_like(h: Handle) -> bool: + return isinstance(h, (Spawn, ProductBirth)) + + +def _motif_phases(catalog: Catalog, motif_id: str) -> dict[int, tuple[Cell, ...]]: + motif = catalog.motifs[motif_id] + if motif.phases is not None: + return motif.phases + return measure_motif(motif.cells).phases + + +_PHASE_OFFSET_CACHE: dict[str, dict[int, tuple[int, int]]] = {} + + +def _motif_phase_offsets(catalog: Catalog, motif_id: str) -> dict[int, tuple[int, int]]: + """Coin min de chaque phase, relativement au coin de la phase de base. + + Les phases stockées par la tranche 1 sont NORMALISÉES (coin à l'origine) : + la forme seule ne dit pas où elle se trouve dans le cycle. Un glider dont + on efface ce décalage inter-phases suit une fausse trajectoire dès t=1 — + c'est le premier écart modèle/replay que la certification a attrapé. + """ + cached = _PHASE_OFFSET_CACHE.get(motif_id) + if cached is not None: + return cached + motif = catalog.motifs[motif_id] + base = normalize(set(motif.cells)) + corners: dict[int, tuple[int, int]] = {} + for k in range(motif.period): + evolved = evolve(k, set(base)) + corners[k] = ( + min(x for x, _ in evolved), + min(y for _, y in evolved), + ) + _PHASE_OFFSET_CACHE[motif_id] = corners + return corners + + +def _phase_position( + catalog: Catalog, motif_id: str, anchor: tuple[int, int], p0: int, t0: int, t: int +) -> tuple[tuple[int, int], int]: + """(position de la forme normalisée courante, index de phase) à l'instant t. + + Le cycle démarre à la phase p0 : l'index de phase à t est (p0 + t - t0) + mod P et le nombre de périodes écoulées est (p0 + t - t0) // P — pas + (t - t0) // P, qui oublierait les périodes déjà parcourues par la phase + initiale (piège mesuré : glider p0=3 décalé d'une diagonale à t=1). + """ + motif = catalog.motifs[motif_id] + tx, ty = motif.translation_step + elapsed = p0 + t - t0 + k = elapsed // motif.period + phase = elapsed % motif.period + corner = _motif_phase_offsets(catalog, motif_id)[phase] + return ( + (anchor[0] + corner[0] + tx * k, anchor[1] + corner[1] + ty * k), + phase, + ) + + +def cells_at(catalog: Catalog, h: Handle, t: int) -> Grid: + """Cellules exactes d'un composant à l'instant t (trajectoire dérivée). + + Pour un motif de période P et de translation par période T, posé avec la + phase p0 à l'ancre `anchor` (position de la forme normalisée de phase 0) + et né à t0 : + phase(t) = (p0 + t - t0) mod P + position(t) = anchor + coin(phase) + T * ((t - t0) // P) + Exact, pas approché : la certification replay le vérifie pas à pas. + """ + if isinstance(h, StaticProduct): + return set(h.cells) if t >= h.t0 else set() + if isinstance(h, Spawn): + motif_id, p0, t0, anchor = h.motif_id, h.phase0, 0, h.anchor + else: + motif_id, p0, t0, anchor = h.motif_id, h.phase0, h.t0, h.anchor + if t < t0: + return set() + (dx, dy), phase = _phase_position(catalog, motif_id, anchor, p0, t0, t) + return {(x + dx, y + dy) for (x, y) in _motif_phases(catalog, motif_id)[phase]} + + +def anchor_at(catalog: Catalog, h: Handle, t: int) -> tuple[int, int] | None: + """Position de la forme normalisée courante d'un composant à l'instant t.""" + if not is_spawn_like(h): + return None + if isinstance(h, Spawn): + motif_id, p0, t0, anchor = h.motif_id, h.phase0, 0, h.anchor + else: + motif_id, p0, t0, anchor = h.motif_id, h.phase0, h.t0, h.anchor + if t < t0: + return None + return _phase_position(catalog, motif_id, anchor, p0, t0, t)[0] + + +def phase_at(catalog: Catalog, h: Handle, t: int) -> int | None: + if not is_spawn_like(h): + return None + motif = catalog.motifs[h.motif_id] + if isinstance(h, Spawn): + p0, t0 = h.phase0, 0 + else: + p0, t0 = h.phase0, h.t0 + if t < t0: + return None + return (p0 + t - t0) % motif.period + + +def _rect_of(cells: Iterable[Cell]) -> tuple[int, int, int, int] | None: + grid = set(cells) + if not grid: + return None + x0 = min(x for x, _ in grid) + y0 = min(y for _, y in grid) + x1 = max(x for x, _ in grid) + y1 = max(y for _, y in grid) + return (x0, y0, x1 - x0 + 1, y1 - y0 + 1) + + +def swept_rect( + catalog: Catalog, h: Handle, t_from: int, t_to: int +) -> tuple[int, int, int, int] | None: + """Sur-ensemble rectangulaire du volume parcouru sur [t_from, t_to]. + + Conservateur : union des cellules aux instants extrêmes et médian (la + dérive est linéaire par période, donc monotone par axe). + """ + if isinstance(h, StaticProduct): + return _rect_of(h.cells) + if t_to < t_start(h): + return None + t_from = max(t_from, t_start(h)) + union: Grid = set() + for t in {t_from, t_to, (t_from + t_to) // 2}: + union |= cells_at(catalog, h, t) + return _rect_of(union) + + +def _grown( + rect: tuple[int, int, int, int], g: int = GROW +) -> tuple[int, int, int, int]: + x0, y0, w, h = rect + return (x0 - g, y0 - g, w + 2 * g, h + 2 * g) + + +def _rects_overlap(a: tuple[int, int, int, int], b: tuple[int, int, int, int]) -> bool: + ax0, ay0, aw, ah = a + bx0, by0, bw, bh = b + return ax0 < bx0 + bw and bx0 < ax0 + aw and ay0 < by0 + bh and by0 < ay0 + ah + + +def _rect_contains_cell( + rect: tuple[int, int, int, int], cell: Cell +) -> bool: + x0, y0, w, h = rect + x, y = cell + return x0 <= x < x0 + w and y0 <= y < y0 + h + + +def _min_chebyshev(a: Grid, b: Grid) -> int: + best = 10**9 + for (x1, y1) in a: + for (x2, y2) in b: + d = max(abs(x1 - x2), abs(y1 - y2)) + if d < best: + best = d + return best + + +# --------------------------------------------------------------------------- +# Fenêtres de réaction : replay solo (mémoïsé) +# --------------------------------------------------------------------------- + +_WINDOW_CACHE: dict[tuple[str, tuple[int, int]], list[Grid]] = {} + + +def reaction_initial_cells(catalog: Catalog, reaction: Reaction) -> Grid: + """Réactifs déclarés posés dans le cadre propre de la réaction.""" + grid: Grid = set() + for placement in reaction.reactants: + phase_cells = _motif_phases(catalog, placement.motif_id)[placement.phase] + dx, dy = placement.offset + grid |= {(x + dx, y + dy) for (x, y) in phase_cells} + return grid + + +def reaction_window_cells( + catalog: Catalog, reaction: Reaction, offset: tuple[int, int] +) -> list[Grid]: + """Évolution solo de la réaction translatée : cellules à t=0..stabilisation.""" + key = (reaction.id, offset) + cached = _WINDOW_CACHE.get(key) + if cached is not None: + return cached + base = reaction_initial_cells(catalog, reaction) + dx, dy = offset + cells = {(x + dx, y + dy) for (x, y) in base} + window = [set(cells)] + for _ in range(reaction.stabilization_time): + cells = step(cells) + window.append(set(cells)) + _WINDOW_CACHE[key] = window + return window + + +def product_handles(catalog: Catalog, event: EventInst) -> list[Handle]: + """Composants nés d'un événement : placements produits, ou cellules brutes.""" + reaction = catalog.reactions[event.reaction_id] + t0 = event.t_fire + reaction.stabilization_time + born: list[Handle] = [] + if reaction.products: + for k, placement in enumerate(reaction.products): + born.append( + ProductBirth( + event_idx=event.idx, + placement_idx=k, + motif_id=placement.motif_id, + anchor=( + placement.offset[0] + event.offset[0], + placement.offset[1] + event.offset[1], + ), + phase0=placement.phase, + t0=t0, + ) + ) + elif reaction.product_cells is not None: + dx, dy = event.offset + born.append( + StaticProduct( + event_idx=event.idx, + cells=tuple((x + dx, y + dy) for (x, y) in reaction.product_cells), + t0=t0, + ) + ) + return born + + +# --------------------------------------------------------------------------- +# Objectifs +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class Goal: + """Objectif borné : comptes finaux, événements exigés, budgets.""" + + name: str + final_min_counts: dict[str, int] + required_events: dict[str, int] + max_events: int + max_components: int + horizon: int + surface: tuple[int, int] + max_gliders: int + spawn_kinds: tuple[str, ...] = ("spaceship",) + + +OBJECTIVES: dict[str, Goal] = { + "two_blocks": Goal( + name="two_blocks", + final_min_counts={"block": 2}, + required_events={}, + max_events=1, + max_components=2, + horizon=24, + surface=(24, 24), + max_gliders=2, + ), + # Variante stricte : la catalyse DOIT réutiliser un bloc produit par un + # événement antérieur (aucun spawn statique autorisé). Le produit de + # glider_pair_two_blocks est une paire de blocs séparés d'une colonne : + # tout épisode catalytique sur l'un des deux voit l'autre à Chebyshev 2, + # et la transmutation du bloc lié engendre des naissances croisées dans + # la colonne intercalaire -- le replay exact rejette tous les candidats. + # Objectif démonstrateur d'impossibilité bornée racinée. + "two_blocks_catalyse": Goal( + name="two_blocks_catalyse", + final_min_counts={"block": 2}, + required_events={"block_catalyses_glider": 1}, + max_events=2, + max_components=3, + horizon=60, + surface=(32, 32), + max_gliders=3, + ), + # Variante libre : même chaîne, mais le bloc catalysé peut être un spawn + # statique (4 composants). Démontreuse FOUND avec réutilisation temporelle + # des événements et contrainte d'interface port/phase active. + "two_blocks_catalyse_free": Goal( + name="two_blocks_catalyse_free", + final_min_counts={"block": 2}, + required_events={ + "block_catalyses_glider": 1, + "glider_pair_two_blocks": 1, + }, + max_events=2, + max_components=4, + horizon=60, + surface=(40, 40), + max_gliders=3, + spawn_kinds=("spaceship", "still_life"), + ), +} + +BLOCK_SHAPE: tuple[Cell, ...] = ((0, 0), (1, 0), (0, 1), (1, 1)) + +# Cible partagée avec la baseline cellulaire de #15571 : deux blocs 2x2 +# séparés par une colonne vide (le produit de glider_pair_two_blocks). +TWO_BLOCKS_TARGET: tuple[Cell, ...] = ( + (0, 0), (1, 0), (0, 1), (1, 1), + (3, 0), (4, 0), (3, 1), (4, 1), +) + + +def blocks_in(cells: Grid) -> list[tuple[int, int]]: + """Positions des blocs 2x2 disjoints présents dans un ensemble de cellules.""" + found: list[tuple[int, int]] = [] + used: Grid = set() + for (x, y) in sorted(cells): + if (x, y) in used: + continue + square = {(x + dx, y + dy) for (dx, dy) in BLOCK_SHAPE} + if square <= cells and not (square & used): + found.append((x, y)) + used |= square + return found + + +def goal_reached( + catalog: Catalog, goal: Goal, events: Sequence[EventInst], live: Sequence[Handle] +) -> bool: + """Évaluation symbolique de l'objectif (le replay certifie ensuite).""" + fired: dict[str, int] = {} + for event in events: + fired[event.reaction_id] = fired.get(event.reaction_id, 0) + 1 + for reaction_id, count in goal.required_events.items(): + if fired.get(reaction_id, 0) < count: + return False + final: Grid = set() + for h in live: + final |= cells_at(catalog, h, goal.horizon) + for motif_id, required in goal.final_min_counts.items(): + if motif_id == "block": + if len(blocks_in(final)) < required: + return False + else: + count = sum( + 1 + for h in live + if is_spawn_like(h) and h.motif_id == motif_id + ) + if count < required: + return False + return True + + +def reaction_useful(catalog: Catalog, goal: Goal, reaction_id: str) -> bool: + """Une réaction contribue-t-elle à l'objectif (filtrage R2) ? + + Heuristique : la réaction est un événement exigé, ou produit (par + placement ou par cellules brutes) un motif compté par l'objectif. + """ + if reaction_id in goal.required_events: + return True + reaction = catalog.reactions[reaction_id] + for placement in reaction.products: + if placement.motif_id in goal.final_min_counts: + return True + if reaction.product_cells is not None: + if "block" in goal.final_min_counts and blocks_in(set(reaction.product_cells)): + return True + return False + + +# --------------------------------------------------------------------------- +# Compatibilité des ports (sémantique de la tranche 3) +# --------------------------------------------------------------------------- + + +def port_compatible(catalog: Catalog, motif_id: str, port_direction: tuple[int, int]) -> bool: + """Compatibilité port <-> motif, selon la sémantique documentée. + + Réactif mobile : les axes non nuls du port doivent être exactement les + axes non nuls de la translation par période, avec les mêmes signes. + Réactif statique (translation nulle) : tout port non nul est accepté — + le port décrit l'orientation du connecteur, pas un mouvement. + """ + motif = catalog.motifs[motif_id] + tx, ty = motif.translation_step + dx, dy = port_direction + if tx == 0 and ty == 0: + return not (dx == 0 and dy == 0) + if (tx == 0) != (dx == 0) or (ty == 0) != (dy == 0): + return False + if dx != 0 and (dx > 0) != (tx > 0): + return False + if dy != 0 and (dy > 0) != (ty > 0): + return False + return True + + +# --------------------------------------------------------------------------- +# Recherche +# --------------------------------------------------------------------------- + + +def _new_counters() -> dict[str, int]: + keys = [ + "candidates", + "r1_anchor_collapsed", + "r1_state_dedup", + "r2_useful", + "r2_type", + "r2_port", + "r3_envelope", + "r3_clearance", + "r4_phase", + "r4_window", + "model_anchor", + "budget", + "replay_rejected", + "certified", + "nodes", + ] + return {k: 0 for k in keys} + + +@dataclass +class SearchState: + spawns: tuple[Spawn, ...] = () + events: tuple[EventInst, ...] = () + live: tuple[Handle, ...] = () + + +def _handle_desc(h: Handle) -> tuple: + """Descripteur structurel : ce dont le futur de l'objet dépend. + + L'identité de bookkeeping (index de l'événement producteur) est remplacée + par la géométrie et le temps de naissance : deux états où le même objet + naît au même endroit au même instant sont isomorphes, quel que soit + l'ordre de tir des épisodes indépendants. + """ + if isinstance(h, Spawn): + return ("spawn", h.motif_id, h.anchor, h.phase0) + if isinstance(h, ProductBirth): + return ("product", h.motif_id, h.anchor, h.t0) + return ("static", tuple(sorted(h.cells)), h.t0) + + +def _state_key(state: SearchState) -> tuple: + """Clé de dédup canonique : multiset des spawns et des événements. + + Deux états de même clé ne diffèrent que par l'ORDRE de tir d'épisodes + indépendants — et la sémantique du module est insensible à cet ordre : + `certify` calcule la scène pas à pas comme une union (fenêtres indexées + par temps, consommation last-writer unique puisqu'un réactif n'est lié + qu'une fois), et `goal_reached` compte des identifiants de réaction. Le + `live` n'est pas dans la clé : il est dérivé de (spawns, events). + + Deux clés TROP GROSSIÈRES ont produit des faux négatifs de complétude + (mesurés) : une clé indépendante de la position avalait le produit d'un + second épisode né ailleurs ; une clé réduite aux vivants + nombre + d'événements confondait deux scènes aux mêmes produits à des temps de tir + différents — sous ablation de R4, un état non certifiant partageant la + clé du témoin était déplié d'abord, le témoin dédupliqué, et la recherche + concluait IMPOSSIBLE_BOUNDED alors qu'un témoin existait. La clé porte + désormais tout ce dont les transitions dépendent. + """ + spawns = tuple( + sorted((s.motif_id, s.anchor, s.phase0) for s in state.spawns) + ) + events = tuple( + sorted( + ( + ev.reaction_id, + ev.offset, + ev.t_fire, + tuple(sorted(_handle_desc(b) for b in ev.bindings)), + ) + for ev in state.events + ) + ) + return (spawns, events) + + +class Searcher: + """Parcours déterministe des scènes compositionnelles bornées.""" + + def __init__( + self, + catalog: Catalog, + goal: Goal, + ablations: frozenset[str] = frozenset(), + allowed_spawns: set[str] | None = None, + node_budget: int = DEFAULT_NODE_BUDGET, + allow_static_spawns: bool = False, + ) -> None: + self.catalog = catalog + self.goal = goal + self.ablations = ablations + self.allowed = allowed_spawns + self.node_budget = node_budget + self.allow_static_spawns = allow_static_spawns + self.counters = _new_counters() + self.seen_states: set[tuple] = set() + self.witness: dict[str, Any] | None = None + self.certified_count = 0 + + # -- critères ---------------------------------------------------------- + + def _end_times( + self, + state: SearchState, + new_spawns: Sequence[Spawn], + event: EventInst, + consumed: set[tuple], + ) -> dict[tuple, int]: + """Instant de fin de vie de chaque composant du candidat. + + Un composant consommé par l'événement cesse d'exister à t_fire : + le balayer au-delà ferait dériver son enveloppe après sa mort et + rejetterait des scènes valides (leçon mesurée sur le premier run). + """ + ends: dict[tuple, int] = {} + for h in list(state.live) + list(new_spawns): + key = handle_key(h) + ends[key] = event.t_fire if key in consumed else self.goal.horizon + return ends + + def _spawn_allowed(self, motif_id: str) -> bool: + if self.allowed is not None and motif_id not in self.allowed: + return False + motif = self.catalog.motifs[motif_id] + if motif.kind in self.goal.spawn_kinds: + return True + return self.allow_static_spawns + + def _extent_grid(self, state: SearchState) -> list[tuple[int, int]]: + """Grille de décalages autour de l'emprise courante (énumération e). + + Deux échelles : une grille fine à ±4 (alignements proches) et un + anneau grossier au pas 4 borné par la surface, pour garer un épisode + indépendant loin de la scène existante -- la liaison sur un handle + existant résout e par équation, l'anneau ne sert donc qu'aux spawns + neufs (scènes multi-épisodes disjointes). + """ + rects = [ + swept_rect(self.catalog, h, 0, self.goal.horizon) + for h in state.live + ] + rects = [r for r in rects if r is not None] + if not rects: + return [(0, 0)] + x0 = min(r[0] for r in rects) - 4 + x1 = max(r[0] + r[2] for r in rects) + 4 + y0 = min(r[1] for r in rects) - 4 + y1 = max(r[1] + r[3] for r in rects) + 4 + fine = [ + (x, y) + for x in range(x0, x1 + 1) + for y in range(y0, y1 + 1) + ] + reach = max(self.goal.surface) // 2 + ring = [ + (x, y) + for x in range(x0 - reach, x1 + reach + 1, 4) + for y in range(y0 - reach, y1 + reach + 1, 4) + if not (x0 <= x <= x1 and y0 <= y <= y1) + ] + return fine + ring + + def _surface_ok( + self, + handles: Sequence[Handle], + events: Sequence[EventInst], + ends: dict[tuple, int], + ) -> bool: + """Emprise globale de la scène (enveloppes + fenêtres) sous budget.""" + rects: list[tuple[int, int, int, int]] = [] + for h in handles: + r = swept_rect(self.catalog, h, 0, ends[handle_key(h)]) + if r is not None: + rects.append(r) + for event in events: + reaction = self.catalog.reactions[event.reaction_id] + window = reaction_window_cells(self.catalog, reaction, event.offset) + ever: Grid = set() + for cells in window: + ever |= cells + r = _rect_of(ever) + if r is not None: + rects.append(r) + if not rects: + return True + x0 = min(r[0] for r in rects) + y0 = min(r[1] for r in rects) + w = max(r[0] + r[2] for r in rects) - x0 + h = max(r[1] + r[3] for r in rects) - y0 + return w <= self.goal.surface[0] and h <= self.goal.surface[1] + + def _spatial_ok( + self, + state: SearchState, + new_spawns: Sequence[Spawn], + event: EventInst, + consumed: set[tuple], + ) -> bool: + """R3 : non-interaction spatiale de toute paire non réagissante. + + Paires réagissantes (consommées par le même événement) : distance + de Chebyshev exacte >= 2 pour tout t < t_fire (elles doivent se + rejoindre à l'instant déclaré, pas avant). + Paires non réagissantes : enveloppes balayées gonflées disjointes + sur la coexistence ; composants vs fenêtres/clearances des événements. + """ + catalog = self.catalog + horizon = self.goal.horizon + handles = list(state.live) + list(new_spawns) + ends = self._end_times(state, new_spawns, event, consumed) + # fenêtres : cet événement + les événements déjà tirés + all_events = list(state.events) + [event] + windows = { + ev.idx: reaction_window_cells( + catalog, catalog.reactions[ev.reaction_id], ev.offset + ) + for ev in all_events + } + for i in range(len(handles)): + for j in range(i + 1, len(handles)): + a, b = handles[i], handles[j] + ka, kb = handle_key(a), handle_key(b) + reacting_pair = ka in consumed and kb in consumed + if reacting_pair: + for t in range(max(t_start(a), t_start(b)), event.t_fire): + if _min_chebyshev( + cells_at(catalog, a, t), cells_at(catalog, b, t) + ) < 2: + return False + else: + t_end = min(ends[ka], ends[kb]) + ra = swept_rect(catalog, a, 0, t_end) + rb = swept_rect(catalog, b, 0, t_end) + if ra and rb and _rects_overlap(_grown(ra), _grown(rb)): + # emprises proches : contrôle exact, instant par instant + # (deux blocs séparés d'une colonne sont à Chebyshev 2 : + # non-interaction réelle que les rectangles se touchent) + for t in range(max(t_start(a), t_start(b)), t_end + 1): + if _min_chebyshev( + cells_at(catalog, a, t), cells_at(catalog, b, t) + ) < 2: + return False + for h in handles: + if handle_key(h) in consumed: + continue + for ev in all_events: + # un produit EST la dernière image de la fenêtre de son + # propre événement : le confronter à elle est un faux positif + # (le replay couvre déjà la transition aux bornes exactes) + if isinstance(h, (ProductBirth, StaticProduct)) and h.event_idx == ev.idx: + continue + reaction = catalog.reactions[ev.reaction_id] + window = windows[ev.idx] + ever: Grid = set() + for cells in window: + ever |= cells + wrect = _rect_of(ever) + if wrect is None: + continue + r = swept_rect(catalog, h, 0, ends[handle_key(h)]) + if r is None: + continue + if not _rects_overlap(_grown(wrect), _grown(r)): + continue + # emprises proches : contrôle exact, instant par instant + for k, cells in enumerate(window): + t = ev.t_fire + k + hc = cells_at(catalog, h, t) + if hc and _min_chebyshev(hc, cells) < 2: + return False + for rect in reaction.clearance: + x0, y0, w, hgt = rect + abs_rect = (x0 + ev.offset[0], y0 + ev.offset[1], w, hgt) + for t in range(ev.t_fire, ev.t_fire + reaction.stabilization_time): + for cell in cells_at(catalog, h, t): + if _rect_contains_cell(abs_rect, cell): + return False + return True + + # -- énumération --------------------------------------------------------- + + def _enumerate_events( + self, state: SearchState, depth: int + ) -> Iterator[tuple[EventInst, tuple[Spawn, ...], set[tuple]]]: + catalog = self.catalog + goal = self.goal + counters = self.counters + ablate = self.ablations + for reaction_id in sorted(catalog.reactions): + reaction = catalog.reactions[reaction_id] + if "r2" not in ablate and not reaction_useful(catalog, goal, reaction_id): + counters["r2_useful"] += 1 + continue + stab = reaction.stabilization_time + for t_fire in range(0, goal.horizon - stab + 1): + option_lists: list[list[tuple[str, Handle | None]]] = [] + feasible = True + for i, decl in enumerate(reaction.reactants): + options: list[tuple[str, Handle | None]] = [] + port = reaction.ports[i] if i < len(reaction.ports) else None + for h in state.live: + if not is_spawn_like(h): + continue + if t_start(h) > t_fire: + counters["r4_window"] += 1 + continue + motif = catalog.motifs[h.motif_id] + if "r2" not in ablate: + if motif.id != decl.motif_id: + counters["r2_type"] += 1 + continue + if port is not None and not port_compatible( + catalog, motif.id, port.direction + ): + counters["r2_port"] += 1 + continue + if "r4" not in ablate: + ph = phase_at(catalog, h, t_fire) + if ph != decl.phase: + counters["r4_phase"] += 1 + continue + options.append(("bind", h)) + if self._spawn_allowed(decl.motif_id): + options.append(("spawn", None)) + if not options: + feasible = False + counters["r2_type"] += 1 + break + option_lists.append(options) + if not feasible: + continue + for combo in itertools.product(*option_lists): + counters["candidates"] += 1 + spawn_slots = [ + i + for i, (kind, _) in enumerate(combo) + if kind == "spawn" + ] + phase_domains = [] + for i in spawn_slots: + decl = reaction.reactants[i] + period = catalog.motifs[decl.motif_id].period + phase_domains.append(range(period)) + for phases in itertools.product(*phase_domains) if phase_domains else [()]: + phase_map = dict(zip(spawn_slots, phases)) + # détermination du décalage e + offset: tuple[int, int] | None = None + consistent = True + for i, (kind, h) in enumerate(combo): + if kind != "bind": + continue + decl = reaction.reactants[i] + anchor = anchor_at(catalog, h, t_fire) # type: ignore[arg-type] + if anchor is None: + consistent = False + break + candidate_e = ( + anchor[0] - decl.offset[0], + anchor[1] - decl.offset[1], + ) + if offset is None: + offset = candidate_e + elif offset != candidate_e: + counters["model_anchor"] += 1 + consistent = False + break + if not consistent: + continue + offset_iter: Iterator[tuple[int, int]] + if offset is not None: + offset_iter = iter([offset]) + elif depth == 0 and "r1" not in ablate: + counters["r1_anchor_collapsed"] += 1 + offset_iter = iter([(0, 0)]) + else: + offset_iter = iter(self._extent_grid(state)) + for e in offset_iter: + # dérivation des spawns (phase énumérée, R4 filtre) + spawns: list[Spawn] = [] + ok = True + for i in spawn_slots: + decl = reaction.reactants[i] + motif = catalog.motifs[decl.motif_id] + tx, ty = motif.translation_step + p0 = phase_map[i] + if "r4" not in ablate: + forced = decl.phase - t_fire + forced %= motif.period + if p0 != forced: + counters["r4_phase"] += 1 + ok = False + break + corner = _motif_phase_offsets( + catalog, decl.motif_id + )[decl.phase] + periods = (p0 + t_fire) // motif.period + anchor = ( + e[0] + decl.offset[0] - corner[0] - tx * periods, + e[1] + decl.offset[1] - corner[1] - ty * periods, + ) + spawns.append( + Spawn(motif_id=decl.motif_id, anchor=anchor, phase0=p0) + ) + if not ok: + continue + bindings = tuple( + h if kind == "bind" else spawns[spawn_slots.index(i)] + for i, (kind, h) in enumerate(combo) + ) + consumed = {handle_key(b) for b in bindings} + event = EventInst( + idx=len(state.events), + reaction_id=reaction_id, + offset=e, + t_fire=t_fire, + bindings=bindings, + ) + # budgets + total_spawns = tuple(state.spawns) + tuple(spawns) + gliders = sum( + 1 + for s in total_spawns + if catalog.motifs[s.motif_id].kind == "spaceship" + ) + if ( + len(total_spawns) > goal.max_components + or gliders > goal.max_gliders + ): + counters["budget"] += 1 + continue + if not self._surface_ok( + list(state.live) + spawns, + list(state.events) + [event], + self._end_times(state, spawns, event, consumed), + ): + counters["budget"] += 1 + continue + # R3 spatial + if "r3" not in ablate and not self._spatial_ok( + state, spawns, event, consumed + ): + counters["r3_envelope"] += 1 + continue + yield event, tuple(spawns), consumed + + # -- parcours ------------------------------------------------------------ + + def run(self) -> dict[str, Any]: + try: + self._search(SearchState(), 0) + except BudgetHit: + if self.witness is not None: + return self._report("FOUND") + return self._report("TIMEOUT") + if self.witness is not None: + return self._report("FOUND") + return self._report("IMPOSSIBLE_BOUNDED") + + def _search(self, state: SearchState, depth: int) -> None: + self.counters["nodes"] += 1 + if self.counters["nodes"] > self.node_budget: + raise BudgetHit + if goal_reached(self.catalog, self.goal, state.events, state.live): + verdict = certify(self.catalog, self.goal, state.spawns, state.events) + if verdict["ok"]: + self.counters["certified"] += 1 + if self.witness is None: + self.witness = { + "spawns": [s.__dict__ | {"anchor": list(s.anchor)} for s in state.spawns], + "events": [ + { + "reaction_id": ev.reaction_id, + "offset": list(ev.offset), + "t_fire": ev.t_fire, + "bindings": [ + handle_key(b).__repr__() for b in ev.bindings + ], + } + for ev in state.events + ], + "initial_cells": sorted( + tuple(c) + for s in state.spawns + for c in cells_at(self.catalog, s, 0) + ), + "final_cells": sorted(tuple(c) for c in verdict["final_cells"]), + "blocks_final": verdict["blocks_final"], + "steps_checked": verdict["steps_checked"], + } + return + self.counters["replay_rejected"] += 1 + return + if depth >= self.goal.max_events: + return + for event, spawns, _consumed in self._enumerate_events(state, depth): + if self.witness is not None: + return + new_live = tuple( + h for h in state.live if handle_key(h) not in {handle_key(b) for b in event.bindings} + ) + tuple(product_handles(self.catalog, event)) + new_state = SearchState( + spawns=state.spawns + spawns, + events=state.events + (event,), + live=new_live, + ) + key = _state_key(new_state) + if "r1" not in self.ablations: + if key in self.seen_states: + self.counters["r1_state_dedup"] += 1 + continue + self.seen_states.add(key) + self._search(new_state, depth + 1) + + def _report(self, verdict: str) -> dict[str, Any]: + return { + "objective": self.goal.name, + "verdict": verdict, + "witness": self.witness, + "stats": dict(self.counters), + } + + +# Convention du fixture (tranche 1+2) : les ports sont listés dans l'ordre des +# réactifs (in0 -> reactants[0], in1 -> reactants[1]). L'énumération indexe +# directement reaction.ports[i]. + + +# --------------------------------------------------------------------------- +# Certification : replay indépendant pas à pas +# --------------------------------------------------------------------------- + + +def certify( + catalog: Catalog, goal: Goal, spawns: Sequence[Spawn], events: Sequence[EventInst] +) -> dict[str, Any]: + """Rejoue la scène entière dans le moteur et vérifie, à chaque pas : + + * scène(t) == union(trajectoires pures) + fenêtres des événements actifs ; + * les zones de clearance ne sont jamais traversées pendant leur fenêtre. + + La trajectoire pure d'un composant non réagissant et la fenêtre d'une + réaction sont deux mondes qui ne se mélangent pas : leur réunion exacte à + chaque pas est la preuve d'absence de rétrointeraction destructive. + """ + handles: list[Handle] = list(spawns) + for event in events: + handles.extend(product_handles(catalog, event)) + consumed_at: dict[tuple, int] = {} + for event in events: + for b in event.bindings: + consumed_at[handle_key(b)] = event.t_fire + windows = { + ev.idx: reaction_window_cells(catalog, catalog.reactions[ev.reaction_id], ev.offset) + for ev in events + } + scene: Grid = set() + for s in spawns: + scene |= cells_at(catalog, s, 0) + initial_cells = set(scene) + mismatches: list[dict[str, Any]] = [] + clearance_violations: list[dict[str, Any]] = [] + final_cells: Grid = set() + for t in range(goal.horizon + 1): + expected: Grid = set() + for h in handles: + key = handle_key(h) + if t < t_start(h): + continue + if consumed_at.get(key, 10**9) <= t: + continue + expected |= cells_at(catalog, h, t) + for ev in events: + reaction = catalog.reactions[ev.reaction_id] + if ev.t_fire <= t < ev.t_fire + reaction.stabilization_time: + expected |= windows[ev.idx][t - ev.t_fire] + if scene != expected: + mismatches.append( + { + "t": t, + "delta": sorted( + tuple(c) for c in (scene ^ expected) + )[:8], + } + ) + for ev in events: + reaction = catalog.reactions[ev.reaction_id] + if ev.t_fire <= t < ev.t_fire + reaction.stabilization_time: + for (x0, y0, w, h) in reaction.clearance: + abs_rect = (x0 + ev.offset[0], y0 + ev.offset[1], w, h) + hit = sorted( + tuple(c) + for c in scene + if _rect_contains_cell(abs_rect, c) + ) + if hit: + clearance_violations.append({"t": t, "rect": abs_rect, "hit": hit[:4]}) + final_cells = set(scene) + if t < goal.horizon: + scene = step(scene) + return { + "ok": not mismatches and not clearance_violations, + "steps_checked": goal.horizon + 1, + "mismatches": mismatches[:5], + "clearance_violations": clearance_violations[:5], + "final_cells": final_cells, + "blocks_final": blocks_in(final_cells), + } + + +# --------------------------------------------------------------------------- +# Comparaison avec la baseline cellulaire de #15571 +# --------------------------------------------------------------------------- + + +def _run_sig(run: dict[str, Any]) -> dict[str, Any]: + # la stabilité compare le CONTENU des répétitions (verdict, nœuds, + # candidats, témoin) — jamais wall_s/peak, qui varient avec la machine + return {k: v for k, v in run.items() if k not in ("wall_s", "peak_mem_kib")} + + +def run_baseline(reps: int = 3) -> dict[str, Any]: + """Baseline B1 : énumération cellulaire de life_synthesize sur la cible. + + `synthesize(1, (0,0), box, max_cells)` retourne TOUTES les formes de + still lifes de taille <= max_cells dans la boîte : la baseline n'accepte + pas de cible, son coût pour atteindre LA forme visée est donc son + énumération complète (appartenance vérifiée a posteriori). Elle produit + une existence, pas une construction — c'est mesuré, pas caché. + """ + from life_synthesize import synthesize + + box, max_cells = 5, 8 + target = normalize(set(TWO_BLOCKS_TARGET)) + runs = [] + for _ in range(reps): + tracemalloc.start() + t0 = time.perf_counter() + found = synthesize(1, (0, 0), box, max_cells) + wall = time.perf_counter() - t0 + _, peak = tracemalloc.get_traced_memory() + tracemalloc.stop() + normalized = [normalize(set(p)) for p in found] + runs.append( + { + "wall_s": round(wall, 3), + "peak_mem_kib": peak // 1024, + "candidates": sum( + comb(box * box, k) for k in range(1, max_cells + 1) + ), + "still_lifes_found": len(found), + "target_found": target in normalized, + } + ) + return { + "engine": "life_synthesize.synthesize (enumeration B1, #15571)", + "spec": "evolve^1 T = T (still lifes), boite 5x5, <= 8 cellules", + "target": [list(c) for c in target], + "runs": runs, + "stable": len({json.dumps(_run_sig(r), sort_keys=True) for r in runs}) == 1, + } + + +def run_compositional(goal: Goal, reps: int = 3, **kwargs: Any) -> dict[str, Any]: + runs = [] + for _ in range(reps): + _WINDOW_CACHE.clear() + tracemalloc.start() + t0 = time.perf_counter() + searcher = Searcher(load_catalog_fixture(kwargs["fixture"]), goal, **{ + k: v for k, v in kwargs.items() if k != "fixture" + }) + report = searcher.run() + wall = time.perf_counter() - t0 + _, peak = tracemalloc.get_traced_memory() + tracemalloc.stop() + runs.append( + { + "wall_s": round(wall, 4), + "peak_mem_kib": peak // 1024, + "verdict": report["verdict"], + "nodes": report["stats"]["nodes"], + "candidates": report["stats"]["candidates"], + } + ) + return { + "engine": "life_compose (recherche compositionnelle, tranche 3+4)", + "runs": runs, + "stable": len({json.dumps(_run_sig(r), sort_keys=True) for r in runs}) == 1, + "report": report, + } + + +def load_catalog_fixture(path: str) -> Catalog: + return load_catalog(path) + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description=( + "Recherche compositionnelle Life : générateur de contraintes + " + "démonstrateur (tranche 3+4 de #15635)." + ) + ) + parser.add_argument( + "--fixture", + default=str(Path(__file__).resolve().parent / "life_components_fixture.json"), + help="catalogue de composants (tranche 1+2)", + ) + parser.add_argument( + "--objective", + default="two_blocks_catalyse", + choices=sorted(OBJECTIVES), + help="objectif borné à résoudre", + ) + parser.add_argument( + "--ablate", + default="", + help="famille de pruning à désactiver (r1, r2, r3 ou r4)", + ) + parser.add_argument("--node-budget", type=int, default=DEFAULT_NODE_BUDGET) + parser.add_argument( + "--restrict-motifs", + default="", + help="liste de motifs autorisés au spawn, séparés par des virgules (contrôle négatif)", + ) + parser.add_argument( + "--full-report", + action="store_true", + help="exécute base + 4 ablations + comparaison baseline (3 répétitions)", + ) + parser.add_argument("--json", action="store_true") + args = parser.parse_args(argv) + + goal = OBJECTIVES[args.objective] + allowed = set(args.restrict_motifs.split(",")) if args.restrict_motifs else None + ablations = frozenset(a for a in args.ablate.split(",") if a) + unknown = ablations - set(ABLATION_FAMILIES) + if unknown: + parser.error(f"familles d'ablation inconnues : {sorted(unknown)}") + + catalog = load_catalog(args.fixture) + + if args.full_report: + payload = { + "objective": args.objective, + "base": run_compositional( + goal, fixture=args.fixture, node_budget=args.node_budget + ), + "ablations": { + fam: run_compositional( + goal, + fixture=args.fixture, + ablations=frozenset([fam]), + node_budget=args.node_budget, + ) + for fam in ABLATION_FAMILIES + }, + } + if args.objective == "two_blocks": + payload["baseline"] = run_baseline(reps=3) + print(json.dumps(payload, ensure_ascii=False, indent=2)) + return 0 + + tracemalloc.start() + t0 = time.perf_counter() + searcher = Searcher( + catalog, + goal, + ablations=ablations, + allowed_spawns=allowed, + node_budget=args.node_budget, + ) + report = searcher.run() + report["wall_s"] = round(time.perf_counter() - t0, 4) + _, peak = tracemalloc.get_traced_memory() + tracemalloc.stop() + report["peak_mem_kib"] = peak // 1024 + + if args.json: + print(json.dumps(report, ensure_ascii=False, indent=2)) + return 0 if report["verdict"] == "FOUND" else 1 + + verdict = report["verdict"] + stats = report["stats"] + print(f"Objectif : {goal.name} — verdict : {verdict} ({report['wall_s']} s)") + print( + f" noeuds={stats['nodes']} candidats={stats['candidates']} " + f"certifiés={stats['certified']} replay_rejetés={stats['replay_rejected']}" + ) + print( + f" pruning : r1(ancre+dedup)={stats['r1_anchor_collapsed'] + stats['r1_state_dedup']} " + f"r2(useful/type/port)={stats['r2_useful'] + stats['r2_type'] + stats['r2_port']} " + f"r3(envelope)={stats['r3_envelope']} r4(phase)={stats['r4_phase']}" + ) + print( + f" rejets modèle : anchor={stats['model_anchor']} budget={stats['budget']} " + f"fenêtre={stats['r4_window']}" + ) + witness = report.get("witness") + if witness: + print(f" témoin : {len(witness['initial_cells'])} cellules initiales") + for ev in witness["events"]: + print( + f" événement {ev['reaction_id']} à t={ev['t_fire']} " + f"décalage={ev['offset']}" + ) + print(f" blocs finaux : {witness['blocks_final']}") + print(f" pas certifiés : {witness['steps_checked']}") + return 0 if verdict == "FOUND" else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/lean/tests/test_life_compose.py b/scripts/lean/tests/test_life_compose.py new file mode 100644 index 0000000000..44f20df2f6 --- /dev/null +++ b/scripts/lean/tests/test_life_compose.py @@ -0,0 +1,496 @@ +"""Tests du générateur de contraintes compositionnelles (tranches 3+4 de #15635). + +Le contrat testé ici est celui des critères 3 a 8 de #15635 : les contraintes +spatiales, temporelles, de ports et de clearance sont toutes ACTIVES et +chacune couverte par un contrôle négatif ; les quatres familles d'élagage +sont instrumentées et ablatables (jamais la certification) ; les verdicts +FOUND / IMPOSSIBLE_BOUNDED / TIMEOUT sont distincts ; le replay indépendant +attrape les erreurs de modèle. + +Deux leçons mesurées de la tranche 3 ont leur test de non-régression : +- la non-interaction universelle est Chebyshev >= 3 ; a distance 2 elle + dépend du contenu (naissances croisées possible) ; +- un placement réactif déclaré doit être ATTEIGNABLE par dérive pure le + long de la direction du port (le placement tranche 1+2 de + block_catalyses_glider démarrait au milieu de l'interaction). +""" + +from __future__ import annotations + +import copy +import json +import sys +from pathlib import Path + +import pytest + +LEAN_DIR = Path(__file__).resolve().parent.parent +FIXTURE = LEAN_DIR / "life_components_fixture.json" + +sys.path.insert(0, str(LEAN_DIR)) + +import life_compose as lc # noqa: E402 +from life_components import load_catalog # noqa: E402 +from life_synthesize import step, synthesize # noqa: E402 + + +CATALOG = load_catalog(FIXTURE) + + +def spawn_cells(motif_id: str, anchor: tuple[int, int], p0: int, t: int) -> set: + return set(lc.cells_at(CATALOG, lc.Spawn(motif_id, anchor, p0), t)) + + +# --------------------------------------------------------------------------- +# Trajectoires exactes (socle du modèle propositionnel) +# --------------------------------------------------------------------------- + + +def test_trajectoires_exactes_contre_le_moteur() -> None: + """cells_at == evolve pour chaque motif, chaque phase, 3 ancres, 13 pas.""" + mismatches = 0 + for motif_id in sorted(CATALOG.motifs): + for p0 in range(CATALOG.motifs[motif_id].period): + for anchor in [(0, 0), (-5, 3), (7, -4)]: + real = spawn_cells(motif_id, anchor, p0, 0) + for t in range(13): + if real != spawn_cells(motif_id, anchor, p0, t): + mismatches += 1 + real = step(real) + assert mismatches == 0 + + +def test_compteur_de_periodes_depends_de_p0() -> None: + """Glider p0=3 a t=1 : une période COMPLETE ecoulee (coin avance d'une + diagonale) -- le compteur est (p0+t)//P, pas t//P.""" + a = spawn_cells("glider", (0, 0), 3, 1) + b = spawn_cells("glider", (1, -1), 0, 0) + assert a == b + + +# --------------------------------------------------------------------------- +# Ports (critere 5 : contraintes d'interface) +# --------------------------------------------------------------------------- + + +def test_semantique_port_mobile_et_statique() -> None: + """in0 mobile = signes de la translation par période ; in1 statique = axe.""" + reaction = CATALOG.reactions["block_catalyses_glider"] + d0 = reaction.ports[0].direction + tx, ty = CATALOG.motifs["glider"].translation_step + assert d0 == (1 if tx > 0 else -1 if tx < 0 else 0, + 1 if ty > 0 else -1 if ty < 0 else 0) + # port statique du bloc : axe connecteur vers la ligne d'approche + assert reaction.ports[1].direction[1] == 0 + + +def test_compatibilite_de_port_filtre_les_directions() -> None: + assert lc.port_compatible(CATALOG, "glider", (1, -1)) + # direction opposée : le glider s'éloignerait du connecteur + assert not lc.port_compatible(CATALOG, "glider", (-1, 1)) + # axe statique sur un reactif mobile : incompatible + assert not lc.port_compatible(CATALOG, "glider", (1, 0)) + # translation orthogonale : pas la ligne du motif + assert not lc.port_compatible(CATALOG, "glider", (1, 1)) + + +def test_congruence_de_phase_a_larrivee() -> None: + """Le reactif mobile doit arriver dans la phase déclarée (R4).""" + reaction = CATALOG.reactions["block_catalyses_glider"] + decl = reaction.reactants[0] + t_fire = 15 + period = CATALOG.motifs["glider"].period + for p0 in range(period): + forced = (decl.phase - t_fire) % period + ph = lc.phase_at(CATALOG, lc.Spawn("glider", (16, 8), p0), t_fire) + assert (ph == decl.phase) == (p0 == forced) + + +# --------------------------------------------------------------------------- +# Physique mesurée : seuils de non-interaction (leçon tranche 3) +# --------------------------------------------------------------------------- + + +def test_chebyshev_2_engendre_des_naissances_croisees() -> None: + """Glider + bloc a distance 2 : naissance combinée mesurée a t=7.""" + block = {(4, 2), (5, 2), (4, 3), (5, 3)} + spawn = lc.Spawn("glider", (-3, 3), 1) + scene = set(lc.cells_at(CATALOG, spawn, 0)) | block + first_dev = None + for t in range(12): + expected = set(lc.cells_at(CATALOG, spawn, t)) | block + if scene != expected: + first_dev = t + break + scene = step(scene) + assert first_dev == 11 # pur jusqu'a t=10 inclus : le depart declare + # la fenetre démarre au DERNIER etat pur ; le produit est un bloc stable + window0 = set(lc.cells_at(CATALOG, spawn, 10)) | block + cells = set(window0) + for _ in range(8): + cells = step(cells) + assert cells == {(2, 3), (3, 3), (2, 4), (3, 4)} + assert step(cells) == cells + + +def test_chebyshev_3_est_sur_sans_interaction() -> None: + """Toute paire d'objets a distance >= 3 n'interagit jamais (propriété).""" + block = {(7, 2), (8, 2), (7, 3), (8, 3)} # a distance 3 du bloc reactif + spawn = lc.Spawn("glider", (-3, 3), 1) + scene = set(lc.cells_at(CATALOG, spawn, 0)) | block + for t in range(16): + expected = set(lc.cells_at(CATALOG, spawn, t)) | block + assert scene == expected, f"interaction inattendue a t={t}" + scene = step(scene) + + +def test_deux_blocs_a_distance_2_sont_stables() -> None: + """Contenu-dépendant : la cible gap-1 est stable malgré la distance 2.""" + target = set(lc.TWO_BLOCKS_TARGET) + assert step(target) == target + + +# --------------------------------------------------------------------------- +# Atteignabilité du placement déclaré (leçon tranche 3) +# --------------------------------------------------------------------------- + + +def _reachability(catalog, doc_reaction: dict) -> tuple[bool, int]: + """Dérive le réactif mobile le long de son port et cherche le fenêtrage. + + Retourne (atteignable, pas de la première déviation) : le placement est + atteignable s'il existe un spawn sur la ligne du port dont la scène reste + pure jusqu'à l'état déclaré exactement. + """ + decl = doc_reaction["reactants"][0] + motif_id = decl["motif_id"] + tx, ty = catalog.motifs[motif_id].translation_step + static = set() + for r in doc_reaction["reactants"][1:]: + phase_cells = lc._motif_phases(catalog, r["motif_id"])[r["phase"]] + dx, dy = r["offset"] + static |= {(x + dx, y + dy) for (x, y) in phase_cells} + # spawn 4 périodes en amont le long de la direction du port + period = catalog.motifs[motif_id].period + corner = lc._motif_phase_offsets(catalog, motif_id)[decl["phase"]] + back = ( + decl["offset"][0] - 4 * tx - corner[0], + decl["offset"][1] - 4 * ty - corner[1], + ) + p0 = decl["phase"] # phase d'arrivée = phase déclarée + spawn = lc.Spawn(motif_id, back, p0) + scene = set(lc.cells_at(catalog, spawn, 0)) | static + declared = set(lc.cells_at(catalog, spawn, 4 * period)) | static + for t in range(4 * period + 1): + expected = set(lc.cells_at(catalog, spawn, t)) | static + if scene != expected: + return False, t + if t == 4 * period: + return scene == declared, t + scene = step(scene) + return False, -1 + + +def test_placement_declare_atteignable_par_derivee() -> None: + doc = json.loads(FIXTURE.read_text(encoding="utf-8")) + entry = [r for r in doc["reactions"] if r["id"] == "block_catalyses_glider"][0] + ok, _ = _reachability(CATALOG, entry) + assert ok + + +def test_placement_milieu_dinteraction_inatteignable() -> None: + """Contrôle négatif : le placement tranche 1+2 (phase 0, bloc (3,0)) + démarrait au milieu de l'interaction -- la dérive dévie avant de l'atteindre.""" + doc = json.loads(FIXTURE.read_text(encoding="utf-8")) + entry = copy.deepcopy( + [r for r in doc["reactions"] if r["id"] == "block_catalyses_glider"][0] + ) + entry["reactants"][0]["phase"] = 0 + entry["reactants"][1]["offset"] = [3, 0] + ok, first_dev = _reachability(CATALOG, entry) + assert not ok + assert first_dev < 4 * CATALOG.motifs["glider"].period + + +# --------------------------------------------------------------------------- +# R3 : élagage spatial et ses non-régressions +# --------------------------------------------------------------------------- + + +def _state_after_e0(): + sp0 = (lc.Spawn("glider", (0, 0), 0), lc.Spawn("glider_mirror", (4, 0), 0)) + ev0 = lc.EventInst( + idx=0, reaction_id="glider_pair_two_blocks", + offset=(0, 0), t_fire=0, bindings=sp0, + ) + live0 = tuple(lc.product_handles(CATALOG, ev0)) + return lc.SearchState(spawns=sp0, events=(ev0,), live=live0), sp0, ev0 + + +def test_produit_pas_confronte_a_sa_propre_fenetre() -> None: + """Non-régression R3 : un candidat E1 distant doit passer _spatial_ok.""" + state0, _, _ = _state_after_e0() + goal = lc.OBJECTIVES["two_blocks_catalyse_free"] + s = lc.Searcher(CATALOG, goal, node_budget=10) + e, tf2 = (20, 4), 15 + gspawn = lc.Spawn("glider", (16, 8), 0) + bspawn = lc.Spawn("block", (24, 6), 0) + ev1 = lc.EventInst( + idx=1, reaction_id="block_catalyses_glider", + offset=e, t_fire=tf2, bindings=(gspawn, bspawn), + ) + consumed = {lc.handle_key(b) for b in ev1.bindings} + assert s._spatial_ok(state0, (gspawn, bspawn), ev1, consumed) + # et la scène complète certifie (le témoin libre est réel) + all_spawns = state0.spawns + (gspawn, bspawn) + new_live = tuple( + h for h in state0.live if lc.handle_key(h) not in consumed + ) + tuple(lc.product_handles(CATALOG, ev1)) + state1 = lc.SearchState(spawns=all_spawns, events=(ev0_of(state0), ev1), live=new_live) + assert lc.goal_reached(CATALOG, goal, state1.events, state1.live) + verdict = lc.certify(CATALOG, goal, state1.spawns, state1.events) + assert verdict["ok"] + assert len(verdict["blocks_final"]) == 3 + + +def ev0_of(state): + return state.events[0] + + +def test_cle_dedup_injective_sur_ancre_et_temps() -> None: + """Non-régression R1 : la clé d'état distingue les ancres ET les tirs. + + Deux faux positifs mesurés, tous deux des faux négatifs de complétude : + (a) une clé indépendante de la position avalait le produit d'un second + épisode ; (b) une clé réduite à la signature des vivants confondait deux + scènes aux mêmes produits à des temps de tir différents — sous ablation + de R4, un état non certifiant partageant la clé du témoin était déplié en + premier, le témoin était dédupliqué, et la recherche concluait + IMPOSSIBLE_BOUNDED alors qu'un témoin existait. + """ + state0, sp0, _ = _state_after_e0() + assert lc._state_key(state0) == lc._state_key(state0) + # (a) même structure, ancre décalée de 9 cellules + sp_b = (lc.Spawn("glider", (9, 9), 0), lc.Spawn("glider_mirror", (13, 9), 0)) + ev_b = lc.EventInst( + idx=0, reaction_id="glider_pair_two_blocks", + offset=(9, 9), t_fire=0, bindings=sp_b, + ) + state_b = lc.SearchState( + spawns=sp_b, events=(ev_b,), live=tuple(lc.product_handles(CATALOG, ev_b)) + ) + # (b) même structure, même ancre, tir décalé d'un pas + ev_c = lc.EventInst( + idx=0, reaction_id="glider_pair_two_blocks", + offset=(0, 0), t_fire=1, bindings=sp0, + ) + state_c = lc.SearchState( + spawns=sp0, events=(ev_c,), live=tuple(lc.product_handles(CATALOG, ev_c)) + ) + assert lc._state_key(state0) != lc._state_key(state_b) + assert lc._state_key(state0) != lc._state_key(state_c) + # et la clé est CANONIQUE : l'ordre de tir des épisodes indépendants ne + # la change pas (certify est insensible à l'ordre : union pas à pas, + # consommation last-writer unique, comptage d'identifiants de réaction) + state_swapped = lc.SearchState( + spawns=sp0, + events=(ev0_of(state0), lc.EventInst( + idx=1, reaction_id="glider_pair_two_blocks", + offset=(9, 9), t_fire=3, bindings=sp_b, + )), + live=(), + ) + state_direct = lc.SearchState( + spawns=sp0, + events=(lc.EventInst( + idx=0, reaction_id="glider_pair_two_blocks", + offset=(9, 9), t_fire=3, bindings=sp_b, + ), ev0_of(state0)), + live=(), + ) + assert lc._state_key(state_swapped) == lc._state_key(state_direct) + + +def test_elagage_rejette_le_spawn_dans_la_fenetre() -> None: + """Contrôle négatif R3 : un bloc posé SUR la fenêtre d'un événement actif + est rejeté (distance 0 aux cellules de fenêtre).""" + state0, _, ev0 = _state_after_e0() + goal = lc.OBJECTIVES["two_blocks_catalyse_free"] + s = lc.Searcher(CATALOG, goal, node_budget=10) + # bloc statique posé sur la trajectoire des gliders de E0 (x in [0,6]) + bspawn = lc.Spawn("block", (2, 1), 0) + gspawn = lc.Spawn("glider", (-40, 44), 0) # loin, n'interagit jamais + ev1 = lc.EventInst( + idx=1, reaction_id="block_catalyses_glider", + offset=(-44, 40), t_fire=55, bindings=(gspawn, bspawn), + ) + consumed = {lc.handle_key(b) for b in ev1.bindings} + assert not s._spatial_ok(state0, (gspawn, bspawn), ev1, consumed) + + +# --------------------------------------------------------------------------- +# Budgets (critere 3 : contraintes bornées) +# --------------------------------------------------------------------------- + + +def test_budgets_composants_et_gliders_rejetes() -> None: + goal = lc.Goal( + name="ping", + final_min_counts={"block": 2}, + required_events={"glider_pair_two_blocks": 1}, + max_events=2, + max_components=2, # la paire seule = 2 composants : budget exact + horizon=24, + surface=(24, 24), + max_gliders=2, + ) + r = lc.run_compositional(goal, fixture=str(FIXTURE), reps=1) + assert r["report"]["verdict"] == "FOUND" + tight = lc.Goal( + name="ping_serre", + final_min_counts={"block": 2}, + required_events={"glider_pair_two_blocks": 1}, + max_events=2, + max_components=1, # impossible : la paire exige 2 spawns + horizon=24, + surface=(24, 24), + max_gliders=2, + ) + r2 = lc.run_compositional(tight, fixture=str(FIXTURE), reps=1) + assert r2["report"]["verdict"] == "IMPOSSIBLE_BOUNDED" + assert r2["report"]["stats"]["budget"] > 0 + + +def test_glider_seuls_impossible_borne() -> None: + """Sans événement, deux gliders restent des gliders : épuisement trivial.""" + goal = lc.Goal( + name="gliders_seuls", + final_min_counts={"block": 2}, + required_events={}, + max_events=0, + max_components=2, + horizon=24, + surface=(24, 24), + max_gliders=2, + ) + r = lc.run_compositional(goal, fixture=str(FIXTURE), reps=1) + assert r["report"]["verdict"] == "IMPOSSIBLE_BOUNDED" + assert r["report"]["stats"]["candidates"] >= 0 # énumération vide mais bornée + + +def test_timeout_distinct_de_impossible() -> None: + """Budget de noeuds épuisé AVANT épuisement de l'espace = TIMEOUT.""" + goal = lc.Goal( + name="large", + final_min_counts={"block": 2}, + required_events={"glider_pair_two_blocks": 1}, + max_events=2, + max_components=2, + horizon=40, + surface=(32, 32), + max_gliders=2, + ) + r = lc.run_compositional(goal, fixture=str(FIXTURE), reps=1, node_budget=1) + assert r["report"]["verdict"] == "TIMEOUT" + + +# --------------------------------------------------------------------------- +# Certification : le replay dispose (critere 7) +# --------------------------------------------------------------------------- + + +def test_replay_rejette_une_erreur_de_modele() -> None: + """Un témoin falsifié (ancre décalée d'une cellule) est rejeté par certify.""" + sp0 = (lc.Spawn("glider", (0, 0), 0), lc.Spawn("glider_mirror", (4, 0), 0)) + ev0 = lc.EventInst( + idx=0, reaction_id="glider_pair_two_blocks", + offset=(0, 0), t_fire=0, bindings=sp0, + ) + goal = lc.OBJECTIVES["two_blocks"] + assert lc.certify(CATALOG, goal, sp0, (ev0,))["ok"] + faux = (lc.Spawn("glider", (1, 0), 0), sp0[1]) # glider décalé + assert not lc.certify(CATALOG, goal, faux, (ev0,))["ok"] + + +def test_clearance_jamais_traversee_pendant_la_fenetre() -> None: + """Contrôle négatif : un bloc tiers posé DANS la zone de clearance + (mais a distance >= 3 de l'épisode, sans perturber la physique) est + attrapé par la règle de clearance de la certification.""" + goal = lc.OBJECTIVES["two_blocks_catalyse_free"] + sp0 = (lc.Spawn("glider", (0, 0), 0), lc.Spawn("glider_mirror", (4, 0), 0)) + ev0 = lc.EventInst( + idx=0, reaction_id="glider_pair_two_blocks", + offset=(0, 0), t_fire=0, bindings=sp0, + ) + e, tf2 = (20, 4), 15 + gspawn = lc.Spawn("glider", (16, 8), 0) + bspawn = lc.Spawn("block", (24, 6), 0) + ev1 = lc.EventInst( + idx=1, reaction_id="block_catalyses_glider", + offset=e, t_fire=tf2, bindings=(gspawn, bspawn), + ) + # coin de clearance (reaction) = (-3,3) : a l'interieur du rect, a + # distance >= 3 de toute cellule de l'episode (aucune interaction physique) + rogue = lc.Spawn("block", (-3 + e[0], 3 + e[1]), 0) + spawns = sp0 + (gspawn, bspawn, rogue) + verdict = lc.certify(CATALOG, goal, spawns, (ev0, ev1)) + assert not verdict["ok"] + assert verdict["clearance_violations"] + # sans le bloc voyou, la même scène certifie + verdict_ok = lc.certify(CATALOG, goal, sp0 + (gspawn, bspawn), (ev0, ev1)) + assert verdict_ok["ok"] + + +# --------------------------------------------------------------------------- +# Verdicts de bout en bout (critere 8 : démonstrateur discriminant) +# --------------------------------------------------------------------------- + + +def test_two_blocks_found_et_deterministe() -> None: + results = [lc.run_compositional(lc.OBJECTIVES["two_blocks"], fixture=str(FIXTURE), reps=1) for _ in range(3)] + assert all(r["report"]["verdict"] == "FOUND" for r in results) + w0 = results[0]["report"]["witness"] + assert all(r["report"]["witness"] == w0 for r in results[1:]) + assert len(w0["blocks_final"]) == 2 + assert w0["steps_checked"] > 0 + + +def test_reutilisation_stricte_impossible_borne() -> None: + """Variante stricte : tout candidat de réutilisation est rejeté par le + replay (naissances croisées a distance 2 pendant la transmutation).""" + r = lc.run_compositional(lc.OBJECTIVES["two_blocks_catalyse"], fixture=str(FIXTURE), reps=1) + assert r["report"]["verdict"] == "IMPOSSIBLE_BOUNDED" + assert r["report"]["stats"]["replay_rejected"] >= 1 + assert r["report"]["witness"] is None + + +def test_variante_libre_found() -> None: + r = lc.run_compositional(lc.OBJECTIVES["two_blocks_catalyse_free"], fixture=str(FIXTURE), reps=1) + assert r["report"]["verdict"] == "FOUND" + fired = {ev["reaction_id"] for ev in r["report"]["witness"]["events"]} + assert fired == {"block_catalyses_glider", "glider_pair_two_blocks"} + assert len(r["report"]["witness"]["blocks_final"]) >= 2 + + +@pytest.mark.parametrize("famille", ["r1", "r2", "r3", "r4"]) +def test_ablations_conservent_le_verdict(famille: str) -> None: + """Ablation d'une famille d'élagage : le verdict de two_blocks tient + (la certification n'est JAMAIS ablatable).""" + r = lc.run_compositional(lc.OBJECTIVES["two_blocks"], fixture=str(FIXTURE), reps=1, ablations=frozenset([famille])) + assert r["report"]["verdict"] == "FOUND" + + +# --------------------------------------------------------------------------- +# Baseline cellulaire #15571 : machinerie rapide (le rapport complet la +# rejoue en reps=3 sur la vraie cible, 1,8 M candidats ~ 58 s) +# --------------------------------------------------------------------------- + + +def test_machinerie_baseline_sur_petite_boite() -> None: + found = synthesize(1, (0, 0), 4, 4) + shapes = {frozenset(s) for s in found} + bloc = frozenset({(0, 0), (1, 0), (0, 1), (1, 1)}) + assert bloc in shapes + for s in found: + cells = set(s) + assert step(cells) == cells # tout résultat est bien un still life