code/moth/integrations/test_integrations.py

"""Offline checks for the moth harness adapters: fake harness payloads against a throwaway stdlib stub of moth.

Each adapter must post exactly the turn's final assistant text, skip subagent / aborted / text-less turns, print
nothing, exit 0, and give up within ~1 s when moth is slow (stub sleeps 2 s) or down (closed port).
The JS adapters run under `node` with fake `pi` / `ctx` / plugin inputs; a stray promise rejection would crash node.
`Display` covers the read-back half against a stub with a real log (LogStub): moth tail, the Hermes desktop pet
(and that Hermes registers no /moth), the Pi widget and /moth, the OpenCode toast, and the `show_in_harness` setting
(moth/config.py): off by default (still posted, nothing polled or drawn), env beats file, and a live toggle of the
file applies from the next turn without a restart. `Serve` checks serve.py's real
HTTP handler (torch faked out): Host allow-list, CORS, the per-install token, POST hygiene, /api/says, X-Frame-Options,
and that "new home" only deletes a folder moth made. `AdapterToLog` runs every adapter against the real handler and
the real Moth.hear_agent: the session ids and labels from each harness's payload end up on the log's cards.
Every adapter must send the token (from MOTH_TOKEN_FILE, set to a
throwaway file here) and its harness name as `source`, and post past an HTTP proxy set in the environment.

  python3 moth/integrations/test_integrations.py      (from ~/moth; about 50 s, mostly deliberate waits; only 127.0.0.1)
"""
import importlib.util
import json
import os
import shutil
import socket
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

HERE = Path(__file__).resolve().parent
TOKEN = "test-token-not-secret"
_tf = tempfile.NamedTemporaryFile("w", suffix=".token", delete=False)
_tf.write(TOKEN + "\n")
_tf.close()
os.environ["MOTH_TOKEN_FILE"] = _tf.name   # every adapter and display started from here reads this one
# moth's show_in_harness setting: never the user's real ~/.moth/config.json or shell; a file that doesn't exist = off
os.environ["MOTH_CONFIG_FILE"] = os.path.join(tempfile.mkdtemp(prefix="moth-cfg-"), "absent.json")
os.environ.pop("MOTH_SHOW_IN_HARNESS", None)
# "gives up within ~1 s": GIVE_UP bounds an adapter run (1 s timeout + interpreter start), SLOW is the slow stub's delay.
# Cold starts and refused loopback connects are slower on Windows / macOS runners, so both grow there, SLOW > GIVE_UP.
GIVE_UP, SLOW = (1.8, 2.0) if sys.platform.startswith("linux") else (3.0, 4.0)


class Stub:
    """Fake moth: records every POSTed {"text", "source"[, "session", "label"]} that carries the token; optionally
    answers only after `delay` seconds."""

    def __init__(self, delay=0.0):
        self.texts, self.sources, self.metas, stub = [], [], [], self

        class H(BaseHTTPRequestHandler):
            def do_POST(self):
                body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
                assert self.path == "/api/agent" and {"text", "source"} <= set(body) <= {"text", "source", "session",
                                                                                         "label"}, (self.path, body)
                assert all(isinstance(v, str) and v for v in body.values()), body
                assert self.headers["X-Moth-Token"] == TOKEN and self.headers["Content-Type"] == "application/json"
                stub.texts.append(body["text"])
                stub.sources.append(body["source"])
                stub.metas.append((body.get("session"), body.get("label")))
                time.sleep(delay)
                try:
                    self.send_response(200)
                    self.end_headers()
                    self.wfile.write(b"{}")
                except OSError:
                    pass

            def log_message(self, *a):
                pass

        self.srv = ThreadingHTTPServer(("127.0.0.1", 0), H)
        self.srv.daemon_threads = True
        self.url = f"http://127.0.0.1:{self.srv.server_port}"
        threading.Thread(target=self.srv.serve_forever, kwargs={"poll_interval": 0.02}, daemon=True).start()

    def close(self):
        self.srv.shutdown()
        self.srv.server_close()


def closed_url():
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()
    return f"http://127.0.0.1:{port}"


def run(cmd, url, stdin=""):
    t = time.monotonic()
    p = subprocess.run(cmd, input=stdin, capture_output=True, text=True, encoding="utf-8", timeout=10,
                       env={**os.environ, "MOTH_URL": url})
    return p, time.monotonic() - t


class Base(unittest.TestCase):
    def setUp(self):
        self.ok, self.slow = Stub(), Stub(delay=SLOW)

    def tearDown(self):
        self.ok.close()
        self.slow.close()

    def quiet(self, p, took):
        self.assertEqual((p.returncode, p.stdout, p.stderr), (0, "", ""))
        self.assertLess(took, GIVE_UP)


class ClaudeCode(Base):
    HOOK = [sys.executable, str(HERE / "claude_code.py")]

    def test_last_assistant_message(self):
        p, took = run(self.HOOK, self.ok.url, json.dumps({"hook_event_name": "Stop", "transcript_path": "/nope",
                                                          "session_id": "0f3c9a2e-aaaa", "cwd": "/home/u/moth-repo/",
                                                          "last_assistant_message": "## Done\nfixed **it**"}))
        self.quiet(p, took)
        self.assertEqual(self.ok.texts, ["## Done\nfixed **it**"])
        self.assertEqual(self.ok.sources, ["claude-code"])
        self.assertEqual(self.ok.metas, [("0f3c9a2e-aaaa", "moth-repo")])

    def test_session_label_fallbacks(self):
        for hook, meta in (({"session_id": "0f3c9a2e-aaaa", "cwd": "/"}, ("0f3c9a2e-aaaa", "0f3c9a2e")),
                           ({"session_id": 5, "cwd": "C:\\Users\\u\\proj"}, (None, "proj")),
                           ({}, (None, None))):
            with self.subTest(hook=hook):
                self.ok.metas.clear()
                self.quiet(*run(self.HOOK, self.ok.url, json.dumps({**hook, "last_assistant_message": "hi"})))
                self.assertEqual(self.ok.metas, [meta])

    def test_ignores_proxy_env(self):
        """A corporate proxy in the environment must not swallow a post to 127.0.0.1."""
        proxy = closed_url()
        p = subprocess.run(self.HOOK, input=json.dumps({"last_assistant_message": "via no proxy"}), text=True,
                           capture_output=True, timeout=10, env={**os.environ, "MOTH_URL": self.ok.url,
                           "http_proxy": proxy, "HTTP_PROXY": proxy, "no_proxy": "", "NO_PROXY": ""})
        self.assertEqual((p.returncode, self.ok.texts), (0, ["via no proxy"]))

    def test_bare_socket_post(self):
        """The plain-socket POST: a host name (localhost may resolve to ::1 first; the stub is on 127.0.0.1 only),
        a path prefix in MOTH_URL kept, and a token with a line break refused (nothing sent), as urllib did."""
        port = self.ok.url.rsplit(":", 1)[1]
        p, took = run(self.HOOK, f"http://localhost:{port}", json.dumps({"last_assistant_message": "by name"}))
        if sys.platform.startswith("linux"):   # Windows may sit out a refused ::1 connect first (~1 s): no timing
            self.quiet(p, took)
        self.assertEqual((p.returncode, self.ok.texts), (0, ["by name"]))
        spec = importlib.util.spec_from_file_location("moth_cc_hook", HERE / "claude_code.py")
        cc = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(cc)
        seen = []

        class Echo(BaseHTTPRequestHandler):
            def do_POST(self):
                seen.append((self.path, self.headers["Host"], self.headers["X-Moth-Token"],
                             self.rfile.read(int(self.headers["Content-Length"]))))
                self.send_response(200)
                self.end_headers()

            def log_message(self, *a):
                pass
        srv = ThreadingHTTPServer(("127.0.0.1", 0), Echo)
        threading.Thread(target=srv.serve_forever, kwargs={"poll_interval": 0.02}, daemon=True).start()
        try:
            url = f"http://127.0.0.1:{srv.server_port}/moth/api/agent"
            cc.http_post(url, b'{"text": "x"}', {"Content-Type": "application/json", "X-Moth-Token": "t0k"})
            self.assertEqual(seen, [("/moth/api/agent", f"127.0.0.1:{srv.server_port}", "t0k", b'{"text": "x"}')])
            with self.assertRaises(ValueError):
                cc.http_post(url, b"{}", {"X-Moth-Token": "a\r\nX-Evil: 1"})
            self.assertEqual(len(seen), 1)
        finally:
            srv.shutdown()
            srv.server_close()
        import ast   # codex.py carries the same function (each adapter is one file): the code must not drift apart

        def body(file):
            fn = next(n for n in ast.parse((HERE / file).read_text(encoding="utf-8")).body
                      if isinstance(n, ast.FunctionDef) and n.name == "http_post")
            return [ast.dump(s) for s in fn.body[1:]]   # minus the docstring
        self.assertEqual(body("claude_code.py"), body("codex.py"))

    def test_empty_message_posts_nothing(self):
        p, took = run(self.HOOK, self.ok.url, json.dumps({"last_assistant_message": "", "transcript_path": "/nope"}))
        self.quiet(p, took)
        self.assertEqual(self.ok.texts, [])

    def test_utf8_payloads(self):
        """Harnesses write raw UTF-8 JSON; the hooks must not decode it with the locale's code page (cp1252 on
        Windows), for the stdin payload (Claude Code, Codex) and Claude Code's transcript fallback alike."""
        text = "it’s fixed — naïve café, 🦋 ok"
        with tempfile.NamedTemporaryFile("wb", suffix=".jsonl", delete=False) as f:
            f.write(json.dumps({"type": "assistant", "message": {"content": [{"type": "text", "text": text}]}},
                               ensure_ascii=False).encode("utf-8") + b"\n")
        try:
            for hook, payload in ((self.HOOK, {"last_assistant_message": text}), (self.HOOK, {"transcript_path": f.name}),
                                  (Codex.HOOK, {"hook_event_name": "Stop", "last_assistant_message": text})):
                with self.subTest(hook=hook[-1], keys=sorted(payload)):
                    self.ok.texts.clear()
                    p = subprocess.run(hook, input=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
                                       capture_output=True, timeout=10, env={**os.environ, "MOTH_URL": self.ok.url})
                    self.assertEqual((p.returncode, p.stdout, p.stderr), (0, b"", b""))
                    self.assertEqual(self.ok.texts, [text])
        finally:
            os.unlink(f.name)

    def test_transcript_fallback_current_turn_only(self):
        lines = [
            {"type": "user", "message": {"content": "first prompt"}},
            {"type": "assistant", "message": {"content": [{"type": "text", "text": "old reply"}]}},
            {"type": "user", "message": {"content": "second prompt"}},
            {"type": "assistant", "message": {"content": [{"type": "text", "text": "looking"},
                                                          {"type": "tool_use", "id": "t1"}]}},
            {"type": "user", "message": {"content": [{"type": "tool_result", "tool_use_id": "t1"}]}},
            {"type": "assistant", "isSidechain": True, "message": {"content": [{"type": "text", "text": "sub"}]}},
            {"type": "assistant", "message": {"content": [{"type": "text", "text": "new reply"}]}},
        ]
        with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as f:
            f.write("\n".join(json.dumps(x) for x in lines) + "\nnot json\n")
        p, took = run(self.HOOK, self.ok.url, json.dumps({"transcript_path": f.name}))
        self.quiet(p, took)
        self.assertEqual(self.ok.texts, ["new reply"])
        with open(f.name, "a") as g:                      # a turn with no text yet: nothing, not "new reply" again
            g.write(json.dumps({"type": "user", "message": {"content": "third"}}) + "\n")
        p, took = run(self.HOOK, self.ok.url, json.dumps({"transcript_path": f.name}))
        self.quiet(p, took)
        self.assertEqual(self.ok.texts, ["new reply"])
        os.unlink(f.name)

    def test_moth_slow_down_or_garbage(self):
        stdin = json.dumps({"last_assistant_message": "hi"})
        p, took = run(self.HOOK, self.slow.url, stdin)
        self.quiet(p, took)
        self.assertEqual(self.slow.texts, ["hi"])
        self.quiet(*run(self.HOOK, closed_url(), stdin))
        self.quiet(*run(self.HOOK, self.ok.url, "not json"))


