jj.wang
4f45d368ad
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.
163 lines
6.9 KiB
Python
163 lines
6.9 KiB
Python
"""In-memory fakes for Plane and executors (no network, no processes)."""
|
|
import copy
|
|
import tempfile
|
|
|
|
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"}
|
|
LABELS = {"model:astra": "l-astra", "needs-game": "l-game", "exec:claude": "l-claude", "budget:30m": "l-b30", "cap:gpu": "l-gpu",
|
|
"tier:none": "l-tnone", "tier:flex": "l-tflex"}
|
|
|
|
|
|
def config_data():
|
|
return {
|
|
"plane": {"base_url": "https://plane.example", "workspace": "ws"},
|
|
"agentd": {"max_rounds_per_thread": 3},
|
|
"terminals": {
|
|
"win10": {"transport": "warpgate-ssh", "target": "win10", "os": "windows", "launch": "schtasks", "repo": "C:\\repo",
|
|
"plane_user": "window10", "token_key": "TOK_W10", "concurrency": 1, "capabilities": ["game"],
|
|
"executors": ["codex"]},
|
|
"oregon": {"transport": "local", "os": "linux", "launch": "subprocess", "repo": "/repo", "plane_user": "oregon",
|
|
"token_key": "TOK_OR", "concurrency": 2, "executors": ["claude-code", "shell"]},
|
|
},
|
|
"executors": {
|
|
"codex": {"run": "codex exec -", "resume": "codex exec resume {thread} -", "models": ["sol", "astra"], "default_effort": "medium"},
|
|
"claude-code": {"run": "claude -p", "resume": "claude -p --resume {thread}", "models": ["sonnet"]},
|
|
"shell": {"run": "{script}", "resume": False},
|
|
},
|
|
"projects": {"GWENT": {
|
|
"id": "proj", "key_prefix": "GW", "reviewer": "oregon", "answerer": "kappiee",
|
|
"states": {"todo": "Todo", "running": "In Progress", "waiting": "Waiting", "review": "Review", "done": "Done"},
|
|
"routing": {"by_assignee": {"window10": "win10", "oregon": "oregon"},
|
|
"by_label": {"model:astra": {"model": "astra", "effort": "high"}, "needs-game": {"serialize_group": "game"},
|
|
"exec:claude": {"executor": "claude-code"}},
|
|
"default_executor": {"win10": "codex", "oregon": "claude-code"}},
|
|
"retry": {"transient_max": 2, "backoff_minutes": [2, 5]},
|
|
"timeout_hours": 4,
|
|
}},
|
|
}
|
|
|
|
|
|
class FakePlane:
|
|
def __init__(self):
|
|
self.issues, self.comments, self.patches, self.n = {}, {}, [], 0
|
|
|
|
def add_issue(self, seq, state="Todo", assignee="window10", labels=(), desc="<p>do <b>it</b></p>"):
|
|
iid = "i%d" % seq
|
|
self.issues[iid] = {"id": iid, "sequence_id": seq, "name": "T-%d" % seq, "description_html": desc,
|
|
"state": STATES[state], "assignees": [MEMBERS[assignee]], "labels": [LABELS[l] for l in labels]}
|
|
self.comments[iid] = []
|
|
return iid
|
|
|
|
def add_comment(self, iid, html):
|
|
self.n += 1
|
|
c = {"id": "c%d" % self.n, "created_at": "2026-10-21T00:00:%02d" % self.n, "comment_html": html}
|
|
self.comments[iid].append(c)
|
|
return c
|
|
|
|
def texts(self, iid):
|
|
from agentd.plane import html_to_text
|
|
return [html_to_text(c["comment_html"]) for c in self.comments[iid]]
|
|
|
|
def state_of(self, iid):
|
|
return {v: k for k, v in STATES.items()}[self.issues[iid]["state"]]
|
|
|
|
def assignee_of(self, iid):
|
|
return {v: k for k, v in MEMBERS.items()}[self.issues[iid]["assignees"][0]]
|
|
|
|
# client API used by core
|
|
def states(self, pid): return dict(STATES)
|
|
def labels(self, pid): return dict(LABELS)
|
|
def members(self): return dict(MEMBERS)
|
|
def list_issues(self, pid): return copy.deepcopy(list(self.issues.values()))
|
|
def list_comments(self, pid, iid): return copy.deepcopy(self.comments[iid])
|
|
def post_comment(self, pid, iid, html): return copy.deepcopy(self.add_comment(iid, html))
|
|
|
|
def set_state(self, pid, iid, state, assignees=None):
|
|
self.patches.append((iid, state, assignees))
|
|
self.issues[iid]["state"] = state
|
|
if assignees is not None:
|
|
self.issues[iid]["assignees"] = assignees
|
|
|
|
|
|
class FakeExecutor:
|
|
def __init__(self, resume=True):
|
|
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 TransportError("relay down")
|
|
h = self.handle(run)
|
|
if "pid.txt" in self.dirs.get(h["run_dir"], ()):
|
|
raise TransportError("HASPID")
|
|
self.started.append(run)
|
|
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):
|
|
return self.done[handle["run_id"]]
|
|
|
|
def kill(self, handle):
|
|
if self.fail_kills:
|
|
self.fail_kills -= 1
|
|
raise RuntimeError("relay down")
|
|
self.killed.append(handle["run_id"])
|
|
|
|
def finish(self, message, exit_code=0, thread="th-1", tail=""):
|
|
run = self.started[-1]
|
|
self.done[run.run_id] = ExecResult(message, exit_code, thread, ["/runs/%s/last.md" % run.run_id], tail)
|
|
|
|
|
|
class Harness:
|
|
def __init__(self, data=None):
|
|
self.cfg = Config(data or config_data())
|
|
self.plane = FakePlane()
|
|
self.execs = {}
|
|
self.clock = [1_000_000.0]
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.d = Agentd(self.cfg, store=Store(self.tmp.name, echo=False), plane_factory=lambda term: self.plane,
|
|
executor_factory=self.executor, now=lambda: self.clock[0])
|
|
|
|
def executor(self, term, name):
|
|
return self.execs.setdefault((term, name), FakeExecutor(resume=name != "shell"))
|
|
|
|
def tick(self):
|
|
return self.d.tick()
|
|
|
|
def rec(self, iid):
|
|
return self.d.store.load()["projects"]["GWENT"]["issues"][iid]
|
|
|
|
def close(self):
|
|
self.tmp.cleanup()
|