fix(server): graceful SIGTERM/SIGINT shutdown + crash-durable queue

Root cause of the "Event loop stopped before Future completed" crash that
wiped the in-memory queue/history (seen 3x today): the fork's _graceful_shutdown
handler (added in a45a441) called event_loop.stop(). The server coroutine passed
to run_until_complete() never completes on its own, so stopping the loop makes
run_until_complete() raise RuntimeError; the surrounding try only caught
KeyboardInterrupt, so the process died non-zero with a traceback and discarded
the queue. The dirty exit also failed to release :8188 promptly, and a duplicate
transient/user `hanzo-studio.service` racing the canonical studio.service turned
that into an EADDRINUSE restart storm.

main.py:
- handler is now idempotent, flags the queue closed, snapshots it, then stops
  the loop; the outer try treats RuntimeError('Event loop stopped before Future
  completed.') as the expected graceful-exit path (logs "Server stopped by
  shutdown signal", exits 0). Verified live in journald.

execution.py:
- opt-in crash-durable PromptQueue behind STUDIO_PERSIST_QUEUE=1. Pending +
  in-flight prompts are snapshotted atomically to user/queue_snapshot.json on
  every change and re-queued on boot. Best-effort: all I/O is guarded so it
  never affects the render path; inert when the flag is unset.

ops/systemd/studio.service:
- enable STUDIO_PERSIST_QUEUE=1 on the single canonical unit.

scripts/studio-service-swap.sh:
- batch-safe reconcile+apply: waits for an empty queue, removes any stray
  duplicate unit, reinstalls the canonical unit, restarts, health-checks.

docs/ops-crash-and-service.md:
- topology (one canonical unit), how to read crash history, the fix, persistence.

tests-unit/execution_test/prompt_queue_persist_test.py:
- round-trip, inert-when-disabled, and corrupt-row-skipping tests.

docs/ops-crash-and-service.md

@@ -0,0 +1,126 @@
+# Hanzo Studio — service & crash operations
+
+How Hanzo Studio runs on the GB10 box, how to read crash history, and the fix
+for the "queue vanished after a restart" class of outage.
+
+## Service topology — ONE canonical unit
+
+Hanzo Studio runs as exactly **one** systemd unit, version-controlled in this
+repo:
+
+- Canonical source: **`ops/systemd/studio.service`**
+- Installed at: `/etc/systemd/system/studio.service` (system unit, `User=z`)
+- Serves on `0.0.0.0:8188`, enabled at boot (`WantedBy=multi-user.target`)
+
+There must be **exactly one** unit bound to :8188. Today's outage was a
+**duplicate-unit collision**: a transient/user `hanzo-studio.service` (from
+`systemd-run`) was created alongside `studio.service`; both had `Restart=` and
+both tried to bind :8188, so they ping-ponged the port in a restart storm
+(`OSError: [Errno 98] address already in use`). The fix is to keep only the
+canonical unit and never re-create a second one.
+
+```bash
+systemctl status studio.service              # health
+sudo systemctl restart studio.service         # graceful restart (drains queue)
+journalctl -u studio.service -f               # live logs
+curl -s http://127.0.0.1:8188/queue           # {"queue_running":[...],"queue_pending":[...]}
+```
+
+Roll out code or unit changes **without interrupting a running batch** (waits
+for an empty queue, removes any stray duplicate, reinstalls the canonical unit,
+restarts, health-checks):
+
+```bash
+scripts/studio-service-swap.sh
+```
+
+## Reading crash history
+
+stdout/stderr go to journald (no separate log file). To inspect crashes:
+
+```bash
+# Everything, newest last:
+journalctl -u studio.service -o short-precise --no-pager
+
+# Just failures / restarts / shutdowns:
+journalctl -u studio.service --no-pager | grep -E 'Traceback|Error|Failed|Main process exited|Scheduled restart|shutting down'
+
+# How did the last process exit? (0 = clean, 1 = crash, killed/9 = SIGKILL/OOM)
+journalctl -u studio.service --no-pager | grep 'Main process exited' | tail
+systemctl show studio.service -p ExecMainStatus -p NRestarts -p Result
+
+# Kernel OOM / GPU VRAM exhaustion:
+sudo dmesg -T | grep -iE 'out of memory|killed process|oom-kill|NV_ERR_NO_MEMORY'
+```
+
+## The crash that ate the queue (root cause + fix)
+
+**Symptom** (seen 3×): on any SIGTERM/SIGINT the process died with
+
+```
+File "main.py", line 507, in <module>
+    event_loop.run_until_complete(x)
+RuntimeError: Event loop stopped before Future completed.
+```
+
+and the in-memory queue + history were lost.
+
+**Root cause.** The fork's signal handler (`_graceful_shutdown` in `main.py`,
+added in `a45a441`) called `event_loop.stop()`. The server coroutine passed to
+`run_until_complete()` never finishes on its own, so stopping the loop makes
+`run_until_complete()` raise `RuntimeError('Event loop stopped before Future
+completed.')`. The surrounding `try` only caught `KeyboardInterrupt`, so the
+process exited **non-zero with a traceback** instead of shutting down cleanly —
+and the dirty exit discarded the queue. Because the exit was dirty, the socket
+was sometimes not released before restart, producing secondary
+`OSError: [Errno 98] address already in use` crashes; the duplicate unit turned
+that into a storm.
+
+**Fix (`main.py`).** The handler now runs once (idempotent), flags the queue
+closed, snapshots it, then stops the loop; the outer `try` treats the resulting
+`RuntimeError('Event loop stopped before Future completed.')` as the expected
+graceful-exit path and logs `Server stopped by shutdown signal`. Verified live:
+
+```
+Received SIGTERM, shutting down gracefully...
+Server stopped by shutdown signal
+```
+
+**Fix (topology).** The duplicate transient/user `hanzo-studio.service` was
+removed; `studio.service` is the single canonical unit.
+
+## Crash-durable queue (STUDIO_PERSIST_QUEUE)
+
+A graceful signal is now handled, but a hard kill can still happen — e.g. a
+SIGKILL/OOM while a batch is queued (observed: a mid-batch SIGKILL dropped 20
+pending prompts). ComfyUI's `PromptQueue` is in-memory, so Studio adds an
+**opt-in durable queue** (`execution.py`), enabled in the unit:
+
+- `Environment=STUDIO_PERSIST_QUEUE=1` in `ops/systemd/studio.service`.
+- On every queue change (submit / start / finish / delete / wipe) the pending
+  **and in-flight** prompts are snapshotted atomically to
+  `user/queue_snapshot.json`.
+- On boot the snapshot is re-queued (an interrupted render restarts from
+  scratch — correct, since it never finished). Look for
+  `[persist-queue] restored N pending prompt(s)` in the log.
+- Persistence is **best-effort**: every read/write is wrapped so a serialization
+  or disk error only logs a warning and never affects rendering. With the flag
+  unset the code path is inert (zero behavior change vs. upstream).
+
+Tests: `tests-unit/execution_test/prompt_queue_persist_test.py` (round-trip,
+inert-when-disabled, corrupt-row skipping). Run standalone:
+`.venv/bin/python tests-unit/execution_test/prompt_queue_persist_test.py`.
+
+## Known-benign shutdown noise
+
+During interpreter teardown you may see a one-shot traceback:
+
+```
+File "studio/model_management.py", line 800, in cleanup_models
+    if current_loaded_models[i].real_model() is None:
+TypeError: 'NoneType' object is not callable
+```
+
+This is an atexit/weakref finalizer racing module teardown (upstream ComfyUI).
+It is cosmetic — it prints after the server has already stopped and does not
+affect the queue or the exit path.

