diff --git a/scripts/check_pr_repoll.py b/scripts/check_pr_repoll.py new file mode 100644 index 0000000000..e71085cde6 --- /dev/null +++ b/scripts/check_pr_repoll.py @@ -0,0 +1,275 @@ +#!/usr/bin/env python3 +"""Organe de garde anti re-sondage de PR (#16144, item 2). + +Pourquoi cet organe existe +-------------------------- +Le sondage repete de la meme PR par une meme session (relectures `gh pr view` +en boucle pendant une attente) est la consommation la plus silencieuse du +quota serveur : mesure ai-01 du 13/09, 9,2 refs/PR cote client, en salves de +3-9 relectures pendant les attentes (corrobore cote serveur : #15881 x30 refs +dans la journee). Le client CC >= 2.1.269 a retire les bornes generiques +(rumination, perimetre) -- l'organe les reintroduit au moment du geste, sur +le pattern de `check_unaddressed_nits.py` : un refus AVANT la lecture, pas +un audit apres. + +Bornes (plan du 14/09, formes corrigees 10:53Z) : + - B2' : refus d'une relecture manuelle si `last_read` date de moins de + `--min-age` minutes ET qu'aucun signal de changement d'etat + n'est survenu depuis (Monitor arme, `--watch`, notification). + - B2-bis : toute lecture manuelle est refusee tant qu'une surveillance + (watch) est vivante -- le watcher notifie de lui-meme, la + lecture manuelle est redondante par construction. + +Usage +----- + python scripts/check_pr_repoll.py check 20992 # pre-lecture (gate) + python scripts/check_pr_repoll.py check 20992 --min-age 20 # seuil B2' explicite + python scripts/check_pr_repoll.py record 20992 # post-lecture + python scripts/check_pr_repoll.py record 20992 --watch # arme la surveillance + python scripts/check_pr_repoll.py record 20992 --watch-end # clot la surveillance + python scripts/check_pr_repoll.py record 20992 --signal monitor # trace un signal + +Registre : JSON `{: {"last_read": ..., "watch_armed_at": ..., +"watch_ended_at": ..., "signals": [{"at": ..., "kind": ...}]}}`, tenu par +l'organe -- le skill appelant n'ecrit jamais le JSON a la main. Emplacement : +`--ledger`, sinon `$PR_POLL_LEDGER`, sinon un emplacement stable hors tempdir (`%LOCALAPPDATA%\\CoursIA` sur Windows, `~/.local/share/CoursIA` sinon) — le tempdir est purgé par les règles de scratch de la flotte et vidait silencieusement le registre (review #21060). + +Fail-open assume : registre absent ou corrompu -> `check` sort 0 + warning +stderr. L'organe garde des tokens ; il ne doit jamais bloquer la +coordination sur un sidecar casse. Un `record` sur registre corrompu +preserve la donnee (renommee `.corrupt-`) avant de repartir +vieux : jamais de destruction sans preuve de preservation. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +DEFAULT_MIN_AGE_MIN = 30 + +# --------------------------------------------------------------------------- +# Horodatage : UTC explicite, jamais d'heure locale habillee d'un Z +# (lecon utc-vs-local : le suffixe Z sur une heure locale est proscrit). +# --------------------------------------------------------------------------- + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _iso(dt: datetime) -> str: + return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _parse_iso(ts: str) -> datetime: + return datetime.fromisoformat(ts.replace("Z", "+00:00")) + + +# --------------------------------------------------------------------------- +# Registre : chargement fail-open, ecriture atomique. +# --------------------------------------------------------------------------- + + +def default_ledger_path() -> Path: + env = os.environ.get("PR_POLL_LEDGER") + if env: + return Path(env) + # Defaut stable hors tempdir (review #21060) : le tempdir est + # purge par les regles de scratch de la flotte ; un registre qui + # disparait avec lui rend le garde inoperant sans le dire. + # Precedent d'emplacement : journal de la tache planifiee prune + # (%LOCALAPPDATA%\CoursIA\prune_task\logs\). + local = os.environ.get("LOCALAPPDATA") + root = Path(local) if local else Path.home() / ".local" / "share" + return root / "CoursIA" / "pr-poll-ledger.json" + + +def load_ledger(path: Path) -> tuple[dict[str, Any] | None, str | None]: + """Charge le registre ; (None, warning) si absent ou corrompu (fail-open).""" + if not path.exists(): + return None, None + try: + raw = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + return None, f"registre illisible ({path}) : {exc} -- fail-open" + if not isinstance(raw, dict): + return None, f"registre non-object ({path}) -- fail-open" + return raw, None + + +def save_ledger(path: Path, ledger: dict[str, Any]) -> str | None: + """Ecrit le registre ; sur corrompu existant, le preserve avant remplacement.""" + warning: str | None = None + if path.exists(): + try: + json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + backup = path.with_name(f"{path.name}.corrupt-{_iso(_now())}") + # Remplacement : caractere ':' interdit dans les noms Windows, sanatise. + backup = path.with_name(backup.name.replace(":", "")) + try: + path.replace(backup) + warning = f"registre corrompu preserve en {backup.name}, repart a vide" + except OSError as exc: + return f"impossible de preserver le registre corrompu ({exc}) -- rien ecrit" + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(f"{path.name}.tmp") + try: + tmp.write_text( + json.dumps(ledger, indent=2, ensure_ascii=False, sort_keys=True) + "\n", + encoding="utf-8", + ) + tmp.replace(path) + except OSError as exc: + return f"ecriture registre echouee ({path}) : {exc}" + return warning + + +# --------------------------------------------------------------------------- +# Coeur pur : la decision B2'/B2-bis, testable sans I/O. +# --------------------------------------------------------------------------- + + +@dataclass +class Decision: + """Verdict du gate `check` : refused=True -> exit 1 au moment du geste.""" + + refused: bool + reason: str = "" + + +def evaluate(entry: dict[str, Any] | None, now: datetime, min_age_min: int) -> Decision: + """Applique B2-bis puis B2' sur une entree de registre. + + B2-bis en premier : une surveillance vivante refuse toute lecture + manuelle, meme si `last_read` est ancien -- le watcher notifie de + lui-meme (le refus ne depend pas de l'age). + """ + if entry is None: + return Decision(False, "aucune lecture enregistree -- premiere lecture libre") + if entry.get("watch_armed_at") and not entry.get("watch_ended_at"): + return Decision( + True, + f"surveillance vivante depuis {entry['watch_armed_at']} (B2-bis) : " + "le watcher notifie de lui-meme, lecture manuelle refusee", + ) + last_read = entry.get("last_read") + if not last_read: + return Decision(False, "pas de last_read -- lecture libre") + read_at = _parse_iso(last_read) + age_min = (now - read_at).total_seconds() / 60.0 + if age_min >= min_age_min: + return Decision(False, f"derniere lecture il y a {age_min:.0f} min (>= {min_age_min} min)") + signals = entry.get("signals") or [] + fresh = [s for s in signals if _parse_iso(s["at"]) >= read_at] + if fresh: + kinds = ", ".join(sorted({s["kind"] for s in fresh})) + return Decision(False, f"signal(s) depuis la lecture ({kinds}) -- changement d'etat") + return Decision( + True, + f"derniere lecture il y a {age_min:.0f} min (< {min_age_min} min) sans signal " + "de changement d'etat (B2') : relecture refusee ; pour forcer, " + "`record --signal ` ou armer un watch", + ) + + +# --------------------------------------------------------------------------- +# CLI : record (post-lecture) / check (pre-lecture). +# --------------------------------------------------------------------------- + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Gate anti re-sondage de PR (#16144 item 2) : refuse la relecture " + "d'une PR lue trop recemment sans changement d'etat.", + ) + parser.add_argument( + "--ledger", + type=Path, + default=None, + help="chemin du registre (defaut : $PR_POLL_LEDGER, sinon un emplacement " + "stable hors tempdir : %%LOCALAPPDATA%%\\CoursIA sur Windows, " + "~/.local/share/CoursIA sinon)", + ) + sub = parser.add_subparsers(dest="verb", required=True) + + p_rec = sub.add_parser("record", help="post-lecture : stamp last_read (+ effets)") + p_rec.add_argument("pr", help="numero de PR (cle du registre)") + flags = p_rec.add_mutually_exclusive_group() + flags.add_argument( + "--watch", + action="store_true", + help="arme la surveillance (watch_armed_at, efface watch_ended_at)", + ) + flags.add_argument( + "--watch-end", + action="store_true", + help="clot la surveillance (watch_ended_at)", + ) + flags.add_argument( + "--signal", + metavar="KIND", + help="trace un signal de changement d'etat (monitor, notification, ...)", + ) + + p_chk = sub.add_parser("check", help="pre-lecture : exit 1 si relecture refusee") + p_chk.add_argument("pr", help="numero de PR (cle du registre)") + p_chk.add_argument( + "--min-age", + type=int, + default=DEFAULT_MIN_AGE_MIN, + metavar="N", + help=f"age minimal en minutes avant relecture libre (defaut {DEFAULT_MIN_AGE_MIN})", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + path = args.ledger if args.ledger is not None else default_ledger_path() + now = _now() + + ledger, warning = load_ledger(path) + if warning: + print(f"WARNING: {warning}", file=sys.stderr) + + if args.verb == "check": + entry = (ledger or {}).get(args.pr) + decision = evaluate(entry, now, args.min_age) + print(f"[pr-repoll] PR {args.pr} : {'REFUS -- ' if decision.refused else 'OK -- '}" + f"{decision.reason}") + return 1 if decision.refused else 0 + + # record : registre absent -> repart vide ; corrompu -> preserve par save_ledger + ledger = ledger or {} + entry = ledger.setdefault(args.pr, {"signals": []}) + entry["signals"] = entry.get("signals") or [] + entry["last_read"] = _iso(now) + if args.watch: + entry["watch_armed_at"] = _iso(now) + entry["watch_ended_at"] = None + entry["signals"].append({"at": _iso(now), "kind": "watch"}) + if args.watch_end: + entry["watch_ended_at"] = _iso(now) + entry["signals"].append({"at": _iso(now), "kind": "watch-end"}) + if args.signal: + entry["signals"].append({"at": _iso(now), "kind": args.signal}) + warn_save = save_ledger(path, ledger) + if warn_save: + print(f"WARNING: {warn_save}", file=sys.stderr) + print(f"[pr-repoll] PR {args.pr} : last_read={entry['last_read']}" + + (" +watch" if args.watch else "") + + (" +watch-end" if args.watch_end else "") + + (f" +signal:{args.signal}" if args.signal else "")) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/tests/test_check_pr_repoll.py b/scripts/tests/test_check_pr_repoll.py new file mode 100644 index 0000000000..18dd4985a3 --- /dev/null +++ b/scripts/tests/test_check_pr_repoll.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python3 +"""Tests de l'organe check_pr_repoll.py (#16144, item 2 -- contrat c.6205821308). + +Les 7 cas contractuels : + 1. pas de registre -> check exit 0 + 2. re-sondage < N sans signal -> check exit 1 + 3. < N avec signal -> check exit 0 + 4. watch vivant -> check exit 1 (meme si last_read ancien) + 5. watch termine -> check exit 0 + 6. last_read > N -> check exit 0 + 7. registre corrompu -> check exit 0 + warning, puis record preserve + le fichier corrompu (backup) et repart propre. +""" + +import json +import os +import sys +import tempfile +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from check_pr_repoll import ( # noqa: E402 + _iso, + build_parser, + default_ledger_path, + evaluate, + main, +) + + +def _ts(minutes_ago: float) -> str: + return _iso(datetime.now(timezone.utc) - timedelta(minutes=minutes_ago)) + + +@pytest.fixture +def ledger(tmp_path: Path) -> Path: + return tmp_path / "pr-poll-ledger.json" + + +def _write(ledger: Path, pr: str, **entry) -> None: + data = json.loads(ledger.read_text(encoding="utf-8")) if ledger.exists() else {} + data[pr] = entry + ledger.write_text(json.dumps(data), encoding="utf-8") + + +def test_1_no_registry_check_passes(ledger: Path, capsys: pytest.CaptureFixture) -> None: + assert main(["--ledger", str(ledger), "check", "1234"]) == 0 + assert "REFUS" not in capsys.readouterr().out + + +def test_2_fresh_read_without_signal_refused(ledger: Path, capsys: pytest.CaptureFixture) -> None: + _write(ledger, "1234", last_read=_ts(5), signals=[]) + assert main(["--ledger", str(ledger), "check", "1234"]) == 1 + assert "B2'" in capsys.readouterr().out + + +def test_3_fresh_read_with_signal_allowed(ledger: Path, capsys: pytest.CaptureFixture) -> None: + _write(ledger, "1234", last_read=_ts(5), + signals=[{"at": _ts(2), "kind": "notification"}]) + assert main(["--ledger", str(ledger), "check", "1234"]) == 0 + assert "changement d'etat" in capsys.readouterr().out + + +def test_4_live_watch_refuses_even_old_read(ledger: Path, capsys: pytest.CaptureFixture) -> None: + _write(ledger, "1234", last_read=_ts(600), watch_armed_at=_ts(590), + watch_ended_at=None, signals=[]) + assert main(["--ledger", str(ledger), "check", "1234"]) == 1 + assert "B2-bis" in capsys.readouterr().out + + +def test_5_ended_watch_allows(ledger: Path, capsys: pytest.CaptureFixture) -> None: + _write(ledger, "1234", last_read=_ts(600), watch_armed_at=_ts(590), + watch_ended_at=_ts(580), signals=[]) + assert main(["--ledger", str(ledger), "check", "1234"]) == 0 + + +def test_6_old_read_allowed(ledger: Path, capsys: pytest.CaptureFixture) -> None: + _write(ledger, "1234", last_read=_ts(90), signals=[]) + assert main(["--ledger", str(ledger), "check", "1234"]) == 0 + + +def test_7_corrupt_registry_fail_open_then_record_preserves( + ledger: Path, capsys: pytest.CaptureFixture) -> None: + ledger.write_text("{corrompu", encoding="utf-8") + # check : fail-open -- exit 0 + warning stderr, jamais de blocage. + assert main(["--ledger", str(ledger), "check", "1234"]) == 0 + err = capsys.readouterr().err + assert "fail-open" in err + # record : le corrompu est preserve (backup .corrupt-*), pas detruit. + assert main(["--ledger", str(ledger), "record", "1234"]) == 0 + backups = list(ledger.parent.glob(f"{ledger.name}.corrupt-*")) + assert len(backups) == 1, "le registre corrompu doit etre preserve avant remplacement" + assert backups[0].read_text(encoding="utf-8") == "{corrompu" + # le nouveau registre est propre et consultable. + data = json.loads(ledger.read_text(encoding="utf-8")) + assert "last_read" in data["1234"] + assert main(["--ledger", str(ledger), "check", "1234"]) == 1 # < N sans signal depuis + + +def test_evaluate_watch_end_signal_beats_fresh_read() -> None: + """Borne hors contrat strict : la fin de watch est un signal -- sous + l'age minimal (60 min < 120), la relecture reste LIBRE parce que le + watcher vient de rendre la main en notifiant (signal >= last_read). + Sans la regle du signal, ce cas serait refuse : c'est le cas + discriminant de B2'.""" + now = datetime.now(timezone.utc) + entry = { + "last_read": _iso(now - timedelta(minutes=60)), + "watch_armed_at": _iso(now - timedelta(minutes=120)), + "watch_ended_at": _iso(now - timedelta(minutes=2)), + "signals": [{"at": _iso(now - timedelta(minutes=2)), "kind": "watch-end"}], + } + assert evaluate(entry, now, 120).refused is False + + +def test_9_default_ledger_is_stable_not_tempdir() -> None: + """Review #21060 : le defaut ne doit plus vivre sous le tempdir -- + purge par les regles de scratch de la flotte, il vidait le registre + et le garde perdait son effet sans le dire.""" + path = default_ledger_path() + assert path.name == "pr-poll-ledger.json" + tmp = Path(tempfile.gettempdir()).resolve() + try: + path.resolve().relative_to(tmp) + under = True + except ValueError: + under = False + assert not under, "le defaut ne doit pas vivre sous le tempdir" + assert "coursia" in [p.lower() for p in path.parts] + + +def test_10_build_parser_format_help_does_not_crash() -> None: + """Review coordinateur #21060 : argparse.help= doit etre exempt de % non + echappes, sinon `format_help()` leve ValueError (3.11) ou la construction + du parser leve en 3.14+. Regression : `%%LOCALAPPDATA%%` ecrit en clair + dans le help= de `--ledger`.""" + help_text = build_parser().format_help() + # verifie que le path documente bien la cible (avec %% reduit a % par argparse) + assert "LOCALAPPDATA" in help_text, help_text + # et qu'aucun format involontaire ne reste (le %% est reduit a % par argparse) + assert "%L" not in help_text.replace("%%", "").replace("LOCALAPPDATA", ""), ( + "le help= doit etre ecrit sans % non echappe susceptible d'etre " + "interprete par argparse comme un format" + )