clanker-saddle/test/web_stack_test.py

#!/usr/bin/env python3
"""Self-contained web-stack feature tests: app + fake saddle daemon.

Unlike integration_test.py (which needs a real saddle daemon and a mock
LLM), this suite spawns everything it needs:

- the clanker-saddle app binary (path from argv[1], or builddir default)
  in a scratch directory with its own web-config.json (statum keys copied
  from the repo, plus `saddle_callback_base_url`) and SQLite database
- an in-process fake saddle daemon speaking the wire contract: POST
  /start (captures the payload), GET /status (scripted state machine),
  GET /output (empty), GET /log?since= (raw PTY chunks, one new chunk
  per request so since-paging is exercised)

It then drives the app the way statum.js does (entrypoint loads + action
POSTs) and asserts the Workstream B features against the wire contract:

3. plan-review blocks hide their duplicate body preview
5. rules CRUD/attach + the effective `rules` array in /start
6. project notes, POST /api/saddle/note, `project_notes` + `callback`
   in /start
   plus the raw-log side: session_raw_log polling + terminal raw view

Prints PASS/FAIL lines Statum-style; exits non-zero on any failure.
"""
import json
import os
import shutil
import socket
import sqlite3
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
APP = os.path.abspath(sys.argv[1]) if len(sys.argv) > 1 else os.path.join(REPO, "builddir", "clanker-saddle")
WORK = os.environ.get("WEBSTACK_WORK", "/tmp/kilo/webstack-e2e")
DB_PATH = os.path.join(WORK, "web.sqlite")

passes = 0
failures = 0


def check(condition, label):
    global passes, failures
    if condition:
        passes += 1
        print("PASS %s" % label)
    else:
        failures += 1
        print("FAIL %s" % label)


def free_port():
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]


APP_PORT = free_port()
DAEMON_PORT = free_port()
WEB = "http://127.0.0.1:%d" % APP_PORT


def http(method, url, data=None, headers=None, timeout=30):
    body = None
    if data is not None:
        body = "&".join("%s=%s" % (k, urllib.parse.quote(str(v))) for k, v in data.items()).encode()
    request = urllib.request.Request(WEB + url if url.startswith("/") else url,
                                     data=body, method=method)
    if data is not None:
        request.add_header("Content-Type", "application/x-www-form-urlencoded")
    for key, value in (headers or {}).items():
        request.add_header(key, value)
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            return response.status, response.read().decode()
    except urllib.error.HTTPError as error:
        return error.code, error.read().decode()


def directives(uri, auth=None):
    headers = {}
    if auth:
        headers["X-Statum-Slot"] = "key=%s; as_at=%s" % (auth["key"], auth["as_at"])
    status, body = http("GET", "/_statum/entrypoint?uri=" + urllib.parse.quote(uri.split("?")[0]),
                        headers=headers)
    if status != 200:
        raise SystemExit("entrypoint %s -> HTTP %d: %s" % (uri, status, body[:300]))
    return json.loads(body)


def slot_public(auth, uri, type_name):
    for directive in directives(uri, auth):
        if directive["type"] == "set":
            content = json.loads(directive["snapshot_frame"]["content"])
            if content["slot"]["type"] == type_name:
                return content["public"]
    raise SystemExit("no %s slot in entrypoint state for %s" % (type_name, uri))


def act(auth, action, fields):
    headers = {"X-Statum-Slot": "key=%s; as_at=%s" % (auth["key"], auth["as_at"])}
    if action.get("private"):
        headers["X-Statum-Private"] = action["private"]
    status, body = http(action.get("method", "POST"), action["uri"], fields, headers)
    if status != 200:
        raise SystemExit("action %s -> HTTP %d: %s" % (action["uri"], status, body[:300]))
    return json.loads(body)


def login(username, password):
    frames = directives("/login")
    content = json.loads(frames[0]["snapshot_frame"]["content"])
    action = content["public"]["login"]
    status, body = http(action.get("method", "POST"), action["uri"],
                        {"username": username, "password": password})
    if status != 200:
        raise SystemExit("login -> HTTP %d: %s" % (status, body[:300]))
    for directive in json.loads(body):
        if directive["type"] == "set":
            frame = json.loads(directive["snapshot_frame"]["content"])
            if frame["slot"]["type"] == "auth":
                return {"key": frame["slot"]["key"], "as_at": frame["as_at"]}
    raise SystemExit("login response carried no auth slot: " + body[:300])


