kappiee/plane-agentd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
19be2296bc51df070247df5ce3f3f736dbf98d60
plane-agentd/tests/test_core.py
T
jj.wang 4f45d368ad Double-start guard: launching marker, probe-and-adopt, fresh-dir relaunch; skip unreachable terminals per tick 
Before the launch call the planned run (run id, terminal, handle, bookkeeping) is saved as a launching
marker. If the launch call fails or agentd dies before recording the run, the next tick probes the run
directory (exit.txt / launch.pid / pid.txt): evidence -> adopt the run; nothing -> relaunch with the same
run id in <run_id>-r2 (-r3 ...); probe unreachable -> keep the marker, which holds the terminal slot.
Transports refuse to clean a run directory holding a pid file. A terminal whose relay fails is skipped
for the rest of the tick (heartbeat.down) while other terminals carry on.
2026-10-21 11:54:40 +00:00

508 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
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, "<p>old context</p>")
self.h.tick()
self.codex().finish("[question] A or B?")
self.h.tick()
self.p.add_comment(iid, "<p>B please</p>")
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.<x>.tier_defaults: tier per model unless a label (`tier:<x>`, `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()
Reference in New Issue View Git Blame Copy Permalink