diff --git a/.claude/rules/git-workflow.md b/.claude/rules/git-workflow.md index c97ea7109c..a1d9424f10 100644 --- a/.claude/rules/git-workflow.md +++ b/.claude/rules/git-workflow.md @@ -86,6 +86,13 @@ python scripts/ci/prune_merged_worktrees.py --apply # applique python scripts/ci/prune_merged_worktrees.py --json # sortie structuree pour sweep dashboard ``` +**Cablage quotidien (#14473)** : `scripts/ci/install_prune_task.py` installe la tache planifiee locale (schtasks DAILY 03:17, journal `%LOCALAPPDATA%\CoursIA\prune_task\logs\`) qui execute la purge en `--apply` — l'appel manuel fin de cycle reste le filet. La prose d'une regle ne s'execute pas seule (regles injectees au demarrage, perdues en crash) : une fois par machine, + +```bash +python scripts/ci/install_prune_task.py --install # garde : REFUSE tant que le fix #14476 n'est pas merge +python scripts/ci/install_prune_task.py --status # etat de la tache +``` + **Critères de retrait (cf issue #14195 acceptance)** : - **REMOVE** : PR MERGED ou CLOSED, branche sans unpushed, pas d'édition source untracked. - **REFUSE** : branche `main`/`master` (jamais le worktree de travail) · PR OPEN (l'itération continue) · commits non poussés vs upstream spécifique (`@{u}` non-`main`) · édition source untracked non tolérée (`.py`, `.ipynb`, `.lean`, `.md`, etc.). diff --git a/scripts/ci/install_prune_task.py b/scripts/ci/install_prune_task.py new file mode 100644 index 0000000000..453245f3f1 --- /dev/null +++ b/scripts/ci/install_prune_task.py @@ -0,0 +1,187 @@ +#!/usr/bin/env python3 +r"""Installateur de la tache planifiee quotidienne prune_merged_worktrees (#14473). + +Le organe (scripts/ci/prune_merged_worktrees.py, #14195) etait livre mais +rappelé par personne : une prescription en prose dans .claude/rules/ ne +s'execute pas (regles injectees au DEMARRAGE de session, fin de cycle +disparait en crash/compaction). Le cablage est necessairement LOCAL : un +workflow GitHub Actions ne voit que son checkout ephemerre, jamais les +worktrees de l'hote (missing-tool-turns-a-guard-green). + +Modes : + --install [--repo PATH] [--time HH:MM] cree la tache planifiee (idempotent) + --status etat de la tache + --uninstall supprime la tache + --run execute la purge (invoqué PAR la tache) : + journal horodate, jamais de couleur/TTY + +Garde de securite (#14476) : --install REFUSE de cabler --apply si le script +cible ne contient pas encore la resolution par numero (def _normalize_subject, +PR #14481) -- installer le cron avec l'ancien predicat d'intersection de +jetons deploierait l'attribution fausse (et destructive) TOUS LES JOURS. + +Journal : %LOCALAPPDATA%\CoursIA\prune_task\logs\prune_YYYYMMDD.log +""" +from __future__ import annotations + +import argparse +import datetime as _dt +import os +import subprocess +import sys +from pathlib import Path + +TASK_NAME = r"CoursIA\prune_merged_worktrees" +THIS_FILE = Path(__file__).resolve() +LOG_DIR = Path(os.environ.get("LOCALAPPDATA", str(Path.home() / "AppData" / "Local"))) / "CoursIA" / "prune_task" / "logs" + +# Marqueur du fix #14476/#14481 dans le script cible -- voir garde ci-dessus. +REQUIRED_FIX_MARKER = "def _normalize_subject" + + +def _run(cmd: list[str], **kw) -> subprocess.CompletedProcess: + return subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", + errors="replace", **kw) + + +def prune_script_path(repo: Path) -> Path: + return repo / "scripts" / "ci" / "prune_merged_worktrees.py" + + +def check_prune_fix_present(repo: Path) -> tuple[bool, str]: + """Le predicat detached-head doit etre la version corrigee (#14476). + + L'ancienne heuristique (intersection de jetons) attribuait n'importe + quelle PR recente partageant un mot du domaine -> retraits faux. Une + tache quotidienne --apply ne doit JAMAIS la deployer. + """ + target = prune_script_path(repo) + if not target.is_file(): + return False, f"introuvable : {target}" + text = target.read_text(encoding="utf-8", errors="replace") + if REQUIRED_FIX_MARKER not in text: + return False, ( + f"{target.name} ne contient pas le fix #14476 " + f"('{REQUIRED_FIX_MARKER}'). Installer le cron --apply avec " + "l'heuristique d'intersection de jetons deploierait " + "l'attribution fausse quotidiennement (cf #14481). " + "Rebase/merge d'abord, puis relancer --install." + ) + return True, "fix #14476 present" + + +def task_command(repo: Path) -> list[str]: + """Commande enregistree dans le planificateur : ce script --run, qui + journalise et appelle l'organe en --apply.""" + return [sys.executable, str(THIS_FILE), "--run", "--repo", str(repo)] + + +def build_schtasks_install(cmd: list[str], time: str) -> list[str]: + """Ligne schtasks /Create : quotidienne, contexte utilisateur courant + (gh auth vit au niveau utilisateur), fenêtre masquee.""" + tr = " ".join(cmd) + return [ + "schtasks", "/Create", "/F", + "/TN", TASK_NAME, + "/SC", "DAILY", + "/ST", time, + "/TR", f'"{tr}"', + ] + + +def log_path_for(day: _dt.date | None = None) -> Path: + day = day or _dt.date.today() + return LOG_DIR / f"prune_{day:%Y%m%d}.log" + + +def task_exists() -> bool: + return _run(["schtasks", "/Query", "/TN", TASK_NAME]).returncode == 0 + + +def cmd_install(repo: Path, time: str) -> int: + ok, msg = check_prune_fix_present(repo) + if not ok: + print(f"REFUSE : {msg}", file=sys.stderr) + return 2 + print(f"garde OK : {msg}") + LOG_DIR.mkdir(parents=True, exist_ok=True) + proc = _run(build_schtasks_install(task_command(repo), time)) + if proc.returncode != 0: + print(f"schtasks /Create echoue (rc={proc.returncode}) : " + f"{proc.stdout.strip()} {proc.stderr.strip()}", file=sys.stderr) + return 2 + print(f"tache installee : {TASK_NAME} quotidienne a {time}") + print(f"commande : {' '.join(task_command(repo))}") + print(f"journal : {log_path_for()}") + print("verification : schtasks /Query /TN " + + TASK_NAME.replace("\\", "\\") + " /V /FO LIST") + return 0 + + +def cmd_status() -> int: + if not task_exists(): + print(f"tache ABSENTE : {TASK_NAME}") + return 1 + proc = _run(["schtasks", "/Query", "/TN", TASK_NAME, "/V", "/FO", "LIST"]) + print(proc.stdout) + return 0 + + +def cmd_uninstall() -> int: + proc = _run(["schtasks", "/Delete", "/TN", TASK_NAME, "/F"]) + if proc.returncode != 0: + print(f"suppression echouee : {proc.stdout.strip()} " + f"{proc.stderr.strip()}", file=sys.stderr) + return 2 + print(f"tache supprimee : {TASK_NAME}") + return 0 + + +def cmd_run(repo: Path) -> int: + """Invoque par la tache planifiee : journal horodate, pas de TTY.""" + LOG_DIR.mkdir(parents=True, exist_ok=True) + log = log_path_for() + stamp = _dt.datetime.now().strftime("%Y-%m-%dT%H:%M:%S") + with log.open("a", encoding="utf-8") as fh: + fh.write(f"\n=== {stamp} run start ===\n") + fh.flush() + proc = subprocess.run( + [sys.executable, str(prune_script_path(repo)), + "--path", str(repo), "--apply"], + stdout=fh, stderr=subprocess.STDOUT, + ) + fh.write(f"=== {_dt.datetime.now().strftime('%Y-%m-%dT%H:%M:%S')} " + f"run end rc={proc.returncode} ===\n") + # exit code non zero si l'organe a echoue -- visible dans le journal + return proc.returncode + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + p.add_argument("--install", action="store_true") + p.add_argument("--status", action="store_true") + p.add_argument("--uninstall", action="store_true") + p.add_argument("--run", action="store_true", + help="mode interne (invoque par la tache planifiee)") + p.add_argument("--repo", type=Path, + default=Path(r"C:\dev\CoursIA"), + help="checkout principal du depot (defaut C:\\dev\\CoursIA)") + p.add_argument("--time", default="03:17", + help="heure quotidienne HH:MM (defaut 03:17, hors heures ouvrables)") + args = p.parse_args(argv) + + repo = args.repo.resolve() + if args.install: + return cmd_install(repo, args.time) + if args.status: + return cmd_status() + if args.uninstall: + return cmd_uninstall() + if args.run: + return cmd_run(repo) + p.print_help() + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/tests/test_install_prune_task.py b/scripts/tests/test_install_prune_task.py new file mode 100644 index 0000000000..54b332fc7e --- /dev/null +++ b/scripts/tests/test_install_prune_task.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python3 +r"""Tests de l'installateur de tache planifiee prune (#14473). + +Pinent : +1. la garde de securite : --install REFUSE si le script cible ne contient + pas le fix #14476 (_normalize_subject) -- un cron --apply quotidien ne + doit JAMAIS deployer l'attribution fausse par intersection de jetons ; +2. l'idempotence : schtasks /Create /F (relançable sans doublon) ; +3. la commande de tache appelle bien --run (le mode journalisant) avec le + --repo explicit, et l'organe en --apply vient de cmd_run seulement ; +4. le journal est horodate a un chemin nomme (LOCALAPPDATA). +""" +from __future__ import annotations + +import subprocess +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] / "ci")) + +import install_prune_task as ipt # noqa: E402 + + +# --------------------------------------------------------------------------- +# garde de securite #14476 +# --------------------------------------------------------------------------- + +class TestPruneFixGuard: + def test_refuse_quand_le_fix_est_absent(self, tmp_path): + repo = tmp_path / "CoursIA" + (repo / "scripts" / "ci").mkdir(parents=True) + # version SANS le fix : l'heuristique d'intersection d'origine + (repo / "scripts" / "ci" / "prune_merged_worktrees.py").write_text( + "def lookup_pr_for_detached_head():\n" + " title_tokens & subj_tokens # intersection de jetons\n", + encoding="utf-8") + ok, msg = ipt.check_prune_fix_present(repo) + assert not ok + assert "#14476" in msg or "#14481" in msg + + def test_accepte_quand_le_fix_est_present(self, tmp_path): + repo = tmp_path / "CoursIA" + (repo / "scripts" / "ci").mkdir(parents=True) + (repo / "scripts" / "ci" / "prune_merged_worktrees.py").write_text( + "def _normalize_subject(s):\n return s.lower()\n", + encoding="utf-8") + ok, _ = ipt.check_prune_fix_present(repo) + assert ok + + def test_refuse_si_le_script_est_absent(self, tmp_path): + ok, msg = ipt.check_prune_fix_present(tmp_path) + assert not ok and "introuvable" in msg + + def test_cmd_install_refuse_sans_fix(self, tmp_path, monkeypatch, capsys): + """Le mode --install entier doit exit 2 AVANT tout appel schtasks.""" + repo = tmp_path / "CoursIA" + (repo / "scripts" / "ci").mkdir(parents=True) + (repo / "scripts" / "ci" / "prune_merged_worktrees.py").write_text( + "# vieille version", encoding="utf-8") + called = [] + monkeypatch.setattr(ipt, "_run", lambda cmd, **kw: called.append(cmd) + or subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")) + rc = ipt.cmd_install(repo, "03:17") + assert rc == 2 + assert called == [] # aucun effet de bord planificateur + assert "REFUSE" in capsys.readouterr().err + + +# --------------------------------------------------------------------------- +# construction de la tache / idempotence +# --------------------------------------------------------------------------- + +class TestTaskConstruction: + def test_commande_de_tache_est_le_mode_run_avec_repo_explicite(self, tmp_path): + cmd = ipt.task_command(tmp_path) + assert "--run" in cmd and "--repo" in cmd + # le python interpreteur est absolu (contexte planificateur sans PATH venv) + assert Path(cmd[0]).is_absolute() + # JAMAIS --apply dans la ligne de tache : il vient de cmd_run + assert "--apply" not in cmd + + def test_schtasks_create_est_force_donc_idempotent(self, tmp_path): + line = ipt.build_schtasks_install(ipt.task_command(tmp_path), "03:17") + assert line[0] == "schtasks" and "/Create" in line and "/F" in line + assert "/SC" in line and "DAILY" in line + assert "/TN" in line and ipt.TASK_NAME in line + # quotidienne a l'heure demandee + i = line.index("/ST") + assert line[i + 1] == "03:17" + + def test_nom_de_tache_namespaced(self): + assert "CoursIA" in ipt.TASK_NAME + + +# --------------------------------------------------------------------------- +# journal +# --------------------------------------------------------------------------- + +class TestJournal: + def test_chemin_nomme_et_hordate(self, tmp_path, monkeypatch): + monkeypatch.setattr(ipt, "LOG_DIR", tmp_path) + import datetime as dt + p = ipt.log_path_for(dt.date(2026, 9, 3)) + assert p == tmp_path / "prune_20260903.log" + assert p.name.startswith("prune_") + + def test_cmd_run_journalise_et_transmet_le_code_retour(self, tmp_path, monkeypatch): + repo = tmp_path / "CoursIA" + (repo / "scripts" / "ci").mkdir(parents=True) + target = repo / "scripts" / "ci" / "prune_merged_worktrees.py" + target.write_text("print('organ output')\n", encoding="utf-8") + monkeypatch.setattr(ipt, "LOG_DIR", tmp_path / "logs") + + recorded = {} + + def fake_run(cmd, stdout=None, stderr=None, **kw): + recorded["cmd"] = cmd + recorded["stdout"] = stdout + # simule un refus d'organe (worktree vivant) -> rc=1 + stdout.write("REFUSE uncommitted_source_changes\n") + return subprocess.CompletedProcess(cmd, 1) + + monkeypatch.setattr(subprocess, "run", fake_run) + rc = ipt.cmd_run(repo) + assert rc == 1 # le code de l'organe traverse + assert recorded["cmd"][-2:] == ["--apply"] or "--apply" in recorded["cmd"] + assert str(repo) in recorded["cmd"] + log = (tmp_path / "logs").glob("prune_*.log") + logs = list(log) + assert len(logs) == 1 + content = logs[0].read_text(encoding="utf-8") + assert "run start" in content and "run end rc=1" in content + assert "REFUSE uncommitted_source_changes" in content