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.

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 subprocess
import sqlite3
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 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("== 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())