feat(ci,#14473): installer for the daily prune_merged_worktrees scheduled task

.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.).

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())

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