res: queue coalesces same-lock gates on descendant commitsHEADmaster

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JEAjUkQRrCYX5MZhWxdtj2

2 files changed, 133 insertions, 4 deletions

diff --git a/tests/test_res.py b/tests/test_res.py
index e4dfb1c..5f9ac92 100644
--- a/tests/test_res.py
+++ b/tests/test_res.py
@@ -566,10 +566,6 @@ class Units(unittest.TestCase):
                          "alias claude='systemd-run --user --scope --quiet --slice=agents.slice claude'")
 
 
-if __name__ == "__main__":
-    unittest.main()
-
-
 class Throttle(unittest.TestCase):
     def test_psi(self):
         text = "some avg10=41.50 avg60=33.20 avg300=12.00 total=99\nfull avg10=30.00 avg60=25.00 avg300=9.00 total=88\n"
@@ -806,3 +802,67 @@ class TaskFit(unittest.TestCase):
         self.assertEqual(R.batch_fit(4, 59, 30), 1)
         self.assertEqual(R.batch_fit(4, 29, 30), 0)
         self.assertEqual(R.batch_fit(4, 0, 30), 0)
+
+
+class Coalesce(unittest.TestCase):
+    """Batch r-740/744/757: one queued gate per lock on the newest commit (ruling 2026-10-07: the queue coalesces)."""
+    # history a ← b ← c (c newest); x on a side branch of a
+    ANC = {("a", "b"), ("a", "c"), ("b", "c"), ("a", "x")}
+
+    def anc(self, x, y):
+        return (x, y) in self.ANC
+
+    def gate(self, id, commit, at, lock="gate@/p"):
+        e = queued(id, 10.0, at=at)
+        e.lock, e.commit = lock, commit
+        return e
+
+    def test_descendant_supersedes_queued_ancestors(self):
+        g1, g2 = self.gate("r-740", "a", T(12)), self.gate("r-744", "b", T(13))
+        run = running("r-730", title="gate z")
+        run.lock, run.commit = "gate@/p", "z"
+        led = R.Ledger(entries=[run, g1, g2])
+        new = self.gate("r-757", "c", T(14))
+        cov, gone = R.coalesce(led, new, self.anc)
+        self.assertEqual((cov, [e.id for e in gone]), (None, ["r-740", "r-744"]))       # running r-730 untouched
+        led.entries.append(new)
+        R.supersede(new, gone, NOW)
+        self.assertEqual([(e.state, e.superseded_by, e.why) for e in (g1, g2)],
+                         [("done", "r-757", "superseded by r-757")] * 2)
+        self.assertEqual((new.covers, new.queued), (["a", "b"], T(12)))                 # keeps the oldest turn
+        self.assertEqual([e.id for e in R._queue(led)], ["r-757"])
+        self.assertEqual(run.state, "running")
+
+    def test_ancestor_is_covered_not_queued(self):
+        g = self.gate("r-757", "c", T(14))
+        led = R.Ledger(entries=[g])
+        for c in ("b", "c"):
+            cov, gone = R.coalesce(led, self.gate("r-760", c, T(15)), self.anc)
+            self.assertIs(cov, g)
+            self.assertEqual(gone, [])
+        R.cover(g, "b")
+        R.cover(g, "c")                                                                  # own commit not listed
+        self.assertEqual(g.covers, ["b"])
+
+    def test_unrelated_other_lock_or_no_commit_kept(self):
+        led = R.Ledger(entries=[self.gate("r-1", "x", T(12)), self.gate("r-2", "a", T(12), lock="gate@/q"),
+                                self.gate("r-3", "", T(12))])
+        self.assertEqual(R.coalesce(led, self.gate("r-9", "c", T(14)), self.anc), (None, []))
+        self.assertEqual(R.coalesce(led, self.gate("r-9", "", T(14)), self.anc), (None, []))
+
+    def test_red_coalesced_gate_names_range(self):
+        e = running("r-757", title="gate c")
+        e.commit, e.covers = "c" * 40, ["a" * 40, "b" * 40]
+        e.state, e.rc, e.ended = "done", 1, T(14, 30)
+        line = R.done_line(e)
+        self.assertIn("covers aaaaaaa,bbbbbbb", line)
+        self.assertIn("aaaaaaa^..ccccccc", line)
+        self.assertIn("bisect", line)
+        e.rc = 0
+        self.assertNotIn("bisect", R.done_line(e))
+        e.covers = []
+        self.assertNotIn("covers", R.done_line(e))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_res_io.py b/tests/test_res_io.py
index c30b21f..3d5b789 100644
--- a/tests/test_res_io.py
+++ b/tests/test_res_io.py
@@ -27,6 +27,7 @@ class Fake:
         self.scopes = {}        # name → {prop: value} for agent session scopes
         self.manager_env = "PATH=/usr/bin\nHOME=/h/u\n"
         self.git = {}           # git subcommand (branch | rev-parse) → stdout