class Codex(Base):
    HOOK = [sys.executable, str(HERE / "codex.py")]

    def test_stop_hook(self):
        p, took = run(self.HOOK, self.ok.url, json.dumps({"hook_event_name": "Stop", "turn_id": "1",
                                                          "session_id": "cx-thread-1", "cwd": "/srv/api",
                                                          "last_assistant_message": "codex done"}))
        self.quiet(p, took)
        self.assertEqual(self.ok.texts, ["codex done"])
        self.assertEqual(self.ok.sources, ["codex"])
        self.assertEqual(self.ok.metas, [("cx-thread-1", "api")])

    def test_subagent_stop_and_null_skipped(self):
        self.quiet(*run(self.HOOK, self.ok.url, json.dumps({"hook_event_name": "SubagentStop",
                                                            "last_assistant_message": "sub"})))
        self.quiet(*run(self.HOOK, self.ok.url, json.dumps({"hook_event_name": "Stop",
                                                            "last_assistant_message": None})))
        self.assertEqual(self.ok.texts, [])

    def test_legacy_notify_argv(self):
        payload = {"type": "agent-turn-complete", "thread-id": "t", "turn-id": "1", "cwd": "/x",
                   "input-messages": ["do it"], "last-assistant-message": "notify done"}
        p, took = run(self.HOOK + [json.dumps(payload)], self.ok.url)
        self.quiet(p, took)
        self.assertEqual(self.ok.texts, ["notify done"])
        self.assertEqual(self.ok.metas, [("t", "x")])

    def test_moth_slow_or_down(self):
        stdin = json.dumps({"hook_event_name": "Stop", "last_assistant_message": "hi"})
        self.quiet(*run(self.HOOK, self.slow.url, stdin))
        self.assertEqual(self.slow.texts, ["hi"])
        self.quiet(*run(self.HOOK, closed_url(), stdin))


class Hermes(Base):
    def load(self, url):
        spec = importlib.util.spec_from_file_location("moth_hermes", HERE / "hermes" / "moth" / "__init__.py")
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)
        mod.URL = url + "/api/agent"
        return mod

    def wait(self, stub, n):
        for _ in range(100):
            if len(stub.texts) >= n:
                return
            time.sleep(0.01)

    def test_posts_final_skips_subagent(self):
        mod = self.load(self.ok.url)
        hooks = []
        mod.register(type("Ctx", (), {"register_hook": lambda self, n, f: hooks.append(n)})())
        self.assertEqual(hooks, ["post_llm_call"])
        kw = dict(session_id="s", user_message="hi", conversation_history=[], model="m")
        mod.on_post_llm_call(assistant_response="sub reply", platform="subagent", **kw)
        mod.on_post_llm_call(assistant_response="   ", platform="cli", **kw)
        mod.on_post_llm_call(assistant_response="hermes reply", platform="cli", **kw)
        self.wait(self.ok, 1)
        time.sleep(0.05)
        self.assertEqual(self.ok.texts, ["hermes reply"])
        self.assertEqual(self.ok.sources, ["hermes"])
        self.assertEqual(self.ok.metas, [("s", os.path.basename(os.environ.get("TERMINAL_CWD") or os.getcwd()))])

    def test_label_from_hermes_cwd(self):
        """Hermes' own per-session cwd (agent.runtime_cwd, a ContextVar) is read in the hook, before the thread."""
        import types
        from unittest import mock
        fake = types.ModuleType("agent.runtime_cwd")
        fake.scoped_session_cwd, fake.scope_terminal_cwd = (lambda: "/gw/sess-proj/"), (lambda: "/term/other")
        mod = self.load(self.ok.url)
        with mock.patch.dict(sys.modules, {"agent": types.ModuleType("agent"), "agent.runtime_cwd": fake}):
            mod.on_post_llm_call(assistant_response="one", platform="telegram", session_id="20261018_1200_ab")
            fake.scoped_session_cwd = lambda: ""
            mod.on_post_llm_call(assistant_response="two", platform="cli", session_id="")

            def refuse():
                raise RuntimeError("terminal scope refused")
            fake.scope_terminal_cwd = refuse
            with mock.patch.dict(os.environ, {"TERMINAL_CWD": "/env/fallback"}):
                mod.on_post_llm_call(assistant_response="three", platform="cli", session_id="s3")
        self.wait(self.ok, 3)
        got = dict(zip(self.ok.texts, self.ok.metas))
        self.assertEqual(got, {"one": ("20261018_1200_ab", "sess-proj"), "two": (None, "other"),
                               "three": ("s3", "fallback")})

    def test_never_blocks(self):
        for url in (self.slow.url, closed_url()):
            mod = self.load(url)
            t = time.monotonic()
            self.assertIsNone(mod.on_post_llm_call(assistant_response="hi", platform="cli"))
            self.assertLess(time.monotonic() - t, 0.1)
        self.wait(self.slow, 1)
        self.assertEqual(self.slow.texts, ["hi"])


# --- JS adapters: one node driver per harness, fed fake events ------------------------------------------------
PI = r"""
import ext from "./adapter.mjs";
const h = {}; ext({ on: (n, f) => { (h[n] ||= []).push(f); } });
const ctx = { cwd: "/home/u/pi-proj", sessionManager: { getSessionId: () => "pi-sess-1" } };
const fire = (n, e) => h[n].forEach((f) => f(e, ctx));
const A = (text, stopReason = "stop") => ({ role: "assistant", stopReason,
  content: [{ type: "thinking", thinking: "hmm" }, { type: "text", text }] });
fire("agent_start", {});
fire("agent_end", { messages: [{ role: "user", content: "hi" },
  { role: "assistant", stopReason: "toolUse", content: [{ type: "text", text: "pi-step" }, { type: "toolCall" }] },
  { role: "toolResult", content: [] }, A("pi-final")] });
fire("agent_settled", { aborted: false });
fire("agent_start", {}); fire("agent_end", { messages: [A("pi-aborted")] }); fire("agent_settled", { aborted: true });
fire("agent_start", {}); fire("agent_end", { messages: [A("pi-error", "error")] }); fire("agent_settled", { aborted: false });
fire("agent_start", {}); fire("agent_end", { messages: [A("x", "error")] });
fire("agent_start", {}); fire("agent_end", { messages: [A("pi-retried")] }); fire("agent_settled", { aborted: false });
fire("agent_end", null); fire("agent_settled", undefined);
"""
PI_EXPECT = ["pi-final", "pi-retried"]

