import datetime as dt
import json
import sys
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from wflib import res as R

UTC = dt.timezone.utc


def T(h, m=0, day=1):
    return dt.datetime(2026, 10, day, h, m, tzinfo=UTC)


NOW = T(14, 10)
GIB = 1024 ** 3


def running(id="r-4", project="proj-a", title="dotnet e2e", mem=10.0, cpus=2, est=40, started=T(14, 0)):
    return R.Entry(id=id, project=project, owner=1, title=title, mem_gb=mem, cpus=cpus, est_min=est,
                   state="running", unit=f"wf-{id}.service", started=started,
                   expires=started + dt.timedelta(minutes=2 * est))


def note(id="r-8", owner=77, mem=3.0, cpus=1, expires=T(14, 30)):
    return R.Entry(id=id, project="proj-b", owner=owner, title="dotnet test", mem_gb=mem, cpus=cpus,
                   est_min=20, state="note", started=T(14, 0), expires=expires)


def facts(available=20.0, units=None, live=(), results=None, slice_gb=0.0, now=NOW):
    return R.Facts(now=now, total_gb=30.0, available_gb=available, nproc=16, units=units or {},
                   live=set(live), results=results or {}, slice_gb=slice_gb)


class Parse(unittest.TestCase):
    def test_size(self):
        self.assertEqual(R.parse_size("10G"), 10.0)
        self.assertEqual(R.parse_size("512M"), 0.5)
        self.assertEqual(R.parse_size("1.5g"), 1.5)
        with self.assertRaisesRegex(R.ResError, "size '10'"):
            R.parse_size("10")

    def test_duration(self):
        self.assertEqual(R.parse_duration("40m"), 40)
        self.assertEqual(R.parse_duration("4h"), 240)
        self.assertEqual(R.parse_duration("1h30m"), 90)
        self.assertEqual(R.parse_duration("90"), 90)
        for bad in ("0m", "x", "", "h"):
            with self.assertRaises(R.ResError):
                R.parse_duration(bad)

    def test_meminfo(self):
        text = "MemTotal:       31457280 kB\nMemFree:  1 kB\nMemAvailable:   12582912 kB\n"
        self.assertEqual(R.meminfo(text), (30.0, 12.0))
        with self.assertRaises(R.ResError):
            R.meminfo("MemFree: 1 kB\n")

    def test_show_units(self):
        text = ("Id=agents.slice\nActiveState=active\nMemoryCurrent=0\n\n"
                "Id=wf-r-1.service\nActiveState=inactive\nMemoryCurrent=[not set]\n")
        self.assertEqual(R.show_units(text), {
            "agents.slice": {"Id": "agents.slice", "ActiveState": "active", "MemoryCurrent": "0"},
            "wf-r-1.service": {"Id": "wf-r-1.service", "ActiveState": "inactive", "MemoryCurrent": "[not set]"}})

    def test_gb_or_none(self):
        self.assertEqual(R.gb_or_none("1073741824"), 1.0)
        self.assertIsNone(R.gb_or_none("[not set]"))
        self.assertIsNone(R.gb_or_none("infinity"))

    def test_formats(self):
        self.assertEqual(R.fmt_gb(2.25), "2.2 GB")
        self.assertEqual(R.fmt_bytes(1288490188), "1.2 GB")
        self.assertEqual(R.fmt_bytes(300 * 1024 ** 2), "300 MB")
        self.assertEqual(R.fmt_dur(240), "4h")
        self.assertEqual(R.fmt_dur(90), "1h30m")
        self.assertEqual(R.fmt_dur(40), "40m")
        self.assertEqual(R.hhmm(T(9, 5)), "09:05")


class ConfigTest(unittest.TestCase):
    def test_defaults_and_override(self):
        cfg = R.load_config("user_reserve_gb = 8\n")
        self.assertEqual((cfg.user_reserve_gb, cfg.user_reserve_cpus, cfg.game_reserve_gb, cfg.game_reserve_cpus,
                          cfg.game_hours, cfg.small_headroom_gb, cfg.scratch_hours), (8.0, 4, 12.0, 8, 4.0, 2.0, 2.0))

    def test_unknown_key(self):
        self.assertEqual(R.load_config("session_mem_gb = 12").session_mem_gb, 12.0)
        self.assertEqual(R.Config().session_mem_gb, 6.0)
        with self.assertRaisesRegex(R.ResError, "unknown key\\(s\\) foo"):
            R.load_config("foo = 1\n")

    def test_negative(self):
        with self.assertRaisesRegex(R.ResError, "game_hours must be a number ≥ 0"):
            R.load_config("game_hours = -1\n")


class LedgerJson(unittest.TestCase):
    def test_roundtrip(self):
        led = R.Ledger(next=9, game_until=T(18), entries=[running(), note()])
        text = R.dumps(led)
        self.assertIn('"started": "2026-10-01T14:00:00+00:00"', text)
        self.assertEqual(R.loads(text), led)

    def test_empty(self):
        self.assertEqual(R.loads(""), R.Ledger())

    def test_corrupt(self):
        for bad in ("{", '{"entries": [{"id": "r-1"}]}', "[]"):
            with self.assertRaisesRegex(R.ResError, "corrupt ledger"):
                R.loads(bad)

    def test_unknown_entry_keys_ignored(self):
        d = json.loads(R.dumps(R.Ledger(next=9, entries=[running()])))
        d["entries"][0]["from_a_newer_version"] = 1
        self.assertEqual(R.loads(json.dumps(d)), R.Ledger(next=9, entries=[running()]))

    def test_unknown_keys_survive_rewrite(self):
        d = json.loads(R.dumps(R.Ledger(next=9, entries=[running()])))
        d["entries"][0]["field_x"] = {"a": 1}
        d["top_x"] = 5
        out = json.loads(R.dumps(R.loads(json.dumps(d))))
        self.assertEqual(out["entries"][0]["field_x"], {"a": 1})
        self.assertEqual(out["top_x"], 5)

    def test_new_id_and_get(self):
        led = R.Ledger(next=3)
        self.assertEqual(led.new_id(), "r-3")
        self.assertEqual(led.next, 4)
        with self.assertRaisesRegex(R.ResError, "no entry 'r-9'"):
            led.get("r-9")


