fix(photon): recover inbound after half-open ("zombie") gRPC stream

spectrum-ts's live-stream consumer (consumeLive in spectrum-ts 3.x) only
reconnects when its inbound async iterator throws or ends. A half-open
("zombie") gRPC socket — where the TCP connection stays ESTABLISHED but the
peer is gone (NAT idle-timeout, network blip, laptop sleep) — makes the
iterator hang forever: no error, no end. The SDK exposes no gRPC keepalive
knob (createClient takes only {address, tls, token}; grpc.keepalive_time_ms
defaults to -1 = pings off), so the inbound stream silently dies and stays
dead until the gateway is restarted. Symptom: the agent's iMessage line goes
"online but deaf" — Photon's cloud-side fallback answers users with "the agent
isn't online right now" and inbound never reaches the gateway.

Fix, entirely in the code we own (no SDK fork):

- Sidecar gains a POST /probe endpoint that drives a cheap unary read
  (space.getMessage on a synthetic id) over the SAME gRPC channel the inbound
  stream uses. A live channel round-trips in ms (server returns not-found,
  which is success for liveness); a zombie hangs. It sends nothing to any user
  and creates no chat (space.get is local in shared/dedicated mode; only the
  message read touches the wire).

- The adapter runs a presence watchdog: it probes on an interval, skips the
  probe when natural inbound traffic already proved liveness within the
  window, and after N consecutive failed probes respawns the sidecar — a fresh
  Spectrum() re-subscribes the stream and re-registers presence. Successful
  probes double as application-level keepalive, helping prevent the zombie
  from forming at all. Respawn is lock-guarded against double-spawn and the
  watchdog is torn down cleanly on disconnect.

Behavioural settings live in config.yaml (extra), bridged to env per the
.env-is-secrets-only convention:
  probe_interval_seconds (60), probe_timeout_seconds (10),
  probe_max_failures (3). A non-positive interval disables the watchdog.

Tests: tests/plugins/platforms/photon/test_presence_watchdog.py covers config
resolution, the disable switch, probe alive/dead(500)/timeout/no-client, the
core N-failures->one-respawn detection, success-resets-failures, stop-then-
start respawn ordering, and lock-guarding — all without spawning Node or
hitting the network.

Contributed by Vaibhav Sharma (X: @vabbyshabby).

plugins/platforms/photon/adapter.py

@@ -311,6 +311,33 @@ def sidecar_deps_installed() -> bool:
     return (_SIDECAR_DIR / "node_modules" / "spectrum-ts").exists()
 
 
+def _coerce_float(value: Any, default: float) -> float:
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def _coerce_int(value: Any, default: int) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def _first_set(*values: Any) -> Any:
+    """Return the first value that is not None.
+
+    Unlike ``a or b``, this preserves an explicit falsy value (e.g. ``0`` or
+    ``""``) so config can intentionally set a zero/empty without it silently
+    falling through to a later source or a default.
+    """
+    for v in values:
+        if v is not None:
+            return v
+    return None
+
+
 def check_requirements() -> bool:
     """Return True when both Python deps and the Node sidecar are available."""
     if not HTTPX_AVAILABLE:
@@ -638,6 +665,42 @@ class PhotonAdapter(BasePlatformAdapter):
         ).lower() not in ("0", "false", "no")
         self._node_bin = os.getenv("PHOTON_NODE_BIN") or shutil.which("node") or "node"
 
