From f61169861ae82a55a2146f185c048cdef827a94f Mon Sep 17 00:00:00 2001 From: aeyeopsdev <275853971+aeyeopsdev@users.noreply.github.com> Date: Sun, 31 May 2026 18:57:44 +0000 Subject: [PATCH] fix(google-chat): allow http inbound without pubsub --- plugins/platforms/google_chat/adapter.py | 161 ++++++++++++---------- plugins/platforms/google_chat/plugin.yaml | 31 +++-- tests/gateway/test_google_chat.py | 49 ++++++- 3 files changed, 154 insertions(+), 87 deletions(-) diff --git a/plugins/platforms/google_chat/adapter.py b/plugins/platforms/google_chat/adapter.py index 0e7e180f4..ae8b80a63 100644 --- a/plugins/platforms/google_chat/adapter.py +++ b/plugins/platforms/google_chat/adapter.py @@ -1,9 +1,9 @@ """ Google Chat platform adapter. -Uses Google Cloud Pub/Sub (pull subscription) for inbound events and the -Google Chat REST API for outbound messages. Pattern parallels Slack Socket -Mode and Telegram long-polling: no public endpoint required. +Uses authenticated HTTP callbacks or Google Cloud Pub/Sub for inbound +events and the Google Chat REST API for outbound messages. Pub/Sub remains +available for no-public-URL deployments. Concurrency model ----------------- @@ -767,33 +767,42 @@ class GoogleChatAdapter(BasePlatformAdapter): ) return credentials - def _validate_config(self) -> Tuple[str, str]: + def _validate_config(self) -> Tuple[str, Optional[str]]: """Return (project_id, subscription_path) after validation. - Raises ValueError with a sanitized message on any config problem. + ``subscription_path`` is ``None`` for HTTP-inbound deployments. Raises + ValueError with a sanitized message on any config problem. """ - project_id = self.config.extra.get("project_id") - subscription = self.config.extra.get("subscription_name") + project_id = (self.config.extra.get("project_id") or "").strip() + subscription = (self.config.extra.get("subscription_name") or "").strip() + http_events_url = (self.config.extra.get("http_events_url") or "").strip() + + if subscription: + match = _SUBSCRIPTION_PATH_RE.match(subscription) + if not match: + raise ValueError( + "GOOGLE_CHAT_SUBSCRIPTION_NAME must match " + "'projects//subscriptions/'." + ) + subscription_project = match.group("project") + if project_id and subscription_project != project_id: + raise ValueError( + "project_id in GOOGLE_CHAT_PROJECT_ID does not match the " + "project embedded in GOOGLE_CHAT_SUBSCRIPTION_NAME." + ) + return project_id or subscription_project, subscription + + if http_events_url: + return project_id, None + if not project_id: raise ValueError( "GOOGLE_CHAT_PROJECT_ID (or GOOGLE_CLOUD_PROJECT) is not set." ) - if not subscription: - raise ValueError( - "GOOGLE_CHAT_SUBSCRIPTION_NAME (or GOOGLE_CHAT_SUBSCRIPTION) is not set." - ) - match = _SUBSCRIPTION_PATH_RE.match(subscription) - if not match: - raise ValueError( - "GOOGLE_CHAT_SUBSCRIPTION_NAME must match " - "'projects//subscriptions/'." - ) - if match.group("project") != project_id: - raise ValueError( - "project_id in GOOGLE_CHAT_PROJECT_ID does not match the " - "project embedded in GOOGLE_CHAT_SUBSCRIPTION_NAME." - ) - return project_id, subscription + raise ValueError( + "GOOGLE_CHAT_SUBSCRIPTION_NAME (or GOOGLE_CHAT_SUBSCRIPTION) is not set. " + "Set GOOGLE_CHAT_HTTP_EVENTS_URL for HTTP callback mode." + ) # ------------------------------------------------------------------ # Loop bridge helpers (thread -> asyncio loop) @@ -1004,36 +1013,37 @@ class GoogleChatAdapter(BasePlatformAdapter): "all threads as fresh)", exc_info=True, ) - # Sanity check: subscription exists / SA has access. - self._subscriber = pubsub_v1.SubscriberClient(credentials=credentials) - try: - await asyncio.to_thread( - lambda: self._subscriber.get_subscription( - request={"subscription": subscription_path} + if subscription_path is not None: + # Sanity check: subscription exists / SA has access. + self._subscriber = pubsub_v1.SubscriberClient(credentials=credentials) + try: + await asyncio.to_thread( + lambda: self._subscriber.get_subscription( + request={"subscription": subscription_path} + ) ) - ) - except gax_exceptions.NotFound: - self._set_fatal_error( - code="subscription_not_found", - message="Pub/Sub subscription not found at configured path", - retryable=False, - ) - return False - except gax_exceptions.PermissionDenied: - self._set_fatal_error( - code="subscription_permission", - message=( - "Service Account lacks roles/pubsub.subscriber on the " - "subscription" - ), - retryable=False, - ) - return False - except Exception as exc: - msg = _redact_sensitive(str(exc)) - logger.error("[GoogleChat] subscription.get failed: %s", msg) - self._set_fatal_error(code="subscription_check", message=msg, retryable=True) - return False + except gax_exceptions.NotFound: + self._set_fatal_error( + code="subscription_not_found", + message="Pub/Sub subscription not found at configured path", + retryable=False, + ) + return False + except gax_exceptions.PermissionDenied: + self._set_fatal_error( + code="subscription_permission", + message=( + "Service Account lacks roles/pubsub.subscriber on the " + "subscription" + ), + retryable=False, + ) + return False + except Exception as exc: + msg = _redact_sensitive(str(exc)) + logger.error("[GoogleChat] subscription.get failed: %s", msg) + self._set_fatal_error(code="subscription_check", message=msg, retryable=True) + return False # Resolve bot user_id (eager): cache first, then members.list. self._bot_user_id = self._load_cached_bot_id() @@ -1047,14 +1057,22 @@ class GoogleChatAdapter(BasePlatformAdapter): "will resolve on first addedToSpace or member lookup" ) - # Start the supervisor task that runs the Pub/Sub pull with exponential - # backoff + jitter on transient errors, bails out after N retries. - self._supervisor_task = asyncio.create_task(self._run_supervisor()) + if subscription_path is not None: + # Start the supervisor task that runs the Pub/Sub pull with exponential + # backoff + jitter on transient errors, bails out after N retries. + self._supervisor_task = asyncio.create_task(self._run_supervisor()) + inbound = "pubsub" + else: + self._supervisor_task = None + inbound = "http" + self._mark_connected() logger.info( - "[GoogleChat] Connected; project=%s, subscription=, " + "[GoogleChat] Connected; project=%s, inbound=%s, subscription=%s, " "bot_user_id=%s, flow_control(msgs=%s, bytes=%s)", - project_id, + project_id or "", + inbound, + "" if subscription_path else "", self._bot_user_id or "", self._max_messages, self._max_bytes, @@ -3182,15 +3200,11 @@ class GoogleChatAdapter(BasePlatformAdapter): def _validate_config(config: PlatformConfig) -> bool: - """Plugin-side config gate: require both Pub/Sub project and subscription. - - Mirrors the legacy dispatch entry in ``gateway/config.py`` so the - registry can decide whether the platform is configured without - importing the legacy table. - """ + """Plugin-side config gate for HTTP callback or Pub/Sub inbound modes.""" extra = getattr(config, "extra", {}) or {} return bool( - extra.get("project_id") and extra.get("subscription_name") + extra.get("http_events_url") + or (extra.get("project_id") and extra.get("subscription_name")) ) @@ -3215,7 +3229,8 @@ def _check_for_registry() -> bool: os.getenv("GOOGLE_CHAT_SUBSCRIPTION_NAME") or os.getenv("GOOGLE_CHAT_SUBSCRIPTION") ) - return bool(project and subscription) + http_events_url = os.getenv("GOOGLE_CHAT_HTTP_EVENTS_URL") + return bool(http_events_url or (project and subscription)) def _is_connected(config: PlatformConfig) -> bool: @@ -3245,12 +3260,16 @@ def _env_enablement() -> Optional[Dict[str, Any]]: os.getenv("GOOGLE_CHAT_SUBSCRIPTION_NAME") or os.getenv("GOOGLE_CHAT_SUBSCRIPTION") ) - if not (project and subscription): + http_events_url = os.getenv("GOOGLE_CHAT_HTTP_EVENTS_URL") + if not (http_events_url or (project and subscription)): return None - seed: Dict[str, Any] = { - "project_id": project, - "subscription_name": subscription, - } + seed: Dict[str, Any] = {} + if project: + seed["project_id"] = project + if subscription: + seed["subscription_name"] = subscription + if http_events_url: + seed["http_events_url"] = http_events_url sa_json = ( os.getenv("GOOGLE_CHAT_SERVICE_ACCOUNT_JSON") or os.getenv("GOOGLE_APPLICATION_CREDENTIALS") @@ -3533,8 +3552,6 @@ def register(ctx) -> None: validate_config=_validate_config, is_connected=_is_connected, required_env=[ - "GOOGLE_CHAT_PROJECT_ID", - "GOOGLE_CHAT_SUBSCRIPTION_NAME", "GOOGLE_CHAT_SERVICE_ACCOUNT_JSON", ], install_hint="pip install 'hermes-agent[google_chat]'", diff --git a/plugins/platforms/google_chat/plugin.yaml b/plugins/platforms/google_chat/plugin.yaml index 1a8b90c43..6cf99cb8a 100644 --- a/plugins/platforms/google_chat/plugin.yaml +++ b/plugins/platforms/google_chat/plugin.yaml @@ -4,31 +4,34 @@ kind: platform version: 1.0.0 description: > Google Chat gateway adapter for Hermes Agent. - Connects via Cloud Pub/Sub pull subscription for inbound events and the - Google Chat REST API for outbound messages — same ergonomics as Slack - Socket Mode or Telegram long-polling, no public URL required. Native - file attachments are delivered via per-user OAuth (each user runs - /setup-files once in their own DM). + Connects through authenticated HTTP callbacks or an optional Cloud Pub/Sub + pull subscription for inbound events, and uses the Google Chat REST API for + outbound messages. Native file attachments are delivered via per-user OAuth + (each user runs /setup-files once in their own DM). author: Ramón Fernández # ``requires_env`` entries are surfaced in ``hermes config`` UI via the # platform-plugin env var injector in ``hermes_cli/config.py``. Using the # rich-dict form lets us contribute description/url/prompt metadata so users # see helpful guidance instead of the auto-generated fallback text. requires_env: - - name: GOOGLE_CHAT_PROJECT_ID - description: "GCP project ID hosting the Pub/Sub topic for Chat events. Falls back to GOOGLE_CLOUD_PROJECT." - prompt: "GCP project ID" - url: "https://console.cloud.google.com/" - password: false - - name: GOOGLE_CHAT_SUBSCRIPTION_NAME - description: "Full Pub/Sub subscription path: projects//subscriptions/. Legacy alias: GOOGLE_CHAT_SUBSCRIPTION." - prompt: "Pub/Sub subscription name" - password: false - name: GOOGLE_CHAT_SERVICE_ACCOUNT_JSON description: "Path to Service Account JSON key (or inline JSON). Leave empty to use Application Default Credentials on Cloud Run / GCE. Falls back to GOOGLE_APPLICATION_CREDENTIALS." prompt: "Path to SA JSON (or empty for ADC)" password: true optional_env: + - name: GOOGLE_CHAT_HTTP_EVENTS_URL + description: "Authenticated HTTP endpoint for Chat message events." + prompt: "HTTP events callback URL" + 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" + url: "https://console.cloud.google.com/" + password: false + - name: GOOGLE_CHAT_SUBSCRIPTION_NAME + description: "Optional Pub/Sub subscription path for pull-mode inbound events." + prompt: "Pub/Sub subscription name" + password: false - name: GOOGLE_CHAT_ALLOWED_USERS description: "Comma-separated user emails allowed to interact with the bot." prompt: "Allowed user emails (comma-separated)" diff --git a/tests/gateway/test_google_chat.py b/tests/gateway/test_google_chat.py index b475b48be..2f739b51f 100644 --- a/tests/gateway/test_google_chat.py +++ b/tests/gateway/test_google_chat.py @@ -253,6 +253,7 @@ class TestEnvConfigLoading: "GOOGLE_CLOUD_PROJECT", "GOOGLE_CHAT_SUBSCRIPTION_NAME", "GOOGLE_CHAT_SUBSCRIPTION", + "GOOGLE_CHAT_HTTP_EVENTS_URL", "GOOGLE_CHAT_SERVICE_ACCOUNT_JSON", "GOOGLE_APPLICATION_CREDENTIALS", "GOOGLE_CHAT_HOME_CHANNEL", @@ -281,7 +282,12 @@ class TestEnvConfigLoading: cfg = load_gateway_config() assert _GC not in cfg.platforms - + def test_http_events_enable_without_pubsub(self, monkeypatch): + self._clean_env(monkeypatch) + monkeypatch.setenv("GOOGLE_CHAT_HTTP_EVENTS_URL", "https://example.test/google-chat/events") + cfg = load_gateway_config() + assert _GC in cfg.platforms + assert cfg.platforms[_GC].extra["http_events_url"] == "https://example.test/google-chat/events" # =========================================================================== @@ -391,6 +397,47 @@ class TestValidateConfig: assert project == "test-project" assert sub == "projects/test-project/subscriptions/test-sub" + def test_http_events_mode_does_not_require_pubsub(self): + cfg = PlatformConfig(enabled=True) + cfg.extra["http_events_url"] = "https://example.test/google-chat/events" + a = GoogleChatAdapter(cfg) + project, sub = a._validate_config() + assert project == "" + assert sub is None + + def test_full_subscription_can_infer_project(self): + cfg = PlatformConfig(enabled=True) + cfg.extra["subscription_name"] = "projects/inferred/subscriptions/sub" + a = GoogleChatAdapter(cfg) + project, sub = a._validate_config() + assert project == "inferred" + assert sub == "projects/inferred/subscriptions/sub" + + +class TestConnectModes: + @pytest.mark.asyncio + async def test_connect_http_mode_skips_pubsub_subscriber(self, tmp_path, monkeypatch): + cfg = PlatformConfig(enabled=True) + cfg.extra.update({ + "http_events_url": "https://example.test/google-chat/events", + "service_account_json": "{}", + }) + a = GoogleChatAdapter(cfg) + a._thread_count_store._path = tmp_path / "google_chat_thread_counts.json" + monkeypatch.setattr(_gc_mod, "_load_google_modules", lambda: True) + monkeypatch.setattr(a, "_load_sa_credentials", MagicMock(return_value=MagicMock())) + monkeypatch.setattr(_gc_mod, "build_service", MagicMock(return_value=MagicMock())) + subscriber_client = MagicMock() + monkeypatch.setattr(_gc_mod, "pubsub_v1", MagicMock(SubscriberClient=subscriber_client)) + a._resolve_bot_user_id = AsyncMock(return_value=None) + + assert await a.connect() is True + subscriber_client.assert_not_called() + assert a._subscription_path is None + assert a._supervisor_task is None + assert a.is_connected is True + await a.disconnect() + # =========================================================================== # _chunk_text