fix(cli): install the stderr log handler at startup and prepare the run state before the TUI launches

strix/interface/main.py

@@ -44,7 +44,7 @@ from strix.interface.utils import (
 )
 from strix.llm.warmup import start_import_warmup, wait_for_import_warmup
 from strix.telemetry import posthog, report_error, scarf, set_scan_phase
-from strix.telemetry.logging import configure_dependency_logging
+from strix.telemetry.logging import setup_console_logging
 
 
 BEDROCK_MODEL_PREFIX = "bedrock/"
@@ -428,7 +428,7 @@ def main() -> None:
         _force_utf8_streams()
         asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
 
-    configure_dependency_logging()
+    setup_console_logging()
 
     if len(sys.argv) == 2 and sys.argv[1] in ("-h", "--help"):
         try:

strix/interface/tui/runtime.py

@@ -167,7 +167,6 @@ class GoTuiRuntime:
         if self.controller.setup_mode:
             self._setup_preflight = asyncio.create_task(self.check_setup_model())
             return self._setup_preflight
-        self.init_run_state()
         self.start_scan()
         return None
 
@@ -418,6 +417,8 @@ class GoTuiRuntime:
         prepare_task: asyncio.Task[None] | None = None
         process: asyncio.subprocess.Process | subprocess.Popen[bytes] | None = None
         try:
+            if not self.controller.setup_mode:
+                self.init_run_state()
             env = child_environment()
             env["STRIX_VERSION"] = package_version()
             command = self.binary_command()

strix/telemetry/logging.py

