fix(cli): heal a zombie remote host daemon instead of polling it

Reuse for a `--server` target was PID-liveness only: a daemon process that is
alive but whose tunnel is gone for good was reused as-is. Every command then
waited out `wait_for_host_online` and failed with "did not come online within
30s", and the only way out was `omnigent host stop` — which is why that became
the standing remedy for a wedged host. Local-mode daemons have had this heal
for a while; remote ones were left out on the grounds that the remote server
owns tunnel health, but that covers transient drops, not a daemon that never
re-registers.

Extend the heal to background remote daemons, with the guard that matters:
tear down only on a definite "the server says offline". `_daemon_host_online`
folded an unreachable server into `False`, so reusing it here would let a VPN
drop or a network blip kill a perfectly healthy daemon. Split the probe into a
tri-state `_daemon_host_state` (online / offline / unknown) and heal only on
`offline`, never on `unknown`; `_daemon_host_online` stays as a thin wrapper so
local mode and the registration wait are unchanged.

The existing age gate still applies, so a daemon that is merely still
connecting is left alone, and foreground `omnigent host` is never touched.

Cost is one extra host-status GET on the remote reuse path; in `_ensure_backend`
it runs concurrently with the auth probe, which is the longer pole.

Co-authored-by: Isaac <no-reply@databricks.com>

omnigent/cli.py

@@ -2523,32 +2523,44 @@ def _normalize_daemon_target(server_url: str | None) -> str:
     return _LOCAL_DAEMON_MARKER if not server_url else server_url.rstrip("/")
 
 
-def _daemon_host_online(record: _HostDaemonRecord, *, timeout_s: float = 2.0) -> bool:
+_DaemonHostState: TypeAlias = Literal["online", "offline", "unknown"]
+
+
+def _daemon_host_state(
+    record: _HostDaemonRecord,
+    *,
+    timeout_s: float = 2.0,
+) -> _DaemonHostState:
     """
-    Probe whether a daemon's host is currently online on its server.
+    Probe a daemon's host registration, separating offline from unreachable.
 
     A daemon process being alive (PID check) does not mean its WebSocket
     tunnel to the Omnigent server is up: the server only reports the host
     ``online`` while a daemon holds an authenticated tunnel and has
     heartbeated within ``HOST_LIVENESS_TTL_S``. After a server restart,
     an ungraceful daemon death, or a flapping tunnel, the daemon can be a
-    "zombie" — alive but not registered. This probe distinguishes the two
-    so reuse can heal instead of polling a zombie until timeout.
+    "zombie" — alive but not registered.
+
+    ``"unknown"`` is deliberately distinct from ``"offline"``: we could not
+    complete the probe (no host id, unreachable server, a non-200, an
+    unparseable body), which says nothing about the daemon. Only a server that
+    answers and reports a non-online status is evidence the tunnel is really
+    gone — the difference between healing a zombie and killing a healthy
+    daemon over a network blip.
 
     :param record: Daemon record to probe.
     :param timeout_s: Per-request HTTP timeout in seconds, e.g. ``2.0``.
-    :returns: ``True`` only when the server reports the record's host id
-        as ``"online"``; ``False`` if the host id is unknown, the server
-        is unreachable, or the host reports offline.
+    :returns: ``"online"`` / ``"offline"`` when the server answered,
+        ``"unknown"`` when the probe could not be completed.
     """
     from omnigent.claude_native_bridge import url_component
 
     host_id = record.host_id or _load_existing_host_id()
     if host_id is None:
-        return False
+        return "unknown"
     base_url = _daemon_base_url(record)
     if base_url is None:
-        return False
+        return "unknown"
     result = _host_http_json(
         base_url=base_url,
         method="GET",
@@ -2557,8 +2569,21 @@ def _daemon_host_online(record: _HostDaemonRecord, *, timeout_s: float = 2.0) ->
         host_id=host_id,
     )
     if result.status_code != 200 or not isinstance(result.body, dict):
