fix(ci,#18494): fenetre agent vivant -- REFUSE le retrait content_on_main d'un worktree actif

scripts/ci/prune_merged_worktrees.py

@@ -74,6 +74,16 @@
    cannot be moved or removed") : le prononcer REFUSE evite un FAILED a
    chaque passe --apply. Les gitignorés **non-cache** (`.env` laissé, ...)
    sont signalés (`ignored=...`) sans bloquer le retrait.
+7. **Fenetre « agent vivant » sur content_on_main (#18494)** : le retrait
+   par contenu deja integre (`HEAD` ancetre de `origin/main`, predicat 5
+   de l'issue #17771) est un critere de CONTENU, pas d'ACTIVITE. Un agent
+   vivant en phase de lecture (worktree propre, branche sans commit propre,
+   avant sa premiere edition) y est indiscernable d'un worktree abandonne.
+   Un marqueur d'activite recent -- mtime de `.lane-owner`, a defaut mtime
+   du dossier -- plus jeune que la fenetre (`--activity-window-h`, defaut
+   6 h) fait REFUSER le retrait (`reason=recent_activity:<age>h`). Des que
+   le marqueur depasse la fenetre, le worktree redevient retirable : le
+   refus protege la phase de travail, il ne conserve rien indefiniment.
 
 Ancre PR : `gh pr list --state all --search "head:<branch>"` (autoritative,
 cf matrice a 4 ancres de `.claude/rules/git-workflow.md` §orphan-branch-scan).
@@ -198,6 +208,7 @@
 import subprocess
 import sys
 import traceback
+import time
 from pathlib import Path
 from typing import Optional
 
@@ -504,6 +515,39 @@ def read_lane_owner(wt_path: str) -> Optional[str]:
     return None
 
 
+def recent_activity_age_hours(wt_path: str,
+                              now: Optional[float] = None) -> float:
+    """Age en heures du marqueur d'activite le plus recent du worktree.
+
+    #18494 : un worktree sans commit propre peut etre celui d'un agent
+    VIVANT en phase de lecture (spawn lance, pas encore sa premiere
+    edition). Deux marqueurs, du plus fiable au moins fiable :
+
+    - mtime de `.lane-owner`, pose par le spawn a la creation ;
+    - a defaut, mtime du dossier du worktree lui-meme (les agents
+      ecrivent tot : `.ipynb`, caches -- en phase de lecture seule, seule
+      la creation du worktree marque le dossier).
+
+    Retourne l'age du plus recent des deux. Chemin illisible : +inf (aucune
+    activite prouvable, le worktree n'est pas protege par cette garde).
+    """
+    now = time.time() if now is None else now
+    candidates: list = []
+    lane_owner = Path(wt_path) / ".lane-owner"
+    try:
+        if lane_owner.exists():
+            candidates.append(lane_owner.stat().st_mtime)
+    except OSError:
+        pass
+    try:
+        candidates.append(Path(wt_path).stat().st_mtime)
+    except OSError:
+        pass
+    if not candidates:
+        return float("inf")
+    return (now - max(candidates)) / 3600.0
+
+
 def same_worktree_path(a: str, b: str) -> bool:
     """Deux chemins de worktree designent-ils le meme repertoire ?
 
@@ -1079,12 +1123,16 @@ def _normalize(s: str) -> str:
 
 
 def diagnose_worktree(wt_path: str, current_path: str,
-                      head_sha: Optional[str] = None) -> WorktreeStatus:
+                      head_sha: Optional[str] = None,
+                      activity_window_h: float = 6.0) -> WorktreeStatus:
     """Diagnostic complet d'un worktree.
 
     `head_sha` (fourni par `list_worktrees`, porcelain) sert uniquement a
     la garde oid du cache de verdicts MERGED (#15369) : sans lui, l'etage
     cache est saute, jamais consulte a l'aveugle.
+
+    `activity_window_h` : fenetre de la garde « agent vivant » (#18494),
+    appliquee au retrait par contenu deja integre (predicat content_on_main).
     """
     info = get_worktree_info(wt_path, current_path)
 
@@ -1332,7 +1380,36 @@ def diagnose_worktree(wt_path: str, current_path: str,
     # committee ni untracked non tolere (sinon ``uncommitted_source_changes``
     # / ``untolerated_untracked`` seraient sortis). Le worktree ne porte
     # plus rien que main ne contienne deja.
+    #
+    # #18494 -- la fenetre « agent vivant ». `content_on_main` est un
+    # critere de CONTENU, pas d'ACTIVITE : un agent vivant traverse
+    # necessairement un etat « worktree propre + branche sans commit
+    # propre » (sa phase de lecture, avant la premiere edition), et un
+    # `--apply` dans cette fenetre detruit son plan de travail en cours.
+    # Un marqueur d'activite recent (`.lane-owner` ou mtime du dossier)
+    # fait donc REFUSER le retrait ; passe la fenetre, plus rien ne le
+    # protege -- le refus couvre la phase de travail, pas la conservation.
     if info["branch"] and head_is_ancestor_of_main(wt_path):
+        age_h = recent_activity_age_hours(wt_path)
+        if age_h < activity_window_h:
+            return WorktreeStatus(
+                path=wt_path,
+                branch=info["branch"],
+                is_current=False,
+                pr_state=None,
+                pr_number=None,
+                pr_url=None,
+                ahead_count=info["ahead_count"],
+                has_source_dirty=info["has_source_dirty"],
+                untracked_paths=info["untracked"],
+                decision="REFUSE",
+                refusal_reason=f"recent_activity:{age_h:.1f}h",
+                has_submodules=info["has_submodules"],
+                blocking_untracked=info.get("blocking_untracked", []),
+                ignored_extra=info.get("ignored_extra", []),
+                lane_owner=info.get("lane_owner"),
+                content_on_main=True,
+            )
         return WorktreeStatus(
             path=wt_path,
             branch=info["branch"],
@@ -1651,6 +1728,17 @@ def main() -> int:
              "sur le dashboard workspace si refused > N (#3895). Desactive "
              "par defaut ; la tache planifiee passe 20.",
     )
+    p.add_argument(
+        "--activity-window-h",
+        type=float,
+        default=6.0,
+        metavar="H",
+        help="Fenetre « agent vivant » (#18494) : un worktree sans commit "
+             "propre dont un marqueur d'activite (.lane-owner, a defaut "
+             "mtime du dossier) est plus jeune que H heures est REFUSE "
+             "(reason=recent_activity:<age>h) au lieu d'etre retire par "
+             "content_on_main. Defaut 6.0.",
+    )
     args = p.parse_args()
 
     cwd = args.path or "."
@@ -1696,7 +1784,8 @@ def main() -> int:
         try:
             statuses.append(
                 diagnose_worktree(
-                    wt["path"], current_path, head_sha=wt.get("head_sha")
+                    wt["path"], current_path, head_sha=wt.get("head_sha"),
+                    activity_window_h=args.activity_window_h,
                 )
             )
         except RuntimeError as e:

scripts/tests/test_prune_merged_worktrees.py

@@ -2023,7 +2023,7 @@ def _run_main(self, monkeypatch, statuses, argv_extra=()):
             {"path": s.path, "branch": s.branch} for s in statuses])
         monkeypatch.setattr(
             pmw, "diagnose_worktree",
-            lambda path, cur, head_sha=None:
+            lambda path, cur, head_sha=None, activity_window_h=6.0:
                 next(s for s in statuses if s.path == path),
         )
         monkeypatch.setattr(

tests/test_prune_merged_worktrees.py

@@ -0,0 +1,173 @@
+#!/usr/bin/env python3
+"""test_prune_merged_worktrees.py -- tests unitaires pour la garde
+« agent vivant » de prune_merged_worktrees.py (#18494).
+
+Deux niveaux :
+
+- fonctions deterministes pures (age du marqueur d'activite) -- CPU seul ;
+- le predicat content_on_main de diagnose_worktree sur un depot git
+  ephemere (init local + origin bare + worktree sans commit propre).
+  La resolution PR y est patchee : ce que ces tests mesurent est la
+  FENETRE D'ACTIVITE, pas l'ancre gh (deja couverte par sa propre
+  couche a trois etages, #15369).
+
+Les tests n'ont PAS besoin de GH_TOKEN ni de reseau.
+"""
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from unittest import mock
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+_REPO = os.path.dirname(_HERE)
+sys.path.insert(0, os.path.join(_REPO, "scripts", "ci"))
+
+import prune_merged_worktrees as pmw  # noqa: E402
+
+
+class TestRecentActivityAgeHours(unittest.TestCase):
+    """Age du marqueur d'activite : le plus jeune des deux gouverne."""
+
+    def test_fresh_directory_is_recent(self):
+        with tempfile.TemporaryDirectory() as d:
+            age = pmw.recent_activity_age_hours(d)
+            self.assertLess(age, 0.1)
+
+    def test_old_directory_no_lane_owner(self):
+        with tempfile.TemporaryDirectory() as d:
+            old = time.time() - 10 * 3600
+            os.utime(d, (old, old))
+            age = pmw.recent_activity_age_hours(d)
+            self.assertGreater(age, 9.9)
+            self.assertLess(age, 10.1)
+
+    def test_fresh_lane_owner_beats_old_directory(self):
+        # Cas de la mesure fondatrice (#18494) : le dossier date du spawn,
+        # le marqueur .lane-owner vient d'etre pose -- actif.
+        with tempfile.TemporaryDirectory() as d:
+            old = time.time() - 10 * 3600
+            os.utime(d, (old, old))
+            with open(os.path.join(d, ".lane-owner"), "w",
+                      encoding="utf-8") as f:
+                f.write("myia-po-2027:CoursIA\n")
+            age = pmw.recent_activity_age_hours(d)
+            self.assertLess(age, 0.1)
+
+    def test_unreadable_path_is_infinite(self):
+        age = pmw.recent_activity_age_hours(
+            os.path.join(tempfile.gettempdir(), "nope-18494", "wt"))
+        self.assertEqual(age, float("inf"))
+
+
+def _run_git(cwd, *args):
+    subprocess.run(
+        ["git", "-C", cwd, *args],
+        capture_output=True, text=True, encoding="utf-8", check=True,
+    )
+
+
+class TestActivityWindowPredicate(unittest.TestCase):
+    """Acceptance #18494 sur un depot ephemere.
+
+    Le worktree cible est celui d'un « agent vivant en phase de lecture » :
+    branche sans commit propre, HEAD ancetre de origin/main, arbre propre.
+    """
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        root = self._tmp.name
+        # Depot de travail + origin bare : origin/main doit exister pour
+        # que head_is_ancestor_of_main tranne (sinon le predicat 5 n'est
+        # meme pas atteint).
+        self.repo = os.path.join(root, "repo")
+        self.origin = os.path.join(root, "origin.git")
+        self.wt = os.path.join(root, "wt-agent")
+        os.makedirs(self.repo)
+        _run_git(self.repo, "init", "-b", "main")
+        _run_git(self.repo, "config", "user.email", "t@example.invalid")
+        _run_git(self.repo, "config", "user.name", "Test 18494")
+        with open(os.path.join(self.repo, "f.txt"), "w",
+                  encoding="utf-8") as f:
+            f.write("init\n")
+        _run_git(self.repo, "add", "f.txt")
+        _run_git(self.repo, "commit", "-m", "init")
+        _run_git(self.repo, "init", "--bare", self.origin)
+        _run_git(self.repo, "remote", "add", "origin", self.origin)
+        _run_git(self.repo, "push", "-u", "origin", "main")
+        # Worktree « agent vivant » : branche neuve sur la pointe de main,
+        # aucun commit propre, aucun fichier.
+        _run_git(self.repo, "worktree", "add", self.wt, "-b", "topic-agent")
+        # Le repo de travail doit voir origin/main (fetch implicite via
+        # push -u ; le worktree partage les refs du depot hebergeur).
+        _run_git(self.wt, "fetch", "origin", "main")
+        # Resolution PR patchee : hermetique, et le cas mesure n'a de
+        # toute facon aucune PR.
+        patcher = mock.patch.object(
+            pmw, "lookup_pr_for_branch", lambda branch, head_sha=None: None)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        patcher2 = mock.patch.object(
+            pmw, "remote_head_for_head",
+            lambda wt_path, branch, head_sha: None)
+        patcher2.start()
+        self.addCleanup(patcher2.stop)
+        pmw.reset_pr_resolution()
+
+    def tearDown(self):
+        pmw.reset_pr_resolution()
+        self._tmp.cleanup()
+
+    def _diagnose(self, window_h=6.0):
+        return pmw.diagnose_worktree(
+            self.wt, current_path=self.repo,
+            head_sha=None, activity_window_h=window_h,
+        )
+
+    def test_active_agent_is_refused_never_removed(self):
+        # Acceptance 1 : dry-run pendant qu'un agent travaille (worktree
+        # cree a l'instant, donc marqueur d'activite frais) -> REFUSE,
+        # jamais REMOVE.
+        status = self._diagnose()
+        self.assertEqual(status.decision, "REFUSE")
+        self.assertTrue(status.refusal_reason.startswith("recent_activity:"))
+        self.assertTrue(status.content_on_main)
+
+    def test_stale_activity_becomes_removable_again(self):
+        # Acceptance 2 : le marqueur depasse la fenetre -> retour au
+        # retrait par contenu deja integre. Le refus protege la phase de
+        # travail, il ne conserve rien indefiniment.
+        old = time.time() - 10 * 3600
+        os.utime(self.wt, (old, old))
+        status = self._diagnose()
+        self.assertEqual(status.decision, "REMOVE")
+        self.assertTrue(status.content_on_main)
+        self.assertIsNone(status.refusal_reason)
+
+    def test_window_parameter_moves_the_boundary(self):
+        # La fenetre est un parametre : avec 12 h, un worktree inactif
+        # depuis 10 h est encore protege.
+        old = time.time() - 10 * 3600
+        os.utime(self.wt, (old, old))
+        self.assertEqual(self._diagnose(window_h=6.0).decision, "REMOVE")
+        self.assertEqual(self._diagnose(window_h=12.0).decision, "REFUSE")
+
+    def test_lane_owner_marker_protects_old_directory(self):
+        # .lane-owner frais sur un dossier vieux : le marqueur le plus
+        # jeune gouverne, le worktree reste protege.
+        old = time.time() - 10 * 3600
+        os.utime(self.wt, (old, old))
+        with open(os.path.join(self.wt, ".lane-owner"), "w",
+                  encoding="utf-8") as f:
+            f.write("myia-po-2027:CoursIA\n")
+        status = self._diagnose()
+        self.assertEqual(status.decision, "REFUSE")
+        self.assertTrue(status.refusal_reason.startswith("recent_activity:"))
+
+
+if __name__ == "__main__":
+    unittest.main()