Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
118 changes: 101 additions & 17 deletions scripts/coordination/install_merge_ready_task.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,25 +7,39 @@
une tache planifiee Windows toutes les 20 minutes qui lance
``merge_ready.py --apply`` sous l'identite coordinateur (myia-ai-01).

Fenetre masquee (audit de flotte 2026-09-24) : python.exe est un programme
console et le --run spawn des enfants console (git, gh), donc chaque tour
faisait flasher une fenetre noire sur le bureau interactif. L'action de la
tache est desormais wscript.exe sur un lanceur VBS qui execute la commande
en fenetre 0 (masquee) et attend la fin : la console cachee est heritee par
tous les enfants console ; pythonw.exe ne conviendrait pas -- sans console
attachee, chaque enfant ouvrirait sa PROPRE fenetre visible.

Modes :
--dry-run [--repo PATH] [--interval N] imprime la commande schtasks
EXACTE sans l'executer
EXACTE et le contenu du
lanceur VBS, sans rien
ecrire ni executer
(discipline UAC : la sortie
du dry-run precede toute
inscription schtasks)
--install [--repo PATH] [--interval N] cree la tache (idempotent)
--install [--repo PATH] [--interval N] ecrit le lanceur VBS puis
cree la tache (idempotent)
--status etat de la tache
--uninstall supprime la tache
--uninstall supprime la tache ET le
lanceur VBS
--run execute l'organe en --apply (invoque PAR la
tache) : journal horodate, jamais de TTY
tache, via le lanceur VBS) : journal
horodate, jamais de TTY

Garde : --install REFUSE de cabler --apply si l'organe cible est absent
(installer un cron vers un fichier qui n'existe pas deployerait un echec
silencieux toutes les 20 minutes).

Journal de la tache : %LOCALAPPDATA%\CoursIA\merge_ready\logs\
merge_ready_YYYYMMDD.log ; journal de l'organe :
%LOCALAPPDATA%\CoursIA\merge_ready\journal.jsonl.
%LOCALAPPDATA%\CoursIA\merge_ready\journal.jsonl ; lanceur masque :
%LOCALAPPDATA%\CoursIA\merge_ready\run_hidden.vbs.
"""
from __future__ import annotations

Expand All @@ -44,6 +58,13 @@
/ "merge_ready"
/ "logs"
)
# Lanceur masque (audit de flotte 2026-09-24) : wscript y execute la
# commande de la tache en fenetre 0, console heritee par les enfants.
VBS_PATH = LOG_DIR.parent / "run_hidden.vbs"
# Chemin ABSOLU : l'action de la tache ne doit pas dependre du PATH au reveil.
WSH_PATH = (
Path(os.environ.get("SystemRoot", r"C:\Windows")) / "System32" / "wscript.exe"
)
# Siege coordinateur (ai-01) : un worktree DEDIE a `origin/main`, pas le
# checkout principal. D:\CoursIA est le siege interactif, souvent sur une
# branche de travail : l'organe y executerait le gate et B.0 de CETTE branche,
Expand Down Expand Up @@ -75,18 +96,60 @@ def check_organ_present(repo: Path) -> tuple[bool, str]:


def task_command(repo: Path) -> list[str]:
"""Commande enregistree dans le planificateur : ce script --run, qui
journalise et appelle l'organe en --apply."""
"""Commande reellement executee a chaque tour : ce script --run, qui
journalise et appelle l'organe en --apply. C'est cette ligne que le
lanceur VBS execute (vbs_content) ; l'action ENREGISTREE dans le
planificateur est wscript (task_action)."""
return [
sys.executable,
str(repo / "scripts" / "coordination" / "install_merge_ready_task.py"),
"--run", "--repo", str(repo),
]


def vbs_content(repo: Path) -> str:
"""Contenu du lanceur run_hidden.vbs : la commande task_command(repo)
executee par WScript.Shell.Run en fenetre 0 (masquee), avec attente de
fin (True) et remontee du code de sortie (WScript.Quit) -- duree et
verdict de la tache restent lisibles dans le planificateur.

Le fichier est ecrit en ASCII : wscript lit les .vbs en ANSI, tout
caractere non-ASCII casserait la ligne de commande silencieusement --
on refuse fort (ValueError) plutot que d'ecrire un lanceur casse.
"""
cmdline = subprocess.list2cmdline(task_command(repo))
try:
cmdline.encode("ascii")
except UnicodeEncodeError as exc:
raise ValueError(
"ligne de commande non-ASCII (wscript lit les .vbs en ANSI, "
f"le lanceur serait casse) : {cmdline!r} -- {exc}"
) from exc
# litteral VBScript : chaque guillemet de la ligne de commande est double
escaped = cmdline.replace('"', '""')
return (
"' Lanceur genere par install_merge_ready_task.py -- ecrase a chaque\n"
"' --install, ne pas editer a la main.\n"
"' Fenetre 0 (masquee, heritee par les enfants console de --run) et\n"
"' attente de fin : le code de sortie remonte a la tache planifiee.\n"
"Dim WshShell\n"
"Set WshShell = CreateObject(\"WScript.Shell\")\n"
f"WScript.Quit WshShell.Run(\"{escaped}\", 0, True)\n"
)


