CST-agent/scripts/e2e_webui_ext.py

"""WebUI /init, /skill, extensions and the thinking switch against a fake API.

  python3 scripts/e2e_webui_ext.py      (CST_AGENT= picks the binary)

Checks: show_thinking is validated, lands in the state and in settings.json;
/api/skill inlines the skill and 404s an unknown one; /api/init sends the
AGENTS.md prompt; /api/extensions lists skills and the built-in roles;
trust and reload work and write settings.json; the changing calls are a 409
while a turn runs, the listing is not. HOME points at a temp dir.
"""
import json, os, shutil, subprocess, sys, tempfile, threading, time, urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

BIN = os.environ.get("CST_AGENT", os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "cst-agent"))
SEEN = []   # the full last user message of every request


class Api(BaseHTTPRequestHandler):
    """Answers "好的"; a prompt with "slow" takes 2 s."""
    def log_message(self, *a):
        pass

    def do_GET(self):
        body = json.dumps({"data": [{"id": "deepseek-flash"}]}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        req = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))) or b"{}")
        users = [m.get("content") for m in req.get("messages", []) if m.get("role") == "user"]
        last = users[-1] if users and isinstance(users[-1], str) else ""
        SEEN.append(last)
        if "slow" in last:
            time.sleep(2)
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.end_headers()
        for o in ({"delta": {"role": "assistant", "content": "好的"}},
                  {"delta": {}, "finish_reason": "stop"}):
            chunk = {"id": "x", "object": "chat.completion.chunk", "model": "deepseek-flash",
                     "choices": [dict(index=0, **o)]}
            self.wfile.write(b"data: " + json.dumps(chunk, ensure_ascii=False).encode() + b"\n\n")
            self.wfile.flush()
        self.wfile.write(b"data: [DONE]\n\n")
        self.wfile.flush()


fails = 0


def check(ok, msg):
    global fails
    print(("PASS " if ok else "FAIL ") + msg)
    fails += 0 if ok else 1


def wait(fn, t=15):
    end = time.time() + t
    while time.time() < end:
        v = fn()
        if v:
            return v
        time.sleep(0.1)
    return None


def call(base, path, obj=None, method=None):
    data = None if obj is None else json.dumps(obj).encode()
    req = urllib.request.Request(base + path, data=data, method=method or ("POST" if data else "GET"),
                                 headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return r.status, json.loads(r.read() or b"{}")
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read() or b"{}")
        except ValueError:
            return e.code, {}


def skill(ws, name, desc, body):
    d = os.path.join(ws, ".agent", "skills", name)
    os.makedirs(d, exist_ok=True)
    with open(os.path.join(d, "SKILL.md"), "w", encoding="utf-8") as f:
        f.write("---\nname: %s\ndescription: %s\n---\n\n%s\n" % (name, desc, body))


api = ThreadingHTTPServer(("127.0.0.1", 0), Api)
threading.Thread(target=api.serve_forever, daemon=True).start()
ws = tempfile.mkdtemp(prefix="cst-ext-")
home = tempfile.mkdtemp(prefix="cst-home-")
skill(ws, "demo", "演示用的 skill", "DEMO-INSTRUCTIONS: 先说你好。")
env = dict(os.environ, HOME=home, XDG_CONFIG_HOME=home + "/.config",
           XDG_DATA_HOME=home + "/.local/share", XDG_CACHE_HOME=home + "/.cache")
for k in ("DEEPSEEK_API_KEY", "OPENAI_API_KEY", "CST_API_KEY"):
    env.pop(k, None)
log = open(ws + "/server.log", "w+")
proc = subprocess.Popen([BIN, "--web", "--web-port", "0", "--workspace", ws, "--new", "--api-key", "sk-x",
                         "--provider", "deepseek", "--model", "deepseek-flash",
                         "--api-base", "http://127.0.0.1:%d/v1" % api.server_address[1],
                         "--protocol", "native-eos", "--approval", "auto"],
                        stdout=log, stderr=subprocess.STDOUT, env=env)


def port():
    log.seek(0)
    for l in log.read().splitlines():
        if l.startswith("webui: http://127.0.0.1:"):
            return l.split(":")[3].split("/")[0]


def settings():
    for name in ("hermit", "cst-agent"):
        path = home + "/.config/%s/settings.json" % name
        try:
            with open(path, encoding="utf-8") as f:
                return json.load(f)
        except OSError:
            continue
    return {}


def busy(base):
    return call(base, "/api/state")[1].get("status_busy")


def turn(base, path, body):
    code, d = call(base, path, body)
    if code == 202:
        time.sleep(0.3)
        wait(lambda: not busy(base), 20)
    return code, d