class Prune(unittest.TestCase):
    def test_prune(self):
        r1, r2 = running("r-1"), running("r-2")
        n3, n5 = note("r-3", owner=77), note("r-5", owner=78, expires=T(14, 5))
        d6 = R.Entry(id="r-6", project="p", owner=1, title="old", mem_gb=1, cpus=1, est_min=1, state="done",
                     ended=NOW - dt.timedelta(hours=25))
        d7 = R.Entry(id="r-7", project="p", owner=1, title="new", mem_gb=1, cpus=1, est_min=1, state="done",
                     ended=NOW - dt.timedelta(hours=23))
        led = R.Ledger(game_until=T(14, 9), entries=[r1, r2, n3, n5, d6, d7])
        f = facts(units={"wf-r-1.service": R.Unit(False)}, live={78}, results={"r-1": (0, 9.4, "")})
        lines = R.prune(led, f)
        self.assertEqual(lines, ["r-1 exited rc=0", "r-2 exited rc=?", "r-3 freed (owner gone)",
                                 "r-5 freed (expired)", "game off (expired)"])
        self.assertEqual([e.id for e in led.entries], ["r-1", "r-2", "r-3", "r-5", "r-7"])
        self.assertEqual((r1.state, r1.rc, r1.peak_gb, r1.why, r1.ended), ("done", 0, 9.4, "exited", NOW))
        self.assertEqual((n3.why, n5.why), ("owner gone", "expired"))
        self.assertIsNone(led.game_until)

    def test_prune_missing_results(self):
        r = running("r-1")
        R.prune(R.Ledger(entries=[r]), facts(units={"wf-r-1.service": R.Unit(False)}))
        self.assertEqual((r.state, r.rc, r.peak_gb), ("done", None, None))

    def test_overdue_running_is_kept(self):
        r = running(started=T(10))
        R.prune(R.Ledger(entries=[r]), facts(units={"wf-r-4.service": R.Unit(True, 4.0)}))
        self.assertEqual(r.state, "running")


class Capacity(unittest.TestCase):
    def setUp(self):
        self.cfg = R.Config()

    def test_budget(self):
        led = R.Ledger(entries=[running(), note()])
        f = facts(units={"wf-r-4.service": R.Unit(True, 4.0)}, live={77})
        # 20 − 6 reserve − 2 headroom − (10−4) unused − 3 note = 3; 16 − 4 − (2+1) = 9
        self.assertEqual(R.budget(self.cfg, led, f), (3.0, 9))

    def test_budget_gaming(self):
        led = R.Ledger(game_until=T(15), entries=[running(), note()])
        f = facts(units={"wf-r-4.service": R.Unit(True, 4.0)}, live={77})
        # 20 − 12 − 2 − 6 − 3 = −3; 16 − 8 − 3 = 5
        self.assertEqual(R.budget(self.cfg, led, f), (-3.0, 5))

    def test_held_over_reservation_is_zero(self):
        self.assertEqual(R.held_gb(running(), facts(units={"wf-r-4.service": R.Unit(True, 12.0)})), 0.0)
        self.assertEqual(R.frees_gb(running(), facts(units={"wf-r-4.service": R.Unit(True, 12.0)})), 12.0)

    def test_fits(self):
        self.assertTrue(R.fits(3.0, 9, (3.0, 9)))
        self.assertFalse(R.fits(3.1, 1, (3.0, 9)))
        self.assertFalse(R.fits(1.0, 10, (3.0, 9)))

    def test_never_fits(self):
        f = facts()
        self.assertEqual(R.never_fits(self.cfg, f, 25.0, 1), "25.0 GB can never fit (max 22.0 GB for agents)")
        self.assertEqual(R.never_fits(self.cfg, f, 1.0, 13), "13 cpus can never fit (max 12 for agents)")
        self.assertIsNone(R.never_fits(self.cfg, f, 22.0, 12))

    def test_queued_total(self):
        q = running("r-9")
        q.state = "queued"
        self.assertEqual(R.queued_total(R.Ledger(entries=[q, running()])), (10.0, 2))


class BusyLine(unittest.TestCase):
    def setUp(self):
        self.cfg = R.Config()
        self.led = R.Ledger(entries=[running()])
        self.f = facts(units={"wf-r-4.service": R.Unit(True, 4.0)})   # budget 6.0 GB, 10 cpus

    def test_memory(self):
        self.assertEqual(R.busy_line(self.cfg, self.led, self.f, 8.0, 1),
                         'busy: 10.0 GB held by proj-a "dotnet e2e" (r-4) until ~14:40; 6.0 GB free for agents; '
                         'retry after ~14:40 or work on something else')

    def test_cpus(self):
        self.assertEqual(R.busy_line(self.cfg, self.led, self.f, 1.0, 12),
                         'busy: 2 cpus held by proj-a "dotnet e2e" (r-4) until ~14:40; 10 cpus free for agents; '
                         'retry after ~14:40 or work on something else')

    def test_no_holder(self):
        self.assertEqual(R.busy_line(self.cfg, R.Ledger(), facts(available=7.0), 1.0, 1),
                         "busy: only 0.0 GB free for agents and no agent job holds any; other programs use the rest; "
                         "retry later or work on something else")

    def test_busy_overdue_holder(self):
        led = R.Ledger(entries=[running(started=T(10))])
        self.assertEqual(R.busy_line(self.cfg, led, self.f, 8.0, 1),
                         'busy: 10.0 GB held by proj-a "dotnet e2e" (r-4) until overdue; 6.0 GB free for agents; '
                         'retry later or work on something else')

    def test_two_holders_earliest_first(self):
        a = running("r-1", project="a", title="A", mem=4.0, cpus=1, est=60)       # ends 15:00
        b = running("r-2", project="b", title="B", mem=4.0, cpus=1, est=30)       # ends 14:30
        f = facts(available=14.0, units={})   # 14 − 6 − 2 − 4 − 4 = −2.0 budget
        self.assertEqual(R.busy_line(self.cfg, R.Ledger(entries=[a, b]), f, 5.0, 1),
                         'busy: 4.0 GB held by b "B" (r-2) until ~14:30, 4.0 GB held by a "A" (r-1) until ~15:00; '
                         '0.0 GB free for agents; retry after ~15:00 or work on something else')