def task_action() -> list[str]:
"""Action enregistree dans le planificateur : wscript execute le lanceur
VBS sans fenetre. //B (mode batch) supprime les dialogues d'erreur de
l'hote, //Nologo la banniere ; le prefixe // distingue les options de
wscript d'un chemin commenceant par /."""
return [str(WSH_PATH), "//B", "//Nologo", str(VBS_PATH)]


def build_schtasks_install(cmd: list[str], interval_minutes: int) -> list[str]:
"""Ligne schtasks /Create : toutes les N minutes, contexte utilisateur
courant (gh auth vit au niveau utilisateur), fenetre masquee."""
courant (gh auth vit au niveau utilisateur). ``cmd`` est l'ACTION de la
tache : le lanceur wscript (task_action), PAS python.exe en direct."""
# Quoter chaque element, jamais la ligne entiere : un /TR "python.exe script.py
# --run" enregistre la ligne comme NOM d'executable, et la tache echoue
# a chaque tour avec 0x80070002 (fichier introuvable) sans rien journaliser.
Expand All @@ -110,12 +173,20 @@ def task_exists() -> bool:


def cmd_dry_run(repo: Path, interval: int) -> int:
"""Imprime la commande EXACTE que --install enregistrerait. N'execute
RIEN : c'est la sortie que la discipline UAC exige de voir avant toute
inscription schtasks."""
command = build_schtasks_install(task_command(repo), interval)
"""Imprime la commande EXACTE que --install enregistrerait et le
contenu EXACT du lanceur VBS qu'il ecrirait. N'execute RIEN et
n'ecrit RIEN : c'est la sortie que la discipline UAC exige de voir
avant toute inscription schtasks."""
try:
vbs = vbs_content(repo)
except ValueError as exc:
print(f"REFUSE : {exc}", file=sys.stderr)
return 2
command = build_schtasks_install(task_action(), interval)
print("DRY-RUN -- commande schtasks que --install enregistrerait :")
print(" ".join(command))
print(f"DRY-RUN -- contenu de {VBS_PATH} que --install ecrirait :")
print(vbs, end="")
print(f"journal de la tache : {log_path_for()}")
return 0

Expand All @@ -126,8 +197,16 @@ def cmd_install(repo: Path, interval: int) -> int:
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), interval))
try:
vbs = vbs_content(repo)
except ValueError as exc:
print(f"REFUSE : {exc}", file=sys.stderr)
return 2
# le lanceur AVANT l'inscription : une tache enregistree ne doit jamais
# pointer vers un fichier absent
VBS_PATH.parent.mkdir(parents=True, exist_ok=True)
VBS_PATH.write_text(vbs, encoding="ascii")
proc = _run(build_schtasks_install(task_action(), interval))
if proc.returncode != 0:
print(
f"schtasks /Create echoue (rc={proc.returncode}) : "
Expand All @@ -136,8 +215,9 @@ def cmd_install(repo: Path, interval: int) -> int:
)
return 2
print(f"tache installee : {TASK_NAME} toutes les {interval} minutes")
print(f"commande : {' '.join(task_command(repo))}")
print(f"journal : {log_path_for()}")
print(f"lanceur : {VBS_PATH}")
print(f"commande : {' '.join(task_command(repo))}")
print(f"journal : {log_path_for()}")
print(
"verification : schtasks /Query /TN " + TASK_NAME + " /V /FO LIST"
)
Expand All @@ -162,6 +242,9 @@ def cmd_uninstall() -> int:
)
return 2
print(f"tache supprimee : {TASK_NAME}")
# la tache etait l'unique consommateur du lanceur : on le retire
VBS_PATH.unlink(missing_ok=True)
print(f"lanceur supprime : {VBS_PATH}")
return 0


Expand Down Expand Up @@ -202,7 +285,8 @@ def sync_repo(repo: Path) -> tuple[bool, str]:


def cmd_run(repo: Path) -> int:
"""Invoque par la tache planifiee : journal horodate, pas de TTY.
"""Invoque par la tache planifiee (via le lanceur VBS masque) : journal
horodate, pas de TTY.

L'organe part en dry-run par defaut a l'ecran, mais le cablage est le
geste : la tache lance --apply (c'est le mandat Q40 -- sans --apply le
Expand Down
120 changes: 112 additions & 8 deletions scripts/tests/test_install_merge_ready_task.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,18 @@

La discipline UAC du depot exige que le dry-run imprime la commande schtasks
exacte SANS l'executer : le test verifie les deux moities (la commande est
imprimee, aucun sous-processus n'est lance).
imprimee, aucun sous-processus n'est lance). Depuis le lanceur masque
(audit de flotte 2026-09-24), le dry-run couvre AUSSI le contenu du .vbs,
sans jamais l'ecrire.
"""