execution.py

@@ -1,7 +1,9 @@
 import copy
 import heapq
 import inspect
+import json
 import logging
+import os
 import sys
 import threading
 import time
@@ -1144,12 +1146,60 @@ class PromptQueue:
         self.currently_running = {}
         self.history = {}
         self.flags = {}
+        # Optional crash-durable queue: snapshot pending + in-flight prompts to
+        # disk on every change, re-queue them on boot. Off by default; enable
+        # with STUDIO_PERSIST_QUEUE=1. Persistence is strictly best-effort and
+        # never affects the render path (all failures are caught and logged).
+        self._persist_path = None
+        if os.environ.get("STUDIO_PERSIST_QUEUE") == "1":
+            import folder_paths
+            self._persist_path = os.path.join(folder_paths.get_user_directory(), "queue_snapshot.json")
+            self._restore_locked()
+
+    def _snapshot_items(self):
+        # Pending queue plus any in-flight items, so a crash mid-render re-queues
+        # the interrupted prompt. Tuples serialize to JSON lists.
+        return [list(it) for it in (list(self.queue) + list(self.currently_running.values()))]
+
+    def _persist_locked(self):
+        if self._persist_path is None:
+            return
+        try:
+            data = json.dumps(self._snapshot_items())
+            tmp = self._persist_path + ".tmp"
+            with open(tmp, "w") as f:
+                f.write(data)
+            os.replace(tmp, self._persist_path)  # atomic
+        except Exception as e:
+            logging.warning("[persist-queue] snapshot failed: %s", e)
+
+    def _restore_locked(self):
+        try:
+            if not os.path.exists(self._persist_path):
+                return
+            with open(self._persist_path) as f:
+                items = json.load(f)
+            restored = 0
+            for it in items:
+                # Minimal validation: (number, prompt_id:str, prompt:dict, ...)
+                if isinstance(it, list) and len(it) >= 3 and isinstance(it[1], str) and isinstance(it[2], dict):
+                    heapq.heappush(self.queue, tuple(it))
+                    restored += 1
+            if restored:
+                logging.info("[persist-queue] restored %d pending prompt(s) from previous run", restored)
+        except Exception as e:
+            logging.warning("[persist-queue] restore failed: %s", e)
+
+    def persist_now(self):
+        with self.mutex:
+            self._persist_locked()
 
     def put(self, item):
         with self.mutex:
             heapq.heappush(self.queue, item)
             self.server.queue_updated()
             self.not_empty.notify()
