test(e2e): stub /health so the host-badge push test isolates useHosts status

test_hosts_changed_frame_updates_host_badge failed at its first
assertion (before any hosts_changed frame): the badge read "status
unknown" instead of the stubbed "online". The test intercepts the
WS /v1/sessions/updates stream to keep liveOnline undefined, but the
open-session GET /health poll is a second, independent source of
host_online — and the real endpoint emits host_online: null for a
session it finds without a host binding. That null reaches
useSessionHostOnline as a live signal, which HostBadge treats as
authoritative "unknown", overriding the useHosts status the test drives.

Patch /health to drop the seeded session from the batch sessions map so
useSessionHostOnline stays undefined ("not observed yet") and the badge
falls back to the useHosts status field — matching the test's stated
intent and the existing snapshot/list route patches. HostBadge behavior
is unchanged; this only repairs the test's mock world, which had the
snapshot claiming host-bound while /health said otherwise.

Co-authored-by: Isaac

omnigent/host/connect.py

@@ -43,6 +43,8 @@ from omnigent.host.frames import (
     HostRemoveWorktreeFrame,
     HostRemoveWorktreeResultFrame,
     HostRunnerExitedFrame,
+    HostRunnerStatusFrame,
+    HostRunnerStatusResultFrame,
     HostStatFrame,
     HostStatResultFrame,
     HostStopRunnerFrame,
@@ -1191,6 +1193,37 @@ class HostProcess:
             status="stopped",
         )
 
+    def _handle_runner_status(
+        self,
+        frame: HostRunnerStatusFrame,
+    ) -> HostRunnerStatusResultFrame:
+        """Answer whether a runner's process is alive, dead, or unknown.
+
+        The host is the authoritative owner of runner liveness: it holds
+        the runner's :class:`subprocess.Popen`. A runner tracked with a
+        still-running process is ``alive`` (covers a runner that is still
+        booting — it is inserted at ``Popen`` time, before its tunnel
+        connects — so the server waits for it). A tracked-but-exited
+        process is ``dead``. A runner this host has no record of is
+        ``unknown`` — it was stopped (``_handle_stop`` popped it) or a
+        fresh post-restart host never spawned it; either way it will never
+        connect, so the server relaunches without waiting.
+
+        :param frame: The status query frame.
+        :returns: Result frame with ``alive`` / ``dead`` / ``unknown``.
+        """
+        handle = self._runners.get(frame.runner_id)
+        if handle is None:
+            status = "unknown"
+        elif handle.proc.poll() is None:
+            status = "alive"
+        else:
+            status = "dead"
+        return HostRunnerStatusResultFrame(
+            request_id=frame.request_id,
+            status=status,
+        )
+
     async def _watch_runner(self, runner_id: str) -> None:
         """Watch a spawned runner and report an unexpected exit.
 
@@ -2068,6 +2101,8 @@ class HostProcess:
             await ws.send(encode_host_frame(await self._handle_launch(frame)))
         elif isinstance(frame, HostStopRunnerFrame):
             await ws.send(encode_host_frame(self._handle_stop(frame)))
+        elif isinstance(frame, HostRunnerStatusFrame):
+            await ws.send(encode_host_frame(self._handle_runner_status(frame)))
         elif isinstance(frame, HostStatFrame):
             await ws.send(encode_host_frame(self._handle_stat(frame)))
         elif isinstance(frame, HostListDirFrame):

omnigent/host/frames.py

@@ -43,6 +43,8 @@ class HostFrameKind(str, Enum):
     STOP_RUNNER = "host.stop_runner"
     STOP_RUNNER_RESULT = "host.stop_runner_result"
     RUNNER_EXITED = "host.runner_exited"
+    RUNNER_STATUS = "host.runner_status"
+    RUNNER_STATUS_RESULT = "host.runner_status_result"
     STAT = "host.stat"
     STAT_RESULT = "host.stat_result"
     LIST_DIR = "host.list_dir"
@@ -203,6 +205,48 @@ class HostRunnerExitedFrame:
     error: str
 
 
+@dataclass
+class HostRunnerStatusFrame:
+    """Server → host: is this runner's process alive, dead, or unknown?
+
+    The host is the authoritative owner of runner-process liveness — it
+    holds each runner's :class:`subprocess.Popen`. The runner tunnel
+    only tells the server "connected right now"; it cannot distinguish a
+    runner that is still booting (will connect) from one that was stopped
+    or died when the host restarted (never will). The message-dispatch
+    path asks this before its connect grace so it waits for a runner that
+    is coming and relaunches immediately for one that is not.
+
+    :param request_id: Unique id for correlating the result, e.g.
+        ``"req_rs_1"``.
+    :param runner_id: Runner to query, e.g. ``"runner_abc123..."``.
+    """
+
+    request_id: str
+    runner_id: str
+
+
+@dataclass
+class HostRunnerStatusResultFrame:
+    """Host → server: liveness of a queried runner.
+
+    :param request_id: Correlates to the :class:`HostRunnerStatusFrame`,
+        e.g. ``"req_rs_1"``.
+    :param status: One of:
+
+        * ``"alive"`` — the host has this runner and its process is
+          running (booting or serving). The runner is coming; wait.
+        * ``"dead"`` — the host has this runner but its process has
+          exited. It will never connect; relaunch now.
+        * ``"unknown"`` — the host has no record of this runner (it was
+          stopped, or a fresh post-restart host never spawned it).
+          Relaunch now.
+    """
+
+    request_id: str
+    status: str
+
+
 @dataclass
 class HostStatFrame:
     """Server → host: stat a path on the host's filesystem.
