feat: add serial console observe mode with fan-out streaming

Allow multiple clients sharing a lease to access the serial console
simultaneously. One client holds the exclusive write token, others
observe the output read-only with scrollback replay.

- Add StreamFanOut state machine and FanOutStreamMixin for drivers
  with exclusive physical streams (jumpstarter/streams/fanout.py)
- CLI: `j serial start-console --observe` for read-only console
- CLI: `j serial pipe --observe` for read-only pipe
- CLI: `j serial release-console` to force-release write token
- CLI: `j serial console-status` to show session info
- Byte-bounded ClientBuffer with drop-oldest overflow policy
- 64KB scrollback ring replayed atomically on observer attach

Signed-off-by: Benny Zlotnik <bzlotnik@redhat.com>
Assisted-by: claude-opus-4.6

Author: bennyz

python/packages/jumpstarter-cli-common/jumpstarter_cli_common/exceptions.py

@@ -20,6 +20,35 @@ def _append_details(base_message: str, details: str) -> str:
     return f"{base_message} Details: {details}" if details else base_message
 
 
+def _extract_console_in_use_message(text: str) -> str | None:
+    """Pull the user-facing exclusive-console message out of a wrapped error string."""
+    marker = "Console in use"
+    idx = text.find(marker)
+    if idx == -1:
+        return None
+    return text[idx:]
+
+
+def _exception_chain(exc: BaseException):
+    """Yield *exc* and related exceptions (groups, __cause__, then __context__)."""
+    seen: set[int] = set()
+
+    def walk(current: BaseException):
+        if id(current) in seen:
+            return
+        seen.add(id(current))
+        yield current
+        if isinstance(current, BaseExceptionGroup):
+            for child in current.exceptions:
+                yield from walk(child)
+        if current.__cause__ is not None:
+            yield from walk(current.__cause__)
+        if current.__context__ is not None and not current.__suppress_context__:
+            yield from walk(current.__context__)
+
+    yield from walk(exc)
+
+
 def _map_runtime_exception(exc: BaseException, message: str, message_lower: str) -> click.ClickException | None:
     if isinstance(exc, TimeoutError):
         timeout_hint = (
@@ -96,6 +125,9 @@ def _map_grpc_exception(exc: BaseException) -> click.ClickException | None:
     code, details = _extract_grpc_code_and_details(exc)
     details_lower = details.lower()
 
+    if console_msg := _extract_console_in_use_message(details):
+        return ClickExceptionRed(console_msg)
+
     if code == "DEADLINE_EXCEEDED":
         return ClickExceptionRed(
             _append_details(
@@ -155,14 +187,15 @@ def _map_common_exception(exc: BaseException) -> click.ClickException | None:
 
 
 def _map_cli_exception(exc: BaseException) -> click.ClickException | None:
-    if common_exc := _map_common_exception(exc):
-        return common_exc
-    if isinstance(exc, JumpstarterException):
-        return ClickExceptionRed(str(exc))
-    if isinstance(exc, KeyboardInterrupt):
-        return ClickExceptionRed("Cancelled by user.")
-    if isinstance(exc, click.ClickException):
-        return exc
+    for candidate in _exception_chain(exc):
+        if common_exc := _map_common_exception(candidate):
+            return common_exc
+        if isinstance(candidate, JumpstarterException):
+            return ClickExceptionRed(str(candidate))
+        if isinstance(candidate, KeyboardInterrupt):
+            return ClickExceptionRed("Cancelled by user.")
+        if isinstance(candidate, click.ClickException):
+            return candidate
     return None
 
 

python/packages/jumpstarter-cli-common/jumpstarter_cli_common/exceptions_test.py

@@ -262,3 +262,64 @@ def fn():
 
     # Original call + exactly one retry = 2 total.
     assert call_count == 2
+
+
+class _MockGrpcError(Exception):
+    def __init__(self, code_name: str, details: str):
+        super().__init__(details)
+        self._code_name = code_name
+        self._details = details
+
+    def code(self):
+        return type("Code", (), {"name": self._code_name})()
+
+    def details(self):
+        return self._details
+
+
+_WRAPPED_CONSOLE_IN_USE = (
+    "Unexpected <class 'jumpstarter.streams.fanout.ExclusiveSessionActive'>: "
+    "Console in use. Use --observe or release-console."
+)
+
+
+def test_handle_exceptions_maps_exclusive_session_active() -> None:
+    from jumpstarter.streams.fanout import ExclusiveSessionActive
+
+    @handle_exceptions
+    def fn():
+        raise ExclusiveSessionActive()
+
+    with pytest.raises(click.ClickException, match="Console in use") as exc_info:
+        fn()
+    assert "Unexpected" not in str(exc_info.value)
+
+
+def test_handle_exceptions_maps_wrapped_console_in_use_grpc_error() -> None:
+    from anyio import BrokenResourceError
+
+    @handle_exceptions
+    def fn():
+        raise BrokenResourceError from _MockGrpcError("UNKNOWN", _WRAPPED_CONSOLE_IN_USE)
+
+    with pytest.raises(click.ClickException, match="Console in use") as exc_info:
+        fn()
+    assert "Unexpected" not in str(exc_info.value)
+
+
+@pytest.mark.anyio
+async def test_async_handle_exceptions_maps_console_in_use_in_exception_group() -> None:
+    from anyio import BrokenResourceError
+
+    @async_handle_exceptions
+    async def fn():
+        try:
+            raise _MockGrpcError("UNKNOWN", _WRAPPED_CONSOLE_IN_USE)
+        except _MockGrpcError as grpc_exc:
+            wrapped = BrokenResourceError()
+            wrapped.__cause__ = grpc_exc
+            raise ExceptionGroup("unhandled errors in a TaskGroup", [wrapped]) from None
+
+    with pytest.raises(click.ClickException, match="Console in use") as exc_info:
+        await fn()
+    assert "Unexpected" not in str(exc_info.value)

python/packages/jumpstarter-cli/jumpstarter_cli/share.py

@@ -69,15 +69,7 @@ def share_list(config, lease: str):
     \b
         $ jmp share list my-lease
     """
-    leases = config.list_leases(only_active=True)
-    target = None
-    for l in leases.leases:
-        if l.name == lease:
-            target = l
-            break
-
-    if target is None:
-        raise click.ClickException(f"Lease {lease!r} not found")
+    target = config.get_lease(lease)
 
     if not target.shared_with:
         click.echo(f"Lease {lease} is not shared with anyone.")

python/packages/jumpstarter-cli/jumpstarter_cli/share_test.py

@@ -1,9 +1,6 @@
 import inspect
 from unittest.mock import Mock, patch
 
-import click
-import pytest
-
 from jumpstarter_cli.share import share_add, share_list, share_remove
 
 
@@ -43,45 +40,26 @@ def test_share_remove_calls_update_lease():
 
 def test_share_list_shows_shared_clients(capsys):
     lease_entry = Mock()
-    lease_entry.name = "my-lease"
     lease_entry.shared_with = ["alice", "bob"]
 
-    leases_result = Mock()
-    leases_result.leases = [lease_entry]
-
     config = Mock()
-    config.list_leases.return_value = leases_result
+    config.get_lease.return_value = lease_entry
 
     inspect.unwrap(share_list.callback)(config=config, lease="my-lease")
 
-    config.list_leases.assert_called_once_with(only_active=True)
+    config.get_lease.assert_called_once_with("my-lease")
     captured = capsys.readouterr()
     assert "my-lease" in captured.out
     assert "alice" in captured.out
     assert "bob" in captured.out
 
 
-def test_share_list_not_found():
-    leases_result = Mock()
-    leases_result.leases = []
-
-    config = Mock()
-    config.list_leases.return_value = leases_result
-
-    with pytest.raises(click.ClickException, match="not found"):
-        inspect.unwrap(share_list.callback)(config=config, lease="no-such-lease")
-
-
 def test_share_list_not_shared(capsys):
     lease_entry = Mock()
-    lease_entry.name = "my-lease"
     lease_entry.shared_with = []
 
-    leases_result = Mock()
-    leases_result.leases = [lease_entry]
-
     config = Mock()
-    config.list_leases.return_value = leases_result
+    config.get_lease.return_value = lease_entry
 
     inspect.unwrap(share_list.callback)(config=config, lease="my-lease")
 

python/packages/jumpstarter-cli/jumpstarter_cli/shell.py

@@ -520,12 +520,21 @@ async def _shell_with_signal_handling(  # noqa: C901
                                     lease, exporter_logs, config, command, tg.cancel_scope
                                 )
                             except BaseExceptionGroup as eg:
+                                # SIGINT cancels in-flight Dial, which is mapped to
+                                # ExporterUnreachableError. If we extract only that
+                                # error we swallow CancelledError and reacquire.
+                                if find_exception_in_group(eg, cancelled_exc_class):
+                                    exit_code = 2
+                                    break
                                 unreachable = find_exception_in_group(eg, ExporterUnreachableError)
                                 if unreachable is None:
                                     raise
                             except ExporterUnreachableError as exc:
                                 unreachable = exc
                             if unreachable is not None:
+                                if tg.cancel_scope.cancel_called:
+                                    exit_code = 2
+                                    break
                                 if lease.lease_ended:
                                     break  # lease expired naturally — exit cleanly
                                 if lease.lease_transferred:
@@ -550,31 +559,39 @@ async def _shell_with_signal_handling(  # noqa: C901
                                 _warn_about_expired_token(lease.name, selector)
                             break
             except BaseExceptionGroup as eg:
-                for exc in eg.exceptions:
-                    if isinstance(exc, TimeoutError):
-                        raise exc from None
-                unreachable_exc = find_exception_in_group(eg, ExporterUnreachableError)
-                if unreachable_exc:
-                    raise unreachable_exc from None
-                offline_exc = find_exception_in_group(eg, ExporterOfflineError)
-                if offline_exc:
-                    raise offline_exc from None
-                lease_exc = find_exception_in_group(eg, LeaseError)
-                if lease_exc:
-                    raise lease_exc from None
-                if lease_used is not None:
-                    if lease_used.lease_ended:
-                        # Lease expired naturally (e.g. during beforeLease hook)
-                        # - exit gracefully instead of showing a scary error
-                        pass
-                    elif lease_used.lease_transferred:
-                        raise ExporterOfflineError(
-                            "Lease has been transferred to another client. Session is no longer valid."
-                        ) from None
-                    else:
-                        raise ExporterOfflineError("Connection to exporter lost") from None
+                if find_exception_in_group(eg, cancelled_exc_class):
+                    token = getattr(config, "token", None)
+                    if lease_used and token:
+                        remaining = get_token_remaining_seconds(token)
+                        if remaining is not None and remaining <= 0:
+                            _warn_about_expired_token(lease_used.name, selector)
+                    exit_code = 2
                 else:
-                    raise
+                    for exc in eg.exceptions:
+                        if isinstance(exc, TimeoutError):
+                            raise exc from None
+                    unreachable_exc = find_exception_in_group(eg, ExporterUnreachableError)
+                    if unreachable_exc:
+                        raise unreachable_exc from None
+                    offline_exc = find_exception_in_group(eg, ExporterOfflineError)
+                    if offline_exc:
+                        raise offline_exc from None
+                    lease_exc = find_exception_in_group(eg, LeaseError)
+                    if lease_exc:
+                        raise lease_exc from None
+                    if lease_used is not None:
+                        if lease_used.lease_ended:
+                            # Lease expired naturally (e.g. during beforeLease hook)
+                            # - exit gracefully instead of showing a scary error
+                            pass
+                        elif lease_used.lease_transferred:
+                            raise ExporterOfflineError(
+                                "Lease has been transferred to another client. Session is no longer valid."
+                            ) from None
+                        else:
+                            raise ExporterOfflineError("Connection to exporter lost") from None
+                    else:
+                        raise
             except cancelled_exc_class:
                 # Check if cancellation was due to token expiry
                 token = getattr(config, "token", None)

python/packages/jumpstarter-cli/jumpstarter_cli/shell_test.py

@@ -1332,3 +1332,76 @@ async def fake_run(*_):
 
         assert exit_code == 0
         assert state["call_count"] == 3
+
+
+class TestRetryLoopUserInterrupt:
+    """SIGINT must exit, not reacquire, even if Dial surfaces as Unreachable."""
+
+    def _config_with_lease(self, lease):
+        config = _DummyConfig()
+
+        @asynccontextmanager
+        async def lease_async(
+            selector, exporter_name, lease_name, duration, portal,
+            acquisition_timeout, retry_timeout=None, dial_timeout=None,
+        ):
+            yield lease
+
+        config.lease_async = lease_async
+        return config
+
+    def _lease(self):
+        lease = Mock()
+        lease.release = True
+        lease.name = "test-lease"
+        lease.exporter_name = "test-exporter"
+        lease.retry_timeout = 10.0
+        lease.lease_ended = False
+        lease.lease_transferred = False
+        return lease
+
+    async def test_does_not_retry_when_cancel_scope_already_cancelled(self):
+        lease = self._lease()
+        config = self._config_with_lease(lease)
+        state = {"call_count": 0}
+
+        async def fake_run(*args):
+            state["call_count"] += 1
+            cancel_scope = args[4]
+            cancel_scope.cancel()
+            raise ExporterUnreachableError("dial cancelled")
+
+        with (
+            patch("jumpstarter_cli.shell._monitor_token_expiry", new_callable=AsyncMock),
+            patch("jumpstarter_cli.shell._run_shell_with_lease_async", side_effect=fake_run),
+        ):
+            exit_code = await _shell_with_signal_handling(
+                config, None, None, None, timedelta(minutes=1), False, (), None
+            )
+
+        assert exit_code == 2
+        assert state["call_count"] == 1
+
+    async def test_does_not_retry_when_unreachable_is_mixed_with_cancellation(self):
+        lease = self._lease()
+        config = self._config_with_lease(lease)
+        state = {"call_count": 0}
+        cancelled_exc_class = anyio.get_cancelled_exc_class()
+
+        async def fake_run(*_):
+            state["call_count"] += 1
+            raise BaseExceptionGroup(
+                "task group",
+                [ExporterUnreachableError("dial cancelled"), cancelled_exc_class()],
+            )
+
+        with (
+            patch("jumpstarter_cli.shell._monitor_token_expiry", new_callable=AsyncMock),
+            patch("jumpstarter_cli.shell._run_shell_with_lease_async", side_effect=fake_run),
+        ):
+            exit_code = await _shell_with_signal_handling(
+                config, None, None, None, timedelta(minutes=1), False, (), None
+            )
+
+        assert exit_code == 2
+        assert state["call_count"] == 1