+            self._persist_locked()
 
     def get(self, timeout=None):
         with self.not_empty:
@@ -1162,6 +1212,7 @@ class PromptQueue:
             self.currently_running[i] = copy.deepcopy(item)
             self.task_counter += 1
             self.server.queue_updated()
+            self._persist_locked()
             return (item, i)
 
     class ExecutionStatus(NamedTuple):
@@ -1190,6 +1241,7 @@ class PromptQueue:
             }
             self.history[prompt[1]].update(history_result)
             self.server.queue_updated()
+            self._persist_locked()
 
     # Note: slow
     def get_current_queue(self):
@@ -1214,6 +1266,7 @@ class PromptQueue:
         with self.mutex:
             self.queue = []
             self.server.queue_updated()
+            self._persist_locked()
 
     def delete_queue_item(self, function):
         with self.mutex:
@@ -1225,6 +1278,7 @@ class PromptQueue:
                         self.queue.pop(x)
                         heapq.heapify(self.queue)
                     self.server.queue_updated()
+                    self._persist_locked()
                     return True
         return False
 

main.py

@@ -494,13 +494,26 @@ if __name__ == "__main__":
 
     event_loop, prompt_server, start_all_func = start_studio()
 
+    # Sentinel: a non-empty list means shutdown is already in progress. Using a
+    # mutable container (not a global) keeps the handler closure-clean.
+    shutdown_started = []
+
     def _graceful_shutdown(signum, frame):
         sig_name = signal.Signals(signum).name
