Fix(guard,#24721): activity-leak blanchit les Lnnn ancrés dans les outputs antérieurs

scripts/notebook_tools/check_activity_leak.py

@@ -167,6 +167,23 @@
 # controle positif qui l'empeche de s'elargir en silence est
 # test_file_target_mask_is_span_local.
 _FILE_TARGET_RE = re.compile(r"[\w./-]+\.(?:ipynb|md)\b")
+# Ancre de ligne IMPRIMEE par une cellule de code anterieure : « (L1380) »
+# dans une « Lecture du resultat » renvoie a une ligne du listing que la
+# cellule au-dessus vient d'imprimer -- meme forme que le renvoi aux lecons,
+# sens inverse, troisieme famille apres les permaliens et les cibles de lien.
+# Mesure du 2026-11-04 (KNOTS-03, PR #24654) : 3 des 6 ancres qui faisaient
+# rougir le marqueur etaient des citations d'output (L1053/L1380/L3657
+# presentes dans le listing imprime au-dessus) ; les 3 autres etaient des
+# renvois de lignes de FICHIER source et restent comptees -- leur forme
+# canonique est la prose (« ligne 234 »). Le masque est porte par
+# l'APPELANT, qui connait l'ordre des cellules : un token ne blanchit que
+# s'il figure dans les outputs des cellules code STRICTEMENT anterieures --
+# une citation suit sa preuve (convention d'ordonnancement des « Lecture du
+# resultat » : l'interpretation vient APRES la cellule interpretee). Comme
+# les familles precedentes, le predicat est le token exact sans regard sur
+# la prose ; l'ecart est assume et fige par
+# test_lesson_tag_anchored_in_prior_output_is_masked.
+_ANCHORED_TOKEN_RE = re.compile(r"\bL\d{3,4}\b")
 _DM_RE = re.compile(r"\bDM\b")
 
 _LANE_RE = re.compile(r"\blanes?\b", re.IGNORECASE)
@@ -310,6 +327,43 @@ def _context_guarded(text: str, at: int, end: int, ok_context: re.Pattern[str])
     return bool(ok_context.search(window))
 
 
+def _output_text_tokens(cell: dict) -> set[str]:
+    """Tokens L<d> imprimes par les outputs d'une cellule code.
+
+    Le ``text`` d'un output nbformat est une chaine OU une liste de lignes :
+    aplatir avant extraction, sinon ``findall`` leve TypeError sur une liste
+    (piege mesure sur #24654). Les outputs non textuels (images, erreurs)
+    sont ignores : une ancre d'output se cite dans du texte imprime.
+    """
+    tokens: set[str] = set()
+    for out in cell.get("outputs", []) or []:
+        if not isinstance(out, dict):
+            continue
+        text = out.get("text")
+        if isinstance(text, list):
+            text = "".join(str(part) for part in text)
+        if not text:
+            continue
+        tokens.update(_ANCHORED_TOKEN_RE.findall(text))
+    return tokens
+
+
+def anchored_line_tokens(cells: list[dict], upto: int) -> frozenset[str]:
+    """Tokens ancrables pour la cellule d'index ``upto`` : ceux imprimes par
+    les cellules code STRICTEMENT anterieures.
+
+    Une citation suit sa preuve (convention d'ordonnancement des « Lecture du
+    resultat ») : un token imprime plus bas ne peut pas etre la source d'une
+    ancre citee plus haut. Les marcheurs accumulent eux-memes au fil du
+    carnet ; cette fonction sert aux tests et aux appels ponctuels.
+    """
+    tokens: set[str] = set()
+    for cell in cells[:upto]:
+        if cell.get("cell_type") == "code":
+            tokens |= _output_text_tokens(cell)
+    return frozenset(tokens)
+
+
 # Marqueurs qui ne comptent QUE s'ils sont accompagnes d'un autre marqueur
 # dans le meme passage. « cellule N » est le cas mesure : dans le corpus il
 # sert massivement de renvoi INTRA-carnet (« Sharpe 0.063 (cellule 12) »,
@@ -321,14 +375,20 @@ def _context_guarded(text: str, at: int, end: int, ok_context: re.Pattern[str])
 CO_MARKER_ONLY = frozenset({"cellule"})
 
 
-def markers_in_text(text: str) -> list[tuple[str, str, int]]:
+def markers_in_text(text: str,
+                    anchored: frozenset[str] = frozenset()) -> list[tuple[str, str, int]]:
     """Rend [(nom, extrait, poids)] pour tous les marqueurs de ``text``.
 
     Les gardes anti-FP s'appliquent par occurrence, sur son contexte :
     un « DM » colle a « Diebold-Mariano » ou « p-value » est le test
     statistique, pas un message du coordinateur. Les marqueurs de
     ``CO_MARKER_ONLY`` ne sont retenus que si un AUTRE marqueur est present
     dans le meme passage.
+
+    ``anchored`` porte les tokens L<d> imprimes par les outputs des cellules
+    code STRICTEMENT anterieures (cf ``anchored_line_tokens``) : un tel token
+    dans la prose est une ancre de citation d'output, pas un renvoi aux
+    lecons -- il est masque pour ``lesson_tag`` seulement.
     """
     out: list[tuple[str, str, int]] = []
     for name, rx, weight in ALL_MARKERS:
@@ -340,6 +400,13 @@ def markers_in_text(text: str) -> list[tuple[str, str, int]]:
             # plus (#23004) -- ses L<d> sont des tailles de systeme.
             haystack = _URL_FRAGMENT_RE.sub(lambda m: " " * len(m.group(0)), text)
             haystack = _FILE_TARGET_RE.sub(lambda m: " " * len(m.group(0)), haystack)
+            for tok in anchored:
+                # 3e famille (#24721) : token imprime par un output ANTERIEUR
+                # = ancre de citation. Bornes de mots : L138 ne blanchit pas
+                # L1380.
+                haystack = re.sub(
+                    r"\b" + re.escape(tok) + r"\b", " " * len(tok), haystack
+                )
         for m in rx.finditer(haystack):
             if name == "dm" and _context_guarded(text, m.start(), m.end(), _DM_OK_CONTEXT):
                 continue
@@ -376,8 +443,9 @@ def preview(self) -> str:
         return (first[:70] + "...") if len(first) > 70 else first
 
 
-def judge_passage(cell_index: int, text: str) -> PassageVerdict:
-    markers = markers_in_text(text)
+def judge_passage(cell_index: int, text: str,
+                  anchored: frozenset[str] = frozenset()) -> PassageVerdict:
+    markers = markers_in_text(text, anchored)
     score = sum(w for _, _, w in markers)
     strong = sum(1 for _, _, w in markers if w >= 2)
     density = score * 1000.0 / max(len(text), 1)
@@ -423,7 +491,8 @@ def dense_reason(v: PassageVerdict) -> str:
 # titre est bas (un seul marqueur fort suffit) parce qu'un titre qui porte
 # un identifiant de cycle a deja tout dit -- le cas SC-09 mesure les titres
 # « Corrigendum c.NNNN » a score 4-12.
-def judge_cell(cell_index: int, md_text: str) -> list[PassageVerdict]:
+def judge_cell(cell_index: int, md_text: str,
+               anchored: frozenset[str] = frozenset()) -> list[PassageVerdict]:
     """Juge chaque paragraphe d'une cellule, avec heritage de section.
 
     La cellule est parcourue dans l'ordre : un paragraphe-titre ouvre une
@@ -440,7 +509,7 @@ def judge_cell(cell_index: int, md_text: str) -> list[PassageVerdict]:
     verdicts: list[PassageVerdict] = []
     section_dense = False
     for para in split_paragraphs(md_text):
-        v = judge_passage(cell_index, para)
+        v = judge_passage(cell_index, para, anchored)
         heading = SECTION_HEADING_RE.match(para)
         is_h1 = bool(heading and len(heading.group(1)) == 1)
         is_section_heading = bool(heading and not is_h1)
@@ -486,14 +555,20 @@ def scan_notebook(nb_path: Path, root: Path | None = None) -> NotebookReport | N
         except ValueError:
             rel = None
     report = NotebookReport(path=rel or str(nb_path))
+    # Ancres d'output (#24721) : accumulees au fil des cellules code, un
+    # markdown ne voit que les outputs STRICTEMENT anterieurs.
+    anchored: set[str] = set()
     for idx, cell in enumerate(nb.get("cells", [])):
+        if cell.get("cell_type") == "code":
+            anchored |= _output_text_tokens(cell)
+            continue
         if cell.get("cell_type") != "markdown":
             continue
         src = cell.get("source", "")
         if isinstance(src, list):
             src = "".join(src)
         report.total_md_chars += len(src)
-        for verdict in judge_cell(idx, src):
+        for verdict in judge_cell(idx, src, frozenset(anchored)):
             report.all_passages.append(verdict)
             if verdict.dense:
                 report.dense_chars += len(verdict.text)
@@ -628,13 +703,19 @@ def scan_diff(diff_range: str) -> tuple[list[AddedPassage], list[str], list[str]
         except (OSError, json.JSONDecodeError) as exc:
             incidents.append(f"{rel} : tete illisible ({str(exc)[:80]})")
             continue
+        # Ancres d'output (#24721) : meme accumulation que scan_notebook --
+        # la base ne fournit pas d'ancres, seule la tete ordonne le carnet juge.
+        anchored: set[str] = set()
         for idx, cell in enumerate(head_nb.get("cells", [])):
+            if cell.get("cell_type") == "code":
+                anchored |= _output_text_tokens(cell)
+                continue
             if cell.get("cell_type") != "markdown":
                 continue
             src = cell.get("source", "")
             if isinstance(src, list):
                 src = "".join(src)
-            for verdict in judge_cell(idx, src):
+            for verdict in judge_cell(idx, src, frozenset(anchored)):
                 key = _paragraph_key(verdict.text)
                 if key in base_keys:
                     continue

scripts/tests/test_check_activity_leak.py

@@ -55,6 +55,8 @@ def _load_organ():
 scan_notebook = _mod.scan_notebook
 scan_diff = _mod.scan_diff
 emit_diff = _mod.emit_diff
+anchored_line_tokens = _mod.anchored_line_tokens
+_output_text_tokens = _mod._output_text_tokens
 _markdown_paragraphs_of_blob = _mod._markdown_paragraphs_of_blob
 
 
@@ -734,5 +736,123 @@ def test_scan_notebook_part_on_fixture(self, tmp_path=None):
         self.assertEqual(len(report.dense_passages), 2)
 
 
+class TestOutputAnchoredLessonTags(unittest.TestCase):
+    """Blanchiment #24721 : L<d> ancre dans l'output d'une cellule anterieure.
+
+    Mesure fondatrice (KNOTS-03, PR #24654, 2026-11-04) : sur 6 ancres qui
+    faisaient rougir ``lesson_tag``, 3 (L1053/L1380/L3657) figuraient dans
+    le listing imprime par la cellule code AU-DESSUS de la « Lecture du
+    resultat » -- des citations d'output, pas des renvois aux lecons.
+    """
+
+    KNOTS_PARA = (
+        "La lecture du resultat est directe : la contrainte de taille "
+        "sature (L1380) des la deuxieme vague, tandis que la barre du "
+        "compteur amorti (L1053) ne bouge plus apres la troisieme."
+    )
+
+    def test_lesson_tag_anchored_in_prior_output_is_masked(self):
+        """Fige l'ecart assume du masque d'ancre (cf commentaire de
+        ``_ANCHORED_TOKEN_RE``) : predicat = token exact, sans regard de
+        prose, symetrique des deux familles precedentes."""
+        anchored = frozenset({"L1380", "L1053"})
+        v = judge_passage(0, self.KNOTS_PARA, anchored)
+        self.assertNotIn("lesson_tag", {n for n, _, _ in v.markers})
+        self.assertFalse(v.dense, "une citation d'output n'est pas un recit de cycle")
+
+        # meme passage, sans ancres : le comportement pre-24721 est preserve
+        v_bare = judge_passage(0, self.KNOTS_PARA)
+        self.assertIn("lesson_tag", {n for n, _, _ in v_bare.markers})
+        self.assertTrue(v_bare.dense)
+
+    def test_anchored_mask_is_scoped_to_lesson_tag(self):
+        """Un token ancre ne blanchit pas le recit de coordination voisin."""
+        para = (
+            "La lane a livre la tranche, verdict `domain: fail` du dossier ; "
+            "voir le listing (L1380) pour le compte rendu de la re-execution."
+        )
+        anchored = frozenset({"L1380"})
+        v = judge_passage(0, para, anchored)
+        self.assertNotIn("lesson_tag", {n for n, _, _ in v.markers})
+        self.assertIn("lane", {n for n, _, _ in v.markers})
+        self.assertIn("verdict", {n for n, _, _ in v.markers})
+
+    def test_posterior_output_does_not_anchor(self):
+        """Une citation suit sa preuve : un token imprime PLUS BAS n'ancre
+        rien (une « Lecture du resultat » ne peut pas citer un listing
+        posterieur)."""
+        cells = [
+            {"cell_type": "markdown", "source": self.KNOTS_PARA},
+            {"cell_type": "code", "source": "print('saturation (L1380)')",
+             "outputs": [{"output_type": "stream", "text": "saturation (L1380)\n"}]},
+        ]
+        # rien avant la cellule 0 : pas d'ancres
+        self.assertEqual(anchored_line_tokens(cells, 0), frozenset())
+        # pour la cellule code 1 elle-meme : son propre output ne compte pas
+        self.assertEqual(anchored_line_tokens(cells, 1), frozenset())
+        # la cellule code 1 ancre pour ce qui SUIT (cellule 2), pas avant
+        self.assertEqual(anchored_line_tokens(cells, 2), frozenset({"L1380"}))
+        v = judge_passage(0, self.KNOTS_PARA, anchored_line_tokens(cells, 0))
+        self.assertIn("lesson_tag", {n for n, _, _ in v.markers})
+
+    def test_output_text_list_form_is_flattened(self):
+        """nbformat : ``text`` d'un output = chaine OU liste de lignes. Le
+        piege TypeError sur une liste est mesure sur #24654."""
+        cell_list = {
+            "cell_type": "code", "source": "print(listing)",
+            "outputs": [{"output_type": "stream",
+                         "text": ["seuil (L1380) atteint\n",
+                                   "compteur (L1053) fige\n"]}],
+        }
+        self.assertEqual(_output_text_tokens(cell_list), {"L1380", "L1053"})
+        cell_str = {
+            "cell_type": "code", "source": "print(listing)",
+            "outputs": [{"output_type": "stream", "text": "seuil (L3657)\n"}],
+        }
+        self.assertEqual(_output_text_tokens(cell_str), {"L3657"})
+
+    def test_word_boundary_anchored_l138_does_not_mask_l1380(self):
+        """L138 ancre ne blanchit pas L1380 : bornes de mots dans le masque."""
+        para = "La lecture du resultat (L1380) reste valable sur les grandes tailles."
+        v = judge_passage(0, para, frozenset({"L138"}))
+        self.assertIn("lesson_tag", {n for n, _, _ in v.markers})
+
+    def test_scan_notebook_threads_anchors_in_order(self):
+        """Bout en bout : scan_notebook accumule les ancres au fil des
+        cellules -- la « Lecture du resultat » jugee apres son listing n'est
+        plus dense, et le meme carnet SANS la cellule code le reste."""
+        import json
+        import tempfile
+
+        def build(with_code_cell):
+            cells = []
+            if with_code_cell:
+                cells.append({
+                    "cell_type": "code", "source": "print(listing)",
+                    "outputs": [{"output_type": "stream",
+                                 "text": "seuil (L1380) atteint, compteur (L1053) fige\n"}],
+                })
+            cells.append({"cell_type": "markdown", "source": self.KNOTS_PARA})
+            return {"cells": cells}
+
+        with tempfile.TemporaryDirectory() as d:
+            path = Path(d) / "fixture.ipynb"
+            path.write_text(json.dumps(build(True)), encoding="utf-8")
+            anchored_report = scan_notebook(path)
+            path.write_text(json.dumps(build(False)), encoding="utf-8")
+            bare_report = scan_notebook(path)
+        self.assertEqual(anchored_report.dense_passages, [],
+                         "le listing anterieur ancre les (Lnnn) de la lecture")
+        self.assertTrue(bare_report.dense_passages,
+                        "sans listing anterieur, les Lnnn restent des lecons")
+
+    def test_url_fragment_frozen_regression_stays_green(self):
+        """Garde-regression : la 3e famille ne deroge pas aux deux premieres
+        -- un fragment de permalien ``#L261`` reste masque SANS ancre."""
+        prose_anchor = "Voir la ligne #L261 du fichier pour le detail du calcul."
+        v = judge_passage(0, prose_anchor, frozenset({"L261"}))
+        self.assertNotIn("lesson_tag", {n for n, _, _ in v.markers})
+
+
 if __name__ == "__main__":
     unittest.main()