try:
    base = "http://127.0.0.1:%s" % wait(port)

    # 1. the thinking switch
    st = call(base, "/api/state")[1]
    check(st.get("show_thinking") is False, "show_thinking starts off")
    check(call(base, "/api/config", {"show_thinking": "yes"})[0] == 400, "show_thinking must be a bool")
    code, _ = call(base, "/api/config", {"show_thinking": True})
    check(code == 200 and call(base, "/api/state")[1].get("show_thinking") is True, "show_thinking on")
    check(settings().get("show_thinking") is True, "show_thinking is saved in settings.json")
    call(base, "/api/config", {"show_thinking": False})
    check(settings() and not settings().get("show_thinking"), "and cleared again (%s)" % settings())

    # 2. the extensions listing
    code, ex = call(base, "/api/extensions")
    check(code == 200 and ex.get("enabled") is True and ex.get("trusted") is False,
          "extensions: enabled, not trusted (%s)" % code)
    names = [s["name"] for s in ex.get("skills", [])]
    check(names == ["demo"] and ex["skills"][0].get("desc") == "演示用的 skill"
          and ex["skills"][0].get("source") == "workspace", "the workspace skill is listed (%s)" % ex.get("skills"))
    roles = [r["name"] for r in ex.get("roles", [])]
    check("explore" in roles and all(r.get("source") for r in ex["roles"]), "built-in roles are listed (%s)" % roles)
    check(ex.get("mcp") == [] and ex.get("plugins") == [] and ex.get("runs") == [], "no MCP, plugins or runs")
    check(ex.get("task_tool") is True and "1 skill" in ex.get("report", ""), "task tool on, report %r" % ex.get("report"))

    # 3. /skill
    code, d = turn(base, "/api/skill", {"name": "demo", "task": "做点事"})
    check(code == 202 and "DEMO-INSTRUCTIONS" in d.get("text", "") and d["text"].endswith("做点事"),
          "skill turn accepted (%s)" % code)
    check(SEEN and "DEMO-INSTRUCTIONS" in SEEN[-1] and "做点事" in SEEN[-1], "the model saw the skill and the task")
    code, d = turn(base, "/api/skill", {"name": "demo"})
    check(code == 202 and SEEN[-1].endswith("Apply this skill now."), "a skill without a task")
    check(call(base, "/api/skill", {"name": "nope"})[0] == 404, "unknown skill: 404")
    check(call(base, "/api/skill", {"name": "../demo"})[0] == 404, "skill ../demo: 404")
    check(call(base, "/api/skill", {})[0] == 400, "skill without a name: 400")
    check(call(base, "/api/skill", {"name": "x" * 100})[0] == 400, "a long skill name: 400")
    check(call(base, "/api/skill", {"name": "demo", "task": 5})[0] == 400, "a non-string task: 400")

    # 4. /init
    code, d = turn(base, "/api/init", {})
    check(code == 202 and "AGENTS.md" in d.get("text", ""), "init accepted (%s)" % code)
    check("AGENTS.md" in SEEN[-1], "the model saw the AGENTS.md prompt")

    # 5. trust and reload
    code, d = call(base, "/api/extensions/trust", {"on": True})
    check(code == 200 and d.get("trusted") is True, "trust the workspace (%s %s)" % (code, d))
    tw = settings().get("trusted_workspaces", [])
    check(any(os.path.realpath(ws) == os.path.realpath(p) for p in tw), "settings.json lists it (%s)" % tw)
    check(call(base, "/api/extensions")[1].get("trusted") is True, "the listing says trusted")
    code, d = call(base, "/api/extensions/trust", {"on": False})
    check(code == 200 and d.get("trusted") is False and not settings().get("trusted_workspaces"),
          "untrust (%s)" % settings().get("trusted_workspaces"))
    check(call(base, "/api/extensions/trust", {"on": "yes"})[0] == 400, "trust needs a bool")
    skill(ws, "zeta", "第二个", "ZETA")
    code, d = call(base, "/api/extensions/reload", {})
    check(code == 200 and "2 skills" in d.get("report", ""), "reload sees the new skill (%s)" % d)
    check([s["name"] for s in call(base, "/api/extensions")[1]["skills"]] == ["demo", "zeta"], "both listed")

    # 6. while a turn runs
    call(base, "/api/turn", {"text": "slow 慢一点"})
    time.sleep(0.4)
    check(call(base, "/api/extensions/reload", {})[0] == 409, "reload during a turn: 409")
    check(call(base, "/api/extensions/trust", {"on": True})[0] == 409, "trust during a turn: 409")
    check(call(base, "/api/skill", {"name": "demo"})[0] == 409, "skill during a turn: 409")
    check(call(base, "/api/init", {})[0] == 409, "init during a turn: 409")
    check(call(base, "/api/extensions")[0] == 200, "the listing works during a turn")
    check(call(base, "/api/config", {"show_thinking": True})[0] == 200, "the thinking switch works during a turn")
    wait(lambda: not busy(base), 20)
    check(not settings().get("trusted_workspaces"), "the refused trust changed nothing")

    # 7. another workspace brings its own skills
    ws2 = tempfile.mkdtemp(prefix="cst-ext2-")
    skill(ws2, "other", "另一个工作区的", "OTHER")
    code, _ = call(base, "/api/workspace", {"path": ws2})
    ex = call(base, "/api/extensions")[1]
    check(code == 200 and [s["name"] for s in ex.get("skills", [])] == ["other"]
          and ex.get("report") == "1 skill", "a workspace switch reloads the extensions (%s)" % ex.get("report"))
    code, d = turn(base, "/api/skill", {"name": "other"})
    check(code == 202 and "OTHER" in SEEN[-1], "and its skill runs")
    shutil.rmtree(ws2, ignore_errors=True)
finally:
    proc.terminate()
    try:
        proc.wait(20)
    except subprocess.TimeoutExpired:
        proc.kill()
        check(False, "server hung on SIGTERM")
    log.close()
    shutil.rmtree(ws, ignore_errors=True)
    shutil.rmtree(home, ignore_errors=True)
    api.shutdown()
print("FAILED:", fails)
sys.exit(1 if fails else 0)