fix(google-chat): allow http inbound without pubsub

fix/verification-admin-route-recovery
aeyeopsdev 2026-05-31 18:57:44 +00:00 committed by Teknium
parent 702473edbd
commit f61169861a
3 changed files with 154 additions and 87 deletions

View File

@ -1,9 +1,9 @@
""" """
Google Chat platform adapter. Google Chat platform adapter.
Uses Google Cloud Pub/Sub (pull subscription) for inbound events and the Uses authenticated HTTP callbacks or Google Cloud Pub/Sub for inbound
Google Chat REST API for outbound messages. Pattern parallels Slack Socket events and the Google Chat REST API for outbound messages. Pub/Sub remains
Mode and Telegram long-polling: no public endpoint required. available for no-public-URL deployments.
Concurrency model Concurrency model
----------------- -----------------
@ -767,33 +767,42 @@ class GoogleChatAdapter(BasePlatformAdapter):
) )
return credentials return credentials
def _validate_config(self) -> Tuple[str, str]: def _validate_config(self) -> Tuple[str, Optional[str]]:
"""Return (project_id, subscription_path) after validation. """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") project_id = (self.config.extra.get("project_id") or "").strip()
subscription = self.config.extra.get("subscription_name") 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/<project>/subscriptions/<sub>'."
)
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: if not project_id:
raise ValueError( raise ValueError(
"GOOGLE_CHAT_PROJECT_ID (or GOOGLE_CLOUD_PROJECT) is not set." "GOOGLE_CHAT_PROJECT_ID (or GOOGLE_CLOUD_PROJECT) is not set."
) )
if not subscription: raise ValueError(
raise ValueError( "GOOGLE_CHAT_SUBSCRIPTION_NAME (or GOOGLE_CHAT_SUBSCRIPTION) is not set. "
"GOOGLE_CHAT_SUBSCRIPTION_NAME (or GOOGLE_CHAT_SUBSCRIPTION) is not set." "Set GOOGLE_CHAT_HTTP_EVENTS_URL for HTTP callback mode."
) )
match = _SUBSCRIPTION_PATH_RE.match(subscription)
if not match:
raise ValueError(
"GOOGLE_CHAT_SUBSCRIPTION_NAME must match "
"'projects/<project>/subscriptions/<sub>'."
)
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
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# Loop bridge helpers (thread -> asyncio loop) # Loop bridge helpers (thread -> asyncio loop)
@ -1004,36 +1013,37 @@ class GoogleChatAdapter(BasePlatformAdapter):
"all threads as fresh)", exc_info=True, "all threads as fresh)", exc_info=True,
) )
# Sanity check: subscription exists / SA has access. if subscription_path is not None:
self._subscriber = pubsub_v1.SubscriberClient(credentials=credentials) # Sanity check: subscription exists / SA has access.
try: self._subscriber = pubsub_v1.SubscriberClient(credentials=credentials)
await asyncio.to_thread( try:
lambda: self._subscriber.get_subscription( await asyncio.to_thread(
request={"subscription": subscription_path} lambda: self._subscriber.get_subscription(
request={"subscription": subscription_path}
)
) )
) except gax_exceptions.NotFound:
except gax_exceptions.NotFound: self._set_fatal_error(
self._set_fatal_error( code="subscription_not_found",
code="subscription_not_found", message="Pub/Sub subscription not found at configured path",
message="Pub/Sub subscription not found at configured path", retryable=False,
retryable=False, )
) return False
return False except gax_exceptions.PermissionDenied:
except gax_exceptions.PermissionDenied: self._set_fatal_error(
self._set_fatal_error( code="subscription_permission",
code="subscription_permission", message=(
message=( "Service Account lacks roles/pubsub.subscriber on the "
"Service Account lacks roles/pubsub.subscriber on the " "subscription"
"subscription" ),
), retryable=False,
retryable=False, )
) return False
return False except Exception as exc:
except Exception as exc: msg = _redact_sensitive(str(exc))
msg = _redact_sensitive(str(exc)) logger.error("[GoogleChat] subscription.get failed: %s", msg)
logger.error("[GoogleChat] subscription.get failed: %s", msg) self._set_fatal_error(code="subscription_check", message=msg, retryable=True)
self._set_fatal_error(code="subscription_check", message=msg, retryable=True) return False
return False
# Resolve bot user_id (eager): cache first, then members.list. # Resolve bot user_id (eager): cache first, then members.list.
self._bot_user_id = self._load_cached_bot_id() self._bot_user_id = self._load_cached_bot_id()
@ -1047,14 +1057,22 @@ class GoogleChatAdapter(BasePlatformAdapter):
"will resolve on first addedToSpace or member lookup" "will resolve on first addedToSpace or member lookup"
) )
# Start the supervisor task that runs the Pub/Sub pull with exponential if subscription_path is not None:
# backoff + jitter on transient errors, bails out after N retries. # Start the supervisor task that runs the Pub/Sub pull with exponential
self._supervisor_task = asyncio.create_task(self._run_supervisor()) # 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() self._mark_connected()
logger.info( logger.info(
"[GoogleChat] Connected; project=%s, subscription=<redacted>, " "[GoogleChat] Connected; project=%s, inbound=%s, subscription=%s, "
"bot_user_id=%s, flow_control(msgs=%s, bytes=%s)", "bot_user_id=%s, flow_control(msgs=%s, bytes=%s)",
project_id, project_id or "<unset>",
inbound,
"<redacted>" if subscription_path else "<none>",
self._bot_user_id or "<unresolved>", self._bot_user_id or "<unresolved>",
self._max_messages, self._max_messages,
self._max_bytes, self._max_bytes,
@ -3182,15 +3200,11 @@ class GoogleChatAdapter(BasePlatformAdapter):
def _validate_config(config: PlatformConfig) -> bool: def _validate_config(config: PlatformConfig) -> bool:
"""Plugin-side config gate: require both Pub/Sub project and subscription. """Plugin-side config gate for HTTP callback or Pub/Sub inbound modes."""
Mirrors the legacy dispatch entry in ``gateway/config.py`` so the
registry can decide whether the platform is configured without
importing the legacy table.
"""
extra = getattr(config, "extra", {}) or {} extra = getattr(config, "extra", {}) or {}
return bool( 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") os.getenv("GOOGLE_CHAT_SUBSCRIPTION_NAME")
or os.getenv("GOOGLE_CHAT_SUBSCRIPTION") 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: def _is_connected(config: PlatformConfig) -> bool:
@ -3245,12 +3260,16 @@ def _env_enablement() -> Optional[Dict[str, Any]]:
os.getenv("GOOGLE_CHAT_SUBSCRIPTION_NAME") os.getenv("GOOGLE_CHAT_SUBSCRIPTION_NAME")
or os.getenv("GOOGLE_CHAT_SUBSCRIPTION") 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 return None
seed: Dict[str, Any] = { seed: Dict[str, Any] = {}
"project_id": project, if project:
"subscription_name": subscription, seed["project_id"] = project
} if subscription:
seed["subscription_name"] = subscription
if http_events_url:
seed["http_events_url"] = http_events_url
sa_json = ( sa_json = (
os.getenv("GOOGLE_CHAT_SERVICE_ACCOUNT_JSON") os.getenv("GOOGLE_CHAT_SERVICE_ACCOUNT_JSON")
or os.getenv("GOOGLE_APPLICATION_CREDENTIALS") or os.getenv("GOOGLE_APPLICATION_CREDENTIALS")
@ -3533,8 +3552,6 @@ def register(ctx) -> None:
validate_config=_validate_config, validate_config=_validate_config,
is_connected=_is_connected, is_connected=_is_connected,
required_env=[ required_env=[
"GOOGLE_CHAT_PROJECT_ID",
"GOOGLE_CHAT_SUBSCRIPTION_NAME",
"GOOGLE_CHAT_SERVICE_ACCOUNT_JSON", "GOOGLE_CHAT_SERVICE_ACCOUNT_JSON",
], ],
install_hint="pip install 'hermes-agent[google_chat]'", install_hint="pip install 'hermes-agent[google_chat]'",

View File

@ -4,31 +4,34 @@ kind: platform
version: 1.0.0 version: 1.0.0
description: > description: >
Google Chat gateway adapter for Hermes Agent. Google Chat gateway adapter for Hermes Agent.
Connects via Cloud Pub/Sub pull subscription for inbound events and the Connects through authenticated HTTP callbacks or an optional Cloud Pub/Sub
Google Chat REST API for outbound messages — same ergonomics as Slack pull subscription for inbound events, and uses the Google Chat REST API for
Socket Mode or Telegram long-polling, no public URL required. Native outbound messages. Native file attachments are delivered via per-user OAuth
file attachments are delivered via per-user OAuth (each user runs (each user runs /setup-files once in their own DM).
/setup-files once in their own DM).
author: Ramón Fernández author: Ramón Fernández
# ``requires_env`` entries are surfaced in ``hermes config`` UI via the # ``requires_env`` entries are surfaced in ``hermes config`` UI via the
# platform-plugin env var injector in ``hermes_cli/config.py``. Using 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 # rich-dict form lets us contribute description/url/prompt metadata so users
# see helpful guidance instead of the auto-generated fallback text. # see helpful guidance instead of the auto-generated fallback text.
requires_env: 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/<proj>/subscriptions/<sub>. Legacy alias: GOOGLE_CHAT_SUBSCRIPTION."
prompt: "Pub/Sub subscription name"
password: false
- name: GOOGLE_CHAT_SERVICE_ACCOUNT_JSON - 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." 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)" prompt: "Path to SA JSON (or empty for ADC)"
password: true password: true
optional_env: 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 - name: GOOGLE_CHAT_ALLOWED_USERS
description: "Comma-separated user emails allowed to interact with the bot." description: "Comma-separated user emails allowed to interact with the bot."
prompt: "Allowed user emails (comma-separated)" prompt: "Allowed user emails (comma-separated)"

View File

@ -253,6 +253,7 @@ class TestEnvConfigLoading:
"GOOGLE_CLOUD_PROJECT", "GOOGLE_CLOUD_PROJECT",
"GOOGLE_CHAT_SUBSCRIPTION_NAME", "GOOGLE_CHAT_SUBSCRIPTION_NAME",
"GOOGLE_CHAT_SUBSCRIPTION", "GOOGLE_CHAT_SUBSCRIPTION",
"GOOGLE_CHAT_HTTP_EVENTS_URL",
"GOOGLE_CHAT_SERVICE_ACCOUNT_JSON", "GOOGLE_CHAT_SERVICE_ACCOUNT_JSON",
"GOOGLE_APPLICATION_CREDENTIALS", "GOOGLE_APPLICATION_CREDENTIALS",
"GOOGLE_CHAT_HOME_CHANNEL", "GOOGLE_CHAT_HOME_CHANNEL",
@ -281,7 +282,12 @@ class TestEnvConfigLoading:
cfg = load_gateway_config() cfg = load_gateway_config()
assert _GC not in cfg.platforms 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 project == "test-project"
assert sub == "projects/test-project/subscriptions/test-sub" 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 # _chunk_text