clanker-saddle/test/integration_test.py

#!/usr/bin/env python3
"""End-to-end integration test: ClankerSaddle web app + saddle daemon.

Drives the web app the way statum.js does (entrypoint state loads plus
action POSTs carrying the auth slot and sealed private blobs) while a
real saddle daemon works on a real git checkout under WORKSPACE/demo,
powered by saddle/test/mock_llm.py with a deterministic three-task
script: an explore task (bash + comment + done), a dependency-gated
follow-up, and a blocking question answered through the web UI.

A fourth section exercises the part 2 surface directly against the
saddle ingest API: bearer-token auth, typed blocks with options, plan,
progress and review updates, quoted comments on the poll endpoint, and
the terminal SSE stream.

Prerequisites (see README.md "Integration testing"):
- the web app running with SPRY_SEED_ADMIN_USERNAME=admin,
  SPRY_SEED_ADMIN_PASSWORD=adminpass123, SADDLE_LLM_BASE_URL pointing at
  the mock (below), SADDLE_PROJECTS_ROOT=$WORKSPACE
- a saddle daemon reachable at SADDLE
- a git repo with at least two commits at $WORKSPACE/demo

Environment:
  WEB            web app base URL      (default http://localhost:8080)
  SADDLE         daemon control URL    (default http://localhost:7100)
  SADDLE_HOST    host:port to register (default localhost:7100)
  SADDLE_NAME    registration name     (default saddle-e2e-1)
  WORKSPACE      demo repo parent dir  (default /tmp/clanker-e2e/workspace)
  DB_PATH        web app SQLite file   (for audit assertions)
  MOCK           mock_llm.py path      (default <repo>/../saddle/test/mock_llm.py)
  MOCK_PORT      local mock port       (default 9999)
  MOCK_CMD       run instead of the local mock (e.g. a podman exec line
                 that starts the mock inside the saddle container); the
                 script skips spawning the mock when set

Prints PASS/FAIL lines Statum-style; exits non-zero on any failure.
"""
import json
import os
import sqlite3
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
WEB = os.environ.get("WEB", "http://localhost:8080")
SADDLE = os.environ.get("SADDLE", "http://localhost:7100")
SADDLE_HOST = os.environ.get("SADDLE_HOST", "localhost:7100")
SADDLE_NAME = os.environ.get("SADDLE_NAME", "saddle-e2e-1")
WORKSPACE = os.environ.get("WORKSPACE", "/tmp/clanker-e2e/workspace")
DB_PATH = os.environ.get("DB_PATH", os.path.join(WORKSPACE, "..", "web", "clanker-saddle.sqlite"))
MOCK = os.environ.get("MOCK", os.path.join(os.path.dirname(REPO), "saddle", "test", "mock_llm.py"))
MOCK_PORT = os.environ.get("MOCK_PORT", "9999")
MOCK_CMD = os.environ.get("MOCK_CMD", "")
DEMO = os.path.join(WORKSPACE, "demo")

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 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 saddle_status():
    try:
        status, body = http("GET", SADDLE + "/status", timeout=8)
        return json.loads(body) if status == 200 else None
    except (urllib.error.URLError, OSError):
        return None


def wait_for(predicate, timeout=60, what="condition", interval=1.0):
    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 wait_task_status(auth, task_id, status, timeout=90):
    return wait_for(lambda: task_state(auth, task_id).get("status") == status or False,
                    timeout=timeout, what="task %d -> %s" % (task_id, status))


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 start_mock():
    responses = [
        "# Inspecting the checkout\nls -la %s" % DEMO,
        "# Checking history\ngit -C %s log --oneline" % DEMO,
        'saddle comment "Inspected the checkout"\nsaddle done "Explored the demo repository"',
        "# Starting the follow-up\necho task-two-running",
        'saddle done "Follow-up audit complete"',
        '# I need to ask\nsaddle block "Which branch should I target?"',
    ]
    argv = [sys.executable, MOCK, "--port", MOCK_PORT]
    for response in responses:
        argv += ["-r", response]
    argv += ["--on-answer", 'saddle done "Branch confirmed: main"', "--delay", "2"]
    return subprocess.Popen(argv, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT)