class Lines(unittest.TestCase):
    def test_run_argv(self):
        e = running("r-3", mem=10.0)
        e.cwd, e.log, e.cmd = "/projects/x", "/s/logs/r-3.log", ["make", "it big", "--fast"]
        self.assertEqual(R.run_argv(e, "/s/logs"), [
            "systemd-run", "--user", "--quiet", "--collect", "--slice=agents-jobs.slice", "--unit=wf-r-3.service",
            "--working-directory=/projects/x", "--setenv=WF_RES_ID=r-3",
            "-p", "MemoryMax=10737418240", "-p", "MemoryHigh=9663676416", "-p", "MemorySwapMax=0", "-p", "Nice=10",
            "-p", "StandardOutput=append:/s/logs/r-3.log", "-p", "StandardError=append:/s/logs/r-3.log",
            "/bin/sh", "-c",
            '"$@"; rc=$?; cat /sys/fs/cgroup$(cut -d: -f3 /proc/self/cgroup)/memory.peak > /s/logs/r-3.peak '
            '2>/dev/null; echo $rc > /s/logs/r-3.rc',
            "sh", "make", "it big", "--fast"])

    def test_run_argv_env(self):
        e = running("r-3", mem=1.0)
        e.cwd, e.log, e.cmd, e.env = "/p", "/s/r-3.log", ["x"], {"B": "2", "A": "1 2"}
        argv = R.run_argv(e, "/s")
        self.assertEqual(argv[argv.index("--working-directory=/p") + 1:][:2], ["--setenv=A=1 2", "--setenv=B=2"])

    def test_env_diff(self):
        self.assertEqual(R.env_diff({"A": "1", "B": "2", "C": "3", "PWD": "/", "OLDPWD": "/", "SHLVL": "1", "_": "/x",
                                     "MY_TOKEN": "t", "API_SECRET": "s", "PGPASSWORD": "p", "SSH_AUTH_SOCK": "/s"},
                                    {"A": "1", "B": "9", "SSH_AUTH_SOCK": "/s"}),
                         {"B": "2", "C": "3"})

    def test_manager_env_parse(self):
        self.assertEqual(R.parse_show_environment("A=1\nB=x=y\n\nbad\n"), {"A": "1", "B": "x=y"})

    def test_entry_lines(self):
        f = facts(units={"wf-r-4.service": R.Unit(True, 4.0)}, live={77})
        q = running("r-9", project="p", title="Q", mem=1.0, cpus=1)
        q.state = "queued"
        d = R.Entry(id="r-2", project="p", owner=1, title="D", mem_gb=1, cpus=1, est_min=1, state="done",
                    ended=T(13, 50), rc=None, peak_gb=None, why="exited")
        led = R.Ledger(entries=[running(), note(), q, d])
        self.assertEqual([R.entry_line(e, led, f) for e in led.entries], [
            'r-4 proj-a "dotnet e2e" running 10.0 GB used 4.0 GB 2 cpu since 14:00 ETA ~14:40',
            'r-8 proj-b "dotnet test" note 3.0 GB 1 cpu until ~14:30',
            'r-9 p "Q" queued #1 1.0 GB 1 cpu',
            'r-2 p "D" done (exited) rc=? peak ? at 13:50'])

    def test_status_lines(self):
        f = facts(units={"wf-r-4.service": R.Unit(True, 4.0)}, slice_gb=5.5)
        self.assertEqual(R.status_lines(R.Config(), R.Ledger(entries=[running()]), f), [
            'r-4 proj-a "dotnet e2e" running 10.0 GB used 4.0 GB 2 cpu since 14:00 ETA ~14:40',
            "agents may use 6.0 GB, 10 cpus now; reserve 6.0 GB/4 cpus",
            "unreserved agent memory 1.5 GB"])
        f.slice_cache_gb = 0.6
        self.assertEqual(R.status_lines(R.Config(), R.Ledger(entries=[running()]), f)[-1],
                         "unreserved agent memory 1.5 GB (0.6 GB of it file cache, reclaimable)")
        f.slice_cache_gb = 3.0      # cache partly inside the jobs' 4.0 GB → capped at the unreserved figure
        self.assertEqual(R.status_lines(R.Config(), R.Ledger(entries=[running()]), f)[-1],
                         "unreserved agent memory 1.5 GB (1.5 GB of it file cache, reclaimable)")

    def test_cache_gb(self):
        # agents.slice memory.stat excerpt from this machine: file 5602209792, shmem (tmpfs, not reclaimable) 3861041152
        self.assertEqual(round(R.cache_gb("anon 11335516160\nfile 5602209792\nkernel 1\nshmem 3861041152\n"), 3), 1.622)
        self.assertEqual(R.cache_gb(""), 0.0)

    def test_status_lines_empty_overdue(self):
        self.assertEqual(R.status_lines(R.Config(), R.Ledger(), facts(available=7.0)), [
            "no reservations", "agents may use 0.0 GB, 12 cpus now; reserve 6.0 GB/4 cpus"])
        r = running(started=T(10))
        self.assertIn("ETA overdue (still running, not killed)", R.entry_line(r, R.Ledger(entries=[r]), facts()))

    def test_status_json(self):
        import json
        d = json.loads(R.status_json(R.Config(), R.Ledger(entries=[running()]),
                                     facts(units={"wf-r-4.service": R.Unit(True, 4.0)})))
        self.assertEqual((d["budget_gb"], d["budget_cpus"], d["gaming"], d["entries"][0]["id"]),
                         (6.0, 10, False, "r-4"))

    def test_done_line(self):
        e = running("r-3", started=T(14, 0))
        e.state, e.ended, e.rc, e.peak_gb = "done", T(14, 37), 0, 9.4
        self.assertEqual(R.done_line(e), "r-3 done rc=0 peak 9.4 GB in 37 min")
        e.rc, e.peak_gb = None, None
        self.assertEqual(R.done_line(e), "r-3 done rc=? peak ? in 37 min")
        e.why = "killed: timeout"
        self.assertEqual(R.done_line(e), "r-3 done rc=? peak ? in 37 min (killed: timeout)")

    def test_prune_killed(self):
        r = running("r-1")
        f = facts(units={"wf-r-1.service": R.Unit(False)}, results={"r-1": (None, 3.9, "killed: timeout")})
        self.assertEqual(R.prune(R.Ledger(entries=[r]), f), ["r-1 killed: timeout"])
        self.assertEqual((r.state, r.rc, r.peak_gb, r.why), ("done", None, 3.9, "killed: timeout"))