-        return False
-    return result.body.get("status") == "online"
+        return "unknown"
+    return "online" if result.body.get("status") == "online" else "offline"
+
+
+def _daemon_host_online(record: _HostDaemonRecord, *, timeout_s: float = 2.0) -> bool:
+    """
+    Probe whether a daemon's host is currently online on its server.
+
+    :param record: Daemon record to probe.
+    :param timeout_s: Per-request HTTP timeout in seconds, e.g. ``2.0``.
+    :returns: ``True`` only when the server reports the record's host id
+        as ``"online"``; ``False`` if the host id is unknown, the server
+        is unreachable, or the host reports offline.
+    """
+    return _daemon_host_state(record, timeout_s=timeout_s) == "online"
 
 
 def _daemon_registry_dir() -> Path:
@@ -2811,6 +2836,38 @@ def _daemon_tunnel_recovers(
     return False
 
 
+def _daemon_host_definitely_offline(
+    record: _HostDaemonRecord,
+    *,
+    grace_s: float = _DAEMON_RECONNECT_GRACE_S,
+) -> bool:
+    """
+    Return whether the server insists a daemon's host is offline.
+
+    Like :func:`_daemon_tunnel_recovers` this polls for up to *grace_s* to let
+    a daemon mid-reconnect re-register, but it answers the stricter question:
+    did the server actually tell us the host is offline? A probe we could not
+    complete (``"unknown"``) is never evidence — it usually means *we* cannot
+    reach the server, and tearing down a healthy remote daemon over the
+    caller's own network blip would be worse than the zombie we are hunting.
+
+    :param record: Daemon record to probe.
+    :param grace_s: Seconds to keep polling for recovery, e.g. ``5.0``.
+    :returns: ``True`` only if the host never reported online during the grace
+        window and the final answer was a definite ``"offline"``.
+    """
+    state = _daemon_host_state(record)
+    if state == "online":
+        return False
+    deadline = time.monotonic() + grace_s
+    while time.monotonic() < deadline:
+        time.sleep(0.5)
+        state = _daemon_host_state(record)
+        if state == "online":
+            return False
+    return state == "offline"
+
+
 def _daemon_host_identity_changed(record: _HostDaemonRecord) -> bool:
     """
     Return whether a daemon record belongs to a different current host id.
@@ -2908,11 +2965,25 @@ def _reuse_existing_daemon_record(target: str) -> _DaemonReuseDecision:
 
     if target != _LOCAL_DAEMON_MARKER:
         # Remote / explicit ``--server`` mode: the daemon connects to a server
-        # we don't own and can't restart, so the config-signature / heal /
-        # "re-run" semantics below don't apply (auth posture is the remote's
-        # concern; its own reconnect loop covers transient tunnel drops). Keep
-        # the original PID-liveness reuse so a live daemon for the URL is
-        # reused as-is.
+        # we don't own and can't restart, so the config-signature and "re-run"
+        # semantics below don't apply — auth posture is the remote's concern.
+        #
+        # Tunnel health still does. A daemon whose tunnel is gone for good is
+        # a zombie either way: every later command waits out
+        # ``wait_for_host_online`` and fails with "did not come online", which
+        # is what made `omnigent host stop` the standing remedy. Heal it the
+        # way local mode does, with one extra guard — only on a definite
+        # "the server says offline", never on a probe we could not complete,
+        # since the daemon's own reconnect loop handles transient drops and we
+        # must not kill a healthy daemon over our own network blip.
+        age_s = time.time() - existing.started_at
+        if (
+            background
+            and age_s >= _DAEMON_REUSE_MIN_AGE_S
+            and _daemon_host_definitely_offline(existing)
+        ):
+            _terminate_host_unit(existing, reason="host tunnel is offline")
+            return _DaemonReuseDecision(reuse=False, config_changed=False)
         return _DaemonReuseDecision(reuse=True, config_changed=False)
 
     if not background:

tests/cli/test_backend.py

@@ -12,6 +12,7 @@ from __future__ import annotations
 
 import hashlib
 import io
+import itertools
 import json
 import re
 import sys
@@ -2327,3 +2328,204 @@ def test_resume_command_defaults_scheme_https(monkeypatch: pytest.MonkeyPatch) -
     assert result.exit_code == 0, result.output
     assert seen == ["https://dbc-x.cloud.databricks.com/omnigent"]
     assert captured["server"] == _expand_marker("https://dbc-x.cloud.databricks.com/omnigent")
+
+
+def _remote_record() -> cli._HostDaemonRecord:
+    """Build a background remote-target daemon record.
+
+    :returns: A ``server``-mode record carrying a log path (background-spawned).
+    """
+    return cli._HostDaemonRecord(
+        pid=4242,
+        target="https://server.example.com",
+        mode="server",
+        server_url="https://server.example.com",
+        log_path="/tmp/daemon.log",
+        started_at=1_000_000,
+        host_id="host_abc",
+        resolved_server_url=None,
+    )
+
+
+def test_daemon_host_state_distinguishes_offline_from_unknown(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A completed probe reports offline; an incomplete one reports unknown."""
+    monkeypatch.setattr(
+        cli,
+        "_host_http_json",
+        lambda **_kw: cli._HostHttpResult(status_code=200, body={"status": "offline"}),
+    )
+    assert cli._daemon_host_state(_online_record()) == "offline"
+
+    monkeypatch.setattr(
+        cli,
+        "_host_http_json",
+        lambda **_kw: cli._HostHttpResult(status_code=0, body="ConnectError: refused"),
+    )
+    assert cli._daemon_host_state(_online_record()) == "unknown"
+
+    # A 404 is not evidence either: the row may be missing because we asked
+    # with different credentials than the daemon registered under.
+    monkeypatch.setattr(
+        cli,
+        "_host_http_json",
+        lambda **_kw: cli._HostHttpResult(status_code=404, body={"detail": "host not found"}),
+    )
+    assert cli._daemon_host_state(_online_record()) == "unknown"
+
+
+def test_daemon_host_definitely_offline_only_on_a_server_answer(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """An unreachable server must never read as a definite offline."""
+    monkeypatch.setattr(cli.time, "sleep", lambda _s: None)
+
+    monkeypatch.setattr(cli, "_daemon_host_state", lambda record, **_kw: "offline")
+    assert cli._daemon_host_definitely_offline(_remote_record(), grace_s=0.0) is True
+
+    monkeypatch.setattr(cli, "_daemon_host_state", lambda record, **_kw: "unknown")
+    assert cli._daemon_host_definitely_offline(_remote_record(), grace_s=0.0) is False
+
+    monkeypatch.setattr(cli, "_daemon_host_state", lambda record, **_kw: "online")
+    assert cli._daemon_host_definitely_offline(_remote_record(), grace_s=0.0) is False
+
+
+def test_daemon_host_definitely_offline_waits_for_a_reconnect(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A daemon that re-registers inside the grace window is not offline."""
+    monkeypatch.setattr(cli.time, "sleep", lambda _s: None)
+    states = iter(["offline", "offline", "online"])
+    monkeypatch.setattr(cli, "_daemon_host_state", lambda record, **_kw: next(states))
+
+    assert cli._daemon_host_definitely_offline(_remote_record(), grace_s=5.0) is False
+
+
+def test_ensure_host_daemon_respawns_zombie_remote_daemon(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """A remote daemon the server calls offline is torn down and respawned.
+
+    Without this every later command waits out ``wait_for_host_online`` and
+    fails with "did not come online", leaving ``omnigent host stop`` as the
+    only remedy.
+    """
+    captured: dict[str, object] = {}
+    _patch_daemon_spawn(monkeypatch, tmp_path, captured)
+    _write_daemon_registry_record(
+        tmp_path,
+        pid=4242,
+        target="https://server.example.com",
+        mode="server",
+        server_url="https://server.example.com",
+        log_path=str(tmp_path / "daemon.log"),
+        started_at=1_000_000,
+    )
+    monkeypatch.setattr(cli, "_pid_alive", lambda pid: True)
+    # Old enough to be eligible for the tunnel-health check.
+    monkeypatch.setattr(cli.time, "time", lambda: 1_000_100.0)
+    monkeypatch.setattr(cli, "_daemon_host_definitely_offline", lambda record, **_kw: True)
+    torn_down: list[str] = []
+    monkeypatch.setattr(
+        cli, "_terminate_host_unit", lambda record, *, reason: torn_down.append(reason)
+    )
+
+    _ensure_host_daemon("https://server.example.com")
+
+    assert torn_down == ["host tunnel is offline"]
+    assert "args" in captured  # a fresh daemon was spawned
+
+
+def test_ensure_host_daemon_keeps_remote_daemon_when_probe_inconclusive(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """An unreachable server must not cost the user a healthy remote daemon.
+
+    The daemon has its own reconnect loop; a blip on the CLI's side is no
+    reason to tear it down.
+    """
+    captured: dict[str, object] = {}
+    _patch_daemon_spawn(monkeypatch, tmp_path, captured)
+    _write_daemon_registry_record(
+        tmp_path,
+        pid=4242,
+        target="https://server.example.com",
+        mode="server",
+        server_url="https://server.example.com",
+        log_path=str(tmp_path / "daemon.log"),
+        started_at=1_000_000,
+    )
+    monkeypatch.setattr(cli, "_pid_alive", lambda pid: True)
+    monkeypatch.setattr(cli.time, "time", lambda: 1_000_100.0)
+    monkeypatch.setattr(cli.time, "sleep", lambda _s: None)
+    # Advance the monotonic clock so the grace window elapses without waiting.
+    ticks = itertools.count(0.0, 10.0)
+    monkeypatch.setattr(cli.time, "monotonic", lambda: next(ticks))
+    monkeypatch.setattr(cli, "_daemon_host_state", lambda record, **_kw: "unknown")
+    torn_down: list[str] = []
+    monkeypatch.setattr(
+        cli, "_terminate_host_unit", lambda record, *, reason: torn_down.append(reason)
+    )
+
+    _ensure_host_daemon("https://server.example.com")
+
+    assert torn_down == []
+    assert "args" not in captured  # reused, not respawned
+
+
+def test_ensure_host_daemon_keeps_young_remote_daemon(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """A just-spawned remote daemon is not judged before it can connect."""
+    captured: dict[str, object] = {}
+    _patch_daemon_spawn(monkeypatch, tmp_path, captured)
+    _write_daemon_registry_record(
+        tmp_path,
+        pid=4242,
+        target="https://server.example.com",
+        mode="server",
+        server_url="https://server.example.com",
+        log_path=str(tmp_path / "daemon.log"),
+        started_at=1_000_000,
+    )
+    monkeypatch.setattr(cli, "_pid_alive", lambda pid: True)
+    # Younger than _DAEMON_REUSE_MIN_AGE_S.
+    monkeypatch.setattr(cli.time, "time", lambda: 1_000_001.0)
+
+    def _must_not_probe(record: object, **_kw: object) -> bool:
+        raise AssertionError("a young daemon must not be probed for tunnel health")
+
+    monkeypatch.setattr(cli, "_daemon_host_definitely_offline", _must_not_probe)
+
+    _ensure_host_daemon("https://server.example.com")
+
+    assert "args" not in captured  # reused, not respawned
+
+
+def test_ensure_host_daemon_keeps_foreground_remote_daemon(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """A foreground ``omnigent host`` is never silently killed by the heal."""
+    captured: dict[str, object] = {}
+    _patch_daemon_spawn(monkeypatch, tmp_path, captured)
+    _write_daemon_registry_record(
+        tmp_path,
+        pid=4242,
+        target="https://server.example.com",
+        mode="server",
+        server_url="https://server.example.com",
+        log_path=None,  # foreground
+        started_at=1_000_000,
+    )
+    monkeypatch.setattr(cli, "_pid_alive", lambda pid: True)
+    monkeypatch.setattr(cli.time, "time", lambda: 1_000_100.0)
+
+    def _must_not_probe(record: object, **_kw: object) -> bool:
+        raise AssertionError("a foreground daemon must not be probed for tunnel health")
+
+    monkeypatch.setattr(cli, "_daemon_host_definitely_offline", _must_not_probe)
+
+    _ensure_host_daemon("https://server.example.com")
+
+    assert "args" not in captured  # reused, not respawned