OPENCODE = r"""
import { Moth } from "./adapter.mjs";
const hooks = await Moth({ client: {}, project: {}, directory: "/x/oc-plugin-dir", worktree: "/x", $: null });
const ev = (type, properties) => hooks.event({ event: { type, properties } });
const msg = (id, sid, role, extra = {}) => ev("message.updated", { info: { id, sessionID: sid, role, ...extra } });
const txt = (id, mid, sid, text, extra = {}) => ev("message.part.updated", { part: { id, messageID: mid, sessionID: sid, type: "text", text, ...extra } });
const idle = (sid) => ev("session.status", { sessionID: sid, status: { type: "idle" } });
await ev("session.created", { info: { id: "s1", title: "t", directory: "/home/u/oc-proj" } });
await msg("u1", "s1", "user"); await txt("p0", "u1", "s1", "user prompt");
await msg("a1", "s1", "assistant"); await txt("p1", "a1", "s1", "oc-step");
await ev("message.part.updated", { part: { id: "t1", messageID: "a1", sessionID: "s1", type: "tool" } });
await msg("a2", "s1", "assistant"); await txt("p2", "a2", "s1", "oc-fi");
await ev("message.part.delta", { sessionID: "s1", messageID: "a2", partID: "p2", field: "text", delta: "nal" });
await txt("p2", "a2", "s1", "oc-final", { time: { start: 1, end: 2 } });
await txt("p3", "a2", "s1", "reminder", { synthetic: true });
await ev("session.status", { sessionID: "s1", status: { type: "busy" } });
await idle("s1"); await ev("session.idle", { sessionID: "s1" });
await ev("session.created", { info: { id: "c1", parentID: "s1" } });
await msg("c1a", "c1", "assistant"); await txt("cp", "c1a", "c1", "oc-child"); await idle("c1");
await msg("a3", "s1", "assistant"); await idle("s1");
await msg("a4", "s1", "assistant"); await txt("p4", "a4", "s1", "oc-aborted");
await msg("a4", "s1", "assistant", { error: { name: "MessageAbortedError", data: {} } }); await idle("s1");
await msg("a5", "s1", "assistant", { summary: true }); await txt("p5", "a5", "s1", "oc-summary"); await idle("s1");
await hooks.event({}); await hooks.event({ event: null });
"""
OPENCODE_EXPECT = ["oc-final"]

DSH = r"""
import { apply, name } from "./adapter.mjs";
if (name !== "moth") throw new Error("name");
const h = {}; apply({ on: (n, f) => { (h[n] ||= []).push(f); } });
const se = (sid, type, data) => h["session/event"].forEach((f) => f({ id: sid }, { seq: 0, time: 0, type, data }));
const am = (turn, text, extra = {}) => ({ turn, step: 0, stream: [], ...extra, message: { id: "m", role: "assistant",
  source: { kind: "model" }, content: [{ type: "reasoning", text: "think" }, { type: "text", text }] } });
se("s1", "turn/start", { turn: 1 }); se("s1", "assistant/message", am(1, "dsh-step"));
se("s1", "tool/call", { turn: 1, step: 0, callId: "c", name: "bash", arguments: "{}" });
se("s1", "assistant/message", am(1, "dsh-final")); se("s1", "turn/end", { turn: 1, reason: { kind: "completed" } });
se("s1", "turn/start", { turn: 2 }); se("s1", "assistant/message", am(2, "dsh-aborted"));
se("s1", "turn/end", { turn: 2, reason: { kind: "aborted", reason: { kind: "user" } } });
se("s1", "turn/start", { turn: 3 }); se("s1", "assistant/message", am(3, "dsh-cut", { interrupted: true }));
se("s1", "turn/end", { turn: 3, reason: { kind: "completed" } });
se("s1", "turn/start", { turn: 4 }); se("s1", "turn/end", { turn: 4, reason: { kind: "completed" } });
h["subagent/start"].forEach((f) => f({ runId: "r", provider: "local", id: "c1", local: true }));
se("c1", "turn/start", { turn: 1 }); se("c1", "assistant/message", am(1, "dsh-child"));
se("c1", "turn/end", { turn: 1, reason: { kind: "completed" } });
se("s1", "turn/start", { turn: 5 }); se("s1", "assistant/message", am(5, "dsh-max"));
se("s1", "turn/end", { turn: 5, reason: { kind: "max-tokens" } });
se("s1", "turn/end", { turn: 6, reason: { kind: "completed" } }); h["session/event"].forEach((f) => f(null, null));
"""
DSH_EXPECT = ["dsh-final", "dsh-max"]


@unittest.skipUnless(shutil.which("node"), "node not installed")
class JsAdapters(Base):
    CASES = {"pi": (HERE / "pi" / "moth.ts", PI, PI_EXPECT),
             "opencode": (HERE / "opencode" / "moth.js", OPENCODE, OPENCODE_EXPECT),
             "dsh": (HERE / "dsh" / "moth.mjs", DSH, DSH_EXPECT)}
    SOURCE = {"pi": "pi", "opencode": "opencode", "dsh": "dsh"}
    META = {"pi": ("pi-sess-1", "pi-proj"), "opencode": ("s1", "oc-proj"), "dsh": ("s1", Path.cwd().name)}

    # node's own start for the baseline: the builtins the adapters load (node:http / node:https; Response loads
    # fetch's undici), no network. A fresh Windows runner spends ~2 s loading them the first time (measured: Pi's
    # import took 2.2 s once, 33 ms after), which a harness that is already running never pays.
    NODE_START = 'import "node:http"; import "node:https"; await new Response("x").text();\n'

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        with tempfile.TemporaryDirectory() as d:   # pay the first-load cost once, outside every timed run
            (Path(d) / "warm.mjs").write_text(cls.NODE_START, encoding="utf-8")
            subprocess.run(["node", str(Path(d) / "warm.mjs")], capture_output=True, timeout=30)

    def drive(self, harness, url):
        """(process, seconds the adapter took): the run's time minus a node start loading the same builtins right
        before it, so the GIVE_UP bound is about the adapter giving up, not node's cold start (3.5 s once on a busy
        Windows runner)."""
        src, driver, _ = self.CASES[harness]
        with tempfile.TemporaryDirectory() as d:
            shutil.copy(src, Path(d) / "adapter.mjs")   # .ts / .js are plain ESM JS; .mjs makes node load them as ESM
            (Path(d) / "driver.mjs").write_text(driver, encoding="utf-8")
            (Path(d) / "start.mjs").write_text(self.NODE_START, encoding="utf-8")
            _, base = run(["node", str(Path(d) / "start.mjs")], url)
            p, took = run(["node", str(Path(d) / "driver.mjs")], url)
            return p, took - base

    def test_posts_only_final_replies(self):
        for show in ("", "1"):   # posting never depends on the show_in_harness setting (default off, on)
            for harness, (_, _, expect) in self.CASES.items():
                with self.subTest(harness, show=show):
                    self.ok.texts.clear()
                    self.ok.sources.clear()
                    self.ok.metas.clear()
                    os.environ["MOTH_SHOW_IN_HARNESS"] = show
                    try:
                        self.quiet(*self.drive(harness, self.ok.url))
                    finally:
                        os.environ.pop("MOTH_SHOW_IN_HARNESS", None)
                    # each post is its own unawaited fetch, so two of them (Pi's pi-final and pi-retried) can land
                    # in either order on a busy machine: which posts arrived matters, not their order
                    self.assertEqual(sorted(self.ok.texts), sorted(expect))
                    self.assertEqual(self.ok.sources, [self.SOURCE[harness]] * len(expect))
                    self.assertEqual(self.ok.metas, [self.META[harness]] * len(expect))

    def test_moth_slow_or_down_never_crashes(self):
        for harness, (_, _, expect) in self.CASES.items():
            with self.subTest(harness):
                self.slow.texts.clear()
                self.quiet(*self.drive(harness, self.slow.url))   # fetch aborted at 1 s: no unhandled rejection
                self.assertEqual(sorted(self.slow.texts), sorted(expect))   # (in any order, as above)
                self.quiet(*self.drive(harness, closed_url()))
                self.quiet(*self.drive(harness, "not a url"))