# journalctl --user -u wf-r-12.service -o cat, verbatim from a real systemd-oomd kill (systemd 258)
OOMD = """Started wf-r-12.service - [systemd-run] /bin/sh -c "x" sh dotnet test.
wf-r-12.service: systemd-oomd killed some process(es) in this unit.
wf-r-12.service: Main process exited, code=killed, status=9/KILL
wf-r-12.service: Failed with result 'oom-kill'.
wf-r-12.service: Consumed 11min 2.837s CPU time, 3.9G memory peak.
"""


class JournalReason(unittest.TestCase):
    def test_oomd(self):
        self.assertEqual(R.journal_reason(OOMD, 4.0),
                         ("killed: oom-kill by systemd-oomd, limit 4.0 GB; raise --mem", 3.9))

    def test_kernel_oom(self):
        text = ("u: A process of this unit has been killed by the OOM killer.\n"
                "u: Main process exited, code=killed, status=9/KILL\n"
                "u: Failed with result 'oom-kill'.\nu: Consumed 2s CPU time, 512M memory peak.\n")
        self.assertEqual(R.journal_reason(text, 0.5),
                         ("killed: oom-kill at MemoryMax, limit 0.5 GB; raise --mem", 0.5))

    def test_signal(self):
        text = "u: Main process exited, code=killed, status=15/TERM\nu: Failed with result 'signal'.\n"
        self.assertEqual(R.journal_reason(text, 1.0), ("killed: signal 15/TERM", None))

    def test_timeout_and_exit(self):
        self.assertEqual(R.journal_reason("u: Failed with result 'timeout'.\n", 1.0), ("killed: timeout", None))
        text = "u: Main process exited, code=exited, status=3/NOTIMPLEMENTED\nu: Failed with result 'exit-code'.\n"
        self.assertEqual(R.journal_reason(text, 1.0), ("failed: exit 3/NOTIMPLEMENTED", None))

    def test_nothing(self):
        self.assertEqual(R.journal_reason("", 1.0), ("", None))
        self.assertEqual(R.journal_reason("u: Deactivated successfully.\nu: Consumed 1s CPU time, 1.5G memory peak.\n", 1.0),
                         ("", 1.5))

def queued(id, mem, cpus=1, at=T(13, 0)):
    e = R.Entry(id=id, project="p", owner=1, title=id, mem_gb=mem, cpus=cpus, est_min=10, state="queued",
                cmd=["x"], queued=at)
    return e


class Queue(unittest.TestCase):
    def test_fifo_blocked_head(self):
        # budget 20 − 6 − 2 = 12 GB
        led = R.Ledger(entries=[queued("r-1", 5, at=T(13, 0)), queued("r-2", 9, at=T(13, 1)),
                                queued("r-3", 1, at=T(13, 2))])
        self.assertEqual([e.id for e in R.to_start(R.Config(), led, facts())], ["r-1"])

    def test_all_fit(self):
        led = R.Ledger(entries=[queued("r-2", 4, at=T(13, 1)), queued("r-1", 4, at=T(13, 0))])
        self.assertEqual([e.id for e in R.to_start(R.Config(), led, facts())], ["r-1", "r-2"])

    def test_estimate(self):
        # running r-4 holds 10 (used 4), ends 14:40; budget 6; queue r-1 (5) ahead of r-2 (4): need 9 − 6 = 3
        led = R.Ledger(entries=[running(), queued("r-1", 5), queued("r-2", 4, at=T(13, 5))])
        f = facts(units={"wf-r-4.service": R.Unit(True, 4.0)})
        self.assertEqual(R.queue_estimate(R.Config(), led, f, led.get("r-2")), T(14, 40))

    def test_estimate_unknown(self):
        led = R.Ledger(entries=[queued("r-1", 20)])
        self.assertIsNone(R.queue_estimate(R.Config(), led, facts(), led.get("r-1")))

class BatchLoan(unittest.TestCase):
    """Real case: batch r-741 claims 6.3 GB, its gate r-757 (10 GB) queued, budget 7.6 → must not deadlock."""

    def setUp(self):
        self.batch = running("r-741", title="wf batch", mem=6.3, cpus=1)
        # 21.9 − 6 reserve − 2 headroom − 6.3 unused claim = 7.6 GB budget
        self.f = facts(available=21.9, units={"wf-r-741.service": R.Unit(True, 0.0)})

    def gate(self, id="r-757", batch="r-741", mem=10.0, at=T(13, 0)):
        e = queued(id, mem, at=at)
        e.by = {"batch": batch} if batch else {}
        return e

    def test_budget_is_real_case(self):
        led = R.Ledger(entries=[self.batch])
        self.assertAlmostEqual(R.budget(R.Config(), led, self.f)[0], 7.6)

    def test_own_gate_borrows_batch_claim(self):
        led = R.Ledger(entries=[self.batch, self.gate()])
        self.assertEqual([e.id for e in R.to_start(R.Config(), led, self.f)], ["r-757"])
        self.assertIsNotNone(R.queue_estimate(R.Config(), led, self.f, led.get("r-757")))

    def test_foreign_job_does_not_borrow(self):
        led = R.Ledger(entries=[self.batch, self.gate(batch="")])
        self.assertEqual(R.to_start(R.Config(), led, self.f), [])
        led = R.Ledger(entries=[self.batch, self.gate(batch="r-1")])     # other batch, not live
        self.assertEqual(R.to_start(R.Config(), led, self.f), [])

    def test_claim_lent_once(self):
        # 7.6 + 6.3 = 13.9: first 10 GB gate fits, second does not (loan used up, 3.9 left)
        led = R.Ledger(entries=[self.batch, self.gate("r-757"), self.gate("r-758", at=T(13, 1))])
        self.assertEqual([e.id for e in R.to_start(R.Config(), led, self.f)], ["r-757"])
        led.get("r-757").state = "running"
        led.get("r-757").unit = "wf-r-757.service"
        self.f.units["wf-r-757.service"] = R.Unit(True, 0.0)
        self.assertEqual(R.loan(led, led.get("r-758")), (0.0, 0))
        self.assertEqual(R.to_start(R.Config(), led, self.f), [])

    def test_dead_batch_gate_dropped(self):
        # r-740: gate of dead batch r-731 heads the queue; r-757 behind it must still start
        dead = self.gate("r-740", batch="r-731", at=T(12, 0))
        led = R.Ledger(entries=[self.batch, dead, self.gate()])
        self.assertEqual(R.prune(led, self.f), ["r-740 dropped (batch r-731 gone)"])
        self.assertEqual((dead.state, dead.why), ("done", "owner gone"))
        self.assertEqual([e.id for e in R.to_start(R.Config(), led, self.f)], ["r-757"])

    def test_batch_ending_in_same_prune_drops_gate(self):
        led = R.Ledger(entries=[self.batch, self.gate()])
        f = facts(units={"wf-r-741.service": R.Unit(False)})
        self.assertEqual(R.prune(led, f), ["r-741 exited rc=?", "r-757 dropped (batch r-741 gone)"])

    def test_live_batch_gate_kept(self):
        led = R.Ledger(entries=[self.batch, self.gate()])
        self.assertEqual(R.prune(led, self.f), [])
        self.assertEqual(led.get("r-757").state, "queued")


