fix(daemon): make build-staleness checks non-blocking on status hot paths

src/leapflow/daemon/service.py

@@ -39,7 +39,7 @@
 from leapflow.daemon.turn_admission import TurnAdmission
 from leapflow.engine import StreamEvent
 from leapflow.memory.protocol import MemoryQuery
-from leapflow.utils.build_info import capture_build_info, is_stale
+from leapflow.utils.build_info import StalenessMonitor, capture_build_info, is_stale
 
 logger = logging.getLogger(__name__)
 
@@ -70,8 +70,13 @@ def __init__(self, settings: Any, *, mock_host: bool = False) -> None:
         self._started_at = time.time()
         # Captured once at daemon startup so status() can tell a developer
         # this process is stale (source changed since it started) instead of
-        # a fixed behavior change looking like a code defect.
+        # a fixed behavior change looking like a code defect. The staleness
+        # check itself shells out to git, so it is wrapped in a non-blocking
+        # cache: status() must return promptly even while other background
+        # work (e.g. a deferred DB worker) is busy (see
+        # test_daemon_event_loop_blocking.py).
         self._build_info = capture_build_info()
+        self._build_staleness = StalenessMonitor(self._build_info)
         self._client_count: Callable[[], int] = lambda: 0
         self._client_leases: Callable[[], list[ClientLeaseSnapshot]] = lambda: []
         self._approval_coordinator = ApprovalCoordinator(
@@ -133,6 +138,7 @@ async def start(self) -> None:
                     self._observation = None
 
     async def shutdown(self) -> None:
+        self._build_staleness.cancel_pending()
         if self._ctx is None:
             return
         ctx = self._ctx
@@ -756,7 +762,12 @@ async def status(self, session_id: str = "") -> dict[str, Any]:
         self._approval_coordinator.prune_stale()
         clients = await asyncio.to_thread(self._safe_client_lease_summaries)
         host = await asyncio.to_thread(host_backend_status, ctx)
-        build_stale = await asyncio.to_thread(is_stale, self._build_info)
+        # Non-blocking: returns the last cached verdict (None on the very
+        # first call) and refreshes it in the background, never awaiting the
+        # git subprocess on this hot path. `is_stale` is looked up here (not
+        # bound at __init__ time) so tests can still
+        # `monkeypatch.setattr(service_module, "is_stale", ...)`.
+        build_stale = self._build_staleness.current(is_stale)
         return {
             "pid": os.getpid(),
             "profile": getattr(settings, "profile", "default"),

src/leapflow/dashboard/server.py

@@ -25,7 +25,7 @@
 from leapflow.dashboard.service import DaemonDataProvider, DashboardViewBuilder
 from leapflow.dashboard.templates import TemplateLibrary
 from leapflow.monitor.types import EVENT_ERROR, EVENT_FINDING, EVENT_HEARTBEAT, EVENT_WATCH_STATE
-from leapflow.utils.build_info import capture_build_info, is_stale
+from leapflow.utils.build_info import StalenessMonitor, capture_build_info, is_stale
 
 logger = logging.getLogger(__name__)
 
@@ -60,8 +60,11 @@ def __init__(
         self._upstream_task: Optional[asyncio.Task[None]] = None
         # Captured once at server startup so /api/server-info and the ViewSpec
         # meta can report whether this long-lived process is still fresh
-        # relative to the source tree (see leapflow.utils.build_info).
+        # relative to the source tree (see leapflow.utils.build_info). Wrapped
+        # in a non-blocking cache so a browser refresh never waits on a git
+        # subprocess.
         self._build_info = capture_build_info()
+        self._build_staleness = StalenessMonitor(self._build_info)
 
     # ── App wiring ─────────────────────────────────────────────────────────
 
@@ -95,6 +98,7 @@ async def serve(self) -> None:
             while True:
                 await asyncio.sleep(3600)
         finally:
+            self._build_staleness.cancel_pending()
             await runner.cleanup()
 
     # ── Auth ───────────────────────────────────────────────────────────────
@@ -160,8 +164,14 @@ async def _handle_server_info(self, request: Any) -> Any:
         return web.json_response(await self._server_info())
 
     async def _server_info(self) -> dict[str, Any]:
-        """Build the {build, stale} payload shared by the view meta and the endpoint."""
-        stale = await asyncio.to_thread(is_stale, self._build_info)
+        """Build the {build, stale} payload shared by the view meta and the endpoint.
+
+        Non-blocking: returns the last cached verdict and refreshes it in the
+        background once due, rather than awaiting the git subprocess on every
+        request. `is_stale` is looked up here (not bound at __init__ time) so
+        tests can still `monkeypatch.setattr(server_module, "is_stale", ...)`.
+        """
+        stale = self._build_staleness.current(is_stale)
         return {"build": self._build_info.to_dict(), "stale": stale}
 
     async def _handle_action(self, request: Any) -> Any:

src/leapflow/utils/build_info.py

@@ -20,13 +20,14 @@
 
 from __future__ import annotations
 
+import asyncio
 import hashlib
 import os
 import subprocess
 import time
 from dataclasses import dataclass
 from pathlib import Path
-from typing import List, Optional, Tuple
+from typing import Callable, List, Optional, Tuple
 
 from leapflow.version import __version__
 
@@ -147,4 +148,73 @@ def is_stale(captured: BuildInfo) -> Optional[bool]:
     return commit != captured.commit or digest != captured.dirty_digest
 
 
-__all__ = ["BuildInfo", "capture_build_info", "is_stale"]
+# Refresh cadence for :class:`StalenessMonitor`. Staleness is a "restart me"
+# hint for a developer, not a real-time signal, so a coarse cadence trades an
+# acceptable amount of freshness for zero risk of blocking a status call.
+_DEFAULT_STALENESS_TTL_S = 30.0
+
+
+class StalenessMonitor:
+    """Non-blocking, TTL-cached wrapper around a staleness check.
+
+    ``is_stale`` re-derives its verdict via a fresh subprocess call every
+    time, which is exactly right for correctness but wrong for a hot path:
+    daemon ``status()`` RPCs and dashboard view handlers must return promptly
+    even while other background work (e.g. a deferred DB worker) is busy, and
+    a caller-facing deadline can be shorter than a loaded git invocation.
+
+    :meth:`current` never awaits the check itself — it returns the
+    last-known verdict (``None`` = "unknown, still checking" before the
+    first refresh completes) and, once ``ttl_s`` has elapsed, schedules a
+    fire-and-forget background refresh via :meth:`refresh`. Callers that need
+    a deterministic, synchronous result (tests, a first-use warmup) can
+    ``await monitor.refresh(checker)`` directly.
+
+    ``checker`` is accepted as a parameter (defaulting to :func:`is_stale`)
+    rather than bound at construction time, so call sites that pass their own
+    module-level ``is_stale`` reference keep working with
+    ``monkeypatch.setattr(module, "is_stale", ...)``-style test doubles.
+    """
+
+    def __init__(self, captured: BuildInfo, ttl_s: float = _DEFAULT_STALENESS_TTL_S) -> None:
+        self._captured = captured
+        self._ttl_s = ttl_s
+        self._value: Optional[bool] = None
+        self._checked_at: float = 0.0
+        self._refresh_task: "Optional[asyncio.Task[Optional[bool]]]" = None
+
+    def current(self, checker: Callable[[BuildInfo], Optional[bool]] = is_stale) -> Optional[bool]:
+        """Return the last-known verdict; never blocks on ``checker``.
+
+        Schedules a background refresh once ``ttl_s`` has elapsed since the
+        last completed check (or immediately, on first use).
+        """
+        due = time.time() - self._checked_at >= self._ttl_s
+        if due and (self._refresh_task is None or self._refresh_task.done()):
+            try:
+                loop = asyncio.get_running_loop()
+            except RuntimeError:
+                loop = None
+            if loop is not None:
+                self._refresh_task = loop.create_task(self.refresh(checker))
+        return self._value
+
+    async def refresh(self, checker: Callable[[BuildInfo], Optional[bool]] = is_stale) -> Optional[bool]:
+        """Synchronously run ``checker`` off-thread and cache the verdict.
+
+        Safe to ``await`` directly when a caller needs a deterministic result
+        (e.g. right after startup, or in a test) instead of racing the
+        background refresh that :meth:`current` schedules.
+        """
+        self._value = await asyncio.to_thread(checker, self._captured)
+        self._checked_at = time.time()
+        return self._value
+
+    def cancel_pending(self) -> None:
+        """Cancel any in-flight background refresh; call during shutdown."""
+        task = self._refresh_task
+        if task is not None and not task.done():
+            task.cancel()
+
+
+__all__ = ["BuildInfo", "StalenessMonitor", "capture_build_info", "is_stale"]

tests/test_build_info.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import os
 import subprocess
 from pathlib import Path
@@ -114,6 +115,92 @@ def test_to_dict_is_plain_json_safe_and_coerces_none_to_empty_string(
     }
 
 
+# ── StalenessMonitor: non-blocking cache around is_stale ─────────────────────
+
+
+def _info(commit: str = "abc123") -> build_info.BuildInfo:
+    return build_info.BuildInfo(
+        version=__version__, commit=commit, dirty_digest="d1", pid=os.getpid(), started_at=0.0,
+    )
+
+
+def test_staleness_monitor_current_is_none_before_first_refresh_completes() -> None:
+    """The first call must return instantly, never blocking on ``checker``."""
+    monitor = build_info.StalenessMonitor(_info())
+
+    verdict = monitor.current(lambda captured: True)
+
+    assert verdict is None  # unknown: the background refresh has not run yet
+
+
+async def test_staleness_monitor_current_reflects_background_refresh_once_it_completes() -> None:
+    monitor = build_info.StalenessMonitor(_info())
+    monitor.current(lambda captured: True)  # schedules the background refresh
+
+    for _ in range(50):
+        if monitor.current(lambda captured: True) is not None:
+            break
+        await asyncio.sleep(0.01)
+
+    assert monitor.current(lambda captured: True) is True
+
+
+async def test_staleness_monitor_does_not_reschedule_refresh_within_ttl() -> None:
+    calls = 0
+
+    def _checker(captured: build_info.BuildInfo) -> Optional[bool]:
+        nonlocal calls
+        calls += 1
+        return False
+
+    monitor = build_info.StalenessMonitor(_info(), ttl_s=60.0)
+    await monitor.refresh(_checker)
+    assert calls == 1
+
+    # Within the TTL window, repeated current() calls must not schedule
+    # another background refresh.
+    for _ in range(5):
+        monitor.current(_checker)
+    await asyncio.sleep(0)
+    assert calls == 1
+
+
+async def test_staleness_monitor_refresh_is_synchronous_and_deterministic() -> None:
+    monitor = build_info.StalenessMonitor(_info())
+
+    verdict = await monitor.refresh(lambda captured: True)
+
+    assert verdict is True
+    assert monitor.current(lambda captured: True) is True
+
+
+def test_staleness_monitor_current_without_running_loop_degrades_gracefully() -> None:
+    """Called from sync code (no running loop), current() must not raise."""
+    monitor = build_info.StalenessMonitor(_info())
+
+    assert monitor.current(lambda captured: True) is None
+
+
+def test_staleness_monitor_cancel_pending_is_noop_without_pending_refresh() -> None:
+    monitor = build_info.StalenessMonitor(_info())
+
+    monitor.cancel_pending()  # must not raise when nothing is in flight
+
+
+async def test_staleness_monitor_cancel_pending_cancels_inflight_refresh() -> None:
+    async def _hang() -> Optional[bool]:
+        await asyncio.sleep(3600)
+        return True
+
+    monitor = build_info.StalenessMonitor(_info())
+    monitor._refresh_task = asyncio.ensure_future(_hang())
+
+    monitor.cancel_pending()
+
+    with pytest.raises(asyncio.CancelledError):
+        await monitor._refresh_task
+
+
 # ── _run_git: graceful degradation on subprocess failure ────────────────────
 
 

tests/test_daemon_rpc.py

@@ -538,6 +538,10 @@ async def test_runtime_service_status_reports_own_build_staleness(tmp_path, monk
     service = service_module.RuntimeLeapService(make_settings(str(tmp_path)), mock_host=True)
     captured_pid = service._build_info.pid
     monkeypatch.setattr(service_module, "is_stale", lambda info: True)
+    # status() itself is a non-blocking cache read (see
+    # test_daemon_event_loop_blocking.py); force one deterministic refresh so
+    # this assertion does not race the background task current() schedules.
+    await service._build_staleness.refresh(service_module.is_stale)
 
     status = await service.status()
 

tests/test_dashboard_launcher.py

@@ -315,6 +315,10 @@ async def test_handle_server_info_reports_captured_build_and_stale_verdict(
     # This process's own fingerprint check is irrelevant to the endpoint's
     # wiring; pin the verdict so the assertion is deterministic.
     monkeypatch.setattr(server_module, "is_stale", lambda info: True)
+    # _server_info() is a non-blocking cache read; force one deterministic
+    # refresh so the assertion below does not race the background task that
+    # current() would otherwise schedule.
+    await server._build_staleness.refresh(server_module.is_stale)
     request = SimpleNamespace(query={"token": "t"}, headers={})
 
     response = await server._handle_server_info(request)