# --- display layer: moth tail, Hermes /moth + desktop pet, Pi widget + /moth, OpenCode toast, serve.py CORS ----------
class LogStub:
    """Fake moth with a real log: /api/agent and /api/chat append events like serve.py; GET /api/log slices it
    (token only), GET /api/says gives moth's lines only (`?source=&session=`: only its reactions to those, like
    serve.py, which tags a reaction with "re": {"source", "session"} of the turn it answers). `epoch` changes when moth
    "restarts"; a poll with an old epoch gets the whole new log. A reaction to an agent turn lands `react` seconds
    after the post (moth thinking); chat answers after `delay`."""

    def __init__(self, delay=0.0, react=0.2):
        self.log, self.paths, self.epoch, stub = [], [], "e1", self

        class H(BaseHTTPRequestHandler):
            def _json(self, obj, code=200):
                b = json.dumps(obj).encode()
                try:
                    self.send_response(code)
                    self.send_header("Content-Type", "application/json")
                    self.send_header("Content-Length", str(len(b)))
                    self.end_headers()
                    self.wfile.write(b)
                except OSError:
                    pass

            def do_GET(self):
                from urllib.parse import parse_qs, urlsplit
                stub.paths.append(("GET", self.path))
                url = urlsplit(self.path)
                q = parse_qs(url.query)
                if url.path == "/api/log" and self.headers.get("X-Moth-Token") != TOKEN:
                    return self._json({"error": "token"}, 401)
                n = min(int(q.get("since", ["0"])[0]), len(stub.log))
                if q.get("epoch", [stub.epoch])[0] != stub.epoch:
                    n = 0
                evs = stub.log[n:]
                if url.path == "/api/says":
                    want = {k: q[k][0] for k in ("source", "session") if k in q}
                    evs = [{"say": e["say"]} for e in evs if "say" in e and
                           all((e.get("re") or {}).get(k) == v for k, v in want.items())]
                self._json({"events": evs, "next": len(stub.log), "epoch": stub.epoch, "tree": []})

            def do_POST(self):
                stub.paths.append(("POST", self.path))
                if self.headers.get("X-Moth-Token") != TOKEN:
                    return self._json({"error": "token"}, 401)
                body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
                text = body["text"]
                if self.path == "/api/agent":
                    stub.log.append({"agent": text})
                    re_ = {k: body[k] for k in ("source", "session") if body.get(k)}
                    threading.Timer(react, stub.log.append, [{"say": f"ooh, {text}", "re": re_}]).start()
                    return self._json({"events": []})
                stub.log.append({"you": text})
                time.sleep(delay)
                evs = [{"tool": "bash", "args": {"command": "ls"}, "result": "notes"}, {"say": f"hi back: {text}"}]
                stub.log.extend(evs)
                self._json({"events": evs})

            def log_message(self, *a):
                pass

        self.srv = ThreadingHTTPServer(("127.0.0.1", 0), H)
        self.srv.daemon_threads = True
        self.url = f"http://127.0.0.1:{self.srv.server_port}"
        threading.Thread(target=self.srv.serve_forever, kwargs={"poll_interval": 0.02}, daemon=True).start()

    def close(self):
        self.srv.shutdown()
        self.srv.server_close()