+        self.real_git = False   # True: git calls run for real (cwd = a synthetic repo)
 
     def __call__(self, argv):
         self.calls.append(list(argv))
@@ -34,6 +35,8 @@ class Fake:
         if key in self.fail:
             return subprocess.CompletedProcess(argv, 1, "", self.fail[key] + "\n")
         out = ""
+        if argv[0] == "git" and self.real_git:
+            return subprocess.run(argv, capture_output=True, text=True)
         if argv[:3] == ["systemctl", "--user", "show"]:
             blocks = []
             for name in argv[5:]:
@@ -685,6 +688,72 @@ class Robust(IOBase):
         self.assertEqual((code, err), (1, "wf: [Errno 13] Permission denied: '/proc/meminfo'\n"))
 
 
+class CoalesceIO(IOBase):
+    """Batch of commits a ← b ← c, each queueing its gate (real repo): one queued gate per lock, on c."""
+
+    def setUp(self):
+        super().setUp()
+        repo = self.box.tmp / "proj"
+        git = lambda *a: subprocess.run(["git", "-C", str(repo), *a], check=True, capture_output=True, text=True).stdout
+        git("init", "-q", "-b", "master")
+        self.sha = {}
+        for c in ("a", "b", "c"):
+            git("-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "--allow-empty", "-m", c)
+            self.sha[c] = git("rev-parse", "HEAD").strip()
+        git("checkout", "-q", "-b", "side", self.sha["a"])
+        git("-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "--allow-empty", "-m", "x")
+        self.sha["x"] = git("rev-parse", "HEAD").strip()
+        self.box.fake.real_git = True
+
+    def gate(self, c, *extra):
+        return self.box.wf("run", "--queue", "--mem", "2G", "--for", "30m", "--title", f"gate {self.sha[c][:7]}",
+                           *extra, "--", "x")
+
+    def test_batch_yields_one_queued_gate(self):
+        self.box.wf("run", "--mem", "2G", "--for", "30m", "--title", "gate running", "--", "x")     # r-1 holds the lock
+        self.assertEqual(self.gate("a")[0], 0)
+        self.box.tick(60)
+        self.gate("b")
+        self.box.tick(60)
+        code, out, _ = self.gate("c")
+        self.assertEqual(code, 0)
+        self.assertIn("r-4 queued, position 1 (lock held by r-1)", out)
+        self.assertIn("supersedes r-3", out)
+        led = self.box.ledger()
+        self.assertEqual([(e.id, e.commit) for e in led.entries if e.state == "queued"], [("r-4", self.sha["c"])])
+        self.assertEqual(led.get("r-4").covers, [self.sha["a"], self.sha["b"]])
+        self.assertEqual(led.get("r-4").queued, led.get("r-2").queued)
+        self.assertEqual([led.get(i).superseded_by for i in ("r-2", "r-3")], ["r-3", "r-4"])     # chain; wait follows it
+        code, out, _ = self.gate("b")                           # late ancestor: covered, nothing new
+        self.assertEqual(code, 0)
+        self.assertTrue(out.startswith("r-4 queued"), out)
+        self.assertIn(f"covers {self.sha['b'][:7]}", out)
+        self.assertEqual(len([e for e in self.box.ledger().entries if e.state == "queued"]), 1)
+        code, out, _ = self.gate("x")                           # side branch: own gate
+        self.assertTrue(out.startswith("r-5 queued"), out)
+        code, out, _ = self.box.wf("run", "--queue", "--mem", "2G", "--for", "30m", "--title", "gate",
+                                   "--commit", "master", "--", "x")
+        self.assertTrue(out.startswith("r-4 queued"), out)      # --commit REV, no hex in title
+        self.assertNotEqual(self.box.wf("run", "--queue", "--mem", "2G", "--for", "30m", "--title", "gate",
+                                        "--commit", "nope", "--", "x")[0], 0)
+
+    def test_wait_follows_superseded_and_red_names_range(self):
+        self.box.wf("run", "--mem", "2G", "--for", "30m", "--title", "gate running", "--", "x")
+        self.gate("a")
+        self.gate("c")
+        self.box.fake.units["wf-r-1.service"] = ("inactive", None)
+        self.box.wf("status")                                   # r-3 starts
+        logs = self.box.env.state / "logs"
+        logs.mkdir(parents=True, exist_ok=True)
+        (logs / "r-3.rc").write_text("1\n")
+        self.box.fake.units["wf-r-3.service"] = ("inactive", None)
+        code, out, _ = self.box.wf("wait", "r-2")
+        self.assertEqual(code, 0)
+        self.assertIn("r-2 superseded by r-3", out)
+        self.assertIn(f"r-3 done rc=1", out)
+        self.assertIn(f"{self.sha['a'][:7]}^..{self.sha['c'][:7]}", out)
+
+
 if __name__ == "__main__":
     unittest.main()