+        if shutdown_started:
+            logging.info(f"Received {sig_name} again, already shutting down...")
+            return
+        shutdown_started.append(True)
         logging.info(f"Received {sig_name}, shutting down gracefully...")
-        # Stop accepting new prompts
+        # Stop accepting new prompts and snapshot any pending work to disk so a
+        # restart can pick it back up (no-op unless STUDIO_PERSIST_QUEUE=1).
         if prompt_server and prompt_server.prompt_queue:
             prompt_server.prompt_queue.set_flag("disable_new", True)
-        # Stop the event loop (will cause run_until_complete to return)
+            prompt_server.prompt_queue.persist_now()
+        # Ask the event loop to stop. Because the server coroutine never finishes
+        # on its own, run_until_complete() then raises RuntimeError('Event loop
+        # stopped before Future completed.'); the except clause below treats that
+        # as the expected graceful-exit path rather than a crash.
         event_loop.call_soon_threadsafe(event_loop.stop)
 
     signal.signal(signal.SIGTERM, _graceful_shutdown)
@@ -512,6 +525,14 @@ if __name__ == "__main__":
         event_loop.run_until_complete(x)
     except KeyboardInterrupt:
         logging.info("\nStopped server")
+    except RuntimeError as e:
+        # A signal handler stopped the loop before the (never-completing) server
+        # coroutine finished; asyncio surfaces this as RuntimeError. This is our
+        # graceful-exit path, not a crash. Anything else is a real error.
+        if "Event loop stopped before Future completed" in str(e):
+            logging.info("Server stopped by shutdown signal")
+        else:
+            raise
 
     logging.info("Cleaning up...")
     cleanup_temp()

ops/systemd/studio.service

@@ -12,6 +12,10 @@ Group=z
 WorkingDirectory=/home/z/work/hanzo/studio
 Environment=HF_HOME=/home/z/.cache/huggingface
 Environment=PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True
+# Crash-durable queue: snapshot pending/in-flight prompts to <user>/queue_snapshot.json
+# on every change and re-queue them on restart, so a kill/OOM mid-batch does not
+# lose queued work. See docs/ops-crash-and-service.md.
+Environment=STUDIO_PERSIST_QUEUE=1
 ExecStart=/home/z/work/hanzo/studio/.venv/bin/python main.py --listen 0.0.0.0 --port 8188 --normalvram --disable-auto-launch --output-directory /home/z/work/hanzo/studio/output
 Restart=on-failure
 RestartSec=8

scripts/studio-service-swap.sh