def main():
    mock = None
    if not MOCK_CMD:
        mock = start_mock()
        time.sleep(0.5)
    else:
        subprocess.Popen(["bash", "-c", MOCK_CMD])
        time.sleep(1.0)

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

        state = slot_public(auth, "/projects", "projects")
        act(auth, state["create"], {
            "name": "demo",
            "git_url": "https://example.com/demo.git",
            "default_branch": "main",
            "summary": "Integration test repository",
        })
        check(any(p["name"] == "demo" for p in slot_public(auth, "/projects", "projects")["projects"]),
              "project demo created")

        state = slot_public(auth, "/tasks", "tasks")
        act(auth, state["create"], {
            "project_id": "1",
            "title": "Explore the demo repository",
            "description": "List the checkout and inspect its git history.",
            "priority": "3",
        })
        act(auth, state["create"], {
            "project_id": "1",
            "title": "Follow-up audit",
            "description": "Runs after the explore task completes.",
            "priority": "4",
        })
        act(auth, state["create"], {
            "project_id": "1",
            "title": "Ask a blocking question",
            "description": "Asks which branch to target, then finishes.",
            "priority": "5",
        })
        titles = [t["title"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]]
        check(titles == ["Explore the demo repository", "Follow-up audit", "Ask a blocking question"],
              "three tasks queued in priority order")

        detail = task_state(auth, 2)
        act(auth, detail["dependency"], {"depends_on_task_id": "1"})
        check(wait_for(lambda: any(int(d.get("id", 0)) == 1
                                   for d in task_state(auth, 2).get("dependencies", [])) or False,
                       timeout=5, what="dependency edge of task 2"),
              "task 2 depends on task 1")

        state = slot_public(auth, "/saddles", "saddles")
        act(auth, state["register"], {"name": SADDLE_NAME, "host": SADDLE_HOST})
        check(any(s["name"] == SADDLE_NAME and s["host"] == SADDLE_HOST
                  for s in slot_public(auth, "/saddles", "saddles")["saddles"]),
              "saddle registered at %s" % SADDLE_HOST)

        print("== explore task ==")
        left_queue = wait_for(
            lambda: (lambda s: s if s != "queued" else False)(task_state(auth, 1).get("status")) or False,
            timeout=15, what="task 1 to leave the queue")
        check(left_queue is not None, "dispatcher picks task 1 up within 15s (status %s)" % left_queue)
        check(wait_for(lambda: (saddle_status() or {}).get("state") in ("working", "blocked", "done") or False,
                       timeout=15, what="saddle active") is not None,
              "saddle daemon receives the task")
        check(wait_task_status(auth, 1, "working", timeout=30) is not None,
              "task 1 flips to working from daemon status")
        check(wait_task_status(auth, 1, "done", timeout=90) is not None, "task 1 completes")

        detail = task_state(auth, 1)
        check("Explored the demo repository" in (detail.get("summary") or ""),
              "task 1 carries the agent's summary")
        check(any(c["author"] == "bot" and "Inspected the checkout" in c["body"]
                  for c in detail.get("comments", [])),
              "agent comment projected into task_comments")
        check(detail.get("sessions") and detail["sessions"][0].get("outcome") == "done",
              "session closed with outcome done")
        check(directives("/tasks/1/terminal", auth) is not None, "terminal page state loads")

        print("== dependency gating ==")
        check(wait_task_status(auth, 2, "done", timeout=90) is not None,
              "task 2 (dependent) dispatches and completes after task 1")

        print("== block and answer ==")
        check(wait_task_status(auth, 3, "blocked", timeout=90) is not None, "task 3 reaches blocked")
        detail = task_state(auth, 3)
        open_blocks = [b for b in detail.get("blocks", []) if not b.get("answered")]
        check(len(open_blocks) == 1 and "Which branch" in open_blocks[0]["question"],
              "block question surfaced in the web UI")
        act(auth, open_blocks[0]["answer_action"], {"answer": "main"})
        check(any(b.get("answer") == "main" for b in task_state(auth, 3).get("blocks", [])),
              "human answer recorded via the web action")
        check(wait_task_status(auth, 3, "done", timeout=90) is not None,
              "answer delivered to daemon; task 3 completes")
        detail = task_state(auth, 3)
        check("Branch confirmed: main" in (detail.get("summary") or ""),
              "task 3 summary reflects the post-answer completion")

        print("== audit trail ==")
        db = sqlite3.connect(DB_PATH)
        rows = db.execute(
            "select command_text, output_text, exit_code from session_commands order by id").fetchall()
        commands = {row[0]: row for row in rows}
        check(any("ls -la" in c and "hello.py" in (r[1] or "") for c, r in commands.items()),
              "ls output captured in session_commands")
        check(any("git" in c and "Initial commit" in (r[1] or "") and r[2] == 0
                  for c, r in commands.items()),
              "git log output captured with exit code 0")
        check(len([c for c in commands if "echo" in c]) >= 1, "echo output captured")
        per_session = db.execute(
            "select session_id, count(*) from session_commands group by session_id order by session_id"
        ).fetchall()
        check(all(count > 0 for _, count in per_session) and len(per_session) == 3,
              "each session carries its own commands (%s)" % per_session)
        db.close()

        print("== part 2 API surface ==")
        db = sqlite3.connect(DB_PATH)

        saddles_state = slot_public(auth, "/saddles", "saddles")
        act(auth, saddles_state["register"], {"name": "saddle-e2e-api", "host": "localhost:7199"})
        api_rows = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
                    if s["name"] == "saddle-e2e-api"]
        check(len(api_rows) == 1 and api_rows[0]["token"].startswith("saddle-"),
              "registration returns an API token")
        api_saddle = int(api_rows[0]["id"])
        api_token = api_rows[0]["token"]

        tasks_state = slot_public(auth, "/tasks", "tasks")
        act(auth, tasks_state["create"], {
            "project_id": "1",
            "title": "Part 2 API surface",
            "description": "Typed blocks, plan, progress, review, quoted comments, SSE",
            "plan": "# Initial plan",
            "priority": "6",
            "task_branch": "task/9-api-surface",
        })
        api_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
                        if t["title"] == "Part 2 API surface"][0])
        detail = task_state(auth, api_task)
        check(detail["branch_from"] == "main",
              "branch_from prefilled from the project default branch")
        check(detail["task_branch"] == "task/9-api-surface", "task_branch stored")

        # Dispatch bookkeeping without a daemon: the ingest API is under
        # test, so the session and saddle/task state are staged directly.
        db.execute("insert into sessions (task_id, saddle_id, started_at)"
                   " values (?, ?, strftime('%s','now'))", (api_task, api_saddle))
        db.execute("update saddles set current_task_id = ?, status = 'idle' where id = ?",
                   (api_task, api_saddle))
        db.execute("update tasks set status = 'working' where id = ?", (api_task,))
        db.commit()
        api_session = db.execute("select max(id) from sessions").fetchone()[0]

        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, None,
                               {"type": "comment", "session_id": api_session, "data": {"body": "x"}})
        check(status == 401, "update without a token is rejected")
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, "saddle-not-a-token",
                               {"type": "comment", "session_id": api_session, "data": {"body": "x"}})
        check(status == 401, "update with an invalid token is rejected")
        status, _ = saddle_api("GET", "/api/saddle/%d/poll" % api_saddle, None)
        check(status == 401, "poll without a token is rejected")

        status, body = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                                  {"type": "comment", "session_id": api_session,
                                   "data": {"body": "Surveyed the module"}})
        check(status == 200 and body.get("ok"), "comment update accepted with token")
        status, body = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                                  {"type": "block", "session_id": api_session, "data": {
                                      "question": "Which database?",
                                      "type": "question",
                                      "options": ["SQLite", "Postgres"],
                                      "body": "# Pick one"}})
        check(status == 200 and body.get("block_id"), "typed block returns block_id")
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                               {"type": "plan", "session_id": api_session,
                                "data": {"plan": "# Plan\n- migrate schemas"}})
        check(status == 200, "plan update accepted")
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                               {"type": "progress", "session_id": api_session,
                                "data": {"summary": "Migrated the schemas"}})
        check(status == 200, "progress update accepted")
        status, _ = saddle_api("POST", "/api/saddle/%d/update" % api_saddle, api_token,
                               {"type": "review", "session_id": api_session,
                                "data": {"summary": "Migration **ready** for review"}})
        check(status == 200, "review update accepted")

        detail = task_state(auth, api_task)
        check(detail["status"] == "review_requested", "review flips the task status")
        check([b["options"] for b in detail["blocks"]
               if b["block_type"] == "question"] == [["SQLite", "Postgres"]],
              "question block carries its options")
        check("- migrate schemas" in detail["plan"], "plan visible on the task page")
        check(detail["progress"][0]["summary"] == "Migrated the schemas",
              "progress history newest-first")
        check(detail["review_pending"] and "ready" in detail["review_summary"],
              "review banner data on the task page")
        check(detail["review_diff_url"].endswith(
            "demo.git/compare/main...task/9-api-surface"), "review diff link constructed")

        dash = slot_public(auth, "/", "dashboard")
        check(any(int(r["task_id"]) == api_task for r in dash["reviews"]),
              "dashboard lists the pending review")
        check(any(b["question"] == "Which database?" for b in dash["open_blocks"]),
              "dashboard lists blocks needing answers")

        question = [b for b in detail["blocks"] if b["block_type"] == "question"][0]
        act(auth, question["answer_action"], {"option": "SQLite"})
        check([b["answer"] for b in task_state(auth, api_task)["blocks"]
              if b["block_type"] == "question"] == ["Human selected: SQLite"],
              "option answer formatted for the agent")
        act(auth, detail["comment"],
            {"body": "Check the migration output", "quote": "ERROR: table demos exists"})
        status, body = saddle_api("GET", "/api/saddle/%d/poll" % api_saddle, api_token)
        check(any(a["answer"] == "Human selected: SQLite" for a in body.get("answers", [])),
              "poll hands over the block answer")
        check(any(c["body"] == "Check the migration output" and "ERROR" in c["quote"]
                  for c in body.get("comments", [])),
              "poll hands over the quoted user comment")
        status, body = saddle_api("GET", "/api/saddle/%d/poll" % api_saddle, api_token)
        check(body.get("answers") == [] and body.get("comments") == [],
              "poll delivers answers and comments exactly once")

        entry_types = {row[0] for row in db.execute(
            "select type from session_commands where session_id = ?", (api_session,))}
        check({"comment", "block", "plan", "progress", "review", "user-comment"} <= entry_types,
              "conversation events logged as typed terminal entries")

        request = urllib.request.Request(WEB + "/terminal-stream/%d" % api_session)
        with urllib.request.urlopen(request, timeout=10) as stream:
            check(stream.headers.get("Content-Type", "").startswith("text/event-stream"),
                  "terminal stream is text/event-stream")
            lines = []
            deadline = time.monotonic() + 5
            while time.monotonic() < deadline and len(lines) < 40:
                line = stream.readline().decode().strip()
                if line:
                    lines.append(line)
                    deadline = min(deadline, time.monotonic() + 2)
            entries = [json.loads(line[6:]) for line in lines
                       if line.startswith("data: ") and line != "data: ping"]
            check(len(entries) >= 7, "terminal stream replays the backlog")
            db.execute("insert into session_commands (session_id, command_text, output_text,"
                       " type, started_at) values (?, 'sse-live-check', 'ok', 'command',"
                       " strftime('%s','now'))", (api_session,))
            db.commit()
            got_live = False
            deadline = time.monotonic() + 6
            while time.monotonic() < deadline and not got_live:
                line = stream.readline().decode().strip()
                if line.startswith("data: ") and "sse-live-check" in line:
                    got_live = True
            check(got_live, "terminal stream pushes new entries within seconds")

        act(auth, detail["cancel"], {"task_id": str(api_task)})
        db.close()

        print("== quiesce ==")
        check(wait_for(
            lambda: all(int(s["current_task_id"] or 0) == 0 and s["status"] == "idle"
                        for s in slot_public(auth, "/saddles", "saddles")["saddles"]) or False,
            timeout=30, what="saddle idle with no current task") is not None,
            "saddle freed and idle once every task finished")
    finally:
        if mock is not None:
            mock.terminate()

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


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