Offer the finished recording as a download

Each link's recording now tees into a footerless FileSink beside the
live binary stream (binary_stream first, then set_sinks with both — the
call replaces the sink set). publish concatenates the four part files
into one recording; footerless parts are what make that a single valid
RRD instead of one with duelling manifests. A hidden DownloadButton
appears with the merged file when the run finishes, and each new run
hides it again.

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

fdanyone/rerun_streaming.py

@@ -283,6 +283,37 @@ def check_stop(spec: RunSpec) -> None:
         raise RunCancelled("Stopped.")
 
 
+LINK_ORDER: tuple[str, ...] = ("begin", "prepare", "gpu", "publish")
+"""Chain links in stream order, and therefore the part-concatenation order."""
+
+
+def rrd_part(spec: RunSpec, name: str) -> Path:
+    """Where one link's footerless recording part lands on disk."""
+
+    part_dir: Path = spec.data_dir / "rrd"
+    part_dir.mkdir(parents=True, exist_ok=True)
+    return part_dir / f"{name}.rrd"
+
+
+def merge_rrd_parts(spec: RunSpec) -> Path:
+    """Concatenate the links' parts into the one recording a visitor downloads.
+
+    Parts are written with ``write_footer=False``: footered parts concatenate
+    into a file with multiple RRD manifests, which ``rrd verify`` rejects.
+    """
+
+    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)
+    return merged
+
+
 def request_stop(spec: RunSpec | None) -> str:
     """Write the run's stop sentinel; the pipeline unwinds at its next hook."""
 

fdanyone_app.py

