feat(gateway): route platform HTTP event callbacks (#36061 salvage)

gateway/platforms/api_server.py

@@ -995,6 +995,10 @@ def __init__(self, config: PlatformConfig):
         # (the /v1/runs path tracks its own in-flight set via
         # _active_run_tasks).
         self._inflight_agent_runs: int = 0
+        # Back-reference to the owning GatewayRunner (set by gateway/run.py)
+        # so /api/platforms/{platform}/events can resolve sibling adapters.
+        # BasePlatformAdapter declares the class-level default of None.
+        self.gateway_runner: Optional[Any] = None
         # Requests admitted before their handler reaches agent bookkeeping.
         # Shutdown counts this reservation so the request cannot slip through
         # the drain between its first await and _run_agent()/task registration.
@@ -1248,6 +1252,131 @@ def _check_auth(self, request: "web.Request") -> Optional["web.Response"]:
             status=401,
         )
 
+    @staticmethod
+    def _normalize_callback_platform(value: str) -> str:
+        normalized = (value or "").strip().lower().replace("-", "_")
+        if not re.fullmatch(r"[a-z0-9_]+", normalized):
+            return ""
+        return normalized
+
+    def _get_platform_callback_adapter(
+        self,
+        request: "web.Request",
+        platform_name: str,
+    ) -> Optional[Any]:
+        injected = request.app.get("platform_event_adapters")
+        if isinstance(injected, dict):
+            adapter = injected.get(platform_name)
+            if adapter is not None:
+                return adapter
+
+        adapter = request.app.get(f"{platform_name}_adapter")
+        if adapter is not None:
+            return adapter
+
+        runner = self.gateway_runner or request.app.get("gateway_runner")
+        adapters = getattr(runner, "adapters", None)
+        if not adapters:
+            return None
+
+        try:
+            from gateway.config import Platform as _Platform
+            return adapters.get(_Platform(platform_name))
+        except Exception:
+            for platform, candidate in adapters.items():
+                if getattr(platform, "value", platform) == platform_name:
+                    return candidate
+        return None
+
+    async def _handle_platform_event_callback(self, request: "web.Request") -> "web.Response":
+        platform_name = self._normalize_callback_platform(
+            request.match_info.get("platform", "")
+        )
+        if not platform_name:
+            return web.json_response(
+                _openai_error(
+                    "Invalid platform name",
+                    code="invalid_platform",
+                ),
+                status=400,
+            )
+
+        adapter = self._get_platform_callback_adapter(request, platform_name)
+        if adapter is None:
+            return web.json_response(
+                _openai_error(
+                    "Platform adapter is not connected",
+                    code="platform_unavailable",
+                ),
+                status=503,
+            )
+
+        verifier = getattr(adapter, "verify_http_event_request", None)
+        dispatcher = getattr(adapter, "dispatch_http_event", None)
+        if verifier is None or dispatcher is None:
+            return web.json_response(
+                _openai_error(
+                    "Platform adapter does not support HTTP events",
+                    code="platform_http_events_unsupported",
+                ),
+                status=503,
+            )
+
+        auth_header = request.headers.get("Authorization", "")
+        try:
+            if asyncio.iscoroutinefunction(verifier):
+                ok, code = await verifier(auth_header)
+            else:
+                # Platform verifiers may do blocking network I/O (e.g. Google
+                # signing-cert fetches) — keep that off the event loop.
+                ok, code = await asyncio.to_thread(verifier, auth_header)
+        except Exception:
+            # Fail closed: a crashing verifier must never admit the event.
+            logger.exception(
+                "Platform HTTP event verifier failed for %s", platform_name
+            )
+            ok, code = False, "platform_event_verifier_error"
+        if not ok:
+            return web.json_response(
+                _openai_error(
+                    "Invalid platform event authorization",
+                    code=code or "invalid_platform_event_authorization",
+                ),
+                status=401,
+            )
+
+        try:
+            payload = await request.json()
+        except Exception:
+            return web.json_response(
+                _openai_error("Invalid JSON in platform event", code="invalid_json"),
+                status=400,
+            )
+
+        if not isinstance(payload, dict):
+            return web.json_response(
+                _openai_error(
+                    "Platform event must be a JSON object",
+                    code="invalid_request",
+                ),
+                status=400,
+            )
+
+        try:
+            result = await dispatcher(payload)
+        except Exception:
+            logger.exception("Platform HTTP event dispatch failed for %s", platform_name)
+            return web.json_response(
+                _openai_error(
+                    "Platform event dispatch failed",
+                    err_type="server_error",
+                    code="platform_event_dispatch_failed",
+                ),
+                status=500,
+            )
+
+        return web.json_response(result if isinstance(result, dict) else {})
+
     # ------------------------------------------------------------------
     # Multi-profile multiplexing (/p/<profile>/…)
     # ------------------------------------------------------------------