class Display(unittest.TestCase):
    ON = {"MOTH_SHOW_IN_HARNESS": "1"}

    def setUp(self):
        self.moth = LogStub()
        self.tmp = tempfile.mkdtemp()
        self.cfg = os.path.join(self.tmp, "config.json")   # per test; absent = the default (off)

    def tearDown(self):
        self.moth.close()
        shutil.rmtree(self.tmp, ignore_errors=True)

    def says_polls(self):
        return [p for m, p in self.moth.paths if m == "GET" and p.startswith("/api/says")]

    def test_tail_watch_prints_new_lines_and_survives_moth_down(self):
        self.moth.log += [{"agent": "old turn"}, {"say": "old line"}]
        p = subprocess.Popen([sys.executable, "-m", "moth.tail", "--watch", "--history", "1", "--url", self.moth.url],
                             cwd=HERE.parents[1], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
                             env={**os.environ, "COLUMNS": "40"})
        time.sleep(0.6)
        self.moth.log += [{"agent": "fixed the parser\nand ran the tests"}, {"say": "a parser! i ate a word once. " * 3}]
        time.sleep(1.4)
        p.terminate()
        out, err = p.communicate(timeout=5)
        self.assertEqual(err, "")
        self.assertNotIn("old turn", out)                       # --history 1: only the newest old event
        self.assertIn("moth> old line", out)
        self.assertIn("agent said: fixed the parser and ran", out)
        self.assertIn("moth> a parser!", out)
        self.assertTrue(all(len(ln) <= 39 for ln in out.splitlines()), out)   # wrapped to the pane
        self.assertFalse(any(m == "POST" for m, _ in self.moth.paths))   # display never posts
        down = subprocess.Popen([sys.executable, "-m", "moth.tail", "--watch", "--url", closed_url()],
                                cwd=HERE.parents[1], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        time.sleep(1.0 if sys.platform.startswith("linux") else 3.5)   # Windows: a refused connect takes ~2 s
        down.terminate()
        out, err = down.communicate(timeout=5)
        self.assertEqual((out.count("moth is asleep"), err), (1, ""))   # said once, not once per retry

    def test_tail_chat_echo_and_restart(self):
        sys.path.insert(0, str(HERE.parents[1]))
        from moth.tail import Tail
        t, shown = Tail(self.moth.url, watch=True, history=0), []
        t.show = shown.append
        t.poll_once()
        t.sent.append("hello moth")
        t.chat("hello moth")                       # the reply comes through the poll, our own line is not echoed
        t.poll_once()
        self.assertEqual([s for s in shown if "awake" not in s], ["  $ ls", "moth> hi back: hello moth"])
        self.moth.log[:] = [{"say": "fresh start"}]    # moth restarted: a new epoch, its log starts over
        self.moth.epoch = "e2"
        t.poll_once()
        self.assertEqual(shown[-1], "moth> fresh start")
        self.moth.log += [{"say": "one"}, {"say": "two"}, {"say": "three"}]   # restarted again, already past our cursor
        self.moth.log[:] = self.moth.log[-3:]
        self.moth.epoch = "e3"
        t.poll_once()
        self.assertEqual(shown[-3:], ["moth> one", "moth> two", "moth> three"])
        self.moth.log += [{"agent": "\x1b[31mred\x1b[0m done", "source": "claude-code"},
                          {"agent": "meh", "source": "codex", "skipped": True}, {"say": "\x1b]0;pwned\x07hi"}]
        t.poll_once()
        self.assertEqual(shown[-3:], ["claude said: red done", "codex said (moth let it pass): meh", "moth> hi"])
        t.url = closed_url()
        t.chat("anyone?")
        self.assertIn("didn't hear that", shown[-1])
        import moth.tail as tail_mod
        t.url, good = self.moth.url, tail_mod.TOKEN_FILE
        tail_mod.TOKEN_FILE = "/nonexistent/token"
        try:
            t.chat("no token")
            self.assertIn("no token in", shown[-1])
            with self.assertRaises(PermissionError):
                t.poll_once()
        finally:
            tail_mod.TOKEN_FILE = good

    def test_tail_labels_and_session_filter(self):
        sys.path.insert(0, str(HERE.parents[1]))
        from moth.tail import Tail
        self.moth.log += [{"agent": "fixed login", "source": "claude-code", "session": "a1", "label": "web"},
                          {"agent": "migrated db", "source": "codex", "session": "b2", "label": "api", "skipped": True},
                          {"agent": "two in web", "source": "codex", "session": "c3", "label": "web (c3c3)"},
                          {"agent": "old adapter", "source": "pi"}, {"say": "ooh"}, {"you": "nice"}]
        every = ["claude · web said: fixed login", "codex · api said (moth let it pass): migrated db",
                 "codex · web (c3c3) said: two in web", "pi said: old adapter", "moth> ooh", "you> nice"]
        for only, keep in ((None, every), ("web", [every[0], every[2]]), ("api", [every[1]]), ("b2", [every[1]]),
                           ("pi", [every[3]]), ("claude", [every[0]]), ("nope", [])):
            with self.subTest(session=only):
                t, shown = Tail(self.moth.url, watch=True, history=10, session=only), []
                t.show = shown.append
                t.poll_once()
                rest = [ln for ln in every[4:]]   # moth's lines and yours always show
                self.assertEqual([s for s in shown if "awake" not in s], keep + rest if only else every)
        p = subprocess.Popen([sys.executable, "-m", "moth.tail", "--watch", "--history", "10", "--session", "api",
                              "--url", self.moth.url], cwd=HERE.parents[1], stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE, text=True, env={**os.environ, "COLUMNS": "100"})
        time.sleep(0.8)
        p.terminate()
        out, err = p.communicate(timeout=5)
        self.assertEqual(err, "")
        self.assertIn("codex · api said (moth let it pass): migrated db", out)
        self.assertNotIn("fixed login", out)
        self.assertIn("moth> ooh", out)

    def test_hermes_has_no_moth_command(self):
        """/moth was removed from Hermes: mid-turn, the gateway hands unknown slash commands to the agent."""
        spec = importlib.util.spec_from_file_location("moth_hermes_cmd", HERE / "hermes" / "moth" / "__init__.py")
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)
        calls = []
        mod.register(type("Ctx", (), {"register_hook": lambda self, n, f: calls.append(("hook", n)),
                                      "register_command": lambda self, n, f, **kw: calls.append(("command", n))})())
        self.assertEqual(calls, [("hook", "post_llm_call")])

    def node(self, driver, src, url, name="adapter.mjs", env=None):
        with tempfile.TemporaryDirectory() as d:
            shutil.copy(src, Path(d) / name)
            (Path(d) / "driver.mjs").write_text(driver, encoding="utf-8")
            p = subprocess.run(["node", str(Path(d) / "driver.mjs")], capture_output=True, text=True, encoding="utf-8",
                               timeout=30,
                               env={**os.environ, "MOTH_URL": url, "MOTH_CONFIG_FILE": self.cfg, **(env or {})})
        self.assertEqual((p.returncode, p.stderr), (0, ""), p.stdout)
        return json.loads(p.stdout.strip().splitlines()[-1])

    @unittest.skipUnless(shutil.which("node"), "node not installed")
    def test_hermes_desktop_pet(self):
        driver = r"""
import plugin from "./plugin.mjs";
const said = [], timers = [];
let tick = null;
const ctx = { pet: { say: (text, o) => { said.push([text, o.id, o.ttlMs]); return () => {}; } },
  setInterval: (fn, ms) => { tick = fn; return () => {}; },
  setTimeout: (fn, ms) => { timers.push([fn, ms]); return () => {}; } };
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const post = (path, body) => fetch(`${process.env.MOTH_URL}${path}`, { method: "POST", body: JSON.stringify(body),
  headers: { "X-Moth-Token": "test-token-not-secret" } });
const add = (text, source = "hermes") => post("/api/agent", { text, source, session: "20261018_120000_ab12" });
// poll (the 2.5 s interval, driven by hand) until moth's reaction is said; not a fixed sleep: slow runners vary
const heard = async (n) => { for (const end = Date.now() + 8000; said.length < n && Date.now() < end;) {
  tick(); await sleep(100); } };
if (plugin.id !== "moth" || typeof plugin.register !== "function") throw new Error("shape");
plugin.register(ctx); await sleep(200);                 // first poll: cursor at the end, old lines skipped
await add("one"); await heard(1);                        // moth's reaction to a Hermes turn
await add("claude's turn", "claude-code"); await post("/api/chat", { text: "hi" });   // not Hermes: never voiced
await sleep(300); tick(); await sleep(200);
await add("x ".repeat(150) + "end"); await heard(2);     // a long line: cut into pieces (the rest on timers)
timers.forEach(([fn]) => fn());
plugin.register({}); plugin.register({ pet: {} });    // older desktops: no pet, no crash
console.log(JSON.stringify({ said, delays: timers.map(([, ms]) => ms) }));
"""
        self.moth.log += [{"say": "old line"}]
        with tempfile.TemporaryDirectory() as d:   # the renderer has no env: point the plugin's URL at the stub
            src = Path(d) / "plugin.js"
            src.write_text((HERE / "hermes" / "moth" / "desktop" / "plugin.js").read_text(encoding="utf-8")
                           .replace("http://127.0.0.1:4747", self.moth.url), encoding="utf-8")
            r = self.node(driver, src, self.moth.url, "plugin.mjs")
        texts = [s[0] for s in r["said"]]
        self.assertEqual(texts[0], "ooh, one")
        self.assertNotIn("old line", texts)
        self.assertFalse(any("claude" in t or "hi back" in t for t in texts), texts)
        self.assertTrue(all("source=hermes" in p for p in self.says_polls()))
        self.assertTrue(all(s[1] == "moth" and 1000 <= s[2] <= 30000 for s in r["said"]))
        self.assertTrue(all(len(t) <= 120 for t in texts), texts)
        self.assertEqual(len(texts), 1 + 3)            # the long line became 3 pieces, shown one after another
        self.assertEqual(len(r["delays"]), 2)
        self.assertEqual({p.split("?")[0] for m, p in self.moth.paths if m == "GET"}, {"/api/says"})   # no token

    @unittest.skipUnless(shutil.which("node"), "node not installed")
    def test_hermes_desktop_pet_moth_down(self):
        driver = r"""
import plugin from "./plugin.mjs";
let tick = null; const said = [];
plugin.register({ pet: { say: (t) => said.push(t) }, setInterval: (fn) => { tick = fn; }, setTimeout: () => () => {} });
await new Promise((r) => setTimeout(r, 300)); tick(); await new Promise((r) => setTimeout(r, 300));
console.log(JSON.stringify({ said }));
"""
        with tempfile.TemporaryDirectory() as d:
            src = (HERE / "hermes" / "moth" / "desktop" / "plugin.js").read_text(encoding="utf-8")
            (Path(d) / "plugin.mjs").write_text(src.replace("http://127.0.0.1:4747", closed_url()), encoding="utf-8")
            (Path(d) / "driver.mjs").write_text(driver, encoding="utf-8")
            p = subprocess.run(["node", str(Path(d) / "driver.mjs")], capture_output=True, text=True, encoding="utf-8",
                               timeout=10)
        self.assertEqual((p.returncode, p.stderr, json.loads(p.stdout)), (0, "", {"said": []}))

    PI_DRIVER = r"""
import ext from "./adapter.mjs";
const h = {}, cmds = {}, widgets = [], notes = [];
ext({ on: (n, f) => { (h[n] ||= []).push(f); }, registerCommand: (n, o) => { cmds[n] = o; } });
const ctx = { hasUI: true, ui: { setWidget: (k, lines) => widgets.push([k, lines]), notify: (m) => notes.push(m) },
  sessionManager: { getSessionId: () => "pi-s1" }, cwd: "/w/shop" };
const fire = (n, e) => h[n].forEach((f) => f(e, ctx));
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
fire("agent_start", {}); await sleep(200);
fire("agent_end", { messages: [{ role: "assistant", stopReason: "stop", content: [{ type: "text", text: "pi done" }] }] });
fire("agent_settled", { aborted: false }); await sleep(1500);
await cmds.moth.handler("hey", ctx); await cmds.moth.handler("", ctx);
const bare = { hasUI: false, ui: { notify: (m) => notes.push(m) } };   // no widget (print mode): the reply is a notice
await cmds.moth.handler("plain", bare);
console.log(JSON.stringify({ widgets, notes, desc: cmds.moth.description }));
"""

    @unittest.skipUnless(shutil.which("node"), "node not installed")
    def test_pi_widget_and_command(self):
        r = self.node(self.PI_DRIVER, HERE / "pi" / "moth.ts", self.moth.url, env=self.ON)
        lines = [w[1][0] for w in r["widgets"]]
        self.assertTrue(all(w[0] == "moth" for w in r["widgets"]))
        self.assertIn("\x1b[33mmoth>\x1b[0m ooh, pi done", lines)
        self.assertEqual(lines[-1], "\x1b[33mmoth>\x1b[0m hi back: hey")
        self.assertIn("usage", r["notes"][0])
        self.assertEqual(r["notes"][1], "moth> hi back: plain")
        self.assertEqual({p.split("?")[0] for m, p in self.moth.paths if m == "GET"}, {"/api/says"})
        r = self.node(self.PI_DRIVER, HERE / "pi" / "moth.ts", closed_url(), env=self.ON)
        self.assertEqual(r["widgets"], [])
        self.assertIn("asleep", r["notes"][0])

    @unittest.skipUnless(shutil.which("node"), "node not installed")
    def test_pi_off_by_default_posts_but_draws_nothing(self):
        """No env, no config file: the turn is still posted (the pet reacts), but nothing is polled or drawn.
        /moth is registered anyway and answers with a one-off notice, never the widget."""
        r = self.node(self.PI_DRIVER, HERE / "pi" / "moth.ts", self.moth.url)
        self.assertIn({"agent": "pi done"}, self.moth.log)   # the stub's card keeps only the text
        self.assertEqual(r["widgets"], [])
        self.assertEqual(r["notes"][0], "moth> hi back: hey")
        self.assertIn("usage", r["notes"][1])
        self.assertEqual(r["notes"][2], "moth> hi back: plain")
        self.assertEqual(self.says_polls(), [])
        Path(self.cfg).write_text('{"show_in_harness": true}')   # the env beats the file, both ways
        r = self.node(self.PI_DRIVER, HERE / "pi" / "moth.ts", self.moth.url, env={"MOTH_SHOW_IN_HARNESS": "0"})
        self.assertEqual((r["widgets"], self.says_polls()), ([], []))
        Path(self.cfg).write_text('{"show_in_harness": "off"}')
        r = self.node(self.PI_DRIVER, HERE / "pi" / "moth.ts", self.moth.url, env={"MOTH_SHOW_IN_HARNESS": "yes"})
        self.assertIn("\x1b[33mmoth>\x1b[0m ooh, pi done", [w[1][0] for w in r["widgets"]])

    @unittest.skipUnless(shutil.which("node"), "node not installed")
    def test_pi_live_toggle_needs_no_restart(self):
        """One Pi process: on, then off (the widget is taken down), then on again. The reaction lands 1 s after the
        post, so a cursor kept from before the off turn would show the off turn's line instead of the new one."""
        self.moth.close()
        self.moth = LogStub(react=1.0)
        driver = r"""
import ext from "./adapter.mjs";
import { writeFileSync } from "node:fs";
const h = {}, widgets = [];
ext({ on: (n, f) => { (h[n] ||= []).push(f); }, registerCommand: () => {} });
const ctx = { hasUI: true, ui: { setWidget: (k, lines) => widgets.push(lines === undefined ? null : lines[0]),
  notify: () => {} }, sessionManager: { getSessionId: () => "pi-s1" } };
const fire = (n, e) => h[n].forEach((f) => f(e, ctx));
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const set = (v) => writeFileSync(process.env.MOTH_CONFIG_FILE, JSON.stringify({ other: 1, show_in_harness: v }));
const turn = async (text) => {
  fire("agent_start", {}); await sleep(200);
  fire("agent_end", { messages: [{ role: "assistant", stopReason: "stop", content: [{ type: "text", text }] }] });
  fire("agent_settled", { aborted: false }); await sleep(1900);
};
set("on"); await turn("one"); set(false); await turn("two"); set(true); await turn("three");
console.log(JSON.stringify({ widgets }));
"""
        r = self.node(driver, HERE / "pi" / "moth.ts", self.moth.url)
        y = "\x1b[33mmoth>\x1b[0m "
        self.assertEqual(r["widgets"], [y + "ooh, one", None, y + "ooh, three"])
        self.assertEqual([e["agent"] for e in self.moth.log if "agent" in e], ["one", "two", "three"])

    @unittest.skipUnless(shutil.which("node"), "node not installed")
    def test_pi_widget_shows_only_its_own_session(self):
        """Another agent finishes right after Pi: moth's reply to it lands last, but Pi's widget shows the reply to
        Pi's own turn (live 2026-10-18: Claude Code's band showed moth's reply to Hermes). No session id: nothing."""
        self.moth.close()
        self.moth = LogStub(react=0.6)
        driver = r"""
import ext from "./adapter.mjs";
const h = {}, widgets = [];
ext({ on: (n, f) => { (h[n] ||= []).push(f); }, registerCommand: () => {} });
const mk = (sid) => ({ hasUI: true, ui: { setWidget: (k, l) => widgets.push(l[0]), notify: () => {} },
  sessionManager: { getSessionId: () => sid } });
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const other = (text) => fetch(`${process.env.MOTH_URL}/api/agent`, { method: "POST",
  headers: { "X-Moth-Token": "test-token-not-secret" }, body: JSON.stringify({ text, source: "claude-code", session: "cc1" }) });
const turn = async (ctx, text) => {
  h.agent_start.forEach((f) => f({}, ctx)); await sleep(200);
  h.agent_end.forEach((f) => f({ messages: [{ role: "assistant", stopReason: "stop", content: [{ type: "text", text }] }] }, ctx));
  h.agent_settled.forEach((f) => f({ aborted: false }, ctx));
  await sleep(100); await other("claude " + text); await sleep(1800);
};
await turn(mk("pi-s1"), "mine");
await turn(mk(""), "no id");   // an older Pi without a session id: no way to tell, so no widget
console.log(JSON.stringify({ widgets }));
"""
        r = self.node(driver, HERE / "pi" / "moth.ts", self.moth.url, env=self.ON)
        self.assertEqual(r["widgets"], ["\x1b[33mmoth>\x1b[0m ooh, mine"])
        self.assertTrue(any("source=pi&session=pi-s1" in p for p in self.says_polls()), self.says_polls())
        self.assertEqual(len([e for e in self.moth.log if "say" in e]), 4)   # the other reactions did happen

    @unittest.skipUnless(shutil.which("node"), "node not installed")
    def test_setting_spellings_match_python(self):
        """Pi's enabled() and moth/config.py read the same env / file values the same way."""
        sys.path.insert(0, str(HERE.parents[1]))
        from moth import config
        cases = [("", None), ("", '{"show_in_harness": true}'), ("", '{"show_in_harness": "On"}'),
                 ("", '{"show_in_harness": "nah"}'), ("", '{"show_in_harness": 1}'), ("", "not json"), ("", "[]"),
                 ("0", '{"show_in_harness": true}'), ("yes", '{"show_in_harness": false}'), ("ON", None),
                 ("off", None), ("maybe", '{"show_in_harness": true}'), (" true ", None), ("no", None)]
        driver = r"""
import { enabled } from "./adapter.mjs";
import { writeFileSync, rmSync } from "node:fs";
const out = [];
for (const [env, file] of JSON.parse(process.env.CASES)) {
  process.env.MOTH_SHOW_IN_HARNESS = env;
  if (file === null) rmSync(process.env.MOTH_CONFIG_FILE, { force: true });
  else writeFileSync(process.env.MOTH_CONFIG_FILE, file);
  out.push(enabled());
}
console.log(JSON.stringify(out));
"""
        js = self.node(driver, HERE / "pi" / "moth.ts", self.moth.url, env={"CASES": json.dumps(cases)})
        py = []
        old = os.environ.get("MOTH_CONFIG_FILE")
        try:
            os.environ["MOTH_CONFIG_FILE"] = self.cfg
            for env, file in cases:
                os.environ["MOTH_SHOW_IN_HARNESS"] = env
                if file is None:
                    Path(self.cfg).unlink(missing_ok=True)
                else:
                    Path(self.cfg).write_text(file)
                py.append(config.show_in_harness())
        finally:
            os.environ.pop("MOTH_SHOW_IN_HARNESS", None)
            os.environ["MOTH_CONFIG_FILE"] = old
        self.assertEqual(js, py)
        self.assertEqual(py, [False, True, True, False, False, False, False, False, True, True, False, True, True,
                              False])

    OC_DRIVER = r"""
import { Moth } from "./adapter.mjs";
const toasts = [];
const client = { tui: { showToast: async (o) => { toasts.push(o.body); return {}; } } };
const hooks = await Moth({ client, project: {}, directory: "/x", worktree: "/x", $: null });
const ev = (type, properties) => hooks.event({ event: { type, properties } });
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
await ev("session.status", { sessionID: "s1", status: { type: "busy" } }); await sleep(200);
await ev("message.updated", { info: { id: "a1", sessionID: "s1", role: "assistant" } });
await ev("message.part.updated", { part: { id: "p1", messageID: "a1", sessionID: "s1", type: "text", text: "oc done" } });
const t = Date.now(); await ev("session.status", { sessionID: "s1", status: { type: "idle" } });
const waited = Date.now() - t; await sleep(1500);
console.log(JSON.stringify({ toasts, waited }));
"""

    @unittest.skipUnless(shutil.which("node"), "node not installed")
    def test_opencode_toast(self):
        r = self.node(self.OC_DRIVER, HERE / "opencode" / "moth.js", self.moth.url, env=self.ON)
        self.assertEqual(r["toasts"], [{"title": "moth", "message": "ooh, oc done", "variant": "info", "duration": 8000}])
        self.assertLess(r["waited"], 50)             # the event hook never waits on the poll
        self.assertEqual(self.node(self.OC_DRIVER, HERE / "opencode" / "moth.js", closed_url(), env=self.ON)["toasts"],
                         [])

    @unittest.skipUnless(shutil.which("node"), "node not installed")
    def test_opencode_off_by_default_posts_but_no_toast(self):
        r = self.node(self.OC_DRIVER, HERE / "opencode" / "moth.js", self.moth.url)
        self.assertIn({"agent": "oc done"}, self.moth.log)
        self.assertEqual((r["toasts"], self.says_polls()), ([], []))
        Path(self.cfg).write_text('{"show_in_harness": "on"}')   # the file alone turns it on
        r = self.node(self.OC_DRIVER, HERE / "opencode" / "moth.js", self.moth.url)
        self.assertEqual([t["message"] for t in r["toasts"]], ["ooh, oc done"])
        r = self.node(self.OC_DRIVER, HERE / "opencode" / "moth.js", self.moth.url, env={"MOTH_SHOW_IN_HARNESS": "0"})
        self.assertEqual(r["toasts"], [])                           # ...and the env beats it

    @unittest.skipUnless(shutil.which("node"), "node not installed")
    def test_opencode_live_toggle_needs_no_restart(self):
        self.moth.close()
        self.moth = LogStub(react=1.0)
        driver = r"""
import { Moth } from "./adapter.mjs";
import { writeFileSync } from "node:fs";
const toasts = [];
const client = { tui: { showToast: async (o) => { toasts.push(o.body.message); return {}; } } };
const hooks = await Moth({ client, project: {}, directory: "/x", worktree: "/x", $: null });
const ev = (type, properties) => hooks.event({ event: { type, properties } });
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const set = (v) => writeFileSync(process.env.MOTH_CONFIG_FILE, JSON.stringify({ show_in_harness: v }));
let n = 0;
const turn = async (text) => {
  n++;
  await ev("session.status", { sessionID: "s1", status: { type: "busy" } }); await sleep(200);
  await ev("message.updated", { info: { id: "a" + n, sessionID: "s1", role: "assistant" } });
  await ev("message.part.updated", { part: { id: "p" + n, messageID: "a" + n, sessionID: "s1", type: "text", text } });
  await ev("session.status", { sessionID: "s1", status: { type: "idle" } }); await sleep(1900);
};
set(true); await turn("one"); set("0"); await turn("two"); set("1"); await turn("three");
console.log(JSON.stringify({ toasts }));
"""
        r = self.node(driver, HERE / "opencode" / "moth.js", self.moth.url)
        self.assertEqual(r["toasts"], ["ooh, one", "ooh, three"])
        self.assertEqual([e["agent"] for e in self.moth.log if "agent" in e], ["one", "two", "three"])


    @unittest.skipUnless(shutil.which("node"), "node not installed")
    def test_opencode_two_sessions_each_get_their_own_toast(self):
        """Two OpenCode sessions finish 0.1 s apart and another harness's turn lands too: each session's poll asks
        for its own reactions only and keeps its own cursor, so both toasts show, once each, and nothing else."""
        self.moth.close()
        self.moth = LogStub(react=0.6)
        driver = r"""
import { Moth } from "./adapter.mjs";
const toasts = [];
const client = { tui: { showToast: async (o) => { toasts.push(o.body.message); return {}; } } };
const hooks = await Moth({ client, project: {}, directory: "/x", worktree: "/x", $: null });
const ev = (type, properties) => hooks.event({ event: { type, properties } });
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
for (const s of ["s1", "s2"]) await ev("session.status", { sessionID: s, status: { type: "busy" } });
await sleep(200);
for (const [s, text] of [["s1", "one"], ["s2", "two"]]) {
  await ev("message.updated", { info: { id: "a" + s, sessionID: s, role: "assistant" } });
  await ev("message.part.updated", { part: { id: "p" + s, messageID: "a" + s, sessionID: s, type: "text", text } });
}
await ev("session.status", { sessionID: "s1", status: { type: "idle" } }); await sleep(100);
await ev("session.status", { sessionID: "s2", status: { type: "idle" } });
await fetch(`${process.env.MOTH_URL}/api/agent`, { method: "POST", headers: { "X-Moth-Token": "test-token-not-secret" },
  body: JSON.stringify({ text: "hermes", source: "hermes", session: "h1" }) });
await sleep(2000);
console.log(JSON.stringify({ toasts }));
"""
        r = self.node(driver, HERE / "opencode" / "moth.js", self.moth.url, env=self.ON)
        self.assertEqual(sorted(r["toasts"]), ["ooh, one", "ooh, two"])

