Add fake demo runtime vertical slice

flashdreams/flashdreams/runtime/demo/__init__.py

@@ -3,8 +3,43 @@
 
 """Experimental shared demo API above the inference runtime API."""
 
+from flashdreams.runtime.demo.drivers import (
+    BatchSessionDriver,
+    DriverInvariantError,
+    run_demo_session,
+)
+from flashdreams.runtime.demo.host import RuntimeHost
 from flashdreams.runtime.demo.outputs import build_output_target
+from flashdreams.runtime.demo.outputs import (
+    NullOutputSink,
+    OutputDecision,
+    OutputSink,
+    SessionInfo,
+)
+from flashdreams.runtime.demo.pipeline import StepOutcome, StepPipeline
 from flashdreams.runtime.demo.replay import run_replay_demo
+from flashdreams.runtime.demo.run_modes import (
+    DefaultErrorPolicy,
+    ErrorAction,
+    InMemorySessionMetricsRecorder,
+    MetricsSnapshot,
+    NoopTransportService,
+    RunContext,
+    RunMode,
+    RunResult,
+    RunSummary,
+    SessionEdges,
+    SingleSessionAdmissionPolicy,
+)
+from flashdreams.runtime.demo.session_inputs import (
+    BatchInputSource,
+    ControlDecision,
+    InputSource,
+    ModelInputProvider,
+    PreparedStep,
+    RealtimeInputSource,
+    UserInputWindow,
+)
 from flashdreams.runtime.demo.spec import (
     DemoAdapter,
     DemoSpec,
@@ -17,14 +52,42 @@
 )
 
 __all__ = [
+    "BatchInputSource",
+    "BatchSessionDriver",
+    "ControlDecision",
+    "DefaultErrorPolicy",
     "DemoAdapter",
     "DemoSpec",
+    "DriverInvariantError",
+    "ErrorAction",
+    "InMemorySessionMetricsRecorder",
+    "InputSource",
+    "MetricsSnapshot",
+    "ModelInputProvider",
     "Mp4OutputSpec",
+    "NoopTransportService",
     "NullOutputSpec",
+    "NullOutputSink",
+    "OutputDecision",
     "OutputSpec",
+    "OutputSink",
     "PreparedScenario",
+    "PreparedStep",
+    "RealtimeInputSource",
+    "RunContext",
+    "RunMode",
+    "RunResult",
+    "RunSummary",
+    "RuntimeHost",
+    "SessionEdges",
+    "SessionInfo",
+    "SingleSessionAdmissionPolicy",
+    "StepOutcome",
+    "StepPipeline",
+    "UserInputWindow",
     "WebRTCAppResources",
     "WebRTCOutputSpec",
     "build_output_target",
+    "run_demo_session",
     "run_replay_demo",
 ]

flashdreams/flashdreams/runtime/demo/drivers.py