@@ -30,10 +30,12 @@ from fdanyone.rerun_streaming import (
     Session,
     begin_phase,
     generate_phase,
+    merge_rrd_parts,
     motion_phase,
     new_spec,
     publish_phase,
     request_stop,
+    rrd_part,
     smoke_generate_phase,
     smoke_motion_phase,
     smoke_publish_phase,
@@ -55,8 +57,17 @@ class Link:
 
         return self.stream.read()
 
+    def close(self) -> None:
+        """Release this link's file sink before the next link (or fork) runs.
 
-def open_link(token: str) -> Link:
+        The part is already complete — a footerless part is whole at the last
+        flush — so this is descriptor hygiene, not a correctness dependency.
+        """
+
+        self.recording.disconnect()
+
+
+def open_link(token: str, part: Path) -> Link:
     """Open this callback's own recording, in whatever process the callback runs in.
 
     Called first thing in every link, GPU ones included, where "this callback"
@@ -75,12 +86,18 @@ def open_link(token: str) -> Link:
 
     rr.cleanup_if_forked_child()
     recording: rr.RecordingStream = rr.RecordingStream(APPLICATION_ID, recording_id=token)
-    return Link(recording=recording, stream=recording.binary_stream())
+    # Order is load-bearing: binary_stream() installs the stream sink, and
+    # set_sinks REPLACES the sink set — so the stream must be re-installed
+    # beside the file. Footerless parts are what make the final concatenation
+    # a single valid recording.
+    stream: rr.BinaryStream = recording.binary_stream()
+    recording.set_sinks(stream, rr.FileSink(str(part), write_footer=False))
+    return Link(recording=recording, stream=stream)
 
 
 def begin(
     video: str | None, start_time: float, seed: int
-) -> Iterator[tuple[RunSpec, bytes | None, str, Any]]:
+) -> Iterator[tuple[RunSpec, bytes | None, str, Any, Any]]:
     """Validate the input on CPU, open a recording, and switch to the outputs."""
 
     if video is None:
@@ -89,17 +106,29 @@ def begin(
         spec: RunSpec = new_spec(Path(video), float(start_time), int(seed))
     except FourDAnyoneError as exc:
         raise gr.Error(str(exc)) from None
-    link: Link = open_link(spec.token)
-    begin_phase(link.recording, spec)
-    yield spec, link.read(), "Preparing the source clip.", gr.Tabs(selected="outputs")
+    link: Link = open_link(spec.token, rrd_part(spec, "begin"))
+    try:
+        begin_phase(link.recording, spec)
+        yield (
+            spec,
+            link.read(),
+            "Preparing the source clip.",
+            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)
-    label: str = source_phase(link.recording, spec)
-    yield link.read(), label
+    link: Link = open_link(spec.token, rrd_part(spec, "prepare"))
+    try:
+        label: str = source_phase(link.recording, spec)
+        yield link.read(), label
+    finally:
+        link.close()
 
 
 @spaces.GPU(duration=GPU_DURATION)
@@ -115,7 +144,7 @@ def run_gpu(spec: RunSpec) -> Iterator[tuple[dict | None, bytes | None, str]]:
     is the only process that may build the recording it logs through.
     """
 
-    link: Link = open_link(spec.token)
+    link: Link = open_link(spec.token, rrd_part(spec, "gpu"))
     session: Session = Session(spec)
     try:
         motion: Iterator[str] = (
@@ -132,6 +161,8 @@ def run_gpu(spec: RunSpec) -> Iterator[tuple[dict | None, bytes | None, str]]:
         )
         for label in generate:
             yield None, link.read(), label
+        # Read before close: disconnect drops the binary sink with the file.
+        closing: bytes | None = link.read()
     except FourDAnyoneError as exc:
         # The banner keeps the reason on screen after the toast dismisses.
         yield None, link.read(), f"Failed: {exc}"
@@ -140,26 +171,32 @@ def run_gpu(spec: RunSpec) -> Iterator[tuple[dict | None, bytes | None, str]]:
         message: str = f"{type(exc).__name__}: {exc}"
         yield None, link.read(), f"Failed: {message}"
         raise gr.Error(message) from exc
-    yield session.summary, link.read(), "Publishing the result."
+    finally:
+        link.close()
+    yield session.summary, closing, "Publishing the result."
 
 
-def publish_cpu(spec: RunSpec, summary: dict | None) -> Iterator[tuple[bytes | None, str]]:
+def publish_cpu(spec: RunSpec, summary: dict | None) -> Iterator[tuple[bytes | None, str, Any]]:
     """Attach the finished rig and its six videos, off the GPU allocation."""
 
     if summary is None:
         raise gr.Error("The generation phase did not publish a result.")
-    link: Link = open_link(spec.token)
+    link: Link = open_link(spec.token, rrd_part(spec, "publish"))
     try:
         label: str = (
             smoke_publish_phase(link.recording, spec)
             if STREAM_SMOKE
             else publish_phase(link.recording, spec, summary)
         )
+        closing: bytes | None = link.read()
     except Exception as exc:
         message: str = f"{type(exc).__name__}: {exc}"
-        yield link.read(), f"Failed: {message}"
+        yield link.read(), f"Failed: {message}", gr.skip()
         raise gr.Error(message) from exc
-    yield link.read(), label
+    finally:
+        link.close()
+    merged: Path = merge_rrd_parts(spec)
+    yield closing, label, gr.DownloadButton(value=str(merged), visible=True)
 
 
 DESCRIPTION: str = """
@@ -255,6 +292,9 @@ def build_demo() -> gr.Blocks:
                 status: gr.Markdown = gr.Markdown(
                     "Upload a clip, or pick the example, then press Run.", elem_id="run-status"
                 )
+                download: gr.DownloadButton = gr.DownloadButton(
+                    "Download the recording (.rrd)", visible=False
+                )
             with gr.Column(scale=3):
                 viewer: Rerun = Rerun(
                     streaming=True,
@@ -262,10 +302,12 @@ def build_demo() -> gr.Blocks:
                     panel_states={"time": "collapsed", "blueprint": "hidden", "selection": "hidden"},
                 )
 
-        started = run_button.click(begin, [video, start_time, seed], [spec, viewer, status, tabs])
+        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])
-        published = generated.success(publish_cpu, [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.

tests/test_app_helpers.py

@@ -256,7 +256,7 @@ def test_a_recording_made_before_a_fork_cannot_be_flushed_after_it() -> None:
     )
 
 
-def test_open_link_gives_a_forked_child_a_recording_it_can_flush() -> None:
+def test_open_link_gives_a_forked_child_a_recording_it_can_flush(tmp_path: Path) -> None:
     """The fix: each link builds its own stream, in whatever process it runs in.
 
     The token is the ``recording_id``, so the parent's rows and the child's are
@@ -264,25 +264,28 @@ def test_open_link_gives_a_forked_child_a_recording_it_can_flush() -> None:
     """
 
     token: str = uuid.uuid4().hex
-    parent: fdanyone_app.Link = fdanyone_app.open_link(token)
+    parent: fdanyone_app.Link = fdanyone_app.open_link(token, tmp_path / "parent.rrd")
     parent.recording.log("log", rr.TextLog("parent"))
     assert parent.read()
+    parent.close()
 
     def body() -> int:
-        child: fdanyone_app.Link = fdanyone_app.open_link(token)
+        child: fdanyone_app.Link = fdanyone_app.open_link(token, tmp_path / "child.rrd")
         assert child.recording.get_recording_id() == token
         child.recording.log("log", rr.TextLog("child"))
         payload: bytes | None = child.read()
         return OK if payload else WRONG
 
     assert _in_child(body) == OK
+    # The dual sink teed the parent's rows into its part file as well.
+    assert (tmp_path / "parent.rrd").stat().st_size > 0
 
 
 RRD_MAGIC: bytes = b"RRF2"
 """First four bytes of every RRD document the SDK emits."""
 
 
-def test_a_stream_per_link_adds_no_framing_the_viewer_did_not_already_get() -> None:
+def test_a_stream_per_link_adds_no_framing_the_viewer_did_not_already_get(tmp_path: Path) -> None:
     """Why splitting one stream into four costs the browser viewer nothing.
 
     Each ``read`` already returns a whole RRD document, magic bytes and manifest
@@ -294,7 +297,7 @@ def test_a_stream_per_link_adds_no_framing_the_viewer_did_not_already_get() -> N
     where that shows up rather than in a blank viewer on the Space.
     """
 
-    link: fdanyone_app.Link = fdanyone_app.open_link(uuid.uuid4().hex)
+    link: fdanyone_app.Link = fdanyone_app.open_link(uuid.uuid4().hex, tmp_path / "part.rrd")
     for row in range(3):
         link.recording.log("log", rr.TextLog(f"row {row}"))
         payload: bytes | None = link.read()