@@ -1339,6 +1468,10 @@ def _http_route_table(self) -> List[tuple]:
             ("POST", "/v1/responses", self._handle_responses),
             ("GET", "/v1/responses/{response_id}", self._handle_get_response),
             ("DELETE", "/v1/responses/{response_id}", self._handle_delete_response),
+            # Generic platform HTTP event callback ingress. Authenticated by
+            # the target adapter's own verifier (platform-signed bearer), NOT
+            # API_SERVER_KEY — external platforms hold no API server key.
+            ("POST", "/api/platforms/{platform}/events", self._handle_platform_event_callback),
             ("GET", "/api/jobs", self._handle_list_jobs),
             ("POST", "/api/jobs", self._handle_create_job),
             ("GET", "/api/jobs/{job_id}", self._handle_get_job),
@@ -5149,6 +5282,8 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
             # native routes first lets those shims no-op instead of shadowing the
             # upstream session-control handlers.
             self._app["api_server_adapter"] = self
+            if self.gateway_runner is not None:
+                self._app["gateway_runner"] = self.gateway_runner
 
             # Start background sweep to clean up orphaned (unconsumed) run streams
             sweep_task = asyncio.create_task(self._sweep_orphaned_runs())

gateway/run.py

@@ -9043,7 +9043,9 @@ def _create_adapter(
             if not check_api_server_requirements():
                 logger.warning("API Server: aiohttp not installed")
                 return None
-            return APIServerAdapter(config)
+            adapter = APIServerAdapter(config)
+            adapter.gateway_runner = self
+            return adapter
 
         elif platform == Platform.WEBHOOK:
             from gateway.platforms.webhook import WebhookAdapter, check_webhook_requirements

plugins/platforms/google_chat/adapter.py

@@ -43,6 +43,8 @@
 import os
 import random
 import re
+import threading
+import time
 from pathlib import Path as _Path
 from typing import Any, Callable, Dict, List, Optional, Tuple
 
@@ -71,6 +73,59 @@
 MediaFileUpload: Any = None  # type: ignore
 
 _google_modules_loaded: bool = False
+_GOOGLE_ID_TOKEN_CERTS_TTL_SECONDS = 300
+_google_id_token_request: Any = None
+_google_id_token_request_lock = threading.Lock()
+
+
+class _CachedGoogleAuthRequest:
+    def __init__(self, request: Any, ttl_seconds: int = _GOOGLE_ID_TOKEN_CERTS_TTL_SECONDS) -> None:
+        self._request = request
+        self._ttl_seconds = ttl_seconds
+        self._lock = threading.Lock()
+        self._cache: Dict[Tuple[str, str], Tuple[float, Any]] = {}
+
+    def __call__(self, url: str, method: str = "GET", **kwargs: Any) -> Any:
+        cache_key = (method.upper(), url)
+        if cache_key[0] != "GET":
+            return self._request(url=url, method=method, **kwargs)
+
+        now = time.monotonic()
+        with self._lock:
+            cached = self._cache.get(cache_key)
+            if cached and cached[0] > now:
+                return cached[1]
+
+        response = self._request(url=url, method=method, **kwargs)
+        if getattr(response, "status", None) == 200:
+            with self._lock:
+                self._cache[cache_key] = (now + self._ttl_seconds, response)
+        return response
+
+
+def _get_google_id_token_request() -> Any:
+    global _google_id_token_request
+    with _google_id_token_request_lock:
+        if _google_id_token_request is None:
+            try:
+                from google.auth.transport import requests as google_requests
+            except ImportError as exc:
+                raise RuntimeError("google-auth is required for Google Chat HTTP callbacks") from exc
+            _google_id_token_request = _CachedGoogleAuthRequest(google_requests.Request())
+        return _google_id_token_request
+
+
+def _verify_google_id_token(token: str, audience: str) -> Dict[str, Any]:
+    try:
+        from google.oauth2 import id_token
+    except ImportError as exc:
+        raise RuntimeError("google-auth is required for Google Chat HTTP callbacks") from exc
+
+    return id_token.verify_oauth2_token(
+        token,
+        _get_google_id_token_request(),
+        audience,
+    )
 
 
 def _load_google_modules() -> bool:
@@ -690,6 +745,21 @@ def __init__(self, config: PlatformConfig):
             self._max_bytes = int(os.getenv("GOOGLE_CHAT_MAX_BYTES", str(16 * 1024 * 1024)))
         except (ValueError, TypeError):
             self._max_bytes = 16 * 1024 * 1024
+        self._http_events_url = (
+            self.config.extra.get("http_events_url")
+            or os.getenv("GOOGLE_CHAT_HTTP_EVENTS_URL", "")
+            or ""
+        ).strip()
+        self._http_events_audience = (
+            self.config.extra.get("http_events_audience")
+            or os.getenv("GOOGLE_CHAT_HTTP_EVENTS_AUDIENCE", "")
+            or self._http_events_url
+        ).strip()
+        self._http_events_service_account_email = (
+            self.config.extra.get("http_events_service_account_email")
+            or os.getenv("GOOGLE_CHAT_HTTP_EVENTS_SERVICE_ACCOUNT_EMAIL", "")
+            or ""
+        ).strip().lower()
 
     # ------------------------------------------------------------------
     # Configuration loading and validation
@@ -1422,6 +1492,62 @@ def _on_pubsub_message(self, message: Any) -> None:
             except Exception:
                 pass
 
+    async def dispatch_http_event(self, envelope: Dict[str, Any]) -> Dict[str, Any]:
+        extracted = self._extract_message_payload(envelope)
+        if extracted is None:
+            return {}
+
+        msg, space, _fmt = extracted
+        sender = msg.get("sender") or {}
+        if sender.get("type") == "BOT":
+            return {}
+
+        msg_name = msg.get("name") or ""
+        if msg_name and self._dedup.is_duplicate(msg_name):
+            return {}
+
+        msg_with_space = dict(msg)
+        if "space" not in msg_with_space and space:
+            msg_with_space["space"] = space
+
+        enriched_env = dict(envelope)
+        if "space" not in enriched_env and space:
+            enriched_env["space"] = space
+
+        await self._dispatch_message(msg_with_space, enriched_env)
+        return {}
+
+    def verify_http_event_request(self, auth_header: str) -> Tuple[bool, str]:
+        if not self._http_events_audience or not self._http_events_service_account_email:
+            return False, "google_chat_http_events_not_configured"
+
+        if not auth_header.startswith("Bearer "):
+            return False, "missing_google_bearer"
+
+        token = auth_header[7:].strip()
+        if not token:
+            return False, "missing_google_bearer"
+
+        try:
+            claims = _verify_google_id_token(token, self._http_events_audience)
+        except Exception as exc:
+            logger.warning(
+                "[GoogleChat] HTTP event bearer verification failed: %s",
+                _redact_sensitive(str(exc)),
+            )
+            return False, "invalid_google_bearer"
+
+        expected = {
+            item.strip().lower()
+            for item in self._http_events_service_account_email.split(",")
+            if item.strip()
+        }
+        claim_email = str(claims.get("email") or "").strip().lower()
+        if not claim_email or claim_email not in expected:
+            return False, "unexpected_google_bearer_identity"
+
+        return True, ""
+
     async def _dispatch_message(self, msg: Dict[str, Any], envelope: Dict[str, Any]) -> None:
         """Translate a Chat message payload to a MessageEvent and hand off.
 
@@ -3270,6 +3396,12 @@ def _env_enablement() -> Optional[Dict[str, Any]]:
         seed["subscription_name"] = subscription
     if http_events_url:
         seed["http_events_url"] = http_events_url
+    http_events_audience = os.getenv("GOOGLE_CHAT_HTTP_EVENTS_AUDIENCE")
+    if http_events_audience:
+        seed["http_events_audience"] = http_events_audience
+    http_events_sa_email = os.getenv("GOOGLE_CHAT_HTTP_EVENTS_SERVICE_ACCOUNT_EMAIL")
+    if http_events_sa_email:
+        seed["http_events_service_account_email"] = http_events_sa_email
     sa_json = (
         os.getenv("GOOGLE_CHAT_SERVICE_ACCOUNT_JSON")
         or os.getenv("GOOGLE_APPLICATION_CREDENTIALS")

plugins/platforms/google_chat/plugin.yaml

@@ -23,6 +23,14 @@ optional_env:
     description: "Authenticated HTTP endpoint for Chat message events."
     prompt: "HTTP events callback URL"
     password: false
+  - name: GOOGLE_CHAT_HTTP_EVENTS_AUDIENCE
+    description: "Expected audience for Google-signed HTTP event bearer tokens. Defaults to GOOGLE_CHAT_HTTP_EVENTS_URL."
+    prompt: "HTTP events token audience"
+    password: false
+  - name: GOOGLE_CHAT_HTTP_EVENTS_SERVICE_ACCOUNT_EMAIL
+    description: "Expected Google service account email for HTTP event bearer tokens."
+    prompt: "HTTP events service account email"
+    password: false
   - name: GOOGLE_CHAT_PROJECT_ID
     description: "GCP project ID for optional Pub/Sub inbound mode. Falls back to GOOGLE_CLOUD_PROJECT."
     prompt: "GCP project ID"