@@ -0,0 +1,222 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Session drivers and helpers for demo runtime vertical slices."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from flashdreams.runtime.interfaces import InferenceSession
+
+from .host import RuntimeHost
+from .outputs import SessionInfo
+from .pipeline import StepPipeline
+from .run_modes import (
+    DriverStatus,
+    RunContext,
+    RunMode,
+    RunResult,
+    SessionEdges,
+    SessionReservation,
+)
+from .session_inputs import ModelInputProvider
+from .spec import DemoAdapter, DemoSpec, PreparedScenario
+
+
+class DriverInvariantError(RuntimeError):
+    """A driver invariant was violated; this is a driver bug, not a run result."""
+
+
+class BatchSessionDriver:
+    """Minimal finite-session driver for Phase 2 fake-model coverage."""
+
+    def run_one_session(
+        self,
+        *,
+        host: RuntimeHost,
+        provider: ModelInputProvider,
+        session_edges: SessionEdges,
+        pipeline: StepPipeline,
+    ) -> RunResult:
+        session: InferenceSession | None = None
+        final_status: DriverStatus = "completed"
+        final_reason: str | None = None
+        final_error: Exception | None = None
+        setup_ok = False
+        try:
+            try:
+                initial_input = host.call(provider.prepare_initial_input)
+                session = host.call(host.start_session, initial_input)
+                session_info = host.call(_session_info, session)
+                session_edges.output_sink.open(session_info)
+                setup_ok = True
+            except Exception as exc:
+                action = session_edges.error_policy.handle_setup_error(exc)
+                if action.drop_chunk or action.result_status == "completed":
+                    raise DriverInvariantError(
+                        "Setup failures must resolve to failed or skipped."
+                    ) from exc
+                session_edges.metrics.record_error(exc, action)
+                final_status = action.result_status
+                final_reason = str(exc)
+                final_error = exc if action.result_status == "failed" else None
+
+            while setup_ok:
+                if session is None:
+                    raise DriverInvariantError("setup_ok was set without a session.")
+                try:
+                    if session_edges.input_source.is_finished():
+                        break
+                    request = host.call(session.next_step_request)
+                    if request is None:
+                        break
+                    user_window = session_edges.input_source.next_window(request)
+                    outcome = host.call(
+                        pipeline.execute_step,
+                        request=request,
+                        user_window=user_window,
+                        provider=provider,
+                        session=session,
+                        output=session_edges.output_sink,
+                        metrics=session_edges.metrics,
+                    )
+                    if outcome.control.reset:
+                        host.call(session.reset, outcome.control.reset_input)
+                        if not outcome.control.provider_already_reset:
+                            host.call(provider.reset, outcome.control.reset_input)
+                        continue
+                    if outcome.control.close_session:
+                        break
+                    if outcome.output.should_stop:
+                        break
+                except DriverInvariantError:
+                    raise
+                except Exception as exc:
+                    action = session_edges.error_policy.handle(exc)
+                    session_edges.metrics.record_error(exc, action)
+                    if action.drop_chunk:
+                        continue
+                    final_status = action.result_status
+                    final_reason = str(exc)
+                    final_error = exc if action.result_status == "failed" else None
+                    break
+        except DriverInvariantError:
+            raise
+        except Exception as exc:
+            final_status = "failed"
+            final_reason = str(exc)
+            final_error = exc
+        finally:
+            if session is not None:
+                host.call(_close_safely, session.close, session_edges)
+            host.call(_close_safely, provider.close, session_edges)
+
+        return session_edges.close_result(
+            status=final_status,
+            reason=final_reason,
+            error=final_error,
+        )
+
+
+def run_demo_session(
+    *,
+    context: RunContext,
+    spec: DemoSpec,
+    scenario: PreparedScenario,
+    adapter: DemoAdapter,
+    run_mode: RunMode,
+    pipeline: StepPipeline,
+    reservation: SessionReservation | None = None,
+) -> RunResult:
+    """Run one prepared demo session through a selected run mode."""
+    reservation = reservation or context.admission.try_reserve()
+    if reservation is None:
+        result = RunResult.rejected(reason="busy")
+        context.run_metrics.record_session(result)
+        return result
+
+    provider: Any | None = None
+    session_edges: SessionEdges | None = None
+    driver_started = False
+    try:
+        create_provider = getattr(adapter, "create_model_input_provider")
+        provider = context.host.call(create_provider, spec, scenario)
+        run_mode.validate_session(
+            spec=spec,
+            scenario=scenario,
+            adapter=adapter,
+            provider=provider,
+        )
+        session_edges = run_mode.create_session_edges(
+            context=context,
+            spec=spec,
+            scenario=scenario,
+            provider=provider,
+            adapter=adapter,
+        )
+        driver = run_mode.select_driver()
+        if not isinstance(driver, BatchSessionDriver):
+            raise TypeError(
+                "Phase 2 run_demo_session supports BatchSessionDriver only, "
+                f"got {type(driver).__name__}."
+            )
+        driver_started = True
+        result = driver.run_one_session(
+            host=context.host,
+            provider=provider,
+            session_edges=session_edges,
+            pipeline=pipeline,
+        )
+        context.run_metrics.record_session(result)
+        return result
+    except DriverInvariantError:
+        raise
+    except Exception as exc:
+        if provider is not None and not driver_started:
+            try:
+                context.host.call(provider.close)
+            except Exception as close_exc:
+                if session_edges is not None:
+                    session_edges.metrics.record_cleanup_error(close_exc)
+                else:
+                    context.run_metrics.record_cleanup_error(close_exc)
+        if session_edges is not None:
+            result = session_edges.close_result(
+                status="failed",
+                reason=str(exc),
+                error=exc,
+            )
+        else:
+            result = RunResult(status="failed", reason=str(exc), error=exc)
+        context.run_metrics.record_session(result)
+        return result
+    finally:
+        reservation.release()
+
+
+def _session_info(session: InferenceSession) -> SessionInfo:
+    session_info = getattr(session, "session_info", None)
+    if not callable(session_info):
+        return SessionInfo()
+    value = session_info()
+    if not isinstance(value, SessionInfo):
+        raise TypeError(
+            "session.session_info() must return SessionInfo, "
+            f"got {type(value).__name__}."
+        )
+    return value
+
+
+def _close_safely(close: Any, session_edges: SessionEdges) -> None:
+    try:
+        close()
+    except Exception as exc:
+        session_edges.metrics.record_cleanup_error(exc)
+
+
+__all__ = [
+    "BatchSessionDriver",
+    "DriverInvariantError",
+    "run_demo_session",
+]

flashdreams/flashdreams/runtime/demo/host.py

@@ -0,0 +1,66 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Minimal runtime host for the Phase 2 demo-session vertical slice."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import TypeVar
+
+from flashdreams.runtime.inputs import InferenceInput
+from flashdreams.runtime.interfaces import InferenceRuntime, InferenceSession
+
+_T = TypeVar("_T")
+
+
+class RuntimeHost:
+    """Thin synchronous host around an :class:`InferenceRuntime`.
+
+    Phase 3 moves the thread-affine worker boundary here. Phase 2 keeps the
+    dispatch direct so fake-model CPU tests can prove the session-driver shape
+    without introducing worker behavior early.
+    """
+
+    def __init__(self, runtime: InferenceRuntime) -> None:
+        self._runtime = runtime
+        self._healthy = True
+
+    @property
+    def runtime(self) -> InferenceRuntime:
+        """Return the hosted runtime."""
+        return self._runtime
+
+    @property
+    def is_healthy(self) -> bool:
+        """Return whether admission should continue accepting sessions."""
+        return self._healthy
+
+    def mark_unhealthy(self) -> None:
+        """Latch the host as unhealthy."""
+        self._healthy = False
+
+    def call(self, func: Callable[..., _T], /, *args: object, **kwargs: object) -> _T:
+        """Run one model-affine callable synchronously."""
+        return func(*args, **kwargs)
+
+    async def call_async(
+        self,
+        func: Callable[..., _T],
+        /,
+        *args: object,
+        **kwargs: object,
+    ) -> _T:
+        """Async-compatible direct dispatch placeholder for Phase 3."""
+        return self.call(func, *args, **kwargs)
+
+    def start_session(self, inputs: InferenceInput) -> InferenceSession:
+        """Start one inference session through the hosted runtime."""
+        return self._runtime.start_session(inputs)
+
+    def close(self) -> None:
+        """Close the hosted runtime."""
+        self._runtime.close()
+
+
+__all__ = ["RuntimeHost"]