Apply the thermo-review survivors (app side)

The chain is three links — begin absorbed the accidental prepare_cpu
boundary (same process, one recording, one part fewer). The generation
result crosses the fork as a typed RunResult. merge_rrd_parts requires
every part non-empty and lands atomically via os.replace. pump owns its
worker's lifetime: an early consumer close requests the stop sentinel
and joins with a bounded window, and run_gpu closes its phase generators
before disconnecting the recording. Vendored: space-streaming@c8893d6.
Validated end-to-end locally: 479 s, all 19 stages once, merged rrd from
exactly three parts.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

PROVENANCE.md

@@ -6,8 +6,8 @@
 | --- | --- |
 | Source repository | <https://github.com/pablovela5620/4DAnyone-5090> |
 | Branch | `space-streaming` |
-| Commit | `d2ac77a55accd22dac2323207e5fac40ee7661e0` |
-| Synced | 2026-08-29T04:45:52Z |
+| Commit | `c8893d657278fe6bd0b3447a87a6cf29f4b4ded4` |
+| Synced | 2026-08-29T05:06:46Z |
 
 ## Excluded from the copy
 

fdanyone/model/loader.py

@@ -403,6 +403,11 @@ def load_pipeline(
         key = _cache_key(checkpoint_path, assets, device, settings)
         if _PIPELINE_CACHE is not None and _PIPELINE_CACHE[0] == key:
             return _PIPELINE_CACHE[1]
+        # Evict BEFORE building: holding the old stack while the new one loads
+        # would put two ~11 GB stacks in memory at once — the exact situation
+        # the single slot exists to prevent.
+        _PIPELINE_CACHE = None
+        gc.collect()
         loaded = _load_pipeline_uncached(
             checkpoint_path=checkpoint_path,
             assets=assets,

fdanyone/model/turbo_lora.py

@@ -124,12 +124,12 @@ def merge_wan_turbo_lora(
                     up = adapter.get_tensor(up_key).to(device=compute_device, dtype=torch.float32)
                     delta = torch.mm(up, down)
                     parameter.add_(delta.to(device=parameter.device, dtype=parameter.dtype))
+                for key, target_name in direct_targets.items():
+                    parameter = parameters[target_name]
+                    delta = adapter.get_tensor(key)
+                    parameter.add_(delta.to(device=parameter.device, dtype=parameter.dtype))
         finally:
             torch.backends.cuda.matmul.allow_tf32 = tf32
-            for key, target_name in direct_targets.items():
-                parameter = parameters[target_name]
-                delta = adapter.get_tensor(key)
-                parameter.add_(delta.to(device=parameter.device, dtype=parameter.dtype))
 
     return TurboLoraReport(
         lora_modules=len(down_keys),

fdanyone/pipeline.py

@@ -238,7 +238,7 @@ def _build_conditioning(
             stderr=subprocess.STDOUT,
         )
     completed: bool = False
-    failure: FourDAnyoneError | None = None
+    failure: Exception | None = None
 
     def _validate_target_render() -> None:
         return_code: int = render_process.wait()
@@ -278,7 +278,9 @@ def _build_conditioning(
 
     def wait_for_targets() -> None:
         # Re-entry (the pipeline-level cleanup) must repeat the original
-        # verdict, not mask an informative failure with a generic one.
+        # verdict, not mask an informative failure with a generic one — and
+        # that holds for ANY failure, not just the domain error: a JSON or OS
+        # error forgotten here would turn into a silent pass on re-entry.
         nonlocal completed, failure
         if completed:
             if failure is not None:
@@ -287,7 +289,7 @@ def _build_conditioning(
         completed = True
         try:
             _validate_target_render()
-        except FourDAnyoneError as exc:
+        except Exception as exc:
             failure = exc
             raise
 
@@ -465,9 +467,15 @@ def prepare_run(
         # Re-decode the worker-produced source before it becomes a model tensor.
         verify_lossless_video(clip, conditioning.source_video)
     except BaseException:
-        if conditioning is not None and conditioning.target_waiter is not None:
-            conditioning.wait_for_target_skeletons()
-        _discard_scratch(scratch)
+        # The waiter can itself raise (its renderer failed too); the scratch
+        # must still be discarded, and the original failure must propagate.
+        try:
+            if conditioning is not None and conditioning.target_waiter is not None:
+                conditioning.wait_for_target_skeletons()
+        except Exception:
+            LOGGER.exception("Deferred target rendering also failed during unwinding")
+        finally:
+            _discard_scratch(scratch)
         raise
     return PreparedRun(
         settings=settings,
@@ -486,6 +494,25 @@ def prepare_run(
     )
 
 
+@dataclass(frozen=True)
+class RunResult:
+    """What one finished run publishes across the process boundary.
+
+    Frozen and picklable: on ZeroGPU it travels from the forked worker back
+    through a Gradio ``State``. ``metadata`` is the exported summary document,
+    kept for CLI and metadata consumers; the typed fields are the contract.
+    """
+
+    result_dir: Path
+    """Published result directory: six views, cameras, metadata."""
+    motion_dir: Path
+    """Published GVHMR motion directory."""
+    total_elapsed_seconds: float
+    """Wall time from pipeline start to the exported result."""
+    metadata: dict
+    """The full exported summary, exactly as written to disk."""
+
+
 def warm_generation_models(
     *,
     model_dir: str,
@@ -526,7 +553,7 @@ def generate_run(
     on_denoise_step: DenoiseStepHook | None = None,
     on_denoise_step_start: DenoiseStepStartHook | None = None,
     on_generate_stage: GenerateStageHook | None = None,
-) -> dict:
+) -> RunResult:
     """Generate and publish every view, consuming the prepared run.
 
     The prompt embedding is fixed by ``prepare_run``, because whether it exists
@@ -568,9 +595,24 @@ def generate_run(
             )
         summary["result_dir"] = str(prepared.result_dir)
         summary["motion_dir"] = str(prepared.motion_dir)
-        return summary
-    finally:
-        release_run(prepared)
+        result = RunResult(
+            result_dir=prepared.result_dir,
+            motion_dir=prepared.motion_dir,
+            total_elapsed_seconds=float(summary["total_pipeline_elapsed_seconds"]),
+            metadata=summary,
+        )
+    except BaseException:
+        # Release failures during unwinding must not replace the failure that
+        # actually explains the run; they are logged instead.
+        try:
+            release_run(prepared)
+        except Exception:
+            LOGGER.exception("Run release failed while unwinding a generation failure")
+        raise
+    # A release failure on the success path stays loud: the renderer's own
+    # verdict is part of the run's validity.
+    release_run(prepared)
+    return result
 
 
 def release_run(prepared: PreparedRun) -> None:
@@ -608,7 +650,7 @@ def run_pipeline(
     on_denoise_step: DenoiseStepHook | None = None,
     on_denoise_step_start: DenoiseStepStartHook | None = None,
     prompt_embedding_path: Path | None = None,
-) -> dict:
+) -> RunResult:
     """Execute inference and publish reusable GVHMR plus 4DAnyone results."""
 
     if seed < 0:

fdanyone/rerun_streaming.py

@@ -33,7 +33,7 @@ from fdanyone.errors import FourDAnyoneError
 from fdanyone.model.inference import _tensor_frames
 from fdanyone.model.tiny_decoder import decode_tiny_target_video, load_tiny_wan_decoder
 from fdanyone.motion.body import BodyMotion, load_body_motion
-from fdanyone.pipeline import PreparedRun, generate_run, prepare_run, warm_generation_models
+from fdanyone.pipeline import PreparedRun, RunResult, generate_run, prepare_run, warm_generation_models
 from fdanyone.video import choose_canonical_fps
 from fdanyone.viz import (
     BOX_COLOR,
@@ -110,8 +110,10 @@ STREAM_SMOKE: bool = os.environ.get("FDANYONE_STREAM_SMOKE") == "1"
 SMOKE_DELAY: float = float(os.environ.get("FDANYONE_SMOKE_DELAY", "0.1"))
 """Seconds between smoke yields; raise it to eyeball each phase in a browser."""
 
-SMOKE_SUMMARY: dict = {"result_dir": "", "total_pipeline_elapsed_seconds": 0.0}
-"""Stand-in for the pipeline's published metadata; the smoke run writes no files."""
+SMOKE_SUMMARY: RunResult = RunResult(
+    result_dir=Path(""), motion_dir=Path(""), total_elapsed_seconds=0.0, metadata={}
+)
+"""Stand-in for the pipeline's published result; the smoke run writes no files."""
 
 # Body evaluation resolves SMPL-X relative to the vendored package's repository
 # root, which is not where a Space keeps its models. An absolute root wins the
@@ -281,8 +283,8 @@ class Session:
     """The run this link was rebuilt from."""
     prepared: PreparedRun | None = None
     """Result of the motion phase, consumed by the generation phase."""
-    summary: dict | None = None
-    """Published metadata the generation phase returns, set once it succeeds."""
+    summary: RunResult | None = None
+    """Published result the generation phase returns, set once it succeeds."""
     source_frames: dict[int, RgbFrame] = field(default_factory=dict)
     """Decoded source stills for the diffusion pane, keyed by canonical frame."""
     events: queue.Queue[str | None] = field(default_factory=queue.Queue)
@@ -311,7 +313,7 @@ def check_stop(spec: RunSpec) -> None:
         raise RunCancelled("Stopped.")
 
 
-LINK_ORDER: tuple[str, ...] = ("begin", "prepare", "gpu", "publish")
+LINK_ORDER: tuple[str, ...] = ("begin", "gpu", "publish")
 """Chain links in stream order, and therefore the part-concatenation order."""
 
 
@@ -328,17 +330,24 @@ def merge_rrd_parts(spec: RunSpec) -> Path:
 
     Parts are written with ``write_footer=False``: footered parts concatenate
     into a file with multiple RRD manifests, which ``rrd verify`` rejects.
+    Every link's part must exist and be non-empty — silently skipping one
+    would offer a plausible-looking partial recording — and the merge lands
+    atomically so a failure never leaves a truncated file at the final path.
     """
 
     import shutil
 
     merged: Path = spec.data_dir / f"4danyone-{spec.token[:8]}.rrd"
-    with merged.open("wb") as out:
-        for name in LINK_ORDER:
-            part: Path = spec.data_dir / "rrd" / f"{name}.rrd"
-            if part.is_file():
-                with part.open("rb") as source:
-                    shutil.copyfileobj(source, out, length=1 << 20)
+    parts: list[Path] = [spec.data_dir / "rrd" / f"{name}.rrd" for name in LINK_ORDER]
+    missing: list[str] = [part.name for part in parts if not part.is_file() or part.stat().st_size == 0]
+    if missing:
+        raise FourDAnyoneError(f"The recording is incomplete; missing parts: {missing}.")
+    staging: Path = merged.with_name(merged.name + ".tmp")
+    with staging.open("wb") as out:
+        for part in parts:
+            with part.open("rb") as source:
+                shutil.copyfileobj(source, out, length=1 << 20)
+    os.replace(staging, merged)
     return merged
 
 
@@ -376,12 +385,24 @@ def pump(session: Session, work: Callable[[], T]) -> Iterator[str]:
 
     worker: threading.Thread = threading.Thread(target=target, daemon=True)
     worker.start()
-    while True:
-        label: str | None = session.events.get()
-        if label is None:
-            break
-        yield label
-    worker.join()
+    try:
+        while True:
+            label: str | None = session.events.get()
+            if label is None:
+                break
+            yield label
+        worker.join()
+    finally:
+        if worker.is_alive():
+            # The consumer vanished mid-run (client disconnect, generator
+            # close). The worker only halts at its next sentinel check, so
+            # request the stop and give it a bounded window; a daemon thread
+            # cannot block interpreter exit, but an unbounded join here would
+            # block the closing request thread for a whole denoise step.
+            request_stop(session.spec)
+            worker.join(timeout=5.0)
+            if worker.is_alive():
+                LOGGER.warning("Pipeline worker still unwinding after its consumer closed.")
     if failure:
         raise failure[0]
     return outcome[0]
@@ -699,7 +720,7 @@ def generate_phase(recording: rr.RecordingStream, session: Session) -> Iterator[
     log_status(recording, "generate: six views published")
 
 
-def publish_phase(recording: rr.RecordingStream, spec: RunSpec, summary: dict) -> str:
+def publish_phase(recording: rr.RecordingStream, spec: RunSpec, summary: RunResult) -> str:
     """Place the finished rig and its six videos, then switch to the final layout.
 
     Everything this reads is a file on disk, so it runs outside the GPU
@@ -707,10 +728,10 @@ def publish_phase(recording: rr.RecordingStream, spec: RunSpec, summary: dict) -
     """
 
     recording.reset_time()
-    log_result(recording, Path(summary["result_dir"]))
+    log_result(recording, summary.result_dir)
     log_status(recording, "done")
     recording.send_blueprint(result_blueprint(spec.fps), make_active=True)
-    elapsed: float = float(summary["total_pipeline_elapsed_seconds"])
+    elapsed: float = summary.total_elapsed_seconds
     return f"Done in {elapsed:.1f}s. Six views at {float(spec.fps):.3f} FPS."
 
 

fdanyone_app.py

@@ -1,6 +1,7 @@
 """Gradio and ZeroGPU wiring for the 4DAnyone Rerun streaming runtime.
 
-The four-link event chain is ``begin -> prepare_cpu -> run_gpu -> publish_cpu``.
+The three-link event chain is ``begin -> run_gpu -> publish_cpu``, one link
+per process boundary.
 ZeroGPU runs decorated callbacks in forked children, so each link creates its
 own recording after ``rr.cleanup_if_forked_child()`` and uses the shared token
 as its recording ID. The merged ``run_gpu`` remains one allocation because its
@@ -26,6 +27,7 @@ from fdanyone.rerun_streaming import (
     EXAMPLE_DIR,
     GPU_DURATION,
     STREAM_SMOKE,
+    RunResult,
     RunSpec,
     Session,
     begin_phase,
@@ -75,7 +77,7 @@ def open_link(token: str, part: Path) -> Link:
     means the forked child. A stream inherited across the fork cannot be used:
     the SDK compares pids on flush and raises "Fork detected during flush". A
     stream built here belongs to this process, and sharing ``token`` as the
-    recording id is what makes the viewer treat all four links as one recording.
+    recording id is what makes the viewer treat all the links as one recording.
 
     ``cleanup_if_forked_child`` drops whatever streams were inherited. The SDK
     registers it with ``os.register_at_fork`` already, and ``rr.init`` calls it
@@ -98,8 +100,14 @@ def open_link(token: str, part: Path) -> Link:
 
 def begin(
     video: str | None, start_time: float, seed: int
-) -> Iterator[tuple[RunSpec, bytes | None, str, Any, Any]]:
-    """Validate the input on CPU, open a recording, and switch to the outputs."""
+) -> Iterator[tuple[Any, bytes | None, str, Any, Any]]:
+    """Validate, open a recording, switch to the outputs, and log the source.
+
+    One CPU link: the earlier split between "begin" and "prepare" callbacks
+    was an accidental process boundary — both ran on the CPU in this process,
+    each with its own recording lifecycle and part file. The first yield lands
+    the tab switch immediately; the second follows with the source clip.
+    """
 
     if video is None:
         raise gr.Error("Upload a video, or pick the bundled example.")
@@ -117,23 +125,14 @@ def begin(
             gr.Tabs(selected="outputs"),
             gr.DownloadButton(visible=False),
         )
-    finally:
-        link.close()
-
-
-def prepare_cpu(spec: RunSpec) -> Iterator[tuple[bytes | None, str]]:
-    """Put the source clip on the frame timeline before any GPU work starts."""
-
-    link: Link = open_link(spec.token, rrd_part(spec, "prepare"))
-    try:
         label: str = source_phase(link.recording, spec)
-        yield link.read(), format_status(spec, label)
+        yield gr.skip(), link.read(), format_status(spec, label), gr.skip(), gr.skip()
     finally:
         link.close()
 
 
 @spaces.GPU(duration=GPU_DURATION)
-def run_gpu(spec: RunSpec) -> Iterator[tuple[dict | None, bytes | None, str]]:
+def run_gpu(spec: RunSpec) -> Iterator[tuple[RunResult | None, bytes | None, str]]:
     """Recover the motion and generate the six views, inside one allocation.
 
     The two phases have to share a process. ``PreparedRun`` carries the decoded
@@ -147,15 +146,17 @@ def run_gpu(spec: RunSpec) -> Iterator[tuple[dict | None, bytes | None, str]]:
 
     link: Link = open_link(spec.token, rrd_part(spec, "gpu"))
     session: Session = Session(spec)
+    motion: Iterator[str] | None = None
+    generate: Iterator[str] | None = None
     try:
-        motion: Iterator[str] = (
+        motion = (
             smoke_motion_phase(link.recording, session)
             if STREAM_SMOKE
             else motion_phase(link.recording, session)
         )
         for label in motion:
             yield None, link.read(), format_status(spec, label)
-        generate: Iterator[str] = (
+        generate = (
             smoke_generate_phase(link.recording, session)
             if STREAM_SMOKE
             else generate_phase(link.recording, session)
@@ -173,11 +174,16 @@ def run_gpu(spec: RunSpec) -> Iterator[tuple[dict | None, bytes | None, str]]:
         yield None, link.read(), f"Failed: {message}"
         raise gr.Error(message) from exc
     finally:
+        # Close the phase generators FIRST: their unwinding (pump's bounded
+        # worker join) must finish while the recording is still connected.
+        for phase in (motion, generate):
+            if phase is not None:
+                phase.close()
         link.close()
     yield session.summary, closing, format_status(spec, "Publishing the result.")
 
 
-def publish_cpu(spec: RunSpec, summary: dict | None) -> Iterator[tuple[bytes | None, str, Any]]:
+def publish_cpu(spec: RunSpec, summary: RunResult | None) -> Iterator[tuple[bytes | None, str, Any]]:
     """Attach the finished rig and its six videos, off the GPU allocation."""
 
     if summary is None:
@@ -306,14 +312,13 @@ def build_demo() -> gr.Blocks:
         started = run_button.click(
             begin, [video, start_time, seed], [spec, viewer, status, tabs, download]
         )
-        prepared = started.success(prepare_cpu, spec, [viewer, status])
-        generated = prepared.success(run_gpu, spec, [summary, viewer, status])
+        generated = started.success(run_gpu, spec, [summary, viewer, status])
         published = generated.success(publish_cpu, [spec, summary], [viewer, status, download])
         # `cancels` detaches the UI at once, but the pipeline itself only halts
         # when it reads the sentinel `request_stop` writes: the worker thread
         # (and on ZeroGPU, the forked child) never sees a Gradio cancellation.
         stop_button.click(
-            request_stop, spec, status, cancels=[started, prepared, generated, published]
+            request_stop, spec, status, cancels=[started, generated, published]
         )
     return demo
 

tests/test_pipeline_lifecycle.py

@@ -62,11 +62,17 @@ def test_generate_run_settles_and_discards_on_success(
     conditioning: _Conditioning = _Conditioning()
     prepared: PreparedRun = _prepared(tmp_path, conditioning)
     monkeypatch.setattr(fdanyone.model.inference, "generate_views", lambda **_: object())
-    monkeypatch.setattr(fdanyone.output, "export_result", lambda **_: {"fps": Fraction(25, 1)})
+    monkeypatch.setattr(
+        fdanyone.output,
+        "export_result",
+        lambda **_: {"fps": Fraction(25, 1), "total_pipeline_elapsed_seconds": 1.5},
+    )
 
-    summary: dict = generate_run(prepared, seed=0)
+    result = generate_run(prepared, seed=0)
 
-    assert summary["result_dir"] == str(prepared.result_dir)
+    assert result.result_dir == prepared.result_dir
+    assert result.total_elapsed_seconds == 1.5
+    assert result.metadata["result_dir"] == str(prepared.result_dir)
     assert conditioning.settled == 1
     assert not prepared.scratch.exists()
 
@@ -94,7 +100,11 @@ def test_generate_run_discards_even_when_settlement_raises(
     conditioning: _Conditioning = _Conditioning(settlement_error=RuntimeError("render worker died"))
     prepared: PreparedRun = _prepared(tmp_path, conditioning)
     monkeypatch.setattr(fdanyone.model.inference, "generate_views", lambda **_: object())
-    monkeypatch.setattr(fdanyone.output, "export_result", lambda **_: {})
+    monkeypatch.setattr(
+        fdanyone.output,
+        "export_result",
+        lambda **_: {"total_pipeline_elapsed_seconds": 0.0},
+    )
 
     with pytest.raises(RuntimeError, match="render worker died"):
         generate_run(prepared, seed=0)