From 4958703975e1fb3bfb626252a93638ffb0e5bbbc Mon Sep 17 00:00:00 2001 From: Jeffgithub0029 Date: Thu, 16 Jul 2026 13:28:29 +0800 Subject: [PATCH] fix(photon): recover sidecar after outbound transport drops (salvaged v2) - Add _sidecar_restart_lock to serialize restarts - Add _ensure_sidecar_running() called before inbound reconnect - Add _restart_sidecar_for_outbound_error() + _is_recoverable_outbound_drop() - Hook into adapter.send() (direct cron/tools/agent path) and _send_with_retry() - Propagate sidecar error_code in _sidecar_call() exceptions - Remove stale inbound fatal hook (isFatalInboundStreamError, FATAL_INBOUND_EXIT_CODE) - Keep only outbound classifier in fatal-errors.mjs - Add test coverage for direct send, send_with_retry, and sidecar restart --- .gitignore | 10 ++ plugins/platforms/photon/adapter.py | 123 +++++++++++++++++- .../platforms/photon/sidecar/fatal-errors.mjs | 38 ++++++ plugins/platforms/photon/sidecar/package.json | 3 +- .../photon/sidecar/test/fatal-errors.test.mjs | 10 ++ .../test_photon_adapter_outbound_recovery.py | 109 ++++++++++++++++ .../test_photon_adapter_sidecar_restart.py | 60 +++++++++ 7 files changed, 348 insertions(+), 5 deletions(-) create mode 100644 plugins/platforms/photon/sidecar/fatal-errors.mjs create mode 100644 plugins/platforms/photon/sidecar/test/fatal-errors.test.mjs create mode 100644 tests/plugins/test_photon_adapter_outbound_recovery.py create mode 100644 tests/plugins/test_photon_adapter_sidecar_restart.py diff --git a/.gitignore b/.gitignore index 6f1b3be6d92b..1f65c52f68ab 100644 --- a/.gitignore +++ b/.gitignore @@ -82,6 +82,16 @@ apps/shared/src/**/*.d.ts apps/desktop/release/ *.tsbuildinfo +# Desktop `tsc -b` emit: generated JS mirrors alongside tracked .ts/.tsx sources +# under apps/desktop/src. These are build artifacts (not hand-written), were never +# meant to be tracked, and get pulled into `hermes update`'s autostash on every +# upgrade — during which a restored stale .js next to a freshly-merged .tsx +# triggers "duplicate identifier" and breaks the desktop rebuild. Ignore them so +# the working tree stays clean across updates. +apps/desktop/src/**/*.js +apps/desktop/src/**/*.js.map +apps/desktop/src/**/*.d.ts + # Web UI assets — synced from @nous-research/ui at build time via # `npm run sync-assets` (see web/package.json). web/public/fonts/ diff --git a/plugins/platforms/photon/adapter.py b/plugins/platforms/photon/adapter.py index 27df34ecf2c4..cffd8f00b633 100644 --- a/plugins/platforms/photon/adapter.py +++ b/plugins/platforms/photon/adapter.py @@ -317,6 +317,7 @@ def __init__(self, config: PlatformConfig): self._sidecar_supervisor_task: Optional[asyncio.Task] = None self._inbound_task: Optional[asyncio.Task] = None self._sidecar_health_task: Optional[asyncio.Task] = None + self._sidecar_restart_lock = asyncio.Lock() self._inbound_running = False self._http_client: Optional["httpx.AsyncClient"] = None self._sidecar_health_interval = 15.0 @@ -512,6 +513,7 @@ async def _inbound_loop(self) -> None: backoff = 1.0 while self._inbound_running: try: + await self._ensure_sidecar_running() async with client.stream( "GET", url, headers=headers, timeout=None, ) as resp: @@ -1078,6 +1080,40 @@ async def _stop_sidecar(self) -> None: self._sidecar_supervisor_task.cancel() self._sidecar_supervisor_task = None + async def _ensure_sidecar_running(self) -> None: + """Restart the supervised sidecar if it exited after initial connect.""" + if not self._autostart_sidecar: + return + proc = self._sidecar_proc + if proc is not None and proc.poll() is None: + return + async with self._sidecar_restart_lock: + proc = self._sidecar_proc + if proc is not None and proc.poll() is None: + return + exit_code = proc.returncode if proc is not None else None + logger.warning( + "[photon] sidecar exited%s; restarting before reconnecting inbound stream", + f" with code {exit_code}" if exit_code is not None else "", + ) + self._sidecar_proc = None + if self._sidecar_supervisor_task is not None: + self._sidecar_supervisor_task.cancel() + self._sidecar_supervisor_task = None + await self._start_sidecar() + + async def _restart_sidecar_for_outbound_error(self, reason: str) -> None: + """Force a clean sidecar restart after a recoverable outbound transport fault.""" + if not self._autostart_sidecar: + return + async with self._sidecar_restart_lock: + logger.warning( + "[photon] restarting sidecar after recoverable outbound error: %s", + reason, + ) + await self._stop_sidecar() + await self._start_sidecar() + # -- Outbound ---------------------------------------------------------- async def send( @@ -1087,7 +1123,40 @@ async def send( reply_to: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, ) -> SendResult: - return await self._sidecar_send(chat_id, self.format_message(content)) + """Send one text message through the sidecar. + + Cron live-delivery calls adapter.send() directly, bypassing the + gateway response retry wrapper. Keep recoverable sidecar restart logic + at this lowest text-send layer so cron, tools, and agent replies all + share it. + """ + text = self.format_message(content) + result = await self._sidecar_send(chat_id, text) + if result.success or not self._is_recoverable_outbound_drop(result.error or ""): + return result + try: + await self._restart_sidecar_for_outbound_error( + "upstream_connection_dropped during direct /send" + ) + except Exception as e: + logger.warning( + "[photon] sidecar restart after direct outbound drop failed: %s", + e, + ) + return result + recovered = await self._sidecar_send(chat_id, text) + if recovered.success: + logger.info("[photon] recovered direct outbound send after sidecar restart") + return recovered + return recovered + + @staticmethod + def _is_recoverable_outbound_drop(error_str: str) -> bool: + return ( + "error_code=upstream_connection_dropped" in error_str + or "upstream_connection_dropped" in error_str + or "[upstream] Connection dropped" in error_str + ) # -- Outbound media (parity with the BlueBubbles iMessage channel) ----- # @@ -1428,6 +1497,18 @@ async def _send_with_retry( if not is_network and self._is_timeout_error(error_str): return result + # If this is a recoverable upstream drop, restart sidecar once before retrying. + if "error_code=upstream_connection_dropped" in error_str: + try: + await self._restart_sidecar_for_outbound_error( + "upstream_connection_dropped during /send" + ) + except Exception as e: + logger.warning( + "[photon] sidecar restart after outbound drop failed: %s", + e, + ) + if is_network: for attempt in range(1, max_retries + 1): delay = base_delay * (2 ** (attempt - 1)) @@ -1454,6 +1535,31 @@ async def _send_with_retry( ) return result + if "error_code=upstream_connection_dropped" in error_str: + try: + await self._restart_sidecar_for_outbound_error( + "upstream_connection_dropped during /send" + ) + except Exception as e: + logger.warning( + "[photon] sidecar restart after outbound drop failed: %s", + e, + ) + else: + recovered = await self.send( + chat_id=chat_id, + content=text, + reply_to=reply_to, + metadata=metadata, + ) + if recovered.success: + logger.info( + "[photon] recovered outbound send after sidecar restart" + ) + return recovered + error_str = recovered.error or error_str + result = recovered + logger.warning( "[photon] Send failed: %s - retrying plain-text message", error_str, @@ -1549,14 +1655,23 @@ async def _sidecar_call(self, path: str, body: Dict[str, Any]) -> Dict[str, Any] headers = {"X-Hermes-Sidecar-Token": self._sidecar_token} async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.post(url, json=body, headers=headers) + payload: Dict[str, Any] = {} + try: + payload = resp.json() or {} + except Exception: + payload = {} if resp.status_code != 200: + error_code = payload.get("error_code") + suffix = f" error_code={error_code}" if error_code else "" raise RuntimeError( - f"Photon sidecar {path} returned {resp.status_code}: {resp.text[:200]}" + f"Photon sidecar {path} returned {resp.status_code}{suffix}: {resp.text[:200]}" ) - data = resp.json() or {} + data = payload if not data.get("ok"): + error_code = data.get("error_code") + suffix = f" error_code={error_code}" if error_code else "" raise RuntimeError( - f"Photon sidecar {path} reported error: {data.get('error')}" + f"Photon sidecar {path} reported error{suffix}: {data.get('error')}" ) return data diff --git a/plugins/platforms/photon/sidecar/fatal-errors.mjs b/plugins/platforms/photon/sidecar/fatal-errors.mjs new file mode 100644 index 000000000000..c2a39f25ae7f --- /dev/null +++ b/plugins/platforms/photon/sidecar/fatal-errors.mjs @@ -0,0 +1,38 @@ +// Fatal-error helpers for the Photon Spectrum sidecar. +// +// Keep this separate from index.mjs so the detection logic can be unit-tested +// without starting Spectrum or requiring real Photon credentials. + +function appendErrorParts(parts, value, seen) { + if (value == null) return; + if (typeof value === "string" || typeof value === "number" || typeof value === "boolean") { + parts.push(String(value)); + return; + } + if (typeof value !== "object") return; + if (seen.has(value)) return; + seen.add(value); + + for (const key of ["name", "message", "stack", "code", "grpcCode", "details", "path"]) { + if (value[key] != null) parts.push(String(value[key])); + } + if (value.cause) appendErrorParts(parts, value.cause, seen); + if (Array.isArray(value.errors)) { + for (const error of value.errors) appendErrorParts(parts, error, seen); + } +} + +export function errorToText(error) { + const parts = []; + appendErrorParts(parts, error, new Set()); + if (!parts.length) return String(error ?? ""); + return parts.join("\n"); +} + +export function classifyRecoverableOutboundError(error) { + const text = errorToText(error); + if (/\[upstream\]\s*Connection dropped/i.test(text)) { + return "upstream_connection_dropped"; + } + return null; +} \ No newline at end of file diff --git a/plugins/platforms/photon/sidecar/package.json b/plugins/platforms/photon/sidecar/package.json index d689c78e1760..c9d042e19b31 100644 --- a/plugins/platforms/photon/sidecar/package.json +++ b/plugins/platforms/photon/sidecar/package.json @@ -7,7 +7,8 @@ "main": "index.mjs", "scripts": { "start": "node index.mjs", - "postinstall": "node patch-spectrum-mixed-attachments.mjs" + "postinstall": "node patch-spectrum-mixed-attachments.mjs", + "test": "node --test test/" }, "engines": { "node": ">=18.17" diff --git a/plugins/platforms/photon/sidecar/test/fatal-errors.test.mjs b/plugins/platforms/photon/sidecar/test/fatal-errors.test.mjs new file mode 100644 index 000000000000..6b48d5ca48d0 --- /dev/null +++ b/plugins/platforms/photon/sidecar/test/fatal-errors.test.mjs @@ -0,0 +1,10 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { classifyRecoverableOutboundError } from "../fatal-errors.mjs"; + +test("classifies upstream connection drops as recoverable outbound send errors", () => { + const error = new Error("ConnectionError: [upstream] Connection dropped"); + + assert.equal(classifyRecoverableOutboundError(error), "upstream_connection_dropped"); +}); \ No newline at end of file diff --git a/tests/plugins/test_photon_adapter_outbound_recovery.py b/tests/plugins/test_photon_adapter_outbound_recovery.py new file mode 100644 index 000000000000..fcf277730ec9 --- /dev/null +++ b/tests/plugins/test_photon_adapter_outbound_recovery.py @@ -0,0 +1,109 @@ +import pytest + +from gateway.config import PlatformConfig +from gateway.platforms.base import SendResult +from plugins.platforms.photon.adapter import PhotonAdapter + + +@pytest.mark.asyncio +async def test_direct_send_restarts_sidecar_after_recoverable_upstream_drop(monkeypatch): + adapter = PhotonAdapter( + PlatformConfig( + enabled=True, + extra={"project_id": "pid", "project_secret": "secret"}, + ) + ) + + calls = {"send": 0, "restart": 0} + + async def fake_sidecar_send(chat_id, text): + calls["send"] += 1 + assert chat_id == "chat-1" + assert text == "hello" + if calls["send"] == 1: + return SendResult( + success=False, + error='Photon sidecar /send returned 500 error_code=upstream_connection_dropped: {"ok":false,"error":"internal sidecar error","error_code":"upstream_connection_dropped"}', + ) + return SendResult(success=True, message_id="m-direct-recovered") + + async def fake_restart(reason: str): + calls["restart"] += 1 + assert "upstream_connection_dropped" in reason + + monkeypatch.setattr(adapter, "_sidecar_send", fake_sidecar_send) + monkeypatch.setattr(adapter, "_restart_sidecar_for_outbound_error", fake_restart) + + result = await adapter.send(chat_id="chat-1", content="hello") + + assert result.success is True + assert result.message_id == "m-direct-recovered" + assert calls == {"send": 2, "restart": 1} + + +@pytest.mark.asyncio +async def test_direct_send_recovers_plain_upstream_drop_text(monkeypatch): + adapter = PhotonAdapter( + PlatformConfig( + enabled=True, + extra={"project_id": "pid", "project_secret": "secret"}, + ) + ) + + calls = {"send": 0, "restart": 0} + + async def fake_sidecar_send(chat_id, text): + calls["send"] += 1 + if calls["send"] == 1: + return SendResult( + success=False, + error="Photon sidecar /send returned 500: ConnectionError: [upstream] Connection dropped", + ) + return SendResult(success=True, message_id="m-plain-text-recovered") + + async def fake_restart(reason: str): + calls["restart"] += 1 + assert "upstream_connection_dropped" in reason + + monkeypatch.setattr(adapter, "_sidecar_send", fake_sidecar_send) + monkeypatch.setattr(adapter, "_restart_sidecar_for_outbound_error", fake_restart) + + result = await adapter.send(chat_id="chat-1", content="hello") + + assert result.success is True + assert result.message_id == "m-plain-text-recovered" + assert calls == {"send": 2, "restart": 1} + + +@pytest.mark.asyncio +async def test_send_with_retry_restarts_sidecar_after_recoverable_upstream_drop(monkeypatch): + adapter = PhotonAdapter( + PlatformConfig( + enabled=True, + extra={"project_id": "pid", "project_secret": "secret"}, + ) + ) + + calls = {"send": 0, "restart": 0} + + async def fake_send(*, chat_id, content, reply_to=None, metadata=None): + calls["send"] += 1 + if calls["send"] == 1: + return SendResult( + success=False, + error='Photon sidecar /send returned 500 error_code=upstream_connection_dropped: {"ok":false,"error":"internal sidecar error","error_code":"upstream_connection_dropped"}', + ) + return SendResult(success=True, message_id="m-recovered") + + async def fake_restart(reason: str): + calls["restart"] += 1 + assert "upstream_connection_dropped" in reason + + monkeypatch.setattr(adapter, "send", fake_send) + monkeypatch.setattr(adapter, "_restart_sidecar_for_outbound_error", fake_restart) + + result = await adapter._send_with_retry("chat-1", "/new") + + assert result.success is True + assert result.message_id == "m-recovered" + assert calls == {"send": 2, "restart": 1} \ No newline at end of file diff --git a/tests/plugins/test_photon_adapter_sidecar_restart.py b/tests/plugins/test_photon_adapter_sidecar_restart.py new file mode 100644 index 000000000000..a16cce366b8c --- /dev/null +++ b/tests/plugins/test_photon_adapter_sidecar_restart.py @@ -0,0 +1,60 @@ +from typing import Any, cast + +import pytest + +from gateway.config import PlatformConfig +from plugins.platforms.photon.adapter import PhotonAdapter + + +class _FakeProc: + def __init__(self, returncode=None): + self.returncode = returncode + + def poll(self): + return self.returncode + + +@pytest.mark.asyncio +async def test_ensure_sidecar_running_restarts_exited_sidecar(monkeypatch): + adapter = PhotonAdapter( + PlatformConfig( + enabled=True, + extra={"project_id": "pid", "project_secret": "secret"}, + ) + ) + adapter._sidecar_proc = cast(Any, _FakeProc(returncode=75)) + adapter._sidecar_supervisor_task = None + calls = [] + + async def fake_start_sidecar(): + calls.append("start") + adapter._sidecar_proc = cast(Any, _FakeProc(returncode=None)) + + monkeypatch.setattr(adapter, "_start_sidecar", fake_start_sidecar) + + await adapter._ensure_sidecar_running() + + assert calls == ["start"] + assert adapter._sidecar_proc is not None + assert adapter._sidecar_proc.poll() is None + + +@pytest.mark.asyncio +async def test_ensure_sidecar_running_keeps_live_sidecar(monkeypatch): + adapter = PhotonAdapter( + PlatformConfig( + enabled=True, + extra={"project_id": "pid", "project_secret": "secret"}, + ) + ) + adapter._sidecar_proc = cast(Any, _FakeProc(returncode=None)) + + async def fail_start_sidecar(): # pragma: no cover - should not be called + raise AssertionError("live sidecar should not restart") + + monkeypatch.setattr(adapter, "_start_sidecar", fail_start_sidecar) + + await adapter._ensure_sidecar_running() + + assert adapter._sidecar_proc is not None + assert adapter._sidecar_proc.poll() is None \ No newline at end of file