import importlib.util
import subprocess
import sys
from pathlib import Path

import pytest

HERE = Path(__file__).resolve().parent
INSTALLER_PATH = HERE.parent / "coordination" / "install_merge_ready_task.py"
_spec = importlib.util.spec_from_file_location(
Expand All @@ -19,16 +24,29 @@
_spec.loader.exec_module(imod)


def test_build_schtasks_toutes_les_20_minutes(tmp_path):
cmd = imod.build_schtasks_install(imod.task_command(tmp_path), 20)
def _redirect_cablage(monkeypatch, tmp_path):
"""Pointe LOG_DIR/VBS_PATH sous tmp_path : aucun test d'ecriture ne
touche le vrai %LOCALAPPDATA% de la machine."""
base = tmp_path / "cablage"
monkeypatch.setattr(imod, "LOG_DIR", base / "logs")
monkeypatch.setattr(imod, "VBS_PATH", base / "run_hidden.vbs")
return base


def test_build_schtasks_toutes_les_20_minutes():
# Adapte (audit de flotte 2026-09-24) : ce test affirmait que le /TR
# executait python.exe (install_merge_ready_task.py --run) ; l'action
# est desormais wscript + lanceur VBS, la commande python vit dans le
# .vbs (voir test_vbs_...).
cmd = imod.build_schtasks_install(imod.task_action(), 20)
assert cmd[0:3] == ["schtasks", "/Create", "/F"]
assert cmd[cmd.index("/TN") + 1] == imod.TASK_NAME
assert cmd[cmd.index("/SC") + 1] == "MINUTE"
assert cmd[cmd.index("/MO") + 1] == "20"
# la tache appelle le wrapper --run de cet installateur (qui journalise
# puis lance l'organe en --apply), pas l'organe nu
tr = cmd[cmd.index("/TR") + 1]
assert "install_merge_ready_task.py" in tr and "--run" in tr
assert "wscript.exe" in tr
assert "//B" in tr and "//Nologo" in tr
assert "run_hidden.vbs" in tr


def test_tr_quote_chaque_element_pas_la_ligne_entiere():
Expand All @@ -42,6 +60,27 @@ def test_tr_quote_chaque_element_pas_la_ligne_entiere():
assert tr != '"' + " ".join(cmd) + '"'


def test_vbs_masque_la_commande_et_remonte_le_code_de_sortie(tmp_path):
# Un repo AVEC espace force list2cmdline a quoter : les guillemets
# doivent etre doubles dans le literal VBScript.
repo = tmp_path / "repo avec espace"
content = imod.vbs_content(repo)
cmdline = subprocess.list2cmdline(imod.task_command(repo))
escaped = cmdline.replace('"', '""')
assert f'"{escaped}"' in content
assert ", 0, True)" in content
assert "WScript.Quit" in content
content.encode("ascii") # wscript lit les .vbs en ANSI


def test_vbs_refuse_une_commande_non_ascii(tmp_path):
# Un .vbs ANSI portant un accent casserait la ligne de commande
# silencieusement : refus fort plutot que lanceur casse.
accent = chr(233) # 'e' accentue
with pytest.raises(ValueError, match="ASCII"):
imod.vbs_content(tmp_path / ("depot" + accent))


def test_dry_run_imprime_la_commande_sans_l_executer(tmp_path, capsys, monkeypatch):
def boom(cmd, **kw):
raise AssertionError(
Expand All @@ -51,26 +90,36 @@ def boom(cmd, **kw):

monkeypatch.setattr(imod, "_run", boom)
monkeypatch.setattr(imod.subprocess, "run", boom)
rc = imod.main(["--dry-run", "--repo", str(tmp_path)])
base = _redirect_cablage(monkeypatch, tmp_path)
rc = imod.main(["--dry-run", "--repo", str(tmp_path / "repo")])
out = capsys.readouterr().out
assert rc == 0
assert "schtasks" in out
assert "/Create" in out
assert "/SC" in out and "MINUTE" in out
assert "/MO" in out and " 20" in out
assert "--run" in out
assert "DRY-RUN" in out
# le /TR cible wscript (pas python.exe), et le contenu VBS est imprime
assert "wscript.exe" in out and "//B" in out and "//Nologo" in out
assert "run_hidden.vbs" in out
assert ", 0, True)" in out and "WScript.Quit" in out
assert "install_merge_ready_task.py" in out and "--run" in out
# RIEN n'est ecrit : ni le lanceur, ni meme son repertoire
assert not (base / "run_hidden.vbs").exists()
assert not base.exists()


def test_install_refuse_organe_absent(tmp_path, capsys, monkeypatch):
def boom(cmd, **kw):
raise AssertionError("un depot vide ne doit jamais atteindre schtasks")

monkeypatch.setattr(imod, "_run", boom)
_redirect_cablage(monkeypatch, tmp_path)
repo = tmp_path / "vide" # sans scripts/coordination/merge_ready.py
rc = imod.main(["--install", "--repo", str(repo)])
assert rc == 2
assert "REFUSE" in capsys.readouterr().err
assert not imod.VBS_PATH.exists()


class _Res:
Expand All @@ -91,6 +140,61 @@ def fake(cmd, **kw):
return fake, calls


def test_install_ecrit_le_vbs_puis_enregistre(tmp_path, capsys, monkeypatch):
repo = tmp_path / "repo"
organ = repo / "scripts" / "coordination" / "merge_ready.py"
organ.parent.mkdir(parents=True)
organ.write_text("# stub d'organe\n", encoding="ascii")
base = _redirect_cablage(monkeypatch, tmp_path)
seen = {}

def fake(cmd, **kw):
if "schtasks" in cmd:
seen["tr"] = cmd[cmd.index("/TR") + 1]
# le lanceur doit exister AVANT l'inscription : la tache ne
# doit jamais pointer vers un fichier absent
seen["vbs_premier"] = imod.VBS_PATH.exists()
return _Res()

monkeypatch.setattr(imod, "_run", fake)
rc = imod.main(["--install", "--repo", str(repo)])
assert rc == 0
vbs = base / "run_hidden.vbs"
assert vbs.is_file()
content = vbs.read_text(encoding="ascii")
assert "install_merge_ready_task.py" in content and "--run" in content
assert ", 0, True)" in content and "WScript.Quit" in content
assert seen["vbs_premier"] is True
assert "wscript.exe" in seen["tr"]
assert "//B" in seen["tr"] and "run_hidden.vbs" in seen["tr"]
assert "python" not in seen["tr"].lower()


def test_uninstall_supprime_tache_et_lanceur(tmp_path, capsys, monkeypatch):
base = _redirect_cablage(monkeypatch, tmp_path)
base.mkdir(parents=True)
imod.VBS_PATH.write_text("' stub\n", encoding="ascii")
fake, calls = _scripted([])
monkeypatch.setattr(imod, "_run", fake)
rc = imod.main(["--uninstall"])
assert rc == 0
assert any("/Delete" in c for c in calls)
assert not imod.VBS_PATH.exists()
assert "lanceur supprime" in capsys.readouterr().out


def test_uninstall_echoue_garde_le_lanceur(tmp_path, monkeypatch):
# si schtasks /Delete echoue, la tache vit encore : son lanceur aussi.
base = _redirect_cablage(monkeypatch, tmp_path)
base.mkdir(parents=True)
imod.VBS_PATH.write_text("' stub\n", encoding="ascii")
fake, _ = _scripted([("/Delete", _Res(rc=1, err="acces refuse"))])
monkeypatch.setattr(imod, "_run", fake)
rc = imod.main(["--uninstall"])
assert rc == 2
assert imod.VBS_PATH.exists()


def test_sync_repo_refuse_hors_main(tmp_path, monkeypatch):
fake, calls = _scripted([("rev-parse", _Res(out="feat/x"))])
monkeypatch.setattr(imod, "_run", fake)
Expand Down
Loading