fix(photon): harden structured sidecar error classes + target_not_allowed

Maintainer follow-up to the #51193 salvage:

- _send_with_retry: permanent classes (auth_or_config, target_not_allowed)
  now short-circuit BEFORE the unconditional plain-text fallback resend,
  including when a retry attempt surfaces one — no more double-sends of
  permanently-failing requests.
- sidecar classifySidecarError: new structured code target_not_allowed for
  Spectrum's 'Target not allowed for this project' AuthenticationError
  (shared/free-tier lines cannot initiate outbound sends to new targets).
  Classification applies to every handler sharing the catch-all
  serverError path (/send, /send-attachment, /react, /typing, ...).
- _standalone_send now parses the structured error body too (it reads
  sidecar responses independently of _sidecar_call) and returns
  error_class/retryable alongside the message.
- target_not_allowed maps to a canonical user-facing message in both
  paths; raw upstream error text never leaks through the structured code.

Closes the actionable halves of #50971, #51897, #52794.

plugins/platforms/photon/adapter.py

@@ -262,6 +262,11 @@ def _sidecar_error_from_response(
     error = str(data.get("error") or text[:200] or "sidecar error")
     error_class = str(data.get("error_class") or "sidecar_error")
     retryable = bool(data.get("retryable"))
+    if error_class == "target_not_allowed":
+        # Structured code path: replace whatever the sidecar echoed with the
+        # canonical user-facing explanation — never raw upstream error text.
+        error = _TARGET_NOT_ALLOWED_MESSAGE
+        retryable = False
     return PhotonSidecarError(
         path=path,
         status_code=status_code,
@@ -270,6 +275,14 @@ def _sidecar_error_from_response(
         retryable=retryable,
     )
 
+# User-facing explanation for the Spectrum "Target not allowed for this
+# project" AuthenticationError. Shared/free-tier lines can only reply to
+# conversations the target initiated; they cannot open new outbound threads.
+_TARGET_NOT_ALLOWED_MESSAGE = (
+    "shared/free-tier Photon lines cannot initiate outbound sends to new "
+    "targets — upgrade to a dedicated line or use another delivery channel"
+)
+
 def _coerce_port(value: Any, default: int) -> int:
     try:
         return int(value)
@@ -1752,6 +1765,21 @@ class PhotonAdapter(BasePlatformAdapter):
             return False
         return any(pat in lowered for pat in _PHOTON_RETRYABLE_PATTERNS)
 
+    @staticmethod
+    def _is_permanent_sidecar_failure(result: SendResult) -> bool:
+        """True when the sidecar classified the failure as permanent.
+
+        ``auth_or_config`` and ``target_not_allowed`` cannot be fixed by
+        retrying or by the plain-text fallback resend — attempting either
+        just double-sends a doomed request (issue #50971).
+        """
+        raw = result.raw_response
+        return (
+            isinstance(raw, dict)
+            and raw.get("retryable") is False
+            and raw.get("error_class") in ("auth_or_config", "target_not_allowed")
+        )
+
     async def _send_with_retry(
         self,
         chat_id: str,
@@ -1777,12 +1805,12 @@ class PhotonAdapter(BasePlatformAdapter):
         if result.success:
             return result
 
-        sidecar_failure = result.raw_response
-        if (
-            isinstance(sidecar_failure, dict)
-            and sidecar_failure.get("error_class") == "auth_or_config"
-            and sidecar_failure.get("retryable") is False
-        ):
+        if self._is_permanent_sidecar_failure(result):
+            # Permanent failure classes: retrying cannot succeed and the
+            # unconditional plain-text fallback below would double-send the
+            # same doomed request. Return the structured failure as-is
+            # (with the user-facing explanation already attached for
+            # target_not_allowed in _sidecar_send).
             return result
 
         error_str = result.error or ""
@@ -1807,6 +1835,10 @@ class PhotonAdapter(BasePlatformAdapter):
                 if result.success:
                     return result
                 error_str = result.error or ""
+                if self._is_permanent_sidecar_failure(result):
+                    # A retry surfaced a permanent class — don't fall through
+                    # to the plain-text resend either.
+                    return result
                 if not (result.retryable or self._is_retryable_error(error_str)):
                     break
             else:
@@ -2037,6 +2069,32 @@ def _cache_inbound_attachment(
 # is not co-resident.  Reuses a live sidecar already listening on the
 # configured port (cron processes cannot spawn the sidecar themselves).
 
+def _standalone_error(resp: Any) -> Dict[str, Any]:
+    """Build the structured error dict for a failed standalone sidecar call.
+
+    Mirrors ``_sidecar_error_from_response``: parse the sidecar body
+    independently, carry the error class + retryability, and swap in the
+    canonical user-facing message for ``target_not_allowed`` (never the raw
+    upstream text).
+    """
+    try:
+        data = resp.json() or {}
+    except Exception:
+        data = {}
+    if not isinstance(data, dict):
+        data = {}
+    error_class = str(data.get("error_class") or "sidecar_error")
+    retryable = bool(data.get("retryable"))
+    if error_class == "target_not_allowed":
+        error = _TARGET_NOT_ALLOWED_MESSAGE
+        retryable = False
+    elif resp.status_code != 200:
+        error = f"sidecar returned {resp.status_code}: {resp.text[:200]}"
+    else:
+        error = str(data.get("error") or "sidecar reported failure")
+    return {"error": error, "error_class": error_class, "retryable": retryable}
+
+
 async def _standalone_send(
     pconfig: PlatformConfig,
     chat_id: str,
@@ -2097,10 +2155,10 @@ async def _standalone_send(
                     f"{base}/send", json=send_body, headers=headers,
                 )
                 if resp.status_code != 200:
-                    return {"error": f"sidecar returned {resp.status_code}: {resp.text[:200]}"}
+                    return _standalone_error(resp)
                 data = resp.json() or {}
                 if not data.get("ok"):
-                    return {"error": data.get("error") or "sidecar reported failure"}
+                    return _standalone_error(resp)
                 last_message_id = data.get("messageId")
 
             # 2. Each attachment as a separate /send-attachment call.
@@ -2125,10 +2183,10 @@ async def _standalone_send(
                     f"{base}/send-attachment", json=att_body, headers=headers,
                 )
                 if resp.status_code != 200:
-                    return {"error": f"sidecar returned {resp.status_code}: {resp.text[:200]}"}
+                    return _standalone_error(resp)
                 data = resp.json() or {}
                 if not data.get("ok"):
-                    return {"error": data.get("error") or "sidecar reported failure"}
+                    return _standalone_error(resp)
                 last_message_id = data.get("messageId") or last_message_id
 
         return {"success": True, "message_id": last_message_id}

plugins/platforms/photon/sidecar/index.mjs

@@ -600,6 +600,15 @@ function classifySidecarError(err) {
   const message = String(err && err.message ? err.message : err || "");
   const lowered = message.toLowerCase();
 
+  // Spectrum raises AuthenticationError("Target not allowed for this
+  // project") from space.send when a shared/free-tier line tries to
+  // initiate an outbound conversation with a new target. That is a plan
+  // limitation, not a transient fault — surface a dedicated class so the
+  // adapter can explain it instead of retrying (issues #50971/#51897).
+  if (lowered.includes("target not allowed")) {
+    return { errorClass: "target_not_allowed", retryable: false };
+  }
+
   if (
     lowered.includes("unauthorized") ||
     lowered.includes("forbidden") ||

tests/plugins/platforms/photon/test_overflow_recovery.py

@@ -520,3 +520,118 @@ async def test_external_disconnect_still_cancels_supervisor(
         "External cleanup must still cancel a running supervisor task"
     )
     assert adapter._sidecar_supervisor_task is None
+
+
+# -- target_not_allowed: shared/free-tier outbound-send restriction ----------
+#
+# Spectrum throws AuthenticationError("Target not allowed for this project")
+# from space.send when a shared/free-tier line initiates an outbound send to
+# a new target. The sidecar classifies it as the structured code
+# `target_not_allowed`; the adapter must treat it as permanent in BOTH
+# _send_with_retry and _standalone_send, surfacing the canonical user-facing
+# message instead of raw upstream error text (issues #50971 / #51897).
+
+
+def test_target_not_allowed_maps_to_canonical_message() -> None:
+    err = photon_adapter._sidecar_error_from_response(
+        "/send",
+        500,
+        '{"ok":false,"error":"internal sidecar error",'
+        '"error_class":"target_not_allowed","retryable":false}',
+    )
+
+    assert err.error_class == "target_not_allowed"
+    assert err.retryable is False
+    assert err.error == photon_adapter._TARGET_NOT_ALLOWED_MESSAGE
+    # No raw upstream text may leak through the structured code path.
+    assert "Target not allowed for this project" not in str(err)
+
+
+def test_target_not_allowed_is_not_legacy_retryable() -> None:
+    error = str(
+        photon_adapter.PhotonSidecarError(
+            path="/send",
+            status_code=500,
+            error=photon_adapter._TARGET_NOT_ALLOWED_MESSAGE,
+            error_class="target_not_allowed",
+            retryable=False,
+        )
+    )
+
+    assert PhotonAdapter._is_retryable_error(error) is False
+
+
+@pytest.mark.asyncio
+async def test_send_with_retry_returns_immediately_on_target_not_allowed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    adapter = _make_adapter(monkeypatch)
+    calls = 0
+
+    async def _fake_sidecar_call(path: str, body: Dict[str, Any]) -> Dict[str, Any]:
+        nonlocal calls
+        calls += 1
+        raise photon_adapter._sidecar_error_from_response(
+            path,
+            500,
+            '{"ok":false,"error":"internal sidecar error",'
+            '"error_class":"target_not_allowed","retryable":false}',
+        )
+
+    monkeypatch.setattr(adapter, "_sidecar_call", _fake_sidecar_call)
+
+    result = await adapter._send_with_retry("space-1", "hello", max_retries=3)
+
+    assert result.success is False
+    assert result.retryable is False
+    assert calls == 1, "permanent target_not_allowed must not be retried or resent"
+    assert photon_adapter._TARGET_NOT_ALLOWED_MESSAGE in (result.error or "")
+    assert isinstance(result.raw_response, dict)
+    assert result.raw_response.get("error_class") == "target_not_allowed"
+
+
+@pytest.mark.asyncio
+async def test_standalone_send_classifies_target_not_allowed(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("PHOTON_SIDECAR_TOKEN", "token")
+
+    class _Resp:
+        status_code = 500
+        text = (
+            '{"ok":false,"error":"internal sidecar error",'
+            '"error_class":"target_not_allowed","retryable":false}'
+        )
+
+        @staticmethod
+        def json() -> Dict[str, Any]:
+            return {
+                "ok": False,
+                "error": "internal sidecar error",
+                "error_class": "target_not_allowed",
+                "retryable": False,
+            }
+
+    class _FakeClient:
+        def __init__(self, *a: Any, **k: Any) -> None:
+            pass
+
+        async def __aenter__(self) -> "_FakeClient":
+            return self
+
+        async def __aexit__(self, *a: Any) -> bool:
+            return False
+
+        async def post(self, *a: Any, **k: Any) -> _Resp:
+            return _Resp()
+
+    monkeypatch.setattr(photon_adapter.httpx, "AsyncClient", _FakeClient)
+
+    result = await photon_adapter._standalone_send(
+        PlatformConfig(name="photon", enabled=True), "space-1", "hello",
+    )
+
+    assert result.get("error") == photon_adapter._TARGET_NOT_ALLOWED_MESSAGE
+    assert result.get("error_class") == "target_not_allowed"
+    assert result.get("retryable") is False
+    assert "Target not allowed for this project" not in str(result)