feat(coordination): merge_ready refuses a twin index collision with current main

scripts/coordination/merge_ready.py

@@ -42,6 +42,14 @@
 5. organe B.0 ``check_unaddressed_nits.py <PR>`` exit 0 -- code de
    retour capture DIRECTEMENT (subprocess.returncode, jamais a travers
    un pipe) ;
+5bis. si la PR touche ``scripts/notebook_tools/twin_pairs.d/`` : organe
+   ``check_twin_index_collisions.py --base origin/main --head <tete>``
+   exit 0, apres ``git fetch origin main pull/<N>/head``. Deux PRs au
+   meme index sont chacune CLEAN contre ``main`` ; la seconde fait
+   rougir ``main`` des que la premiere est mergee. La relecture a lieu
+   PR par PR, contre le ``main`` du moment, donc apres les merges deja
+   faits dans ce run. Collision -> skip ; fetch ou organe illisible ->
+   skip fail-closed ;
 6. REST ``repos/jsboige/CoursIA/pulls/<N>`` : ``mergeable_state`` ==
    ``clean`` (jusqu'a 12 relectures a 10 s d'intervalle pendant ``unknown`` --
    apres un merge les PRs soeurs passent ``unknown``), et ``head.sha``
@@ -124,6 +132,9 @@
 COORDINATOR_USER = "myia-ai-01"
 GATE_PATH = SCRIPTS_DIR / "check_adjoint_prevalidation.py"
 NITS_PATH = SCRIPTS_DIR / "check_unaddressed_nits.py"
+TWIN_PATH = SCRIPTS_DIR / "notebook_tools" / "check_twin_index_collisions.py"
+TWIN_REGISTRY_PREFIX = "scripts/notebook_tools/twin_pairs.d/"
+REPO_ROOT = SCRIPTS_DIR.parent
 
 # Parapluies et branches de campagne GELES par un veto user, exemption des
 # redressements comprise : definition et historique portes par le module
@@ -559,6 +570,40 @@ def run_nits(runner: Runner, pr: int, gh_env: dict[str, str]) -> int:
     return res.returncode
 
 
+def twin_collision_reason(
+    runner: Runner, pr: int, head: str, files: list[dict]
+) -> str | None:
+    """Etape 5bis : collision d'index twin-pairs contre le ``main`` du moment.
+
+    Ne coute rien aux PRs qui ne touchent pas le registre twin. Pour les
+    autres, un fetch puis l'organe partage : ``None`` si l'organe rend 0,
+    un motif de skip sinon. L'organe rend 1 sur collision, 2 quand il n'a
+    pas pu lire deux revisions -- et « je n'ai pas pu lire » n'est pas
+    « c'est propre » : skip, jamais merge.
+    """
+    touched = any(
+        str((row or {}).get("path") or "").startswith(TWIN_REGISTRY_PREFIX)
+        for row in files
+    )
+    if not touched:
+        return None
+    fetch = runner.run(
+        ["git", "-C", str(REPO_ROOT), "fetch", "--quiet", "origin", "main",
+         f"pull/{pr}/head"]
+    )
+    if fetch.returncode != 0:
+        return "twin-collision-unreadable:fetch"
+    res = runner.run(
+        [sys.executable, str(TWIN_PATH), "--repo", str(REPO_ROOT),
+         "--base", "origin/main", "--head", head]
+    )
+    if res.returncode == 0:
+        return None
+    if res.returncode == 1:
+        return "twin-index-collision"
+    return f"twin-collision-unreadable:rc={res.returncode}"
+
+
 def mergeable_state_and_head(
     runner: Runner, pr: int, gh_env: dict[str, str]
 ) -> tuple[str, str]:
@@ -610,7 +655,7 @@ def merge_pr(runner: Runner, pr: int, head: str, gh_env: dict[str, str]) -> None
 def evaluate_pr(
     view: dict, pr: int, runner: Runner, gh_env: dict[str, str]
 ) -> PRVerdict:
-    """Applique dans l'ordre les 6 controles pre-merge. Tout echec = skip nomme.
+    """Applique dans l'ordre les controles pre-merge (1 a 6, 5bis compris). Tout echec = skip nomme.
 
     Les controles bon marche (brouillon, commentaire, perimetre, tag) passent
     AVANT le gate couteux ; le gate avant les organes B.0 ; le REST en
@@ -658,6 +703,10 @@ def skip(reason: str) -> PRVerdict:
     # 5. organe B.0 (re-verification reelle).
     if run_nits(runner, pr, gh_env) != 0:
         return skip("b0-organ-blocked")
+    # 5bis. collision d'index twin contre le main du moment.
+    reason = twin_collision_reason(runner, pr, gate_head, view.get("files") or [])
+    if reason is not None:
+        return skip(reason)
     # 6. REST : mergeable + tete.
     state, live_head = mergeable_state_and_head(runner, pr, gh_env)
     if state != "clean":

