#!/usr/bin/env python3 """The web-stack test harness (plan 16 R5, arch 18): everything the scenario bundles share, plus the runner (sequential by default, `--parallel [N]` for capped parallel shards). Layout: test/harness.py (this file — the FakeSaddle daemon, the http/act/login/slot_public wire helpers, wait_for/quiet_window/db/check, the per-shard fixtures via start_bundle/Ctx, and run_bundle + run_all_parallel); test/bundles/.py (one module per bundle, its run(ctx) driving that shard's scenarios); test/web_stack_test.py (the thin entry shim meson invokes; it also stays directly runnable for the per-bundle shard processes the parallel runner spawns). 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 - in-process fake saddle daemons speaking the wire contract: POST /register (plan 10: receives {saddle_token, web_base_url}, answers {ok, ssh_public_key, version}), POST /wake (records the wake; the daemon then asks POST /api/saddle/next-task), POST /task-update (cancel/comment/subtask-status), GET /status (scripted reflection — exceptional use only), and the phase-2 push channels home: POST /api/saddle/chunks (raw PTY batches) and POST /api/saddle/results (round entries + state transitions). The retired /start, /output and /log routes 404 like a post-phase-2 daemon. 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 assignments 6. project notes, POST /api/saddle/note, `project_notes` + `callback` in assignments plus the raw-log side: pushed chunks -> session_raw_log + terminal raw view Plan 10 phase 1 (work acquisition flip): - per-saddle tokens: registration mints + /register-delivers them, the row is the single credential store (the api_tokens fallback died with arch 24), removal revokes - token auth on every /api/saddle/* endpoint - next-task allocation: concurrent asks race exactly one winner, affinity pinning for parked tasks, resume_notice on suspended resumes - dispatcher: wake -> ask -> allocate roundtrip, the lost-wake guard (no ask after a wake -> offline, re-wake the next capable), and the pre-flip /answer + /start resume paths kept as the rollback route Plan 10 phase 2 (telemetry flip): - pushed chunks: idempotent on (session_id, seq), ordered, landing in session_raw_log so the SSE terminal keeps working - pushed results: entries store like the /output path did (redelivery idempotent on the daemon entry ids), blocked raises the block row + parks + bumps the pulse (sealed refresh shows it), done/failed complete/fail the task, suspended parks for the affinity resume - answer resumes, both daemon shapes: a SLEEPING affine saddle (its post-park ask returned 204) is woken, its ask allocating the resume with the verdict riding the assignment; a daemon parked MID-ROUND (its round holds the block's answer wait open, so it can never ask) gets the verdict delivered into that wait via POST /answer — same session, no wake. A daemon dying with a parked-with-answered task never requeues it: the verdict and session survive for the resume - staleness: a silent runner gets ONE exceptional GET /status probe — alive keeps it running, dead requeues the task with an agent comment and marks the saddle offline; a recovered daemon's ask flips offline rows back to idle - the deleted layers are gone: /output, /log and /start 404 on the daemon side, and nothing on the web side calls them Plan 10 phase 3 (capability awareness + UI): - assignment capability context: {saddle_tags, fleet_tags} on every payload, GET /api/saddle/capabilities (per-saddle token) feeding the daemon's `saddle capabilities` CLI - subtask callback capabilities: explicit-only attachment validated against the operator tag set, unknown tags rejected with the valid list in the error - task page: status block pinned to the top, blocking sections newest-first (sealed GET state) - home dashboard: pinned OPEN blocks as newest-first link-through cards (no inline answer forms) above the cross-task activity timeline (last ~50 events newest-first) - lost-wake-guard regression: a saddle that asks SLOWLY after its wake (busy winding down) still claims the task — the wake is never silently dropped, no re-wake storm fires, and the never-asks guard path (offline + re-wake of the next capable) stays green Prints PASS/FAIL lines Statum-style; exits non-zero on any failure. Plan 11 task D+E (UI + notifications), the `push` shard: - task page: progress feed above the status card, status + description in a 50/50 grid, edit expander under the history expander, plan behind an expander, answered blocks behind an expander; queue screen: New Task expander above the filters; dashboard waiting-on-drafts block with links to where Enqueue lives (pages shard) - real Web Push over VAPID: keypair generated at startup into web-config.json, GET /sw.js service worker with click-through, subscribe/unsubscribe + per-task Notify me actions, block events and waiting-on-drafts pushed globally, first start/failure/completion to opted-in users, best-effort queue with dispatcher-tick retry, 410/404 endpoint pruning — payloads verified protocol-level (RFC 8291 decryption + ES256 JWT signature) because the harness owns the browser-side keys Heredoc completion reports (pages shard): - terminal tasks render tasks.summary (set by done/fail) as a markdown completion-report card below the status/description grid — "Completion report" when done, "Failure summary" when failed, no card when empty; the summary renders through the shipped stm-markdown renderer (headings/code, no syntax leaks, HTML escaped) and the dashboard timeline snippets its first line Speed pass (plan 11, task 0): - SHARDING: the scenario groups are partitioned into seven independent bundles (pages, dispatch, subtasks, lineage, allocation, lifecycle, answers), each running in its own process against its own app instance, fake daemons and SQLite database. The default invocation runs the bundles SEQUENTIALLY, in-process, one after another — the unbounded parallel fan-out (one app + fake daemons per shard, all at once) OOM'd the machine. `--bundle NAME` runs one shard for debugging; `--parallel [N]` opts back into parallel shards, at most N at a time (default 4). - Within a bundle, scenario groups SHARE the app+daemon instance and carry per-group state as database rows (task/session/block fixtures), never process restarts; the one deliberate restart (the startup token migration) lives in the answers bundle. - EVENT-DRIVEN WAITS: wait_for polls at 50ms; the fixed sleeps were converted either to observable-condition waits (daemon-side counters, e.g. failed-delivery hits) or to quiet_window() — a bounded negative-assertion poll that FAILS LOUDLY the moment the event it guards against happens (a sleep could silently pass). - ENV KNOBS: SADDLE_DISPATCH_TICK_S=1 (dispatcher cadence; default 5 in production), SADDLE_HTTP_TIMEOUT_S=2, SADDLE_LOST_WAKE_S=2, SADDLE_STALE_S=3, plus the fake daemon's scripted timings — the wire-traffic shapes stay realistic, just faster. """ import hashlib import json import os import re 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 # Web Push protocol verification (plan 11 task E): the test harness # owns the browser-side keys, so it can do everything the push service # does — decrypt the RFC 8291 body and verify the ES256 VAPID JWT. import base64 from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.asymmetric.utils import encode_dss_signature from cryptography.hazmat.primitives.ciphers.aead import AESGCM from cryptography.hazmat.primitives.kdf.hkdf import HKDF from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat 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") # --- speed-pass tuning ----------------------------------------------------- # The fake daemon's scripted timings and the dispatch-tick cadence the # quiet windows are sized against. All overridable for debugging. TICK_S = float(os.environ.get("SADDLE_DISPATCH_TICK_S", "1") or 1) FAKE_WORK_S = float(os.environ.get("WEBSTACK_FAKE_WORK_S", "1.2") or 1.2) FAKE_FAIL_S = float(os.environ.get("WEBSTACK_FAKE_FAIL_S", "0.5") or 0.5) FAKE_HEARTBEAT_S = float(os.environ.get("WEBSTACK_FAKE_HEARTBEAT_S", "0.4") or 0.4) FAKE_TRIGGER_S = float(os.environ.get("WEBSTACK_FAKE_TRIGGER_S", "0.05") or 0.05) SHARD_TIMEOUT_S = float(os.environ.get("WEBSTACK_SHARD_TIMEOUT_S", "600") or 600) def quiet_seconds(): """A negative-assertion window spanning several dispatch ticks (the converted fixed sleeps were 3-7s at the old 5s tick; the bound never exceeds the old timeout).""" return min(2 * TICK_S + 0.8, 7.0) WORK = os.environ.get("WEBSTACK_WORK", "/tmp/kilo/webstack-e2e") DB_PATH = os.path.join(WORK, "web.sqlite") passes = 0 failures = 0 # The current shard's fixture context (goal_state reads its auth — the # restart scenarios re-login and re-point ctx.auth mid-run) CTX = None 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 bundle_env(name): """Point the module's runtime globals at one shard's scratch dir, database and ports (each bundle is a process with its own app, daemons and SQLite file).""" global WORK, DB_PATH, APP_PORT, DAEMON_PORT, WEB base = os.environ.get("WEBSTACK_WORK", "/tmp/kilo/webstack-e2e") WORK = "%s-%s" % (base, name) DB_PATH = os.path.join(WORK, "web.sqlite") 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: pairs = [] for k, v in data.items(): for item in (v if isinstance(v, (list, tuple)) else [v]): pairs.append((k, item)) body = "&".join("%s=%s" % (k, urllib.parse.quote(str(item))) for k, item in pairs).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) # Retry dropped/refused connections: the freshly started app can # close a very early keep-alive connection once while warming up, # and its first seconds can wobble under load. for attempt in range(6): 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 >= 5: raise time.sleep(0.75) 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 = directive["snapshot_frame"].get("content") if not content: continue content = json.loads(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.05): 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)) dump_threads() return None def quiet_window(seconds, predicate, what): """A converted fixed sleep: asserts a NEGATIVE condition (nothing may happen) across a window spanning several dispatch ticks, polling at 50ms. The moment the condition breaks the window fails loudly (diagnostics + thread dump) instead of silently passing a sleep.""" deadline = time.monotonic() + seconds while time.monotonic() < deadline: value = predicate() if value is False or value is None: remaining = deadline - time.monotonic() print(" (quiet window '%s' broken after %.2fs of %.2fs — the" " guarded event DID happen)" % (what, seconds - remaining, seconds)) dump_threads() return False time.sleep(0.05) return True def dump_threads(): """Stacks of suite threads not parked in sleep — pinpoints a wedged fake lifecycle when a wait times out.""" import traceback shown = 0 for thread_id, frame in sys._current_frames().items(): stack = "".join(traceback.format_stack(frame)[-4:]) if (("harness.py" in stack or "bundles" in stack or "web_stack_test.py" in stack) and ("sleep" not in stack) and shown < 6): shown += 1 print(" --- thread %s ---\n%s" % (thread_id, stack)) 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 enqueue_task(auth, task_id): """The human-only Enqueue action (plan 11): draft -> queued.""" detail = task_state(auth, task_id) assert detail["status"] == "draft", "task %d not in draft (%s)" % (task_id, detail["status"]) act(auth, detail["enqueue"], {}) return task_state(auth, task_id)["status"] def create_queued(auth, fields): """Create via the queue form (lands in draft) and enqueue it — the standard 'this task should dispatch' fixture of the plan-11 world.""" act(auth, slot_public(auth, "/tasks", "tasks")["create"], fields) task_id = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"] if t["title"] == fields["title"]][0]) enqueue_task(auth, task_id) return task_id 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) def set_model_config(**fields): """Writes fields of the model_config singleton row (the Admin → Model page's storage) — the env-var config is GONE, so the bundles that used to point the advisor at a scripted LLM via ADVISOR_LLM_BASE_URL + an app restart now write the row instead. Only touches the named fields; the row is created first when the shard's app has not read its configuration yet. INSERT OR IGNORE, not SELECT-then-INSERT: the app's own lazy config reader seeds the same singleton concurrently (ModelConfigService.current), and a plain INSERT loses that race with a UNIQUE violation.""" conn = db() try: conn.execute("insert or ignore into model_config (id) values (1)") for name, value in fields.items(): conn.execute("update model_config set %s = ? where id = 1" % name, (value,)) conn.commit() finally: conn.close() def http_get_page(auth, uri): """The compiled page HTML for a statum route (auth via the slot header), for asserting page-level embeds the slot state cannot see.""" status, body = http("GET", uri, headers={ "X-Statum-Slot": "key=%s; as_at=%s" % (auth["key"], auth["as_at"])}) if status != 200: raise SystemExit("GET %s -> HTTP %d" % (uri, status)) return body def goal_state(goal_id): """The goal slot for the CURRENT shard's admin login (the x7 per-bundle duplicates collapsed here, plan 16 R5). Reads the runner's fixture context — the restart scenarios re-login and re-point ctx.auth mid-run, and late binding follows.""" return slot_public(CTX.auth, "/goals/%d" % goal_id, "goal") def saddle_card(auth, name): rows = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == name] return rows[0] def queued_count(): conn = db() try: return conn.execute( "select count(*) from tasks where status = 'queued'").fetchone()[0] finally: conn.close() def command_rows(session): conn = db() try: return conn.execute( "select kind, payload from events" " where session_id = ? order by id", (session,)).fetchall() finally: conn.close() def event_rows(task_id, kinds=None): """The task's ledger rows (kind, payload JSON), oldest first.""" conn = db() try: if kinds: marks = ",".join("?" for _ in kinds) return conn.execute( "select kind, payload from events" " where scope_type = 'task' and scope_id = ? and kind in (%s)" " order by id" % marks, (task_id,) + tuple(kinds)).fetchall() return conn.execute( "select kind, payload from events" " where scope_type = 'task' and scope_id = ? order by id", (task_id,)).fetchall() finally: conn.close() class FakeSaddle: """In-process saddle daemon implementing the wire contract. Phase-2 push behaviour: the daemon asks POST /api/saddle/next-task whenever idle-awake (at registration, on POST /wake, after its task reaches a terminal state, and when a /status probe observes it parked idle — throttled), accepts assignments like POST /start used to, then PUSHES its telemetry home: raw PTY chunks via POST /api/saddle/chunks and round results with state transitions via POST /api/saddle/results (working at round start, terminal states when the scripted lifecycle ends them). /output, /log and /start are gone (404) like a post-phase-2 daemon; /task-update serves the cancel/comment/subtask-status kinds; POST /answer receives a delivered verdict when the round is parked in a block's answer wait (409 when nothing waits — the real daemon's contract; the resumed round continues after the response). Every trigger can be disabled per test for determinism; ask_now() performs a synchronous manual ask. """ PROMPT = "bbarrow@fedora:/workspace/task-1$ " def __init__(self): self.lock = threading.Lock() self.start_payloads = [] # captured assignment payloads (/start + next-task 200s) self.state = "idle" self.summary = "" self.state_entered = time.monotonic() self.fail_next = False # next task drives the task to failed self.start_counts = {} # task id -> assignment count self.resumed_pending = False # crash-resume flag on the next /status self.suspend_on_title = None # this task (by title) parks as suspended self.output_seq = 100 # daemon-global entry ids for pushed results self.last_activity_epoch = int(time.time()) self.hang_next = {} # path suffix -> hang count (never responds) self.abort_failures = 0 # cancel task-updates failing (500) while > 0 self.abort_log = [] # task ids aborted via task-update cancel self.running_task = 0 # task id the daemon believes it runs self.running_title = "" # its title (suspend targeting) self.session_id = None # the web-side run row of the running task self.hold_state = False # freeze lifecycle + heartbeat pushes self.ssh_public_key = "" # reported on /status + /register self.workspace_root = "/workspace" # goal 8 task 6: reported on /register self.started_mono = time.monotonic() self.freeze_activity = False # hold last_activity_epoch still self.silent = False # suppress pushes (staleness tests) self.dead = False # kill(): no serving AND no calls home self.status_hits = 0 # GET /status arrivals (probe accounting) self.comment_log = [] # recorded /task-update comment payloads self.scripted_corpus = None # chunk batch pushed when a task starts # --- plan 10 phase 1: ask/wake/register --- self.token = None # per-saddle token from POST /register self.register_log = [] # recorded /register bodies self.wake_log = [] # recorded POST /wake arrivals self.asks = [] # [{"status": int, "payload": dict|None}] self.ask_on_register = True self.ask_on_wake = True self.ask_on_terminal = True self.ask_on_idle_status = True self.ask_delay = 0.0 # seconds; a slow daemon delays its wake-triggered ask self.serve_register = True # False: 404 /register (pre-flip daemon) self.last_ask_trigger = 0.0 self.port = None self.ask_threads = [] # scheduled ask-trigger threads (settle_asks joins them) # --- answer delivery (the parked-session resume) --- self.pending_block = None # the block payload this daemon waits on (state blocked) self.answer_log = [] # recorded POST /answer bodies self.answer_push_delay = 0.2 # the resumed round's first push, after the response # --- plan 12: goal-agent scripting --- self.goal_script = [] # behaviours consumed per goal-run assignment self.goal_behaviour = None # the current round's popped behaviour self.goal_assignments = [] # captured goal-run assignment payloads def set_state(self, state, summary=None): """Advances the scripted lifecycle.""" self.state = state if summary is not None: self.summary = summary self.state_entered = time.monotonic() def touch_activity(self): if not self.freeze_activity: self.last_activity_epoch = int(time.time()) def queue_output(self, entry_type, command, output): """Pushes one command_output event home (the daemon's entry log now rides the events channel; the old results-entries path is gone).""" self.push_events([{"kind": "command_output", "payload": {"command": command, "text": output}}]) # --- plan 10 phase 2: the push channels --- def push_events(self, events): """Pushes a batch of ledger events home (plan 17 §7.2): each element a {seq?, kind, actor?, ts?, payload?} dict; a missing seq is assigned from the per-session monotonic counter. Pushes are suppressed while silent/hold_state (the staleness tests' silent runner).""" if self.dead or self.silent or self.token is None or self.session_id is None: return batch = [] with self.lock: for event in events: item = dict(event) item["seq"] = int(item.pop("seq", 0) or self.next_push_seq()) item.setdefault("kind", "progress") item.setdefault("actor", "model") item.setdefault("payload", {}) batch.append(item) try: saddle_api("POST", "/api/saddle/chunks", self.token, {"task_id": self.running_task, "session_id": self.session_id, "events": batch}) except (OSError, ValueError): pass # app restarting / shutting down mid-test def push_chunks(self, chunks): """Legacy-shaped helper: raw texts (or {seq, text} dicts, a redelivery) push as progress events carrying the text.""" batch = [] for chunk in chunks: if isinstance(chunk, dict): batch.append({"seq": chunk.get("seq"), "kind": chunk.get("kind", "progress"), "payload": chunk.get("payload", {"text": chunk.get("text", "")})}) else: batch.append({"kind": "progress", "payload": {"text": chunk}}) self.push_events(batch) def next_push_seq(self): # The daemon-side per-session event counter for ad-hoc pushes. self._push_seq = getattr(self, "_push_seq", 0) + 1 return self._push_seq def push_results(self, state, entries=None, block=None, summary=None, session_id=None, reason=None): """Pushes one round-results payload home (state transition plus its entries). Returns the HTTP status when it went out, None when suppressed.""" if self.dead or self.silent: return None if self.token is None: return None target = session_id if session_id is not None else self.session_id if target is None: return None payload = {"session_id": target, "state": state} if block is not None: payload["block"] = block if state == "blocked": # The daemon parks its round on this block's answer wait # (the real daemon's wait_for_answer). self.pending_block = dict(block) if summary is not None: payload["summary"] = summary if reason is not None: payload["reason"] = reason try: status, _ = saddle_api("POST", "/api/saddle/results", self.token, payload) if status != 200: print(" [fake %s] push %s -> HTTP %s (task=%s)" % (self.port, state, status, self.running_task)) return status except (OSError, ValueError) as error: print(" [fake %s] push %s failed: %r (task=%s)" % (self.port, state, error, self.running_task)) return None # --- plan 10 phase 1: the ask path --- def ask_now(self): """Synchronously asks the web for work (a phase-1 daemon's idle ask). A 200 assignment is accepted exactly like POST /start; returns (status, payload|None).""" token = self.token if token is None: raise SystemExit("fake daemon asked before receiving its token") status, payload = saddle_api("POST", "/api/saddle/next-task", token) with self.lock: self.asks.append({"status": status, "payload": payload}) if status == 200: self.accept_assignment(payload) return status, payload def _trigger_ask(self, reason): """Schedules a delayed ask (the daemon woke / went idle-awake). ask_delay models a SLOW asker — a daemon still winding down its round when the wake lands: it honours the wake, but its ask arrives late (still inside the web's lost-wake guard window).""" def run(): if self.state in ("working", "failing"): return # a running daemon does not ask if self.ask_delay > 0: time.sleep(self.ask_delay) for attempt in range(3): try: status, _ = self.ask_now() if status in (200, 204): return print(" (ask %s -> HTTP %s, attempt %d)" % (reason, status, attempt + 1)) except Exception as error: # app restarting / blips: retry print(" (ask %s failed: %r, attempt %d)" % (reason, error, attempt + 1)) time.sleep(1.0) def starter(): time.sleep(FAKE_TRIGGER_S) run() thread = threading.Thread(target=starter, daemon=True) with self.lock: self.ask_threads.append(thread) thread.start() def settle_asks(self, timeout=15): """Joins every ask-trigger thread scheduled so far — the event-driven replacement for the fixed sleeps that waited for in-flight asks to land after triggers were frozen.""" with self.lock: threads = [t for t in self.ask_threads if t.is_alive()] deadline = time.monotonic() + timeout for thread in threads: thread.join(timeout=max(0.1, deadline - time.monotonic())) with self.lock: self.ask_threads = [t for t in self.ask_threads if t.is_alive()] return all(not thread.is_alive() for thread in threads) def maybe_trigger_ask(self, reason, throttle=False): """Schedules an idle-awake ask. Protocol events (wake, abort, round end, park) ask singularly and immediately; only the /status-observation path throttles (a burst of probes must not hammer the ask endpoint).""" now = time.monotonic() if throttle and now - self.last_ask_trigger < 1.0: return self.last_ask_trigger = now self._trigger_ask(reason) def accept_assignment(self, payload): """A 200 next-task assignment: adopted exactly like POST /start.""" self.start_payloads.append(payload) task = payload.get("task") or {} task_id = task.get("id") self.running_task = int(task_id or 0) self.running_title = task.get("title") or "" self.session_id = payload.get("session_id") # The assignment's answers resolve any block this daemon was # parked on (resume-carries-answer). self.pending_block = None count = self.start_counts.get(task_id, 0) + 1 self.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.resumed_pending = True # Plan 12: a goal-run assignment plays its scripted behaviour # instead of the plain done push (plan submission, parking, # question, completion, failure — consumed one per round). self.goal_behaviour = None if payload.get("kind") == "goal_run": self.goal_assignments.append(payload) self.goal_behaviour = self.goal_script.pop(0) if self.goal_script else ("suspend",) self.set_state("failing" if self.fail_next else "working") self.fail_next = False # The round's opening telemetry: the §7.2 event batch (turn 1: # lifecycle, turn text, tool call + result, usage) plus the # working state push. A new session restarts the per-session # seq counter; a scripted corpus supplements the batch. self._push_seq = 0 opening = [ {"kind": "lifecycle", "actor": "harness", "payload": {"action": "start", "text": "round started"}}, {"kind": "turn_text", "payload": {"turn": 1, "text": "Starting the assigned task."}}, {"kind": "tool_call", "payload": {"turn": 1, "tool": "run_command", "args": {"command": "echo hello"}}}, {"kind": "tool_result", "actor": "harness", "payload": {"turn": 1, "stdout": "hello\n", "exit_code": 0}}, {"kind": "usage", "actor": "harness", "payload": {"turn": 1, "input_tokens": 120, "output_tokens": 30}}, ] if self.scripted_corpus is not None: opening = [dict(chunk) for chunk in self.scripted_corpus] self.push_events(opening) self.push_results("working") class Handler(BaseHTTPRequestHandler): daemon = None # set on the class before serving protocol_version = "HTTP/1.1" def log_message(self, fmt, *args): pass def maybe_hang(self, path): """True when this request should hang forever (never respond), consuming one unit of the path's hang budget.""" for suffix, count in list(self.daemon.hang_next.items()): if count > 0 and path.startswith(suffix): self.daemon.hang_next[suffix] = count - 1 return True return False 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): if self.maybe_hang(self.path): time.sleep(600) return self.daemon.touch_activity() length = int(self.headers.get("Content-Length") or 0) raw = self.rfile.read(length).decode() if length else "{}" if self.path == "/register": # Plan 10: the web delivers the per-saddle token + its # own base URL; the daemon persists the token for its # calls home and reports its ssh key + version. if not self.daemon.serve_register: self._send(404, {"error": "no such route"}) return try: payload = json.loads(raw) except ValueError: payload = {"raw": raw} self.daemon.register_log.append(payload) if payload.get("saddle_token"): self.daemon.token = payload["saddle_token"] self._send(200, {"ok": True, "ssh_public_key": self.daemon.ssh_public_key, "version": "0.1-fake", "workspace_root": self.daemon.workspace_root}) if self.daemon.ask_on_register: self.daemon.maybe_trigger_ask("register") return if self.path == "/wake": # Plan 10: a capable task is ready; a sleeping daemon # asks for work (immediately — a wake is a direct # command, never throttled). self.daemon.wake_log.append(time.monotonic()) self._send(200, {"ok": True}) if self.daemon.ask_on_wake: self.daemon.maybe_trigger_ask("wake") return if self.path == "/answer": # Plan 11 (daemon Task A): parks FREE the daemon, so no # block ever waits in-process — the verdict travels in # the resuming allocation's resume_notice. Always 409, # exactly like the real daemon's contract. try: payload = json.loads(raw) except ValueError: self._send(400, {"error": "bad json"}) return self.daemon.answer_log.append({ "refused": True, "answer": payload.get("answer", "")}) self._send(409, {"error": "no block is waiting for an answer — parks resume via allocation"}) return if self.path == "/task-update": # Plan 10: cancel (abort machinery) and comment # (injected turn) — delivered directly to a RUNNING # daemon, no auth. try: payload = json.loads(raw) except ValueError: self._send(400, {"error": "bad json"}) return kind = payload.get("kind") if kind == "cancel": if self.daemon.abort_failures > 0: self.daemon.abort_failures -= 1 self._send(500, {"error": "cancel failed (simulated)"}) return aborted = self.daemon.running_task self.daemon.set_state("idle") self.daemon.summary = "" self.daemon.running_task = 0 self.daemon.session_id = None self.daemon.abort_log.append(aborted) self._send(200, {"ok": True, "aborted_task_id": aborted}) # An aborted daemon returns to its idle loop: it asks. self.daemon.maybe_trigger_ask("aborted") elif kind == "comment": self.daemon.comment_log.append(payload) self._send(200, {"ok": True}) elif kind == "answer": # Goal 28 task F (busy-answer delivery): the web # injects a block verdict into the RUNNING round — # a human answered while this session was already # running, so no allocation resume can deliver it. self.daemon.answer_log.append({ "injected": True, "text": payload.get("text", ""), "task_id": payload.get("task_id")}) self._send(200, {"ok": True}) else: self._send(400, {"error": "unknown task-update kind"}) return self._send(404, {"error": "no such route"}) self.close_connection = True def do_GET(self): if self.maybe_hang(self.path): time.sleep(600) return if self.path != "/status": # /status itself is not daemon activity: the liveness # signal must reflect loop rounds / command completions. self.daemon.touch_activity() if self.path == "/status": self.daemon.status_hits += 1 response = {"state": self.daemon.state, "summary": self.daemon.summary, "uptime_seconds": int(time.monotonic() - self.daemon.started_mono), "last_activity_epoch": self.daemon.last_activity_epoch, "ssh_public_key": self.daemon.ssh_public_key, # The wire contract's task confirmation (the # probe-recovery flip): a running daemon names # the task it runs; an idle one reports null. # Without it every answered probe looks like a # crash orphan and requeues the task. "task_id": self.daemon.running_task or None} if getattr(self.daemon, "status_extra", None): response.update(self.daemon.status_extra) if self.daemon.resumed_pending: response["resumed"] = True self.daemon.resumed_pending = False # A parked/terminal daemon is idle-awake: a phase-1 # daemon asks (and sleeps on 204). Throttled so status # poll bursts do not hammer the ask endpoint. if self.daemon.ask_on_idle_status and self.daemon.state in ( "idle", "done", "failed", "suspended"): self.daemon.maybe_trigger_ask("idle-status", throttle=True) self._send(200, response) else: self._send(404, {"error": "no such route"}) self.close_connection = True def run_lifecycle(self): """Applies the scripted lifecycle on a timer (like a real daemon whose task ends on its own): working heartbeats chunks (the flush timer's stand-in, keeping the staleness clock fed), works for FAKE_WORK_S, then pushes done/suspended; failing pushes failed after FAKE_FAIL_S — and a task END makes the idle-awake daemon ask again.""" while True: time.sleep(0.1) if self.hold_state or self.silent: # hold_state freezes the script; silent models a daemon # mid-round with nothing to flush — its round has not # ended, so no terminal push fires while quiet. continue if self.state == "failing": if time.monotonic() - self.state_entered >= FAKE_FAIL_S: self.set_state("failed", "bash -n rejected the script (simulated)") self.push_results("failed", summary="bash -n rejected the script (simulated)") if self.ask_on_terminal: self.maybe_trigger_ask("terminal") elif self.state == "working": if time.monotonic() - self.state_entered >= FAKE_WORK_S: if self.goal_behaviour is not None: # A goal-agent round: play the scripted # behaviour (the results channel carries the # goal-context verbs). behaviour, self.goal_behaviour = self.goal_behaviour, None kind = behaviour[0] if kind == "plan": self.set_state("blocked") self.push_results("blocked", block={ "type": "plan", "question": "Plan submitted for review", "body": behaviour[1]}) if self.ask_on_idle_status: self.maybe_trigger_ask("parked") elif kind == "question": self.set_state("blocked") self.push_results("blocked", block={ "type": "question", "question": behaviour[1], "options": list(behaviour[2]) if len(behaviour) > 2 else None}) if self.ask_on_idle_status: self.maybe_trigger_ask("parked") elif kind == "suspend": self.set_state("suspended") self.push_results("suspended") if self.ask_on_idle_status: self.maybe_trigger_ask("parked") elif kind == "fail": self.set_state("failed", behaviour[1]) self.push_results("failed", summary=behaviour[1]) if self.ask_on_terminal: self.maybe_trigger_ask("terminal") else: # "done": the completion report self.set_state("done", behaviour[1]) self.push_results("done", summary=behaviour[1]) if self.ask_on_terminal: self.maybe_trigger_ask("terminal") elif self.suspend_on_title and self.suspend_on_title == self.running_title: # `saddle suspend`: park until a fresh assignment # resumes the workspace (title-targeted one-shot, # so an interleaved dispatch cannot eat the flag). self.set_state("suspended") self.suspend_on_title = None self.push_results("suspended") # A parked daemon is idle-awake: it asks (and # sleeps on 204) so a later wake can reach it. if self.ask_on_idle_status: self.maybe_trigger_ask("parked") else: self.set_state("done", "fake daemon finished") self.push_results("done", summary="fake daemon finished") if self.ask_on_terminal: self.maybe_trigger_ask("terminal") elif time.monotonic() - getattr(self, "last_heartbeat", 0) >= FAKE_HEARTBEAT_S: # The flush timer: output keeps flowing, so chunks # keep arriving (the staleness window never opens on # a healthy runner). self.last_heartbeat = time.monotonic() self.push_chunks(["working-heartbeat\r\n"]) def serve(self, port=None): """Binds the daemon's HTTP server. A requested port that is already taken (free_port() hands out recyclable ports — two calls can collide) falls back to an OS-assigned one instead of silently dying and impersonating whatever owns that port.""" if port is None: port = DAEMON_PORT handler = type("BoundHandler", (FakeSaddle.Handler,), {"daemon": self}) self.server = None try: self.server = ThreadingHTTPServer(("127.0.0.1", port), handler) except OSError: print("(fake daemon: port %d taken, falling back)" % port) self.server = ThreadingHTTPServer(("127.0.0.1", 0), handler) self.port = self.server.server_address[1] threading.Thread(target=self.server.serve_forever, daemon=True).start() threading.Thread(target=self.run_lifecycle, daemon=True).start() return self.server def kill(self): """The daemon dies: it stops listening (connections refused) AND stops calling home — no pushes, no asks (a dead process has no outbound anything; in-flight handler threads may linger but the wire contract's traffic is one-directional now).""" self.dead = True if getattr(self, "server", None) is not None: self.server.shutdown() self.server.server_close() self.server = None def start_app(extra_env=None): 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) if extra_env: env.update(extra_env) env.update({ "SPRY_DB_PATH": DB_PATH, "SPRY_SEED_ADMIN_USERNAME": "admin", "SPRY_SEED_ADMIN_PASSWORD": "adminpass123", # The one-time model-config bootstrap: the app seeds the # model_config row from this env on its FIRST read (the row # is the live configuration afterwards — the Admin → Model # page owns it). "SADDLE_LLM_BASE_URL": "http://127.0.0.1:%d" % DAEMON_PORT, "SADDLE_PROJECTS_ROOT": os.path.join(WORK, "workspace"), "SADDLE_HTTP_TIMEOUT_S": "2", "SADDLE_LOST_WAKE_S": "2", "SADDLE_STALE_S": "3", "SADDLE_DISPATCH_TICK_S": "1", # Plan 12: the goal-agent resume coalescing window shrinks to # one second so the event-driven resume tests settle quickly. "SADDLE_GOAL_RESUME_S": os.environ.get("SADDLE_GOAL_RESUME_S", "1"), # Goal 28 (ghost-working sweep): compress the stale window in # tests — the ghostworking bundle sets its own default here. "SADDLE_STALE_WORKING_S": os.environ.get("SADDLE_STALE_WORKING_S", "6"), # Never inherit a caller-poisoned LD_LIBRARY_PATH (e.g. one that # resolves the ancient /lib64 copies): empty makes the loader use # the binary's RUNPATH, which is the dev stack it was built # against. "LD_LIBRARY_PATH": "", }) log_previous = os.path.join(WORK, "app.log.1") if os.path.exists(os.path.join(WORK, "app.log")): shutil.move(os.path.join(WORK, "app.log"), log_previous) log = open(os.path.join(WORK, "app.log"), "w") command = [APP, str(APP_PORT)] if os.environ.get("WEBSTACK_GDB"): # Debugging aid: run the app under gdb; a crash backtrace lands in # app.log. WEBSTACK_GDB=1 meson test ... to enable. command = ["gdb", "-batch", "-quiet", "-ex", "handle SIGPIPE nostop noprint pass", "-ex", "run", "-ex", "bt", "--args", APP, str(APP_PORT)] process = subprocess.Popen(command, cwd=WORK, env=env, stdout=log, stderr=subprocess.STDOUT) # The ready probe must require OUR process alive: a free_port() # TOCTOU can hand this port to another shard's app — ours then dies # at bind (MHD error()) while the probe happily answers from the # intruder (same admin seed), and every check would silently drive # the WRONG app. poll() != None with a serving port is exactly that # shape: take the exit branch, the start_bundle retry re-ports. outcome = wait_for(lambda: (app_ready() and process.poll() is None and "ready") or (process.poll() is not None and "app-exited-%d" % process.poll()) or False, timeout=30, what="app ready") if outcome is None: raise SystemExit("app did not come up; see %s/app.log" % WORK) if str(outcome).startswith("app-exited-"): raise SystemExit("app exited during startup (%s); see %s/app.log" % outcome, WORK) return process def app_ready(): try: status, _ = http("GET", "/login", timeout=3) return status == 200 except (urllib.error.URLError, OSError): return False def dump_app_stacks(pid): """Best-effort thread backtrace of the (possibly wedged) app.""" try: fds = os.listdir("/proc/%d/fd" % pid) print("(app open fds at dump: %d)" % len(fds)) kinds = {} for fd in fds[:1200]: try: target = os.readlink("/proc/%d/fd/%s" % (pid, fd)) except OSError: continue key = target.split(":", 1)[0] if ":" in target else target kinds[key] = kinds.get(key, 0) + 1 print("(fd kinds: %s)" % sorted(kinds.items(), key=lambda kv: -kv[1])[:6]) except OSError: pass for command in (["eu-stack", "-p", str(pid)], ["gdb", "-batch", "-quiet", "-p", str(pid), "-ex", "thread apply all bt"]): try: out = subprocess.run(command, capture_output=True, timeout=60) text = out.stdout.decode(errors="replace") + out.stderr.decode(errors="replace") if text.strip(): print("(app stacks via %s)\n%s" % (command[0], text[:12000])) return except (OSError, subprocess.SubprocessError): continue # ============================================================================= # Fake push service + Web Push protocol helpers (plan 11 task E) # ============================================================================= def b64url(raw): return base64.urlsafe_b64encode(raw).rstrip(b"=").decode() def b64url_dec(text): return base64.urlsafe_b64decode(text + "=" * ((4 - len(text) % 4) % 4)) class FakePushService: """In-process Web Push endpoint server: every subscription the tests register points at a path here (a distinct "browser endpoint"). Responses are scriptable per path — a fixed status (the 410-prune fixture) or N transient 500s (the retry-next-tick fixture).""" def __init__(self): self.requests = [] # [{"path", "headers", "body", "at"}] self.fail_next = {} # path -> remaining 500s self.status_for = {} # path -> fixed status self.lock = threading.Lock() self.port = None self.server = None def serve(self): service = self class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, fmt, *args): pass def do_POST(self): length = int(self.headers.get("Content-Length") or 0) body = self.rfile.read(length) status = 201 if self.path in service.status_for: status = service.status_for[self.path] elif service.fail_next.get(self.path, 0) > 0: service.fail_next[self.path] -= 1 status = 500 with service.lock: service.requests.append({ "path": self.path, "headers": {key.lower(): value for key, value in self.headers.items()}, "body": body, "at": time.monotonic(), }) payload = b"{}" self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.port = self.server.server_address[1] threading.Thread(target=self.server.serve_forever, daemon=True).start() return self.server def url(self, path): return "http://127.0.0.1:%d%s" % (self.port, path) def hits(self, prefix): with self.lock: return [request for request in self.requests if request["path"].startswith(prefix)] def decrypt_webpush(request, client_private_key, client_public, auth_secret): """RFC 8291 decryption of one captured push: returns (record_size, payload dict) — the payload builder's exact output.""" body = request["body"] salt = body[:16] record_size = int.from_bytes(body[16:20], "big") key_id_len = body[20] eph_public = body[21:21 + key_id_len] ciphertext = body[21 + key_id_len:] eph_point = ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), eph_public) shared = client_private_key.exchange(ec.ECDH(), eph_point) info = b"WebPush: info\x00" + eph_public + client_public cek = HKDF(algorithm=hashes.SHA256(), length=16, salt=salt, info=info).derive(shared + auth_secret) nonce = HKDF(algorithm=hashes.SHA256(), length=12, salt=salt, info=info).derive(shared + auth_secret) plaintext = AESGCM(cek).decrypt(nonce, ciphertext, None) return record_size, json.loads(plaintext[:-1]) # strip the 0x02 delimiter def verify_vapid(request, vapid_public, vapid_private): """Verifies the ES256 VAPID JWT on one captured push; returns its claims dict (aud/exp/sub) — raises on a bad signature.""" header = request["headers"].get("authorization", "") assert header.startswith("vapid t="), "no vapid authorization header" token_part = header[len("vapid t="):].split(",")[0] sent_key = header.split("k=")[1] head_b64, claims_b64, sig_b64 = token_part.split(".") signature = b64url_dec(sig_b64) key = ec.derive_private_key( int.from_bytes(b64url_dec(vapid_private), "big"), ec.SECP256R1()) key.public_key().verify( encode_dss_signature(int.from_bytes(signature[:32], "big"), int.from_bytes(signature[32:], "big")), (head_b64 + "." + claims_b64).encode(), ec.ECDSA(hashes.SHA256())) claims = json.loads(b64url_dec(claims_b64)) assert sent_key == vapid_public, "the k= key is not the app's VAPID key" return claims def action_notify(result): """The message of the first notify directive an action returned.""" for directive in result: if directive.get("type") == "notify": return directive.get("message", "") return None # ============================================================================= # Shard orchestration # ============================================================================= class Ctx: """Per-bundle shared fixtures: the app process, the admin auth slot and the primary fake saddle (registered + token delivered).""" def start_bundle(name): """Boots one shard's environment: scratch dir, database, ports, the primary fake daemon and the app, then the shared fixtures every bundle builds on (the demo project + the registered fake saddle). Scenario groups inside the bundle share this app+daemon instance and sequence their state as DB rows, never process restarts.""" global APP_PORT, WEB, DAEMON_PORT bundle_env(name) fake = FakeSaddle() fake.serve() DAEMON_PORT = fake.port # the actually-bound port (fallback-safe) # A rare framework-level crash (stack corruption in Statum slot # signing during the FIRST login — shared-library codegen bug, moved # twice already) kills the app before any test runs; a clean restart # (with fresh ports) sidesteps it. Only tolerated while nothing has # been checked yet. auth = None app = None for attempt in range(3): APP_PORT = free_port() WEB = "http://127.0.0.1:%d" % APP_PORT if os.path.exists(WORK): shutil.rmtree(WORK) os.makedirs(os.path.join(WORK, "workspace")) try: app = start_app() auth = login("admin", "adminpass123") break except (SystemExit, OSError, urllib.error.URLError): if passes > 0: raise if app is not None and app.poll() is None: app.terminate() try: app.wait(timeout=5) except subprocess.TimeoutExpired: app.kill() app = None print("(app died before login; restarting cleanly, attempt %d)" % (attempt + 1)) if app is None or auth is None: raise SystemExit("app never came up; see %s/app.log" % WORK) state = slot_public(auth, "/projects", "projects") act(auth, state["create"], { "name": "demo", "git_url": "https://example.com/demo.git", "default_branch": "main", }) saddles_state = slot_public(auth, "/saddles", "saddles") act(auth, saddles_state["register"], {"name": "fake-saddle", "host": "127.0.0.1:%d" % fake.port}) registered = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"] if s["name"] == "fake-saddle"] if wait_for(lambda: len(fake.register_log) > 0 or False, timeout=15, what="/register delivery") is None: raise SystemExit("fake saddle never received /register") ctx = Ctx() ctx.name = name ctx.app = app ctx.auth = auth ctx.fake = fake ctx.fake_id = int(registered[0]["id"]) ctx.fake_token = registered[0]["token"] return ctx BUNDLES_ORDER = ("pages", "t7log", "dispatch", "subtasks", "lineage", "allocation", "lifecycle", "answers", "answerwake", "ghostworking", "turncappark", "modelconfig", "push", "goals", "replay", "goalui", "forcewake", "users", "theme", "stacks", "notes", "resurrect", "t4lists", "t8fleet", "t9library", "t10views", "origins", "parity", "understanding", "discoveries", "knowledge_cycle", "advisor", "chatui", "advisor_arc", "advisor_fix", "advisor_refine", "advisor_utf8", "canon", "projectdelete", "canon_deploy") def run_bundle(name): """Runs one shard to completion; returns a failure count.""" global CTX started = time.monotonic() ctx = start_bundle(name) CTX = ctx import importlib bundle = importlib.import_module("bundles." + name) # Runtime globals the star-import captured at module load can be # stale (the sequential debug path reuses modules across shards) — # re-point them at this shard's environment. The shared helpers # read them from harness globals already. for runtime_name in ("WORK", "DB_PATH", "APP_PORT", "DAEMON_PORT", "WEB"): setattr(bundle, runtime_name, globals()[runtime_name]) runner = bundle.run try: runner(ctx) finally: # Exit-code diagnostic: distinguishes a crashed app (negative # returncode, e.g. -11 SIGSEGV) from a clean shutdown. When # tearing down mid-exception, also capture the still-alive # app's stacks before killing it (a wedged-but-alive server # otherwise leaves no evidence). print("(app returncode at teardown: %s)" % ctx.app.poll()) if ctx.app.poll() is None and sys.exc_info()[0] is not None: dump_app_stacks(ctx.app.pid) ctx.fake.kill() try: ctx.app.terminate() try: ctx.app.wait(timeout=5) except subprocess.TimeoutExpired: ctx.app.kill() check(ctx.app.poll() in (0, -15), "bundle %s app shut down cleanly (exit %s)" % (name, ctx.app.poll())) except OSError: pass print("%d passed, %d failed" % (passes, failures)) print("[bundle %s] %.1fs" % (name, time.monotonic() - started)) return failures def run_all_parallel(limit): """The explicit opt-in: one subprocess per bundle (each with its own app, daemons and database), at most `limit` in flight at a time (every shard is an app plus fake daemons — an unbounded fan-out OOMs the machine), verdicts aggregated into one exit code and one combined check-count summary.""" base = os.environ.get("WEBSTACK_WORK", "/tmp/kilo/webstack-e2e") log_dir = os.path.join(os.path.dirname(base) or "/tmp/kilo", "webstack-shards") os.makedirs(log_dir, exist_ok=True) started = time.monotonic() limit = max(1, int(limit)) pending = list(BUNDLES_ORDER) running = [] finished = [] while pending or running: while pending and len(running) < limit: name = pending.pop(0) out_path = os.path.join(log_dir, "bundle-%s.log" % name) handle = open(out_path, "w") proc = subprocess.Popen( [sys.executable, "-u", os.path.join(os.path.dirname(os.path.abspath(__file__)), "web_stack_test.py"), APP, "--bundle", name], stdout=handle, stderr=subprocess.STDOUT) running.append((name, proc, handle, out_path, time.monotonic())) time.sleep(0.2) still = [] for name, proc, handle, out_path, spawned in running: if proc.poll() is None: if time.monotonic() - spawned > SHARD_TIMEOUT_S: proc.kill() proc.wait() else: still.append((name, proc, handle, out_path, spawned)) continue handle.close() finished.append((name, proc, out_path)) running = still total_pass = total_fail = 0 broken = [] by_name = {name: (proc, out_path) for name, proc, out_path in finished} for name in BUNDLES_ORDER: proc, out_path = by_name[name] text = open(out_path).read() match = re.search(r"(\d+) passed, (\d+) failed", text) took = re.search(r"\[bundle %s\] ([0-9.]+)s" % name, text) bundle_pass = int(match.group(1)) if match else 0 bundle_fail = int(match.group(2)) if match else -1 total_pass += max(bundle_pass, 0) total_fail += max(bundle_fail, 0) status = "ok" if proc.returncode != 0 or bundle_fail != 0 or match is None: status = "BROKEN (exit %s)" % proc.returncode broken.append(name) print("[shard %-10s] %6ss %3d passed %3d failed %s" % (name, took.group(1) if took else "?", bundle_pass, bundle_fail, status)) if status != "ok": tail = "\n".join(text.rstrip().split("\n")[-60:]) print("--- web-stack/%s tail (full log: %s) ---" % (name, out_path)) print(tail) print("--- end web-stack/%s tail ---" % name) print("shards wall: %.1fs (logs under %s)" % (time.monotonic() - started, log_dir)) print("ALL SHARDS: %d passed, %d failed%s" % (total_pass, total_fail, "" if not broken else " BROKEN: " + ", ".join(broken))) return 1 if broken or total_fail else 0 def main(): if not os.path.exists(APP): raise SystemExit("app binary not found at %s (build first)" % APP) argv = sys.argv[2:] if "--bundle" in argv: sys.exit(1 if run_bundle(argv[argv.index("--bundle") + 1]) else 0) if "--parallel" in argv: at = argv.index("--parallel") limit = 4 if len(argv) > at + 1 and argv[at + 1].isdigit(): limit = int(argv[at + 1]) return run_all_parallel(limit) for name in BUNDLES_ORDER: if run_bundle(name): return 1 return 0 if __name__ == "__main__": sys.exit(main())