From 9a9ea2d778f143df3eefd7c921ef13eeeee8b8f4 Mon Sep 17 00:00:00 2001 From: jsboige Date: Fri, 9 Oct 2026 15:23:12 +0200 Subject: [PATCH] feat(picker,#18203): geste 5 -- brancher le journal des tirages dans main() Trois points de consigne (reparation, tapis, volee ponderee) via le helper best-effort _log_tirage_safe : un defaut de journal n'arrete jamais un tirage, et la consigne est inerte sous pytest sans TIRAGE_JOURNAL_PATH (meme commutateur que le cache et la sonde delivered). Module scripts/coordination/tirage_journal.py (merge #19945), API log_tirage verrouillee fichier interprocessus. Une invocation = une entree. Co-Authored-By: Claude Haiku 4.5 (1M context) --- scripts/pick_idle_grain.py | 78 ++++ scripts/tests/test_pick_idle_grain_journal.py | 336 ++++++++++++++++++ 2 files changed, 414 insertions(+) create mode 100644 scripts/tests/test_pick_idle_grain_journal.py diff --git a/scripts/pick_idle_grain.py b/scripts/pick_idle_grain.py index 968c1a12c5..e1beb09363 100644 --- a/scripts/pick_idle_grain.py +++ b/scripts/pick_idle_grain.py @@ -197,6 +197,20 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import gh_identity +# Geste 5 #18203 : journal des tirages, module separe par construction +# (scripts/coordination/tirage_journal.py, merge #19945). Import best-effort : +# un environnement sans le module tire SANS journal plutot que de refuser -- +# le journal est une couche d'observabilite, jamais un gate. +try: + import tirage_journal +except ImportError: + try: + sys.path.insert(0, os.path.join( + os.path.dirname(os.path.abspath(__file__)), "coordination")) + import tirage_journal + except ImportError: + tirage_journal = None # type: ignore[assignment] + REPO = "jsboige/CoursIA" # c.1115 voie 1 (msg-20260912T165428-k6rbfc, ai-01 spec) : klass `delivered` @@ -5614,6 +5628,40 @@ def claims_probe(numbers): return belt_picks, belt_withheld, state +def _log_tirage_safe(*, lane, candidates, retained, urn, mode): + """Geste 5 #18203 : consigner le tirage dans le journal inter-process. + + Best-effort par construction : le journal est une COUCHE + D'OBSERVABILITE, pas un gate -- un defaut d'ecriture (chemin non + inscriptible, verrou fichier expire) ne doit jamais priver la lane + du grain que la commande vient de lui rendre. + + Sous pytest, la consigne est INERTE sans ``TIRAGE_JOURNAL_PATH`` + explicite (meme commutateur d'environnement que le cache et la sonde + delivered, cf main()) : un test unitaire n'ecrit pas dans le state + dir machine. Un test qui veut le journal pose la variable vers un + chemin temporaire. + """ + if tirage_journal is None: + print("(journal de tirage indisponible : module absent)", + file=sys.stderr) + return None + if ("PYTEST_CURRENT_TEST" in os.environ + and not os.environ.get("TIRAGE_JOURNAL_PATH")): + return None + try: + record = tirage_journal.log_tirage( + lane=lane, candidates=list(candidates), retained=retained, + urn=urn, mode=mode) + except Exception as exc: # noqa: BLE001 - observabilite best-effort + print(f"(journal de tirage NON ECRIT : {exc})", file=sys.stderr) + return None + print(f"(journal de tirage : draw_id={record.draw_id} mode={mode} " + f"retenu={retained} candidats={len(list(candidates))})", + file=sys.stderr) + return record + + def main(argv: list[str] | None = None) -> int: # Console Windows cp1252 : un titre d'issue portant un caractere hors table # (fleche U+2192 etc.) fait crasher le print en UnicodeEncodeError et perd @@ -5980,6 +6028,19 @@ def main(argv: list[str] | None = None) -> int: if wip_hit: assignment = ((assignment + "+") if assignment else "") + "drainer-son-wip" grain = grain or (backlog.get("wip_prs") or [None])[0] + if not args.belt: + # Geste 5 #18203 : le chemin reparation est un tirage a part + # entiere -- c'est le plus frequent pour une lane chargee. Sans + # cette entree, le journal ne verrait que les tirages propres + # et sur-estimerait la part du pool reellement proposee. En + # mode belt, pas de consigne ici : le tapis journalise sa propre + # proposition plus bas (une invocation = une entree). + _log_tirage_safe( + lane=args.lane, + candidates=[r["number"] for r in (backlog.get("red") or [])], + retained=(grain.get("number") if isinstance(grain, dict) + else grain), + urn="repair", mode="repair") if args.belt and args.json: # Conserve pour fusion dans la sortie tapis plus bas. repair_payload = { @@ -6211,6 +6272,15 @@ def main(argv: list[str] | None = None) -> int: merged_pr_probe=merged_pr_signal)) belt_delivered_failures = belt_pick_state["failures"] belt_probe_budget_hit = [belt_pick_state["budget_hit"]] + # Geste 5 #18203 : consigner le tirage du tapis -- candidats servis + # et tete retenue. Les ecartes (claims tenus, signal de livraison) + # restent dans la sortie (`withheld`) ; le journal porte la + # proposition faite a la lane, pas l'arbitrage qui l'a precedee. + _log_tirage_safe( + lane=args.lane, + candidates=[it["number"] for it in belt_picks], + retained=belt_picks[0]["number"] if belt_picks else None, + urn="belt", mode="belt") # Banniere legere : le tapis ne refuse jamais, mais rappelle # les DWELL/zone pour le lecteur (information sans journal). if not args.json: @@ -6450,6 +6520,14 @@ def main(argv: list[str] | None = None) -> int: long_visits=long_visits) withheld.extend(claim_conflicts) delivery = recent_delivery(picks) + # Geste 5 #18203 : consigner la volee ponderee. L'urne journalisee est + # celle du grain retenu (tete de la poignee servie) ; une poignee vide + # se consigne aussi -- c'est un fait de tirage, pas une absence. + _log_tirage_safe( + lane=args.lane, + candidates=[p["number"] for p in picks], + retained=picks[0]["number"] if picks else None, + urn=(picks[0]["klass"] if picks else "aucun"), mode="weighted") # Calcule AVANT la branche --json : sans ca, l'avertissement de cache # disparaissait sur la surface que les lanes utilisent reellement (`.vibe/ diff --git a/scripts/tests/test_pick_idle_grain_journal.py b/scripts/tests/test_pick_idle_grain_journal.py new file mode 100644 index 0000000000..3a193e1374 --- /dev/null +++ b/scripts/tests/test_pick_idle_grain_journal.py @@ -0,0 +1,336 @@ +"""Tests du geste 5 #18203 : integration du journal des tirages dans le picker. + +Le module ``scripts/coordination/tirage_journal.py`` (merge #19945) porte +l'API verrouillee ; ces tests verifient les TROIS points de consigne de +``main()`` (reparation, tapis, volee ponderee) et le caractere best-effort +du helper : un defaut de journal ne doit jamais priver la lane du grain que +la commande vient de lui rendre. + +Fichier separe de ``test_pick_idle_grain.py`` et +``test_pick_idle_grain_belt.py`` deliberement : ces derniers sont portes +par des PRs ouvertes (#20052, #19594) -- les helpers sont dupliques ici +plutot que partages, pour que chaque PR avance sans rebase croise. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "coordination")) + +import pick_idle_grain as pig # noqa: E402 +import tirage_journal # noqa: E402 + +LANE = "myia-po-2024:CoursIA-2" + + +def _journal_path(tmp_path): + return tmp_path / "journal.jsonl" + + +# ================================================================== +# Helper _log_tirage_safe : best-effort par construction +# ================================================================== + + +def test_helper_writes_record_and_returns_it(monkeypatch, tmp_path): + """Controle positif : la consigne ecrit un TirageRecord complet.""" + jp = _journal_path(tmp_path) + monkeypatch.setenv("TIRAGE_JOURNAL_PATH", str(jp)) + rec = pig._log_tirage_safe(lane=LANE, candidates=[11, 12], retained=11, + urn="belt", mode="belt") + assert rec is not None and rec.draw_id, ( + "le helper rend l'enregistrement ecrit -- le tirage est traçable" + ) + rows = tirage_journal.read_journal(jp) + assert len(rows) == 1 + assert rows[0].lane == LANE + assert rows[0].candidates == [11, 12] + assert rows[0].retained == 11 + assert rows[0].urn == "belt" + assert rows[0].mode == "belt" + + +def test_helper_is_inert_under_pytest_without_env(monkeypatch, tmp_path, + capsys): + """Sous pytest sans TIRAGE_JOURNAL_PATH : INERT, pas un echec bavard. + + Meme commutateur d'environnement que le cache et la sonde delivered : + un test unitaire n'ecrit pas dans le state dir machine, et stderr + reste propre -- l'inertie n'est pas une panne. + """ + monkeypatch.delenv("TIRAGE_JOURNAL_PATH", raising=False) + rec = pig._log_tirage_safe(lane=LANE, candidates=[1], retained=1, + urn="belt", mode="belt") + assert rec is None + captured = capsys.readouterr() + assert "journal" not in captured.err.lower(), ( + f"inertie bavarde sur stderr : {captured.err!r}" + ) + assert not _journal_path(tmp_path).exists() + + +def test_helper_swallows_journal_failure(monkeypatch, tmp_path, capsys): + """Un defaut d'ecriture ne lève PAS : stderr une ligne, grain rendu. + + Le journal est une couche d'observabilite, jamais un gate -- un + disque plein ne doit pas priver la lane de son tirage. + """ + monkeypatch.setenv("TIRAGE_JOURNAL_PATH", str(_journal_path(tmp_path))) + + def boom(**kwargs): + raise OSError("disque plein") + + monkeypatch.setattr(pig.tirage_journal, "log_tirage", boom) + rec = pig._log_tirage_safe(lane=LANE, candidates=[5], retained=None, + urn="belt", mode="belt") + assert rec is None + assert "NON ECRIT" in capsys.readouterr().err + + +def test_helper_survives_absent_module(monkeypatch, capsys): + """Clone sans le module : le picker tire SANS journal, en le disant. + + L'import best-effort met ``tirage_journal = None`` ; le helper rend + None avec une ligne stderr -- jamais une exception. + """ + monkeypatch.setattr(pig, "tirage_journal", None) + rec = pig._log_tirage_safe(lane=LANE, candidates=[5], retained=5, + urn="belt", mode="belt") + assert rec is None + assert "indisponible" in capsys.readouterr().err + + +# ================================================================== +# Fixtures reseau : neutralisation hermetique (doctrine #19913 -- un +# test n'atteint JAMAIS le reseau par defaut) +# ================================================================== + + +def _make_item(number, created, klass="grain", genre="docs"): + return {"number": number, "klass": klass, "age": 200, "idle": 10, + "genre": genre, "labels": [], "title": f"issue #{number}", + "created_at": created, "last_delivery_stamp": None, + "updated_at": created, "weight": 1.0} + + +def _state_red(number): + """Etat GraphQL d'une PR rouge, forme rendue par fetch_pr_states. + + Forme NESTEE complete (reviews.nodes, commits.nodes -> + statusCheckRollup.contexts.nodes) : une forme aplatie est silencieusement + illisible par ``blocking_causes`` -- le garde rouge ne declenche pas et + le test sort sur le reseau reel (mesure pendant l'ecriture de ce test). + """ + return { + "number": number, "mergeable": "MERGEABLE", + "reviews": {"nodes": []}, + "commits": {"nodes": [{"commit": {"statusCheckRollup": {"contexts": {"nodes": [ + {"name": "PR gate", "conclusion": "FAILURE", "isRequired": True, + "completedAt": "2026-10-08T00:00:00Z"} + ]}}}}]}, + } + + +def _patch_common_network(monkeypatch): + """Le socle commun a tous les chemins de main() hors reseau.""" + monkeypatch.setattr(pig.gh_identity, "pin_gh_token", + lambda *a, **k: None) + monkeypatch.setattr(pig, "fetch_open_prs", lambda: []) + monkeypatch.setattr(pig, "fetch_pr_states", lambda nums: {}) + monkeypatch.setattr(pig, "unaddressed_review_points", lambda nums: {}) + monkeypatch.setattr(pig, "fetch_lane_record_prs", lambda **k: ([], None)) + monkeypatch.setattr(pig, "fetch_main_head_probe", lambda *a, **k: None) + monkeypatch.setattr(pig, "fetch_visits", lambda *a, **k: ({}, None)) + monkeypatch.setattr(pig, "fetch_series_visits", + lambda **k: ({}, {}, None)) + monkeypatch.setattr(pig, "fetch_merged", lambda *a, **k: ([], None)) + # Doctrine #19913 : main() injecte les sondes REELLES au tapis + # (delivered_probe=has_delivered_signal) -- on les remplace par les + # inertes pour que la boucle ne sorte jamais sur un vrai gh. + monkeypatch.setattr(pig, "has_delivered_signal", + pig.delivered_probe_inert) + monkeypatch.setattr(pig, "merged_pr_signal", pig.merged_pr_probe_inert) + + +# ================================================================== +# Chemin REPARATION : le plus frequent pour une lane chargee -- sans +# cette entree, le journal sur-estimerait la part du pool proposee. +# ================================================================== + + +def test_repair_path_journals_the_assignment(monkeypatch, tmp_path): + jp = _journal_path(tmp_path) + monkeypatch.setenv("TIRAGE_JOURNAL_PATH", str(jp)) + _patch_common_network(monkeypatch) + red = _state_red(0) + created = (pig.NOW + - pig.dt.timedelta(hours=30)).strftime("%Y-%m-%dT%H:%M:%SZ") + prs = [{"number": 401, "title": "pr 401", + "body": f"Grain: MED/guard -- lane {LANE}", + "createdAt": created, "isDraft": False}] + monkeypatch.setattr(pig, "fetch_open_prs", lambda: prs) + monkeypatch.setattr(pig, "fetch_pr_states", + lambda nums: {n: red for n in nums}) + + rc = pig.main(["--lane", LANE]) + + assert rc == 0 + rows = tirage_journal.read_journal(jp) + assert len(rows) == 1, "une invocation = une entree" + assert rows[0].mode == "repair" + assert rows[0].urn == "repair" + assert rows[0].candidates == [401] + assert rows[0].retained == 401, ( + "le grain journalise est la PR a reparer, pas un numero de pool" + ) + + +def test_repair_json_path_also_journals(monkeypatch, tmp_path, capsys): + """Le consommateur --json a la meme trace que la sortie texte.""" + jp = _journal_path(tmp_path) + monkeypatch.setenv("TIRAGE_JOURNAL_PATH", str(jp)) + _patch_common_network(monkeypatch) + red = _state_red(0) + created = (pig.NOW + - pig.dt.timedelta(hours=30)).strftime("%Y-%m-%dT%H:%M:%SZ") + prs = [{"number": 402, "title": "pr 402", + "body": f"Grain: MED/guard -- lane {LANE}", + "createdAt": created, "isDraft": False}] + monkeypatch.setattr(pig, "fetch_open_prs", lambda: prs) + monkeypatch.setattr(pig, "fetch_pr_states", + lambda nums: {n: red for n in nums}) + + rc = pig.main(["--lane", LANE, "--json"]) + + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["mode"] == "repair" + rows = tirage_journal.read_journal(jp) + assert len(rows) == 1 + assert rows[0].mode == "repair" + assert rows[0].retained == 402 + + +# ================================================================== +# Chemin TAPIS (--belt) : candidats servis + tete retenue +# ================================================================== + + +def test_belt_path_journals_head_of_queue(monkeypatch, tmp_path, capsys): + jp = _journal_path(tmp_path) + monkeypatch.setenv("TIRAGE_JOURNAL_PATH", str(jp)) + _patch_common_network(monkeypatch) + pool = [ + _make_item(7001, "2025-01-01T00:00:00Z"), + _make_item(7002, "2025-02-01T00:00:00Z"), + _make_item(7003, "2025-03-01T00:00:00Z"), + ] + monkeypatch.setattr(pig, "fetch_pool", lambda **k: (pool, None)) + monkeypatch.setattr(pig, "check_claims", + lambda nums, lane: {n: (pig.CLAIM_CODE_FREE, "libre") + for n in nums}) + monkeypatch.setattr(pig, "latest_claim_stamp", lambda n: None) + + rc = pig.main(["--lane", LANE, "--belt", "--json"]) + + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["mode"] == "belt" + picks = [p["number"] for p in payload["picks"]] + rows = tirage_journal.read_journal(jp) + assert len(rows) == 1, "une invocation = une entree, garde rouge ou non" + assert rows[0].mode == "belt" + assert rows[0].urn == "belt" + assert rows[0].candidates == picks, ( + "les candidats journalises sont les picks servis, pas la file entiere" + ) + assert rows[0].retained == picks[0] + assert rows[0].retained == 7001, ( + "jamais servie et la plus ancienne : la tete du tapis doit etre 7001" + ) + + +def test_belt_with_red_guard_journals_once(monkeypatch, tmp_path): + """Garde rouge + tapis : UNE entree belt, pas une repair + une belt. + + Une invocation = une entree : le rappel rouge est fusionne dans la + sortie du tapis, sa consigne l'est aussi. + """ + jp = _journal_path(tmp_path) + monkeypatch.setenv("TIRAGE_JOURNAL_PATH", str(jp)) + _patch_common_network(monkeypatch) + red = _state_red(0) + created = (pig.NOW + - pig.dt.timedelta(hours=30)).strftime("%Y-%m-%dT%H:%M:%SZ") + prs = [{"number": 403, "title": "pr 403", + "body": f"Grain: MED/guard -- lane {LANE}", + "createdAt": created, "isDraft": False}] + monkeypatch.setattr(pig, "fetch_open_prs", lambda: prs) + monkeypatch.setattr(pig, "fetch_pr_states", + lambda nums: {n: red for n in nums}) + pool = [_make_item(7011, "2025-01-01T00:00:00Z")] + monkeypatch.setattr(pig, "fetch_pool", lambda **k: (pool, None)) + monkeypatch.setattr(pig, "check_claims", + lambda nums, lane: {n: (pig.CLAIM_CODE_FREE, "libre") + for n in nums}) + monkeypatch.setattr(pig, "latest_claim_stamp", lambda n: None) + + rc = pig.main(["--lane", LANE, "--belt", "--json"]) + + assert rc == 0 + rows = tirage_journal.read_journal(jp) + assert len(rows) == 1, ( + f"une invocation belt avec garde rouge doit produire UNE entree, " + f"got {[r.mode for r in rows]}" + ) + assert rows[0].mode == "belt" + + +# ================================================================== +# Chemin PONDERE : la poignee servie et sa tete +# ================================================================== + + +def test_weighted_path_journals_picks(monkeypatch, tmp_path, capsys): + jp = _journal_path(tmp_path) + monkeypatch.setenv("TIRAGE_JOURNAL_PATH", str(jp)) + _patch_common_network(monkeypatch) + pool = [_make_item(7101, "2025-01-01T00:00:00Z"), + _make_item(7102, "2025-02-01T00:00:00Z")] + monkeypatch.setattr(pig, "fetch_pool", lambda **k: (pool, None)) + # La machinerie de tirage elle-meme est couverte par les tests + # existants ; ici on borne la poignee rendue pour verifier la + # CONSIGNE, pas la ponderation. + drawn = [ + {"number": 7101, "klass": "grain", "age": 200, "idle": 10, + "genre": "docs", "weight": 1.0, "title": "issue #7101", + "visits": 0}, + {"number": 7102, "klass": "grain", "age": 190, "idle": 10, + "genre": "docs", "weight": 1.0, "title": "issue #7102", + "visits": 0}, + ] + monkeypatch.setattr(pig, "fetch_merged_grains", lambda: ([], None)) + monkeypatch.setattr(pig, "draw_unclaimed", + lambda *a, **k: (drawn, {}, [])) + monkeypatch.setattr(pig, "recent_delivery", lambda picks: {}) + + rc = pig.main(["--lane", LANE, "--json"]) + + assert rc == 0 + rows = tirage_journal.read_journal(jp) + assert len(rows) == 1 + assert rows[0].mode == "weighted" + assert rows[0].candidates == [7101, 7102] + assert rows[0].retained == 7101 + assert rows[0].urn == "grain", ( + "l'urne journalisee est celle du grain retenu" + ) + # La sortie machine reste du JSON pur : la confirmation journal va sur + # stderr, jamais sur stdout. + captured = capsys.readouterr() + json.loads(captured.out) + assert "journal de tirage" in captured.err