@@ -582,6 +626,8 @@ HostFrame = (
     | HostStopRunnerFrame
     | HostStopRunnerResultFrame
     | HostRunnerExitedFrame
+    | HostRunnerStatusFrame
+    | HostRunnerStatusResultFrame
     | HostStatFrame
     | HostStatResultFrame
     | HostListDirFrame
@@ -693,6 +739,22 @@ def encode_host_frame(frame: HostFrame) -> str:
                 "error": frame.error,
             }
         )
+    if isinstance(frame, HostRunnerStatusFrame):
+        return _encode_payload(
+            {
+                "kind": HostFrameKind.RUNNER_STATUS.value,
+                "request_id": frame.request_id,
+                "runner_id": frame.runner_id,
+            }
+        )
+    if isinstance(frame, HostRunnerStatusResultFrame):
+        return _encode_payload(
+            {
+                "kind": HostFrameKind.RUNNER_STATUS_RESULT.value,
+                "request_id": frame.request_id,
+                "status": frame.status,
+            }
+        )
     if isinstance(frame, HostStatFrame):
         return _encode_payload(
             {
@@ -915,6 +977,10 @@ def _decode_known_host_frame(
             return _decode_stop_runner_result(msg)
         case HostFrameKind.RUNNER_EXITED:
             return _decode_runner_exited(msg)
+        case HostFrameKind.RUNNER_STATUS:
+            return _decode_runner_status(msg)
+        case HostFrameKind.RUNNER_STATUS_RESULT:
+            return _decode_runner_status_result(msg)
         case HostFrameKind.STAT:
             return _decode_stat(msg)
         case HostFrameKind.STAT_RESULT:
@@ -1034,6 +1100,32 @@ def _decode_runner_exited(msg: dict[str, Any]) -> HostRunnerExitedFrame:
     )
 
 
+def _decode_runner_status(msg: dict[str, Any]) -> HostRunnerStatusFrame:
+    """Decode a host.runner_status request frame.
+
+    :param msg: Decoded frame object.
+    :returns: Typed host.runner_status frame.
+    """
+    return HostRunnerStatusFrame(
+        request_id=_required_str(msg, "request_id"),
+        runner_id=_required_str(msg, "runner_id"),
+    )
+
+
+def _decode_runner_status_result(
+    msg: dict[str, Any],
+) -> HostRunnerStatusResultFrame:
+    """Decode a host.runner_status_result frame.
+
+    :param msg: Decoded frame object.
+    :returns: Typed host.runner_status_result frame.
+    """
+    return HostRunnerStatusResultFrame(
+        request_id=_required_str(msg, "request_id"),
+        status=_required_str(msg, "status"),
+    )
+
+
 def _decode_stat(msg: dict[str, Any]) -> HostStatFrame:
     """Decode a host.stat request frame.
 

omnigent/server/host_registry.py

@@ -157,6 +157,11 @@ class HostConnection:
     :param pending_stops: Per-``request_id`` futures for
         in-flight ``host.stop_runner`` requests. Resolved when
         the host sends ``host.stop_runner_result``.
+    :param pending_runner_status: Per-``request_id`` futures for
+        in-flight ``host.runner_status`` queries. Resolved when the
+        host sends ``host.runner_status_result``. Values carry the
+        single ``status`` field (``"alive"`` / ``"dead"`` /
+        ``"unknown"``).
     :param pending_stats: Per-``request_id`` futures for in-flight
         ``host.stat`` requests. Resolved when the host sends
         ``host.stat_result``. The dict values carry the full
@@ -206,6 +211,9 @@ class HostConnection:
     pending_stops: dict[str, asyncio.Future[dict[str, str | None]]] = field(
         default_factory=dict,
     )
+    pending_runner_status: dict[str, asyncio.Future[dict[str, str | None]]] = field(
+        default_factory=dict,
+    )
     pending_stats: dict[str, asyncio.Future[dict[str, Any]]] = field(
         default_factory=dict,
     )

omnigent/server/routes/host_tunnel.py

@@ -36,6 +36,7 @@ from omnigent.host.frames import (
     HostListWorktreesResultFrame,
     HostRemoveWorktreeResultFrame,
     HostRunnerExitedFrame,
+    HostRunnerStatusResultFrame,
     HostStatResultFrame,
     HostStopRunnerResultFrame,
     decode_host_frame,
@@ -483,6 +484,12 @@ async def _receive_loop(
                 await on_runner_exited(frame.runner_id, frame.error)
             continue
 
+        if isinstance(frame, HostRunnerStatusResultFrame):
+            status_future = conn.pending_runner_status.pop(frame.request_id, None)
+            if status_future is not None and not status_future.done():
+                status_future.set_result({"status": frame.status})
+            continue
+
         if isinstance(frame, HostStatResultFrame):
             stat_future = conn.pending_stats.pop(frame.request_id, None)
             if stat_future is not None and not stat_future.done():

omnigent/server/routes/sessions.py

@@ -608,6 +608,12 @@ _NATIVE_TERMINAL_ENSURE_FAILED_CODE = "native_terminal_ensure_failed"
 _NATIVE_POLICY_NOT_ENFORCED_CODE = "native_policy_not_enforced"
 _HOST_BOUND_RUNNER_CONNECT_GRACE_S = 10.0
 _HOST_RELAUNCH_RUNNER_CONNECT_TIMEOUT_S = 30.0
+# Wait budget for the host's ``host.runner_status`` reply. The host answers
+# from an in-memory dict (a ``Popen.poll()``), so the round-trip is just the
+# tunnel latency. Kept short: this gates the connect grace, and a slow/absent
+# reply falls through to the grace wait (the prior blind-wait behavior), so
+# the query can only make the cold path faster, never slower.
+_HOST_RUNNER_STATUS_TIMEOUT_S = 3.0
 _MANAGED_RESUMABLE_TUNNEL_STALE_S = 30.0
 # How often the runner-connect wait re-checks the crash-report store while
 # racing the event-driven connect signal. Small enough that conviction is
@@ -6160,6 +6166,148 @@ async def _get_runner_client(
     return cast("httpx.AsyncClient | None", get_runner_client())
 
 
+async def _query_host_runner_status(
+    host_conn: HostConnection,
+    host_registry: HostRegistry,
+    runner_id: str,
+) -> str | None:
+    """
+    Ask a host whether a runner's process is alive, dead, or unknown.
+
+    The host owns runner-process liveness (it holds the ``Popen``), so it
+    can answer the one question the server's tunnel registry cannot: is an
+    absent-from-the-tunnel runner still coming (booting) or gone for good
+    (stopped, crashed, or lost to a host restart)? Used before the connect
+    grace so the dispatch path waits only for a runner that is coming.
+
+    :param host_conn: Live host connection to query.
+    :param host_registry: Registry used to enqueue the outbound frame.
+    :param runner_id: Runner to ask about, e.g. ``"runner_abc123..."``.
+    :returns: ``"alive"``, ``"dead"``, or ``"unknown"`` from the host; or
+        ``None`` when the host didn't reply in time, the connection
+        dropped, or the host is too old to support the query. ``None``
+        means "no authoritative answer" — the caller falls back to the
+        plain connect grace, preserving the prior blind-wait behavior.
+    """
+    from omnigent.host.frames import HostRunnerStatusFrame, encode_host_frame
+
+    request_id = secrets.token_hex(8)
+    future: asyncio.Future[dict[str, str | None]] = asyncio.get_running_loop().create_future()
+    host_conn.pending_runner_status[request_id] = future
+    frame = encode_host_frame(HostRunnerStatusFrame(request_id=request_id, runner_id=runner_id))
+    try:
+        try:
+            host_registry.send_text(host_conn, frame)
+        except ConnectionError:
+            return None
+        result = await asyncio.wait_for(
+            future,
+            timeout=_HOST_RUNNER_STATUS_TIMEOUT_S,
+        )
+    except asyncio.TimeoutError:
+        return None
+    except Exception:  # noqa: BLE001
+        # Defensive: this query only ever *speeds up* the connect grace, so
+        # any unexpected failure (e.g. the future resolved with an error)
+        # must degrade to "no verdict" and fall back to the wait rather than
+        # break the message POST. CancelledError is a BaseException and still
+        # propagates, so the race helper's cancel/drain is unaffected.
+        _logger.warning(
+            "host.runner_status query for runner %s failed; falling back to grace",
+            runner_id,
+            exc_info=True,
+        )
+        return None
+    finally:
+        host_conn.pending_runner_status.pop(request_id, None)
+    return result.get("status")
+
+
+async def _wait_for_host_bound_runner_client(
+    session_id: str,
+    runner_router: RunnerRouter | None,
+    tunnel_registry: TunnelRegistry | None,
+    *,
+    runner_id: str,
+    timeout_s: float,
+    runner_exit_reports: RunnerExitReports | None,
+    host_conn: HostConnection,
+    host_registry: HostRegistry,
+) -> httpx.AsyncClient | None:
+    """
+    Wait for a host-bound runner to connect, ending early if the host
+    reports it already gone.
+
+    Races the connect grace (:func:`_wait_for_runner_client`) against a
+    one-shot ``host.runner_status`` query, because they answer different
+    questions and either can settle the outcome first:
+
+    * The runner connecting — or a crash report — resolves the wait exactly
+      as :func:`_wait_for_runner_client` does. This is ground truth and
+      always wins when it lands first.
+    * Concurrently, the host — the authoritative owner of runner-process
+      liveness — may report the runner ``dead`` or ``unknown`` (stopped,
+      crashed, or lost to a host restart). That means it will never
+      connect, so the wait ends immediately and the caller relaunches
+      without burning the rest of the grace.
+
+    Running the query *alongside* the wait rather than before it is what
+    keeps the query strictly a speed-up: a host that is too old to answer,
+    slow, or silent (verdict ``None`` / ``"alive"``) never shortcuts the
+    wait, so the connect grace runs its normal course with no added
+    latency.
+
+    :param session_id: Session/conversation identifier.
+    :param runner_router: The ``RunnerRouter`` instance, or ``None``.
+    :param tunnel_registry: The server's ``TunnelRegistry``, or ``None``.
+    :param runner_id: Runner id expected to connect.
+    :param timeout_s: Maximum seconds to wait for the connect.
+    :param runner_exit_reports: Crash-report store consulted by the
+        connect wait to abort early on a reported death.
+    :param host_conn: Live host connection to query for liveness.
+    :param host_registry: Registry used to enqueue the query frame.
+    :returns: The runner HTTP client if it connected, otherwise ``None``
+        (timed out, crash report, or host-confirmed dead/unknown).
+    """
+    connect_task = asyncio.ensure_future(
+        _wait_for_runner_client(
+            session_id,
+            runner_router,
+            tunnel_registry,
+            runner_id=runner_id,
+            timeout_s=timeout_s,
+            runner_exit_reports=runner_exit_reports,
+        )
+    )
+    status_task = asyncio.ensure_future(
+        _query_host_runner_status(host_conn, host_registry, runner_id)
+    )
+    try:
+        done, _pending = await asyncio.wait(
+            {connect_task, status_task},
+            return_when=asyncio.FIRST_COMPLETED,
+        )
+        # The connect settling is authoritative (client, timeout, or crash
+        # report) — the host's opinion no longer matters once it lands.
+        if connect_task in done:
+            return connect_task.result()
+        # Only the status query has resolved so far.
+        if status_task.result() in ("dead", "unknown"):
+            # Host confirms the runner will never connect — stop waiting.
+            return None
+        # No verdict ("alive" or an unavailable/too-old/slow host): let the
+        # connect grace run to its natural conclusion.
+        return await connect_task
+    finally:
+        outstanding = [t for t in (connect_task, status_task) if not t.done()]
+        for task in outstanding:
+            task.cancel()
+        if outstanding:
+            # Drain the cancelled task(s); return_exceptions swallows the
+            # CancelledError so cleanup never masks the real return/raise.
+            await asyncio.gather(*outstanding, return_exceptions=True)
+
+
 async def _wait_for_runner_client(
     session_id: str,
     runner_router: RunnerRouter | None,
@@ -20363,10 +20511,23 @@ def create_sessions_router(
                 runner_client = await _get_runner_client(session_id, runner_router)
         if runner_client is None and conv.host_id is not None:
             _tunnel_registry = getattr(request.app.state, "tunnel_registry", None)
+            _grace_host_reg = getattr(request.app.state, "host_registry", None)
+            _grace_host_conn = (
+                _grace_host_reg.get(conv.host_id) if _grace_host_reg is not None else None
+            )
             # A just-created host session already has a runner_id before
             # the runner's tunnel is registered. The Web UI can post the
             # first message during that gap; wait briefly for the pinned
-            # runner before treating it as dead and replacing it.
+            # runner before treating it as dead and replacing it — but end
+            # that wait early when the runner is not actually coming. The
+            # host owns runner-process liveness (it holds the Popen), so we
+            # race a ``host.runner_status`` query against the connect grace:
+            # a booting runner connects (or reads "alive") and we forward,
+            # while one that was stopped, crashed, or lost to a host restart
+            # reads "dead"/"unknown" and cuts the wait short so the relaunch
+            # below runs at once. A host that is offline, too old to answer,
+            # or slow yields no verdict and the grace runs its normal
+            # course, so the query only ever speeds up the cold path.
             if conv.runner_id is not None and _HOST_BOUND_RUNNER_CONNECT_GRACE_S > 0:
                 _logger.info(
                     "Waiting up to %.1fs for host-bound runner %s to register "
@@ -20375,14 +20536,28 @@ def create_sessions_router(
                     conv.runner_id,
                     session_id,
                 )
-                runner_client = await _wait_for_runner_client(
-                    session_id,
-                    runner_router,
-                    _tunnel_registry,
-                    runner_id=conv.runner_id,
-                    timeout_s=_HOST_BOUND_RUNNER_CONNECT_GRACE_S,
-                    runner_exit_reports=runner_exit_reports,
-                )
+                if _grace_host_conn is not None:
+                    runner_client = await _wait_for_host_bound_runner_client(
+                        session_id,
+                        runner_router,
+                        _tunnel_registry,
+                        runner_id=conv.runner_id,
+                        timeout_s=_HOST_BOUND_RUNNER_CONNECT_GRACE_S,
+                        runner_exit_reports=runner_exit_reports,
+                        host_conn=_grace_host_conn,
+                        host_registry=_grace_host_reg,
+                    )
+                else:
+                    # Host tunnel absent: no one to query, so this is the
+                    # plain connect grace (unchanged pre-existing behavior).
+                    runner_client = await _wait_for_runner_client(
+                        session_id,
+                        runner_router,
+                        _tunnel_registry,
+                        runner_id=conv.runner_id,
+                        timeout_s=_HOST_BOUND_RUNNER_CONNECT_GRACE_S,
+                        runner_exit_reports=runner_exit_reports,
+                    )
             # Runner is dead or still not spawned for a host-bound
             # session. Ask the host to launch one, then re-fetch the
             # runner client and wait briefly for it to connect before

tests/e2e_ui/sessions/test_hosts_changed_push.py

@@ -115,6 +115,44 @@ def _patch_session_list(page: Page, session_id: str) -> None:
     page.route(re.compile(r"/v1/sessions(\?|$)"), _handler)
 
 
+def _patch_health_drops_session(page: Page, session_id: str) -> None:
+    """Drop ``session_id`` from ``GET /health`` batch responses.
+
+    The badge draws ``host_online`` from two independent sources: the
+    ``WS /v1/sessions/updates`` stream (intercepted here) and the
+    open-session ``/health`` poll. The real ``/health`` always emits
+    ``host_online`` for a session it finds — ``null`` when the session
+    isn't host-bound — and that ``null`` reaches ``useSessionHostOnline``
+    as a live signal, which ``HostBadge`` treats as authoritative
+    "unknown" and renders over the ``useHosts`` status this test drives.
+
+    Dropping the id from the ``sessions`` map leaves it *absent* (not
+    ``null``), so ``useSessionHostOnline`` stays ``undefined`` — "not
+    observed yet" — and the badge falls back to the ``useHosts`` status
+    field, exactly as the test intends. This mirrors the snapshot/list
+    patches: the browser is placed in a host-bound view whose liveness
+    comes solely from the controlled ``useHosts`` payload.
+    """
+
+    def _handler(route):  # type: ignore[no-untyped-def]
+        req = route.request
+        if req.method != "GET" or urlparse(req.url).path != "/health":
+            route.continue_()
+            return
+        resp = route.fetch()
+        payload = resp.json()
+        sessions = payload.get("sessions") if isinstance(payload, dict) else None
+        if isinstance(sessions, dict):
+            sessions.pop(session_id, None)
+        route.fulfill(
+            status=200,
+            headers={**resp.headers, "content-type": "application/json"},
+            body=json.dumps(payload),
+        )
+
+    page.route(re.compile(r"/health(\?|$)"), _handler)
+
+
 # ---------------------------------------------------------------------------
 # Tests
 # ---------------------------------------------------------------------------
@@ -154,6 +192,7 @@ def test_hosts_changed_frame_updates_host_badge(
     # Register routes before navigation so they're active on first request.
     _patch_session_snapshot(page, session_id)
     _patch_session_list(page, session_id)
+    _patch_health_drops_session(page, session_id)
     page.route_web_socket(re.compile(r"/v1/sessions/updates"), _handle_ws)
 
     # Start with "online" so we can observe a transition to "offline".

tests/host/test_connect.py

@@ -31,6 +31,8 @@ from omnigent.host.frames import (
     HostListDirFrame,
     HostListDirResultFrame,
     HostRunnerExitedFrame,
+    HostRunnerStatusFrame,
+    HostRunnerStatusResultFrame,
     HostStatFrame,
     HostStatResultFrame,
     HostStopRunnerFrame,
@@ -746,6 +748,77 @@ def test_handle_stop_unknown_runner() -> None:
     assert "unknown runner" in (result.error or "")
 
 
+def test_handle_runner_status_alive_for_running_process(tmp_path: Path) -> None:
+    """
+    Verify ``_handle_runner_status`` reports ``alive`` for a tracked
+    runner whose process is still running.
+
+    This is the still-booting / still-serving case: the server must wait
+    for this runner's tunnel rather than relaunch a healthy process.
+    """
+    host = _make_host_process()
+    proc = subprocess.Popen(
+        ["sleep", "60"],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    try:
+        host._runners["runner_live"] = _RunnerHandle(
+            proc=proc, log_path=tmp_path / "runner-live.log"
+        )
+        result = host._handle_runner_status(
+            HostRunnerStatusFrame(request_id="req_rs", runner_id="runner_live")
+        )
+        assert isinstance(result, HostRunnerStatusResultFrame)
+        assert result.request_id == "req_rs"
+        assert result.status == "alive"
+    finally:
+        proc.terminate()
+        proc.wait()
+
+
+def test_handle_runner_status_dead_for_exited_process(tmp_path: Path) -> None:
+    """
+    Verify ``_handle_runner_status`` reports ``dead`` for a tracked
+    runner whose process has exited.
+
+    A tracked-but-exited runner will never connect its tunnel, so the
+    server must relaunch immediately instead of burning the connect
+    grace on it.
+    """
+    host = _make_host_process()
+    proc = subprocess.Popen(
+        ["true"],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    proc.wait()  # ensure the process has exited before we query
+    host._runners["runner_gone"] = _RunnerHandle(proc=proc, log_path=tmp_path / "runner-gone.log")
+    result = host._handle_runner_status(
+        HostRunnerStatusFrame(request_id="req_rs", runner_id="runner_gone")
+    )
+    assert isinstance(result, HostRunnerStatusResultFrame)
+    assert result.status == "dead"
+
+
+def test_handle_runner_status_unknown_for_untracked_runner() -> None:
+    """
+    Verify ``_handle_runner_status`` reports ``unknown`` for a runner
+    this host has no record of.
+
+    Covers a runner that was stopped (``_handle_stop`` popped it) and a
+    fresh post-restart host that never spawned it — the exact
+    host-restart case that used to strand the server on a full connect
+    grace. Both must read ``unknown`` so the server relaunches at once.
+    """
+    host = _make_host_process()
+    result = host._handle_runner_status(
+        HostRunnerStatusFrame(request_id="req_rs", runner_id="runner_never_seen")
+    )
+    assert isinstance(result, HostRunnerStatusResultFrame)
+    assert result.status == "unknown"
+
+
 def test_alive_runner_ids_cleans_dead(tmp_path: Path) -> None:
     """
     Verify that _alive_runner_ids removes dead processes and returns

tests/host/test_frames.py

@@ -25,6 +25,8 @@ from omnigent.host.frames import (
     HostRemoveWorktreeFrame,
     HostRemoveWorktreeResultFrame,
     HostRunnerExitedFrame,
+    HostRunnerStatusFrame,
+    HostRunnerStatusResultFrame,
     HostStatFrame,
     HostStatResultFrame,
     HostStopRunnerFrame,
@@ -390,6 +392,44 @@ def test_runner_exited_frame_missing_error_raises() -> None:
         decode_host_frame('{"kind": "host.runner_exited", "runner_id": "runner_abc123"}')
 
 
+def test_runner_status_frame_round_trip() -> None:
+    """
+    Verify HostRunnerStatusFrame survives encode → decode.
+
+    A garbled runner_id here would make the host answer about the wrong
+    runner, so the dispatch path could wait for (or relaunch) the wrong
+    one.
+    """
+    original = HostRunnerStatusFrame(
+        request_id="req_rs_1",
+        runner_id="runner_token_abc",
+    )
+    decoded = decode_host_frame(encode_host_frame(original))
+    assert isinstance(decoded, HostRunnerStatusFrame)
+    assert decoded.request_id == "req_rs_1"
+    assert decoded.runner_id == "runner_token_abc"
+
+
+def test_runner_status_result_frame_round_trip() -> None:
+    """
+    Verify HostRunnerStatusResultFrame survives encode → decode for each
+    verdict.
+
+    ``status`` is the entire signal the dispatch gate acts on — a lossy
+    round-trip would make the server wait when it should relaunch, or
+    vice versa.
+    """
+    for status in ("alive", "dead", "unknown"):
+        original = HostRunnerStatusResultFrame(
+            request_id="req_rs_1",
+            status=status,
+        )
+        decoded = decode_host_frame(encode_host_frame(original))
+        assert isinstance(decoded, HostRunnerStatusResultFrame)
+        assert decoded.request_id == "req_rs_1"
+        assert decoded.status == status
+
+
 def test_decode_unknown_kind_raises() -> None:
     """
     Verify that an unknown frame kind raises ValueError.

tests/server/integration/test_session_host_launch.py

@@ -31,6 +31,8 @@ from omnigent.host.frames import (
     HostHelloFrame,
     HostLaunchRunnerFrame,
     HostLaunchRunnerResultFrame,
+    HostRunnerStatusFrame,
+    HostRunnerStatusResultFrame,
     HostStatFrame,
     HostStatResultFrame,
     HostStopRunnerFrame,
@@ -347,6 +349,56 @@ async def _wait_for_launch(
     return None
 
 
+async def _answer_runner_status_then_wait_for_launch(
+    comm: ApplicationCommunicator,
+    *,
+    status: str,
+    budget_s: float,
+) -> HostLaunchRunnerFrame | None:
+    """Reply to a ``host.runner_status`` query, then return the launch frame.
+
+    Models the host-owned liveness race: reads the host's outbound frames
+    (skipping interleaved pings), answers the first
+    :class:`HostRunnerStatusFrame` with *status* so the dispatch gate's
+    query resolves, and then returns the first
+    :class:`HostLaunchRunnerFrame` the relaunch sends. Used to prove that a
+    ``dead``/``unknown`` verdict cuts the connect grace short — the launch
+    arrives well inside *budget_s* even though the grace is much longer.
+
+    :param comm: Connected host communicator.
+    :param status: Verdict to answer the query with (``"alive"`` /
+        ``"dead"`` / ``"unknown"``).
+    :param budget_s: Seconds to wait on each receive, e.g. ``2.0``.
+    :returns: The launch frame the relaunch sent, or ``None`` if none
+        arrived within the budget.
+    """
+    answered = False
+    try:
+        for _ in range(40):
+            output = await comm.receive_output(timeout=budget_s)
+            if output["type"] != "websocket.send":
+                continue
+            frame = decode_host_frame(output["text"])
+            if isinstance(frame, HostRunnerStatusFrame) and not answered:
+                answered = True
+                await comm.send_input(
+                    {
+                        "type": "websocket.receive",
+                        "text": encode_host_frame(
+                            HostRunnerStatusResultFrame(
+                                request_id=frame.request_id,
+                                status=status,
+                            )
+                        ),
+                    }
+                )
+            elif isinstance(frame, HostLaunchRunnerFrame):
+                return frame
+    except asyncio.TimeoutError:
+        return None
+    return None
+
+
 async def test_inline_launch_binds_runner_and_returns_host(
     client: httpx.AsyncClient,
     app: FastAPI,
@@ -855,6 +907,75 @@ async def test_stopped_host_session_message_relaunches_runner(
     )
 
 
+async def test_host_reports_runner_unknown_skips_connect_grace(
+    client: httpx.AsyncClient,
+    app: FastAPI,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A host verdict of ``unknown`` relaunches without burning the grace.
+
+    This is the host-restart / non-sticky-Stop case: the runner is gone
+    (the host has no record of it), so the pinned ``runner_id`` will never
+    connect. With a generously long connect grace, the ``host.runner_status``
+    query must race the wait and cut it short — the ``host.launch_runner``
+    relaunch has to arrive far inside the grace window, not after it.
+
+    The grace is set to 5s while the launch is expected within a 2s
+    per-receive budget: comfortably longer than the sub-second query
+    round-trip but far shorter than the full grace, so a regression that
+    reinstated the blind wait (ignoring the verdict) would blow the budget
+    and fail here.
+
+    Mutation check: make the dispatch gate ignore the ``dead``/``unknown``
+    verdict (always wait the full grace) and the launch arrives ~5s later —
+    ``_answer_runner_status_then_wait_for_launch`` times out and returns
+    ``None``, failing the assertion.
+    """
+    from omnigent.runtime import set_runner_client
+    from omnigent.server.routes import sessions as sessions_module
+
+    # Long grace: a blind wait would take this long; the verdict must beat it.
+    monkeypatch.setattr(sessions_module, "_HOST_BOUND_RUNNER_CONNECT_GRACE_S", 5.0)
+
+    comm = await _connect_host(app)
+    session = await _inline_launch_session(client, comm)
+    session_id = session["id"]
+
+    # No runner client resolves: the message path enters the grace/relaunch
+    # block, where the liveness race runs.
+    set_runner_client(None)
+    post_task = asyncio.create_task(
+        client.post(
+            f"/v1/sessions/{session_id}/events",
+            json={
+                "type": "message",
+                "data": {"role": "user", "content": [{"type": "input_text", "text": "hi"}]},
+            },
+        )
+    )
+    try:
+        launch_frame = await _answer_runner_status_then_wait_for_launch(
+            comm, status="unknown", budget_s=2.0
+        )
+    finally:
+        # No runner ever connects, so the post would otherwise ride the
+        # ~30s relaunch wait — cancel once we've seen the launch frame.
+        post_task.cancel()
+        # return_exceptions drains the cancelled POST without re-raising.
+        await asyncio.gather(post_task, return_exceptions=True)
+
+    assert launch_frame is not None, (
+        "an 'unknown' host verdict must cut the connect grace short and "
+        "relaunch immediately; no host.launch_runner arrived within the "
+        "budget, so the dispatch gate is still waiting out the full grace"
+    )
+    assert launch_frame.workspace == _WORKSPACE
+    # The relaunch mints a fresh runner_id (runner_id rotation itself is
+    # pinned by test_stopped_host_session_message_relaunches_runner); here
+    # the point is purely that the verdict cut the grace short.
+    assert launch_frame.binding_token, "relaunch frame should carry a fresh binding token"
+
+
 async def test_host_session_message_relaunches_offline_runner(
     client: httpx.AsyncClient,
     app: FastAPI,

tests/server/routes/test_runner_status_query.py

@@ -0,0 +1,188 @@
+"""Tests for ``_query_host_runner_status`` — the host-owned liveness query.
+
+The host owns runner-process liveness (it holds the ``Popen``). Before the
+message-dispatch connect grace, the server asks the host whether an
+absent-from-the-tunnel runner is still coming (``alive``) or gone for good
+(``dead`` / ``unknown``). A verdict of ``None`` means "no authoritative
+answer" (host too old, slow, or the connection dropped), and the caller
+falls back to the plain grace wait — so the query can only ever speed up
+the cold path, never slow it down.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from omnigent.host.frames import (
+    HostRunnerStatusFrame,
+    HostRunnerStatusResultFrame,
+    decode_host_frame,
+)
+from omnigent.server.routes.sessions import _query_host_runner_status
+
+pytestmark = pytest.mark.asyncio
+
+
+class _FakeHostConn:
+    """Minimal ``HostConnection`` stand-in for the runner-status query.
+
+    :param host_id: Host id, only used in error paths / logging.
+    """
+
+    def __init__(self, host_id: str = "host_test") -> None:
+        """Initialize with an empty pending-query map."""
+        self.host_id = host_id
+        self.pending_runner_status: dict[str, asyncio.Future[dict[str, str | None]]] = {}
+
+
+class _ReplyingRegistry:
+    """Registry stand-in that replies to the query with a fixed status.
+
+    On ``send_text`` it decodes the outbound frame, finds the matching
+    pending future on the connection, and resolves it with ``reply`` — the
+    same round-trip the real host tunnel performs, without a socket.
+
+    :param reply: Status to answer with (``"alive"`` / ``"dead"`` /
+        ``"unknown"``).
+    """
+
+    def __init__(self, conn: _FakeHostConn, reply: str) -> None:
+        """Record the connection to resolve against and the canned reply."""
+        self._conn = conn
+        self._reply = reply
+        self.sent: list[str] = []
+
+    def send_text(self, conn: _FakeHostConn, data: str) -> None:
+        """Decode the query and immediately resolve its pending future.
+
+        :param conn: Host connection the frame is bound for.
+        :param data: Encoded ``host.runner_status`` frame.
+        """
+        self.sent.append(data)
+        frame = decode_host_frame(data)
+        assert isinstance(frame, HostRunnerStatusFrame)
+        future = conn.pending_runner_status.get(frame.request_id)
+        if future is not None and not future.done():
+            future.set_result({"status": self._reply})
+
+
+class _SilentRegistry:
+    """Registry stand-in that sends but never replies (forces a timeout)."""
+
+    def __init__(self) -> None:
+        """Initialize with an empty send log."""
+        self.sent: list[str] = []
+
+    def send_text(self, conn: _FakeHostConn, data: str) -> None:
+        """Record the frame but leave the pending future unresolved."""
+        self.sent.append(data)
+
+
+class _BrokenRegistry:
+    """Registry stand-in whose ``send_text`` raises ``ConnectionError``."""
+
+    def send_text(self, conn: _FakeHostConn, data: str) -> None:
+        """Simulate a host connection that dropped before the send landed."""
+        raise ConnectionError("host connection lost")
+
+
+class _FaultingRegistry:
+    """Registry stand-in that resolves the pending future with an exception.
+
+    Models a receive loop that somehow completed the future with an error
+    rather than a status dict — the defensive path must map this to ``None``
+    rather than let it break the message POST.
+    """
+
+    def send_text(self, conn: _FakeHostConn, data: str) -> None:
+        """Resolve the query's pending future with an exception."""
+        frame = decode_host_frame(data)
+        assert isinstance(frame, HostRunnerStatusFrame)
+        future = conn.pending_runner_status.get(frame.request_id)
+        if future is not None and not future.done():
+            future.set_exception(RuntimeError("receive loop blew up"))
+
+
+@pytest.mark.parametrize("verdict", ["alive", "dead", "unknown"])
+async def test_query_returns_host_verdict(verdict: str) -> None:
+    """Each host verdict is returned verbatim to the caller.
+
+    ``alive`` drives "wait for the connect", ``dead`` / ``unknown`` drive
+    "relaunch now" — the dispatch gate depends on these passing through
+    unchanged.
+    """
+    conn = _FakeHostConn()
+    registry = _ReplyingRegistry(conn, verdict)
+
+    result = await _query_host_runner_status(conn, registry, "runner_x")  # type: ignore[arg-type]
+
+    assert result == verdict
+    # The outbound frame targeted the queried runner.
+    assert len(registry.sent) == 1
+    frame = decode_host_frame(registry.sent[0])
+    assert isinstance(frame, HostRunnerStatusFrame)
+    assert frame.runner_id == "runner_x"
+    # The pending entry is cleaned up on the reply path.
+    assert conn.pending_runner_status == {}
+
+
+async def test_query_times_out_to_none(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A host that never replies yields ``None`` (fall back to the grace).
+
+    ``None`` must not be read as "dead" — a slow or too-old host should
+    still get the benefit of the connect grace, so the query returning
+    ``None`` preserves the prior blind-wait behavior.
+    """
+    monkeypatch.setattr("omnigent.server.routes.sessions._HOST_RUNNER_STATUS_TIMEOUT_S", 0.05)
+    conn = _FakeHostConn()
+    registry = _SilentRegistry()
+
+    result = await _query_host_runner_status(conn, registry, "runner_slow")  # type: ignore[arg-type]
+
+    assert result is None
+    # The pending entry is cleaned up even on the timeout path.
+    assert conn.pending_runner_status == {}
+
+
+async def test_query_connection_error_returns_none() -> None:
+    """A dropped host connection yields ``None`` rather than raising.
+
+    The dispatch path treats ``None`` as "no verdict" and falls back to
+    the grace/relaunch flow; a raised error here would surface as a 500
+    on the message POST instead.
+    """
+    conn = _FakeHostConn()
+    registry = _BrokenRegistry()
+
+    result = await _query_host_runner_status(conn, registry, "runner_x")  # type: ignore[arg-type]
+
+    assert result is None
+    assert conn.pending_runner_status == {}
+
+
+async def test_query_future_exception_returns_none() -> None:
+    """A future resolved with an exception degrades to ``None``, not a raise.
+
+    Defensive contract: the query only ever speeds up the connect grace, so
+    an unexpected failure must fall back to the wait rather than surface as
+    a 500 on the message POST.
+    """
+    conn = _FakeHostConn()
+    registry = _FaultingRegistry()
+
+    result = await _query_host_runner_status(conn, registry, "runner_x")  # type: ignore[arg-type]
+
+    assert result is None
+    assert conn.pending_runner_status == {}
+
+
+async def test_runner_status_result_field_shape() -> None:
+    """The result frame carries exactly the ``status`` the gate reads.
+
+    A sanity check on the wire contract the query helper relies on: the
+    host answers with a single ``status`` string.
+    """
+    frame = HostRunnerStatusResultFrame(request_id="r", status="alive")
+    assert frame.status == "alive"