Unify serving paths

flashdreams/flashdreams/infra/postprocess/__init__.py

@@ -37,12 +37,14 @@
     VideoPostprocessStepStats,
     VideoPostprocessStream,
     create_runner_postprocess_stream,
+    create_video_postprocess_stream,
 )
 
 __all__ = [
     "VideoPostprocessStream",
     "VideoPostprocessStepStats",
     "create_runner_postprocess_stream",
+    "create_video_postprocess_stream",
     "VideoChunk",
     "VideoPostProcessor",
     "VideoPostProcessorConfig",

flashdreams/flashdreams/infra/postprocess/stream.py

@@ -212,15 +212,17 @@ def _prepare(self, output: Tensor) -> None:
         self._prepared = True
 
 
-def create_runner_postprocess_stream(
-    config: RunnerConfigT,
+def create_video_postprocess_stream(
     *,
+    postprocess: VideoPostprocessChainConfig,
+    output_layout: VideoTensorLayout,
+    fps: float | None,
+    per_view: bool,
     world_size: int,
     is_rank_zero: bool = True,
-    fps: float | None = None,
+    profile: bool = False,
 ) -> VideoPostprocessStream | None:
-    """Create a runner post-processing stream, or ``None`` when skipped."""
-    postprocess = getattr(config, "postprocess")
+    """Create a post-processing stream for one generated video rollout."""
     if not postprocess.is_enabled():
         return None
     postprocess.validate_execution(world_size=world_size)
@@ -230,7 +232,27 @@ def create_runner_postprocess_stream(
         and not postprocess.requires_all_ranks(world_size=world_size)
     ):
         return None
+    return VideoPostprocessStream(
+        postprocess=postprocess,
+        output_layout=output_layout,
+        fps=fps,
+        per_view=per_view,
+        world_size=world_size,
+        profile=profile,
+    )
 
+
+def create_runner_postprocess_stream(
+    config: RunnerConfigT,
+    *,
+    world_size: int,
+    is_rank_zero: bool = True,
+    fps: float | None = None,
+) -> VideoPostprocessStream | None:
+    """Create a runner post-processing stream, or ``None`` when skipped."""
+    postprocess = getattr(config, "postprocess")
+    if not postprocess.is_enabled():
+        return None
     output_layout = getattr(config, "postprocess_output_layout")
     if output_layout is None:
         raise ValueError(
@@ -242,12 +264,13 @@ def create_runner_postprocess_stream(
     if configured_fps is None:
         configured_fps = getattr(config, "fps", getattr(config, "output_fps", None))
 
-    return VideoPostprocessStream(
+    return create_video_postprocess_stream(
         postprocess=postprocess,
         output_layout=output_layout,
         fps=configured_fps,
         per_view=getattr(config, "postprocess_per_view"),
         world_size=world_size,
+        is_rank_zero=is_rank_zero,
         profile=bool(
             getattr(getattr(config, "pipeline", None), "enable_sync_and_profile", False)
         ),

flashdreams/flashdreams/infra/results.py

@@ -0,0 +1,136 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Generated inference result contracts shared by runtimes and consumers."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Any
+
+from torch import Tensor
+
+from flashdreams.infra.postprocess import VideoTensorLayout
+from flashdreams.infra.time import TimeWindow
+
+if TYPE_CHECKING:
+    from flashdreams.infra.video_output import LazyRGBFrame
+
+
+@dataclass(frozen=True, kw_only=True, slots=True)
+class StepResult:
+    """Generated output and metadata returned by one inference step.
+
+    Video results use :meth:`from_video_chunk`, which records a required tensor
+    layout and derives the frame count once. Non-video results may use the
+    regular constructor without a layout.
+    """
+
+    __hash__ = None
+
+    step_index: int
+    output: Any = None
+    frame_count: int = 0
+    layout: VideoTensorLayout | None = None
+    output_window: TimeWindow | None = None
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+    metrics: Mapping[str, float | int] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        if self.step_index < 0:
+            raise ValueError("StepResult.step_index must be >= 0.")
+        if self.frame_count < 0:
+            raise ValueError("StepResult.frame_count must be >= 0.")
+        if self.layout is not None:
+            from flashdreams.infra.video_output import infer_video_num_frames
+
+            video_chunk = self.video_chunk
+            derived_frame_count = infer_video_num_frames(
+                video_chunk,
+                layout=self.layout,
+            )
+            if self.frame_count not in (0, derived_frame_count):
+                raise ValueError(
+                    "StepResult.frame_count does not match the declared video "
+                    f"layout: expected {derived_frame_count}, got {self.frame_count}."
+                )
+            object.__setattr__(self, "frame_count", derived_frame_count)
+        object.__setattr__(self, "metadata", MappingProxyType(dict(self.metadata)))
+        object.__setattr__(self, "metrics", MappingProxyType(dict(self.metrics)))
+
+    @classmethod
+    def from_video_chunk(
+        cls,
+        *,
+        step_index: int,
+        video_chunk: Tensor,
+        layout: VideoTensorLayout,
+        output_window: TimeWindow | None = None,
+        metadata: Mapping[str, Any] | None = None,
+        metrics: Mapping[str, float | int] | None = None,
+    ) -> StepResult:
+        """Build one layout-aware generated-video result."""
+        return cls(
+            step_index=step_index,
+            output=video_chunk,
+            layout=layout,
+            output_window=output_window,
+            metadata=dict(metadata or {}),
+            metrics=dict(metrics or {}),
+        )
+
+    @property
+    def video_chunk(self) -> Tensor:
+        """Return the video tensor or fail if this is not a video result."""
+        if self.layout is None:
+            raise ValueError("StepResult.layout is required for video output.")
+        if not isinstance(self.output, Tensor):
+            raise TypeError(
+                "A video StepResult requires a torch.Tensor output, "
+                f"got {type(self.output).__name__}."
+            )
+        return self.output
+
+    def lazy_rgb_frames(
+        self,
+        *,
+        batch_index: int = 0,
+        view_index: int = 0,
+        record_cuda_event: bool = True,
+    ) -> list[LazyRGBFrame]:
+        """Expose this video result as lazy per-frame RGB handles."""
+        from flashdreams.infra.video_output import lazy_rgb_frames_from_video_tensor
+
+        return lazy_rgb_frames_from_video_tensor(
+            self.video_chunk,
+            layout=self._video_layout(),
+            batch_index=batch_index,
+            view_index=view_index,
+            record_cuda_event=record_cuda_event,
+        )
+
+    def video_hwc_uint8(
+        self,
+        *,
+        batch_index: int = 0,
+        view_index: int = 0,
+    ) -> Tensor:
+        """Return this video result as uint8 ``[T,H,W,C]`` on its device."""
+        from flashdreams.infra.video_output import video_tensor_to_hwc_uint8
+
+        return video_tensor_to_hwc_uint8(
+            self.video_chunk,
+            layout=self._video_layout(),
+            batch_index=batch_index,
+            view_index=view_index,
+        )
+
+    def _video_layout(self) -> VideoTensorLayout:
+        if self.layout is None:
+            raise ValueError("StepResult.layout is required for video output.")
+        return self.layout
+
+
+__all__ = ["StepResult"]

flashdreams/flashdreams/infra/runner.py

@@ -63,6 +63,9 @@ class RunnerConfig(InstantiateConfig):
     per-runner ``--help`` (it's metadata, not a knob); a non-empty
     value is enforced for in-tree runners by the registry test."""
 
+    output_adapter: Annotated[str | None, tyro.conf.Suppress] = None
+    """Optional ``module:attribute`` implementing non-CLI output capabilities."""
+
     pipeline: StreamInferencePipelineConfig
     """Wrapped pipeline config; the runner instantiates and drives it."""
 
@@ -183,9 +186,8 @@ def create_video_output_stream(
         self,
         *,
         fps: float | None = None,
-        move_to_cpu: bool = True,
     ) -> VideoOutputStream:
-        """Create the standard runner video output stream for one rollout."""
+        """Create the standard post-processing stream for one rollout."""
         layout = self.config.postprocess_output_layout
         if layout is None:
             raise ValueError(
@@ -194,8 +196,6 @@ def create_video_output_stream(
         return VideoOutputStream(
             postprocess_stream=self.create_postprocess_stream(fps=fps),
             output_layout=layout,
-            collect_output=self.is_rank_zero,
-            move_to_cpu=move_to_cpu,
         )
 
     @abstractmethod

flashdreams/flashdreams/infra/time.py

@@ -0,0 +1,32 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared time-domain value objects."""
+
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, kw_only=True, slots=True)
+class TimeWindow:
+    """Half-open time window in seconds since session start."""
+
+    start_s: float
+    end_s: float
+
+    def __post_init__(self) -> None:
+        if not math.isfinite(self.start_s) or not math.isfinite(self.end_s):
+            raise ValueError("TimeWindow bounds must be finite seconds.")
+        if self.start_s < 0 or self.end_s < 0:
+            raise ValueError("TimeWindow bounds must be non-negative.")
+        if self.end_s < self.start_s:
+            raise ValueError("TimeWindow.end_s must be >= start_s.")
+
+    def contains(self, timestamp_s: float) -> bool:
+        """Return whether ``timestamp_s`` falls within this half-open window."""
+        return self.start_s <= timestamp_s < self.end_s
+
+
+__all__ = ["TimeWindow"]