class Game(unittest.TestCase):
    def setUp(self):
        self.cfg = R.Config()

    def test_slice_props(self):
        self.assertEqual(R.slice_props(self.cfg, R.Ledger(), facts()),
                         {"CPUWeight": "20", "IOWeight": "20", "MemoryHigh": str(24 * GIB)})
        on = R.Ledger(game_until=T(18))
        self.assertEqual(R.slice_props(self.cfg, on, facts(slice_gb=10.0))["MemoryHigh"], str(18 * GIB))
        self.assertEqual(R.slice_props(self.cfg, on, facts(slice_gb=20.0)),
                         {"CPUWeight": "5", "IOWeight": "5", "MemoryHigh": str(20 * GIB)})

    def test_shortfall(self):
        a = running("r-4", project="proj-a", title="dotnet e2e", mem=2.5, est=55, started=T(14, 10))   # ~15:05
        b = running("r-7", project="proj-b", title="r5 rebuild", mem=9.0, est=80, started=T(14, 10))    # ~15:30
        f = facts(available=11.5, units={"wf-r-4.service": R.Unit(True, 2.5), "wf-r-7.service": R.Unit(True, 9.0)})
        led = R.Ledger(game_until=T(18), entries=[a, b])
        # free for you 11.5 − 0 unused = 11.5 → short 0.5: r-4 alone frees 2.5
        self.assertEqual(R.shortfall_lines(self.cfg, led, f), [
            'short 0.5 GB of 12.0 GB: r-4 proj-a "dotnet e2e" 2.5 GB ~15:05',
            "full reserve free ~15:05 (est.); free now: wf res release r-4 --stop"])
        f.available_gb = 8.9          # short 3.1: needs both
        self.assertEqual(R.shortfall_lines(self.cfg, led, f), [
            'short 3.1 GB of 12.0 GB: r-4 proj-a "dotnet e2e" 2.5 GB ~15:05, r-7 proj-b "r5 rebuild" 9.0 GB ~15:30',
            "full reserve free ~15:30 (est.); free now: wf res release r-7 --stop"])
        f.available_gb = 13.0
        self.assertEqual(R.shortfall_lines(self.cfg, led, f), [])

    def test_shortfall_not_agents(self):
        self.assertEqual(R.shortfall_lines(self.cfg, R.Ledger(game_until=T(18)), facts(available=5.0)), [
            "short 7.0 GB of 12.0 GB: other programs, not agent jobs",
            "agent jobs alone cannot free it; close other programs"])

    def test_game_on_lines(self):
        led = R.Ledger(game_until=T(18, 10))
        self.assertEqual(R.game_on_lines(self.cfg, led, facts(available=20.0), 240),
                         ["game on until 18:10 (4h); CPU/IO now yours"])

    def test_status_shows_game(self):
        lines = R.status_lines(self.cfg, R.Ledger(game_until=T(18, 10)), facts(available=5.0))
        self.assertEqual(lines[1:], ["agents may use 0.0 GB, 8 cpus now; reserve 12.0 GB/8 cpus",
                                     "game on until 18:10 (4h left)",
                                     "short 7.0 GB of 12.0 GB: other programs, not agent jobs",
                                     "agent jobs alone cannot free it; close other programs"])

class Clean(unittest.TestCase):
    def test_scratch_victims(self):
        now = 10_000_000.0
        h = 3600
        tree = {
            "/s/old-file.png": (now - 3 * h, {}),
            "/s/-projects-a": (now - 3 * h, {"/s/-projects-a/s1": now - 3 * h}),
            "/s/-projects-b": (now - 60, {"/s/-projects-b/s1": now - 5 * h, "/s/-projects-b/s2": now - 60}),
            "/s/new.txt": (now - 60, {}),
        }
        self.assertEqual(R.scratch_victims(tree, now, 2.0),
                         ["/s/-projects-a", "/s/-projects-b/s1", "/s/old-file.png"])
        # live sessions keep their dir however idle; a stale top holding one is pruned per child
        self.assertEqual(R.scratch_victims(tree, now, 2.0, keep={"s1"}), ["/s/old-file.png"])
        tree["/s/-projects-a"] = (now - 3 * h, {"/s/-projects-a/s1": now - 3 * h, "/s/-projects-a/s9": now - 3 * h})
        self.assertEqual(R.scratch_victims(tree, now, 2.0, keep={"s1"}), ["/s/-projects-a/s9", "/s/old-file.png"])

    def test_live_session_id(self):
        text = '{"pid": 817151, "sessionId": "4ad140ab-69ca", "procStart": "31379851", "kind": "interactive"}'
        self.assertEqual(R.live_session_id(text, "31379851"), "4ad140ab-69ca")
        self.assertIsNone(R.live_session_id(text, "999"))               # pid reused by another process
        self.assertEqual(R.live_session_id('{"sessionId": "x"}', "5"), "x")   # no procStart recorded → trust pid
        self.assertIsNone(R.live_session_id("{oops", "5"))
        self.assertIsNone(R.live_session_id('{"pid": 1}', "5"))

    def test_cleanup_rule(self):
        self.assertEqual(R.cleanup_rule("out/prof"), ("out/prof", None))
        self.assertEqual(R.cleanup_rule("out/history-logs/*.log:30d"), ("out/history-logs/*.log", 30))
        for bad in ("/etc", "../x", "out/../../x"):
            with self.assertRaisesRegex(R.ResError, "cleanup pattern"):
                R.cleanup_rule(bad)

    def test_proc_name(self):
        # background sessions run the versioned binary: comm is the version, argv0 the path or `claude`
        self.assertEqual(R.proc_name("2.1.283", "claude"), "claude")
        self.assertEqual(R.proc_name("2.1.283", "claude bg-pty-host --bg-pty-host /tmp/x.sock"), "claude")  # rewritten title
        self.assertEqual(R.proc_name("2.1.283", "/h/u/.local/share/claude/versions/2.1.283"), "claude")
        self.assertEqual(R.proc_name("claude", "/h/u/.local/bin/claude"), "claude")
        self.assertEqual(R.proc_name("2.1.284", "ugrep"), "2.1.284")    # tool re-exec of the binary: not a session
        self.assertEqual(R.proc_name("bash", "/bin/bash"), "bash")
        self.assertEqual(R.proc_name("x", ""), "x")                     # kernel thread / unreadable cmdline

    def test_adopt_groups(self):
        procs = [
            (100, 1, "bash", "/u/app.slice/tab1.scope"),
            (101, 100, "claude", "/u/app.slice/tab1.scope"),      # root, outside
            (102, 101, "bash", "/u/app.slice/tab1.scope"),        # its shell
            (103, 102, "make", "/u/app.slice/tab1.scope"),
            (104, 101, "claude", "/u/app.slice/tab1.scope"),      # child claude: not a root, but a descendant
            (105, 101, "job", "/u/agents.slice/agents-jobs.slice/wf-r-1.service"),   # already inside: skipped
            (200, 1, "claude", "/u/agents.slice/run-9.scope"),    # root already inside
            (300, 1, "vim", "/u/app.slice/tab2.scope"),
        ]
        self.assertEqual(R.adopt_groups(procs), {101: [101, 102, 103, 104]})

    def test_warning_lines(self):
        self.assertEqual(R.warning_lines(30.0, 7.6, 2), [
            "warning: /tmp (RAM) holds 7.6 GB; wf res clean",
            "warning: 2 claude sessions outside agents.slice (wf res adopt)"])
        self.assertEqual(R.warning_lines(30.0, 7.4, 0), [])

