From 1469a587fd3884e79a8f408770ed176c806c2ef9 Mon Sep 17 00:00:00 2001 From: Jesse Archer Date: Tue, 11 Aug 2026 07:14:16 +0000 Subject: [PATCH 1/5] Preserve setup-time driver invariants Keep setup-time DriverInvariantError exceptions out of normal setup error policy handling in both batch and realtime demo drivers. Add regression coverage for setup invariant propagation and document MP4 generation handling across realtime resets. --- .../flashdreams/runtime/demo/drivers.py | 4 +++ .../flashdreams/runtime/demo/outputs.py | 2 ++ .../test_demo_runtime_realtime_driver.py | 35 +++++++++++++++++++ .../tests/test_demo_runtime_vertical_slice.py | 31 ++++++++++++++++ 4 files changed, 72 insertions(+) diff --git a/flashdreams/flashdreams/runtime/demo/drivers.py b/flashdreams/flashdreams/runtime/demo/drivers.py index 9bdd40db8..27179ac32 100644 --- a/flashdreams/flashdreams/runtime/demo/drivers.py +++ b/flashdreams/flashdreams/runtime/demo/drivers.py @@ -65,6 +65,8 @@ def run_one_session( session_info = host.call(_session_info, session) session_edges.output_sink.open(session_info) setup_ok = True + except DriverInvariantError: + raise except Exception as exc: action = session_edges.error_policy.handle_setup_error(exc) if action.drop_chunk or action.result_status == "completed": @@ -206,6 +208,8 @@ async def run_one_session( session_edges.output_sink.open(session_info) session_edges.output_sink.begin_generation(generation) setup_ok = True + except DriverInvariantError: + raise except Exception as exc: action = session_edges.error_policy.handle_setup_error(exc) if action.drop_chunk or action.result_status == "completed": diff --git a/flashdreams/flashdreams/runtime/demo/outputs.py b/flashdreams/flashdreams/runtime/demo/outputs.py index 866acbb00..42123e747 100644 --- a/flashdreams/flashdreams/runtime/demo/outputs.py +++ b/flashdreams/flashdreams/runtime/demo/outputs.py @@ -169,6 +169,8 @@ def open(self, session_info: SessionInfo) -> None: def begin_generation(self, generation: int) -> None: if generation < 0: raise ValueError("generation must be >= 0.") + # MP4 recording is continuous across realtime resets; WebRTC is the sink + # that drops stale generations. def write(self, result: StepResult) -> OutputDecision: if not self._opened or self._closed or self._collector is None: diff --git a/flashdreams/tests/test_demo_runtime_realtime_driver.py b/flashdreams/tests/test_demo_runtime_realtime_driver.py index 5c79b83f7..7195218ec 100644 --- a/flashdreams/tests/test_demo_runtime_realtime_driver.py +++ b/flashdreams/tests/test_demo_runtime_realtime_driver.py @@ -223,6 +223,41 @@ async def test_realtime_driver_invariant_finalizes_edges_before_reraising() -> N assert metrics.closed +@pytest.mark.asyncio +async def test_realtime_driver_setup_invariant_reraises_without_error_policy() -> None: + runtime = _FakeRealtimeRuntime(session=_FakeRealtimeSession(num_steps=1)) + host = RuntimeHost(runtime) + provider = _FakeRealtimeProvider( + fail_initial=DriverInvariantError("setup invariant") + ) + output = _RecordingOutputSink() + transport = _RecordingTransport() + metrics = InMemorySessionMetricsRecorder() + edges = _edges( + output=output, + transport=transport, + metrics=metrics, + error_policy=_SetupPolicy(result_status="failed"), + ) + + try: + with pytest.raises(DriverInvariantError, match="setup invariant"): + await RealtimeSessionDriver().run_one_session( + host=host, + provider=provider, + session_edges=edges, + pipeline=StepPipeline(), + ) + finally: + host.close() + + assert provider.close_count == 1 + assert output.close_count == 1 + assert transport.close_count == 1 + assert metrics.closed + assert metrics.errors == [] + + @pytest.mark.asyncio async def test_realtime_step_invariant_reraises_without_error_policy() -> None: runtime = _FakeRealtimeRuntime(session=_FakeRealtimeSession(num_steps=1)) diff --git a/flashdreams/tests/test_demo_runtime_vertical_slice.py b/flashdreams/tests/test_demo_runtime_vertical_slice.py index 1f430a459..826684033 100644 --- a/flashdreams/tests/test_demo_runtime_vertical_slice.py +++ b/flashdreams/tests/test_demo_runtime_vertical_slice.py @@ -572,6 +572,37 @@ def test_setup_failure_can_return_skipped_but_not_completed() -> None: assert provider.close_count == 1 +def test_batch_driver_setup_invariant_reraises_without_error_policy() -> None: + provider = _FakeVideoModelInputProvider( + fail_initial=DriverInvariantError("setup invariant") + ) + output = _RecordingOutputSink() + transport = _RecordingTransport() + metrics = InMemorySessionMetricsRecorder() + edges = SessionEdges( + input_source=_FakeBatchInputSource(num_windows=1), + output_sink=output, + cleanup_tasks=set(), + metrics=metrics, + error_policy=_SetupPolicy(result_status="failed"), + transport=transport, + ) + + with pytest.raises(DriverInvariantError, match="setup invariant"): + BatchSessionDriver().run_one_session( + host=RuntimeHost(_FakeVideoRuntime(session=_FakeVideoSession(num_steps=1))), + provider=provider, + session_edges=edges, + pipeline=StepPipeline(), + ) + + assert output.close_count == 1 + assert transport.close_count == 1 + assert metrics.closed + assert metrics.errors == [] + assert provider.close_count == 1 + + def test_batch_driver_invariant_finalizes_edges_when_host_closed() -> None: runtime = _FakeVideoRuntime(session=_FakeVideoSession(num_steps=1)) host = RuntimeHost(runtime) From 766e056c87d21955333e923f88e38676ba02462a Mon Sep 17 00:00:00 2001 From: Jesse Archer Date: Tue, 11 Aug 2026 07:44:22 +0000 Subject: [PATCH 2/5] Move LingBot shared provider off input mappings Make the migrated LingBot demo provider convert UserInputWindow values directly into model InferenceInput, including trace slicing, live camera integration, skipped WebRTC input catch-up, and text-event prompt updates. Allow shared replay runs to use mapping-free model providers while keeping legacy compatibility runners mapping-gated, and update LingBot parity coverage against the old mapping path. --- .../flashdreams/runtime/demo/replay.py | 63 ++- integrations/lingbot/lingbot/demo/adapter.py | 47 +- .../lingbot/lingbot/demo/providers.py | 430 +++++++++++++++--- integrations/lingbot/lingbot/runtime.py | 6 +- integrations/lingbot/tests/test_demo_api.py | 33 +- .../lingbot/tests/test_demo_providers.py | 151 +++++- 6 files changed, 560 insertions(+), 170 deletions(-) diff --git a/flashdreams/flashdreams/runtime/demo/replay.py b/flashdreams/flashdreams/runtime/demo/replay.py index fa2f507a9..3528f3993 100644 --- a/flashdreams/flashdreams/runtime/demo/replay.py +++ b/flashdreams/flashdreams/runtime/demo/replay.py @@ -90,16 +90,12 @@ def run_replay_demo( raise ValueError("run_replay_demo does not support WebRTC output.") prepared = adapter.prepare_scenario(spec) - mapping = prepared.mapping or adapter.default_input_mapping() - if mapping is None: - raise ValueError( - "Demo scenario did not provide an input mapping, and the adapter " - "has no default input mapping." - ) + mapping = _scenario_mapping(prepared=prepared, adapter=adapter) if spec.config is None: raise RuntimeError("DemoSpec.config was not initialized.") if runner is not None: + mapping = _require_replay_mapping(mapping) return _run_replay_demo_with_compat_runner( spec=spec, adapter=adapter, @@ -197,18 +193,26 @@ def _run_replay_demo_with_run_mode( spec: DemoSpec, adapter: DemoAdapter, prepared: "PreparedScenario", - mapping: InputMapping, + mapping: InputMapping | None, output_sink_factory: OutputSinkFactory, metrics: MetricsRecorder | None, ) -> RunResult: config = _require_config(spec) - _validate_replay_mapping( - adapter=adapter, - config=config, - mapping=mapping, - source_schema=prepared.source_schema, - canonicalizer=prepared.canonicalizer, - ) + if mapping is None: + if not callable(getattr(adapter, "create_model_input_provider", None)): + raise ValueError( + "Demo scenario did not provide an input mapping, and the adapter " + "has no model input provider or default input mapping." + ) + adapter.validate_config(config) + else: + _validate_replay_mapping( + adapter=adapter, + config=config, + mapping=mapping, + source_schema=prepared.source_schema, + canonicalizer=prepared.canonicalizer, + ) request_state = _ReplayStepRequestState() runtime = _ReplayRuntimeAdapter( runtime=adapter.create_runtime(config), @@ -325,7 +329,7 @@ def __init__( self, *, adapter: DemoAdapter, - mapping: InputMapping, + mapping: InputMapping | None, request_state: "_ReplayStepRequestState", ) -> None: self._adapter = adapter @@ -370,6 +374,11 @@ def create_model_input_provider( create_provider = getattr(self._adapter, "create_model_input_provider", None) if callable(create_provider): return create_provider(spec, scenario) + if self._mapping is None: + raise ValueError( + "Replay adapter requires an input mapping when no model input " + "provider is available." + ) return _ReplayMappingModelInputProvider( adapter=self._adapter, scenario=scenario, @@ -498,7 +507,7 @@ def next_window(self, request: StepRequirements) -> UserInputWindow: return UserInputWindow( start_s=window.start_s, end_s=window.end_s, - inputs=self._user_inputs, + inputs=self._user_inputs.window(window), ) @@ -589,6 +598,28 @@ def _require_config(spec: DemoSpec) -> InferenceConfig: return spec.config +def _scenario_mapping( + *, + prepared: PreparedScenario, + adapter: DemoAdapter, +) -> InputMapping | None: + if prepared.mapping is not None: + return prepared.mapping + default_input_mapping = getattr(adapter, "default_input_mapping", None) + if not callable(default_input_mapping): + return None + return default_input_mapping() + + +def _require_replay_mapping(mapping: InputMapping | None) -> InputMapping: + if mapping is not None: + return mapping + raise ValueError( + "Compatibility replay runners require an input mapping; the prepared " + "scenario and adapter did not provide one." + ) + + def _all_user_inputs_window(user_inputs: UserInputs) -> TimeWindow: if not user_inputs.events: return TimeWindow(start_s=0.0, end_s=3600.0) diff --git a/integrations/lingbot/lingbot/demo/adapter.py b/integrations/lingbot/lingbot/demo/adapter.py index fedd0a2c2..777e1a59a 100644 --- a/integrations/lingbot/lingbot/demo/adapter.py +++ b/integrations/lingbot/lingbot/demo/adapter.py @@ -9,7 +9,6 @@ from typing import Any from flashdreams.runtime import ( - InputCanonicalizer, UserInputCapability, UserInputs, UserInputSchema, @@ -22,10 +21,6 @@ WebRTCOutputSpec, ) from flashdreams.runtime.interfaces import InferenceRuntime -from lingbot.input_mapping import ( - KeyboardToCameraCommand, - TextEventSelection, -) from lingbot.runtime import ( FIELD_FPS, FIELD_PIXEL_HEIGHT, @@ -37,7 +32,11 @@ inference_input_from_replay_inputs, ) -from .providers import LingbotInputProvider +from .providers import ( + PROVIDER_INPUTS_METADATA_KEY, + LingbotInputProvider, + create_lingbot_provider_inputs, +) from .spec import ( resolve_replay_inputs, resolve_text_event_prompts, @@ -93,27 +92,11 @@ def prepare_scenario(self, spec: DemoSpec) -> PreparedScenario: ) text_event_prompts = resolve_text_event_prompts(scenario) user_inputs = resolve_user_input_events(scenario) - if live_camera or _camera_source(scenario) == "events": - # Live control still needs the scenario's calibration, so the trace - # is loaded for its intrinsics and world scale and then discarded - # as a trajectory source. - trace = self.create_input_mapping(replay_inputs).camera_trace - mapping = self.create_live_input_mapping( - fps=replay_inputs.fps, - base_intrinsics=trace.intrinsics[0], - # A trace's world scale is derived from how far its poses - # travel, so a stationary example yields 0. Live control has no - # trajectory to normalize against, so it falls back to the same - # unit scale the live runtime uses. - world_scale=trace.world_scale or 1.0, - prompt=replay_inputs.prompt, - text_event_prompts=text_event_prompts, - ) - else: - mapping = self.create_input_mapping( - replay_inputs, - text_event_prompts=text_event_prompts, - ) + provider_inputs = create_lingbot_provider_inputs( + replay_inputs, + live_camera=live_camera or _camera_source(scenario) == "events", + text_event_prompts=text_event_prompts, + ) return PreparedScenario( initial_inputs=inference_input_from_replay_inputs(replay_inputs), user_inputs=user_inputs, @@ -122,11 +105,10 @@ def prepare_scenario(self, spec: DemoSpec) -> PreparedScenario: include_keyboard=live_camera, include_text_events=live_camera and bool(text_event_prompts), ), - canonicalizer=_canonicalizer(text_event_prompts), - mapping=mapping, metadata={ "model_id": self.model_id, "preset_id": self.preset_id(spec.config), + PROVIDER_INPUTS_METADATA_KEY: provider_inputs, }, ) @@ -246,13 +228,6 @@ def _source_schema( ) -def _canonicalizer(text_event_prompts: Mapping[str, str] | None) -> InputCanonicalizer: - converters: list[Any] = [KeyboardToCameraCommand()] - if text_event_prompts: - converters.append(TextEventSelection()) - return InputCanonicalizer(converters) - - __all__ = [ "LingbotDemoAdapter", "ReplayRuntimeFactory", diff --git a/integrations/lingbot/lingbot/demo/providers.py b/integrations/lingbot/lingbot/demo/providers.py index b11db6834..f95083a9f 100644 --- a/integrations/lingbot/lingbot/demo/providers.py +++ b/integrations/lingbot/lingbot/demo/providers.py @@ -5,16 +5,18 @@ from __future__ import annotations -from collections.abc import Mapping +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from types import MappingProxyType from typing import Any +import numpy as np +import torch from flashdreams.runtime import ( - CanonicalInputs, InferenceInput, InferenceInputSchema, - StepRequest, StepRequirements, - TimeWindow, UserInputs, ) from flashdreams.runtime.demo import ( @@ -23,23 +25,174 @@ ProviderCapabilities, UserInputWindow, ) +from flashdreams.runtime.keyboard import DEFAULT_SUPPORTED_KEYS, KeyboardState from flashdreams.serving.webrtc.services import ( WEBRTC_SKIPPED_INPUTS_METADATA_KEY, WEBRTC_SKIPPED_WINDOW_METADATA_KEY, ) -from lingbot.input_mapping import CAMERA_COMMAND, LingbotInputMapping -from lingbot.runtime import LingbotModelAdapter +from lingbot.controls import CameraPoseIntegrator, PoseSegment +from lingbot.runtime import ( + FIELD_PROMPT, + FIELD_WORLD_SCALE, + LingbotModelAdapter, + LingbotReplayInputs, +) +FIELD_CAMERA_TRAJECTORY = "camera_trajectory" +FIELD_CAMERA_INTRINSICS = "camera_intrinsics" +FIELD_TOTAL_CAMERA_FRAMES = "total_camera_frames" +PROVIDER_INPUTS_METADATA_KEY = "lingbot_provider_inputs" + +_INTRINSICS_REFERENCE_HEIGHT = 480 +_INTRINSICS_REFERENCE_WIDTH = 832 +_CLEAR_STATES = frozenset({"clear", "release", "off", "none"}) +_TRIGGER_STATES = frozenset({"trigger", "hold", "on"}) + + +@dataclass(frozen=True, kw_only=True, slots=True) +class LingbotCameraTrace: + """Fixed LingBot camera trajectory loaded for a shared demo scenario.""" + + __hash__ = None + + poses: torch.Tensor + intrinsics: torch.Tensor + world_scale: float + + def __post_init__(self) -> None: + if self.poses.ndim != 3 or self.poses.shape[1:] != (4, 4): + raise ValueError( + f"LingbotCameraTrace.poses must be [T, 4, 4], got " + f"{tuple(self.poses.shape)}." + ) + if self.intrinsics.ndim != 2 or self.intrinsics.shape[1] != 4: + raise ValueError( + f"LingbotCameraTrace.intrinsics must be [T, 4], got " + f"{tuple(self.intrinsics.shape)}." + ) + if self.world_scale < 0: + raise ValueError("LingbotCameraTrace.world_scale must be >= 0.") + + @property + def frame_count(self) -> int: + return int(self.poses.shape[0]) -class LingbotInputProvider: - """Convert shared user-input windows into Lingbot model inputs. - Lingbot's existing mapping API still accepts the legacy ``StepRequest`` - shape, because the runtime owns frame-start metadata today. This provider is - the model-owned bridge from shared demo drivers to that mapping boundary; - the bridge stays here so WebRTC transport code never has to know Lingbot's - camera or prompt semantics. - """ +@dataclass(frozen=True, kw_only=True, slots=True) +class LingbotProviderInputs: + """Provider-owned LingBot input setup for a migrated shared demo scenario.""" + + fps: int + trace: LingbotCameraTrace | None = None + base_intrinsics: torch.Tensor | Sequence[float] | None = None + world_scale: float | None = None + prompt: str = "" + text_event_prompts: Mapping[str, str] = field(default_factory=dict) + + def __post_init__(self) -> None: + if self.fps <= 0: + raise ValueError("LingbotProviderInputs.fps must be > 0.") + object.__setattr__( + self, + "text_event_prompts", + MappingProxyType( + {str(key): str(value) for key, value in self.text_event_prompts.items()} + ), + ) + if self.trace is not None: + object.__setattr__(self, "world_scale", self.trace.world_scale) + object.__setattr__(self, "base_intrinsics", None) + return + if self.base_intrinsics is None: + raise ValueError( + "Live LingBot provider inputs require base_intrinsics." + ) + intrinsics = torch.as_tensor(self.base_intrinsics, dtype=torch.float32).reshape( + 4 + ) + object.__setattr__(self, "base_intrinsics", intrinsics.clone()) + if self.world_scale is None or self.world_scale <= 0: + raise ValueError("Live LingBot provider inputs require world_scale > 0.") + + @property + def live_camera(self) -> bool: + return self.trace is None + + +def create_lingbot_provider_inputs( + replay_inputs: LingbotReplayInputs, + *, + live_camera: bool, + text_event_prompts: Mapping[str, str] | None = None, +) -> LingbotProviderInputs: + """Build provider-owned input setup from resolved LingBot replay inputs.""" + trace = load_camera_trace( + camera_poses_path=replay_inputs.camera_poses_path, + camera_intrinsics_path=replay_inputs.camera_intrinsics_path, + pixel_height=replay_inputs.pixel_height, + pixel_width=replay_inputs.pixel_width, + intrinsics_reference_height=_INTRINSICS_REFERENCE_HEIGHT, + intrinsics_reference_width=_INTRINSICS_REFERENCE_WIDTH, + world_scale=replay_inputs.world_scale, + ) + if live_camera: + return LingbotProviderInputs( + fps=replay_inputs.fps, + base_intrinsics=trace.intrinsics[0], + # A trace's world scale is derived from pose spread, so stationary + # examples fall back to the unit live-control scale. + world_scale=trace.world_scale or 1.0, + prompt=replay_inputs.prompt, + text_event_prompts=text_event_prompts or {}, + ) + return LingbotProviderInputs( + fps=replay_inputs.fps, + trace=trace, + prompt=replay_inputs.prompt, + text_event_prompts=text_event_prompts or {}, + ) + + +def load_camera_trace( + *, + camera_poses_path: str | Path, + camera_intrinsics_path: str | Path, + pixel_height: int, + pixel_width: int, + intrinsics_reference_height: int, + intrinsics_reference_width: int, + world_scale: float | None = None, +) -> LingbotCameraTrace: + """Load and preprocess a fixed LingBot camera trajectory from ``.npy`` files.""" + from lingbot.encoder.utils import ( # noqa: PLC0415 + get_Ks_transformed, + preprocess_example_poses, + ) + + intrinsics = torch.from_numpy( + np.asarray(np.load(camera_intrinsics_path), dtype=np.float32) + ) + intrinsics = get_Ks_transformed( + intrinsics, + height_org=intrinsics_reference_height, + width_org=intrinsics_reference_width, + height_resize=pixel_height, + width_resize=pixel_width, + height_final=pixel_height, + width_final=pixel_width, + ) + poses, inferred_world_scale = preprocess_example_poses( + np.asarray(np.load(camera_poses_path)) + ) + return LingbotCameraTrace( + poses=torch.from_numpy(np.ascontiguousarray(poses)).to(torch.float32), + intrinsics=intrinsics.to(torch.float32), + world_scale=float(inferred_world_scale if world_scale is None else world_scale), + ) + + +class LingbotInputProvider: + """Convert shared user-input windows directly into LingBot model inputs.""" def __init__( self, @@ -47,17 +200,17 @@ def __init__( scenario: PreparedScenario, inference_input_schema: InferenceInputSchema | None = None, ) -> None: - mapping = scenario.mapping - if not isinstance(mapping, LingbotInputMapping): + provider_inputs = scenario.metadata.get(PROVIDER_INPUTS_METADATA_KEY) + if not isinstance(provider_inputs, LingbotProviderInputs): raise TypeError( - "LingbotInputProvider requires PreparedScenario.mapping to be " - f"LingbotInputMapping, got {type(mapping).__name__}." + "LingbotInputProvider requires PreparedScenario.metadata" + f"[{PROVIDER_INPUTS_METADATA_KEY!r}] to be LingbotProviderInputs." ) if inference_input_schema is None: inference_input_schema = LingbotModelAdapter().inference_input_schema self.capabilities = ProviderCapabilities( - supports_realtime_clock=_supports_realtime_clock(mapping), + supports_realtime_clock=provider_inputs.live_camera, supports_recorded_input=True, supports_reset=True, deterministic_given_inputs=True, @@ -65,20 +218,31 @@ def __init__( inference_input_schema=inference_input_schema, ) self._scenario = scenario - self._mapping = mapping + self._inputs = provider_inputs self._step_base_inputs = InferenceInput( step=scenario.initial_inputs.step, metadata=scenario.initial_inputs.metadata, ) self._next_frame_start = 0 + self._keyboard_state = KeyboardState(supported_keys=DEFAULT_SUPPORTED_KEYS) + self._integrator = ( + CameraPoseIntegrator() if provider_inputs.live_camera else None + ) + self._active_text_event_id: str | None = None + self._applied_text_event_id: str | None = None self._closed = False def prepare_initial_input(self) -> InferenceInput: self._require_open() self._reset_state() - return self._mapping.map_global_conditioning_inputs( - canonical_inputs=CanonicalInputs(), - inference_input=self._scenario.initial_inputs, + payload = dict(self._scenario.initial_inputs.global_conditioning) + payload[FIELD_WORLD_SCALE] = self._inputs.world_scale + if self._inputs.trace is not None: + payload[FIELD_TOTAL_CAMERA_FRAMES] = self._inputs.trace.frame_count + return InferenceInput( + global_conditioning=payload, + step=self._scenario.initial_inputs.step, + metadata=self._scenario.initial_inputs.metadata, ) def prepare_step( @@ -92,25 +256,50 @@ def prepare_step( return PreparedStep(control=user_window.control) self._advance_skipped_input_state(user_window) - legacy_request = self._legacy_step_request( - request=request, - user_window=user_window, + metadata: dict[str, Any] = dict(request.metadata) + num_frames = _metadata_positive_int( + metadata, + "num_frames", + default=request.input_frame_count, ) - assert legacy_request.user_input_window is not None - canonical_inputs = self._scenario.canonicalizer.canonicalize( - user_window.inputs, - window=legacy_request.user_input_window, - source_schema=self._scenario.source_schema, + frame_start = _metadata_int( + metadata, + "frame_start", + default=self._next_frame_start, ) - inference_input = self._mapping.map_step_inputs( - canonical_inputs=canonical_inputs, - inference_input=self._step_base_inputs, - request=legacy_request, + if self._inputs.trace is not None: + self._consume_window_inputs( + user_window.inputs, + start_s=user_window.start_s, + end_s=user_window.end_s, + collect_camera_segments=False, + ) + poses, intrinsics = self._slice_trace( + frame_start=frame_start, + num_frames=num_frames, + ) + else: + segments = self._consume_window_inputs( + user_window.inputs, + start_s=user_window.start_s, + end_s=user_window.end_s, + collect_camera_segments=True, + ) + poses, intrinsics = self._integrate_live_camera( + segments=segments, + start_s=user_window.start_s, + end_s=user_window.end_s, + num_frames=num_frames, + ) + step = dict(self._step_base_inputs.step) + step[FIELD_CAMERA_TRAJECTORY] = poses + step[FIELD_CAMERA_INTRINSICS] = intrinsics + inference_input = InferenceInput( + global_conditioning=self._text_event_update(), + step=step, + metadata=self._step_base_inputs.metadata, ) - self._next_frame_start = _required_int( - legacy_request.metadata, - "frame_start", - ) + _required_positive_int(legacy_request.metadata, "num_frames") + self._next_frame_start = frame_start + num_frames return PreparedStep(inference_input=inference_input) def reset(self, inputs: InferenceInput | None = None) -> None: @@ -121,36 +310,126 @@ def reset(self, inputs: InferenceInput | None = None) -> None: def close(self) -> None: self._closed = True - def _legacy_step_request( + def _slice_trace( self, *, - request: StepRequirements, - user_window: UserInputWindow, - ) -> StepRequest: - metadata: dict[str, Any] = dict(request.metadata) - metadata["num_frames"] = _metadata_positive_int( - metadata, - "num_frames", - default=request.input_frame_count, + frame_start: int, + num_frames: int, + ) -> tuple[torch.Tensor, torch.Tensor]: + trace = self._inputs.trace + if trace is None: + raise RuntimeError("Cannot slice a missing LingBot camera trace.") + frame_end = frame_start + num_frames + if frame_end > trace.frame_count: + raise ValueError( + f"Lingbot camera trace has {trace.frame_count} frames, but " + f"step needs frames [{frame_start}, {frame_end})." + ) + return ( + trace.poses[frame_start:frame_end], + trace.intrinsics[frame_start:frame_end], ) - metadata["frame_start"] = _metadata_int( - metadata, - "frame_start", - default=self._next_frame_start, + + def _integrate_live_camera( + self, + *, + segments: list[PoseSegment], + start_s: float, + end_s: float, + num_frames: int, + ) -> tuple[torch.Tensor, torch.Tensor]: + if self._integrator is None: + raise RuntimeError("Cannot integrate LingBot camera without an integrator.") + base_intrinsics = self._inputs.base_intrinsics + if not isinstance(base_intrinsics, torch.Tensor): + raise RuntimeError("Live LingBot camera control requires base_intrinsics.") + frame_times = [ + start_s + (index + 1) / self._inputs.fps for index in range(num_frames) + ] + frame_times[-1] = min(frame_times[-1], end_s) + poses = self._integrator.integrate_chunk( + segments=segments, + frame_times=frame_times, ) - return StepRequest( - step_index=request.step_index, - inference_input_schema=request.inference_input_schema, - user_input_window=TimeWindow( - start_s=user_window.start_s, - end_s=user_window.end_s, - ), - metadata=metadata, + poses_t = torch.from_numpy(np.ascontiguousarray(poses)).to(torch.float32) + poses_t = poses_t.reshape(num_frames, 4, 4) + intrinsics_t = base_intrinsics.reshape(1, 4).repeat(num_frames, 1) + return poses_t, intrinsics_t + + def _consume_window_inputs( + self, + user_inputs: UserInputs, + *, + start_s: float, + end_s: float, + collect_camera_segments: bool, + ) -> list[PoseSegment]: + segments: list[PoseSegment] = [] + segment_start = start_s + previous_keys = self._keyboard_state.resolved_effective_keys() + + for event in user_inputs.events: + if event.event_type in {"key_down", "key_up"} and self._inputs.live_camera: + key = event.payload.get("key") + if not isinstance(key, str): + continue + edge_t = min(max(float(event.timestamp_s), start_s), end_s) + if collect_camera_segments and edge_t > segment_start: + segments.append((segment_start, edge_t, previous_keys)) + segment_start = edge_t + self._keyboard_state.apply_event( + event="keydown" if event.event_type == "key_down" else "keyup", + key=key, + ) + previous_keys = self._keyboard_state.resolved_effective_keys() + continue + if event.event_type == "text_event": + self._apply_text_event(event.payload) + + if collect_camera_segments: + if end_s > segment_start or not segments: + segments.append((segment_start, end_s, previous_keys)) + return segments + return [] + + def _apply_text_event(self, payload: Mapping[str, Any]) -> None: + event_id = payload.get("event_id") + state = str(payload.get("state", "trigger")).strip().lower() + if state and state not in _CLEAR_STATES and state not in _TRIGGER_STATES: + raise ValueError( + f"Unsupported text event state {state!r}. Supported states: " + f"{sorted(_CLEAR_STATES | _TRIGGER_STATES)}." + ) + if event_id is None or state in _CLEAR_STATES: + self._active_text_event_id = None + return + self._active_text_event_id = str(event_id) + + def _text_event_update(self) -> Mapping[str, Any]: + if not self._inputs.text_event_prompts: + return {} + event_id = self._active_text_event_id + if event_id == self._applied_text_event_id: + return {} + if event_id is not None and event_id not in self._inputs.text_event_prompts: + supported = ", ".join(sorted(self._inputs.text_event_prompts)) + raise ValueError( + f"Unknown Lingbot text event_id={event_id!r}. Supported: {supported}" + ) + self._applied_text_event_id = event_id + prompt = ( + self._inputs.prompt + if event_id is None + else self._inputs.text_event_prompts[event_id] ) + return {} if prompt is None else {FIELD_PROMPT: prompt} def _reset_state(self) -> None: - self._scenario.canonicalizer.reset() - self._mapping.reset() + self._keyboard_state = KeyboardState(supported_keys=DEFAULT_SUPPORTED_KEYS) + if self._integrator is not None: + self._integrator.reset() + self._active_text_event_id = None + self._applied_text_event_id = None self._next_frame_start = 0 def _advance_skipped_input_state(self, user_window: UserInputWindow) -> None: @@ -170,24 +449,17 @@ def _advance_skipped_input_state(self, user_window: UserInputWindow) -> None: end_s = float(end_value) if end_s <= start_s: return - self._scenario.canonicalizer.canonicalize( + self._consume_window_inputs( skipped_inputs, - window=TimeWindow(start_s=start_s, end_s=end_s), - source_schema=self._scenario.source_schema, + start_s=start_s, + end_s=end_s, + collect_camera_segments=False, ) def _require_open(self) -> None: if self._closed: raise RuntimeError("LingbotInputProvider is closed.") - -def _supports_realtime_clock(mapping: LingbotInputMapping) -> bool: - return any( - modality.name == CAMERA_COMMAND.name - for modality in mapping.mapping_schema.consumes - ) - - def _metadata_int( metadata: Mapping[str, Any], name: str, @@ -226,4 +498,14 @@ def _required_positive_int(metadata: Mapping[str, Any], name: str) -> int: return value -__all__ = ["LingbotInputProvider"] +__all__ = [ + "FIELD_CAMERA_INTRINSICS", + "FIELD_CAMERA_TRAJECTORY", + "FIELD_TOTAL_CAMERA_FRAMES", + "PROVIDER_INPUTS_METADATA_KEY", + "LingbotCameraTrace", + "LingbotInputProvider", + "LingbotProviderInputs", + "create_lingbot_provider_inputs", + "load_camera_trace", +] diff --git a/integrations/lingbot/lingbot/runtime.py b/integrations/lingbot/lingbot/runtime.py index ef7b549ef..908306c65 100644 --- a/integrations/lingbot/lingbot/runtime.py +++ b/integrations/lingbot/lingbot/runtime.py @@ -118,8 +118,8 @@ class LingbotSessionInputs: """Session-global Lingbot state established at session start or reset. The camera trajectory is deliberately absent: it arrives per step through - ``InferenceInput.step``, built by the selected input mapping from either a - fixed trace or live user events. + ``InferenceInput.step``, built by the selected input provider or legacy + mapping from either a fixed trace or live user events. """ prompt: str @@ -580,7 +580,7 @@ def _require_step_tensor( if name not in inputs.step: raise ValueError( f"Lingbot step inputs are missing {name!r}. The selected input " - f"mapping must produce it for every step." + f"provider or mapping must produce it for every step." ) value = inputs.step[name] if not isinstance(value, torch.Tensor): diff --git a/integrations/lingbot/tests/test_demo_api.py b/integrations/lingbot/tests/test_demo_api.py index 98444a7c5..d2cbd14db 100644 --- a/integrations/lingbot/tests/test_demo_api.py +++ b/integrations/lingbot/tests/test_demo_api.py @@ -20,6 +20,7 @@ LingbotWebRTCScenario, ) from lingbot.demo.app import _replay_spec, _webrtc_spec, parse_args +from lingbot.demo.providers import PROVIDER_INPUTS_METADATA_KEY from lingbot.demo.replay import ( LingbotReplayRuntime, LingbotReplayRuntimeOptions, @@ -196,9 +197,14 @@ def test_lingbot_replay_adapter_accepts_null_output(tmp_path: Path) -> None: assert prepared.initial_inputs.global_conditioning[FIELD_FIRST_FRAME_PATH] == image assert prepared.initial_inputs.global_conditioning[FIELD_TOTAL_BLOCKS] == 1 + assert prepared.mapping is None + assert prepared.canonicalizer.converters == () + assert PROVIDER_INPUTS_METADATA_KEY in prepared.metadata -def test_lingbot_replay_demo_uses_shared_runner(tmp_path: Path) -> None: +def test_lingbot_replay_demo_rejects_compat_runner_without_mapping( + tmp_path: Path, +) -> None: image = tmp_path / "image.jpg" poses = tmp_path / "poses.npy" intrinsics = tmp_path / "intrinsics.npy" @@ -232,24 +238,15 @@ def fake_runner(**kwargs: Any) -> Sequence[OutputArtifact]: ), ) - result = run_replay_demo( - spec=spec, - adapter=adapter, - output_target_factory=lambda output_spec: output, - runner=fake_runner, - ) + with pytest.raises(ValueError, match="Compatibility replay runners require"): + run_replay_demo( + spec=spec, + adapter=adapter, + output_target_factory=lambda output_spec: output, + runner=fake_runner, + ) - assert result.status == "completed" - assert result.artifacts == ( - OutputArtifact(kind="video/mp4", uri="memory://lingbot"), - ) - assert len(calls) == 1 - assert calls[0]["adapter"] is adapter - assert calls[0]["config"] == spec.config - inputs = calls[0]["initial_inputs"].global_conditioning - assert inputs[FIELD_PROMPT] == "drive through a city" - assert inputs[FIELD_FIRST_FRAME_PATH] == image - assert inputs[FIELD_TOTAL_BLOCKS] == 1 + assert calls == [] def test_lingbot_replay_demo_run_mode_uses_model_provider( diff --git a/integrations/lingbot/tests/test_demo_providers.py b/integrations/lingbot/tests/test_demo_providers.py index db7425da9..1d042ea95 100644 --- a/integrations/lingbot/tests/test_demo_providers.py +++ b/integrations/lingbot/tests/test_demo_providers.py @@ -3,6 +3,7 @@ from __future__ import annotations +from dataclasses import dataclass from pathlib import Path from typing import Any @@ -10,21 +11,29 @@ import pytest import torch from lingbot.demo import DEFAULT_LINGBOT_PRESET, LINGBOT_MODEL_ID, LingbotDemoAdapter -from lingbot.demo.providers import LingbotInputProvider -from lingbot.input_mapping import ( +from lingbot.demo.providers import ( FIELD_CAMERA_TRAJECTORY, FIELD_TOTAL_CAMERA_FRAMES, + PROVIDER_INPUTS_METADATA_KEY, + LingbotInputProvider, +) +from lingbot.input_mapping import ( + KeyboardToCameraCommand, LingbotInputMapping, + TextEventSelection, ) from lingbot.runtime import ( FIELD_FIRST_FRAME_PATH, FIELD_PROMPT, FIELD_TOTAL_BLOCKS, + LingbotReplayInputs, ) from flashdreams.runtime import ( + CanonicalInputs, InferenceConfig, InferenceInput, + InputCanonicalizer, StepRequest, StepRequirements, TimeWindow, @@ -43,22 +52,20 @@ def test_lingbot_provider_initial_input_matches_mapping_path(tmp_path: Path) -> None: adapter = LingbotDemoAdapter() - expected = _prepared_scenario(tmp_path, adapter=adapter) actual = _prepared_scenario(tmp_path, adapter=adapter) - assert isinstance(expected.mapping, LingbotInputMapping) + legacy = _legacy_bridge(tmp_path, adapter=adapter) + assert actual.mapping is None + assert actual.canonicalizer.converters == () + assert PROVIDER_INPUTS_METADATA_KEY in actual.metadata provider = LingbotInputProvider( scenario=actual, inference_input_schema=adapter.inference_input_schema, ) - expected_initial = expected.mapping.map_global_conditioning_inputs( - canonical_inputs=expected.canonicalizer.canonicalize( - UserInputs(), - window=TimeWindow(start_s=0.0, end_s=0.0), - source_schema=expected.source_schema, - ), - inference_input=expected.initial_inputs, + expected_initial = legacy.mapping.map_global_conditioning_inputs( + canonical_inputs=CanonicalInputs(), + inference_input=actual.initial_inputs, ) actual_initial = provider.prepare_initial_input() @@ -75,15 +82,21 @@ def test_lingbot_provider_initial_input_matches_mapping_path(tmp_path: Path) -> def test_lingbot_provider_trace_steps_match_mapping_path(tmp_path: Path) -> None: adapter = LingbotDemoAdapter() - expected = _prepared_scenario(tmp_path, adapter=adapter) actual = _prepared_scenario(tmp_path, adapter=adapter) + legacy = _legacy_bridge(tmp_path, adapter=adapter) provider = LingbotInputProvider( scenario=actual, inference_input_schema=adapter.inference_input_schema, ) provider.prepare_initial_input() - expected_first = _legacy_step(expected, step_index=0, frame_start=0, num_frames=4) + expected_first = _legacy_step( + actual, + legacy, + step_index=0, + frame_start=0, + num_frames=4, + ) actual_first = _provider_step( provider, step_index=0, @@ -91,7 +104,13 @@ def test_lingbot_provider_trace_steps_match_mapping_path(tmp_path: Path) -> None num_frames=4, inputs=actual.user_inputs, ) - expected_second = _legacy_step(expected, step_index=1, frame_start=4, num_frames=4) + expected_second = _legacy_step( + actual, + legacy, + step_index=1, + frame_start=4, + num_frames=4, + ) actual_second = _provider_step( provider, step_index=1, @@ -118,17 +137,16 @@ def test_lingbot_provider_trace_steps_match_mapping_path(tmp_path: Path) -> None def test_lingbot_provider_uses_driver_user_window_inputs(tmp_path: Path) -> None: adapter = LingbotDemoAdapter() scenario_events = ({"t": 10.0, "type": "key_down", "key": "a"},) - expected = _prepared_scenario( + actual = _prepared_scenario( tmp_path, adapter=adapter, camera_source="events", events=scenario_events, ) - actual = _prepared_scenario( + legacy = _legacy_bridge( tmp_path, adapter=adapter, camera_source="events", - events=scenario_events, ) provider = LingbotInputProvider( scenario=actual, @@ -146,7 +164,8 @@ def test_lingbot_provider_uses_driver_user_window_inputs(tmp_path: Path) -> None provider.prepare_initial_input() expected_step = _legacy_step( - expected, + actual, + legacy, step_index=0, frame_start=0, num_frames=4, @@ -202,6 +221,16 @@ def test_lingbot_provider_folds_webrtc_skipped_inputs_into_state( provider.prepare_initial_input() idle_provider.prepare_initial_input() + legacy = _legacy_bridge( + tmp_path, + adapter=adapter, + camera_source="events", + ) + legacy.canonicalizer.canonicalize( + skipped_inputs, + window=TimeWindow(start_s=0.0, end_s=0.25), + source_schema=with_skip.source_schema, + ) actual = _provider_step( provider, step_index=0, @@ -220,24 +249,40 @@ def test_lingbot_provider_folds_webrtc_skipped_inputs_into_state( num_frames=4, inputs=UserInputs(), ) + expected = _legacy_step( + with_skip, + legacy, + step_index=0, + frame_start=4, + num_frames=4, + inputs=UserInputs(), + ) poses = actual.step[FIELD_CAMERA_TRAJECTORY] + assert torch.allclose(poses, expected.step[FIELD_CAMERA_TRAJECTORY]) assert not torch.allclose(poses, expected_idle.step[FIELD_CAMERA_TRAJECTORY]) assert not torch.allclose(poses[0], poses[-1]) def test_lingbot_provider_reset_clears_text_event_state(tmp_path: Path) -> None: adapter = LingbotDemoAdapter() + text_events = {"storm": "a violent storm"} actual = _prepared_scenario( tmp_path, adapter=adapter, camera_source="events", - text_events={"storm": "a violent storm"}, + text_events=text_events, events=( {"t": 10.0, "type": "key_down", "key": "w"}, {"t": 10.1, "type": "text_event", "event_id": "storm"}, ), ) + legacy = _legacy_bridge( + tmp_path, + adapter=adapter, + camera_source="events", + text_events=text_events, + ) provider = LingbotInputProvider( scenario=actual, inference_input_schema=adapter.inference_input_schema, @@ -252,6 +297,14 @@ def test_lingbot_provider_reset_clears_text_event_state(tmp_path: Path) -> None: num_frames=4, inputs=first_inputs, ) + expected_first = _legacy_step( + actual, + legacy, + step_index=0, + frame_start=0, + num_frames=4, + inputs=first_inputs, + ) repeated = _provider_step( provider, step_index=1, @@ -259,6 +312,14 @@ def test_lingbot_provider_reset_clears_text_event_state(tmp_path: Path) -> None: num_frames=4, inputs=UserInputs(), ) + expected_repeated = _legacy_step( + actual, + legacy, + step_index=1, + frame_start=4, + num_frames=4, + inputs=UserInputs(), + ) provider.reset() after_reset = _provider_step( provider, @@ -268,6 +329,8 @@ def test_lingbot_provider_reset_clears_text_event_state(tmp_path: Path) -> None: inputs=first_inputs, ) + assert first.global_conditioning == expected_first.global_conditioning + assert repeated.global_conditioning == expected_repeated.global_conditioning assert first.global_conditioning[FIELD_PROMPT] == "a violent storm" assert repeated.global_conditioning == {} assert after_reset.global_conditioning[FIELD_PROMPT] == "a violent storm" @@ -358,22 +421,64 @@ def _write_camera_assets(poses: Path, intrinsics: Path, *, frames: int = 32) -> ) +@dataclass(slots=True) +class _LegacyBridge: + mapping: LingbotInputMapping + canonicalizer: InputCanonicalizer + + +def _legacy_bridge( + tmp_path: Path, + *, + adapter: LingbotDemoAdapter, + camera_source: str = "trace", + text_events: dict[str, str] | None = None, +) -> _LegacyBridge: + replay_inputs = LingbotReplayInputs( + prompt="drive through a city", + first_frame_path=tmp_path / "image.jpg", + camera_poses_path=tmp_path / "poses.npy", + camera_intrinsics_path=tmp_path / "intrinsics.npy", + total_blocks=2, + ) + if camera_source == "events": + trace = adapter.create_input_mapping(replay_inputs).camera_trace + mapping = adapter.create_live_input_mapping( + fps=replay_inputs.fps, + base_intrinsics=trace.intrinsics[0], + world_scale=trace.world_scale or 1.0, + prompt=replay_inputs.prompt, + text_event_prompts=text_events, + ) + else: + mapping = adapter.create_input_mapping( + replay_inputs, + text_event_prompts=text_events, + ) + converters: list[Any] = [KeyboardToCameraCommand()] + if text_events: + converters.append(TextEventSelection()) + return _LegacyBridge( + mapping=mapping, + canonicalizer=InputCanonicalizer(converters), + ) + + def _legacy_step( scenario: PreparedScenario, + legacy: _LegacyBridge, *, step_index: int, frame_start: int, num_frames: int, inputs: UserInputs | None = None, ) -> InferenceInput: - mapping = scenario.mapping - assert isinstance(mapping, LingbotInputMapping) window = TimeWindow( start_s=frame_start / 16, end_s=(frame_start + num_frames) / 16, ) - return mapping.map_step_inputs( - canonical_inputs=scenario.canonicalizer.canonicalize( + return legacy.mapping.map_step_inputs( + canonical_inputs=legacy.canonicalizer.canonicalize( scenario.user_inputs if inputs is None else inputs, window=window, source_schema=scenario.source_schema, From 94f1b127c22c2d6c8d9e227dc136375628ea47b4 Mon Sep 17 00:00:00 2001 From: Jesse Archer Date: Tue, 11 Aug 2026 07:56:45 +0000 Subject: [PATCH 3/5] Decouple LingBot WebRTC startup from legacy input mapping Update the LingBot WebRTC inference-session startup path so shared demo sessions only require an initialized rollout, not the legacy input mapping. Clarify the remaining direct-session compatibility path and add assertions that the migrated LingBot WebRTC demo uses a mapping-free shared scenario. --- .../lingbot/lingbot/webrtc/session.py | 20 +++++++++--------- integrations/lingbot/tests/test_demo_api.py | 3 +++ .../tests/test_webrtc_session_branch.py | 21 ++++++------------- 3 files changed, 19 insertions(+), 25 deletions(-) diff --git a/integrations/lingbot/lingbot/webrtc/session.py b/integrations/lingbot/lingbot/webrtc/session.py index 03c1d2b90..57f0a4cdc 100644 --- a/integrations/lingbot/lingbot/webrtc/session.py +++ b/integrations/lingbot/lingbot/webrtc/session.py @@ -631,15 +631,15 @@ async def trigger_event( async def start_inference_session(self) -> LingbotWebRTCInferenceSession: """Return an ``InferenceSession`` view of the current rollout. - The shared manager canonicalizes raw key and text events and maps them - into per-step model inputs before stepping the session. + Shared demo providers prepare per-step model inputs before handing them + to this session. The legacy direct WebRTC path may still expose + ``input_mapping``/``input_canonicalizer`` to the shared manager, but + starting a session only requires an initialized rollout. """ if self._closed: raise LingbotRuntimeError("Runtime is closed.") - if self._input_mapping is None: - raise LingbotRuntimeError( - "Runtime input mapping is not initialized; reset the rollout first." - ) + if not self._is_runtime_initialized(): + raise LingbotRuntimeError("Runtime is not initialized.") return LingbotWebRTCInferenceSession(runtime=self) @property @@ -689,7 +689,7 @@ def _build_input_layers_sync(self, text_events: tuple[TextEventSpec, ...]) -> No self._input_mapping.set_base_prompt(self._prompt or "") def _next_step_request_sync(self) -> StepRequest: - """Describe the next mapped-input chunk for the session branch.""" + """Describe the next provider-prepared chunk for the session branch.""" if self._model_session is None: raise LingbotRuntimeError("Runtime is not initialized.") step_index = self._model_session.step_index @@ -1135,8 +1135,8 @@ def _generate_from_camera_inputs( ) -> StepResult: """Generate one chunk from an already-resolved camera trajectory. - Shared by the segment path and the mapped-input session path so both - reach the model through identical conditioning. + Shared by the segment path and the provider-prepared session path so + both reach the model through identical conditioning. """ if self._pipeline is None or self._model_session is None: raise LingbotRuntimeError("Runtime is not initialized.") @@ -1235,7 +1235,7 @@ class LingbotWebRTCInferenceSession: The rollout itself is owned by :class:`LingbotInferenceRuntime`; this only adapts it to the runtime-API stepping surface so the shared manager can - drive it with mapped inputs. + drive it with provider-prepared inputs. """ def __init__(self, *, runtime: LingbotInferenceRuntime) -> None: diff --git a/integrations/lingbot/tests/test_demo_api.py b/integrations/lingbot/tests/test_demo_api.py index d2cbd14db..07e1f57f5 100644 --- a/integrations/lingbot/tests/test_demo_api.py +++ b/integrations/lingbot/tests/test_demo_api.py @@ -707,6 +707,9 @@ def test_lingbot_webrtc_demo_uses_shared_manager_with_model_config( assert manager._shared_spec.input_mode == "keyboard-driving" assert isinstance(manager._shared_spec.output, WebRTCOutputSpec) assert manager._shared_scenario is not None + assert manager._shared_scenario.mapping is None + assert manager._shared_scenario.canonicalizer.converters == () + assert manager._needs_legacy_segment_metadata() is False provider = manager._shared_adapter.create_model_input_provider( manager._shared_spec, manager._shared_scenario, diff --git a/integrations/lingbot/tests/test_webrtc_session_branch.py b/integrations/lingbot/tests/test_webrtc_session_branch.py index 1a37d60f6..386de7b06 100644 --- a/integrations/lingbot/tests/test_webrtc_session_branch.py +++ b/integrations/lingbot/tests/test_webrtc_session_branch.py @@ -1,12 +1,12 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 -"""The manager's ``InferenceSession`` branch must preserve camera controls. +"""The manager's legacy ``InferenceSession`` branch must preserve camera controls. -The session branch buffers raw events, canonicalizes them over the chunk -window, and maps them into per-step ``InferenceInput``. The resulting camera -trajectory must match the direct resampler/integrator reference, or moving -LingBot's live path onto the runtime API would silently change how it drives. +The old direct WebRTC session branch buffers raw events, canonicalizes them +over the chunk window, and maps them into per-step ``InferenceInput``. The +resulting camera trajectory must match the direct resampler/integrator +reference while this compatibility path remains available. """ from __future__ import annotations @@ -457,15 +457,6 @@ def finalize( ), ) runtime._model_session._cache = object() - runtime._input_canonicalizer = InputCanonicalizer( - [KeyboardToCameraCommand(), TextEventSelection()] - ) - runtime._input_mapping = LingbotInputMapping( - fps=_FPS, - base_intrinsics=_BASE_INTRINSICS, - world_scale=1.0, - text_event_prompts={}, - ) try: inference_session = asyncio.run(runtime.start_inference_session()) @@ -502,5 +493,5 @@ def test_session_start_requires_an_initialized_rollout() -> None: runtime = LingbotInferenceRuntime(config=LingbotRuntimeConfig(device="cpu")) - with pytest.raises(LingbotRuntimeError, match="input mapping is not initialized"): + with pytest.raises(LingbotRuntimeError, match="Runtime is not initialized"): asyncio.run(runtime.start_inference_session()) From 58eaa6181c2c771dd5569e1f28e21754dae4d3d7 Mon Sep 17 00:00:00 2001 From: Jesse Archer Date: Tue, 11 Aug 2026 08:05:37 +0000 Subject: [PATCH 4/5] Fix ruff errors --- integrations/lingbot/lingbot/demo/providers.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/integrations/lingbot/lingbot/demo/providers.py b/integrations/lingbot/lingbot/demo/providers.py index f95083a9f..9d44a1e97 100644 --- a/integrations/lingbot/lingbot/demo/providers.py +++ b/integrations/lingbot/lingbot/demo/providers.py @@ -13,6 +13,7 @@ import numpy as np import torch + from flashdreams.runtime import ( InferenceInput, InferenceInputSchema, @@ -104,9 +105,7 @@ def __post_init__(self) -> None: object.__setattr__(self, "base_intrinsics", None) return if self.base_intrinsics is None: - raise ValueError( - "Live LingBot provider inputs require base_intrinsics." - ) + raise ValueError("Live LingBot provider inputs require base_intrinsics.") intrinsics = torch.as_tensor(self.base_intrinsics, dtype=torch.float32).reshape( 4 ) @@ -460,6 +459,7 @@ def _require_open(self) -> None: if self._closed: raise RuntimeError("LingbotInputProvider is closed.") + def _metadata_int( metadata: Mapping[str, Any], name: str, From 0faa70cf1842f89670ee6d23ff5d9f1a63e87f3d Mon Sep 17 00:00:00 2001 From: Jesse Archer Date: Tue, 11 Aug 2026 08:23:27 +0000 Subject: [PATCH 5/5] Update LingBot input mapping test for shared provider path Adjust the event-driven LingBot scenario test to assert the migrated mapping-free shared provider behavior. The test now drives the scenario through LingbotInputProvider instead of expecting PreparedScenario.mapping to contain a legacy LingbotInputMapping. --- .../lingbot/tests/test_input_mapping.py | 38 +++++++++++-------- 1 file changed, 23 insertions(+), 15 deletions(-) diff --git a/integrations/lingbot/tests/test_input_mapping.py b/integrations/lingbot/tests/test_input_mapping.py index 87ffb0b63..d9003ebed 100644 --- a/integrations/lingbot/tests/test_input_mapping.py +++ b/integrations/lingbot/tests/test_input_mapping.py @@ -8,6 +8,7 @@ import numpy as np import pytest import torch +from lingbot.demo.providers import PROVIDER_INPUTS_METADATA_KEY, LingbotInputProvider from lingbot.demo.spec import resolve_text_event_prompts, resolve_user_input_events from lingbot.input_mapping import ( CAMERA_COMMAND, @@ -25,12 +26,14 @@ InferenceInput, InputCanonicalizer, StepRequest, + StepRequirements, TimeWindow, UserInputCapability, UserInputEvent, UserInputs, UserInputSchema, ) +from flashdreams.runtime.demo import UserInputWindow pytestmark = pytest.mark.ci_cpu @@ -384,7 +387,7 @@ def test_declared_modalities_match_what_the_converters_produce() -> None: assert not empty.supports(CAMERA_COMMAND) -def test_event_driven_scenario_builds_a_live_mapping(tmp_path: Path) -> None: +def test_event_driven_scenario_uses_shared_input_provider(tmp_path: Path) -> None: """A scenario can drive the camera from events instead of the pose trace.""" from lingbot.demo.adapter import LingbotDemoAdapter from lingbot.runtime import LINGBOT_MODEL_ID @@ -426,26 +429,31 @@ def test_event_driven_scenario_builds_a_live_mapping(tmp_path: Path) -> None: prepared = LingbotDemoAdapter().prepare_scenario(spec) - assert isinstance(prepared.mapping, LingbotInputMapping) - assert prepared.mapping.mapping_schema.consumes == (CAMERA_COMMAND, TEXT_EVENT) + assert prepared.mapping is None + assert prepared.canonicalizer.converters == () + assert PROVIDER_INPUTS_METADATA_KEY in prepared.metadata assert len(prepared.user_inputs.events) == 2 - # The declared source must actually cover the trace it carries, or the - # canonicalizer silently drops the keyboard converter. - canonical_schema = prepared.canonicalizer.canonical_schema(prepared.source_schema) - assert canonical_schema.supports(CAMERA_COMMAND) - assert canonical_schema.supports(TEXT_EVENT) request = _step_request(step_index=0, frame_start=0, num_frames=4) assert request.user_input_window is not None - step_inputs = prepared.mapping.map_step_inputs( - canonical_inputs=prepared.canonicalizer.canonicalize( - prepared.user_inputs, - window=request.user_input_window, - source_schema=prepared.source_schema, + provider = LingbotInputProvider( + scenario=prepared, + inference_input_schema=LingbotDemoAdapter().inference_input_schema, + ) + step = provider.prepare_step( + request=StepRequirements( + step_index=request.step_index, + input_frame_count=4, + metadata=request.metadata, + ), + user_window=UserInputWindow( + start_s=request.user_input_window.start_s, + end_s=request.user_input_window.end_s, + inputs=prepared.user_inputs, ), - inference_input=InferenceInput(), - request=request, ) + assert step.inference_input is not None + step_inputs = step.inference_input poses = step_inputs.step[FIELD_CAMERA_TRAJECTORY] assert poses.shape == (4, 4, 4) assert not torch.allclose(poses[0], poses[-1])