import json import unittest from tests.fakes import Harness, config_data class CoreTest(unittest.TestCase): def setUp(self): self.h = Harness() self.p = self.h.plane def tearDown(self): self.h.close() def codex(self): return self.h.execs[("win10", "codex")] def test_pick_starts_lowest_todo_and_marks_in_progress(self): a, b = self.p.add_issue(7), self.p.add_issue(3) self.p.add_issue(1, state="Backlog") self.p.add_issue(2, assignee="kappiee") # not routed self.h.tick() runs = self.codex().started self.assertEqual([r.run_id for r in runs], ["GW-3-1"]) # concurrency 1 on win10 self.assertEqual(self.p.state_of(b), "In Progress") self.assertEqual(self.p.state_of(a), "Todo") self.assertTrue(self.p.texts(b)[-1].startswith("[worker] 第 1 轮开始：codex / sol / medium @ win10，新线程")) self.assertIsNone(runs[0].thread_id) self.assertIn("do it", runs[0].prompt) def test_review_question_progress_transitions(self): for tag, state, who in (("review", "Review", "oregon"), ("question", "Waiting", "kappiee"), ("blocked", "Waiting", "kappiee"), ("progress", "Todo", "window10")): with self.subTest(tag=tag): h = Harness() iid = h.plane.add_issue(1) h.tick() ex = h.execs[("win10", "codex")] ex.finish("[%s] text" % tag) h.tick() if tag == "progress": # picked again in the same tick, resuming the thread self.assertEqual(len(ex.started), 2) self.assertEqual(ex.started[1].thread_id, "th-1") self.assertEqual(h.plane.state_of(iid), "In Progress") self.assertIn(("i1", "s-todo", ["u-w10"]), h.plane.patches) else: self.assertEqual((h.plane.state_of(iid), h.plane.assignee_of(iid)), (state, who)) texts = h.plane.texts(iid) result = [t for t in texts if t.startswith("[codex]")][0] self.assertIn("第 1 轮结束（exit=0，run=GW-1-1，终端 win10", result) self.assertIn("[%s] text" % tag, result) self.assertEqual(h.rec(iid)["tags"], [tag]) h.close() def test_untagged_message_becomes_review(self): iid = self.p.add_issue(1) self.h.tick() self.codex().finish("all done") self.h.tick() self.assertEqual(self.p.state_of(iid), "Review") self.assertIn("未按协议标注首行", self.p.texts(iid)[-1]) def test_resume_sends_only_new_comments(self): iid = self.p.add_issue(1) self.p.add_comment(iid, "

old context

") self.h.tick() self.codex().finish("[question] A or B?") self.h.tick() self.p.add_comment(iid, "

B please