@@ -0,0 +1,100 @@
+#!/usr/bin/env bash
+# studio-service-swap.sh — batch-safe reconcile + apply for Hanzo Studio.
+#
+# Hanzo Studio runs as exactly ONE unit: the repo-canonical system unit
+# ops/systemd/studio.service (installed at /etc/systemd/system/studio.service,
+# serving :8188). Today's outage was caused by a DUPLICATE unit (a transient/
+# user `hanzo-studio.service`) racing it for the port. This script enforces the
+# single-unit invariant and rolls out code/unit changes WITHOUT interrupting an
+# in-flight render batch:
+#
+#   1. wait until the render queue is empty
+#   2. remove any stray duplicate units (user/transient hanzo-studio.service)
+#   3. (re)install ops/systemd/studio.service -> /etc, enable it, daemon-reload
+#   4. restart studio.service so it picks up the latest code + unit env
+#   5. health-check :8188
+#
+# Idempotent; safe to re-run. Needs sudo for the system unit.
+#
+# Usage:
+#   scripts/studio-service-swap.sh                 # wait for idle, then apply
+#   STUDIO_SWAP_TIMEOUT=0 scripts/...              # apply immediately (skip wait)
+set -euo pipefail
+
+REPO="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+UNIT_SRC="$REPO/ops/systemd/studio.service"
+UNIT_DST="/etc/systemd/system/studio.service"
+PORT="${STUDIO_PORT:-8188}"
+QUEUE_URL="http://127.0.0.1:${PORT}/queue"
+WAIT_TIMEOUT="${STUDIO_SWAP_TIMEOUT:-7200}"   # seconds to wait for empty queue (0 = don't wait)
+POLL_INTERVAL="${STUDIO_SWAP_POLL:-15}"
+
+log() { printf '[swap] %s\n' "$*"; }
+
+queue_state() {
+  # -> busy | idle | down
+  local body
+  body="$(curl -fsS -m 5 "$QUEUE_URL" 2>/dev/null)" || { echo down; return; }
+  python3 - "$body" <<'PY'
+import json, sys
+try:
+    q = json.loads(sys.argv[1])
+    n = len(q.get("queue_running", [])) + len(q.get("queue_pending", []))
+    print("busy" if n > 0 else "idle")
+except Exception:
+    print("down")
+PY
+}
+
+wait_for_idle() {
+  [ "$WAIT_TIMEOUT" -eq 0 ] && { log "STUDIO_SWAP_TIMEOUT=0 -> applying without waiting"; return; }
+  local waited=0 st
+  while :; do
+    st="$(queue_state)"
+    case "$st" in
+      idle) log "queue empty -> proceeding"; return ;;
+      down) log "server not reachable -> proceeding"; return ;;
+      busy) log "queue busy; ${waited}s/${WAIT_TIMEOUT}s elapsed, recheck in ${POLL_INTERVAL}s" ;;
+    esac
+    if (( waited >= WAIT_TIMEOUT )); then
+      log "ERROR: still busy after ${WAIT_TIMEOUT}s; aborting to protect the batch"; exit 1
+    fi
+    sleep "$POLL_INTERVAL"; waited=$((waited + POLL_INTERVAL))
+  done
+}
+
+wait_for_idle
+
+# 2. Kill off any duplicate units that could race :8188.
+if systemctl --user list-units --all 2>/dev/null | grep -q 'hanzo-studio\.service'; then
+  log "removing stray user hanzo-studio.service"
+  systemctl --user disable --now hanzo-studio.service 2>/dev/null || true
+  rm -f "$HOME/.config/systemd/user/hanzo-studio.service" \
+        "$HOME/.config/systemd/user/default.target.wants/hanzo-studio.service"
+  systemctl --user daemon-reload || true
+fi
+
+# 3. Install/refresh the canonical system unit from the repo.
+if ! sudo cmp -s "$UNIT_SRC" "$UNIT_DST" 2>/dev/null; then
+  log "installing canonical unit: $UNIT_SRC -> $UNIT_DST"
+  sudo install -m 0644 "$UNIT_SRC" "$UNIT_DST"
+fi
+sudo systemctl enable studio.service
+sudo systemctl daemon-reload
+
+# 4. Restart to pick up latest code + unit env (queue is empty at this point).
+log "restarting studio.service"
+sudo systemctl restart studio.service
+
+# 5. Health-check.
+log "waiting for :$PORT ..."
+for _ in $(seq 1 60); do
+  if curl -fsS -m 3 "$QUEUE_URL" >/dev/null 2>&1; then
+    log "OK — studio.service is serving on :$PORT (single canonical unit)"
+    exit 0
+  fi
+  sleep 3
+done
+log "ERROR: server did not answer on :$PORT within 180s"
+journalctl -u studio.service -n 40 --no-pager || true
+exit 1

tests-unit/execution_test/prompt_queue_persist_test.py