def wait_for(predicate, timeout=60, what="condition", interval=0.5):
    deadline = time.monotonic() + timeout
    last = None
    while time.monotonic() < deadline:
        last = predicate()
        if last is not False and last is not None:
            return last
        time.sleep(interval)
    print("  (timeout waiting for %s; last: %s)" % (what, last))
    return None


def task_state(auth, task_id):
    return slot_public(auth, "/tasks/%d" % task_id, "task")


def saddle_api(method, path, token, payload=None):
    """Raw call against the saddle ingest API; returns (status, json)."""
    request = urllib.request.Request(WEB + path,
                                     data=json.dumps(payload or {}).encode(), method=method)
    request.add_header("Content-Type", "application/json")
    if token is not None:
        request.add_header("Authorization", "Bearer " + token)
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            return response.status, json.loads(response.read().decode() or "{}")
    except urllib.error.HTTPError as error:
        body = error.read().decode()
        try:
            return error.code, json.loads(body)
        except ValueError:
            return error.code, {"raw": body}


def db():
    return sqlite3.connect(DB_PATH)


class FakeSaddle:
    """In-process saddle daemon implementing the wire contract."""

    PROMPT = "bbarrow@fedora:/workspace/task-1$ "

    def __init__(self):
        self.lock = threading.Lock()
        self.start_payloads = []       # captured POST /start bodies
        self.chunks = []               # [{seq, text}] — the raw log ring
        self.next_seq = 1
        self.state = "idle"
        self.summary = ""
        self.status_polls_after_start = 0

    def append_chunk(self, text):
        with self.lock:
            self.chunks.append({"seq": self.next_seq, "text": text})
            self.next_seq += 1

    def log_since(self, since):
        with self.lock:
            chunks = [c for c in self.chunks if c["seq"] > since]
            return chunks, self.next_seq

    class Handler(BaseHTTPRequestHandler):
        daemon = None  # set on the class before serving

        def log_message(self, fmt, *args):
            pass

        def _send(self, status, obj):
            body = json.dumps(obj).encode()
            self.send_response(status)
            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):
            length = int(self.headers.get("Content-Length") or 0)
            raw = self.rfile.read(length).decode() if length else "{}"
            if self.path == "/start":
                try:
                    payload = json.loads(raw)
                except ValueError:
                    payload = {"raw": raw}
                self.daemon.start_payloads.append(payload)
                self.daemon.append_chunk(self.daemon.PROMPT)
                self.daemon.append_chunk("echo hello\r\n")
                self.daemon.append_chunk("hello\r\n")
                self.daemon.state = "working"
                self.daemon.status_polls_after_start = 0
                self._send(200, {})
            elif self.path == "/answer":
                self._send(200, {})
            else:
                self._send(404, {"error": "no such route"})

        def do_GET(self):
            if self.path == "/status":
                if self.daemon.state == "working":
                    self.daemon.status_polls_after_start += 1
                    if self.daemon.status_polls_after_start >= 2:
                        self.daemon.state = "done"
                        self.daemon.summary = "fake daemon finished"
                self._send(200, {"state": self.daemon.state,
                                 "summary": self.daemon.summary})
            elif self.path.startswith("/output"):
                self._send(200, [])
            elif self.path.startswith("/log"):
                query = urllib.parse.urlparse(self.path).query
                since = int(urllib.parse.parse_qs(query).get("since", ["0"])[0])
                chunks, nxt = self.daemon.log_since(since)
                # Simulate the live PTY: each poll produces one more chunk.
                self.daemon.append_chunk("stream-line-%d\r\n" % self.daemon.next_seq)
                self._send(200, {"chunks": chunks, "next": nxt})
            else:
                self._send(404, {"error": "no such route"})

    def serve(self):
        handler = type("BoundHandler", (FakeSaddle.Handler,), {"daemon": self})
        server = ThreadingHTTPServer(("127.0.0.1", DAEMON_PORT), handler)
        threading.Thread(target=server.serve_forever, daemon=True).start()
        return server


