diff --git a/src/lerobot/rollout/configs.py b/src/lerobot/rollout/configs.py index 1195b6180..8e8c06e65 100644 --- a/src/lerobot/rollout/configs.py +++ b/src/lerobot/rollout/configs.py @@ -241,8 +241,9 @@ class RolloutConfig: duration: float = 0.0 # 0 = infinite (24/7 mode) # Interactive session: control the rollout from stdin with chat-style # commands (/start, /reset, /stop) while hardware and policy stay warm. - # The robot does not move until /start is received. Currently limited to - # --strategy.type=base. + # The robot does not move until /start is received, and console logs are + # muted while the session runs so they don't interleave with the prompt. + # Currently limited to --strategy.type=base. interactive: bool = False interpolation_multiplier: int = 1 device: str | None = None diff --git a/src/lerobot/rollout/inference/base.py b/src/lerobot/rollout/inference/base.py index f269aa5fe..cad1753e4 100644 --- a/src/lerobot/rollout/inference/base.py +++ b/src/lerobot/rollout/inference/base.py @@ -87,3 +87,8 @@ class InferenceEngine(abc.ABC): def failed(self) -> bool: """True if an unrecoverable error occurred in the backend.""" return False + + @property + def failure_traceback(self) -> str | None: + """Formatted traceback of the unrecoverable error, when ``failed`` is True.""" + return None diff --git a/src/lerobot/rollout/inference/rtc.py b/src/lerobot/rollout/inference/rtc.py index c21e9f787..ed32de5d8 100644 --- a/src/lerobot/rollout/inference/rtc.py +++ b/src/lerobot/rollout/inference/rtc.py @@ -147,6 +147,7 @@ class RTCInferenceEngine(InferenceEngine): self._compile_warmup_done = Event() self._shutdown_event = Event() self._rtc_error = Event() + self._failure_traceback: str | None = None self._global_shutdown_event = shutdown_event self._rtc_thread: Thread | None = None @@ -193,6 +194,15 @@ class RTCInferenceEngine(InferenceEngine): """True if the RTC background thread exited due to an unrecoverable error.""" return self._rtc_error.is_set() + @property + def failure_traceback(self) -> str | None: + """Traceback captured when the RTC thread died (see ``failed``). + + Kept on the engine so consumers that mute console logging (the + interactive session) can still surface the fatal error. + """ + return self._failure_traceback + @property def action_queue(self) -> ActionQueue | None: """The shared action queue between the RTC thread and the main loop.""" @@ -390,8 +400,9 @@ class RTCInferenceEngine(InferenceEngine): time.sleep(_RTC_IDLE_SLEEP_S) except Exception as e: + self._failure_traceback = traceback.format_exc() logger.error("Fatal error in RTC thread: %s", e) - logger.error(traceback.format_exc()) + logger.error(self._failure_traceback) self._rtc_error.set() # Unblock any warmup waiters so the main loop doesn't spin forever self._compile_warmup_done.set() diff --git a/src/lerobot/rollout/interactive.py b/src/lerobot/rollout/interactive.py index 52b3ed897..a9155679b 100644 --- a/src/lerobot/rollout/interactive.py +++ b/src/lerobot/rollout/interactive.py @@ -33,6 +33,15 @@ which every strategy control loop already polls as propagate through the linked event's parent, so Ctrl-C behaves exactly as in non-interactive runs. +While the session runs, console log handlers are muted (including +non-propagating library loggers like ``transformers``) and Python warnings +are suppressed, so system output does not interleave with the chat prompt; +only the session's own output is shown. File log handlers are unaffected, +and console logging resumes when the session ends (so teardown logs are +visible). A fatal inference-engine error is still surfaced: the session +prints the engine's captured traceback. Run without ``--interactive`` to +see the full live log output. + The command table is intentionally a name → handler mapping so future commands (``/subtask``, ``/ask`` — see the language-runtime work in PR #4183/#4234) can be registered without restructuring the parser or the @@ -46,6 +55,7 @@ import os import select import sys import time +import warnings from collections.abc import Callable from dataclasses import dataclass from threading import Event, Thread @@ -62,6 +72,35 @@ logger = logging.getLogger(__name__) _BANNER_RULE = "─" * 60 +def _mute_console_log_handlers() -> list[tuple[logging.Handler, int]]: + """Mute console log handlers for the interactive session. + + System logs (policy, robot, control loop) contend with the chat prompt + for the terminal, so raise every console handler above ``CRITICAL`` + while the session runs. All loggers are covered, not just the root: + libraries like ``transformers`` and ``datasets`` attach their own + stderr handlers with ``propagate=False``. File handlers are left + untouched — anyone who wants a persistent log can attach one — and the + previous levels are returned so :func:`_restore_log_handlers` can undo + the muting. + """ + loggers = [logging.getLogger()] + loggers += [lg for lg in logging.Logger.manager.loggerDict.values() if isinstance(lg, logging.Logger)] + muted = [] + for lg in loggers: + for handler in lg.handlers: + if isinstance(handler, logging.StreamHandler) and not isinstance(handler, logging.FileHandler): + muted.append((handler, handler.level)) + handler.setLevel(logging.CRITICAL + 1) + return muted + + +def _restore_log_handlers(muted: list[tuple[logging.Handler, int]]) -> None: + """Restore handler levels changed by :func:`_mute_console_log_handlers`.""" + for handler, level in muted: + handler.setLevel(level) + + class LinkedEvent(Event): """A ``threading.Event`` whose ``is_set`` also reflects a parent event. @@ -340,12 +379,16 @@ class InteractiveSession: def run(self) -> None: """Run the session until ``/stop``, EOF, engine failure, or a shutdown signal.""" play_sounds = self._ctx.runtime.cfg.play_sounds - self._print(self._render_banner()) - self._listener.start() + muted_handlers: list[tuple[logging.Handler, int]] = [] + saved_warning_filters = warnings.filters[:] try: + muted_handlers = _mute_console_log_handlers() + warnings.simplefilter("ignore") + self._print(self._render_banner()) + self._listener.start() while not self._global_shutdown.is_set(): if self._ctx.policy.inference.failed: - self._print("Inference engine failed — shutting down. See the log for the error.") + self._report_engine_failure() break if self._stop_requested.is_set(): break @@ -361,8 +404,20 @@ class InteractiveSession: self._wake.clear() finally: self._listener.stop() + # Restore before log_say so teardown logs are visible again. + _restore_log_handlers(muted_handlers) + warnings.filters[:] = saved_warning_filters log_say("Interactive session ended", play_sounds) + def _report_engine_failure(self) -> None: + """Surface a fatal engine error despite the muted console logging.""" + self._print("Inference engine failed — shutting down.") + failure_traceback = self._ctx.policy.inference.failure_traceback + if failure_traceback: + self._print(failure_traceback) + else: + self._print("Re-run without --interactive=true to see the error output.") + def _run_segment(self) -> None: """Execute one ``strategy.run`` segment until interrupted or finished.""" engine = self._ctx.policy.inference @@ -394,6 +449,7 @@ class InteractiveSession: def _reset_robot(self) -> None: """Pause inference and return the robot to its initial position.""" + self._print("Resetting — returning the robot to its initial position...") self._ctx.policy.inference.pause() log_say("Resetting robot to initial position", self._ctx.runtime.cfg.play_sounds) if self._ctx.hardware.initial_position: @@ -465,6 +521,7 @@ class InteractiveSession: f"{_BANNER_RULE}\n" "Interactive rollout session — the robot will NOT move until you type /start.\n" f"{self._render_help()}\n" + "System logs and warnings are muted during the session; they resume when it ends.\n" f"{_BANNER_RULE}" ) diff --git a/tests/test_interactive_rollout.py b/tests/test_interactive_rollout.py index 827e5b2dc..bfaab820b 100644 --- a/tests/test_interactive_rollout.py +++ b/tests/test_interactive_rollout.py @@ -17,6 +17,8 @@ from __future__ import annotations import contextlib +import io +import logging import os import time from threading import Event, Thread @@ -184,8 +186,6 @@ def test_stdin_listener_handler_errors_do_not_kill_reader(): def test_stdin_listener_blocking_fallback(): """Streams without a file descriptor (e.g. StringIO) use the blocking readline path.""" - import io - lines: list[str] = [] eof = Event() listener = StdinCommandListener(lines.append, on_eof=eof.set, stream=io.StringIO("/start\n\n/help\n")) @@ -207,6 +207,7 @@ def _make_session(input_stream, run_behavior=None): stop_event = LinkedEvent(parent) engine = MagicMock() engine.failed = False + engine.failure_traceback = None ctx = SimpleNamespace( runtime=SimpleNamespace( cfg=SimpleNamespace(play_sounds=False), @@ -351,11 +352,12 @@ def test_session_exits_on_parent_shutdown(): assert not thread.is_alive() -def test_session_stops_on_engine_failure(): +def test_session_stops_on_engine_failure(capsys): def failing_run(c): - # Mimic the RTC thread's fatal-error path: flag the failure and set - # the shutdown event the engine was built with (the LinkedEvent). + # Mimic the RTC thread's fatal-error path: flag the failure, capture + # the traceback, and set the engine's shutdown event (the LinkedEvent). c.policy.inference.failed = True + c.policy.inference.failure_traceback = "RuntimeError: boom-traceback" c.runtime.shutdown_event.set() with _pipe_stream() as (reader, _writer): @@ -364,8 +366,10 @@ def test_session_stops_on_engine_failure(): session._handle_line("/start") thread.join(timeout=2.0) assert not thread.is_alive() - # A failed engine ends the session instead of returning to idle. + # A failed engine ends the session instead of returning to idle, and + # the captured traceback is surfaced despite the muted console logs. strategy.return_to_initial_position.assert_not_called() + assert "boom-traceback" in capsys.readouterr().out def test_session_stops_on_engine_failure_while_idle(): @@ -447,6 +451,58 @@ def test_session_commands_via_stream(): assert strategy.run.call_count == 1 +def test_session_mutes_console_logging_and_restores_on_exit(): + import warnings + + root = logging.getLogger() + console_handler = logging.StreamHandler(io.StringIO()) + console_handler.setLevel(logging.INFO) + root.addHandler(console_handler) + # Libraries like transformers attach their own console handler with + # propagate=False; those must be muted too. + lib_logger = logging.getLogger("test_interactive_fake_lib") + lib_logger.propagate = False + lib_handler = logging.StreamHandler(io.StringIO()) + lib_handler.setLevel(logging.WARNING) + lib_logger.addHandler(lib_handler) + n_warning_filters = len(warnings.filters) + try: + with _pipe_stream() as (reader, _writer): + session, _strategy, _engine, _parent, _run_started = _make_session(reader) + thread = _start_session_thread(session) + assert _wait_for( + lambda: console_handler.level == logging.CRITICAL + 1 + and lib_handler.level == logging.CRITICAL + 1 + ) + session._handle_line("/stop") + thread.join(timeout=2.0) + assert console_handler.level == logging.INFO + assert lib_handler.level == logging.WARNING + assert len(warnings.filters) == n_warning_filters + finally: + root.removeHandler(console_handler) + lib_logger.removeHandler(lib_handler) + + +def test_session_does_not_mute_file_log_handlers(tmp_path): + root = logging.getLogger() + file_handler = logging.FileHandler(tmp_path / "session.log") + file_handler.setLevel(logging.INFO) + root.addHandler(file_handler) + try: + with _pipe_stream() as (reader, _writer): + session, _strategy, _engine, _parent, run_started = _make_session(reader) + thread = _start_session_thread(session) + session._handle_line("/start") + assert _wait_for(run_started.is_set) + assert file_handler.level == logging.INFO + session._handle_line("/stop") + thread.join(timeout=2.0) + finally: + root.removeHandler(file_handler) + file_handler.close() + + def test_session_drives_real_base_strategy(): """End-to-end with a real BaseStrategy control loop (only hardware/engine mocked).""" from lerobot.rollout import BaseStrategy, BaseStrategyConfig