def load_serve():
    """The real serve.py, with torch and the model code faked out (no model is loaded)."""
    import types
    from unittest import mock
    fakes = {n: types.ModuleType(n) for n in ("torch", "tokenizers", "tiny_agent.checkpoint", "tiny_agent.generate")}
    fakes["torch"].no_grad = lambda: (lambda f: f)
    fakes["tokenizers"].Tokenizer = object
    fakes["tiny_agent.checkpoint"].load_model = None
    fakes["tiny_agent.generate"].KVCache = fakes["tiny_agent.generate"].append = None
    with mock.patch.dict(sys.modules, fakes):
        spec = importlib.util.spec_from_file_location("moth_serve_http", HERE.parent / "serve.py")
        serve = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(serve)
    return serve


class Serve(unittest.TestCase):
    """serve.py's real HTTP handler in front of a fake engine."""

    @classmethod
    def setUpClass(cls):
        cls.serve = load_serve()

    def setUp(self):
        import types
        serve, self.tmp = self.serve, tempfile.mkdtemp()
        moth = types.SimpleNamespace(log=serve.Log(), lock=threading.RLock(), heard=[], chats=[],
                                     home=serve.Home(os.path.join(self.tmp, "home"), 1, None),
                                     a=types.SimpleNamespace(home=os.path.join(self.tmp, "home"), pool_file=None))
        moth.log.add({"you": "my boss is awful"}, {"agent": "Fixed the bug."}, {"say": "hi"},
                     {"tool": "bash", "args": {"command": "ls"}, "result": "diary"})
        moth.hear_agent = lambda text, source, session=None, label=None: (
            moth.heard.append((text, source, session, label)) or "queued")
        moth.agents = lambda: {"active": 1, "skipped": 0,
                               "sessions": [{"source": "codex", "session": "s9", "label": "secret-proj"}]}
        moth.say_to = lambda text: moth.chats.append(text) or [{"say": "ok"}]
        moth.forgot = []
        moth.forget = lambda everything=False: moth.forgot.append(everything) or moth.log.clear()
        self.moth = moth
        self.srv = ThreadingHTTPServer(("127.0.0.1", 0), serve.make_handler(moth, TOKEN))
        self.srv.daemon_threads = True
        self.port = self.srv.server_port
        threading.Thread(target=self.srv.serve_forever, daemon=True).start()

    def tearDown(self):
        self.srv.shutdown()
        self.srv.server_close()
        shutil.rmtree(self.tmp, ignore_errors=True)

    def req(self, method, path, headers=None, body=None):
        """-> (status, lower-cased headers, body). A header set to None is left out (Host included)."""
        import http.client
        c = http.client.HTTPConnection("127.0.0.1", self.port, timeout=3)
        c.putrequest(method, path, skip_host=True, skip_accept_encoding=True)
        h = {"Host": f"127.0.0.1:{self.port}"}
        if body is not None:
            h["Content-Length"] = str(len(body))
        h.update(headers or {})
        for k, v in h.items():
            if v is not None:
                c.putheader(k, v)
        c.endheaders(body)
        r = c.getresponse()
        data = r.read()
        c.close()
        return r.status, {k.lower(): v for k, v in r.getheaders()}, data

    def post(self, path, obj=None, raw=None, **headers):
        h = {"Content-Type": "application/json", "X-Moth-Token": TOKEN, **headers}
        return self.req("POST", path, h, raw if raw is not None else json.dumps(obj or {}).encode())

    def test_host_allow_list(self):
        for host in ("evil.example:%d", "127.0.0.1", "localhost", "127.0.0.1:1", "localhost.evil.example:%d",
                     "192.168.1.5:%d", None):
            h = host % self.port if host and "%d" in host else host
            with self.subTest(host=h):
                for method, path in (("GET", "/"), ("GET", "/api/says"), ("OPTIONS", "/api/chat")):
                    self.assertEqual(self.req(method, path, {"Host": h})[0], 403)
                self.assertEqual(self.post("/api/chat", {"text": "hi"}, Host=h)[0], 403)
        for h in (f"127.0.0.1:{self.port}", f"localhost:{self.port}", f"LOCALHOST:{self.port}"):
            self.assertEqual(self.req("GET", "/api/says", {"Host": h})[0], 200)
        self.assertEqual(self.moth.chats, [])

    def test_page_carries_token_and_no_cors(self):
        code, h, body = self.req("GET", "/", {"Origin": "http://localhost:5173"})
        self.assertEqual(code, 200)
        self.assertIn(f'content="{TOKEN}"'.encode(), body)
        self.assertNotIn(b"__MOTH_TOKEN__", body)
        self.assertEqual((h["x-frame-options"], h["cache-control"]), ("DENY", "no-store"))
        self.assertFalse([k for k in h if k.startswith("access-control")])

    def test_full_log_needs_token(self):
        for tok in (None, "", "wrong", TOKEN + "x"):
            with self.subTest(token=tok):
                self.assertEqual(self.req("GET", "/api/log?since=0", {"X-Moth-Token": tok})[0], 401)
                self.assertEqual(self.req("GET", "/api/home", {"X-Moth-Token": tok})[0], 401)
        code, h, body = self.req("GET", "/api/log?since=0", {"X-Moth-Token": TOKEN})
        d = json.loads(body)
        self.assertEqual((code, len(d["events"]), d["next"], d["epoch"]), (200, 4, 4, self.moth.log.epoch))
        self.assertEqual(h["x-frame-options"], "DENY")
        self.assertEqual(json.loads(self.req("GET", "/api/log?since=3", {"X-Moth-Token": TOKEN})[2])["events"][0]["tool"],
                         "bash")

    def test_says_view(self):
        code, h, body = self.req("GET", "/api/says?since=0", {"Origin": "null"})
        d = json.loads(body)
        self.assertEqual((code, d["events"], d["next"]), (200, [{"say": "hi"}], 4))   # no vent, no agent reply
        self.assertNotIn(b"boss", body)
        self.assertEqual(json.loads(self.req("GET", "/api/says?since=1000000000")[2])["events"], [])
        for bad in ("abc", "-1", "1e9", "1.5", "99999999999999999999"):
            with self.subTest(since=bad):
                self.assertEqual(self.req("GET", f"/api/says?since={bad}")[0], 400)
                self.assertEqual(self.req("GET", f"/api/log?since={bad}", {"X-Moth-Token": TOKEN})[0], 400)

    def test_says_per_session(self):
        """?source=&session=: only moth's reactions to that harness / session (tagged "re" by Moth.react), never chat
        or another session's line, and never the tag itself; no params: every line, as before (pet, page)."""
        self.moth.log.add({"agent": "a", "source": "claude-code", "session": "cc-1", "label": "secret-proj"},
                          {"agent": "b", "source": "hermes", "session": "h-1"},
                          {"say": "to hermes", "re": {"source": "hermes", "session": "h-1"}},
                          {"say": "to claude", "re": {"source": "claude-code", "session": "cc-1"}},
                          {"you": "hey"}, {"say": "chat reply"})
        says = lambda q: json.loads(self.req("GET", f"/api/says?since=4&{q}")[2])   # noqa: E731
        for q, want in (("", ["to hermes", "to claude", "chat reply"]),
                        ("source=claude-code&session=cc-1", ["to claude"]), ("session=cc-1", ["to claude"]),
                        ("source=Hermes", ["to hermes"]), ("source=pi", []), ("session=cc-2", []),
                        ("source=hermes&session=cc-1", []), ("session=%21%21", [])):   # cleans to nothing: no match
            with self.subTest(q=q):
                d = says(q)
                self.assertEqual(d["events"], [{"say": t} for t in want])   # stripped to {"say"}: no ids, no labels
                self.assertEqual(d["next"], 10)                               # the cursor still counts every event
        self.assertNotIn(b"cc-1", self.req("GET", "/api/says?since=0&session=cc-1")[2])
        self.assertNotIn(b"secret", self.req("GET", "/api/says?since=0&source=claude-code")[2])
        log = json.loads(self.req("GET", "/api/log?since=6&session=cc-1", {"X-Moth-Token": TOKEN})[2])
        self.assertEqual(len(log["events"]), 4)   # the full log ignores the filter

    def test_cors(self):
        local = ["http://localhost:5173", "https://localhost", "http://127.0.0.1:5174"]
        pets = ["app://hermes", "app://-", "file://", "null"]
        foreign = ["https://evil.example", "http://localhost.evil.example", "http://127.0.0.1.evil.example:80",
                   "http://192.168.1.5:4747", "https://localhost@evil.example", "app://x/../y", "NULL"]

        def acao(path, origin, **h):
            return self.req("GET", path, {"Origin": origin, **h})[1].get("access-control-allow-origin")

        for o in local:
            with self.subTest(origin=o):
                self.assertEqual(acao("/api/log?since=0", o, **{"X-Moth-Token": TOKEN}), o)
                self.assertEqual(acao("/api/says", o), o)
                code, h, _ = self.req("OPTIONS", "/api/chat", {"Origin": o, "Access-Control-Request-Method": "POST"})
                self.assertEqual((code, h["access-control-allow-origin"], h["vary"]), (204, o, "Origin"))
                self.assertEqual(h["access-control-allow-headers"], "Content-Type, X-Moth-Token")
        for o in pets:   # desktop shells: moth's lines only
            with self.subTest(origin=o):
                self.assertEqual(acao("/api/says", o), o)
                self.assertIsNone(acao("/api/log?since=0", o, **{"X-Moth-Token": TOKEN}))
                self.assertFalse([k for k in self.req("OPTIONS", "/api/chat", {"Origin": o})[1]
                                  if k.startswith("access-control")])
        for o in foreign:
            with self.subTest(origin=o):
                self.assertIsNone(acao("/api/says", o))
                self.assertIsNone(acao("/api/log?since=0", o, **{"X-Moth-Token": TOKEN}))

    def test_post_hygiene(self):
        P = "/api/chat"
        self.assertEqual(self.post(P, {"text": "hi"}, **{"X-Moth-Token": None})[0], 401)
        self.assertEqual(self.post(P, {"text": "hi"}, **{"X-Moth-Token": "nope"})[0], 401)
        for o in ("https://evil.example", "null", "file://", "app://hermes"):
            self.assertEqual(self.post(P, {"text": "hi"}, Origin=o)[0], 403)
        self.assertEqual(self.post(P, {"text": "hi"}, **{"Content-Type": "text/plain"})[0], 415)
        self.assertEqual(self.post(P, {"text": "hi"}, **{"Content-Type": None})[0], 415)
        self.assertEqual(self.post(P, raw=b"x" * (64 * 1024 + 1))[0], 413)
        self.assertEqual(self.req("POST", P, {"Content-Type": "application/json", "X-Moth-Token": TOKEN})[0], 411)
        for raw in (b"{nope", b"[1, 2]", b'"text"', b"\xff\xfe", b'{"text": 5}', b'{"text": "   "}', b"null"):
            with self.subTest(raw=raw):
                self.assertEqual(self.post(P, raw=raw)[0], 400)
        self.assertEqual(self.post("/api/nope", {"text": "hi"})[0], 404)
        self.assertEqual(self.moth.chats, [])
        code, h, body = self.post(P, {"text": " hi moth "}, Origin=f"http://127.0.0.1:{self.port}",
                                  **{"Content-Type": "application/json; charset=utf-8"})
        self.assertEqual((code, json.loads(body)["events"], self.moth.chats), (200, [{"say": "ok"}], ["hi moth"]))
        self.assertEqual(h["x-frame-options"], "DENY")

    def test_agent_source(self):
        for src, want in (("claude-code", "claude-code"), ("Codex", "codex"), ("evil <b>", None), (5, None), (None, None)):
            code, _, body = self.post("/api/agent", {"text": "Done.", "source": src})
            self.assertEqual((code, json.loads(body)["status"]), (200, "queued"))
            self.assertEqual(self.moth.heard[-1][:2], ("Done.", want))
        self.assertEqual(self.post("/api/agent", {"text": ""})[0], 400)

    def test_agent_session_label(self):
        for session, label, want in (("0f3c9a2e-11", "moth-repo", ("0f3c9a2e-11", "moth-repo")),
                                     ("ses_ab:12.x", "\x1b[31mmy\n  proj\x1b[0m\x07", ("ses_ab:12.x", "my proj")),
                                     ("../evil id<b>" + "x" * 99, "L" * 99, ("..evilidbx" + "x" * 70, "L" * 40)),
                                     (5, ["x"], (None, None)), ("", "   ", (None, None)), (None, None, (None, None))):
            with self.subTest(session=session, label=label):
                code = self.post("/api/agent", {"text": "Done.", "source": "codex", "session": session,
                                                "label": label})[0]
                self.assertEqual((code, self.moth.heard[-1]), (200, ("Done.", "codex") + want))
        log = json.loads(self.req("GET", "/api/log?since=0", {"X-Moth-Token": TOKEN})[2])
        self.assertEqual(log["agents"]["sessions"][0]["label"], "secret-proj")
        says = self.req("GET", "/api/says?since=0")[2]
        self.assertNotIn(b"secret-proj", says)                  # folder names only behind the token
        self.assertNotIn(b"agents", says)

    def test_reset_only_deletes_moths_own_home(self):
        home = self.moth.home.root
        e0 = self.moth.log.epoch
        code, _, body = self.post("/api/reset", {})
        self.assertEqual((code, json.loads(body)["epoch"] != e0, len(self.moth.log)), (200, True, 0))
        with open(os.path.join(home, "diary", "mine.txt"), "w") as f:
            f.write("keep me")
        self.assertEqual(self.post("/api/reset", {"new_home": True})[0], 200)   # moth made it: replaced
        self.assertFalse(os.path.exists(os.path.join(home, "diary", "mine.txt")))
        self.assertTrue(self.moth.home.owned)
        os.remove(os.path.join(home, self.serve.MARKER))   # e.g. an older home, or --home pointed at your own folder
        with open(os.path.join(home, "precious.txt"), "w") as f:
            f.write("mine")
        code, _, body = self.post("/api/reset", {"new_home": True})
        self.assertEqual(code, 409)
        self.assertIn("won't delete", json.loads(body)["error"])
        self.assertTrue(os.path.exists(os.path.join(home, "precious.txt")))
        self.assertEqual(self.post("/api/reset", {"new_home": True}, **{"X-Moth-Token": None})[0], 401)

    def test_reset_everything_is_moth_forget(self):
        """"new chat" keeps moth's phrase memory; {"everything": true} (what `moth forget` sends) wipes it too."""
        self.assertEqual(self.post("/api/reset", {})[0], 200)
        self.assertEqual(self.post("/api/reset", {"everything": "yes"})[0], 200)   # only a real true counts
        self.assertEqual(self.post("/api/reset", {"everything": True})[0], 200)
        self.assertEqual(self.moth.forgot, [False, False, True])
        self.assertEqual(self.post("/api/reset", {"everything": True}, **{"X-Moth-Token": None})[0], 401)
        self.assertEqual(self.moth.forgot, [False, False, True])


