feat: report bytecode cache status to the launcher on startup

bec_widgets/applications/startup_profiler.py

@@ -90,3 +90,69 @@ class StartupProfiler:
 
 
 startup_profiler = StartupProfiler(_ORIGIN)
+
+
+# --- cold-start (bytecode cache) detection ---------------------------------
+#
+# On the very first launch (or after a Python/env update) the interpreter has to
+# compile every imported module to bytecode, which dominates the "module imports"
+# stage — especially on NFS. We sample a few files from the heavy pure-Python
+# packages *before* importing them and report the cache hit-rate to the launcher,
+# so its banner can tell the user why the first start takes longer.
+
+_PROBE_PACKAGES = ("bec_widgets", "bec_lib", "pyqtgraph", "qtpy")
+
+
+def _bytecode_cache_status(max_files_per_pkg: int = 40) -> tuple[int, int]:
+    """Return ``(checked, cached)`` counts for sampled .py files of the heavy packages.
+
+    Uses ``importlib.util.find_spec`` (locates without executing the modules) and
+    ``cache_from_source``; capped per package so the probe stays cheap even on NFS.
+    """
+    import importlib.util
+
+    checked = cached = 0
+    for pkg in _PROBE_PACKAGES:
+        try:
+            spec = importlib.util.find_spec(pkg)
+        except (ImportError, ValueError):
+            continue
+        if spec is None or not spec.origin or not spec.origin.endswith(".py"):
+            continue
+        sampled = 0
+        for root, _dirs, files in os.walk(os.path.dirname(spec.origin)):
+            for name in files:
+                if not name.endswith(".py"):
+                    continue
+                try:
+                    cache = importlib.util.cache_from_source(os.path.join(root, name))
+                except (ValueError, NotImplementedError):
+                    continue
+                checked += 1
+                if os.path.exists(cache):
+                    cached += 1
+                sampled += 1
+                if sampled >= max_files_per_pkg:
+                    break
+            if sampled >= max_files_per_pkg:
+                break
+    return checked, cached
+
+
+def _report_bytecode_cache() -> None:
+    """Stream the cache status to the launcher banner. Never breaks startup."""
+    if launch_progress is None or not launch_progress.enabled:
+        return
+    try:
+        checked, cached = _bytecode_cache_status()
+        if not checked:
+            return
+        pct = round(100 * cached / checked)
+        launch_progress.emit_info(cold_start=pct < 50, bytecode_cached_pct=pct)
+    except Exception:  # pragma: no cover - diagnostics must never break startup
+        pass
+
+
+# Runs at first import, i.e. before main_app/companion_app perform their heavy
+# imports — the launcher learns about a cold start while those imports run.
+_report_bytecode_cache()

bec_widgets/utils/launch_progress.py

@@ -60,6 +60,12 @@ class LaunchProgressClient:
             }
         )
 
+    def emit_info(self, **fields: object) -> bool:
+        """Send an informational message (e.g. cold-start / bytecode-cache status)."""
+        payload: dict[str, object] = {"t": "info"}
+        payload.update(fields)
+        return self._send(payload)
+
     def emit_ready(self, total_ms: float | None = None) -> bool:
         payload: dict[str, object] = {"t": "ready"}
         if total_ms is not None:

tests/unit_tests/test_launch_progress.py

@@ -106,6 +106,36 @@ def test_client_streams_hello_stage_and_ready(monkeypatch, server):
     assert ready and ready[0]["total_ms"] == 27710.0
 
 
+def test_client_streams_info_message(monkeypatch, server):
+    _client_env(monkeypatch, server.path)
+    client = lp.LaunchProgressClient()
+    assert client.emit_info(cold_start=True, bytecode_cached_pct=12) is True
+
+    messages = server.read_lines()
+    info = [m for m in messages if m["t"] == "info"]
+    assert info and info[0]["cold_start"] is True
+    assert info[0]["bytecode_cached_pct"] == 12
+
+
+def test_bytecode_cache_probe_reports_and_streams(monkeypatch, server):
+    from bec_widgets.applications import startup_profiler as sp
+
+    checked, cached = sp._bytecode_cache_status()
+    # The probe must find real sampled files in this env and never exceed bounds.
+    assert checked > 0
+    assert 0 <= cached <= checked
+
+    _client_env(monkeypatch, server.path)
+    monkeypatch.setattr(sp, "launch_progress", lp.LaunchProgressClient())
+    sp._report_bytecode_cache()
+
+    messages = server.read_lines()
+    info = [m for m in messages if m["t"] == "info"]
+    assert info
+    assert isinstance(info[0]["cold_start"], bool)
+    assert 0 <= info[0]["bytecode_cached_pct"] <= 100
+
+
 def test_client_never_raises_on_bad_socket(monkeypatch):
     _client_env(monkeypatch, _short_socket_path("nonexistent"))
     client = lp.LaunchProgressClient()