surface worker errors instead of blocking on the frame queue

app.py

@@ -55,6 +55,7 @@ GITHUB = "https://github.com/JonathanColetti/LiveWan"
 FPS = 16
 BLOCK_SECONDS = 0.75  # 3 latent frames -> 12 pixel frames at 16 fps
 NO_STEER = "don't steer, stay on the opening prompt"
+STALL_SECONDS = 45  # a block is ~0.3 s; this only trips when something is wrong
 
 # ---------------------------------------------------------------- bootstrap
 
@@ -201,19 +202,30 @@ def run(world: str, steer_to: str, steer_at: float, seconds: float, seed: int):
     engine.start(world=WORLDS[world], seed=int(seed))
     frames, pending_steer, last_push = [], steer_idx is not None, 0.0
 
+    # The worker runs in a thread and reports failure by setting status.state rather
+    # than raising here, so poll it between frames: a long blocking get() would spend
+    # the whole GPU reservation waiting on a stream that is already dead.
+    ended = ""
+    last_frame_at = time.perf_counter()
     try:
         while len(frames) < total_frames:
-            if engine.status.state == "error":
-                raise gr.Error(f"engine error: {engine.status.error}")
             try:
-                jpg = engine.frames.get(timeout=60)
+                jpg = engine.frames.get(timeout=0.5)
             except queue.Empty:
-                # The worker sets state to idle and puts the reason in `detail` when
-                # a stream ends on its own (the 1024-latent-frame RoPE ceiling).
-                if engine.status.state != "streaming":
+                state = engine.status.state
+                if state == "error":
+                    raise gr.Error(f"engine error: {engine.status.error}")
+                if state != "streaming":
+                    # The worker ends the stream itself at the 1024-latent-frame
+                    # RoPE ceiling and puts the reason in `detail`.
+                    ended = engine.status.detail
                     break
-                raise gr.Error("the stream stalled waiting for a block")
+                if time.perf_counter() - last_frame_at > STALL_SECONDS:
+                    raise gr.Error(
+                        f"no block in {STALL_SECONDS} s (state={state})")
+                continue
 
+            last_frame_at = time.perf_counter()
             frames.append(cv2.imdecode(np.frombuffer(jpg, np.uint8),
                                        cv2.IMREAD_COLOR)[:, :, ::-1])
 
@@ -226,14 +238,17 @@ def run(world: str, steer_to: str, steer_at: float, seconds: float, seed: int):
                 last_push = now
                 yield frames[-1], hud(engine.stats(), now - t0), None
     finally:
+        stats = engine.stats()
+        print(f"[run] {len(frames)} frames, {stats['blocks']} blocks, "
+              f"state={engine.status.state!r} detail={engine.status.detail!r} "
+              f"error={engine.status.error!r}", flush=True)
         engine.stop()
 
     if not frames:
         raise gr.Error("no frames were produced")
-    note = engine.status.detail
-    line = hud(engine.stats(), time.perf_counter() - t0)
-    if note:
-        line += f"\n\n_{note}_"
+    line = hud(stats, time.perf_counter() - t0)
+    if ended:
+        line += f"\n\n_{ended}_"
     yield frames[-1], line, write_mp4(frames)