scripts/tests/test_merge_ready.py

@@ -99,6 +99,8 @@ def __init__(
         pulls: list[dict] | None = None,
         merge_rc: int = 0,
         gate_stderr: str = "",
+        fetch_rc: int = 0,
+        twin_rc: int = 0,
     ):
         self.token = token
         self.token_rc = token_rc
@@ -112,6 +114,8 @@ def __init__(
         # stderr du gate : c'est lui qui porte un motif de portee generale
         # (jeton refuse, quota) quand le gate echoue POUR TOUTE la passe.
         self.gate_stderr = gate_stderr
+        self.fetch_rc = fetch_rc
+        self.twin_rc = twin_rc
         self.calls: list[tuple[list[str], dict | None]] = []
         self.sleeps: list[float] = []
 
@@ -163,6 +167,10 @@ def run(self, cmd: list[str], env: dict | None = None) -> mr.RunResult:
             return mr.RunResult(self.gate_rc, payload, self.gate_stderr)
         if len(c) > 1 and "check_unaddressed_nits.py" in c[1]:
             return mr.RunResult(self.nits_rc, "", "")
+        if c[:1] == ["git"] and "fetch" in c:
+            return mr.RunResult(self.fetch_rc, "", "")
+        if len(c) > 1 and "check_twin_index_collisions.py" in c[1]:
+            return mr.RunResult(self.twin_rc, "", "")
         raise AssertionError("commande non scriptee : " + " ".join(c))
 
     def sleep(self, seconds: float) -> None:
@@ -675,3 +683,61 @@ def test_hold_file_override(tmp_path):
     )
     assert rc == 0
     assert lines[-1]["reason"] == "hold:ordre de stack"
+
+
+# --- 5bis. collision d'index twin-pairs -----------------------------------------
+
+TWIN_FILE = "scripts/notebook_tools/twin_pairs.d/sw-5-linked-data/0012-2026-09-25-lane.yaml"
+
+
+def test_twin_organ_not_called_when_registry_untouched(tmp_path):
+    """Une PR hors registre twin ne paie ni fetch ni organe."""
+    runner = ScriptedRunner()
+    rc, lines, _ = run_organ(tmp_path, runner)
+    assert rc == 0
+    assert lines[0]["verdict"] == "would-merge"
+    assert not any("check_twin_index_collisions.py" in f for f in runner.flat())
+    assert not any(cmd[:1] == ["git"] for cmd in runner.cmds())
+
+
+def test_twin_collision_skips(tmp_path):
+    view = default_view(files=("src/a.py", TWIN_FILE))
+    runner = ScriptedRunner(views={123: view}, twin_rc=1)
+    rc, lines, _ = run_organ(tmp_path, runner)
+    assert rc == 0
+    assert lines[0]["verdict"] == "skipped"
+    assert lines[0]["reason"] == "twin-index-collision"
+
+
+def test_twin_clean_merges_and_compares_the_gated_head(tmp_path):
+    view = default_view(files=(TWIN_FILE,))
+    runner = ScriptedRunner(views={123: view}, twin_rc=0)
+    rc, lines, _ = run_organ(tmp_path, runner)
+    assert lines[0]["verdict"] == "would-merge"
+    twin = [c for c in runner.cmds() if len(c) > 1 and "check_twin_index_collisions.py" in c[1]]
+    assert len(twin) == 1
+    assert twin[0][twin[0].index("--head") + 1] == HEAD
+    assert twin[0][twin[0].index("--base") + 1] == "origin/main"
+    # le fetch precede l'organe : le main compare est celui du moment
+    flat = runner.flat()
+    fetch_at = next(i for i, f in enumerate(flat) if f.startswith("git ") and " fetch " in f)
+    twin_at = next(i for i, f in enumerate(flat) if "check_twin_index_collisions.py" in f)
+    assert fetch_at < twin_at
+    assert "pull/123/head" in flat[fetch_at]
+
+
+def test_twin_organ_unreadable_is_fail_closed(tmp_path):
+    view = default_view(files=(TWIN_FILE,))
+    runner = ScriptedRunner(views={123: view}, twin_rc=2)
+    rc, lines, _ = run_organ(tmp_path, runner)
+    assert lines[0]["verdict"] == "skipped"
+    assert lines[0]["reason"] == "twin-collision-unreadable:rc=2"
+
+
+def test_twin_fetch_failure_is_fail_closed(tmp_path):
+    view = default_view(files=(TWIN_FILE,))
+    runner = ScriptedRunner(views={123: view}, fetch_rc=128)
+    rc, lines, _ = run_organ(tmp_path, runner)
+    assert lines[0]["verdict"] == "skipped"
+    assert lines[0]["reason"] == "twin-collision-unreadable:fetch"
+    assert not any("check_twin_index_collisions.py" in f for f in runner.flat())