+        # Presence watchdog. spectrum-ts only reconnects when its inbound
+        # iterator throws or ends; a half-open ("zombie") gRPC socket makes the
+        # iterator hang forever (no error, no end), so inbound silently dies
+        # until the sidecar is restarted. We periodically drive a cheap upstream
+        # unary read via the sidecar's /probe endpoint; repeated failures mean
+        # the stream is dead -> respawn the sidecar (fresh Spectrum() = fresh
+        # gRPC stream). A successful probe is also a real round-trip that keeps
+        # the channel warm, helping prevent the zombie forming at all.
+        # Behavioural settings -> config.yaml (extra), bridged to env.
+        # Use _first_set (not ``or``) so an explicit 0 is honored — ``0 or X``
+        # would silently fall through to the default and you could never
+        # disable the watchdog with probe_interval_seconds: 0.
+        self._probe_interval = _coerce_float(
+            _first_set(
+                extra.get("probe_interval_seconds"),
+                os.getenv("PHOTON_PROBE_INTERVAL_SECONDS"),
+            ),
+            60.0,
+        )
+        self._probe_timeout = _coerce_float(
+            _first_set(
+                extra.get("probe_timeout_seconds"),
+                os.getenv("PHOTON_PROBE_TIMEOUT_SECONDS"),
+            ),
+            10.0,
+        )
+        self._probe_max_failures = _coerce_int(
+            _first_set(
+                extra.get("probe_max_failures"),
+                os.getenv("PHOTON_PROBE_MAX_FAILURES"),
+            ),
+            3,
+        )
+        # A non-positive interval disables the watchdog entirely (escape hatch).
+        self._probe_enabled = self._probe_interval > 0
+
         # With markdown on, format_message preserves fences and the sidecar's
         # markdown() builder renders them (or degrades them readably).
         self.supports_code_blocks = _markdown_enabled()
@@ -650,6 +713,15 @@ class PhotonAdapter(BasePlatformAdapter):
         self._inbound_running = False
         self._http_client: Optional["httpx.AsyncClient"] = None
         self._sidecar_health_interval = 15.0
+        # Presence-watchdog runtime state.
+        self._watchdog_task: Optional[asyncio.Task] = None
+        self._watchdog_running = False
+        self._probe_failures = 0
+        # Monotonic timestamp of the last real upstream activity (inbound
+        # message or a successful probe). The watchdog skips its own probe when
+        # natural traffic already proved the channel live within the interval.
+        self._last_upstream_activity = 0.0
+        self._respawn_lock: Optional[asyncio.Lock] = None
         # Lightweight in-memory dedup. The gRPC stream is at-least-once, so we
         # may see the same messageId more than once (e.g. after a reconnect).
         self._seen_messages: Dict[str, float] = {}
@@ -796,6 +868,16 @@ class PhotonAdapter(BasePlatformAdapter):
             self._monitor_sidecar_health()
         )
 