class Units(unittest.TestCase):
    def test_adopt_argv(self):
        self.assertEqual(R.adopt_argv(101, [101, 102]), [
            "busctl", "--user", "call", "org.freedesktop.systemd1", "/org/freedesktop/systemd1",
            "org.freedesktop.systemd1.Manager", "StartTransientUnit", "ssa(sv)a(sa(sv))",
            "wf-claude-101.scope", "fail", "2", "PIDs", "au", "2", "101", "102", "Slice", "s", "agents.slice", "0"])

    def test_texts(self):
        self.assertEqual(R.service_text("/usr/bin/python3", "/projects/public/workflow/wf.py"),
                         "[Unit]\nDescription=wf res tick\n\n[Service]\nType=oneshot\n"
                         "ExecStart=/usr/bin/python3 /projects/public/workflow/wf.py res tick\n")
        self.assertEqual(R.timer_text(),
                         "[Unit]\nDescription=wf res tick every minute\n\n[Timer]\nOnBootSec=1min\n"
                         "OnUnitActiveSec=1min\n\n[Install]\nWantedBy=timers.target\n")
        self.assertEqual(R.shell_init_line(),
                         "alias claude='systemd-run --user --scope --quiet --slice=agents.slice claude'")


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"
        self.assertEqual(R.psi_some_avg60(text), 33.2)
        self.assertIsNone(R.psi_some_avg60(""))

    def test_lines(self):
        hot = running("r-1", mem=4.0)            # 3.5 of 4.0 GB (≥ 0.85×), stall 33% → warn
        cache = running("r-2", mem=4.0)          # full of page cache, no stall → quiet
        small = running("r-3", mem=10.0)         # stalled but far below its limit (machine pressure) → quiet
        f = facts(units={"wf-r-1.service": R.Unit(True, 3.5, 33.2), "wf-r-2.service": R.Unit(True, 3.9, 0.0),
                         "wf-r-3.service": R.Unit(True, 2.0, 50.0)})
        self.assertEqual(R.throttle_lines(R.Ledger(entries=[hot, cache, small]), f), [
            "r-1 throttled at its memory limit (3.5 of 4.0 GB, stalled 33% of the last minute): "
            "likely too small; `wf res release r-1 --stop` and re-run with a bigger --mem"])


class Owner(unittest.TestCase):
    REC = [{"lane": "fast", "pid": 11, "socket": "/s/a"}, {"lane": "slow", "pid": 22, "socket": "/s/b"}]

    def test_from_env_branch_and_session_record(self):
        env = {"CLAUDE_PID": "22", "CLAUDE_CODE_MESSAGING_SOCKET": "/s/b", "WF_RES_ID": "r-7"}
        self.assertEqual(R.owner_by(env, "slow/t-res-owner", self.REC),
                         {"name": "slow session", "task": "t-res-owner", "batch": "r-7", "address": "uds:/s/b"})

    def test_explicit_name_wins_then_env(self):
        env = {"CLAUDE_PID": "22", "WF_SESSION_NAME": "pilot", "WF_TASK": "t-x"}
        self.assertEqual(R.owner_by(env, "slow/t-y", self.REC, "worker-3"), {"name": "worker-3", "task": "t-x"})
        self.assertEqual(R.owner_by(env, "", self.REC), {"name": "pilot", "task": "t-x"})

    def test_nothing_known(self):
        self.assertEqual(R.owner_by({"CLAUDE_PID": "99"}, "master", self.REC), {})
        self.assertEqual(R.owner_by({}, "feature-x", [{"pid": 1}]), {})

    def test_status_and_throttle_name_owner(self):
        e = running("r-1", mem=4.0)
        e.by = {"name": "slow session", "task": "t-a", "batch": "r-7", "address": "uds:/s/b"}
        f = facts(units={"wf-r-1.service": R.Unit(True, 3.5, 33.2)})
        led = R.Ledger(entries=[e])
        self.assertTrue(R.status_lines(R.Config(), led, f)[0].endswith(
            " [by slow session t-a batch r-7, message uds:/s/b]"))
        self.assertTrue(R.throttle_lines(led, f)[0].endswith(
            "bigger --mem; started by slow session t-a batch r-7, message uds:/s/b"))

    def test_ledger_roundtrip_and_old_entries(self):
        e = running("r-1")
        e.by = {"task": "t-a"}
        self.assertEqual(R.loads(R.dumps(R.Ledger(entries=[e]))).get("r-1").by, {"task": "t-a"})
        d = json.loads(R.dumps(R.Ledger(entries=[running("r-2")])))
        for k in ("by", "lock"):                        # entries written before these fields
            del d["entries"][0][k]
        old = R.loads(json.dumps(d)).get("r-2")
        self.assertEqual((old.by, old.lock), ({}, ""))