") self.p.issues[iid].update(state="s-todo", assignees=["u-w10"]) self.h.tick() run = self.codex().started[-1] self.assertEqual(run.thread_id, "th-1") self.assertIn("B please", run.prompt) self.assertNotIn("old context", run.prompt) self.assertNotIn("A or B?", run.prompt) self.assertIn("续线程 th-1", self.p.texts(iid)[-1]) def test_transient_failure_retries_with_backoff_then_waits(self): iid = self.p.add_issue(1) self.h.tick() self.codex().finish("", exit_code=1, tail="Connection reset by peer") self.h.tick() note = self.p.texts(iid)[-1] self.assertIn("类型=crash（临时性）", note) self.assertIn("重试 1/2", note) self.assertEqual(self.p.state_of(iid), "In Progress") self.assertEqual(len(self.codex().started), 1) # backoff 2 min not elapsed self.h.clock[0] += 121 self.h.tick() self.assertEqual(len(self.codex().started), 2) self.assertEqual(self.codex().started[1].thread_id, "th-1") # resumes the original thread self.assertIn("临时性错误中断", self.codex().started[1].prompt) self.codex().finish("", exit_code=1, tail="HTTP 503 Service Unavailable") self.h.tick() self.h.clock[0] += 301 self.h.tick() self.assertEqual(len(self.codex().started), 3) self.codex().finish("", exit_code=1, tail="banner exchange failed") self.h.tick() self.assertEqual((self.p.state_of(iid), self.p.assignee_of(iid)), ("Waiting", "kappiee")) self.assertIn("不再自动重试", self.p.texts(iid)[-1]) self.assertEqual(self.h.rec(iid)["retries"], 0) def test_logic_failure_on_fresh_thread_goes_waiting(self): iid = self.p.add_issue(1) self.h.tick() self.codex().finish("", exit_code=2, thread=None, tail="panic: bad config") self.h.tick() self.assertEqual(self.p.state_of(iid), "Waiting") self.assertIn("类型=crash，run=GW-1-1，exit=2，重试 0/2", self.p.texts(iid)[-1]) self.assertIn("panic: bad config", self.p.texts(iid)[-1]) def test_resume_failure_rebuilds_thread_with_handoff(self): iid = self.p.add_issue(1) self.h.tick() self.codex().finish("[progress] half way") self.h.tick() # finishes round 1, starts round 2 resuming th-1 self.assertEqual(self.codex().started[1].thread_id, "th-1") self.codex().finish("", exit_code=1, thread=None, tail="thread not found") self.h.tick() # round 2 fails -> rebuild immediately run = self.codex().started[2] self.assertIsNone(run.thread_id) self.assertIn("## 交接摘要", run.prompt) self.assertIn("续线程失败", run.prompt) self.assertIn("[progress] half way", run.prompt) self.assertIn("新线程（重建：续线程失败（crash），附交接摘要）", self.p.texts(iid)[-1]) def test_thread_rebuilt_after_max_rounds(self): iid = self.p.add_issue(1) self.h.tick() for _ in range(3): self.codex().finish("[progress] more") self.h.tick() runs = self.codex().started self.assertEqual([r.thread_id for r in runs], [None, "th-1", "th-1", None]) self.assertIn("线程已续 3 轮（上限 3）", runs[3].prompt) self.assertIn("progress, progress, progress", runs[3].prompt) self.assertEqual(self.h.rec(iid)["thread_rounds"], 1) def test_timeout_kills_and_waits(self): iid = self.p.add_issue(1, labels=["budget:30m"]) self.h.tick() self.h.clock[0] += 29 * 60 self.h.tick() self.assertEqual(self.codex().killed, []) self.h.clock[0] += 2 * 60 self.h.tick() self.assertEqual(self.codex().killed, ["GW-1-1"]) self.assertEqual(self.p.state_of(iid), "Waiting") self.assertIn("类型=timeout", self.p.texts(iid)[-1]) self.assertIn("0.5 h", self.p.texts(iid)[-1]) def test_serialize_group_and_concurrency(self): a = self.p.add_issue(1, assignee="oregon", labels=["needs-game"]) b = self.p.add_issue(2, assignee="oregon", labels=["needs-game"]) c = self.p.add_issue(3, assignee="oregon") d = self.p.add_issue(4, assignee="oregon") self.h.tick() started = [r.run_id for r in self.h.execs[("oregon", "claude-code")].started] self.assertEqual(started, ["GW-1-1", "GW-3-1"]) # 2 slots; GW-2 waits for the game group self.assertEqual([self.p.state_of(x) for x in (a, b, c, d)], ["In Progress", "Todo", "In Progress", "Todo"]) def test_concurrency_zero_pauses_terminal(self): self.h.cfg.terminals["win10"]["concurrency"] = 0 w = self.p.add_issue(1) o = self.p.add_issue(2, assignee="oregon") self.h.tick() self.assertNotIn(("win10", "codex"), self.h.execs) # not even probed (busy_processes) self.assertEqual((self.p.state_of(w), self.p.state_of(o)), ("Todo", "In Progress")) def test_label_routing(self): self.p.add_issue(1, labels=["model:astra"]) self.p.add_issue(2, assignee="oregon", labels=["cap:gpu"]) # oregon lacks the capability self.h.tick() run = self.codex().started[0] self.assertEqual((run.model, run.effort), ("astra", "high")) self.assertNotIn(("oregon", "claude-code"), self.h.execs) def test_human_moving_item_during_run_is_respected(self): iid = self.p.add_issue(1) self.h.tick() self.p.issues[iid]["state"] = "s-done" self.codex().finish("[review] ok") self.h.tick() self.assertEqual(self.p.state_of(iid), "Done") self.assertIn("[review] ok", self.p.texts(iid)[-1]) def test_post_failure_keeps_run_in_flight(self): iid = self.p.add_issue(1) self.h.tick() self.codex().finish("[review] ok") orig = self.p.post_comment self.p.post_comment = lambda *a: (_ for _ in ()).throw(RuntimeError("plane down")) self.h.tick() self.assertIn("inflight", self.h.rec(iid)) self.p.post_comment = orig self.h.tick() self.assertNotIn("inflight", self.h.rec(iid)) self.assertEqual(self.p.state_of(iid), "Review") def test_heartbeat_written(self): self.p.add_issue(1) self.h.tick() import json, pathlib hb = json.loads((pathlib.Path(self.h.tmp.name) / "heartbeat.json").read_text()) self.assertEqual(hb["inflight"][0]["run"], "GW-1-1") self.assertEqual(hb["errors"], []) # ---------------------------------------------------------------- reviewer decisions / fixed defects def test_item_moved_out_of_in_progress_cancels_run(self): a = self.p.add_issue(1, labels=["needs-game"]) b = self.p.add_issue(2, labels=["needs-game"]) self.h.tick() self.codex().finish("[progress] step 1") self.h.tick() # round 2 resumes th-1 self.assertEqual(self.codex().started[-1].run_id, "GW-1-2") self.p.issues[a]["state"] = "s-wait" # a person moves it while round 2 runs self.h.tick() self.assertEqual(self.codex().killed, ["GW-1-2"]) self.assertIn("[worker] 运行已取消：工作项状态被改为 Waiting（run=GW-1-2）", self.p.texts(a)[-1]) rec = self.h.rec(a) self.assertNotIn("inflight", rec) self.assertEqual((rec["thread_id"], rec["thread_on"]), ("th-1", ["win10", "codex"])) self.assertEqual(self.p.state_of(a), "Waiting") # agentd does not touch the state # terminal slot (concurrency 1) and serialize group were released in the same tick self.assertEqual(self.codex().started[-1].run_id, "GW-2-1") self.assertEqual(self.p.state_of(b), "In Progress") # back to Todo later: resumes the kept thread self.codex().finish("[review] b done") self.p.issues[a]["state"] = "s-todo" self.h.tick() self.assertEqual((self.codex().started[-1].run_id, self.codex().started[-1].thread_id), ("GW-1-3", "th-1")) def test_cancel_keeps_slot_when_kill_fails(self): a = self.p.add_issue(1) self.p.add_issue(2) self.h.tick() self.codex().fail_kills = 1 self.p.issues[a]["state"] = "s-bl" self.h.tick() self.assertIn("inflight", self.h.rec(a)) self.assertEqual(len(self.codex().started), 1) # GW-2 must not share win10 with a live run self.h.tick() self.assertNotIn("inflight", self.h.rec(a)) self.assertIn("被改为 Backlog", self.p.texts(a)[-1]) self.assertEqual(self.codex().started[-1].run_id, "GW-2-1") def test_transient_retries_do_not_use_up_thread_rounds(self): iid = self.p.add_issue(1) self.h.tick() self.codex().finish("[progress] 1") self.h.tick() self.codex().finish("[progress] 2") self.h.tick() # round 3: thread_rounds == 3 == max self.assertEqual(self.h.rec(iid)["thread_rounds"], 3) self.codex().finish("", exit_code=1, tail="stream disconnected before completion") self.h.tick() self.h.clock[0] += 121 self.h.tick() retry = self.codex().started[-1] self.assertEqual((retry.run_id, retry.thread_id), ("GW-1-4", "th-1")) # same thread, new run id self.assertEqual(self.h.rec(iid)["thread_rounds"], 3) self.codex().finish("[progress] 3") self.h.tick() self.assertIsNone(self.codex().started[-1].thread_id) # next real round rebuilds self.assertIn("线程已续 3 轮（上限 3）", self.p.texts(iid)[-1]) def test_failed_start_keeps_retry_marker(self): iid = self.p.add_issue(1) self.h.tick() self.codex().finish("", exit_code=1, tail="Connection reset by peer") self.h.tick() self.h.clock[0] += 121 self.codex().fail_starts = 1 self.h.tick() # relay down while relaunching self.assertIn("retry", self.h.rec(iid)) self.assertEqual(self.p.state_of(iid), "In Progress") self.h.tick() self.assertEqual(self.codex().started[-1].thread_id, "th-1") self.assertIn("inflight", self.h.rec(iid)) def test_progress_cap_moves_to_waiting(self): data = config_data() data["projects"]["GWENT"]["max_progress_rounds"] = 2 h = Harness(data) iid = h.plane.add_issue(1) h.tick() ex = h.execs[("win10", "codex")] ex.finish("[progress] 1") h.tick() self.assertEqual(len(ex.started), 2) ex.finish("[progress] 2") h.tick() self.assertEqual(len(ex.started), 2) self.assertEqual((h.plane.state_of(iid), h.plane.assignee_of(iid)), ("Waiting", "kappiee")) self.assertIn("已连续 2 轮 [progress]", h.plane.texts(iid)[-1]) h.plane.issues[iid].update(state="s-todo", assignees=["u-w10"]) # a person lets it continue: fresh allowance h.tick() ex.finish("[progress] 3") h.tick() self.assertEqual(len(ex.started), 4) h.close() def test_tagged_message_counts_as_success_even_with_nonzero_exit(self): iid = self.p.add_issue(1) self.h.tick() self.codex().finish("[review] done, but a hook failed", exit_code=1) self.h.tick() self.assertEqual(self.p.state_of(iid), "Review") self.assertIn("第 1 轮结束（exit=1，", self.p.texts(iid)[-1]) def test_untagged_message_with_nonzero_exit_is_a_failure(self): iid = self.p.add_issue(1) self.h.tick() self.codex().finish("I gave up, tests timed out", exit_code=1, tail="") self.h.tick() self.assertEqual(self.p.state_of(iid), "Waiting") # agent prose never makes it transient self.assertIn("类型=crash，run=GW-1-1", self.p.texts(iid)[-1]) def test_rebuild_after_resume_failure_happens_once(self): iid = self.p.add_issue(1) self.h.tick() self.codex().finish("[progress] half way") self.h.tick() self.codex().finish("", exit_code=1, thread=None, tail="thread not found") self.h.tick() # immediate rebuild (round 3, fresh thread) self.assertIsNone(self.codex().started[2].thread_id) self.assertNotIn("重试", self.p.texts(iid)[-1]) self.codex().finish("", exit_code=1, thread=None, tail="thread not found") self.h.tick() self.assertEqual(len(self.codex().started), 3) # no second rebuild self.assertEqual((self.p.state_of(iid), self.p.assignee_of(iid)), ("Waiting", "kappiee")) def test_state_saved_right_after_each_start(self): self.p.add_issue(1, assignee="oregon") self.p.add_issue(2, assignee="oregon") saves = [] orig = self.h.d.store.save self.h.d.store.save = lambda data: (saves.append(sorted( ("L" if r.get("launching") else "") + (r.get("inflight") or r.get("launching"))["run_id"] for r in data["projects"]["GWENT"]["issues"].values() if r.get("inflight") or r.get("launching"))), orig(data)) self.h.tick() # the launching marker is on disk before each launch call, the run right after it (a crash mid-tick cannot # orphan it or start it twice), then the end-of-tick save self.assertEqual(saves, [["LGW-1-1"], ["GW-1-1"], ["GW-1-1", "LGW-2-1"], ["GW-1-1", "GW-2-1"], ["GW-1-1", "GW-2-1"]]) class EnvFailureTest(unittest.TestCase): def test_env_failure_pauses_group_until_item_leaves_waiting(self): data = config_data() data["projects"]["GWENT"]["retry"]["env_patterns"] = ["foreground window lost"] h = Harness(data) a = h.plane.add_issue(1, labels=["needs-game"]) b = h.plane.add_issue(2, labels=["needs-game"]) h.tick() ex = h.execs[("win10", "codex")] ex.finish("", exit_code=1, tail="foreground window lost") h.tick() self.assertEqual(h.plane.state_of(a), "Waiting") self.assertEqual(h.plane.state_of(b), "Todo") # group paused self.assertIn("暂停派发", h.plane.texts(a)[-1]) h.plane.issues[a]["state"] = "s-done" h.tick() self.assertEqual(h.plane.state_of(b), "In Progress") h.close() class TierDefaultTest(unittest.TestCase): """executors..tier_defaults: tier per model unless a label (`tier:`, `tier:none`) sets one.""" def run_with(self, labels): data = config_data() data["executors"]["codex"]["tier_defaults"] = {"sol": "fast"} h = Harness(data) try: iid = h.plane.add_issue(1, labels=labels) h.tick() return h.execs[("win10", "codex")].started[0].tier, h.plane.texts(iid)[-1] finally: h.close() def test_default_model_gets_its_tier(self): tier, text = self.run_with(()) self.assertEqual(tier, "fast") self.assertIn("codex / sol / medium / fast @ win10", text) def test_other_model_has_no_default(self): self.assertIsNone(self.run_with(("model:astra",))[0]) def test_labels_override(self): self.assertEqual(self.run_with(("tier:flex",))[0], "flex") self.assertIsNone(self.run_with(("tier:none",))[0]) if __name__ == "__main__": unittest.main() class DoubleStartGuardTest(unittest.TestCase): """A launch whose outcome is unknown is probed on the next tick: adopted if it exists, relaunched in a fresh directory if not; never started twice in the same directory.""" def setUp(self): self.h = Harness() self.p = self.h.plane def tearDown(self): self.h.close() def codex(self): return self.h.execs[("win10", "codex")] def test_lost_launch_is_adopted_not_restarted(self): iid = self.p.add_issue(1) self.h.d.executor("win10", "codex").lost_starts = 1 self.h.tick() # run starts, but the relay reports a timeout rec = self.h.rec(iid) self.assertNotIn("inflight", rec) self.assertEqual((rec["launching"]["run_id"], rec["launching"]["handle"]["run_dir"]), ("GW-1-1", "/runs/GW-1-1")) self.assertEqual(self.p.state_of(iid), "Todo") self.h.tick() # probe finds launch.pid/pid.txt -> adopt self.assertEqual(len(self.codex().started), 1) rec = self.h.rec(iid) self.assertNotIn("launching", rec) self.assertEqual((rec["inflight"]["run_id"], rec["runs"]), ("GW-1-1", 1)) self.assertEqual(self.p.state_of(iid), "In Progress") self.assertIn("启动未确认、已接管", self.p.texts(iid)[-1]) self.codex().finish("[review] ok") self.h.tick() self.assertEqual(self.p.state_of(iid), "Review") def test_failed_launch_relaunches_same_run_id_in_fresh_dir(self): iid = self.p.add_issue(1) self.codex_ex = self.h.d.executor("win10", "codex") self.codex_ex.fail_starts = 1 self.h.tick() # nothing launched self.assertIn("launching", self.h.rec(iid)) self.h.tick() # probe: nothing -> relaunch as GW-1-1 in GW-1-1-r2 run = self.codex().started[-1] self.assertEqual((run.run_id, run.dir_name), ("GW-1-1", "GW-1-1-r2")) rec = self.h.rec(iid) self.assertEqual(rec["inflight"]["handle"]["run_dir"], "/runs/GW-1-1-r2") self.assertNotIn("relaunch", rec) self.assertIn("重新启动（目录 GW-1-1-r2）", self.p.texts(iid)[-1]) def test_unprobed_marker_holds_the_terminal_slot(self): a = self.p.add_issue(1) b = self.p.add_issue(2) ex = self.h.d.executor("win10", "codex") ex.lost_starts = 1 self.h.tick() self.assertEqual([r.run_id for r in ex.started], ["GW-1-1"]) # GW-2 not started next to a maybe-live run ex.fail_probes = 1 self.h.tick() # relay down: marker kept, win10 skipped, GW-2 still waits self.assertIn("launching", self.h.rec(a)) self.assertEqual(len(ex.started), 1) self.assertEqual(self.h.d.store.load()["projects"]["GWENT"]["issues"].get(b), None) self.h.tick() # relay back: adopt GW-1-1; concurrency 1 keeps GW-2 waiting self.assertEqual(self.h.rec(a)["inflight"]["run_id"], "GW-1-1") self.assertEqual(len(ex.started), 1) def test_marker_dropped_when_item_moved_away(self): iid = self.p.add_issue(1) self.h.d.executor("win10", "codex").fail_starts = 1 self.h.tick() self.p.issues[iid]["state"] = "s-bl" # a person parks it self.h.tick() rec = self.h.rec(iid) self.assertNotIn("launching", rec) self.assertNotIn("relaunch", rec) self.assertEqual(self.codex().started, []) def test_never_restarts_into_a_dir_with_a_pid_file(self): iid = self.p.add_issue(1) ex = self.h.d.executor("win10", "codex") ex.dirs["/runs/GW-1-1"] = {"pid.txt"} # leftover of a launch the state does not know about self.h.tick() # HASPID -> marker self.assertIn("launching", self.h.rec(iid)) self.h.tick() # probe finds pid.txt -> adopt instead of wiping self.assertEqual(self.h.rec(iid)["inflight"]["handle"]["run_dir"], "/runs/GW-1-1") self.assertEqual(ex.started, []) class RelayDownTest(unittest.TestCase): def test_down_terminal_is_skipped_and_others_continue(self): data = config_data() data["terminals"]["win10"]["concurrency"] = 2 h = Harness(data) w1, w2 = h.plane.add_issue(1), h.plane.add_issue(2) o = h.plane.add_issue(3, assignee="oregon") h.tick() win, ore = h.execs[("win10", "codex")], h.execs[("oregon", "claude-code")] self.assertEqual(len(win.started), 2) ore.finish("[review] done") w3 = h.plane.add_issue(4) win.fail_polls, win.polls[:] = 1, [] h.tick() self.assertEqual(len(win.polls), 1) # second win10 run not polled once the relay failed self.assertEqual(h.plane.state_of(o), "Review") # oregon carried on self.assertEqual(h.plane.state_of(w3), "Todo") # nothing new started on win10 hb = json.loads((h.d.store.dir / "heartbeat.json").read_text()) self.assertEqual((hb["down"], hb["errors"]), (["win10"], [])) self.assertTrue(all("inflight" in h.rec(i) for i in (w1, w2))) win.finish("[review] w") h.tick() # relay back: polled again self.assertIn("win10", {r["terminal"] for r in json.loads((h.d.store.dir / "heartbeat.json").read_text())["inflight"]}) h.close()