plane-agentd/tests/fakes.py

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

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


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 = [], [], {}

    def start(self, run):
        self.started.append(run)
        return {"run_dir": "/runs/" + run.run_id, "run_id": run.run_id, "thread_id": run.thread_id}

    def poll(self, handle):
        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):
        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()