import json import os import sys import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from wflib import tasks as T from wflib import lanes as L from test_cli import Cli from test_check import Base, CLEAN SESS = """\ # Tasks — demo ## Awaiting your decision ## Pending - **t-solo** [P1] (1h): Solo. - Sessions: solo — lib bump Model: sonnet - **t-own** [P1] (1h, interactive): Own. - **t-owner** [P2] (1h): Owner. Sessions: owner - **t-par** [P2] (<1h): Par. Sessions: parallel - **t-plain** [P3] (<1h): Plain. ## Needs human ## Deferred """ class SessionsLineTest(unittest.TestCase): def setUp(self): self.doc = T.parse(SESS) def test_values(self): self.assertEqual([(i.id, i.sessions) for i in self.doc.section("pending").items], [("t-solo", "solo"), ("t-own", "owner"), ("t-owner", "owner"), ("t-par", "parallel"), ("t-plain", "parallel")]) def test_set_replaces_in_place_and_keeps_tail_order(self): T.set_fields(self.doc, "t-solo", sessions="owner") self.assertEqual(self.doc.item("t-solo").body, [" Sessions: owner", " Model: sonnet"]) def test_set_adds_before_model(self): T.set_fields(self.doc, "t-plain", sessions="solo") self.assertEqual(self.doc.item("t-plain").body, [" Sessions: solo"]) def test_set_empty_removes(self): T.set_fields(self.doc, "t-owner", sessions="") self.assertEqual(self.doc.item("t-owner").body, []) def test_set_drops_interactive_flag(self): T.set_fields(self.doc, "t-own", sessions="owner") item = self.doc.item("t-own") self.assertEqual(item.lines(), ["- **t-own** [P1] (1h): Own.", " Sessions: owner"]) def test_set_bad_value(self): with self.assertRaises(T.TaskError): T.set_fields(self.doc, "t-plain", sessions="many") def test_note_goes_before_sessions_line(self): T.add_note(self.doc, "t-owner", "x") self.assertEqual(self.doc.item("t-owner").body, [" - x", " Sessions: owner"]) class SessionsPickTest(unittest.TestCase): def setUp(self): self.doc = T.parse(SESS) def pick(self, lane, model, **kw): return L.pick(self.doc, set(), L.DEFAULT_LANES, "1h", lane, model, **kw) def test_alone_solo_picked_owner_skipped(self): item, skipped = self.pick(None, "sonnet") self.assertEqual(item.id, "t-solo") item, skipped = self.pick("slow", "opus", others=1) self.assertEqual((item.id, [(i.id, why) for i, why in skipped]), ("t-par", [("t-solo", "solo: 1 other live session"), ("t-own", "owner: needs the owner (wf next --owner)"), ("t-owner", "owner: needs the owner (wf next --owner)")])) def test_owner_present(self): item, _ = self.pick("slow", "opus", others=1, owner=True) self.assertEqual(item.id, "t-own") def test_solo_skipped_with_other_live_sessions(self): item, skipped = self.pick(None, "sonnet", others=1) self.assertEqual((item, [(i.id, why) for i, why in skipped]), (None, [("t-solo", "solo: 1 other live session")])) _, skipped = self.pick(None, "sonnet", others=2) self.assertEqual(skipped[0][1], "solo: 2 other live sessions") def test_solo_running(self): self.assertEqual(T.solo_running(self.doc, {"t-solo": "sonnet session uds:/a"}), ("t-solo", "sonnet session uds:/a")) self.assertIsNone(T.solo_running(self.doc, {"t-par": "x"})) self.assertIsNone(T.solo_running(self.doc, {})) def test_solo_done_block(self): sessions = {"opus": {"socket": "/o", "alive": True, "pid": 1}, "haiku": {"socket": "/h", "alive": False, "pid": 2}, "sonnet": {"socket": "/s", "alive": True, "pid": 3}} self.assertEqual(L.solo_done_block(["t-solo"], sessions, "3"), ["notify opus uds:/o: solo t-solo done, run wf next"]) self.assertEqual(L.solo_done_block([], sessions, "3"), []) class SessionsCheckTest(Base): def test_bad_word_error_interactive_warning(self): text = CLEAN.replace("- **t-one** [P1] (1h)", "- **t-one** [P1] (1h, interactive)").replace( "## Needs human", "- **t-x** [P3] (1h): X.\n Sessions: lots\n\n## Needs human") self.tasks(text) errors, warnings = self.run_check() self.assertTrue(any(e.endswith("t-x: Sessions 'lots' (want parallel, solo, owner)") for e in errors), errors) self.assertTrue(any(w.endswith("t-one: 'interactive' flag: write 'Sessions: owner' " "(wf set t-one --sessions owner)") for w in warnings), warnings) class SessionsCliTest(Cli): tasks_text = SESS def env(self, name, pid): sock = self.root / f"{name}.sock" sock.write_text("") return {"CLAUDE_CODE_MESSAGING_SOCKET": str(sock), "CLAUDE_PID": str(pid), "CLAUDE_CODE_SESSION_ID": name} def register(self, lane, model, sock, pid): d = self.root / ".wf" / "sessions" d.mkdir(parents=True, exist_ok=True) (d / f"{lane}.json").write_text(json.dumps({"lane": lane, "model": model, "socket": str(sock), "pid": pid, "session": "s", "at": "2026-10-04T10:00"})) def test_list_marks(self): self.assertEqual(self.ok("list"), "t-solo P1 1h - sonnet slow [solo] Solo\n" "t-own P1 1h - opus slow [owner] Own\n" "t-owner P2 1h - opus slow [owner] Owner\n" "t-par P2 <1h - opus fast Par\n" "t-plain P3 <1h - opus fast Plain\n" "pending 5 · human 0 · awaiting 0 · deferred 0\n") def test_add_and_set(self): self.ok("add", "Four.", "-p", "3", "-e", "1h", "--sessions", "solo", "--model", "haiku") self.assertEqual(self.item("t-four"), "- **t-four** [P3] (1h): Four.\n Sessions: solo\n Model: haiku\n") self.ok("set", "t-four", "--sessions", "") self.assertEqual(self.item("t-four"), "- **t-four** [P3] (1h): Four.\n Model: haiku\n") self.fails("set", "t-four", "--sessions", "many", code=2) self.fails("add", "Five.", "-p", "3", "-e", "1h", "--sessions", "many", code=2) def test_interactive_option_is_old_spelling(self): code, out, err = self.wf("add", "Four.", "-p", "3", "-e", "1h", "--interactive") self.assertEqual((code, err), (0, "wf: --interactive is now --sessions owner\n")) self.assertEqual(self.item("t-four"), "- **t-four** [P3] (1h): Four.\n Sessions: owner\n") code, out, err = self.wf("set", "t-plain", "--interactive", "yes") self.assertEqual((code, err), (0, "wf: --interactive is now --sessions owner\n")) self.assertEqual(self.item("t-plain"), "- **t-plain** [P3] (<1h): Plain.\n Sessions: owner\n") def test_next_owner(self): self.ok("done", "t-solo", "-m", "ok") out = self.ok("next", "--lane", "slow", "--as", "opus", env={"CLAUDE_PID": "", "CLAUDE_CODE_MESSAGING_SOCKET": ""}) self.assertIn("- t-own: owner: needs the owner (wf next --owner)\n", out) self.assertIn("===== Next task =====\n- **t-par**", out) # slow empty → fallback fast self.assertTrue(self.ok("next", "--lane", "slow", "--as", "opus", "--owner", "--brief").startswith("- **t-own**")) def test_next_solo_with_other_live_session(self): sock = self.root / "o.sock" sock.write_text("") self.register("fast", "opus", sock, os.getppid()) err = self.fails("next", "--as", "sonnet", env=self.env("s", os.getpid())) self.assertEqual(err, "wf: nothing pickable for all lanes (sonnet) in Pending\n") out = self.wf("next", "--as", "sonnet", env=self.env("s", os.getpid()))[1] self.assertIn("- t-solo: solo: 1 other live session\n", out) def test_solo_in_progress_blocks_others_and_done_notifies(self): s, o = self.env("s", os.getpid()), self.env("o", os.getppid()) self.ok("next", "--lane", "slow", "--as", "sonnet", "--brief", env=s) self.ok("status", "t-solo", "progress", "x", env=s) code, out, err = self.wf("next", "--lane", "fast", "--as", "opus", env=o) self.assertEqual((code, err), (1, f"wf: solo t-solo in progress by sonnet session uds:{self.root / 's.sock'}: " "wait (its wf done notifies you)\n")) self.assertNotIn("Next task", out) out = self.ok("done", "t-solo", "-m", "ok", env=s) self.assertIn(f"notify fast uds:{self.root / 'o.sock'}: solo t-solo done, run wf next\n", out) self.assertTrue(self.ok("next", "--lane", "fast", "--as", "opus", "--brief", env=o).startswith("- **t-par**")) if __name__ == "__main__": unittest.main()