@@ -0,0 +1,108 @@
+"""Tests for the crash-durable PromptQueue snapshot (STUDIO_PERSIST_QUEUE=1).
+
+Verifies that pending + in-flight prompts survive a process restart: they are
+snapshotted to disk on every queue change and re-queued when a fresh PromptQueue
+is constructed. Also verifies the feature is inert when the flag is unset.
+
+Runs under pytest, or standalone:  python prompt_queue_persist_test.py
+"""
+import json
+import os
+import sys
+import tempfile
+
+# Import from the repo root regardless of where pytest is invoked from.
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+
+import folder_paths  # noqa: E402
+from execution import PromptQueue  # noqa: E402
+
+
+class _StubServer:
+    """PromptQueue only calls server.queue_updated(); everything else is unused."""
+    def queue_updated(self):
+        pass
+
+
+def _item(number, prompt_id):
+    # Mirrors ComfyUI's queue tuple: (number, prompt_id, prompt, extra_data, outputs)
+    return (number, prompt_id, {"1": {"class_type": "Noop", "inputs": {}}}, {"client_id": "t"}, ["1"])
+
+
+def _fresh_user_dir():
+    d = tempfile.mkdtemp(prefix="studio-persist-test-")
+    folder_paths.set_user_directory(d)
+    return os.path.join(d, "queue_snapshot.json")
+
+
+def test_persist_and_restore_round_trip():
+    os.environ["STUDIO_PERSIST_QUEUE"] = "1"
+    snap = _fresh_user_dir()
+    try:
+        q1 = PromptQueue(_StubServer())
+        assert q1._persist_path == snap
+        assert not os.path.exists(snap), "no snapshot should exist before first change"
+
+        q1.put(_item(1, "a"))
+        q1.put(_item(2, "b"))
+        assert os.path.exists(snap), "put() must write a snapshot"
+        with open(snap) as f:
+            assert len(json.load(f)) == 2
+
+        # Start rendering 'a' (highest priority == lowest number). It moves to
+        # currently_running; the snapshot must still cover BOTH so an in-flight
+        # crash re-queues it.
+        item, ident = q1.get(timeout=1)
+        assert item[1] == "a"
+        with open(snap) as f:
+            assert len(json.load(f)) == 2
+
+        # Finish 'a'. Snapshot now holds only the still-pending 'b'.
+        q1.task_done(ident, {}, None)
+        with open(snap) as f:
+            assert [row[1] for row in json.load(f)] == ["b"]
+
+        # Simulate a restart: a brand-new PromptQueue must re-queue 'b'.
+        q2 = PromptQueue(_StubServer())
+        running, queued = q2.get_current_queue_volatile()
+        assert running == []
+        assert len(queued) == 1 and queued[0][1] == "b"
+        # Restored item is a real queue tuple that can be popped and run.
+        item2, _ = q2.get(timeout=1)
+        assert item2[1] == "b" and isinstance(item2[2], dict)
+        print("OK test_persist_and_restore_round_trip")
+    finally:
+        os.environ.pop("STUDIO_PERSIST_QUEUE", None)
+
+
+def test_disabled_by_default_is_inert():
+    os.environ.pop("STUDIO_PERSIST_QUEUE", None)
+    snap = _fresh_user_dir()
+    q = PromptQueue(_StubServer())
+    assert q._persist_path is None
+    q.put(_item(1, "a"))
+    q.get(timeout=1)
+    assert not os.path.exists(snap), "flag unset -> nothing written to disk"
+    print("OK test_disabled_by_default_is_inert")
+
+
+def test_restore_skips_corrupt_rows():
+    os.environ["STUDIO_PERSIST_QUEUE"] = "1"
+    snap = _fresh_user_dir()
+    try:
+        # One valid row, plus junk that must be ignored without raising.
+        with open(snap, "w") as f:
+            json.dump([[3, "good", {"1": {}}, {}, ["1"]], "garbage", [1, 2], {"x": 1}], f)
+        q = PromptQueue(_StubServer())
+        _, queued = q.get_current_queue_volatile()
+        assert len(queued) == 1 and queued[0][1] == "good"
+        print("OK test_restore_skips_corrupt_rows")
+    finally:
+        os.environ.pop("STUDIO_PERSIST_QUEUE", None)
+
+
+if __name__ == "__main__":
+    test_persist_and_restore_round_trip()
+    test_disabled_by_default_is_inert()
+    test_restore_skips_corrupt_rows()
+    print("\nALL PASS")