#!/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) # One retry on a dropped connection: the freshly started app can # close a very early keep-alive connection once while warming up. for attempt in range(2): 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() except (urllib.error.URLError, OSError): if attempt > 0: raise time.sleep(0.5) 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 wait_task_done(auth, task_id, timeout=90): return wait_for(lambda: task_state(auth, task_id).get("status") == "done" or False, timeout=timeout, what="task %d -> done" % task_id) def task_id_of(title): conn = db() try: return conn.execute("select id from tasks where title = ?", (title,)).fetchone()[0] finally: conn.close() 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 self.fail_next = False # next /start drives the task to failed self.start_counts = {} # task id -> /start count self.resumed_pending = False # crash-resume flag on the next /status self.pending_output = [] # entries drained by the next /output poll self.output_seq = 100 def queue_output(self, entry_type, command, output): """Queues a typed entry for the dispatcher's next /output poll.""" with self.lock: self.pending_output.append({ "id": self.output_seq, "command": command, "output": output, "exit_code": 1, "type": entry_type, "timestamp": "2026-09-06T00:00:00Z", }) self.output_seq += 1 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: # `since` is the inclusive resume point (wire contract: a # response ending at seq 12 carries next=13, and polling # with since=13 must still deliver the chunk numbered 13). 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 protocol_version = "HTTP/1.1" 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) task_id = (payload.get("task") or {}).get("id") count = self.daemon.start_counts.get(task_id, 0) + 1 self.daemon.start_counts[task_id] = count if count > 1: # Crash-resume contract: re-pushing a task whose # workspace still exists restores it; the first # /status after the restore reports "resumed": true. self.daemon.resumed_pending = True 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 = "failing" if self.daemon.fail_next else "working" self.daemon.fail_next = False 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": response = {"state": self.daemon.state, "summary": self.daemon.summary} if self.daemon.resumed_pending: response["resumed"] = True self.daemon.resumed_pending = False if self.daemon.state in ("working", "failing"): self.daemon.status_polls_after_start += 1 if self.daemon.state == "failing" and self.daemon.status_polls_after_start >= 1: self.daemon.state = "failed" self.daemon.summary = "bash -n rejected the script (simulated)" elif self.daemon.state == "working" and self.daemon.status_polls_after_start >= 2: self.daemon.state = "done" self.daemon.summary = "fake daemon finished" self._send(200, response) elif self.path.startswith("/output"): with self.daemon.lock: entries = self.daemon.pending_output self.daemon.pending_output = [] self._send(200, entries) elif self.path.startswith("/log"): query = urllib.parse.urlparse(self.path).query since = int(urllib.parse.parse_qs(query).get("since", ["0"])[0]) # Simulate the live PTY: each poll produces one more chunk # (before slicing, so the poller actually sees it). self.daemon.append_chunk("stream-line-%d\r\n" % self.daemon.next_seq) chunks, nxt = self.daemon.log_since(since) self._send(200, {"chunks": chunks, "next": nxt}) else: self._send(404, {"error": "no such route"}) self.close_connection = True 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)") print("== feature 5: rules ==") rules_state = slot_public(auth, "/rules", "rules") for name, body in [ ("Use Invercargill", "Never use libgee; use Invercargill enumerables."), ("Write Vala docs", "Every public class carries a doc comment."), ("Prefer meson", "Builds go through meson; no makefiles."), ("Co-dev rule", "Keep the wire contract exactly."), ("Reference rule", "Must never ride along."), ]: act(auth, rules_state["create"], {"name": name, "body": body}) listed = {r["name"]: r for r in slot_public(auth, "/rules", "rules")["rules"]} check(len(listed) == 5, "five rules created and listed on /rules") def rule_id(name): rows = [r for r in slot_public(auth, "/rules", "rules")["rules"] if r["name"] == name] return int(rows[0]["id"]) state = slot_public(auth, "/projects", "projects") act(auth, state["create"], {"name": "co", "git_url": "https://example.com/co.git"}) act(auth, state["create"], {"name": "ref", "git_url": "https://example.com/ref.git"}) co_id = int([p["id"] for p in slot_public(auth, "/projects", "projects")["projects"] if p["name"] == "co"][0]) ref_id = int([p["id"] for p in slot_public(auth, "/projects", "projects")["projects"] if p["name"] == "ref"][0]) project = slot_public(auth, "/projects/1", "project") act(auth, project["add_link"], {"linked_project_id": str(co_id), "relationship": "co-developed"}) act(auth, project["add_link"], {"linked_project_id": str(ref_id), "relationship": "reference"}) act(auth, project["attach_rule"], {"rule_id": str(rule_id("Use Invercargill"))}) act(auth, project["attach_rule"], {"rule_id": str(rule_id("Prefer meson"))}) project = slot_public(auth, "/projects/1", "project") check(sorted(r["name"] for r in project["rules"]) == ["Prefer meson", "Use Invercargill"], "project page lists its attached rules") co_project = slot_public(auth, "/projects/%d" % co_id, "project") act(auth, co_project["attach_rule"], {"rule_id": str(rule_id("Co-dev rule"))}) ref_project = slot_public(auth, "/projects/%d" % ref_id, "project") act(auth, ref_project["attach_rule"], {"rule_id": str(rule_id("Reference rule"))}) tasks_state = slot_public(auth, "/tasks", "tasks") act(auth, tasks_state["create"], { "project_id": "1", "title": "Rules dispatch", "description": "Effective rule set rides the /start payload.", "priority": "2", }) rule_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"] if t["title"] == "Rules dispatch"][0]) detail = task_state(auth, rule_task) act(auth, detail["attach_rule"], {"rule_id": str(rule_id("Write Vala docs"))}) act(auth, detail["attach_rule"], {"rule_id": str(rule_id("Use Invercargill"))}) detail = task_state(auth, rule_task) check(sorted(r["name"] for r in detail["rules"]) == ["Use Invercargill", "Write Vala docs"], "task page lists its attached rules") check(wait_for(lambda: len(fake.start_payloads) > 0 or False, timeout=30, what="/start push") is not None, "dispatcher pushes the task to the fake saddle") payload = fake.start_payloads[-1] rule_names = [r["name"] for r in payload.get("rules", [])] check(rule_names == ["Write Vala docs", "Use Invercargill", "Prefer meson", "Co-dev rule"], "effective rules in /start: task, primary, co-developed order (%s)" % rule_names) check(all(r.get("body") for r in payload.get("rules", [])), "rules carry their bodies") check("Reference rule" not in rule_names, "reference project rules excluded") detail = task_state(auth, rule_task) act(auth, detail["rules"][0]["detach"], {}) check([r["name"] for r in task_state(auth, rule_task)["rules"]] == ["Use Invercargill"], "detached task rule disappears") rules_rows = slot_public(auth, "/rules", "rules")["rules"] docs_row = [r for r in rules_rows if r["name"] == "Write Vala docs"][0] act(auth, docs_row["update"], {"name": "Write Vala docs", "body": "Doc comments or bust."}) check(any(r["body"] == "Doc comments or bust." for r in slot_public(auth, "/rules", "rules")["rules"]), "rule edit saves the body") check(wait_task_done(auth, rule_task, timeout=60) is not None, "dispatched task completes against the fake daemon") print("== feature 6: project notes + callback ==") conn = db() callback_rows = conn.execute( "select token from api_tokens where saddle_id = 0").fetchall() conn.close() check(len(callback_rows) == 1 and callback_rows[0][0].startswith("saddle-callback-"), "dedicated saddle-callback token minted at startup") callback_token = callback_rows[0][0] status, _ = saddle_api("POST", "/api/saddle/note", None, {"project": "demo", "body": "x"}) check(status == 401, "note endpoint rejects a missing token") status, _ = saddle_api("POST", "/api/saddle/note", "saddle-callback-not-real", {"project": "demo", "body": "x"}) check(status == 401, "note endpoint rejects a wrong token") status, body = saddle_api("POST", "/api/saddle/note", callback_token, {"task_id": rule_task, "project": "demo", "body": "style.css tokens live in :root"}) check(status == 200 and body.get("ok") is True, "note accepted with the callback token") status, _ = saddle_api("POST", "/api/saddle/note", callback_token, {"project": "no-such-project", "body": "x"}) check(status == 400, "unknown project rejected") status, _ = saddle_api("POST", "/api/saddle/note", callback_token, {"project": "demo", "body": " "}) check(status == 400, "empty note body rejected") project = slot_public(auth, "/projects/1", "project") check(any("style.css tokens" in n["body"] and int(n["source_task_id"]) == rule_task for n in project["notes"]), "project page lists the note with its source task") act(auth, [n for n in project["notes"] if "style.css tokens" in n["body"]][0]["delete"], {}) check(all("style.css tokens" not in n["body"] for n in slot_public(auth, "/projects/1", "project")["notes"]), "note deleted from the project page") # /start payload: notes of primary + co-developed (not reference), # oldest first; callback carries the configured base URL + token. saddle_api("POST", "/api/saddle/note", callback_token, {"project": "demo", "body": "note-one"}) saddle_api("POST", "/api/saddle/note", callback_token, {"project": "demo", "body": "note-two"}) saddle_api("POST", "/api/saddle/note", callback_token, {"project": "co", "body": "co-note"}) conn = db() conn.execute("insert into project_notes (project_id, body, created_at)" " values (?, 'ref-note-must-not-ship', strftime('%s','now'))", (ref_id,)) conn.commit() conn.close() act(auth, slot_public(auth, "/tasks", "tasks")["create"], { "project_id": "1", "title": "Notes dispatch", "description": "Project notes ride the /start payload.", "priority": "2", }) check(wait_for(lambda: len(fake.start_payloads) >= 2 or False, timeout=30, what="second /start push") is not None, "dispatcher pushes the notes task") payload = fake.start_payloads[-1] notes = payload.get("project_notes", []) check([n["body"] for n in notes] == ["note-one", "note-two", "co-note"], "project_notes: primary + co-developed, oldest first (%s)" % [n["body"] for n in notes]) check(all(n.get("project") for n in notes) and notes[2]["project"] == "co", "project_notes carry the project name") check("ref-note-must-not-ship" not in [n["body"] for n in notes], "reference project notes excluded") callback = payload.get("callback", {}) check(callback.get("base_url") == WEB and callback.get("token") == callback_token, "callback carries the configured base URL and the minted token") # entry cap: 30 of 33 candidates conn = db() for i in range(33): conn.execute("insert into project_notes (project_id, body, created_at)" " values (?, ?, strftime('%s','now'))", (1, "bulk-%02d" % i)) conn.commit() conn.close() act(auth, slot_public(auth, "/tasks", "tasks")["create"], { "project_id": "1", "title": "Entry cap dispatch", "priority": "2", }) check(wait_for(lambda: len(fake.start_payloads) >= 3 or False, timeout=30, what="third /start push") is not None, "dispatcher pushes the entry-cap task") notes = fake.start_payloads[-1].get("project_notes", []) check(len(notes) == 30 and notes[0]["body"] == "note-one" and notes[1]["body"] == "note-two", "project_notes capped at 30 entries, oldest kept (%d)" % len(notes)) # char cap: the giant note is truncated so the total stays <= 20000 conn = db() conn.execute("delete from project_notes where body like 'bulk-%' or body = 'co-note'") conn.execute("insert into project_notes (project_id, body, created_at)" " values (?, ?, strftime('%s','now'))", (1, "giant!" + "x" * 25000,)) conn.commit() conn.close() act(auth, slot_public(auth, "/tasks", "tasks")["create"], { "project_id": "1", "title": "Char cap dispatch", "priority": "2", }) check(wait_for(lambda: len(fake.start_payloads) >= 4 or False, timeout=30, what="fourth /start push") is not None, "dispatcher pushes the char-cap task") notes = fake.start_payloads[-1].get("project_notes", []) total = sum(len(n["body"]) for n in notes) check(len(notes) == 3 and total == 20000 and notes[2]["body"].startswith("giant!"), "project_notes char cap: head kept, giant truncated (entries=%d, chars=%d)" % (len(notes), total)) print("== raw log polling + terminal raw view ==") # Wait for the last dispatch to finish so its final polls ran. check(wait_task_done(auth, task_id_of("Char cap dispatch"), timeout=60) is not None, "char-cap task completes") conn = db() raw_session = conn.execute( "select s.id from sessions s join tasks t on t.id = s.task_id" " where t.title = 'Char cap dispatch' order by s.id desc limit 1").fetchone()[0] rows = conn.execute("select seq, chunk from session_raw_log where session_id = ?" " order by seq", (raw_session,)).fetchall() conn.close() check(len(rows) >= 3, "session_raw_log captured the daemon's chunks (%d)" % len(rows)) seqs = [r[0] for r in rows] check(seqs == sorted(seqs) and len(set(seqs)) == len(seqs), "raw chunks stored once, seq strictly increasing") text = "".join(r[1] for r in rows) check(fake.PROMPT in text and "echo hello" in text and "hello" in text, "raw log keeps the prompt and echoed output byte-exact") check("stream-line-" in text, "incremental /log polls append later chunks") terminal = slot_public(auth, "/tasks/%d/terminal" % task_id_of("Char cap dispatch"), "terminal") check(terminal.get("has_raw_log") is True and int(terminal["session_id"]) == raw_session, "terminal view selects raw rendering for the raw session") # Old sessions (typed entries, no raw rows) keep the fallback view. conn = db() conn.execute("insert into session_commands (session_id, command_text, output_text," " type, started_at) values (?, 'ls', 'old-output', 'command'," " strftime('%s','now'))", (session_id,)) conn.commit() conn.close() legacy = slot_public(auth, "/tasks/%d/terminal" % plan_task, "terminal") check(legacy.get("has_raw_log") is False, "legacy session falls back to the entry-type terminal view") request = urllib.request.Request(WEB + "/terminal-stream/%d" % raw_session) with urllib.request.urlopen(request, timeout=15) as stream: raw_events = [] deadline = time.monotonic() + 8 while time.monotonic() < deadline and len(raw_events) < 3: line = stream.readline().decode().strip() if not line.startswith("data: ") or line == "data: ping": continue event = json.loads(line[6:]) if event.get("type") == "raw": raw_events.append(event) check(len(raw_events) >= 3 and raw_events[0]["text"] == fake.PROMPT and raw_events[0]["seq"] < raw_events[1]["seq"], "terminal stream serves raw chunks in seq order (prompt first)") print("== capabilities: tag-based routing ==") saddles_state = slot_public(auth, "/saddles", "saddles") act(auth, saddles_state["create_capability"], {"name": "gpu"}) act(auth, saddles_state["create_capability"], {"name": "bigmem"}) saddles_state = slot_public(auth, "/saddles", "saddles") caps = {c["name"]: c for c in saddles_state["capabilities"]} check(len(caps) == 2 and all(int(c["saddle_count"]) == 0 and int(c["task_count"]) == 0 for c in caps.values()), "capabilities created and listed with zero usage") def saddle_card(name): rows = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == name] return rows[0] fake_card = saddle_card("fake-saddle") check(fake_card["capabilities"] == [], "saddle card starts with no capability chips") # A task requiring 'gpu' waits while the only idle saddle lacks it; # a later task without requirements still dispatches (skip-and-continue). tasks_state = slot_public(auth, "/tasks", "tasks") act(auth, tasks_state["create"], { "project_id": "1", "title": "GPU task", "priority": "2", }) gpu_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"] if t["title"] == "GPU task"][0]) detail = task_state(auth, gpu_task) act(auth, detail["attach_capability"], {"capability_id": str(int(caps["gpu"]["id"]))}) detail = task_state(auth, gpu_task) check([c["name"] for c in detail["capabilities"]] == ["gpu"], "task page lists the required capability") # The only idle saddle lacks the tag, so the queue flags the wait # (read before any other task claims the idle saddle). queue = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"] if int(t["id"]) == gpu_task][0] check(queue["status"] == "queued" and queue["capabilities"] == ["gpu"] and queue.get("waiting_for_saddle") is True, "queue row shows the waiting task with its tag and wait flag") act(auth, tasks_state["create"], { "project_id": "1", "title": "Any-saddle task", "priority": "3", }) any_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"] if t["title"] == "Any-saddle task"][0]) check(wait_for(lambda: task_state(auth, any_task).get("status") != "queued" or False, timeout=30, what="uncapable task dispatches") is not None, "task without requirements dispatches past the waiting task") with open(os.path.join(WORK, "app.log")) as handle: app_log = handle.read() check("waits for a capable saddle" in app_log and "requires: gpu" in app_log, "dispatcher logs the capability wait") # Attaching the tag to the saddle releases the task. fake_card = saddle_card("fake-saddle") act(auth, fake_card["attach_capability"], {"capability_id": str(int(caps["gpu"]["id"]))}) fake_card = saddle_card("fake-saddle") check([c["name"] for c in fake_card["capabilities"]] == ["gpu"], "saddle card shows the attached chip") check(wait_task_done(auth, gpu_task, timeout=60) is not None, "task dispatches once the saddle carries the required tag") queue = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"] if int(t["id"]) == gpu_task][0] check(queue.get("waiting_for_saddle") is not True, "wait flag clears once dispatched") # Multi-tag: gpu + bigmem required, only gpu carried → waits; adding # bigmem completes the superset. act(auth, tasks_state["create"], { "project_id": "1", "title": "Big task", "priority": "2", }) big_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"] if t["title"] == "Big task"][0]) detail = task_state(auth, big_task) act(auth, detail["attach_capability"], {"capability_id": str(int(caps["gpu"]["id"]))}) act(auth, detail["attach_capability"], {"capability_id": str(int(caps["bigmem"]["id"]))}) check(wait_for(lambda: ([t for t in slot_public(auth, "/tasks", "tasks")["tasks"] if int(t["id"]) == big_task][0].get("waiting_for_saddle") is True) or False, timeout=30, what="big task waits") is not None, "task waits when the saddle's tags are not a superset") fake_card = saddle_card("fake-saddle") act(auth, fake_card["attach_capability"], {"capability_id": str(int(caps["bigmem"]["id"]))}) check(wait_task_done(auth, big_task, timeout=60) is not None, "task dispatches once the superset is complete") # Detach a requirement; delete a capability everywhere. detail = task_state(auth, big_task) act(auth, [c for c in detail["capabilities"] if c["name"] == "bigmem"][0]["detach"], {}) check([c["name"] for c in task_state(auth, big_task)["capabilities"]] == ["gpu"], "detached requirement disappears from the task") bigmem_row = [c for c in slot_public(auth, "/saddles", "saddles")["capabilities"] if c["name"] == "bigmem"][0] act(auth, bigmem_row["delete"], {}) saddles_state = slot_public(auth, "/saddles", "saddles") check(all(c["name"] != "bigmem" for c in saddles_state["capabilities"]), "deleted capability gone from the catalogue") check([c["name"] for c in task_state(auth, big_task)["capabilities"]] == ["gpu"], "deleting a capability drops its task requirements") check(all(c["name"] != "bigmem" for c in saddle_card("fake-saddle")["capabilities"]), "deleting a capability drops it from saddles") print("== retry + resume ==") # A daemon-side failure fails the task with its explanation. fake.fail_next = True act(auth, slot_public(auth, "/tasks", "tasks")["create"], { "project_id": "1", "title": "Flaky task", "priority": "2", }) flaky = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"] if t["title"] == "Flaky task"][0]) check(wait_for(lambda: task_state(auth, flaky).get("status") == "failed" or False, timeout=60, what="flaky task fails") is not None, "daemon failure fails the task") detail = task_state(auth, flaky) check("bash -n rejected" in (detail.get("summary") or ""), "failure summary shows the daemon's explanation") # Retry: summary moves into a comment and clears, task requeues, # history stays, and the re-dispatch restores the workspace. comments_before = len(detail.get("comments", [])) progress_before = len(detail.get("progress", [])) act(auth, detail["retry"], {}) detail = task_state(auth, flaky) check(detail["status"] == "queued" and (detail.get("summary") or "") == "", "retry requeues the task and clears the stale summary") check(any("bash -n rejected" in c["body"] for c in detail["comments"]), "failure explanation kept as a bot comment") check(any(c["body"] == "Task retried by user" for c in detail["comments"]), "retry lands its audit comment") check(len(detail["comments"]) >= comments_before + 1, "retry keeps the comment history") check(wait_task_done(auth, flaky, timeout=60) is not None, "retried task dispatches again and completes") detail = task_state(auth, flaky) check(any("restored the session context" in p["summary"] for p in detail.get("progress", [])), "daemon resume reported on the first status after the restore") check(len(detail["progress"]) > progress_before, "resume note appended to the progress timeline") # Manual resume: a task stuck working on an offline saddle. act(auth, slot_public(auth, "/tasks", "tasks")["create"], { "project_id": "1", "title": "Stuck task", "priority": "2", }) stuck = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"] if t["title"] == "Stuck task"][0]) check(wait_for(lambda: task_state(auth, stuck).get("status") == "working" or False, timeout=60, what="stuck task starts working") is not None, "stuck task reaches working") conn = db() conn.execute("update saddles set status = 'offline' where name = 'fake-saddle'") conn.commit() conn.close() detail = task_state(auth, stuck) check(detail.get("resume_available") is True, "task page offers resume while its saddle is offline") act(auth, detail["resume"], {}) detail = task_state(auth, stuck) check(detail["status"] == "queued", "manual resume requeues the stuck task") check(any(c["body"] == "Saddle became unreachable; task requeued" for c in detail["comments"]), "manual resume lands the interruption comment") check(all(s["status"] == "offline" for s in slot_public(auth, "/saddles", "saddles")["saddles"]), "saddle stays offline after the manual requeue") # Bringing the saddle back lets the requeued task resume and finish. fake.state = "idle" fake_card = saddle_card("fake-saddle") act(auth, fake_card["refresh"], {}) check(any(s["status"] == "idle" for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == "fake-saddle"), "refreshed saddle returns idle") check(wait_task_done(auth, stuck, timeout=60) is not None, "requeued task resumes on the refreshed saddle and completes") print("== rejected entries (non-fatal) ==") # Ingest API: a rejected entry stores on the session and streams # through SSE; the task is untouched. status, _ = saddle_api("POST", "/api/saddle/%d/update" % fake_id, fake_token, {"type": "output", "session_id": session_id, "data": { "command": "bash -n script.sh", "output": "syntax error near line 3 (recoverable)", "exit_code": 1, "type": "rejected"}}) check(status == 200, "rejected entry accepted by the ingest API") request = urllib.request.Request(WEB + "/terminal-stream/%d" % session_id) with urllib.request.urlopen(request, timeout=15) as stream: rejected_event = None deadline = time.monotonic() + 8 while time.monotonic() < deadline and rejected_event is None: line = stream.readline().decode().strip() if not line.startswith("data: ") or line == "data: ping": continue event = json.loads(line[6:]) if event.get("type") == "rejected": rejected_event = event check(rejected_event is not None and "syntax error near line 3" in (rejected_event.get("output") or ""), "terminal stream carries rejected entries with their reason") # The page template renders them with the distinct non-fatal style. status, page = http("GET", "/tasks/%d/terminal" % stuck, timeout=15) check(status == 200 and "term-rejected" in page, "terminal page renders rejected entries in their own style") # Dispatch path: a rejected entry from /output flows into the audit # trail without failing (or finishing) the task. fake.queue_output("rejected", "echo 'too long' | saddle", "max_tokens exceeded (recoverable)") act(auth, slot_public(auth, "/tasks", "tasks")["create"], { "project_id": "1", "title": "Rejected ride", "priority": "2", }) rejected_task = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"] if t["title"] == "Rejected ride"][0]) check(wait_task_done(auth, rejected_task, timeout=60) is not None, "task completes normally past rejected entries") conn = db() stored = conn.execute( "select count(*) from session_commands sc join sessions s on s.id = sc.session_id" " where s.task_id = ? and sc.type = 'rejected'", (rejected_task,)).fetchone()[0] conn.close() check(stored >= 1, "rejected entries stored in the audit trail (%d rows)" % stored) check(task_state(auth, rejected_task)["status"] == "done" and "fake daemon finished" in (task_state(auth, rejected_task).get("summary") or ""), "rejected entries never fail the task") 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())