fix(photon): inspect port listeners off the gateway event loop

`_reap_stale_sidecar` is `async`, but it identified the processes holding
the sidecar port with two blocking helpers called inline:

* `_find_listener_pids` -> `subprocess.run(["lsof", ...], timeout=5.0)`
* `_pid_is_sidecar` -> `subprocess.run(["ps", ...], timeout=5.0)`, once
  per candidate pid

so the inspection can hold the shared gateway loop for 5 + 5·N seconds
while nothing else on it is serviced. It only runs once the /healthz probe
finds something already listening — the orphaned-sidecar recovery path —
and `_reap_stale_sidecar` is awaited from `_start_sidecar`, which runs on
every reconnect (`connect(is_reconnect=True)`). The stall therefore lands
on a live gateway that is still serving every other platform, right when a
crashed sidecar has already left an orphan behind.

Move the whole inspection to one `asyncio.to_thread` hop (one hop rather
than N+1 round trips). The reaping semantics are untouched: SIGTERM for
verified orphans, SIGKILL escalation, and both foreign-listener
RuntimeErrors behave exactly as before.

Same off-the-loop class as the inbound-image decision (#66688) and the
cron-fire verifier.

Adds a regression test asserting both the lsof lookup and the per-pid ps
check execute on a worker thread rather than the loop thread.

plugins/platforms/photon/adapter.py

@@ -1115,9 +1115,20 @@ class PhotonAdapter(BasePlatformAdapter):
                 )
         except httpx.RequestError:
             return  # nothing listening — the normal case
-        pids = self._find_listener_pids(self._sidecar_port)
-        stale = [pid for pid in pids if self._pid_is_sidecar(pid)]
-        foreign = [pid for pid in pids if pid not in stale]
+        # Off the event loop: _find_listener_pids shells out to `lsof`
+        # (timeout=5s) and _pid_is_sidecar runs a `ps` per candidate pid
+        # (timeout=5s each), so this inspection can hold the loop for
+        # 5 + 5·N seconds. _reap_stale_sidecar is awaited from
+        # _start_sidecar, which runs on every reconnect — exactly when a
+        # crashed gateway left an orphan behind — so the stall lands on a
+        # live gateway that is still serving every other platform. One hop
+        # covers the whole inspection instead of N+1 round trips.
+        def _inspect():
+            found = self._find_listener_pids(self._sidecar_port)
+            mine = [pid for pid in found if self._pid_is_sidecar(pid)]
+            return mine, [pid for pid in found if pid not in mine]
+
+        stale, foreign = await asyncio.to_thread(_inspect)
         if not stale:
             raise RuntimeError(
                 f"port {self._sidecar_port} is in use by another process "

tests/plugins/platforms/photon/test_sidecar_lifecycle.py

@@ -206,3 +206,50 @@ async def test_start_sidecar_rejects_empty_node_modules(
 
     with pytest.raises(RuntimeError, match="not installed"):
         await adapter._start_sidecar()
+
+
+@pytest.mark.asyncio
+async def test_reap_inspects_listeners_off_the_event_loop(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Listener inspection must not block the shared gateway event loop.
+
+    ``_find_listener_pids`` shells out to ``lsof`` (timeout=5s) and
+    ``_pid_is_sidecar`` runs a ``ps`` per candidate pid (timeout=5s each), so
+    inline this holds the loop for 5 + 5·N seconds. ``_reap_stale_sidecar`` is
+    awaited from ``_start_sidecar``, which runs on every reconnect — exactly
+    when a crashed gateway left an orphan — so the stall lands on a live
+    gateway still serving every other platform.
+    """
+    import threading
+
+    adapter = _make_adapter(monkeypatch)
+    monkeypatch.setattr(photon_adapter.sys, "platform", "linux")
+    monkeypatch.setattr(photon_adapter.httpx, "AsyncClient", _ProbeClient)
+
+    main_thread = threading.current_thread()
+    seen: Dict[str, Any] = {}
+
+    def _fake_find(port: int) -> List[int]:
+        seen["find"] = threading.current_thread()
+        return [4242]
+
+    def _fake_is_sidecar(pid: int) -> bool:
+        seen["ps"] = threading.current_thread()
+        return True
+
+    monkeypatch.setattr(adapter, "_find_listener_pids", _fake_find)
+    monkeypatch.setattr(adapter, "_pid_is_sidecar", _fake_is_sidecar)
+    monkeypatch.setattr(adapter, "_pid_alive", lambda pid: False)
+    _capture_kills(monkeypatch)
+
+    await adapter._reap_stale_sidecar()
+
+    assert seen.get("find") is not None, "lsof lookup never ran"
+    assert seen.get("ps") is not None, "ps check never ran"
+    for label in ("find", "ps"):
+        assert seen[label] is not main_thread, (
+            f"{label} ran on the event-loop thread; the listener inspection "
+            "must be dispatched via asyncio.to_thread so a 5s lsof/ps spawn "
+            "can't freeze every other platform on the gateway loop"
+        )