+        # Start the presence watchdog (detects a half-open/zombie gRPC stream
+        # the SDK can't see and respawns the sidecar to recover inbound).
+        self._last_upstream_activity = time.monotonic()
+        if self._probe_enabled and self._autostart_sidecar:
+            self._respawn_lock = asyncio.Lock()
+            self._watchdog_running = True
+            self._watchdog_task = asyncio.get_event_loop().create_task(
+                self._presence_watchdog()
+            )
+
         self._mark_connected()
         logger.info(
             "[photon] connected — sidecar on %s:%d, streaming inbound over gRPC",
@@ -805,6 +887,9 @@ class PhotonAdapter(BasePlatformAdapter):
 
     async def disconnect(self) -> None:
         self._inbound_running = False
+        # Stop the watchdog first so it can't trigger a respawn while we tear
+        # the sidecar down.
+        await self._stop_watchdog()
         if self._sidecar_health_task is not None:
             task = self._sidecar_health_task
             self._sidecar_health_task = None
@@ -965,6 +1050,12 @@ class PhotonAdapter(BasePlatformAdapter):
             break
 
     async def _on_inbound_line(self, line: str) -> None:
+        # Any line on the inbound stream — message OR heartbeat — proves the
+        # upstream gRPC channel is live, so reset the watchdog's failure count
+        # and activity clock. (Heartbeats arrive as blank lines, already
+        # filtered before this method, but a real message is the strongest
+        # liveness signal we have.)
+        self._note_upstream_activity()
         try:
             event = json.loads(line)
         except json.JSONDecodeError:
@@ -1611,6 +1702,126 @@ class PhotonAdapter(BasePlatformAdapter):
                     self._sidecar_supervisor_task.cancel()
                 self._sidecar_supervisor_task = None
 
+    # -- Presence watchdog -------------------------------------------------
+
+    def _note_upstream_activity(self) -> None:
+        """Record proof the upstream gRPC channel is live, and clear failures.
+
+        Called on every inbound line and after every successful probe.
+        """
+        self._last_upstream_activity = time.monotonic()
+        self._probe_failures = 0
+
+    async def _probe_once(self) -> bool:
+        """Drive one upstream liveness probe via the sidecar's ``/probe``.
+
+        Returns True if the sidecar confirmed a live gRPC round-trip within
+        the timeout, False on timeout/error (a likely zombie stream).
+        """
+        client = self._http_client
+        if client is None:
+            return False
+        url = f"http://{self._sidecar_bind}:{self._sidecar_port}/probe"
+        try:
+            resp = await client.post(
+                url,
+                headers={"X-Hermes-Sidecar-Token": self._sidecar_token},
+                timeout=self._probe_timeout,
+            )
+        except Exception as e:
+            logger.debug("[photon] probe transport error: %s", e)
+            return False
+        return resp.status_code == 200
+
+    async def _respawn_sidecar(self, reason: str) -> None:
+        """Tear down and restart the sidecar to recover a dead gRPC stream.
+
+        A fresh ``Spectrum()`` re-subscribes the inbound stream and
+        re-registers presence with Photon cloud. The inbound loop's existing
+        reconnect logic re-opens the loopback NDJSON stream automatically once
+        the new sidecar's ``/inbound`` is up. Guarded by a lock so overlapping
+        triggers (watchdog + a manual call) can't double-spawn.
+        """
+        lock = self._respawn_lock
+        if lock is None:
+            lock = self._respawn_lock = asyncio.Lock()
+        if lock.locked():
+            logger.info("[photon] respawn already in progress; skipping")
+            return
+        async with lock:
+            logger.warning(
+                "[photon] presence watchdog: %s — respawning sidecar", reason
+            )
+            try:
+                await self._stop_sidecar()
+            except Exception:
+                logger.exception("[photon] error stopping sidecar during respawn")
+            try:
+                await self._start_sidecar()
+            except Exception:
+                logger.exception(
+                    "[photon] failed to respawn sidecar; watchdog will retry"
+                )
+                return
+            # Fresh sidecar -> fresh stream. Reset the activity clock so we give
+            # it a full interval before probing again, and clear failures.
+            self._note_upstream_activity()
+            logger.info("[photon] presence watchdog: sidecar respawned, gRPC stream renewed")
+
+    async def _presence_watchdog(self) -> None:
+        """Periodically confirm the upstream gRPC stream is alive.
+
+        spectrum-ts only recovers when its inbound iterator throws or ends; a
+        half-open ("zombie") socket makes it hang forever with no error, so
+        inbound silently dies. We probe the channel on an interval (skipping
+        the probe when natural inbound traffic already proved liveness within
+        the window). After ``_probe_max_failures`` consecutive failed probes we
+        respawn the sidecar to force a fresh stream.
+        """
+        # Stagger the first probe so a fleet of restarts doesn't synchronize,
+        # and so a freshly-started sidecar isn't probed before it's warm.
+        await asyncio.sleep(self._probe_interval)
+        while self._watchdog_running:
+            try:
+                # Skip our own probe if inbound traffic already proved the
+                # channel live within the last interval (cheaper, and avoids
+                # piling synthetic reads on a busy line).
+                idle = time.monotonic() - self._last_upstream_activity
+                if idle < self._probe_interval:
+                    await asyncio.sleep(self._probe_interval - idle)
+                    continue
+
+                alive = await self._probe_once()
+                if alive:
+                    self._note_upstream_activity()
+                else:
+                    self._probe_failures += 1
+                    logger.warning(
+                        "[photon] presence probe failed (%d/%d)",
+                        self._probe_failures, self._probe_max_failures,
+                    )
+                    if self._probe_failures >= self._probe_max_failures:
+                        await self._respawn_sidecar(
+                            f"{self._probe_failures} consecutive probe failures"
+                        )
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                logger.exception("[photon] presence watchdog iteration failed")
+            await asyncio.sleep(self._probe_interval)
+
+    async def _stop_watchdog(self) -> None:
+        self._watchdog_running = False
+        if self._watchdog_task is not None:
+            self._watchdog_task.cancel()
+            try:
+                await self._watchdog_task
+            except asyncio.CancelledError:
+                pass
+            except Exception:
+                pass
+            self._watchdog_task = None
+
     # -- Outbound ----------------------------------------------------------
 
     async def send(

plugins/platforms/photon/sidecar/index.mjs

@@ -211,6 +211,14 @@ console.log = (...args) => {
   originalConsoleLog(...args);
 };
 
+// Upstream liveness probe (see `/probe` handler). A synthetic DM space id +
+// a unique-per-probe bogus message id drive a cheap unary read over the same
+// gRPC channel the inbound stream uses, so we can tell a live channel from a
+// half-open ("zombie") one. `space.get` is purely local in shared/dedicated
+// mode (no chat is created or messaged); only the message read hits the wire.
+const PROBE_SPACE_ID = process.env.PHOTON_PROBE_SPACE_ID || "any;-;+10000000000";
+const PROBE_MSG_PREFIX = "hermes-liveness-probe-";
+
 if (!projectId || !projectSecret || !sharedToken) {
   console.error(
     "photon-sidecar: PHOTON_PROJECT_ID, PHOTON_PROJECT_SECRET and " +
@@ -831,6 +839,40 @@ const server = http.createServer(async (req, res) => {
     if (req.url === "/healthz") {
       return ok(res, { stream: streamHealthSnapshot() });
     }
+    if (req.url === "/probe") {
+      // Upstream liveness probe. Drives a cheap unary read over the SAME gRPC
+      // channel the inbound stream uses. On a live channel this round-trips
+      // fast (the server returns "not found" for the synthetic id, which the
+      // SDK surfaces as a thrown error — that's a SUCCESS for our purposes:
+      // the wire is alive). On a half-open zombie socket the call hangs and
+      // the caller's timeout fires, so the adapter can respawn us. It sends
+      // nothing to any user and creates no chat (space.get is local; only the
+      // message read touches the network).
+      if (typeof app?.stop !== "function") {
+        // app failed to construct — definitely not live.
+        return serverError(res);
+      }
+      try {
+        const im = imessage(app);
+        const space = await im.space.get(PROBE_SPACE_ID);
+        const probeId = PROBE_MSG_PREFIX + Date.now() + "-" + Math.random().toString(36).slice(2);
+        try {
+          await space.getMessage(probeId);
+        } catch {
+          // Expected: the synthetic id doesn't exist. Reaching here means the
+          // unary call completed a round-trip — the channel is ALIVE.
+        }
+        return ok(res, { alive: true });
+      } catch (e) {
+        // space.get is local; an error here is unexpected but means we can't
+        // even build a probe — report not-alive so the adapter recycles us.
+        console.error(
+          "photon-sidecar: probe failed: " +
+            (e && e.message ? e.message : String(e))
+        );
+        return serverError(res);
+      }
+    }
     if (req.url === "/shutdown") {
       ok(res, {});
       setTimeout(() => process.kill(process.pid, "SIGTERM"), 50);

tests/plugins/platforms/photon/test_presence_watchdog.py

@@ -0,0 +1,231 @@
+"""Presence-watchdog tests.
+
+spectrum-ts only reconnects when its inbound iterator throws or ends; a
+half-open ("zombie") gRPC socket makes the iterator hang forever (no error, no
+end), so inbound silently dies until the sidecar is restarted. The adapter's
+presence watchdog probes the upstream channel via the sidecar's ``/probe``
+endpoint and respawns the sidecar after repeated probe failures.
+
+These tests exercise the watchdog's decision logic (probe -> count failures ->
+respawn; success resets; recent inbound traffic skips the probe) without
+spawning Node, binding ports, or hitting the network.
+"""
+from __future__ import annotations
+
+import time
+from typing import Any, List
+
+import pytest
+
+from gateway.config import PlatformConfig
+from plugins.platforms.photon.adapter import PhotonAdapter
+
+
+def _make_adapter(monkeypatch: pytest.MonkeyPatch, **extra: Any) -> PhotonAdapter:
+    monkeypatch.setenv("PHOTON_PROJECT_ID", "test-project-id")
+    monkeypatch.setenv("PHOTON_PROJECT_SECRET", "test-project-secret")
+    cfg = PlatformConfig(enabled=True, token="", extra=dict(extra))
+    return PhotonAdapter(cfg)
+
+
+def test_probe_config_defaults(monkeypatch: pytest.MonkeyPatch) -> None:
+    a = _make_adapter(monkeypatch)
+    assert a._probe_interval == 60.0
+    assert a._probe_timeout == 10.0
+    assert a._probe_max_failures == 3
+    assert a._probe_enabled is True
+
+
+def test_probe_config_from_extra(monkeypatch: pytest.MonkeyPatch) -> None:
+    a = _make_adapter(
+        monkeypatch,
+        probe_interval_seconds=30,
+        probe_timeout_seconds=5,
+        probe_max_failures=2,
+    )
+    assert a._probe_interval == 30.0
+    assert a._probe_timeout == 5.0
+    assert a._probe_max_failures == 2
+    assert a._probe_enabled is True
+
+
+def test_probe_disabled_when_interval_nonpositive(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    a = _make_adapter(monkeypatch, probe_interval_seconds=0)
+    assert a._probe_enabled is False
+
+
+def test_note_activity_resets_failures(monkeypatch: pytest.MonkeyPatch) -> None:
+    a = _make_adapter(monkeypatch)
+    a._probe_failures = 2
+    before = a._last_upstream_activity
+    time.sleep(0.001)
+    a._note_upstream_activity()
+    assert a._probe_failures == 0
+    assert a._last_upstream_activity > before
+
+
+@pytest.mark.asyncio
+async def test_probe_once_alive_on_200(monkeypatch: pytest.MonkeyPatch) -> None:
+    a = _make_adapter(monkeypatch)
+
+    class _Resp:
+        status_code = 200
+
+    class _Client:
+        async def post(self, *args: Any, **kwargs: Any) -> Any:
+            return _Resp()
+
+    a._http_client = _Client()  # type: ignore[assignment]
+    assert await a._probe_once() is True
+
+
+@pytest.mark.asyncio
+async def test_probe_once_dead_on_500(monkeypatch: pytest.MonkeyPatch) -> None:
+    a = _make_adapter(monkeypatch)
+
+    class _Resp:
+        status_code = 500
+
+    class _Client:
+        async def post(self, *args: Any, **kwargs: Any) -> Any:
+            return _Resp()
+
+    a._http_client = _Client()  # type: ignore[assignment]
+    assert await a._probe_once() is False
+
+
+@pytest.mark.asyncio
+async def test_probe_once_dead_on_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
+    a = _make_adapter(monkeypatch)
+
+    class _Client:
+        async def post(self, *args: Any, **kwargs: Any) -> Any:
+            raise TimeoutError("zombie stream — probe hung")
+
+    a._http_client = _Client()  # type: ignore[assignment]
+    # A hung/half-open probe must read as NOT alive (this is the zombie case).
+    assert await a._probe_once() is False
+
+
+@pytest.mark.asyncio
+async def test_probe_once_false_without_client(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    a = _make_adapter(monkeypatch)
+    a._http_client = None
+    assert await a._probe_once() is False
+
+
+@pytest.mark.asyncio
+async def test_respawn_after_max_failures(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The core fix: N consecutive dead probes -> exactly one respawn."""
+    a = _make_adapter(monkeypatch, probe_max_failures=3)
+
+    respawns: List[str] = []
+
+    async def _fake_respawn(reason: str) -> None:
+        respawns.append(reason)
+        a._note_upstream_activity()  # mirror real respawn (clears failures)
+
+    async def _dead_probe() -> bool:
+        return False
+
+    monkeypatch.setattr(a, "_respawn_sidecar", _fake_respawn)
+    monkeypatch.setattr(a, "_probe_once", _dead_probe)
+
+    # Simulate the watchdog's per-iteration decision logic directly (no sleeps).
+    a._last_upstream_activity = time.monotonic() - 999  # force a probe each time
+    for _ in range(3):
+        alive = await a._probe_once()
+        assert alive is False
+        a._probe_failures += 1
+        if a._probe_failures >= a._probe_max_failures:
+            await a._respawn_sidecar("test")
+
+    assert respawns == ["test"]
+    assert a._probe_failures == 0  # reset by the (faked) respawn
+
+
+@pytest.mark.asyncio
+async def test_success_resets_failure_count(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A live probe between dead ones prevents a respawn (failures reset)."""
+    a = _make_adapter(monkeypatch, probe_max_failures=3)
+
+    respawns: List[str] = []
+
+    async def _fake_respawn(reason: str) -> None:
+        respawns.append(reason)
+
+    monkeypatch.setattr(a, "_respawn_sidecar", _fake_respawn)
+
+    # Two failures, then a success, then two more failures: never hits 3 in a row.
+    sequence = [False, False, True, False, False]
+    for alive in sequence:
+        if alive:
+            a._note_upstream_activity()
+        else:
+            a._probe_failures += 1
+            if a._probe_failures >= a._probe_max_failures:
+                await a._respawn_sidecar("should-not-fire")
+
+    assert respawns == []
+    assert a._probe_failures == 2
+
+
+@pytest.mark.asyncio
+async def test_respawn_calls_stop_then_start(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Respawn tears the sidecar down and brings a fresh one up, in order."""
+    a = _make_adapter(monkeypatch)
+    calls: List[str] = []
+
+    async def _fake_stop() -> None:
+        calls.append("stop")
+
+    async def _fake_start() -> None:
+        calls.append("start")
+
+    monkeypatch.setattr(a, "_stop_sidecar", _fake_stop)
+    monkeypatch.setattr(a, "_start_sidecar", _fake_start)
+
+    await a._respawn_sidecar("test reason")
+    assert calls == ["stop", "start"]
+    # A fresh stream means failures are cleared.
+    assert a._probe_failures == 0
+
+
+@pytest.mark.asyncio
+async def test_respawn_is_lock_guarded(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A second respawn while one is in flight is skipped, not double-spawned."""
+    import asyncio
+
+    a = _make_adapter(monkeypatch)
+    starts: List[str] = []
+
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    async def _slow_stop() -> None:
+        started.set()
+        await release.wait()
+
+    async def _fake_start() -> None:
+        starts.append("start")
+
+    monkeypatch.setattr(a, "_stop_sidecar", _slow_stop)
+    monkeypatch.setattr(a, "_start_sidecar", _fake_start)
+    a._respawn_lock = asyncio.Lock()
+
+    first = asyncio.create_task(a._respawn_sidecar("first"))
+    await started.wait()  # first respawn is now holding the lock inside _stop
+    # Second call should see the lock held and return immediately.
+    await a._respawn_sidecar("second")
+    assert starts == []  # second did not proceed to start
+    release.set()
+    await first
+    assert starts == ["start"]  # only the first respawn started a new sidecar