"""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="

do it

"): 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()