def start_app():
    config = json.load(open(os.path.join(REPO, "web-config.json")))
    config["saddle_callback_base_url"] = WEB
    with open(os.path.join(WORK, "web-config.json"), "w") as handle:
        json.dump(config, handle)

    env = dict(os.environ)
    env.update({
        "SPRY_DB_PATH": DB_PATH,
        "SPRY_SEED_ADMIN_USERNAME": "admin",
        "SPRY_SEED_ADMIN_PASSWORD": "adminpass123",
        "SADDLE_LLM_BASE_URL": "http://127.0.0.1:%d" % DAEMON_PORT,
        "SADDLE_PROJECTS_ROOT": os.path.join(WORK, "workspace"),
    })
    log = open(os.path.join(WORK, "app.log"), "w")
    process = subprocess.Popen([APP, str(APP_PORT)], cwd=WORK, env=env,
                               stdout=log, stderr=subprocess.STDOUT)
    if wait_for(lambda: app_ready() or False, timeout=30, what="app ready") is None:
        raise SystemExit("app did not come up; see %s/app.log" % WORK)
    return process


def app_ready():
    try:
        status, _ = http("GET", "/login", timeout=3)
        return status == 200
    except (urllib.error.URLError, OSError):
        return False


def main():
    if not os.path.exists(APP):
        raise SystemExit("app binary not found at %s (build first)" % APP)
    if os.path.exists(WORK):
        shutil.rmtree(WORK)
    os.makedirs(os.path.join(WORK, "workspace"))

    fake = FakeSaddle()
    fake.serve()
    app = start_app()

    try:
        auth = login("admin", "adminpass123")
        check(True, "login yields an auth slot")

        print("== feature 3: plan review hides duplicate body ==")
        state = slot_public(auth, "/projects", "projects")
        act(auth, state["create"], {
            "name": "demo",
            "git_url": "https://example.com/demo.git",
            "default_branch": "main",
        })
        conn = db()
        conn.execute("insert into tasks (project_id, title, status, priority,"
                     " is_design_task, plan, created_at)"
                     " values (1, 'Plan review task', 'working', 5, 1, '# The plan\n\n1. Do it',"
                     " strftime('%s','now'))")
        conn.commit()
        plan_task = conn.execute("select max(id) from tasks").fetchone()[0]

        saddles_state = slot_public(auth, "/saddles", "saddles")
        act(auth, saddles_state["register"], {"name": "fake-saddle",
                                              "host": "127.0.0.1:%d" % DAEMON_PORT})
        registered = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                      if s["name"] == "fake-saddle"]
        check(len(registered) == 1 and registered[0]["token"].startswith("saddle-"),
              "fake saddle registered with an API token")
        fake_id = int(registered[0]["id"])
        fake_token = registered[0]["token"]

        conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at, outcome)"
                     " values (?, ?, strftime('%s','now'), strftime('%s','now'), 'done')",
                     (plan_task, fake_id))
        conn.commit()
        session_id = conn.execute("select max(id) from sessions").fetchone()[0]
        conn.close()

        status, body = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
                                  {"type": "block", "session_id": session_id, "data": {
                                      "question": "Plan submitted for review",
                                      "type": "rfc",
                                      "body": "# The plan\n\n1. Do it"}})
        check(status == 200 and body.get("block_id"), "plan-review block raised")
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token,
                               {"type": "block", "session_id": session_id, "data": {
                                   "question": "Which database?",
                                   "type": "question",
                                   "options": ["SQLite", "Postgres"],
                                   "body": "# Storage choice"}})
        check(status == 200, "question block raised")

        detail = task_state(auth, plan_task)
        blocks = {b["block_type"]: b for b in detail["blocks"]}
        check(blocks["rfc"].get("is_plan_review") is True,
              "rfc block flagged as the plan review")
        check(blocks["rfc"].get("show_body") is False,
              "pending plan review hides the duplicate body preview")
        check(blocks["question"].get("show_body") is True,
              "question block keeps its body preview")

        act(auth, blocks["rfc"]["answer_action"], {"answer": "Plan approved. Proceed."})
        detail = task_state(auth, plan_task)
        rfc = [b for b in detail["blocks"] if b["block_type"] == "rfc"][0]
        check(rfc.get("show_body") is True,
              "answered plan review shows the body again (review section gone)")
    finally:
        app.terminate()
        try:
            app.wait(timeout=5)
        except subprocess.TimeoutExpired:
            app.kill()

    print("%d passed, %d failed" % (passes, failures))
    return 1 if failures else 0


if __name__ == "__main__":
    sys.exit(main())