diff --git a/agentd/core.py b/agentd/core.py index 27a3fa8..2ca243d 100644 --- a/agentd/core.py +++ b/agentd/core.py @@ -3,7 +3,13 @@ on failure post a structured note and retry (transient) or hand over (Waiting); kill on timeout; (b) start the lowest-sequence Todo items whose terminal has a free slot and whose serialize group is idle; (c) write the heartbeat. -Everything is deterministic; models only run inside executors.""" +Everything is deterministic; models only run inside executors. + +Double-start guard: before the launch call a `launching` marker (the planned run and its handle) is saved. If the +launch call fails or the process dies before the run is recorded, the next tick probes the run directory: launch +evidence (exit.txt / launch.pid / pid.txt) -> adopt the run; nothing -> relaunch with the same run id in a fresh +directory (`-r2`, -r3 ...); probe unreachable -> keep the marker (it holds the terminal slot). +A terminal whose relay fails is skipped for the rest of the tick; other terminals carry on.""" import os import re import time @@ -15,7 +21,7 @@ from .executors import ExecResult, Run, make_executor from .executors.base import fill from .plane import Plane, html_to_text, text_to_html from .store import Store -from .transports import make_transport +from .transports import TransportError, make_transport TAGS = ("review", "question", "blocked", "progress") WORKER = "[worker]" @@ -143,7 +149,7 @@ class Ctx: class Agentd: def __init__(self, cfg, store=None, plane_factory=None, executor_factory=None, transport_factory=None, now=time.time): self.cfg = cfg - self.store = store or Store(cfg.state_dir) + self.store = store or Store(cfg.state_dir, log_dir=cfg.log_dir) self.now = now self.log = self.store.log self.plane_factory = plane_factory or (lambda term: Plane( @@ -154,6 +160,12 @@ class Agentd: def _reset_caches(self): self._clients, self._transports, self._executors, self._ready = {}, {}, {}, {} + self._down = set() # terminals whose relay failed during this tick + + def mark_down(self, term, exc): + if term not in self._down: + self.log("%s unreachable, skipping it for the rest of this tick: %s" % (term, exc)) + self._down.add(term) # ---------------------------------------------------------------- caches def client(self, terminal): @@ -195,9 +207,11 @@ class Agentd: self.log("%s tick error: %r" % (name, exc)) finally: self.store.save(data) - self.store.heartbeat({"errors": errors, "inflight": [ + self.store.heartbeat({"errors": errors, "down": sorted(self._down), "inflight": [ {"project": p, "run": r["inflight"]["run_id"], "terminal": r["inflight"]["terminal"]} - for p, ps in data["projects"].items() for r in ps["issues"].values() if r.get("inflight")]}) + for p, ps in data["projects"].items() for r in ps["issues"].values() if r.get("inflight")], + "launching": [{"project": p, "run": r["launching"]["run_id"], "terminal": r["launching"]["terminal"]} + for p, ps in data["projects"].items() for r in ps["issues"].values() if r.get("launching")]}) return data def loop(self, interval): @@ -216,14 +230,43 @@ class Agentd: pstate = data["projects"].setdefault(name, {"issues": {}}) ctx = Ctx(name, pcfg, self.reader(pcfg)) for iid, rec in list(pstate["issues"].items()): - if rec.get("inflight"): - try: + run = rec.get("inflight") or rec.get("launching") + if not run or run["terminal"] in self._down: + continue + try: + if rec.get("inflight"): self.poll(ctx, data, iid, rec) - except Exception as exc: - self.log("%s poll error: %r" % (ctx.key(rec.get("seq")), exc)) - self.store.save(data) # persist each round's outcome (thread id, retry) before touching the next + else: + self.resolve_launch(ctx, iid, rec) + except TransportError as exc: + self.mark_down(run["terminal"], exc) + except Exception as exc: + self.log("%s poll error: %r" % (ctx.key(rec.get("seq")), exc)) + self.store.save(data) # persist each round's outcome (thread id, retry) before touching the next self.pick(ctx, data, pstate, only_issue) + def resolve_launch(self, ctx, iid, rec): + """A launch that was never confirmed (the launch call failed or agentd died mid-start): probe the run + directory and adopt the run if it exists; otherwise relaunch in a fresh directory, never wiping one.""" + m = rec["launching"] + key = ctx.key(rec["seq"]) + present = self.executor(m["terminal"], m["executor"]).probe(m["handle"]) # TransportError -> marker kept + if present: + self.log("%s run %s: unconfirmed launch found on %s (%s); adopting it" % ( + key, m["run_id"], m["terminal"], ",".join(present))) + return self.launched(ctx, iid, rec, m, m["handle"], adopted=True) + rec.pop("launching") + issue = ctx.issues.get(iid) + state = issue and issue.get("state") + if state == ctx.state_id("todo") or (state == ctx.state_id("running") and rec.get("retry")): + rec["relaunch"] = {"run_id": m["run_id"], "attempt": m.get("attempt", 1) + 1} + self.log("%s run %s: nothing at %s; relaunching in %s-r%d" % ( + key, m["run_id"], m["handle"].get("run_dir"), m["run_id"], rec["relaunch"]["attempt"])) + else: + rec.pop("relaunch", None) + self.log("%s run %s: nothing at %s and the item is no longer dispatchable; marker dropped" % ( + key, m["run_id"], m["handle"].get("run_dir"))) + # ---------------------------------------------------------------- (a) poll / finish def poll(self, ctx, data, iid, rec): fl = rec["inflight"] @@ -404,7 +447,7 @@ class Agentd: load, groups = Counter(), set() for p in data["projects"].values(): for r in p["issues"].values(): - fl = r.get("inflight") + fl = r.get("inflight") or r.get("launching") # an unconfirmed launch holds its slot if fl: load[fl["terminal"]] += 1 if fl.get("group"): @@ -422,7 +465,7 @@ class Agentd: if only_issue and iid != only_issue: continue rec = pstate["issues"].get(iid, {}) - if rec.get("inflight"): + if rec.get("inflight") or rec.get("launching"): continue if issue.get("state") == todo: cands.append(issue) @@ -438,15 +481,21 @@ class Agentd: self.log("%s not dispatched: %s" % (ctx.key(issue.get("sequence_id")), route)) continue term, group = route["terminal"], route["group"] + if term in self._down: + continue if load[term] >= self.cfg.terminals[term].get("concurrency", 1) or (group and group in groups): continue if not self.terminal_ready(term, load[term]): continue try: - self.start(ctx, pstate, issue, route) + self.start(ctx, data, pstate, issue, route) except Exception as exc: self.log("%s start failed on %s: %r" % (ctx.key(issue.get("sequence_id")), term, exc)) - continue + if isinstance(exc, TransportError): + self.mark_down(term, exc) + if not pstate["issues"].get(issue["id"], {}).get("launching"): + continue + # the launch call failed but may have started the run: its marker keeps the slot until probed self.store.save(data) # a launched run must be on disk before anything else can fail load[term] += 1 if group: @@ -454,14 +503,17 @@ class Agentd: def terminal_ready(self, term, our_load): """Optional guard against processes we did not start (e.g. a manual codex session).""" + if term in self._down: + return False if term not in self._ready: names = self.cfg.terminals[term].get("busy_processes", []) ready = True for name in names: count = self.transport(term).process_count(name) if count is None: - self.log("%s unreachable; skipping" % term) + self.mark_down(term, "busy-process probe failed") ready = False + break elif count > 0 and our_load == 0: self.log("%s busy (%d %s running that agentd did not start)" % (term, count, name)) ready = False @@ -522,7 +574,7 @@ class Agentd: return "" return p.read_text(encoding="utf-8") - def start(self, ctx, pstate, issue, route): + def start(self, ctx, data, pstate, issue, route): iid, seq = issue["id"], issue["sequence_id"] key = ctx.key(seq) rec = pstate["issues"].setdefault(iid, new_rec(seq)) @@ -552,25 +604,11 @@ class Agentd: intro = fill(ctx.cfg.get("intro", DEFAULT_INTRO), dict(terminal=term, executor=exname, repo=tcfg["repo"], prefix=ctx.prefix, key=key)) prompt = build_prompt(key, issue, comments, seen, bool(thread), self.protocol(pcfg), intro, handoff, retry_note) - run = Run(run_id, tcfg["repo"], prompt, route["model"], route["effort"], route["tier"], thread, route["script"]) - handle = ex.start(run) - rec.pop("retry", None) - rec.pop("rebuild", None) - if not is_retry: - rec["retries"] = 0 - now = self.now() - # transient retries continue the same logical round: they do not use up max_rounds_per_thread - rounds = (rec.get("thread_rounds", 0) + (0 if is_retry else 1)) if thread else 1 - rec.update(runs=k, thread_rounds=rounds, inflight={ - "run_id": run_id, "terminal": term, "executor": exname, "handle": handle, "round": k, - "started": now, "deadline": now + route["timeout_hours"] * 3600, "resumed": bool(thread), - "group": route["group"], "model": route["model"], "effort": route["effort"]}) - if not thread: - rec.pop("thread_id", None) - rec.pop("thread_on", None) - rec["seen_comments"] = sorted(seen | {c["id"] for c in comments if c.get("id")}) - rec["last_comment_at"] = max([c.get("created_at", "") for c in comments] + [rec.get("last_comment_at", "")]) - client = self.client(term) + relaunch = rec.get("relaunch") or {} + attempt = relaunch.get("attempt", 1) if relaunch.get("run_id") == run_id else 1 + dir_name = run_id if attempt == 1 else "%s-r%d" % (run_id, attempt) + run = Run(run_id, tcfg["repo"], prompt, route["model"], route["effort"], route["tier"], thread, route["script"], + dir_name=dir_name) if thread: mode = "续线程 %s" % thread[:12] elif reason: @@ -579,10 +617,53 @@ class Agentd: mode = "新线程（附交接摘要）" if handoff else "新线程" if is_retry: mode += "，重试 %d" % rec.get("retries", 0) - detail = " / ".join(x for x in (exname, route["model"], route["effort"], route["tier"]) if x) + if attempt > 1: + mode += "，重新启动（目录 %s）" % dir_name + # Everything the bookkeeping needs once the run is known to exist; saved before the launch call so that a + # launch whose outcome is unknown (relay timeout, crash) is probed and adopted instead of started twice. + plan = {"run_id": run_id, "terminal": term, "executor": exname, "attempt": attempt, "at": self.now(), + "handle": ex.handle(run), "round": k, "budget_s": route["timeout_hours"] * 3600, + "resumed": bool(thread), "group": route["group"], "model": route["model"], "effort": route["effort"], + "is_retry": is_retry, + # transient retries continue the same logical round: they do not use up max_rounds_per_thread + "thread_rounds": (rec.get("thread_rounds", 0) + (0 if is_retry else 1)) if thread else 1, + "seen": sorted(seen | {c["id"] for c in comments if c.get("id")}), + "last_comment_at": max([c.get("created_at", "") for c in comments] + [rec.get("last_comment_at", "")]), + "mode": mode, + "detail": " / ".join(x for x in (exname, route["model"], route["effort"], route["tier"]) if x)} + rec["launching"] = plan + self.store.save(data) + handle = ex.start(run) # raises -> the marker stays and the next tick probes the run directory + self.launched(ctx, iid, rec, plan, handle) + + def launched(self, ctx, iid, rec, plan, handle, adopted=False): + """The run exists on the terminal: record it as in flight, then move the item and post the start note.""" + key, term = ctx.key(rec["seq"]), plan["terminal"] + rec.pop("launching", None) + rec.pop("relaunch", None) + rec.pop("retry", None) + rec.pop("rebuild", None) + if not plan["is_retry"]: + rec["retries"] = 0 + rec.update(runs=plan["round"], thread_rounds=plan["thread_rounds"], inflight={ + "run_id": plan["run_id"], "terminal": term, "executor": plan["executor"], "handle": handle, + "round": plan["round"], "started": plan["at"], "deadline": plan["at"] + plan["budget_s"], + "resumed": plan["resumed"], "group": plan["group"], "model": plan["model"], "effort": plan["effort"]}) + if not plan["resumed"]: + rec.pop("thread_id", None) + rec.pop("thread_on", None) + rec["seen_comments"] = plan["seen"] + rec["last_comment_at"] = plan["last_comment_at"] + mode = plan["mode"] + ("，启动未确认、已接管" if adopted else "") + issue = ctx.issues.get(iid) + client = self.client(term) try: - self.move(ctx, client, iid, "running", tcfg["plane_user"]) - self.post(ctx, client, iid, rec, "%s 第 %d 轮开始：%s @ %s，%s，run=%s。" % (WORKER, k, detail, term, mode, run_id)) + # an adopted run whose item a person moved meanwhile is left alone: poll cancels it next tick + if issue is not None and issue.get("state") in (ctx.state_id("todo"), ctx.state_id("running")): + self.move(ctx, client, iid, "running", self.cfg.terminals[term]["plane_user"]) + self.post(ctx, client, iid, rec, "%s 第 %d 轮开始：%s @ %s，%s，run=%s。" % ( + WORKER, plan["round"], plan["detail"], term, mode, plan["run_id"])) except Exception as exc: self.log("%s bookkeeping after start failed: %r" % (key, exc)) - self.log("%s run %s started on %s (%s, %s)" % (key, run_id, term, detail, mode)) + self.log("%s run %s %s on %s (%s, %s)" % (key, plan["run_id"], "adopted" if adopted else "started", term, + plan["detail"], mode)) diff --git a/agentd/executors/base.py b/agentd/executors/base.py index 37f0c14..181e87d 100644 --- a/agentd/executors/base.py +++ b/agentd/executors/base.py @@ -17,6 +17,7 @@ class Run: tier: Optional[str] = None thread_id: Optional[str] = None # set => resume this thread script: Optional[str] = None # shell executor + dir_name: Optional[str] = None # run directory / task name; default run_id (`-r2` on a relaunch) @dataclass @@ -35,9 +36,15 @@ class Executor: def __init__(self, name, cfg, terminal, tcfg, transport): self.name, self.cfg, self.terminal, self.tcfg, self.transport = name, cfg, terminal, tcfg, transport + def handle(self, run): # -> the handle `start(run)` will return, computed without touching the terminal + raise NotImplementedError + def start(self, run): # -> handle (dict) raise NotImplementedError + def probe(self, handle): # -> names of launch evidence present on the terminal ([] = never launched) + raise NotImplementedError + def poll(self, handle): # -> "running" | "finished" raise NotImplementedError @@ -84,11 +91,15 @@ class RunnerExecutor(Executor): program = "& " + program return (program + " " + rest).strip() + def handle(self, run): + name = run.dir_name or run.run_id + return {"run_dir": self.transport.join(self.run_root(), name), "task": "agentd-" + name, "thread_id": run.thread_id} + def start(self, run): t = self.transport - run_dir = t.join(self.run_root(), run.run_id) + h = self.handle(run) + run_dir, task = h["run_dir"], h["task"] script_name, script = t.runner(run_dir, run.repo, self.command(run, run_dir), self.stdout_name) - task = "agentd-" + run.run_id files = {"prompt.md": run.prompt, script_name: script} if hasattr(t, "start_run"): # remote POSIX: stage + launch in one relay round trip t.start_run(task, run_dir, files, script_name) @@ -97,7 +108,13 @@ class RunnerExecutor(Executor): for name, text in files.items(): t.write_text(t.join(run_dir, name), text) t.launch(task, t.join(run_dir, script_name), run_dir) - return {"run_dir": run_dir, "task": task, "thread_id": run.thread_id} + return h + + PROBE_FILES = ("exit.txt", "launch.pid", "pid.txt") + + def probe(self, handle): + """Launch evidence in the run directory (one transport call; raises TransportError when unreachable).""" + return self.transport.probe(handle["run_dir"], self.PROBE_FILES) def poll(self, handle): text = self.transport.read_text(self.transport.join(handle["run_dir"], "exit.txt")) diff --git a/agentd/transports/local.py b/agentd/transports/local.py index e5563fa..3a328c5 100644 --- a/agentd/transports/local.py +++ b/agentd/transports/local.py @@ -4,6 +4,10 @@ import signal import subprocess from pathlib import Path +from . import TransportError + +PID_FILES = ("pid.txt", "launch.pid") # a run directory holding one of these is never cleaned + # Shared with the remote POSIX transport (warpgate_ssh): plain sh + `date +FMT`, so it also runs on macOS. RUNNER = """#!/usr/bin/env bash cd {cwd} || {{ echo "exit=97 at $(date +%Y-%m-%dT%H:%M:%S%z) (cd failed)" > {d}/exit.txt; exit 97; }} @@ -40,6 +44,8 @@ class LocalTransport: p = Path(path) p.mkdir(parents=True, exist_ok=True) if clean: + if any((p / n).exists() for n in PID_FILES): + raise TransportError("refusing to clean %s: it holds a pid file (HASPID)" % p) for f in p.iterdir(): if f.is_file(): f.unlink() @@ -51,6 +57,9 @@ class LocalTransport: p = Path(path) return p.read_text(encoding="utf-8", errors="replace") if p.exists() else None + def probe(self, run_dir, names): + return [n for n in names if os.path.exists(os.path.join(run_dir, n))] + def grep_first(self, path, needle): text = self.read_text(path) or "" return next((line for line in text.splitlines() if needle in line), None) diff --git a/agentd/transports/warpgate_ssh.py b/agentd/transports/warpgate_ssh.py index eaad1a4..2114536 100644 --- a/agentd/transports/warpgate_ssh.py +++ b/agentd/transports/warpgate_ssh.py @@ -113,9 +113,21 @@ class WindowsWarpgateTransport: def mkdir(self, path, clean=False): q = self.quote(path) script = "New-Item -ItemType Directory -Force -Path %s | Out-Null\n" % q - if clean: + if clean: # never wipe a directory a runner has already started in + script += ("if ((Test-Path -LiteralPath (Join-Path %s 'pid.txt')) -or (Test-Path -LiteralPath (Join-Path %s 'launch.pid'))) " + "{ 'HASPID'; return }\n" % (q, q)) script += "Get-ChildItem -LiteralPath %s -File | Remove-Item -Force\n" % q - self.ps(script + "'MKDIR=ok'") + out = self.ps(script + "'MKDIR=ok'") + if "MKDIR=ok" not in out: + raise TransportError("mkdir %s failed: %s" % (path, out.strip()[-300:])) + + def probe(self, run_dir, names): + out = self.ps("$d = %s\n'PROBE=' + ((@(%s) | Where-Object { Test-Path -LiteralPath (Join-Path $d $_) }) -join ',')" + % (self.quote(run_dir), ",".join(self.quote(n) for n in names))) + m = re.search(r"PROBE=([^\n]*)", out) + if not m: + raise TransportError("probe %s: unexpected relay output: %s" % (run_dir, out.strip()[-200:])) + return [n for n in m.group(1).strip().split(",") if n] def write_text(self, path, text): fd, tmp = tempfile.mkstemp(prefix="agentd-", suffix=".txt") @@ -230,7 +242,9 @@ class PosixWarpgateTransport: def mkdir_script(self, path, clean=False): q = self.quote(path) s = "mkdir -p %s || exit 1\n" % q - if clean: + if clean: # never wipe a directory a runner has already started in + s += "if [ -e %s ] || [ -e %s ]; then echo HASPID; exit 1; fi\n" % ( + self.quote(posixpath.join(path, "pid.txt")), self.quote(posixpath.join(path, "launch.pid"))) s += "find %s -maxdepth 1 -type f -exec rm -f {} + || exit 1\n" % q return s @@ -275,6 +289,12 @@ class PosixWarpgateTransport: raise TransportError("read %s: unexpected relay output" % path) return _b64_payload(out, "B64") + def probe(self, run_dir, names): + d = self.quote(run_dir) + out = self.sh("printf PROBE=; for f in %s; do [ -e %s/\"$f\" ] && printf '%%s,' \"$f\"; done; echo" + % (" ".join(self.quote(n) for n in names), d), timeout=60, marker="PROBE=") + return [n for n in re.search(r"PROBE=([^\n]*)", out).group(1).split(",") if n] + def _b64_cmd(self, pipeline): out = self.sh("printf B64=; { %s; } 2>/dev/null | base64 | tr -d '\\n'; echo" % pipeline) text = _b64_payload(out, "B64") diff --git a/tests/fakes.py b/tests/fakes.py index 57b1a1e..73714d8 100644 --- a/tests/fakes.py +++ b/tests/fakes.py @@ -6,6 +6,7 @@ from agentd.config import Config from agentd.core import Agentd from agentd.executors import ExecResult from agentd.store import Store +from agentd.transports import TransportError STATES = {"Backlog": "s-bl", "Todo": "s-todo", "In Progress": "s-run", "Waiting": "s-wait", "Review": "s-rev", "Done": "s-done"} MEMBERS = {"window10": "u-w10", "oregon": "u-or", "kappiee": "u-k"} @@ -89,15 +90,39 @@ class FakeExecutor: self.resume = resume self.started, self.killed, self.done = [], [], {} self.fail_starts, self.fail_kills = 0, 0 + self.lost_starts = 0 # the run launches but the relay reports a failure (e.g. timeout) + self.fail_probes = self.fail_polls = 0 + self.dirs = {} # run_dir -> files on the "terminal" + self.polls = [] + + def handle(self, run): + return {"run_dir": "/runs/" + (run.dir_name or run.run_id), "run_id": run.run_id, "thread_id": run.thread_id} def start(self, run): if self.fail_starts: self.fail_starts -= 1 - raise RuntimeError("relay down") + raise TransportError("relay down") + h = self.handle(run) + if "pid.txt" in self.dirs.get(h["run_dir"], ()): + raise TransportError("HASPID") self.started.append(run) - return {"run_dir": "/runs/" + run.run_id, "run_id": run.run_id, "thread_id": run.thread_id} + self.dirs[h["run_dir"]] = {"launch.pid", "pid.txt"} + if self.lost_starts: + self.lost_starts -= 1 + raise TransportError("relay timeout after 150s") + return h + + def probe(self, handle): + if self.fail_probes: + self.fail_probes -= 1 + raise TransportError("relay down") + return sorted(self.dirs.get(handle["run_dir"], ())) def poll(self, handle): + self.polls.append(handle["run_id"]) + if self.fail_polls: + self.fail_polls -= 1 + raise TransportError("relay down") return "finished" if handle["run_id"] in self.done else "running" def collect(self, handle): diff --git a/tests/test_core.py b/tests/test_core.py index e1d3873..d374eba 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -1,3 +1,4 @@ +import json import unittest from tests.fakes import Harness, config_data @@ -337,10 +338,13 @@ class CoreTest(unittest.TestCase): saves = [] orig = self.h.d.store.save self.h.d.store.save = lambda data: (saves.append(sorted( - r["inflight"]["run_id"] for r in data["projects"]["GWENT"]["issues"].values() if r.get("inflight"))), orig(data)) + ("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() - # one save per launched run (a crash mid-tick cannot orphan it), then the end-of-tick save - self.assertEqual(saves, [["GW-1-1"], ["GW-1-1", "GW-2-1"], ["GW-1-1", "GW-2-1"]]) + # 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): @@ -391,3 +395,113 @@ class TierDefaultTest(unittest.TestCase): 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() diff --git a/tests/test_posix_remote.py b/tests/test_posix_remote.py index 2e7b4c9..13c9617 100644 --- a/tests/test_posix_remote.py +++ b/tests/test_posix_remote.py @@ -203,3 +203,17 @@ class PosixRemoteTest(unittest.TestCase): if __name__ == "__main__": unittest.main() + + def test_probe_dir_name_and_pid_guard(self): + ex = make_executor(self.cfg, "mac", "shell", self.t) + run = Run("GW-5-1", self.cfg.terminals["mac"]["repo"], "", script="sleep 0.2", dir_name="GW-5-1-r2") + self.assertEqual(ex.probe(ex.handle(run)), []) + h = ex.start(run) + self.assertEqual((os.path.basename(h["run_dir"]), h["task"]), ("GW-5-1-r2", "agentd-GW-5-1-r2")) + self.assertTrue(wait(ex, h)) + self.assertEqual(ex.probe(h), ["exit.txt", "launch.pid", "pid.txt"]) + n = len(self.t.calls) + with self.assertRaises(AssertionError): # LocalBash: marker LAUNCHED= missing (HASPID) + ex.start(run) + self.assertIn("HASPID", self.t.calls[n]) + self.assertTrue(os.path.exists(os.path.join(h["run_dir"], "exit.txt"))) # nothing wiped diff --git a/tests/test_transports.py b/tests/test_transports.py index 0a0e5db..9b1f338 100644 --- a/tests/test_transports.py +++ b/tests/test_transports.py @@ -120,3 +120,30 @@ class StoreTest(unittest.TestCase): if __name__ == "__main__": unittest.main() + + +class GuardTest(unittest.TestCase): + def test_local_clean_refuses_dir_with_pid_file(self): + from agentd.transports import TransportError + with tempfile.TemporaryDirectory() as tmp: + t = LocalTransport(Config(config_data()), "oregon") + (Path(tmp) / "prompt.md").write_text("x") + t.mkdir(tmp, clean=True) + self.assertEqual(t.probe(tmp, ("exit.txt", "pid.txt")), []) + (Path(tmp) / "pid.txt").write_text("1") + with self.assertRaises(TransportError): + t.mkdir(tmp, clean=True) + self.assertEqual(t.probe(tmp, ("exit.txt", "pid.txt")), ["pid.txt"]) + + def test_windows_probe_and_clean_guard(self): + from agentd.transports import TransportError + cfg = config.load(ROOT / "config" / "agentd.example.yaml") + t = WindowsWarpgateTransport(cfg, "win10") + scripts = [] + t.ps = lambda s, timeout=150: (scripts.append(s), "PROBE=launch.pid,pid.txt\n")[1] + self.assertEqual(t.probe("C:\\r\\GW-1-1", ("exit.txt", "launch.pid", "pid.txt")), ["launch.pid", "pid.txt"]) + self.assertIn("Join-Path $d $_", scripts[-1]) + t.ps = lambda s, timeout=150: (scripts.append(s), "HASPID\n")[1] + with self.assertRaises(TransportError): + t.mkdir("C:\\r\\GW-1-1", clean=True) + self.assertIn("'pid.txt'", scripts[-1])