Fix(picker,#22636): le veto « PR fusionnee < 90 j » ne masque plus une consolidation au-dela de 48 h

scripts/pick_idle_grain.py

@@ -2023,6 +2023,7 @@ def delivered_signal_reason(
     failures: list[int] | None = None,
     merged_pr_probe=None,
     skip_marker: bool = False,
+    skip_merged_pr: bool = False,
 ) -> str | None:
     """Pourquoi ecarter ce candidat de l'urne `grain`, ou ``None``.
 
@@ -2065,6 +2066,13 @@ def delivered_signal_reason(
     # de cloture a plus de 48 h reste ecartee par son LABEL ou par une PR
     # MERGEE recente (une consolidation accumule les PRs : ces deux surfaces
     # disent « un tranche est livree », pas « la consolidation est finie »).
+    # `skip_merged_pr` neutralise la surface « PR MERGEE » a son tour, pour
+    # la seule classe `consolidation` et seulement quand la derniere
+    # livraison par PR est DEJA CONNUE et depasse CONSOLIDATION_LAG_H : une
+    # PR fusionnee de plus de 48 h ne dit plus qu'une tranche vient d'etre
+    # livree, et la sonde n'a plus rien a mesurer. Le label reste intact,
+    # et la voie du tapis ne leve ce court-circuit qu'avec un horodatage
+    # connu -- sans lui, la sonde garde son comportement anterieur.
     if not skip_marker:
         verdict = probe(item["number"], lane)
         if verdict == DELIVERED_SIGNAL_UNPROBED:
@@ -2091,6 +2099,11 @@ def delivered_signal_reason(
     # cycles de suite par le tapis narrow-cache. La sonde est partagee
     # avec `open_cover_reason` via le meme `_counted_probe` parent (meme
     # plafond `DELIVERED_SIGNAL_MAX_PROBES`).
+    # Court-circuit #22500 : consolidation deja datee de plus de 48 h --
+    # ni requete `gh`, ni consommation du budget de sonde, ni repli
+    # UNPROBED qui renverrait le candidat sur le tapis sans le servir.
+    if skip_merged_pr:
+        return None
     merged = merged_pr_probe(item["number"], lane)
     if merged == DELIVERED_SIGNAL_UNPROBED:
         return None
@@ -8104,11 +8117,27 @@ def claims_probe(numbers):
             # lecture d'evenement remplace le filtre recent pour cette classe.
             # Le marqueur n'ecarte plus au-dela de 48 h ; la surface
             # « commentaire » de la sonde pleine est neutralisee elle aussi
-            # (`skip_marker`), mais PAS les deux autres (label, PR MERGEE) --
-            # une consolidation accumule les PRs, ces deux surfaces disent
-            # « une tranche est livree », pas « la consolidation est finie ».
+            # (`skip_marker`), le LABEL reste intact (il continue d'ecarter,
+            # quel que soit l'age), et la surface « PR MERGEE » est
+            # neutralisee seulement quand la derniere livraison par PR est
+            # deja connue et depasse 48 h (`skip_merged_pr`) : une
+            # consolidation accumule les PRs, et une tranche mergee depuis
+            # plus de 48 h ne dit plus « la consolidation est finie ».
             marker_expired = False
+            merged_pr_expired = False
             if is_consolidation(it):
+                # #22500 -- la surface « PR MERGEE » cesse elle aussi
+                # d'ecarter la consolidation au-dela de 48 h : le timestamp
+                # de la derniere livraison par PR est DEJA calcule en amont
+                # (`last_delivery_stamp`, corpus `fetch_merged` partage) --
+                # le reutiliser evite une requete `gh` supplementaire.
+                # Age inconnu (aucune livraison dans la fenetre, ou corpus
+                # indisponible) -> la sonde garde son comportement anterieur.
+                delivery_age_h = _visit_age_hours(
+                    it.get("last_delivery_stamp"), NOW)
+                merged_pr_expired = bool(
+                    delivery_age_h is not None
+                    and delivery_age_h >= CONSOLIDATION_LAG_H)
                 verdict = (False if expired()
                            else user_delivery_event_probe(n))
                 if verdict is None:
@@ -8147,7 +8176,8 @@ def claims_probe(numbers):
                 continue
             reason = delivered_signal_reason(
                 it, args.lane, counted_probe, state["failures"],
-                counted_merged_pr_probe, skip_marker=marker_expired)
+                counted_merged_pr_probe, skip_marker=marker_expired,
+                skip_merged_pr=merged_pr_expired)
             if reason is not None:
                 belt_withheld.append(
                     (it, "LIVRAISON : " + reason

scripts/tests/test_pick_idle_grain_22500_passe2.py

@@ -68,10 +68,10 @@ def _free_claims(items):
     return {it["number"]: (pig.CLAIM_CODE_FREE, "libre") for it in items}
 
 
-def _run(pool, *, event_probe=None, **kw):
+def _run(pool, *, event_probe=None, probe_budget=8, **kw):
     return pig.belt_pick_with_replacements(
         pool, _free_claims(pool), _belt_args(grains=len(pool) + 2),
-        probe_budget=8, delivery_event_probe=event_probe, **kw)
+        probe_budget=probe_budget, delivery_event_probe=event_probe, **kw)
 
 
 # --- point 1 : une consolidation livree n'ecarte que 48 h ------------------
@@ -129,16 +129,99 @@ def test_le_marqueur_neutralise_ne_touche_pas_le_label():
     assert withheld and "candidate-delivered" in withheld[0][1]
 
 
-def test_le_marqueur_neutralise_ne_touche_pas_la_pr_mergee():
-    """Meme chose pour la surface « PR MERGEE < 90 j » : une consolidation
-    accumule les PRs, son etat de livraison ne se lit pas dans le merge d'une
-    tranche. La regle des 48 h neutralise le marqueur, rien d'autre."""
+def test_consolidation_pr_mergee_72h_est_servie():
+    """Une consolidation dont la derniere livraison par PR a plus de 48 h
+    n'est plus ecartee par la surface « PR MERGEE < 90 j » : une tranche
+    mergee depuis 72 h ne dit pas « la consolidation est finie ».
+
+    L'age se lit sur `last_delivery_stamp`, deja calcule en amont par le
+    corpus `fetch_merged` partage : la sonde PR n'a alors plus rien a
+    mesurer, et elle ne doit meme pas etre appelee (une requete `gh` de
+    moins, et aucune consommation du budget de sonde). La sonde ci-dessous
+    rend True, c'est-a-dire le veto historique des 90 jours : sur la version
+    anterieure, elle etait appelee et la consolidation restait invisible."""
     item = _consol(5005, visit_h=72)
+    appels = []
+
+    def _sonde_pr(n, lane=None):
+        appels.append(n)
+        return True
+
     picks, withheld, _state = _run(
         [item], event_probe=lambda n: (True, True, _iso(72)),
-        merged_pr_probe=lambda n, lane=None: True)
-    assert picks == []
-    assert withheld and "PR MERGEE" in withheld[0][1]
+        merged_pr_probe=_sonde_pr)
+    assert [p["number"] for p in picks] == [5005], (
+        "la consolidation doit etre servie, "
+        f"withheld={[w[1] for w in withheld]}")
+    assert withheld == []
+    assert appels == [], (
+        "la sonde PR ne doit pas etre appelee pour une consolidation dont "
+        "la derniere livraison par PR depasse 48 h")
+
+
+def test_consolidation_pr_mergee_2h_est_retenue():
+    """Le veto « PR MERGEE » joue encore tant que la derniere livraison par
+    PR a moins de 48 h : une tranche vient d'etre mergee, la consolidation
+    n'est pas a re-servir.
+
+    La lecture d'evenement rend ici un marqueur ANCIEN (72 h) alors que la
+    derniere PR date de 2 h : le filtre recent ne retient donc pas l'issue,
+    et c'est bien la surface PR seule qui est mesuree. Le grain libre en
+    second sert de temoin : la consolidation retenue est remplacee, et elle
+    n'est PAS re-servie derriere -- « retenue » veut dire qu'elle reste
+    invisible pour CE tirage, pas seulement qu'un motif est affiche."""
+    item = _consol(5008, visit_h=2)
+    libre = _item(5011, visit_h=24 * 40)
+    pool = sorted([libre, item], key=lambda it: pig.belt_sort_key(it, now=NOW))
+    picks, withheld, _state = _run(
+        pool, event_probe=lambda n: (True, True, _iso(72)),
+        merged_pr_probe=lambda n, lane=None: n == 5008, probe_budget=12)
+    assert [p["number"] for p in picks] == [5011], (
+        "seul le grain libre doit etre servi ; la consolidation retenue ne "
+        "doit pas revenir comme remplacement")
+    reasons = [w[1] for w in withheld if w[0]["number"] == 5008]
+    assert reasons and all("PR MERGEE" in r for r in reasons), (
+        f"la consolidation doit etre retenue par le veto PR : {reasons}")
+
+
+def test_grain_ordinaire_pr_mergee_72h_est_toujours_retenu():
+    """La regle des 90 jours ne change pas pour un grain ordinaire : une PR
+    mergee il y a 72 h reste un veto, meme quand le marqueur de commentaire
+    est ancien. La neutralisation des 48 h est propre a la consolidation.
+
+    Le grain libre en second sert de temoin : le grain ordinaire retenu est
+    remplace sans etre re-servi derriere, exactement comme avant."""
+    item = _item(5009, visit_h=72)
+    libre = _item(5012, visit_h=24 * 40)
+    pool = sorted([libre, item], key=lambda it: pig.belt_sort_key(it, now=NOW))
+    picks, withheld, _state = _run(
+        pool, recent_marker_probe=lambda n: False,
+        merged_pr_probe=lambda n, lane=None: n == 5009, probe_budget=12)
+    assert [p["number"] for p in picks] == [5012]
+    reasons = [w[1] for w in withheld if w[0]["number"] == 5009]
+    assert reasons and all("PR MERGEE" in r for r in reasons)
+
+
+def test_label_candidate_delivered_retenu_quel_que_soit_l_age():
+    """Le label `candidate-delivered` n'est pas touche par la regle des 48 h :
+    il continue d'ecarter l'issue, quel que soit l'age de la derniere
+    livraison par PR, et sans qu'aucune sonde soit necessaire. Le grain
+    libre en second sert de temoin : l'issue labellisee est remplacee sans
+    etre re-servie derriere, a 2 h comme a 72 h."""
+    for age in (2, 72):
+        item = _item(5010 + age, visit_h=age,
+                     labels=["consolidation", "candidate-delivered"])
+        libre = _item(5100 + age, visit_h=24 * 40)
+        pool = sorted([libre, item],
+                      key=lambda it: pig.belt_sort_key(it, now=NOW))
+        picks, withheld, _state = _run(
+            pool, event_probe=lambda n, a=age: (True, True, _iso(a)),
+            merged_pr_probe=lambda n, lane=None, a=age: n == 5010 + a,
+            probe_budget=12)
+        assert [p["number"] for p in picks] == [5100 + age], (
+            f"age={age} h : le label doit retenir l'issue")
+        reasons = [w[1] for w in withheld if w[0]["number"] == 5010 + age]
+        assert reasons and all("candidate-delivered" in r for r in reasons)
 
 
 def test_une_lecture_en_echec_ne_neutralise_aucun_marqueur():