Photon/Envoy overflow errors retry and resilience

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #50339 fix(photon): ride through Spectrum/Envoy upstream overflow (retry + typing cooldown + crash detection)

  merged to main

  files: plugins/platforms/photon/adapter.py, tests/plugins/platforms/photon/test_overflow_recovery.py

  Show the diff

  • plugins/platforms/photon/adapter.py

    @@ -85,6 +85,20 @@
     
     _SIDECAR_DIR = Path(__file__).parent / "sidecar"
     
    +# Photon / Envoy / spectrum-ts error substrings that indicate a transient
    +# upstream overload rather than a permanent failure.  These are not in the
    +# core _RETRYABLE_ERROR_PATTERNS because they are specific to this adapter.
    +_PHOTON_RETRYABLE_PATTERNS = (
    +    "internal sidecar error",
    +    "upstream connect error",
    +    "reset reason: overflow",
    +)
    +
    +# Minimum seconds between typing-indicator calls for the same chat.
    +# iMessage is a personal channel — suppressing rapid repeats reduces
    +# upstream gRPC pressure during Photon overflow events.
    +_TYPING_COOLDOWN_SECONDS = 5.0
    +
     # Group-chat mention wake words. When ``require_mention`` is enabled, group
     # messages are ignored unless they match one of these patterns — same
     # behavior and defaults as the BlueBubbles iMessage channel so the two
    @@ -234,6 +248,8 @@ def __init__(self, config: PlatformConfig):
             # react action default to "the message that triggered me" without
             # requiring the model to thread message ids through tool calls.
             self._last_inbound_by_chat: Dict[str, str] = {}
    +        # Last time we sent a typing indicator per chat, for cooldown gating.
    +        self._typing_last_sent: Dict[str, float] = {}
     
             # Group-chat mention gating (parity with BlueBubbles). When enabled,
             # group messages are ignored unless they match a wake word; DMs are
    @@ -839,6 +855,21 @@ async def _supervise_sidecar(self, proc: subprocess.Popen) -> None:
                     logger.info("[photon-sidecar] %s", line.decode("utf-8", "replace").rstrip())
             except Exception as e:  # pragma: no cover - defensive
                 logger.warning("[photon-sidecar] supervisor exited: %s", e)
    +        if self._inbound_running:
    +            exit_code = proc.poll()
    +            logger.error(
    +                "[photon] sidecar exited unexpectedly (code %s) — triggering reconnect",
    +                exit_code,
    +            )
    +            self._set_fatal_error(
    +                "SIDECAR_CRASHED",
    +                f"Photon sidecar exited unexpectedly (code {exit_code})",
    +                retryable=True,
    +            )
    +            try:
    +                await self._notify_fatal_error()
    +            except Exception as exc:  # pragma: no cover - defensive
    +                logger.warning("[photon] fatal-error notification failed: %s", exc)
     
         async def _stop_sidecar(self) -> None:
             proc = self._sidecar_proc
    @@ -988,6 +1019,10 @@ async def send_animation(
             )
     
         async def send_typing(self, chat_id: str, metadata=None) -> None:
    +        now = time.time()
    +        if now - self._typing_last_sent.get(chat_id, 0.0) < _TYPING_COOLDOWN_SECONDS:
    +            return
    +        self._typing_last_sent[chat_id] = now
             try:
                 await self._sidecar_call(
                     "/typing", {"spaceId": chat_id, "state": "start"}
    @@ -996,6 +1031,7 @@ async def send_typing(self, chat_id: str, metadata=None) -> None:
                 logger.debug("[photon] send_typing failed: %s", e)
     
         async def stop_typing(self, chat_id: str) -> None:
    +        self._typing_last_sent.pop(chat_id, None)
             try:
                 await self._sidecar_call(
                     "/typing", {"spaceId": chat_id, "state": "stop"}
    @@ -1189,13 +1225,22 @@ def format_message(self, content: str) -> str:
                 return content
             return strip_markdown(content)
     
    +    @staticmethod
    +    def _is_retryable_error(error: Optional[str]) -> bool:
    +        if BasePlatformAdapter._is_retryable_error(error):
    +            return True
    +        if not error:
    +            return False
    +        lowered = error.lower()
    +        return any(pat in lowered for pat in _PHOTON_RETRYABLE_PATTERNS)
    +
         async def _send_with_retry(
             self,
             chat_id: str,
             content: str,
             reply_to: Optional[str] = None,
             metadata: Any = None,
    -        max_retries: int = 2,
    +        max_retries: int = 1,
             base_delay: float = 2.0,
         ) -> SendResult:
             """Retry sends without the generic Markdown banner.

  • tests/plugins/platforms/photon/test_overflow_recovery.py

    @@ -0,0 +1,197 @@
    +"""Photon adapter resilience to transient Spectrum/Envoy upstream overflow.
    +
    +Covers the three behaviors that let the adapter ride through a Photon
    +"reset reason: overflow" event instead of degrading delivery and silently
    +dying (issue #50185):
    +
    +  1. ``_is_retryable_error`` classifies the Envoy/sidecar overflow strings as
    +     retryable so ``_send_with_retry`` actually engages its backoff loop.
    +  2. ``send_typing`` is rate-gated per chat, and ``stop_typing`` resets the
    +     gate so the next turn's typing indicator fires immediately.
    +  3. ``_supervise_sidecar`` detects an unexpected sidecar exit and raises a
    +     ``retryable=True`` fatal so the gateway reconnect watcher revives the
    +     platform — instead of returning silently and leaving ``_inbound_loop``
    +     spinning against a dead port.
    +
    +No Node sidecar is spawned and no ports are bound.
    +"""
    +from __future__ import annotations
    +
    +from typing import Any, Dict
    +
    +import pytest
    +
    +from gateway.config import PlatformConfig
    +from plugins.platforms.photon.adapter import PhotonAdapter
    +
    +
    +def _make_adapter(monkeypatch: pytest.MonkeyPatch) -> PhotonAdapter:
    +    monkeypatch.setenv("PHOTON_PROJECT_ID", "test-project-id")
    +    monkeypatch.setenv("PHOTON_PROJECT_SECRET", "test-project-secret")
    +    cfg = PlatformConfig(enabled=True, token="", extra={})
    +    return PhotonAdapter(cfg)
    +
    +
    +# -- Gap 1: retryable classification of overflow errors ---------------------
    +
    [redacted:email](
    +    "error",
    +    [
    +        "UNAVAILABLE: internal sidecar error",
    +        "upstream connect error or disconnect/reset before headers",
    +        "reset reason: overflow",
    +        # Case-insensitive: real strings arrive with mixed case.
    +        "Internal Sidecar Error",
    +    ],
    +)
    +def test_overflow_strings_classified_retryable(error: str) -> None:
    +    assert PhotonAdapter._is_retryable_error(error) is True
    +
    +
    +def test_unrelated_error_not_retryable() -> None:
    +    # A genuine permanent failure must NOT be retried.
    +    assert PhotonAdapter._is_retryable_error("400 bad request: invalid spaceId") is False
    +    assert PhotonAdapter._is_retryable_error(None) is False
    +
    +
    +def test_base_network_patterns_still_match() -> None:
    +    # The override delegates to the base classifier first, so generic
    +    # network strings keep working.
    +    assert PhotonAdapter._is_retryable_error("ConnectError: connection refused") is True
    +
    +
    +# -- Gap 2: typing-indicator cooldown ---------------------------------------
    +
    [redacted:email]
    +async def test_typing_cooldown_suppresses_rapid_repeats(
    +    monkeypatch: pytest.MonkeyPatch,
    +) -> None:
    +    adapter = _make_adapter(monkeypatch)
    +    calls: list[Dict[str, Any]] = []
    +
    +    async def _fake_call(path: str, payload: Dict[str, Any]) -> Any:
    +        calls.append(payload)
    +        return {"ok": True}
    +
    +    monkeypatch.setattr(adapter, "_sidecar_call", _fake_call)
    +
    +    # First call fires; immediate repeats are suppressed by the cooldown.
    +    await adapter.send_typing("chat-1")
    +    await adapter.send_typing("chat-1")
    +    await adapter.send_typing("chat-1")
    +
    +    assert len(calls) == 1
    +
    +
    [redacted:email]
    +async def test_typing_cooldown_is_per_chat(
    +    monkeypatch: pytest.MonkeyPatch,
    +) -> None:
    +    adapter = _make_adapter(monkeypatch)
    +    calls: list[str] = []
    +
    +    async def _fake_call(path: str, payload: Dict[str, Any]) -> Any:
    +        calls.append(payload["spaceId"])
    +        return {"ok": True}
    +
    +    monkeypatch.setattr(adapter, "_sidecar_call", _fake_call)
    +
    +    # Different chats have independent cooldowns.
    +    await adapter.send_typing("chat-1")
    +    await adapter.send_typing("chat-2")
    +
    +    assert calls == ["chat-1", "chat-2"]
    +
    +
    [redacted:email]
    +async def test_stop_typing_resets_cooldown(
    +    monkeypatch: pytest.MonkeyPatch,
    +) -> None:
    +    adapter = _make_adapter(monkeypatch)
    +    starts = 0
    +
    +    async def _fake_call(path: str, payload: Dict[str, Any]) -> Any:
    +        nonlocal starts
    +        if payload.get("state") == "start":
    +            starts += 1
    +        return {"ok": True}
    +
    +    monkeypatch.setattr(adapter, "_sidecar_call", _fake_call)
    +
    +    # A start, then a stop (end of turn), then a start for the next turn must
    +    # fire immediately — the cooldown only suppresses rapid consecutive starts
    +    # without an intervening stop.
    +    await adapter.send_typing("chat-1")
    +    await adapter.stop_typing("chat-1")
    +    await adapter.send_typing("chat-1")
    +
    +    assert starts == 2
    +
    +
    +# -- Gap 3: sidecar crash detection -----------------------------------------
    +
    +class _EofStdout:
    +    """A proc.stdout whose readline() reports immediate EOF (dead sidecar)."""
    +
    +    def readline(self) -> bytes:
    +        return b""
    +
    +
    +class _DeadProc:
    +    """Minimal subprocess.Popen stand-in for a sidecar that has exited."""
    +
    +    def __init__(self, exit_code: int = 1) -> None:
    +        self.stdout = _EofStdout()
    +        self.stdin = None
    +        self._exit_code = exit_code
    +
    +    def poll(self) -> int:
    +        return self._exit_code
    +
    +
    [redacted:email]
    +async def test_unexpected_sidecar_exit_raises_retryable_fatal(
    +    monkeypatch: pytest.MonkeyPatch,
    +) -> None:
    +    adapter = _make_adapter(monkeypatch)
    +    # Simulate a live session whose sidecar then dies underneath it.
    +    adapter._inbound_running = True
    +
    +    notified: list[bool] = []
    +
    +    async def _fake_notify() -> None:
    +        notified.append(True)
    +
    +    monkeypatch.setattr(adapter, "_notify_fatal_error", _fake_notify)
    +
    +    await adapter._supervise_sidecar(_DeadProc(exit_code=137))  # type: ignore[arg-type]
    +
    +    assert adapter.has_fatal_error is True
    +    assert adapter.fatal_error_code == "SIDECAR_CRASHED"
    +    # retryable=True routes the platform into the reconnect watcher rather
    +    # than crashing the whole gateway.
    +    assert adapter.fatal_error_retryable is True
    +    assert adapter._running is False
    +    assert notified == [True]
    +
    +
    [redacted:email]
    +async def test_clean_shutdown_does_not_raise_fatal(
    +    monkeypatch: pytest.MonkeyPatch,
    +) -> None:
    +    adapter = _make_adapter(monkeypatch)
    +    # disconnect() sets _inbound_running = False before stopping the sidecar,
    +    # so the detection block must NOT fire on a clean shutdown.
    +    adapter._inbound_running = False
    +
    +    notified: list[bool] = []
    +
    +    async def _fake_notify() -> None:
    +        notified.append(True)
    +
    +    monkeypatch.setattr(adapter, "_notify_fatal_error", _fake_notify)
    +
    +    await adapter._supervise_sidecar(_DeadProc(exit_code=0))  # type: ignore[arg-type]
    +
    +    assert adapter.has_fatal_error is False
    +    assert notified == []