class BareUnits(unittest.TestCase):
    def test_bare_units(self):
        blocks = R.show_units(
            "Id=wh-t27.service\nTransient=yes\nSlice=app.slice\nMemoryCurrent=2147483648\n\n"
            "Id=run-u42.service\nTransient=yes\nSlice=app.slice\nMemoryCurrent=[not set]\n\n"
            "Id=app-org.kde.konsole@65c0.service\nTransient=yes\nSlice=app.slice\nMemoryCurrent=1\n\n"
            "Id=dbus-:1.1-org.kde.kwalletd6@0.service\nTransient=yes\nSlice=app.slice\nMemoryCurrent=1\n\n"
            "Id=wf-r-3.service\nTransient=yes\nSlice=agents-jobs.slice\nMemoryCurrent=1\n\n"
            "Id=pipewire.service\nTransient=no\nSlice=session.slice\nMemoryCurrent=1\n")
        self.assertEqual(R.bare_units(blocks), [("run-u42.service", None), ("wh-t27.service", 2.0)])

    def test_warning(self):
        self.assertEqual(R.warning_lines(30.0, 0.0, 0, [("wh-t27.service", 2.0), ("run-u42.service", None)]), [
            "warning: 2 jobs outside wf res (bare systemd-run): wh-t27.service 2.0 GB, run-u42.service ? GB; "
            "start jobs with wf res run, also from project scripts"])


class Hook(unittest.TestCase):
    BASH = {"tool_name": "Bash", "tool_input": {"command": "dotnet run --project Desktop", "timeout": 5000}}

    def test_gaming_hides_display(self):
        out = R.hook_output(self.BASH, T(16, 0), NOW)
        self.assertEqual(out, {"hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "updatedInput": {"command": "unset DISPLAY WAYLAND_DISPLAY; dotnet run --project Desktop", "timeout": 5000},
            "additionalContext": "wf res game mode until 16:00: this command has no display, so GUI windows fail. "
                                 "Run headless/offscreen, or do other work until game off."}})

    def test_quiet_when_not_gaming_or_not_bash(self):
        self.assertIsNone(R.hook_output(self.BASH, None, NOW))
        self.assertIsNone(R.hook_output(self.BASH, T(14, 0), NOW))          # expired
        self.assertIsNone(R.hook_output({"tool_name": "Read", "tool_input": {"file_path": "x"}}, T(16, 0), NOW))
        self.assertIsNone(R.hook_output({"tool_name": "Bash", "tool_input": {}}, T(16, 0), NOW))


class Litter(unittest.TestCase):
    def test_victims(self):
        now, h = 10_000_000.0, 3600
        entries = [("/t/aB3xYz", "emptydir", now - 3 * h),                  # old empty dir → go
                   ("/t/MSBuildTemp1000", "emptydir", now - 5 * h),          # → go
                   ("/t/fresh", "emptydir", now - 60),                       # too new
                   ("/t/full", "dir", now - 9 * h),                          # has content → never
                   ("/t/clr-debug-pipe-111-222-in", "other", now - 60),     # pid 111 dead → go
                   ("/t/dotnet-diagnostic-333-444-socket", "other", now),    # pid 333 alive
                   ("/t/notes.txt", "file", now - 9 * h)]                    # file → never
        alive = {333}.__contains__
        self.assertEqual(R.litter_victims(entries, now, 2.0, alive),
                         ["/t/MSBuildTemp1000", "/t/aB3xYz", "/t/clr-debug-pipe-111-222-in"])
class SessionCap(unittest.TestCase):
    BLOCKS = {
        "wf-claude-1.scope": {"Slice": "agents.slice", "MemoryHigh": "infinity", "MemoryCurrent": str(GIB)},
        "wf-claude-2.scope": {"Slice": "agents.slice", "MemoryHigh": str(10 * GIB), "MemoryCurrent": str(GIB)},
        "run-u7.scope": {"Slice": "agents.slice", "MemoryHigh": str(4 * GIB), "MemoryCurrent": str(GIB)},
        "podman-pause.scope": {"Slice": "user.slice", "MemoryHigh": "infinity", "MemoryCurrent": str(GIB)},
    }

    def test_caps_to_set(self):
        self.assertEqual(R.session_caps(self.BLOCKS, 10.0),
                         [("run-u7.scope", str(10 * GIB)), ("wf-claude-1.scope", str(10 * GIB))])
        self.assertEqual(R.session_caps(self.BLOCKS, 0.0),          # 0 = off → lift caps
                         [("run-u7.scope", "infinity"), ("wf-claude-2.scope", "infinity")])

    def test_warn_at_cap(self):
        blocks = {"wf-claude-5.scope": {"Slice": "agents.slice", "MemoryCurrent": str(int(9.6 * GIB))},
                  "wf-claude-6.scope": {"Slice": "agents.slice", "MemoryCurrent": str(int(9.8 * GIB))},
                  "wf-claude-7.scope": {"Slice": "agents.slice", "MemoryCurrent": str(int(2 * GIB))}}
        stall = {"wf-claude-5.scope": 45.0, "wf-claude-6.scope": 1.0, "wf-claude-7.scope": 90.0}
        self.assertEqual(R.session_cap_lines(blocks, stall, 10.0), [
            "warning: claude session 5 at its memory cap (9.6 of 10.0 GB, stalled 45% of the last minute): "
            "run big work with wf res run"])
        self.assertEqual(R.session_cap_lines(blocks, stall, 0.0), [])


def run(title="gate 30735e0", peak=1.0, minutes=10.0, project="proj", mem=12.0, est=60, rc=0, id="r-1"):
    return {"id": id, "project": project, "title": title, "mem_gb": mem, "est_min": est, "rc": rc,
            "peak_gb": peak, "min": minutes}


TEN = [run(id=f"r-{i}", title=f"gate {i:07x}a", peak=float(i), minutes=6.0 * i) for i in range(1, 11)]