class AdapterToLog(unittest.TestCase):
    """Labels end to end: every adapter, fed its harness's payload, against serve.py's real handler and real
    Moth.hear_agent (a Moth without weights or tokenizer: only the agent-message half, react_every 0, no window)."""

    @classmethod
    def setUpClass(cls):
        cls.serve = load_serve()

    def setUp(self):
        serve, self.tmp = self.serve, tempfile.mkdtemp()
        m = serve.Moth.__new__(serve.Moth)
        m.log, m.lock, m.wake, m.clock = serve.Log(), threading.RLock(), threading.Condition(threading.RLock()), time.monotonic
        m.sessions, m.pending, m.batch, m.batch_until = {}, None, None, 0.0
        m.react_every, m.batch_window, m.last_react = 0, 0, float("-inf")
        m.home = serve.Home(os.path.join(self.tmp, "home"), 1, None)
        self.m = m
        self.srv = ThreadingHTTPServer(("127.0.0.1", 0), serve.make_handler(m, TOKEN))
        self.srv.daemon_threads = True
        self.url = f"http://127.0.0.1:{self.srv.server_port}"
        threading.Thread(target=self.srv.serve_forever, daemon=True).start()

    def tearDown(self):
        self.srv.shutdown()
        self.srv.server_close()
        shutil.rmtree(self.tmp, ignore_errors=True)

    def cards(self, n):
        for _ in range(200):
            cards = [e for e in self.m.log.since(0)["events"] if "agent" in e]
            if len(cards) >= n:
                return cards
            time.sleep(0.01)
        return cards

    def test_labels_flow_to_the_log(self):
        want = []
        run([sys.executable, str(HERE / "claude_code.py")], self.url,
            json.dumps({"session_id": "cc-1", "cwd": "/home/u/moth-repo", "last_assistant_message": "claude done"}))
        want.append(("claude done", "claude-code", "cc-1", "moth-repo"))
        run([sys.executable, str(HERE / "codex.py")], self.url,
            json.dumps({"hook_event_name": "Stop", "session_id": "cx-1", "cwd": "/srv/api", "last_assistant_message": "codex done"}))
        want.append(("codex done", "codex", "cx-1", "api"))
        spec = importlib.util.spec_from_file_location("moth_hermes_e2e", HERE / "hermes" / "moth" / "__init__.py")
        hermes = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(hermes)
        hermes.URL, hermes._cwd = self.url + "/api/agent", lambda: "/home/u/notes"
        hermes.on_post_llm_call(assistant_response="hermes done", platform="cli", session_id="hm-1")
        want.append(("hermes done", "hermes", "hm-1", "notes"))
        self.cards(3)
        if shutil.which("node"):
            for harness, (src, driver, expect) in JsAdapters.CASES.items():
                with tempfile.TemporaryDirectory() as d:
                    shutil.copy(src, Path(d) / "adapter.mjs")
                    (Path(d) / "driver.mjs").write_text(driver, encoding="utf-8")
                    run(["node", str(Path(d) / "driver.mjs")], self.url)
                want += [(t, JsAdapters.SOURCE[harness]) + JsAdapters.META[harness] for t in expect]
                self.cards(len(want))
        got = [(c["agent"], c.get("source"), c.get("session"), c.get("label")) for c in self.cards(len(want))]
        self.assertEqual(sorted(got), sorted(want))
        self.assertEqual(self.m.agents()["active"], len({w[1:3] for w in want}))


if __name__ == "__main__":
    unittest.main(verbosity=2)