@@ -59,6 +59,7 @@ _NOISY_LIBS: tuple[str, ...] = (
 
 
 _HANDLER_TAG = "_strix_scan_handler"
+_STREAM_TAG = "_strix_stream_handler"
 
 
 # ``openai.agents`` is the openai-agents SDK's canonical logger root.
@@ -116,6 +117,48 @@ def _silence_urllib3_finalizer_noise() -> None:
     sys.unraisablehook = hook
 
 
+class _CurrentStderrHandler(logging.StreamHandler):  # type: ignore[type-arg]
+    def __init__(self, level: int) -> None:
+        logging.Handler.__init__(self, level)
+
+    @property
+    def stream(self) -> object:
+        return sys.stderr
+
+    @stream.setter
+    def stream(self, _value: object) -> None:
+        pass
+
+
+def _debug_enabled(debug: bool | None) -> bool:
+    if debug is not None:
+        return debug
+    return (os.environ.get("STRIX_DEBUG") or "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _stream_handler(debug: bool) -> logging.Handler:
+    handler = _CurrentStderrHandler(logging.DEBUG if debug else logging.ERROR)
+    handler.setFormatter(logging.Formatter(_FORMAT, datefmt=_DATEFMT))
+    handler.addFilter(_StrixContextFilter())
+    handler.addFilter(_StdoutQuietFilter())
+    setattr(handler, _STREAM_TAG, True)
+    return handler
+
+
+def _has_stream_handler(tracked: logging.Logger) -> bool:
+    return any(getattr(handler, _STREAM_TAG, False) for handler in tracked.handlers)
+
+
+def setup_console_logging(*, debug: bool | None = None) -> None:
+    configure_dependency_logging()
+    for name in _TRACKED_ROOTS:
+        tracked = logging.getLogger(name)
+        tracked.setLevel(logging.DEBUG)
+        tracked.propagate = False
+        if not _has_stream_handler(tracked):
+            tracked.addHandler(_stream_handler(_debug_enabled(debug)))
+
+
 def setup_scan_logging(run_dir: Path, *, debug: bool | None = None) -> Callable[[], None]:
     """Attach scan-scoped handlers; return a teardown callable.
 
@@ -134,38 +177,25 @@ def setup_scan_logging(run_dir: Path, *, debug: bool | None = None) -> Callable[
     """
     configure_dependency_logging()
 
-    if debug is None:
-        debug = (os.environ.get("STRIX_DEBUG") or "").strip().lower() in {
-            "1",
-            "true",
-            "yes",
-            "on",
-        }
+    debug = _debug_enabled(debug)
 
     run_dir.mkdir(parents=True, exist_ok=True)
     log_path = run_dir / "strix.log"
 
-    formatter = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
-    context_filter = _StrixContextFilter()
-
     file_handler = logging.FileHandler(log_path, mode="a", encoding="utf-8")
     file_handler.setLevel(logging.DEBUG)
-    file_handler.setFormatter(formatter)
-    file_handler.addFilter(context_filter)
+    file_handler.setFormatter(logging.Formatter(_FORMAT, datefmt=_DATEFMT))
+    file_handler.addFilter(_StrixContextFilter())
     setattr(file_handler, _HANDLER_TAG, True)
 
-    stream_handler = logging.StreamHandler()
-    stream_handler.setLevel(logging.DEBUG if debug else logging.ERROR)
-    stream_handler.setFormatter(formatter)
-    stream_handler.addFilter(context_filter)
-    stream_handler.addFilter(_StdoutQuietFilter())
-    setattr(stream_handler, _HANDLER_TAG, True)
-
     tracked_loggers = [logging.getLogger(name) for name in _TRACKED_ROOTS]
     for tracked in tracked_loggers:
         tracked.setLevel(logging.DEBUG)
         tracked.addHandler(file_handler)
-        tracked.addHandler(stream_handler)
+        if not _has_stream_handler(tracked):
+            stream_handler = _stream_handler(debug)
+            setattr(stream_handler, _HANDLER_TAG, True)
+            tracked.addHandler(stream_handler)
         tracked.propagate = False
 
     for name in _NOISY_LIBS:

tests/test_console_logging.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+import io
+import logging
+import sys
+from typing import TYPE_CHECKING
+
+import pytest
+
+from strix.telemetry import logging as strix_logging
+
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+    from pathlib import Path
+
+
+@pytest.fixture
+def clean_loggers() -> Iterator[None]:
+    tracked = [logging.getLogger(name) for name in strix_logging._TRACKED_ROOTS]
+    saved = [(logger.handlers[:], logger.level, logger.propagate) for logger in tracked]
+    for logger in tracked:
+        logger.handlers = []
+    try:
+        yield
+    finally:
+        for logger, (handlers, level, propagate) in zip(tracked, saved, strict=True):
+            logger.handlers = handlers
+            logger.setLevel(level)
+            logger.propagate = propagate
+
+
+def _stderr_handlers(logger: logging.Logger) -> list[logging.Handler]:
+    return [h for h in logger.handlers if getattr(h, strix_logging._STREAM_TAG, False)]
+
+
+@pytest.mark.usefixtures("clean_loggers")
+def test_console_logging_hides_warnings_and_shows_errors(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("STRIX_DEBUG", raising=False)
+    err = io.StringIO()
+    monkeypatch.setattr(sys, "stderr", err)
+    strix_logging.setup_console_logging()
+
+    logger = logging.getLogger("strix.llm.request_log")
+    logger.warning("llm_request outcome=error")
+    logger.error("boom")
+
+    out = err.getvalue()
+    assert "llm_request" not in out
+    assert "ERROR" in out
+    assert "strix.llm.request_log: boom" in out
+
+
+@pytest.mark.usefixtures("clean_loggers")
+def test_console_logging_follows_the_current_stderr(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("STRIX_DEBUG", raising=False)
+    first = io.StringIO()
+    monkeypatch.setattr(sys, "stderr", first)
+    strix_logging.setup_console_logging()
+    second = io.StringIO()
+    monkeypatch.setattr(sys, "stderr", second)
+
+    logging.getLogger("strix.test").error("after swap")
+
+    assert first.getvalue() == ""
+    assert "after swap" in second.getvalue()
+
+
+@pytest.mark.usefixtures("clean_loggers")
+def test_console_logging_is_idempotent() -> None:
+    strix_logging.setup_console_logging()
+    strix_logging.setup_console_logging()
+
+    for name in strix_logging._TRACKED_ROOTS:
+        assert len(_stderr_handlers(logging.getLogger(name))) == 1
+
+
+@pytest.mark.usefixtures("clean_loggers")
+def test_scan_logging_reuses_the_console_handler(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.delenv("STRIX_DEBUG", raising=False)
+    err = io.StringIO()
+    monkeypatch.setattr(sys, "stderr", err)
+    strix_logging.setup_console_logging()
+    teardown = strix_logging.setup_scan_logging(tmp_path)
+
+    strix_logger = logging.getLogger("strix")
+    assert len(_stderr_handlers(strix_logger)) == 1
+    logging.getLogger("strix.test").error("once")
+    assert err.getvalue().count("once") == 1
+    assert "once" in (tmp_path / "strix.log").read_text(encoding="utf-8")
+
+    teardown()
+    assert len(_stderr_handlers(strix_logger)) == 1
+    assert not any(isinstance(h, logging.FileHandler) for h in strix_logger.handlers)
+
+
+@pytest.mark.usefixtures("clean_loggers")
+def test_scan_logging_alone_still_attaches_a_stderr_handler(tmp_path: Path) -> None:
+    teardown = strix_logging.setup_scan_logging(tmp_path)
+    strix_logger = logging.getLogger("strix")
+    assert len(_stderr_handlers(strix_logger)) == 1
+    teardown()
+    assert _stderr_handlers(strix_logger) == []

tests/test_go_tui_runtime.py

@@ -197,7 +197,7 @@ async def _send_message(connection: socket.socket, message: dict[str, Any]) -> N
 
 
 @pytest.mark.asyncio
-async def test_runtime_does_not_initialize_or_scan_before_ready(
+async def test_runtime_prepares_state_before_launch_and_scans_only_after_ready(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     runtime_args = args()
@@ -241,7 +241,7 @@ async def test_runtime_does_not_initialize_or_scan_before_ready(
     try:
         hello = await _receive_message(child)
         assert hello["type"] == "hello"
-        assert calls == []
+        assert calls == ["state"]
         await _send_message(
             child,
             {
@@ -1027,7 +1027,7 @@ async def test_direct_launch_starts_the_prepared_scan_without_a_model_round_trip
 
     assert runtime._start_preparation() is None
 
-    assert order == ["state", "scan"]
+    assert order == ["scan"]
     assert runtime.controller.setup_mode is False
     assert runtime.controller.scan_state == "running"
 

tests/test_main_launch.py

@@ -3,14 +3,14 @@ from __future__ import annotations
 import argparse
 import importlib
 import sys
-from typing import TYPE_CHECKING, Any
+from types import SimpleNamespace
+from typing import Any
 
-
-if TYPE_CHECKING:
-    import pytest
+import pytest
 
 
 cli_main: Any = importlib.import_module("strix.interface.main")
+report_state_module: Any = importlib.import_module("strix.report.state")
 
 
 def _launch(monkeypatch: pytest.MonkeyPatch, *, needs_setup: bool) -> list[str]:
@@ -26,7 +26,7 @@ def _launch(monkeypatch: pytest.MonkeyPatch, *, needs_setup: bool) -> list[str]:
         calls.append("tui")
 
     monkeypatch.setattr(sys, "argv", ["strix", "--target", "https://example.com"])
-    monkeypatch.setattr(cli_main, "configure_dependency_logging", lambda: None)
+    monkeypatch.setattr(cli_main, "setup_console_logging", lambda: None)
     monkeypatch.setattr(cli_main, "start_import_warmup", lambda: None)
     monkeypatch.setattr(cli_main, "parse_arguments", lambda: args)
     monkeypatch.setattr(cli_main, "start_background_check", lambda: None)
@@ -51,3 +51,65 @@ def test_start_screen_launch_defers_the_model_check_to_the_tui(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     assert _launch(monkeypatch, needs_setup=True) == ["tui"]
+
+
+def test_direct_launch_with_a_bad_key_prints_the_panel_and_exits_before_the_tui(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    calls: list[str] = []
+    failure = cli_main.ModelConnectionError("openai/gpt-4o", RuntimeError("Error code: 401"))
+
+    async def warm_up_llm() -> None:
+        raise failure
+
+    monkeypatch.setattr(cli_main, "warm_up_llm", warm_up_llm)
+    monkeypatch.setattr(cli_main, "report_error", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(
+        cli_main,
+        "_print_model_connection_error",
+        lambda exc, model: calls.append(f"panel:{model}:{exc}"),
+    )
+    monkeypatch.setattr(cli_main, "persist_current", lambda: calls.append("persist"))
+    monkeypatch.setattr(cli_main, "prepare_run", lambda _args: calls.append("prepare"))
+    monkeypatch.setattr(cli_main, "set_scan_phase", lambda _phase: None)
+
+    with pytest.raises(SystemExit) as exit_info:
+        cli_main._bootstrap_scan(argparse.Namespace(non_interactive=False, needs_setup=False))
+
+    assert exit_info.value.code == 1
+    assert calls == ["panel:openai/gpt-4o:Error code: 401"]
+
+
+def test_tui_startup_failure_marks_the_prepared_run_failed(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls: list[str] = []
+    report_state = SimpleNamespace(cleanup=lambda status: calls.append(f"cleanup:{status}"))
+    args = argparse.Namespace(
+        non_interactive=False, needs_setup=False, run_name="run", fail_on=None
+    )
+
+    async def run_tui(_args: argparse.Namespace) -> None:
+        raise cli_main.InteractiveSetupUnavailableError("no sidecar")
+
+    monkeypatch.setattr(sys, "argv", ["strix", "--target", "https://example.com"])
+    monkeypatch.setattr(cli_main, "setup_console_logging", lambda: None)
+    monkeypatch.setattr(cli_main, "start_import_warmup", lambda: None)
+    monkeypatch.setattr(cli_main, "parse_arguments", lambda: args)
+    monkeypatch.setattr(cli_main, "start_background_check", lambda: None)
+    monkeypatch.setattr(cli_main, "prompt_update_if_available", lambda _console: False)
+    monkeypatch.setattr(cli_main, "check_docker_installed", lambda: None)
+    monkeypatch.setattr(cli_main, "pull_docker_image", lambda: None)
+    monkeypatch.setattr(cli_main, "validate_environment", lambda: None)
+    monkeypatch.setattr(cli_main, "wait_for_import_warmup", lambda: None)
+    monkeypatch.setattr(cli_main, "_bootstrap_scan", lambda _args: calls.append("bootstrap"))
+    monkeypatch.setattr(cli_main, "run_tui", run_tui)
+    monkeypatch.setattr(cli_main, "report_error", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(cli_main, "_print_error_panel", lambda *_args: calls.append("panel"))
+    monkeypatch.setattr(cli_main.posthog, "end", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(cli_main.scarf, "end", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(report_state_module, "get_global_report_state", lambda: report_state)
+
+    with pytest.raises(SystemExit) as exit_info:
+        cli_main.main()
+
+    assert exit_info.value.code == 1
+    assert calls == ["bootstrap", "panel", "cleanup:failed"]