class History(unittest.TestCase):
    def test_title_kind(self):
        for title, kind in (("gate 30735e0", "gate"), ("gate 2c91cac7 (fix optimizer test)", "gate"),
                            ("mem-profile 16x5M (t-optimizer-x)", "mem-profile 16x5M"), ("wf-batch", "wf-batch"),
                            ("gate 97e6044b direct", "gate 97e6044b direct"), ("build deadbeef", "build deadbeef"),
                            ("gate a3cbde1 0cf9a79", "gate"), ("abc1234", "abc1234"), ("  x  ", "x")):
            self.assertEqual(R.title_kind(title), kind, title)

    def test_pct_nearest_rank(self):
        xs = [5.0, 1.0, 3.0, 2.0, 4.0]
        self.assertEqual([R.pct(xs, p) for p in (0.5, 0.9, 0.95, 0.2)], [3.0, 5.0, 5.0, 1.0])

    def test_history_record(self):
        e = running(project="p", title="gate x", mem=10.0, est=40)
        e.state, e.ended, e.rc, e.peak_gb = "done", T(14, 13, ), 0, 9.1
        e.ended = e.started + dt.timedelta(minutes=13, seconds=30)
        self.assertEqual(R.history_record(e), {"id": "r-4", "project": "p", "title": "gate x", "mem_gb": 10.0,
                                               "est_min": 40, "rc": 0, "peak_gb": 9.1, "min": 13.5})
        for change in ({"rc": 1}, {"peak_gb": None}, {"state": "running"}, {"ended": None}):
            bad = R.Entry(**{**{f.name: getattr(e, f.name) for f in R.fields(R.Entry)}, **change})
            self.assertIsNone(R.history_record(bad), change)

    def test_suggest_p95_p90(self):
        # peaks 1..10 → p95 = 10 → ×1.15 = 11.5; durations 6..60 → p90 = 54 → ×1.5 = 81
        self.assertEqual(R.suggest(TEN, "proj", "gate"), (11.5, 81, 10))

    def test_suggest_floors_and_rounding(self):
        tiny = [run(id=f"r-{i}", peak=0.01, minutes=1.0) for i in range(3)]
        self.assertEqual(R.suggest(tiny, "proj", "gate"), (0.2, 5, 3))
        odd = [run(id=f"r-{i}", peak=1.01, minutes=7.1) for i in range(3)]
        self.assertEqual(R.suggest(odd, "proj", "gate"), (1.2, 11, 3))     # 1.1615 → 1.2 up; 10.65 → 11 up

    def test_suggest_needs_three_matching(self):
        two = TEN[:2] + [run(id="r-a", project="other"), run(id="r-b", title="build"),
                         run(id="r-c", rc=1), run(id="r-d", peak=None)]
        self.assertIsNone(R.suggest(two, "proj", "gate"))
        self.assertIsNone(R.suggest([], "proj", "gate"))

    def test_hint_line(self):
        sug = (11.5, 81, 10)
        line = "hint: history says ~11.5 GB / 81 min (10 runs)"
        self.assertEqual(R.hint_line(23.1, 60, sug), line)
        self.assertEqual(R.hint_line(10.0, 163, sug), line)
        self.assertIsNone(R.hint_line(23.0, 162, sug))
        self.assertIsNone(R.hint_line(99.0, 999, None))

    def test_merge_runs(self):
        led = R.Ledger(entries=[running(id="r-4", project="proj", title="gate 1234567"), running(id="r-5")])
        led.entries[0].state, led.entries[0].rc, led.entries[0].peak_gb = "done", 0, 2.0
        led.entries[0].ended = led.entries[0].started + dt.timedelta(minutes=3)
        merged = R.all_runs([run(id="r-1"), run(id="r-4", peak=7.0)], led)
        self.assertEqual([(r["id"], r["peak_gb"]) for r in merged], [("r-1", 1.0), ("r-4", 7.0)])
        merged = R.all_runs([run(id="r-1")], led)
        self.assertEqual([(r["id"], r["peak_gb"], r["min"]) for r in merged], [("r-1", 1.0, 10.0), ("r-4", 2.0, 3.0)])

    def test_hist_lines(self):
        runs = TEN + [run(id="r-x", title="wf-batch", peak=3.0, minutes=20.0, mem=6.0, est=400),
                      run(id="r-y", project="other", title="build")]
        self.assertEqual(R.hist_lines(runs, "proj"), [
            "proj · gate · n 10 · req 12.0 GB · peak 5.0/10.0 GB · est 1h · dur 30m/54m · suggest 11.5 GB 1h21m",
            "proj · wf-batch · n 1 · req 6.0 GB · peak 3.0/3.0 GB · est 6h40m · dur 20m/20m · suggest - (< 3 runs)"])
        self.assertEqual(len(R.hist_lines(runs, None)), 3)
        self.assertEqual(R.hist_lines([], None), ["no finished runs with a peak yet"])


ORCH_LOG = """\
2026-10-06T10:00:00 slow opus t-a done abc1234 10m00s
2026-10-06T10:10:00 fast haiku t-b handback - 20m30s
2026-10-06T10:20:00 slow opus t-c done+gate-red abc1234 1h05m
2026-10-06T10:30:00 fast haiku t-d post-check-red abc1234 50m00s (branch x still there)
2026-10-06T10:40:00 fast haiku t-e done abc1234 0m00s
2026-10-06T10:50:00 slow opus t-f done abc1234 -
2026-10-06T11:00:00 cloud opus t-g done abc1234 3h00m
2026-10-06T11:10:00 slow sonnet t-h awaiting - 40m00s
21:30 pilot batch 1 rc=0 done=t-a stopped= added=
ALERT wf-pilot proj: awaiting a-1
2026-10-06T11:20:00 slow opus t-i done abc1234 30m00s
"""


class TaskFit(unittest.TestCase):
    def test_durations_from_log(self):
        # done/handback rows of local lanes with a real duration only (minutes)
        self.assertEqual(R.orch_durations(ORCH_LOG), [10.0, 20.5, 65.0, 30.0])

    def test_p90(self):
        # 4 rows: nearest rank ceil(0.9*4)=4 → 65 min
        self.assertEqual(R.task_p90(ORCH_LOG), (65, 4))
        self.assertEqual(R.task_p90(""), (30, 0))
        two = "2026-10-06T10:00:00 slow opus t-a done abc 5m00s\n" * 2
        self.assertEqual(R.task_p90(two), (30, 2))             # < 3 runs → default 30 min
        odd = "2026-10-06T10:00:00 slow opus t-a done abc 4m10s\n" * 3
        self.assertEqual(R.task_p90(odd), (5, 3))              # rounded up to whole minutes

    def test_batch_fit(self):
        self.assertEqual(R.batch_fit(4, 150, 30), 4)
        self.assertEqual(R.batch_fit(4, 90, 30), 3)
        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()