diff --git a/charts/openab/README.md b/charts/openab/README.md index dfac33a38..2bcf05788 100644 --- a/charts/openab/README.md +++ b/charts/openab/README.md @@ -48,6 +48,7 @@ Each agent lives under `agents.`. | `stt.baseUrl` | STT API base URL. | `"https://api.groq.com/openai/v1"` | | `gateway.enabled` | Enable the gateway config block for webhook-based platforms. | `false` | | `gateway.deploy` | Deploy the gateway Deployment and Service. | `true` | +| `gateway.serviceAccountName` | Existing ServiceAccount attached to the gateway pod. Empty preserves the Kubernetes default identity and does not inherit per-agent or chart-global values. Required for GKE Workload Identity when `gateway.googleChat.useAdc=true`; the chart does not create or annotate the ServiceAccount. | `""` | | `cron.usercronEnabled` | Enable user-provided cron configuration. | `false` | | `cronjobs` | Config-driven scheduled messages for an agent. | `[]` | | `persistence.enabled` | Enable persistent storage for auth and settings. | `true` | diff --git a/charts/openab/templates/gateway.yaml b/charts/openab/templates/gateway.yaml index 2a89dc79a..7f45c6900 100644 --- a/charts/openab/templates/gateway.yaml +++ b/charts/openab/templates/gateway.yaml @@ -4,6 +4,7 @@ {{- $gwCfg := omit $cfg "nameOverride" }} {{- $d := dict "ctx" $ "agent" (printf "%s-gateway" $name) "cfg" $gwCfg }} {{- $agentD := dict "ctx" $ "agent" $name "cfg" $cfg }} +{{- $gatewaySvcAcct := (($cfg.gateway).serviceAccountName) }} {{- $hasTeams := and (($cfg.gateway).teams).appId (($cfg.gateway).teams).appSecret }} {{- $hasTelegram := (($cfg.gateway).telegram).botToken }} {{- $hasLine := (($cfg.gateway).line).channelSecret }} @@ -32,6 +33,9 @@ spec: securityContext: {{- toYaml . | nindent 8 }} {{- end }} + {{- if $gatewaySvcAcct }} + serviceAccountName: {{ $gatewaySvcAcct | quote }} + {{- end }} containers: - name: gateway image: {{ printf "%s:%s" (($cfg.gateway).image | default "ghcr.io/openabdev/openab-gateway") (($cfg.gateway).tag | default $.Chart.AppVersion) }} @@ -157,10 +161,16 @@ spec: value: "false" {{- end }} {{- end }} - {{- $hasGoogleChat := or (($cfg.gateway).googleChat).saKeyJson (($cfg.gateway).googleChat).accessToken (($cfg.gateway).googleChat).audience }} + {{- $hasGoogleChat := or (($cfg.gateway).googleChat).saKeyJson (($cfg.gateway).googleChat).accessToken (($cfg.gateway).googleChat).audience (($cfg.gateway).googleChat).useAdc }} {{- if $hasGoogleChat }} - name: GOOGLE_CHAT_ENABLED value: "true" + {{- if (($cfg.gateway).googleChat).useAdc }} + - name: GOOGLE_CHAT_USE_ADC + value: "true" + - name: GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT + value: {{ required "gateway.googleChat.adcTargetServiceAccount is required when useAdc=true" (($cfg.gateway).googleChat).adcTargetServiceAccount | quote }} + {{- end }} {{- if (($cfg.gateway).googleChat).audience }} - name: GOOGLE_CHAT_AUDIENCE value: {{ ($cfg.gateway).googleChat.audience | quote }} diff --git a/charts/openab/tests/gateway_serviceaccount_test.yaml b/charts/openab/tests/gateway_serviceaccount_test.yaml new file mode 100644 index 000000000..2eea62b55 --- /dev/null +++ b/charts/openab/tests/gateway_serviceaccount_test.yaml @@ -0,0 +1,67 @@ +suite: gateway serviceAccountName support + +templates: + - templates/gateway.yaml + +set: + agents.kiro.gateway.enabled: true + +tests: + - it: does not render a gateway serviceAccountName when no value is set + documentIndex: 0 + asserts: + - notExists: + path: spec.template.spec.serviceAccountName + + - it: does not inherit the chart-global serviceAccountName + set: + serviceAccountName: global-runtime + documentIndex: 0 + asserts: + - notExists: + path: spec.template.spec.serviceAccountName + + - it: does not inherit per-agent or global serviceAccountName + set: + serviceAccountName: global-runtime + agents.kiro.serviceAccountName: agent-runtime + documentIndex: 0 + asserts: + - notExists: + path: spec.template.spec.serviceAccountName + + - it: gateway serviceAccountName overrides per-agent and global values + set: + serviceAccountName: global-runtime + agents.kiro.serviceAccountName: agent-runtime + agents.kiro.gateway.serviceAccountName: gateway-runtime + documentIndex: 0 + asserts: + - equal: + path: spec.template.spec.serviceAccountName + value: gateway-runtime + + - it: renders ADC target and the dedicated gateway runtime service account together + set: + agents.kiro.gateway.serviceAccountName: googlechat-runtime + agents.kiro.gateway.googleChat.useAdc: true + agents.kiro.gateway.googleChat.adcTargetServiceAccount: openab-chat@project.iam.gserviceaccount.com + documentIndex: 0 + asserts: + - equal: + path: spec.template.spec.serviceAccountName + value: googlechat-runtime + - contains: + path: spec.template.spec.containers[0].env + content: + name: GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT + value: openab-chat@project.iam.gserviceaccount.com + + - it: quotes numeric-looking gateway serviceAccountName as a string + set: + agents.kiro.gateway.serviceAccountName: "123" + documentIndex: 0 + asserts: + - equal: + path: spec.template.spec.serviceAccountName + value: "123" diff --git a/charts/openab/values.yaml b/charts/openab/values.yaml index ce1ac7409..218e3b3c0 100644 --- a/charts/openab/values.yaml +++ b/charts/openab/values.yaml @@ -395,6 +395,11 @@ agents: gateway: enabled: false # set to true + provide url to enable the [gateway] config block deploy: true # set to false to skip Gateway Deployment/Service (config-only mode) + # Existing Kubernetes ServiceAccount for the gateway pod. Empty omits + # serviceAccountName and preserves the Kubernetes default identity; it + # does not inherit per-agent or chart-global serviceAccountName values. + # Required on GKE when googleChat.useAdc uses Workload Identity. + serviceAccountName: "" url: "" # e.g. ws://openab-gateway:8080/ws platform: "telegram" # default platform when gateway is enabled token: "" # optional shared secret (injected via GATEWAY_WS_TOKEN env var) @@ -459,6 +464,8 @@ agents: audience: "" # JWT audience → GOOGLE_CHAT_AUDIENCE (set to your webhook URL to enable JWT verification) saKeyJson: "" # Service account key JSON string → GOOGLE_CHAT_SA_KEY_JSON (recommended, auto-refresh) accessToken: "" # Static OAuth2 access token → GOOGLE_CHAT_ACCESS_TOKEN (fallback, 1-hour TTL) + useAdc: false # Keyless ADC → GOOGLE_CHAT_USE_ADC. Runtime SA impersonates the distinct adcTargetServiceAccount; no SA key file. Needs roles/iam.serviceAccountTokenCreator on the target + iamcredentials.googleapis.com. Self-impersonation is prohibited. + adcTargetServiceAccount: "" # Dedicated Chat-app SA email → GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT. Required when useAdc=true; MUST differ from the pod/runtime SA. webhookPath: "" # Gateway default: /webhook/googlechat → GOOGLE_CHAT_WEBHOOK_PATH # WeCom (企业微信) adapter config (gateway-side env vars) # See docs/wecom.md for full setup guide diff --git a/config.toml.example b/config.toml.example index 12d0eee80..e5cf7df3d 100644 --- a/config.toml.example +++ b/config.toml.example @@ -123,6 +123,9 @@ allowed_channels = ["1234567890"] # ↑ omitted + non-empty list → auto- # sa_key_json = "${GOOGLE_CHAT_SA_KEY_JSON}" # inline SA key; wins over sa_key_file # sa_key_file = "/etc/openab/sa.json" # env fallback: GOOGLE_CHAT_SA_KEY_FILE # access_token = "${GOOGLE_CHAT_ACCESS_TOKEN}" # static token alternative +# use_adc = true # keyless ADC via GCE metadata + IAM Credentials; runtime SA impersonates a DISTINCT target (self-impersonation is prohibited). env: GOOGLE_CHAT_USE_ADC +# adc_target_service_account = "chat-bot@project.iam.gserviceaccount.com" # required with use_adc; env: GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT +# Auth precedence: successfully loaded SA key > ADC > static access_token. # audience = "projects//..." # enables webhook JWT verification (L1) # webhook_path = "/webhook/googlechat" # env fallback: GOOGLE_CHAT_WEBHOOK_PATH # allow_all_users = false # env fallback: GOOGLE_CHAT_ALLOW_ALL_USERS diff --git a/crates/openab-core/src/adapter.rs b/crates/openab-core/src/adapter.rs index fa7e95dba..2bf3a7c6d 100644 --- a/crates/openab-core/src/adapter.rs +++ b/crates/openab-core/src/adapter.rs @@ -35,6 +35,25 @@ fn reply_message_limit(platform: &str, adapter_limit: usize) -> usize { } } +/// Whether to use cosmetic streaming (placeholder + in-place edits) for +/// `platform`, given the adapter's own preference. Forces send-once for: +/// - `acp`: streams append-only `agent_message_chunk` deltas, not edits. +/// - platforms in `NON_STREAMING_PLATFORMS` — no message-edit API, or an edit +/// API that cannot be driven per-token (e.g. `googlechat`) — see +/// `NON_STREAMING_PLATFORMS` for the per-platform rationale and +/// `platform_supports_streaming`. +/// +/// This is the embedded/unified dispatch gate; the WebSocket +/// `run_gateway_adapter` path applies the same `platform_supports_streaming` +/// check but has no `acp` case (ACP is embedded-only). The two gates are +/// siblings: adding a platform to `NON_STREAMING_PLATFORMS` covers both +/// paths, while an embedded-only carve-out belongs here. +fn resolve_streaming(platform: &str, adapter_prefers_streaming: bool) -> bool { + platform != "acp" + && crate::gateway::platform_supports_streaming(platform) + && adapter_prefers_streaming +} + /// Parse `[[key:value]]` directives from the beginning of agent output. /// Returns parsed directives and the remaining content (directives stripped). pub fn parse_output_directives(content: &str) -> (OutputDirectives, String) { @@ -701,15 +720,14 @@ impl AdapterRouter { let adapter = adapter.clone(); let thread_channel = thread_channel.clone(); let message_limit = reply_message_limit(&thread_channel.platform, adapter.message_limit()); - // ACP must not inherit the unified adapter's Telegram streaming flag (wrong - // coupling): it streams append-only `agent_message_chunk` deltas built from the - // post+edit (`edit_message` snapshot) path, i.e. streaming=false. Decide it - // explicitly by platform rather than by whatever Telegram happens to be set to. - let streaming = if thread_channel.platform == "acp" { - false - } else { - adapter.use_streaming(other_bot_present) - }; + // Decide streaming explicitly by platform, not by whatever the unified + // adapter's Telegram flag happens to be. ACP streams append-only deltas + // (not cosmetic edits); Google Chat / LINE can't sustain in-place edits. + // See `resolve_streaming`. + let streaming = resolve_streaming( + &thread_channel.platform, + adapter.use_streaming(other_bot_present), + ); // Keep the full turn text (incl. inter-tool narration) when streaming // (it was already shown live) OR when `[reactions] narration_display` is // set. Otherwise a send-once turn delivers only the final answer block. @@ -1746,6 +1764,20 @@ mod tests { assert_eq!(crate::format::split_message(&long, reply_message_limit("acp", 4096)).len(), 1); } + #[test] + fn resolve_streaming_forces_send_once_for_acp_and_googlechat() { + // Editable platforms honor the adapter's own streaming preference. + assert!(resolve_streaming("discord", true)); + assert!(!resolve_streaming("discord", false)); + assert!(resolve_streaming("telegram", true)); + // ACP streams append-only deltas, not cosmetic edits → always send-once. + assert!(!resolve_streaming("acp", true)); + // Google Chat: synthetic id can't be patched (400 INVALID_ARGUMENT) → send-once regardless of pref. + assert!(!resolve_streaming("googlechat", true)); + // LINE: no edit API → send-once. + assert!(!resolve_streaming("line", true)); + } + #[test] fn select_delivery_text_send_once_keeps_only_final_block() { // Simulates: narration "n1" → tool (answer_start→2) → narration "n2" diff --git a/crates/openab-core/src/config.rs b/crates/openab-core/src/config.rs index a9bc26abd..fdd89dcb3 100644 --- a/crates/openab-core/src/config.rs +++ b/crates/openab-core/src/config.rs @@ -823,11 +823,11 @@ impl TelegramConfig { } } -/// `true` when env var == "1" or "true" (case-insensitive); default `false`. -/// Matches the legacy `TELEGRAM_TRUSTED_SOURCE_ONLY` semantics. +/// `true` when the trimmed env var is "1" or "true" (case-insensitive); +/// default `false`. Matches the legacy `TELEGRAM_TRUSTED_SOURCE_ONLY` semantics. fn env_flag_true_one(key: &str) -> bool { std::env::var(key) - .map(|v| v == "1" || v.eq_ignore_ascii_case("true")) + .map(|v| v.trim() == "1" || v.trim().eq_ignore_ascii_case("true")) .unwrap_or(false) } @@ -1225,6 +1225,16 @@ pub struct GoogleChatConfig { /// checked when `allow_all_users` resolves to `false`. Env fallback: /// `GOOGLE_CHAT_ALLOWED_USERS` (comma-separated). pub allowed_users: Option>, + /// Use keyless ADC (GCE metadata server + IAM Credentials + /// `generateAccessToken`) to mint the `chat.bot` token for a distinct target + /// service account. Env fallback: `GOOGLE_CHAT_USE_ADC` (`true`/`1`; default + /// false). Ignored when a configured SA key loads successfully. + pub use_adc: Option, + /// Dedicated Google Chat service account impersonated by the attached + /// runtime service account. Required when `use_adc=true`; MUST differ from + /// the runtime identity because Google prohibits access-token + /// self-impersonation. Env: `GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT`. + pub adc_target_service_account: Option, } /// Fully resolved Google Chat settings (config → env → default applied). @@ -1234,6 +1244,8 @@ pub struct ResolvedGoogleChat { pub sa_key_json: Option, pub sa_key_file: Option, pub access_token: Option, + pub use_adc: bool, + pub adc_target_service_account: Option, pub audience: Option, pub webhook_path: String, pub allow_all_users: bool, @@ -1242,13 +1254,13 @@ pub struct ResolvedGoogleChat { impl GoogleChatConfig { /// Resolve every field: config value (if set) → `GOOGLE_CHAT_*` env → - /// default. String fields filter empty strings from `${}` expansion. + /// default. String fields filter empty or whitespace-only `${}` expansion. pub fn resolve(&self) -> ResolvedGoogleChat { let opt_str = |cfg: &Option, env: &str| -> Option { cfg.as_ref() - .filter(|s| !s.is_empty()) + .filter(|s| !s.trim().is_empty()) .cloned() - .or_else(|| std::env::var(env).ok()) + .or_else(|| std::env::var(env).ok().filter(|s| !s.trim().is_empty())) }; ResolvedGoogleChat { enabled: self.enabled.unwrap_or_else(|| { @@ -1259,6 +1271,13 @@ impl GoogleChatConfig { sa_key_json: opt_str(&self.sa_key_json, "GOOGLE_CHAT_SA_KEY_JSON"), sa_key_file: opt_str(&self.sa_key_file, "GOOGLE_CHAT_SA_KEY_FILE"), access_token: opt_str(&self.access_token, "GOOGLE_CHAT_ACCESS_TOKEN"), + use_adc: self + .use_adc + .unwrap_or_else(|| env_flag_true_one("GOOGLE_CHAT_USE_ADC")), + adc_target_service_account: opt_str( + &self.adc_target_service_account, + "GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT", + ), audience: opt_str(&self.audience, "GOOGLE_CHAT_AUDIENCE"), webhook_path: opt_str(&self.webhook_path, "GOOGLE_CHAT_WEBHOOK_PATH") .unwrap_or_else(|| "/webhook/googlechat".into()), @@ -2966,6 +2985,8 @@ allowed_users = ["U1234567890abcdef0123456789abcdef"] "GOOGLE_CHAT_SA_KEY_JSON", "GOOGLE_CHAT_SA_KEY_FILE", "GOOGLE_CHAT_ACCESS_TOKEN", + "GOOGLE_CHAT_USE_ADC", + "GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT", "GOOGLE_CHAT_AUDIENCE", "GOOGLE_CHAT_WEBHOOK_PATH", ] { @@ -2974,9 +2995,39 @@ allowed_users = ["U1234567890abcdef0123456789abcdef"] // --- defaults --- let r = GoogleChatConfig::default().resolve(); assert!(!r.enabled); + assert!(!r.use_adc); assert!(r.audience.is_none()); assert_eq!(r.webhook_path, "/webhook/googlechat"); + // --- whitespace-only secrets/targets are absent at the config boundary --- + std::env::set_var("GOOGLE_CHAT_ACCESS_TOKEN", " \t "); + std::env::set_var("GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT", " \t "); + let r = GoogleChatConfig::default().resolve(); + assert!(r.access_token.is_none()); + assert!(r.adc_target_service_account.is_none()); + std::env::remove_var("GOOGLE_CHAT_ACCESS_TOKEN"); + std::env::remove_var("GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT"); + + // --- identity-selecting bool env accepts common case/whitespace forms --- + std::env::set_var("GOOGLE_CHAT_USE_ADC", " True "); + assert!(GoogleChatConfig::default().resolve().use_adc); + std::env::remove_var("GOOGLE_CHAT_USE_ADC"); + + // --- use_adc: config value resolves without touching env --- + let r = GoogleChatConfig { + use_adc: Some(true), + adc_target_service_account: Some( + "chat-bot@project.iam.gserviceaccount.com".into(), + ), + ..Default::default() + } + .resolve(); + assert!(r.use_adc); + assert_eq!( + r.adc_target_service_account.as_deref(), + Some("chat-bot@project.iam.gserviceaccount.com") + ); + // --- config wins over env --- std::env::set_var("GOOGLE_CHAT_ENABLED", "true"); std::env::set_var("GOOGLE_CHAT_AUDIENCE", "env-aud"); @@ -2989,9 +3040,9 @@ allowed_users = ["U1234567890abcdef0123456789abcdef"] assert!(!r.enabled); // config false wins over env true assert_eq!(r.audience.as_deref(), Some("cfg-aud")); - // --- empty-string ${} expansion falls through to env --- + // --- whitespace-only ${} expansion falls through to env --- let cfg = GoogleChatConfig { - audience: Some("".into()), + audience: Some(" \t ".into()), ..Default::default() }; let r = cfg.resolve(); diff --git a/crates/openab-core/src/gateway.rs b/crates/openab-core/src/gateway.rs index a3b74adbd..e63416b56 100644 --- a/crates/openab-core/src/gateway.rs +++ b/crates/openab-core/src/gateway.rs @@ -10,8 +10,12 @@ use tokio::sync::Mutex; use tokio_tungstenite::tungstenite::Message; use tracing::{error, info, warn}; -/// Timeout for waiting on gateway reply acknowledgement. +/// Legacy timeout for streaming platforms that may not acknowledge normal sends. const GATEWAY_REPLY_TIMEOUT_SECS: u64 = 5; +/// Acknowledged send-once replies can include bounded auth refresh (up to three +/// 10-second requests), so allow that work to finish; unlike the legacy path, +/// timeout is an error because the adapter promised a delivery response. +const ACKED_GATEWAY_REPLY_TIMEOUT_SECS: u64 = 35; /// Platforms whose gateway adapter emits a `GatewayResponse` for `edit_message` /// so core can observe edit success or failure (used to gate the per-edit @@ -40,29 +44,50 @@ fn platform_acks_writes(platform: &str) -> bool { EDIT_RESPONSE_PLATFORMS.contains(&platform) } -/// Gateway platforms whose messaging API cannot edit a message after it is sent. +/// Platforms whose gateway adapters acknowledge normal send replies with a +/// `GatewayResponse`. This capability is independent of cosmetic streaming: +/// Google Chat is send-once, but its adapter reports API/auth failures and core +/// must retain the request id to observe them. +const REPLY_RESPONSE_PLATFORMS: &[&str] = &["googlechat"]; + +fn platform_acks_replies(platform: &str) -> bool { + REPLY_RESPONSE_PLATFORMS.contains(&platform) +} + +/// Preserve legacy request/response waits for streaming adapters while also +/// supporting send-once adapters that explicitly acknowledge delivery. +fn reply_requires_ack(platform: &str, streaming: bool) -> bool { + streaming || platform_acks_replies(platform) +} + +/// Platforms where cosmetic (typewriter) streaming is not viable, so replies +/// are forced send-once regardless of the configured `streaming` flag. /// -/// Cosmetic (typewriter) streaming works by posting a placeholder and then -/// repeatedly editing it in place with the growing text. On a platform with no -/// edit endpoint, each of those "edits" is delivered as a brand-new message -/// instead — so the user sees the same reply posted several times, each copy -/// longer than the last. Streaming is therefore force-disabled (send-once) for -/// these platforms regardless of the configured `streaming` flag. +/// Cosmetic streaming posts a placeholder and repeatedly edits it with growing +/// text. Without a usable edit path, those updates become duplicate messages: /// -/// LINE's Messaging API only exposes reply/push (no edit), so it lives here. -/// (The in-process unified adapter additionally hard-drops stray edit_message -/// commands in the LINE adapter itself — see `dispatch_line_reply`.) +/// - `line` / `lineworks`: no message-edit API. +/// - `googlechat`: editing requires a real message resource name, but unified +/// fallback IDs are synthetic; its per-space quota also makes rapid edits +/// unsuitable. See . /// -/// NOTE: like `EDIT_RESPONSE_PLATFORMS`, this is platform-identity standing in -/// for a *capability*. The right long-term model is a capability handshake at -/// gateway-connect time ("can this adapter edit messages?"); until that exists, -/// any new gateway platform that lacks a message-edit API MUST be added here. -const NON_EDITABLE_PLATFORMS: &[&str] = &["line", "lineworks"]; +/// NOTE: like `EDIT_RESPONSE_PLATFORMS`, this is platform identity standing in +/// for a capability. Replace it with a negotiated capability when available; +/// until then, add every platform that cannot support cosmetic edits here. +const NON_STREAMING_PLATFORMS: &[&str] = &["line", "lineworks", "googlechat"]; /// Whether cosmetic streaming (placeholder + in-place edits) is possible on -/// `platform`. See `NON_EDITABLE_PLATFORMS`. -fn platform_supports_streaming(platform: &str) -> bool { - !NON_EDITABLE_PLATFORMS.contains(&platform) +/// `platform`. See `NON_STREAMING_PLATFORMS`. `pub(crate)` so the shared +/// dispatch path (`AdapterRouter::stream_prompt_blocks`) can force send-once on +/// these platforms too, not just the WebSocket `run_gateway_adapter` path. +/// +/// Sibling gate: the embedded/unified dispatch path wraps this in +/// `adapter::resolve_streaming`, which additionally forces send-once for +/// `acp` (embedded-only, streams append-only deltas). An embedded-only +/// non-streaming platform must be handled there — adding it to the shared +/// list above covers both paths, but a platform-specific carve-out does not. +pub(crate) fn platform_supports_streaming(platform: &str) -> bool { + !NON_STREAMING_PLATFORMS.contains(&platform) } /// Shared filter parameters for gateway event gating. @@ -213,6 +238,17 @@ struct GatewayResponse { error: Option, } +fn gateway_delivery_result(resp: GatewayResponse) -> Result { + if resp.success { + Ok(resp.message_id.unwrap_or_else(|| "gw_sent".into())) + } else { + Err(anyhow::anyhow!( + "gateway reported failure: {}", + resp.error.unwrap_or_else(|| "unspecified error".to_string()) + )) + } +} + // --- GatewayAdapter: ChatAdapter over WebSocket --- type PendingRequests = Arc>>>; @@ -262,7 +298,7 @@ impl GatewayAdapter { content: &str, quote_message_id: Option<&str>, ) -> Result { - let req_id = if self.streaming { + let req_id = if reply_requires_ack(self.platform_name, self.streaming) { Some(format!("req_{}", uuid::Uuid::new_v4())) } else { None @@ -298,33 +334,42 @@ impl GatewayAdapter { return Err(e.into()); } let msg_id = if let (Some(rx), Some(ref id)) = (pending_rx, &req_id) { - match tokio::time::timeout(std::time::Duration::from_secs(GATEWAY_REPLY_TIMEOUT_SECS), rx).await { - Ok(Ok(resp)) if resp.success => resp.message_id.unwrap_or_else(|| "gw_sent".into()), - Ok(Ok(resp)) => { - // Gateway explicitly reported failure (success=false). Surface - // as Err so dispatch sets ❌ instead of 🆗 over an incomplete - // delivery. Examples: Feishu edit cap reached after append-new - // fallback also failed; chunked send delivered N/M chunks. - let err_msg = resp.error.clone() - .unwrap_or_else(|| "gateway reported failure".to_string()); - tracing::warn!(request_id = %id, error = %err_msg, "gateway replied with failure"); - return Err(anyhow::anyhow!("gateway reported failure: {err_msg}")); - } + let ack_required = platform_acks_replies(self.platform_name); + let timeout_secs = if ack_required { + ACKED_GATEWAY_REPLY_TIMEOUT_SECS + } else { + GATEWAY_REPLY_TIMEOUT_SECS + }; + match tokio::time::timeout(std::time::Duration::from_secs(timeout_secs), rx).await { + Ok(Ok(resp)) => match gateway_delivery_result(resp) { + Ok(message_id) => message_id, + Err(e) => { + tracing::warn!(request_id = %id, error = %e, "gateway replied with failure"); + return Err(e); + } + }, Ok(Err(_)) => { - // Channel closed (gateway shutting down or pending dropped). - // Maintain legacy behavior — adapters that don't implement - // GatewayResponse for all reply types (LINE, Teams) rely on - // this for non-failure outcomes. + if ack_required { + return Err(anyhow::anyhow!( + "gateway acknowledgement channel closed for {}", + self.platform_name + )); + } + // Legacy streaming adapters may not acknowledge normal sends. tracing::warn!(request_id = %id, "gateway response channel closed"); "gw_sent".into() } Err(_) => { - // Timeout. Many adapters (LINE, Teams) intentionally do not - // emit GatewayResponse for replies, so timeout is the expected - // path for them. Maintain legacy behavior to avoid breaking - // platforms that have not yet wired request/response feedback. - tracing::warn!(request_id = %id, "gateway reply timed out"); self.pending.lock().await.remove(id); + if ack_required { + return Err(anyhow::anyhow!( + "gateway delivery acknowledgement timed out after {timeout_secs}s for {}", + self.platform_name + )); + } + // Preserve legacy behavior for adapters that do not promise + // a GatewayResponse for normal sends. + tracing::warn!(request_id = %id, "gateway reply timed out"); "gw_sent".into() } } @@ -1669,17 +1714,41 @@ mod tests { assert!(!platform_supports_streaming("line")); } + #[test] + fn googlechat_rate_limit_forces_send_once() { + // Google Chat has an edit API, but the unified adapter's synthetic + // message id is not a valid resource name, so patch returns 400 + // INVALID_ARGUMENT; the documented 1 write/sec-per-space quota further + // constrains high-frequency editing. Force send-once. + assert!(!platform_supports_streaming("googlechat")); + } + + #[test] + fn googlechat_send_once_still_requires_delivery_ack() { + assert!(platform_acks_replies("googlechat")); + assert!(reply_requires_ack("googlechat", false)); + assert!(!reply_requires_ack("line", false)); + // Preserve legacy behavior: streaming adapters still carry request IDs. + assert!(reply_requires_ack("discord", true)); + } + + #[test] + fn acknowledged_reply_failure_is_propagated() { + let err = gateway_delivery_result(GatewayResponse { + schema: "openab.gateway.response.v1".into(), + request_id: "req_test".into(), + success: false, + thread_id: None, + message_id: None, + error: Some("googlechat API returned 403".into()), + }) + .expect_err("success=false must reach core as Err"); + assert!(err.to_string().contains("googlechat API returned 403")); + } + #[test] fn editable_platforms_still_allow_streaming() { - for platform in [ - "telegram", - "slack", - "discord", - "feishu", - "teams", - "googlechat", - "wecom", - ] { + for platform in ["telegram", "slack", "discord", "feishu", "teams", "wecom"] { assert!( platform_supports_streaming(platform), "{platform} should still support streaming", diff --git a/crates/openab-gateway/src/adapters/googlechat.rs b/crates/openab-gateway/src/adapters/googlechat.rs index 12d274ee4..03866cdb0 100644 --- a/crates/openab-gateway/src/adapters/googlechat.rs +++ b/crates/openab-gateway/src/adapters/googlechat.rs @@ -7,7 +7,7 @@ use jsonwebtoken::{decode, Algorithm, DecodingKey, Validation}; use serde::Deserialize; use std::sync::Arc; use std::time::Instant; -use tokio::sync::RwLock; +use tokio::sync::{Mutex, RwLock}; use tracing::{error, info, warn}; pub const GOOGLE_CHAT_API_BASE: &str = "https://chat.googleapis.com/v1"; @@ -21,6 +21,21 @@ const AUDIO_MAX_DOWNLOAD: u64 = 25 * 1024 * 1024; // 25 MB /// Per-request timeout for Google Chat Media API downloads. Prevents a hung /// connection from blocking the spawned download task indefinitely. const MEDIA_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + +/// Bound every token-mint request (SA-key exchange, metadata, IAM Credentials) +/// so a hung connection cannot stall senders queued behind the token cache's +/// write lock (the refresh runs while holding it) or prevent the ADC → static +/// token degradation path from engaging. +const TOKEN_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +/// Bound each Google Chat API mutation independently. The enclosing delivery +/// deadline below is stricter than core's 35-second acknowledgement window and +/// covers serialization queue wait, token resolution, and every sequential chunk. +const CHAT_API_REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +/// End-to-end budget for one outbound Google Chat reply. Core waits 35 seconds +/// for an acknowledged gateway reply; finishing within 30 seconds leaves room +/// for response serialization and WebSocket/broadcast scheduling. +const GOOGLE_CHAT_DELIVERY_DEADLINE: std::time::Duration = + std::time::Duration::from_secs(30); /// Cap on text file attachments per message (matches Discord/Slack). const TEXT_FILE_COUNT_CAP: usize = 5; /// Cap on aggregate text file bytes per message (matches Discord/Slack 1 MB). @@ -259,27 +274,180 @@ impl GoogleChatJwtVerifier { } } +/// Accept an external token value only when it contains non-whitespace bytes. +/// All three OAuth boundaries (SA-key exchange, metadata base token, IAM +/// Credentials mint) share this validator so their failure semantics cannot +/// drift independently. +fn non_empty_token(value: &str) -> Option<&str> { + (!value.trim().is_empty()).then_some(value) +} + +/// Normalize and validate the dedicated ADC target identity before it can be +/// interpolated into an IAM Credentials resource path. Only user-managed +/// service-account emails are accepted: this intentionally rejects numeric +/// unique IDs (which could alias the runtime SA and bypass the equality guard), +/// trailing-dot aliases, and URL/path metacharacters. +fn normalize_adc_target_service_account(value: Option) -> Result, String> { + let Some(raw) = value else { + return Ok(None); + }; + let email = raw.trim().to_ascii_lowercase(); + if email.is_empty() { + return Ok(None); + } + + const SUFFIX: &str = ".iam.gserviceaccount.com"; + let (account, domain) = email.split_once('@').ok_or_else(|| { + "adc_target_service_account must be a service-account email, not an ID".to_string() + })?; + let project = domain.strip_suffix(SUFFIX).ok_or_else(|| { + format!("adc_target_service_account must end with {SUFFIX}") + })?; + let valid_label = |label: &str| { + !label.is_empty() + && label + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-') + && label + .as_bytes() + .first() + .is_some_and(u8::is_ascii_alphanumeric) + && label + .as_bytes() + .last() + .is_some_and(u8::is_ascii_alphanumeric) + }; + if !valid_label(account) || !valid_label(project) { + return Err( + "adc_target_service_account must use account@project.iam.gserviceaccount.com with only lowercase letters, digits, and interior hyphens" + .into(), + ); + } + + Ok(Some(email)) +} + +/// Google Chat edit targets must be full `spaces/{space}/messages/{message}` +/// resource names. Reject synthetic ids and incomplete `spaces/` prefixes +/// before token resolution or network I/O. +fn is_google_chat_message_name(name: &str) -> bool { + let safe_segment = |segment: &str| { + !segment.is_empty() + && !segment + .bytes() + .any(|b| matches!(b, b'?' | b'#' | b'&' | b'=' | b'%')) + }; + let mut parts = name.split('/'); + matches!( + ( + parts.next(), + parts.next(), + parts.next(), + parts.next(), + parts.next(), + ), + (Some("spaces"), Some(space), Some("messages"), Some(message), None) + if safe_segment(space) && safe_segment(message) + ) +} + // --- Adapter (encapsulates all Google Chat state) --- pub struct GoogleChatAdapter { pub token_cache: Option, - pub access_token: Option, + pub(crate) metadata_source: Option, + pub(crate) access_token: Option, pub jwt_verifier: Option, + /// Conservatively serializes normal sends across the adapter. Google Chat's + /// quota is per-space; every caller starts its deadline before waiting on + /// this lock so the current coarse queue cannot outlive core's ack window. + delivery_lock: Mutex<()>, pub client: reqwest::Client, pub api_base: String, } +/// Named construction parts for [`GoogleChatAdapter::from_parts`], so call +/// sites name each field instead of counting five positional arguments. +#[derive(Default)] +pub(crate) struct GoogleChatParts { + pub sa_key_json: Option, + pub sa_key_file: Option, + pub access_token: Option, + pub audience: Option, + pub use_adc: bool, + /// Dedicated Google Chat service account impersonated by the attached + /// workload identity. It MUST differ from the metadata server's default SA; + /// Google prohibits access-token self-impersonation. + pub adc_target_service_account: Option, +} + +impl GoogleChatParts { + /// Read the env-only credential contract once so standalone constructors + /// cannot drift on ADC parsing or target trimming. + pub(crate) fn from_env() -> Self { + Self { + sa_key_json: std::env::var("GOOGLE_CHAT_SA_KEY_JSON").ok(), + sa_key_file: std::env::var("GOOGLE_CHAT_SA_KEY_FILE").ok(), + access_token: std::env::var("GOOGLE_CHAT_ACCESS_TOKEN").ok(), + audience: std::env::var("GOOGLE_CHAT_AUDIENCE").ok(), + use_adc: std::env::var("GOOGLE_CHAT_USE_ADC") + .map(|v| v.trim() == "1" || v.trim().eq_ignore_ascii_case("true")) + .unwrap_or(false), + adc_target_service_account: std::env::var( + "GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT", + ) + .ok() + .filter(|s| !s.trim().is_empty()), + } + } +} + +/// Emit one correlated normal-reply result when core supplied a request id. +/// Missing-adapter paths use the same contract as configured delivery failures, +/// so core never waits for an acknowledgement that cannot arrive. +pub(crate) fn emit_delivery_response( + event_tx: &tokio::sync::broadcast::Sender, + request_id: &Option, + result: Result, +) { + let Some(request_id) = request_id else { + return; + }; + let (success, message_id, error) = match result { + Ok(name) => (true, Some(name), None), + Err(error) => (false, None, Some(error)), + }; + let response = crate::schema::GatewayResponse { + schema: "openab.gateway.response.v1".into(), + request_id: request_id.clone(), + success, + thread_id: None, + message_id, + error, + }; + if let Ok(json) = serde_json::to_string(&response) { + let _ = event_tx.send(json); + } +} + impl GoogleChatAdapter { /// Build an adapter from resolved parts (#1379): SA key JSON (inline wins - /// over file path), optional static access token, optional JWT audience. + /// over file path), optional static access token, optional JWT audience, + /// and keyless ADC via GCE metadata + IAM Credentials for a distinct + /// `adc_target_service_account`. Auth precedence at send time: + /// SA key > ADC target > static token. /// Shared by env-derived construction and `apply_googlechat_config`. - pub(crate) fn from_parts( - sa_key_json: Option, - sa_key_file: Option, - access_token: Option, - audience: Option, - ) -> Self { + pub(crate) fn from_parts(parts: GoogleChatParts) -> Self { use tracing::{info, warn}; + let GoogleChatParts { + sa_key_json, + sa_key_file, + access_token, + audience, + use_adc, + adc_target_service_account, + } = parts; + let key_configured = sa_key_json.is_some() || sa_key_file.is_some(); let token_cache = sa_key_json .or_else(|| { sa_key_file.and_then(|path| { @@ -299,7 +467,55 @@ impl GoogleChatAdapter { info!("googlechat webhook JWT verification enabled (audience={aud})"); GoogleChatJwtVerifier::new(aud) }); - Self::new(token_cache, access_token, jwt_verifier) + // Precedence at send time (see `get_token`): SA key > ADC > static token. + // Validate at the adapter boundary so env-only/config-first construction + // and every future caller share the same IAM resource-path invariant. + let adc_target = if use_adc && token_cache.is_none() { + match normalize_adc_target_service_account(adc_target_service_account) { + Ok(target) => target, + Err(e) => { + error!("googlechat ADC target is invalid: {e}; ADC is disabled"); + None + } + } + } else { + None + }; + if use_adc && token_cache.is_none() { + if key_configured { + // A key WAS configured but failed to load. Don't switch identity + // silently: name the distinct target identity selected instead. + match adc_target.as_deref() { + Some(target) => warn!( + target_service_account = %target, + "Google Chat SA key was configured but could not be loaded; \ + falling back to keyless ADC impersonation of the configured \ + target — this is NOT the failed key identity" + ), + None => error!( + "Google Chat SA key was configured but could not be loaded, and \ + use_adc=true has no adc_target_service_account; ADC is disabled" + ), + } + } else if let Some(target) = adc_target.as_deref() { + info!( + target_service_account = %target, + "googlechat keyless ADC enabled (distinct target, chat.bot via IAM Credentials)" + ); + } else { + error!( + "googlechat use_adc=true has no adc_target_service_account; ADC is disabled" + ); + } + } + let mut adapter = Self::new(token_cache, access_token, jwt_verifier); + // Install ADC only when it can actually be consulted and a distinct + // target is configured. A loaded SA key wins outright; use_adc without + // a target fails closed instead of attempting prohibited self-impersonation. + if use_adc && adapter.token_cache.is_none() { + adapter.metadata_source = adc_target.map(MetadataTokenSource::new); + } + adapter } pub fn new( @@ -307,15 +523,29 @@ impl GoogleChatAdapter { access_token: Option, jwt_verifier: Option, ) -> Self { + // Static credentials enter through config and env construction paths. + // Reject whitespace-only values here so every caller shares the same + // outbound bearer boundary without altering valid token bytes. + let access_token = access_token.filter(|token| non_empty_token(token).is_some()); Self { token_cache, + metadata_source: None, access_token, jwt_verifier, + delivery_lock: Mutex::new(()), client: reqwest::Client::new(), api_base: GOOGLE_CHAT_API_BASE.into(), } } + /// Resolve the outbound bearer token. Precedence: SA key (`token_cache`) + /// > ADC (`metadata_source`) > static `access_token`. + /// + /// Failure behavior is asymmetric by design: an SA-key exchange error + /// hard-fails (`None` — the operator explicitly configured that identity), + /// while an ADC mint error can fall through to an explicitly configured + /// static token. The static token is opaque and may represent a different + /// principal, so that degradation is logged as a possible identity switch. async fn get_token(&self) -> Option { if let Some(ref cache) = self.token_cache { match cache.get_token(&self.client).await { @@ -326,7 +556,29 @@ impl GoogleChatAdapter { } } } - self.access_token.clone() + if let Some(ref src) = self.metadata_source { + match src.get_token().await { + Ok(t) => return Some(t), + Err(e) => { + // A configured static token is an explicit fallback, but it + // is opaque: the adapter cannot prove it represents the same + // principal as the ADC target. Name the possible identity + // switch rather than claiming equivalence. + if self.access_token.is_some() { + error!( + "googlechat ADC token mint failed ({e}); falling back to \ + configured static access_token (possible identity switch)" + ); + } else { + error!("googlechat ADC token mint failed: {e}"); + } + } + } + } + self.access_token + .as_deref() + .and_then(non_empty_token) + .map(str::to_owned) } async fn edit_message(&self, message_name: &str, text: &str) { @@ -342,7 +594,15 @@ impl GoogleChatAdapter { ); let body = serde_json::json!({ "text": formatted }); - match self.client.patch(&url).bearer_auth(&token).json(&body).send().await { + match self + .client + .patch(&url) + .bearer_auth(&token) + .json(&body) + .timeout(CHAT_API_REQUEST_TIMEOUT) + .send() + .await + { Ok(r) if r.status().is_success() => { tracing::trace!(message_name = %message_name, "googlechat message edited"); } @@ -357,136 +617,137 @@ impl GoogleChatAdapter { } } - pub async fn handle_reply( + /// Deliver one normal Google Chat reply and return the created message + /// resource name. The absolute deadline starts before waiting for the + /// serialization lock, so queueing + token resolution + every chunk all fit + /// inside core's acknowledgement window. + pub async fn deliver_message(&self, reply: &GatewayReply) -> Result { + self.deliver_message_before( + reply, + tokio::time::Instant::now() + GOOGLE_CHAT_DELIVERY_DEADLINE, + ) + .await + } + + async fn deliver_message_before( &self, reply: &GatewayReply, - event_tx: &tokio::sync::broadcast::Sender, - ) { - // Command routing - match reply.command.as_deref() { - Some("add_reaction") | Some("remove_reaction") | Some("create_topic") => return, - Some("edit_message") => { - self.edit_message(&reply.reply_to, &reply.content.text).await; - return; - } - _ => {} - } + deadline: tokio::time::Instant, + ) -> Result { + let _delivery_guard = tokio::time::timeout_at(deadline, self.delivery_lock.lock()) + .await + .map_err(|_| { + format!( + "googlechat delivery queue timed out after {}s before sending any chunk", + GOOGLE_CHAT_DELIVERY_DEADLINE.as_secs() + ) + })?; + self.deliver_message_inner(reply, deadline).await + } + async fn deliver_message_inner( + &self, + reply: &GatewayReply, + deadline: tokio::time::Instant, + ) -> Result { info!( space = %reply.channel.id, thread_id = ?reply.channel.thread_id, "gateway → googlechat" ); - let Some(token) = self.get_token().await else { - info!( - text = %reply.content.text, - "googlechat reply (dry-run, no credentials configured)" - ); - if let Some(ref req_id) = reply.request_id { - let resp = crate::schema::GatewayResponse { - schema: "openab.gateway.response.v1".into(), - request_id: req_id.clone(), - success: false, - thread_id: None, - message_id: None, - error: Some("no credentials configured".into()), - }; - if let Ok(json) = serde_json::to_string(&resp) { - let _ = event_tx.send(json); - } - } - return; - }; - - let text = &reply.content.text; - let chunks = split_text(text, GOOGLE_CHAT_MESSAGE_LIMIT); - - // Empty message: short-circuit, send failure ack and skip API call + let token = tokio::time::timeout_at(deadline, self.get_token()) + .await + .map_err(|_| "googlechat token resolution timed out before sending any chunk".to_string())? + .ok_or_else(|| "no credentials configured".to_string())?; + let chunks = split_text(&reply.content.text, GOOGLE_CHAT_MESSAGE_LIMIT); if chunks.is_empty() { - if let Some(ref req_id) = reply.request_id { - let resp = crate::schema::GatewayResponse { - schema: "openab.gateway.response.v1".into(), - request_id: req_id.clone(), - success: false, - thread_id: None, - message_id: None, - error: Some("empty message".into()), - }; - if let Ok(json) = serde_json::to_string(&resp) { - let _ = event_tx.send(json); - } - } - return; + return Err("empty message".into()); } - if chunks.len() == 1 { - let result = send_message( - &self.client, - &token, - &reply.channel.id, - reply.channel.thread_id.as_deref(), - text, - &self.api_base, - ) - .await; - - if let Some(ref req_id) = reply.request_id { - let (success, message_id, error) = match result { - Ok(name) => (true, Some(name), None), - Err(e) => (false, None, Some(e)), - }; - let resp = crate::schema::GatewayResponse { - schema: "openab.gateway.response.v1".into(), - request_id: req_id.clone(), - success, - thread_id: None, - message_id, - error, - }; - if let Ok(json) = serde_json::to_string(&resp) { - let _ = event_tx.send(json); - } - } - } else { - let mut first_msg_name: Option = None; - let mut first_error: Option = None; - for chunk in chunks { - match send_message( + let total = chunks.len(); + let mut first_message_name = None; + for (index, chunk) in chunks.into_iter().enumerate() { + let result = tokio::time::timeout_at( + deadline, + send_message( &self.client, &token, &reply.channel.id, reply.channel.thread_id.as_deref(), chunk, &self.api_base, + ), + ) + .await + .map_err(|_| { + format!( + "chunk {}/{} timed out after {} successful chunk(s)", + index + 1, + total, + index ) - .await - { - Ok(name) => { - if first_msg_name.is_none() { - first_msg_name = Some(name); - } - } - Err(e) => { - if first_error.is_none() { - first_error = Some(e); - } + })?; + match result { + Ok(name) => { + if first_message_name.is_none() { + first_message_name = Some(name); } } + Err(e) => { + return Err(format!( + "chunk {}/{} failed after {} successful chunk(s): {e}", + index + 1, + total, + index + )); + } } - if let Some(ref req_id) = reply.request_id { - let resp = crate::schema::GatewayResponse { - schema: "openab.gateway.response.v1".into(), - request_id: req_id.clone(), - success: first_msg_name.is_some() && first_error.is_none(), - thread_id: None, - message_id: first_msg_name, - error: first_error, - }; - if let Ok(json) = serde_json::to_string(&resp) { - let _ = event_tx.send(json); + } + + first_message_name.ok_or_else(|| "googlechat delivery produced no message".into()) + } + + pub async fn handle_reply( + &self, + reply: &GatewayReply, + event_tx: &tokio::sync::broadcast::Sender, + ) { + // Command routing + match reply.command.as_deref() { + // Google Chat does not support these gateway commands. Return before + // token resolution or send-path logging/network I/O; in particular, + // `delete_message` must not fall through as an empty send. + Some("add_reaction") + | Some("remove_reaction") + | Some("create_topic") + | Some("delete_message") => return, + Some("edit_message") => { + // Google Chat is send-once (see core's `NON_STREAMING_PLATFORMS`): + // the unified adapter's synthetic `unified_` id is not a + // valid `spaces/*/messages/*` resource name, and `patch` rejects + // it with 400 INVALID_ARGUMENT before any edit applies. Refuse + // non-resource-name ids here instead of sending a doomed request, + // so a future caller cannot silently reintroduce that failure. + if !is_google_chat_message_name(&reply.reply_to) { + tracing::warn!( + reply_to = %reply.reply_to, + "googlechat edit_message ignored: not a message resource name \ + (synthetic ids cannot be patched)" + ); + return; } + self.edit_message(&reply.reply_to, &reply.content.text).await; + return; } + _ => {} + } + + let result = self.deliver_message(reply).await; + if reply.request_id.is_some() { + emit_delivery_response(event_tx, &reply.request_id, result); + } else if let Err(e) = result { + error!(error = %e, "googlechat reply delivery failed without acknowledgement id"); } } } @@ -795,6 +1056,8 @@ impl GoogleChatTokenCache { .and_then(|v| v.as_str()) .ok_or("missing private_key in SA key")? .to_string(); + jsonwebtoken::EncodingKey::from_rsa_pem(pkey.as_bytes()) + .map_err(|e| format!("invalid private_key in SA key: {e}"))?; Ok(Self { token: RwLock::new(None), sa_email: email, @@ -806,27 +1069,47 @@ impl GoogleChatTokenCache { { let guard = self.token.read().await; if let Some((ref tok, ref ts, ttl)) = *guard { - if ts.elapsed().as_secs() < ttl.saturating_sub(TOKEN_REFRESH_MARGIN_SECS) { + if ts.elapsed().as_secs() < refresh_threshold(ttl) { return Ok(tok.clone()); } } } let mut guard = self.token.write().await; if let Some((ref tok, ref ts, ttl)) = *guard { - if ts.elapsed().as_secs() < ttl.saturating_sub(TOKEN_REFRESH_MARGIN_SECS) { + if ts.elapsed().as_secs() < refresh_threshold(ttl) { return Ok(tok.clone()); } } - let (new_token, expire) = self.refresh(client).await?; - *guard = Some((new_token.clone(), Instant::now(), expire)); - info!("googlechat access token refreshed (expires in {expire}s)"); - Ok(new_token) + match self.refresh(client).await { + Ok((new_token, expire)) => { + *guard = Some((new_token.clone(), Instant::now(), expire)); + info!("googlechat access token refreshed (expires in {expire}s)"); + Ok(new_token) + } + Err(e) => { + // Serve the still-valid cached token on a transient exchange + // failure instead of dropping the reply. + if let Some((ref tok, ref ts, ttl)) = *guard { + let elapsed = ts.elapsed().as_secs(); + if elapsed < ttl { + warn!( + "googlechat token refresh failed ({e}); serving cached token \ + still valid for {}s", + ttl - elapsed + ); + return Ok(tok.clone()); + } + } + Err(e) + } + } } async fn refresh(&self, client: &reqwest::Client) -> Result<(String, u64), String> { let jwt = self.build_jwt().map_err(|e| format!("JWT build error: {e}"))?; let resp = client .post("https://oauth2.googleapis.com/token") + .timeout(TOKEN_REQUEST_TIMEOUT) .form(&[ ("grant_type", "urn:ietf:params:oauth:grant-type:jwt-bearer"), ("assertion", &jwt), @@ -843,12 +1126,15 @@ impl GoogleChatTokenCache { let token = body .get("access_token") .and_then(|v| v.as_str()) + // Boundary validation: an empty token would be cached as "valid" + // and fail every send with 401 until the refresh threshold. + .and_then(non_empty_token) .ok_or_else(|| { let err = body .get("error_description") .and_then(|v| v.as_str()) .unwrap_or("unknown error"); - format!("token exchange failed: {err}") + format!("token exchange returned missing/empty access_token: {err}") })? .to_string(); @@ -882,6 +1168,370 @@ impl GoogleChatTokenCache { } } +// --- Keyless ADC token source (GCE metadata → IAM Credentials) --- + +/// Google Chat's `chat.bot` scope for the minted token. +const ADC_CHAT_BOT_SCOPE: &str = "https://www.googleapis.com/auth/chat.bot"; +/// Lifetime we request from IAM Credentials for the impersonated token, and +/// the fallback TTL we cache it under. IAM caps impersonated tokens at 3600s. +const ADC_TOKEN_LIFETIME_SECS: u64 = 3600; +/// After a failed refresh while a cached token remains valid, reuse that token +/// for this cooldown instead of making each queued sender repeat all metadata/ +/// IAM calls serially under the write lock. +const ADC_REFRESH_RETRY_COOLDOWN_SECS: u64 = 30; + +/// Cache TTL (seconds) to use for a minted token, derived from the IAM +/// response's `expireTime`. Falls back to the full lifetime when the field is +/// missing/unparseable, and clamps to `[0, ADC_TOKEN_LIFETIME_SECS]` so an +/// org-policy-shortened token isn't cached past its real expiry (0 forces a +/// fresh mint next call rather than serving a dead token). +fn ttl_from_expire_time(expire_time: &str, now: chrono::DateTime) -> u64 { + match chrono::DateTime::parse_from_rfc3339(expire_time) { + Ok(exp) => (exp.with_timezone(&chrono::Utc) - now) + .num_seconds() + .clamp(0, ADC_TOKEN_LIFETIME_SECS as i64) as u64, + Err(_) => ADC_TOKEN_LIFETIME_SECS, + } +} + +/// Age (seconds) at which a cached token must be refreshed: its ttl minus a +/// margin, where the margin is capped at half the ttl. A fixed 300 s margin +/// would make `elapsed < ttl - 300` always false for any `ttl <= 300`, forcing +/// a re-mint on every single send; capping keeps short-lived tokens cacheable. +/// For `ttl = 0` the threshold is 0, so an expired token is never served from +/// the cache. +fn refresh_threshold(ttl: u64) -> u64 { + ttl.saturating_sub(TOKEN_REFRESH_MARGIN_SECS.min(ttl / 2)) +} + +/// Mints a `chat.bot`-scoped access token **without** a service-account key +/// file, using two distinct identities. Flow, per refresh: +/// 1. read the attached runtime SA's email + base token from GCE metadata +/// 2. call IAM Credentials `generateAccessToken` for the separately configured +/// Google Chat target SA and request the `chat.bot` scope +/// +/// The runtime SA requires `roles/iam.serviceAccountTokenCreator` on the target +/// SA. The identities MUST differ: Google prohibits using a service account's +/// short-lived access token to generate another access token for itself. +pub struct MetadataTokenSource { + token: RwLock>, + refresh_retry_after: RwLock>, + target_service_account: String, + // Private: only `new` (prod, fixed trusted hosts) or the in-module + // `with_bases` (tests, mock server) may set these. Unexported ⇒ no in-process + // caller can retarget the metadata bearer to an arbitrary host. + metadata_base: String, + iam_credentials_base: String, + // Metadata is plaintext/link-local: never redirect it or allow a proxy hop. + metadata_client: reqwest::Client, + // IAM is a public HTTPS API and must retain the deployment's configured + // egress proxy while still refusing bearer-carrying redirects. + iam_client: reqwest::Client, +} + +impl MetadataTokenSource { + /// Production constructor for a target already normalized and validated by + /// [`normalize_adc_target_service_account`]. Kept private so callers cannot + /// bypass that invariant. + fn new(target_service_account: String) -> Self { + Self::with_bases( + target_service_account, + "http://metadata.google.internal".into(), + "https://iamcredentials.googleapis.com".into(), + ) + } + + /// Construct with explicit endpoint bases. Prod always goes through `new` + /// with HTTPS IAM; tests point these at a mock server. + fn with_bases( + target_service_account: String, + metadata_base: String, + iam_credentials_base: String, + ) -> Self { + let metadata_client = Self::build_metadata_client(reqwest::Client::builder()); + let iam_client = Self::build_iam_client(reqwest::Client::builder()); + Self::with_bases_and_clients( + target_service_account, + metadata_base, + iam_credentials_base, + metadata_client, + iam_client, + ) + } + + fn build_metadata_client(builder: reqwest::ClientBuilder) -> reqwest::Client { + builder + // The metadata bearer must never follow a redirect onto a third + // host, nor traverse a proxy (the plaintext metadata token would + // then transit an operator/attacker-controlled hop). Fail loud + // rather than silently falling back to an unsafe default client. + .redirect(reqwest::redirect::Policy::none()) + .no_proxy() + .build() + .expect("no-redirect, no-proxy metadata client must build") + } + + fn build_iam_client(builder: reqwest::ClientBuilder) -> reqwest::Client { + builder + // IAM is HTTPS egress, so preserve HTTP(S)_PROXY/system proxy + // routing while preventing a bearer-carrying redirect. + .redirect(reqwest::redirect::Policy::none()) + .build() + .expect("no-redirect IAM client must build") + } + + fn with_bases_and_clients( + target_service_account: String, + metadata_base: String, + iam_credentials_base: String, + metadata_client: reqwest::Client, + iam_client: reqwest::Client, + ) -> Self { + Self { + token: RwLock::new(None), + refresh_retry_after: RwLock::new(None), + target_service_account, + metadata_base, + iam_credentials_base, + metadata_client, + iam_client, + } + } + + /// Cached-or-refresh, mirroring [`GoogleChatTokenCache::get_token`]: + /// double-checked locking around the RwLock so only one refresh runs. + pub async fn get_token(&self) -> Result { + { + let guard = self.token.read().await; + if let Some((ref tok, ref ts, ttl)) = *guard { + if ts.elapsed().as_secs() < refresh_threshold(ttl) { + return Ok(tok.clone()); + } + } + } + let mut guard = self.token.write().await; + if let Some((ref tok, ref ts, ttl)) = *guard { + let elapsed = ts.elapsed().as_secs(); + if elapsed < refresh_threshold(ttl) { + return Ok(tok.clone()); + } + // A previous refresh failed but this token is still valid. While + // the cooldown is active, queued/subsequent callers reuse it + // immediately instead of repeating up to three HTTP timeouts. + if elapsed < ttl + && self + .refresh_retry_after + .read() + .await + .is_some_and(|deadline| deadline > Instant::now()) + { + return Ok(tok.clone()); + } + } + match self.refresh().await { + Ok(minted) => { + *self.refresh_retry_after.write().await = None; + let MintedToken { + token, + ttl, + runtime_service_account, + target_service_account, + } = minted; + if ttl == 0 { + // Freshly minted but already at/after its expireTime — almost + // always local clock skew (Google validates against its own + // clock, so the token may still work). Serve it once, but do + // not cache a dead token: the next call re-mints. + warn!( + runtime_service_account = %runtime_service_account, + target_service_account = %target_service_account, + "googlechat ADC minted a token with ttl=0 (clock skew?); \ + serving once without caching" + ); + return Ok(token); + } + *guard = Some((token.clone(), Instant::now(), ttl)); + info!( + runtime_service_account = %runtime_service_account, + target_service_account = %target_service_account, + "googlechat ADC token minted (distinct target, chat.bot, ttl {ttl}s)" + ); + Ok(token) + } + Err(e) => { + // During a transient metadata/IAM failure, serve the cached + // token while it is still valid rather than dropping the reply. + if let Some((ref tok, ref ts, ttl)) = *guard { + let elapsed = ts.elapsed().as_secs(); + if elapsed < ttl { + *self.refresh_retry_after.write().await = Some( + Instant::now() + + std::time::Duration::from_secs( + ADC_REFRESH_RETRY_COOLDOWN_SECS, + ), + ); + warn!( + "googlechat ADC refresh failed ({e}); serving cached token \ + still valid for {}s; retry suppressed for {}s", + ttl - elapsed, + ADC_REFRESH_RETRY_COOLDOWN_SECS + ); + return Ok(tok.clone()); + } + } + Err(e) + } + } + } + + async fn refresh(&self) -> Result { + // Metadata and IAM have different proxy requirements. Both clients + // reject redirects, but only the plaintext metadata path disables proxying. + let metadata_client = &self.metadata_client; + // 1. Default SA email from the GCE metadata server. + let email = metadata_client + .get(format!( + "{}/computeMetadata/v1/instance/service-accounts/default/email", + self.metadata_base + )) + .header("Metadata-Flavor", "Google") + .timeout(TOKEN_REQUEST_TIMEOUT) + .send() + .await + .map_err(|e| format!("metadata email request failed: {e}"))? + .error_for_status() + .map_err(|e| format!("metadata email status: {e}"))? + .text() + .await + .map_err(|e| format!("metadata email read failed: {e}"))?; + let runtime_service_account = email.trim(); + if runtime_service_account.is_empty() { + return Err("metadata returned empty runtime SA email".into()); + } + if runtime_service_account.eq_ignore_ascii_case(&self.target_service_account) { + return Err(format!( + "ADC target service account must differ from runtime service account \ + (self-impersonation is prohibited): {runtime_service_account}" + )); + } + + // 2. Base access token for the default SA from the metadata server. + let base: serde_json::Value = metadata_client + .get(format!( + "{}/computeMetadata/v1/instance/service-accounts/default/token", + self.metadata_base + )) + .header("Metadata-Flavor", "Google") + .timeout(TOKEN_REQUEST_TIMEOUT) + .send() + .await + .map_err(|e| format!("metadata token request failed: {e}"))? + .error_for_status() + .map_err(|e| format!("metadata token status: {e}"))? + .json() + .await + .map_err(|e| format!("metadata token parse failed: {e}"))?; + let base_token = base + .get("access_token") + .and_then(|v| v.as_str()) + .and_then(non_empty_token) + .ok_or("metadata token response missing or empty access_token")?; + + // 3. Exchange the runtime SA's base token for a chat.bot-scoped token + // for the distinct configured Chat-app service account. + let url = format!( + "{}/v1/projects/-/serviceAccounts/{}:generateAccessToken", + self.iam_credentials_base, self.target_service_account + ); + let resp = self + .iam_client + .post(&url) + .bearer_auth(base_token) + .json(&serde_json::json!({ + "scope": [ADC_CHAT_BOT_SCOPE], + "lifetime": format!("{ADC_TOKEN_LIFETIME_SECS}s"), + })) + .timeout(TOKEN_REQUEST_TIMEOUT) + .send() + .await + .map_err(|e| format!("generateAccessToken request failed: {e}"))?; + let status = resp.status(); + if !status.is_success() { + // Classify the common GCP causes so on-call can act on the log + // line directly instead of decoding GCP error prose (mirrors the + // operator guidance in docs/google-chat.md Option C). + let body: String = resp + .text() + .await + .unwrap_or_default() + .chars() + .take(400) + .collect(); + let reason = classify_generate_access_token_error(status.as_u16(), &body); + return Err(format!( + "generateAccessToken failed (status {status}, reason={reason}): {body}" + )); + } + let resp: serde_json::Value = resp + .json() + .await + .map_err(|e| format!("generateAccessToken parse failed: {e}"))?; + let token = resp + .get("accessToken") + .and_then(|v| v.as_str()) + // Boundary validation: an empty token would be cached as "valid" + // for its full TTL and bypass the static-token degradation path. + .and_then(non_empty_token) + .ok_or("generateAccessToken response missing or empty accessToken")? + .to_string(); + // Cache under the server-granted lifetime (respects an org policy that + // shortens impersonated tokens below the requested 3600s), falling back + // to the full lifetime when expireTime is absent/unparseable. + let ttl = resp + .get("expireTime") + .and_then(|v| v.as_str()) + .map(|e| ttl_from_expire_time(e, chrono::Utc::now())) + .unwrap_or(ADC_TOKEN_LIFETIME_SECS); + Ok(MintedToken { + token, + ttl, + runtime_service_account: runtime_service_account.to_string(), + target_service_account: self.target_service_account.clone(), + }) + } +} + +/// A successfully minted ADC token plus the source and target identities for +/// audit logging of the supported two-service-account impersonation flow. +struct MintedToken { + token: String, + ttl: u64, + runtime_service_account: String, + target_service_account: String, +} + +/// Best-effort classification of common GCP `generateAccessToken` failures. +/// Ordering matters: the insufficient-scope 403 body says "scopes" (not +/// "permission"), which is the documented signal distinguishing it from a +/// missing `roles/iam.serviceAccountTokenCreator` binding. +fn classify_generate_access_token_error(status: u16, body: &str) -> &'static str { + let b = body.to_ascii_lowercase(); + if b.contains("failed_precondition") || b.contains("same service account") { + "self_impersonation_prohibited" + } else if b.contains("api has not been used") + || b.contains("service_disabled") + || b.contains("is disabled") + { + "api_not_enabled" + } else if status == 403 && b.contains("scopes") { + "insufficient_scope" + } else if status == 403 { + "missing_role" + } else { + "unclassified" + } +} + /// Convert markdown to Google Chat native formatting. /// /// Called by both `send_message` and `edit_message`. Assumes the caller passes @@ -1115,6 +1765,7 @@ async fn send_message( .post(&url) .bearer_auth(token) .json(&body) + .timeout(CHAT_API_REQUEST_TIMEOUT) .send() .await; @@ -1122,11 +1773,15 @@ async fn send_message( Ok(r) if r.status().is_success() => { let body = r.text().await.unwrap_or_default(); let parsed: serde_json::Value = serde_json::from_str(&body).unwrap_or_default(); - parsed - .get("name") - .and_then(|v| v.as_str()) - .map(String::from) - .ok_or_else(|| "missing message name in response".into()) + if let Some(name) = parsed.get("name").and_then(|v| v.as_str()) { + Ok(name.to_string()) + } else { + warn!( + response_body = %body, + "googlechat accepted message but omitted its resource name; using synthetic receipt" + ); + Ok("googlechat_sent".into()) + } } Ok(r) => { let status = r.status(); @@ -1786,6 +2441,12 @@ mod tests { // --- Token cache tests --- + const VALID_SA_KEY_JSON: &str = r#"{ + "type": "service_account", + "client_email": "test@test.iam.gserviceaccount.com", + "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQDdcQN1uyXnPVop\n76hhCUfZswb7/2AAqHPiikNp8Kkxx5R9Up/vGCEJDxbZ0iZy/oPVs27Fa3LtM+9D\n3p9qpbQLZLw3AhK3R/VoS4Ex0p6EkvcxqfQ4OolaJlDG+/myFEuZLqEs164DFUIJ\ntr4dygVdU+jFy6A+h7Q5T1QXArG4pR6Ap7DuBJhgEtvPqAn6U7JxAMEELcAZbhHU\nARjfiks2uFMaew1DKrH1qLzIj159ZmhgrXM/4Vc1lSgKMYyqUN3Kued05/DXd2rd\ngfpX7af9tyYsD6Lh1bwREJ4wDOYSewUZxj35mhZhdxHDlZa8wzXQdN8Z04eUghEL\nA+9whYa9AgMBAAECggEAEB0mNE3/Dxmu0vhml1EWkmftrS7DLKkVbbnD+BSUK1Qr\noQb/LmXiGYeokQcy1xFgRI+/Esyj21D5K0Yq8ZbHSED3VUVoWT/6QrGj4B1Efb84\nD7wfUmwoDJBXNnOlkujZK3dyMRIsznqgiJZstTw7MbRmbuZHbeVwHu9/3gDLL/Vp\nXET6KvhGbe5ssmBAefQhOsB+L0aKh4h0HuJdeD+Q7Xhc9gTKRZBwWs6xJgCUqLCr\n5XI7MqjMGDmDN8CImos8zOHCXCYJvIj5BHT1iVkdYn+VmeSDCh+hFcTxmVnSoYf9\nf7G14C160QeEnl0gGxE3yMqMvhI9tb9nXR9vRZjqnQKBgQD18YdRsZLhi9ZnZzXI\nMs1/BCslDnka6yt6q0jYlkzRn/mCyg9QzyFeQpD1fyoWVclRKnYbJygvA90ZkHCN\nXxn7M+0yYWxO9hF+ghK/bhGsVQE/nElLLumt5P8JH0rKB/NvTHpjvM4Q26A23mMp\n9/46gWEHpeIhJL852k2H+a9fZwKBgQDmfwEbVW9wLNDmYaG3z4su7B/n4DHWZrt/\nF5+XziZiEza2NIVFsa03ZaFPHRIK1m6ExDFnnbBd34CU0o3SMsxrTj92VEDb4LdF\nw9foEZnq0wMWT6y9qCZysYQCIJ/4Uy3Un98sbXc1HBJZ9lWq3baaeDttxvyhifNq\nPsmZtPzmOwKBgC3VDceelOWtPo5UgIRHW15BM50bPlxS2O5qPxAFqlkiO8gwyXvg\nrbI4K3Vkdj5lTDfw9sOGn4lraeeasC7YOypB+gD6gMmSN55gtQexhl+cE7h78nit\nTGTYmOJlT3Wo16e1E9XEWI5xr0CqXsZybZEPjTp0olhU1cH9OZeOYy0fAoGAUjp0\n1p+ABfC3BblGzCBKcw7hwwMERIyZzxlKYgm1P7/DAPVzpg1g0iZ7iZHBYgRloQ+s\n4F4tERAu+uiyl45vxsg/c6NTEB32w/i+CZhd5JwqucbqxS47qScTBP9Gknx6GSR/\npYXXxSailV1/6lj2T90ctmkKr0ZbhEep/B/JKQkCgYBx6jih7r21GZMrGLI4mN2H\nirN6VYrNklGBhgIBToJ9cp3iTApyr1pwYQSGAhjkA7ERS4UelhYOY6k466mBX+eR\nIUWbl2pgsjuySiq2WPEnjaI1pXoPGl5xDHVCEXH1+wUq4+iLCUuIeZ9mLRKY9xVT\nEE0jaWDa0qXlgQH+iAsJNw==\n-----END PRIVATE KEY-----\n" + }"#; + #[test] fn token_cache_rejects_invalid_json() { let result = GoogleChatTokenCache::new("not json"); @@ -1801,52 +2462,754 @@ mod tests { } #[test] - fn token_cache_accepts_valid_sa_key() { - let key = r#"{ - "type": "service_account", + fn token_cache_rejects_invalid_pem() { + let key = serde_json::json!({ "client_email": "test@test.iam.gserviceaccount.com", - "private_key": "-----BEGIN RSA PRIVATE KEY-----\nMIIBogIBAAJBALvRE+oCMiEhtfO5ufaVc9wGPUMgPGxmVFiMPC/NMxmCSiMGNO9h\nCOyByeF78QHp4gOW/lgVU8MJkv33hVMbOr0CAwEAAQJAD2k/cFR5MIkw1PFcm98K\n9MqYKGpJCmGBjFY0ek0FHoC14d/hpAGaoWMjNaAyjU/IbGv1fj8C5MfFRal0fV/L\nAQIhAP0T6FPJMm3O4bM18kMHnOP2+Y5kxMpVxCCjkVNH7D09AiEAvXEQJYwR+PFs\njDDhEm4VPmk+lKJoQlopj8TN5gQV8DECIBcXbU+LPWx4H+qRElhCB1B5a9mYmpY\nV6LFPnvSfHqNAiEAiNj5+A6E7WJ50il+5NG5yn7gXh8vNxdCYIw5qx6C2bECIBmW\nVGVRhSmNsmDMJFsGIdKJsnEXpizIVHtfpXsS4j9X\n-----END RSA PRIVATE KEY-----\n" - }"#; - let result = GoogleChatTokenCache::new(key); - assert!(result.is_ok()); + "private_key": "-----BEGIN PRIVATE KEY-----\nnot-a-real-key\n-----END PRIVATE KEY-----\n", + }) + .to_string(); + let error = match GoogleChatTokenCache::new(&key) { + Err(error) => error, + Ok(_) => panic!("expected invalid PEM to be rejected during construction"), + }; + assert!(error.contains("invalid private_key in SA key"), "{error}"); } - // --- Bot filtering logic test --- - #[test] - fn bot_user_type_detected() { - let json = make_envelope("hello", None, "BOT", "DM", None); - let envelope: GoogleChatEnvelope = serde_json::from_str(&json).unwrap(); - let chat = envelope.chat.unwrap(); - let sender = chat - .message_payload - .as_ref() - .and_then(|p| p.message.as_ref()) - .and_then(|m| m.sender.as_ref()) - .or(chat.user.as_ref()); - let is_bot = sender.map(|s| s.user_type == "BOT").unwrap_or(false); - assert!(is_bot); + fn token_cache_accepts_valid_sa_key() { + assert!(GoogleChatTokenCache::new(VALID_SA_KEY_JSON).is_ok()); } - // --- JWT verifier tests --- + // --- Keyless ADC (MetadataTokenSource) tests --- - #[tokio::test] - async fn jwt_rejects_missing_bearer_prefix() { - let verifier = GoogleChatJwtVerifier::new("123456".into()); - let result = verifier.verify("NotBearer xyz").await; - assert!(result.is_err()); - assert!(result.unwrap_err().contains("Bearer")); + #[test] + fn ttl_from_expire_time_derives_and_clamps() { + use chrono::{DateTime, Utc}; + let now: DateTime = "2026-08-25T00:00:00Z".parse().unwrap(); + // Normal: 30 min out → 1800s. + assert_eq!(ttl_from_expire_time("2026-08-25T00:30:00Z", now), 1800); + // Beyond the 3600s cap → clamped to 3600. + assert_eq!(ttl_from_expire_time("2026-08-25T05:00:00Z", now), 3600); + // Already expired → 0 (forces refresh next call, never caches a dead token). + assert_eq!(ttl_from_expire_time("2026-08-24T23:00:00Z", now), 0); + // Unparseable → safe fallback to the full lifetime. + assert_eq!(ttl_from_expire_time("not-a-timestamp", now), 3600); } #[tokio::test] - async fn jwt_rejects_invalid_token() { - let verifier = GoogleChatJwtVerifier::new("123456".into()); - let result = verifier.verify("Bearer not.a.valid.jwt").await; - assert!(result.is_err()); - } + #[ignore = "integration: uses a local HTTP server"] + async fn metadata_token_source_mints_chat_bot_token() { + use wiremock::matchers::{header, method, path, path_regex}; + use wiremock::{Mock, MockServer, ResponseTemplate}; - #[tokio::test] - async fn jwt_rejects_empty_bearer() { + let server = MockServer::start().await; + // GCE metadata: default SA email. + Mock::given(method("GET")) + .and(path( + "/computeMetadata/v1/instance/service-accounts/default/email", + )) + .and(header("Metadata-Flavor", "Google")) + .respond_with( + ResponseTemplate::new(200) + .set_body_string("openab-host@dev-seba.iam.gserviceaccount.com"), + ) + .mount(&server) + .await; + // GCE metadata: base access token. + Mock::given(method("GET")) + .and(path( + "/computeMetadata/v1/instance/service-accounts/default/token", + )) + .and(header("Metadata-Flavor", "Google")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "access_token": "base-tok", + "expires_in": 3600, + "token_type": "Bearer" + }))) + .mount(&server) + .await; + // IAM Credentials: runtime SA impersonates distinct Chat target. + Mock::given(method("POST")) + .and(path_regex( + r"/v1/projects/-/serviceAccounts/.*:generateAccessToken", + )) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "accessToken": "chat-bot-tok", + "expireTime": "2099-01-01T00:00:00Z" + }))) + .mount(&server) + .await; + + let src = MetadataTokenSource::with_bases( + "chat-bot@project.iam.gserviceaccount.com".into(), + server.uri(), + server.uri(), + ); + + let token = src + .get_token() + .await + .expect("should mint a chat.bot token"); + assert_eq!(token, "chat-bot-tok"); + } + + #[tokio::test] + #[ignore = "integration: uses a local HTTP server"] + async fn metadata_token_source_rejects_self_impersonation_before_token_request() { + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let server = MockServer::start().await; + let same_sa = "runtime@project.iam.gserviceaccount.com"; + Mock::given(method("GET")) + .and(path( + "/computeMetadata/v1/instance/service-accounts/default/email", + )) + .respond_with(ResponseTemplate::new(200).set_body_string(same_sa)) + .expect(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path( + "/computeMetadata/v1/instance/service-accounts/default/token", + )) + .respond_with(ResponseTemplate::new(200)) + .expect(0) + .mount(&server) + .await; + + let src = MetadataTokenSource::with_bases( + same_sa.into(), + server.uri(), + server.uri(), + ); + let err = src + .get_token() + .await + .expect_err("same runtime/target SA must be rejected"); + assert!(err.contains("self-impersonation is prohibited"), "{err}"); + } + + #[tokio::test] + #[ignore = "integration: validates reqwest proxy routing with local HTTP servers"] + async fn metadata_bypasses_proxy_while_iam_uses_proxy_client() { + use wiremock::matchers::{header, method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let metadata_server = MockServer::start().await; + let iam_proxy = MockServer::start().await; + Mock::given(method("GET")) + .and(path( + "/computeMetadata/v1/instance/service-accounts/default/email", + )) + .respond_with( + ResponseTemplate::new(200) + .set_body_string("runtime@project.iam.gserviceaccount.com"), + ) + .expect(1) + .mount(&metadata_server) + .await; + Mock::given(method("GET")) + .and(path( + "/computeMetadata/v1/instance/service-accounts/default/token", + )) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "access_token": "base-tok", "expires_in": 3600 + }))) + .expect(1) + .mount(&metadata_server) + .await; + Mock::given(method("POST")) + .and(header("authorization", "Bearer base-tok")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "accessToken": "chat-bot-tok" + }))) + .expect(1) + .mount(&iam_proxy) + .await; + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(500)) + .expect(0) + .mount(&iam_proxy) + .await; + + // Start both production policy helpers from a builder carrying the + // same explicit proxy. Metadata must clear it; IAM must retain it. + let metadata_client = MetadataTokenSource::build_metadata_client( + reqwest::Client::builder() + .proxy(reqwest::Proxy::all(iam_proxy.uri()).unwrap()), + ); + let iam_client = MetadataTokenSource::build_iam_client( + reqwest::Client::builder() + .proxy(reqwest::Proxy::all(iam_proxy.uri()).unwrap()), + ); + let src = MetadataTokenSource::with_bases_and_clients( + "chat-bot@project.iam.gserviceaccount.com".into(), + metadata_server.uri(), + "http://iam.invalid".into(), + metadata_client, + iam_client, + ); + + assert_eq!(src.get_token().await.unwrap(), "chat-bot-tok"); + } + + #[tokio::test] + #[ignore = "integration: uses a local HTTP server"] + async fn adc_refresh_failure_cooldown_prevents_queued_retry_storm() { + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path( + "/computeMetadata/v1/instance/service-accounts/default/email", + )) + .respond_with( + ResponseTemplate::new(200) + .set_body_string("runtime@project.iam.gserviceaccount.com"), + ) + .expect(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path( + "/computeMetadata/v1/instance/service-accounts/default/token", + )) + .respond_with(ResponseTemplate::new(503)) + .expect(1) + .mount(&server) + .await; + + let src = MetadataTokenSource::with_bases( + "chat-bot@project.iam.gserviceaccount.com".into(), + server.uri(), + server.uri(), + ); + *src.token.write().await = Some(( + "cached-token".into(), + Instant::now() - std::time::Duration::from_secs(6), + 10, + )); + + // First call attempts refresh, gets 503, serves the still-valid token, + // and starts cooldown. Second call must reuse it without another GET. + assert_eq!(src.get_token().await.unwrap(), "cached-token"); + assert_eq!(src.get_token().await.unwrap(), "cached-token"); + } + + #[test] + fn from_parts_use_adc_toggles_metadata_source() { + let with = GoogleChatAdapter::from_parts(GoogleChatParts { + use_adc: true, + adc_target_service_account: Some("chat-bot@project.iam.gserviceaccount.com".into()), + ..Default::default() + }); + assert!(with.metadata_source.is_some(), "use_adc=true → ADC source"); + let without = GoogleChatAdapter::from_parts(GoogleChatParts::default()); + assert!( + without.metadata_source.is_none(), + "use_adc=false → no ADC source" + ); + } + + #[test] + fn from_parts_use_adc_without_target_fails_closed() { + use std::io::Write; + use std::sync::{Arc as StdArc, Mutex as StdMutex}; + + #[derive(Clone)] + struct Capture(StdArc>>); + impl Write for Capture { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + let buffer = StdArc::new(StdMutex::new(Vec::new())); + let capture = Capture(buffer.clone()); + let subscriber = tracing_subscriber::fmt() + .with_writer(move || capture.clone()) + .with_ansi(false) + .without_time() + .finish(); + let adapter = tracing::subscriber::with_default(subscriber, || { + GoogleChatAdapter::from_parts(GoogleChatParts { + use_adc: true, + ..Default::default() + }) + }); + assert!( + adapter.metadata_source.is_none(), + "use_adc without adc_target_service_account must not install ADC" + ); + let output = String::from_utf8(buffer.lock().unwrap().clone()).unwrap(); + assert!( + output.contains( + "googlechat use_adc=true has no adc_target_service_account; ADC is disabled" + ), + "missing ADC target must emit a startup diagnostic: {output}" + ); + } + + #[test] + fn from_parts_use_adc_with_invalid_target_fails_closed() { + let adapter = GoogleChatAdapter::from_parts(GoogleChatParts { + use_adc: true, + adc_target_service_account: Some("123456789012345678901".into()), + ..Default::default() + }); + assert!( + adapter.metadata_source.is_none(), + "numeric unique IDs must not bypass the service-account email invariant" + ); + } + + #[test] + fn from_parts_malformed_key_with_use_adc_installs_adc() { + // Regression: a configured-but-malformed SA key parses to + // token_cache=None; with use_adc=true the ADC source is still installed + // (from_parts logs a warning naming the identity switch). + let adapter = GoogleChatAdapter::from_parts(GoogleChatParts { + sa_key_json: Some("not valid json".into()), + use_adc: true, + adc_target_service_account: Some("chat-bot@project.iam.gserviceaccount.com".into()), + ..Default::default() + }); + assert!( + adapter.token_cache.is_none(), + "malformed SA key → no SA-key cache" + ); + assert!( + adapter.metadata_source.is_some(), + "use_adc=true → ADC source installed even when a key was configured but failed to load" + ); + } + + #[test] + fn from_parts_invalid_pem_with_use_adc_installs_adc() { + let key = serde_json::json!({ + "client_email": "sa@example.iam.gserviceaccount.com", + "private_key": "-----BEGIN PRIVATE KEY-----\nnot-a-real-key\n-----END PRIVATE KEY-----\n", + }) + .to_string(); + let adapter = GoogleChatAdapter::from_parts(GoogleChatParts { + sa_key_json: Some(key), + use_adc: true, + adc_target_service_account: Some("chat-bot@project.iam.gserviceaccount.com".into()), + ..Default::default() + }); + assert!(adapter.token_cache.is_none(), "invalid PEM must not load"); + assert!( + adapter.metadata_source.is_some(), + "invalid PEM must leave the configured ADC fallback selectable" + ); + } + + #[test] + #[ignore = "integration: exercises filesystem error handling"] + fn from_parts_unreadable_key_file_with_use_adc_installs_adc() { + // Regression: an unreadable/absent key FILE also yields token_cache=None + // and must not suppress ADC. + let adapter = GoogleChatAdapter::from_parts(GoogleChatParts { + sa_key_file: Some("/nonexistent/path/sa-key.json".into()), + use_adc: true, + adc_target_service_account: Some("chat-bot@project.iam.gserviceaccount.com".into()), + ..Default::default() + }); + assert!(adapter.token_cache.is_none(), "unreadable key file → no cache"); + assert!( + adapter.metadata_source.is_some(), + "use_adc=true → ADC source installed when the key file could not be read" + ); + } + + #[test] + fn from_parts_loaded_key_suppresses_metadata_source() { + // A successfully loaded and PEM-validated SA key wins outright at send + // time, so no dead ADC source is installed behind it. + let adapter = GoogleChatAdapter::from_parts(GoogleChatParts { + sa_key_json: Some(VALID_SA_KEY_JSON.into()), + use_adc: true, + adc_target_service_account: Some("chat-bot@project.iam.gserviceaccount.com".into()), + ..Default::default() + }); + assert!(adapter.token_cache.is_some(), "valid SA key loads cache"); + assert!( + adapter.metadata_source.is_none(), + "loaded SA key suppresses ADC source" + ); + } + + #[test] + fn classify_generate_access_token_error_covers_documented_cases() { + assert_eq!( + classify_generate_access_token_error( + 400, + r#"{"error":{"status":"FAILED_PRECONDITION","message":"You can't create a token for the same service account that you used to authenticate the request."}}"#, + ), + "self_impersonation_prohibited" + ); + // Insufficient scope: the 403 body says "scopes" — the documented + // signal distinguishing it from a missing IAM role. + assert_eq!( + classify_generate_access_token_error( + 403, + r#"{"error":{"status":"PERMISSION_DENIED","message":"Request had insufficient authentication scopes."}}"# + ), + "insufficient_scope" + ); + // Missing serviceAccountTokenCreator: 403 without the scopes wording. + assert_eq!( + classify_generate_access_token_error( + 403, + r#"{"error":{"status":"IAM_PERMISSION_DENIED","message":"Permission 'iam.serviceAccounts.getAccessToken' denied on resource"}}"# + ), + "missing_role" + ); + // IAM Credentials API not enabled. + assert_eq!( + classify_generate_access_token_error( + 403, + r#"{"error":{"message":"IAM Service Account Credentials API has not been used in project 123 before or it is disabled."}}"# + ), + "api_not_enabled" + ); + // Anything else stays unclassified rather than guessing. + assert_eq!(classify_generate_access_token_error(500, "boom"), "unclassified"); + } + + #[test] + fn external_token_values_must_be_non_whitespace() { + assert_eq!(None::<&str>.and_then(non_empty_token), None); + assert_eq!(non_empty_token(""), None); + assert_eq!(non_empty_token(" \t\n"), None); + assert_eq!(non_empty_token(" token "), Some(" token ")); + } + + #[test] + fn adc_target_requires_user_managed_service_account_email() { + assert_eq!( + normalize_adc_target_service_account(Some( + " Chat-Bot@Project-1.iam.gserviceaccount.com ".into() + )) + .unwrap() + .as_deref(), + Some("chat-bot@project-1.iam.gserviceaccount.com") + ); + for invalid in [ + "123456789012345678901", + "runtime@project-1.iam.gserviceaccount.com.", + "runtime@project-1.iam.gserviceaccount.com?x=y", + "runtime@project-1.iam.gserviceaccount.com/../other", + "-runtime@project-1.iam.gserviceaccount.com", + ] { + assert!( + normalize_adc_target_service_account(Some(invalid.into())).is_err(), + "accepted invalid ADC target: {invalid}" + ); + } + } + + #[tokio::test] + async fn delivery_deadline_includes_wait_for_serialization_lock() { + let adapter = GoogleChatAdapter::new(None, Some("token".into()), None); + let held = adapter.delivery_lock.lock().await; + let reply = GatewayReply { + schema: "openab.gateway.reply.v1".into(), + reply_to: "orig".into(), + platform: "googlechat".into(), + channel: ReplyChannel { + id: "spaces/TEST".into(), + thread_id: None, + }, + content: Content { + content_type: "text".into(), + attachments: vec![], + text: "hello".into(), + }, + command: None, + request_id: Some("req_queue".into()), + quote_message_id: None, + }; + + let err = adapter + .deliver_message_before( + &reply, + tokio::time::Instant::now() + std::time::Duration::from_millis(1), + ) + .await + .expect_err("queued delivery must time out before the held lock is released"); + drop(held); + assert!(err.contains("delivery queue timed out"), "{err}"); + assert!(err.contains("before sending any chunk"), "{err}"); + } + + #[tokio::test] + async fn static_access_token_rejects_whitespace_at_adapter_boundary() { + let adapter = GoogleChatAdapter::new(None, Some(" \t\n".into()), None); + assert!(adapter.access_token.is_none()); + assert_eq!(adapter.get_token().await, None); + } + + #[tokio::test] + #[ignore = "integration: uses a local HTTP server"] + async fn adc_failure_does_not_fall_back_to_whitespace_static_token() { + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path( + "/computeMetadata/v1/instance/service-accounts/default/email", + )) + .respond_with(ResponseTemplate::new(503)) + .expect(1) + .mount(&server) + .await; + + let mut adapter = GoogleChatAdapter::new(None, Some(" \t\n".into()), None); + adapter.metadata_source = Some(MetadataTokenSource::with_bases( + "chat-bot@project.iam.gserviceaccount.com".into(), + server.uri(), + server.uri(), + )); + + assert_eq!(adapter.get_token().await, None); + assert!(adapter.access_token.is_none()); + } + + #[test] + fn edit_targets_require_full_message_resource_names() { + assert!(is_google_chat_message_name("spaces/SP/messages/msg1")); + assert!(!is_google_chat_message_name("unified_a1b2c3d4")); + assert!(!is_google_chat_message_name("spaces/")); + assert!(!is_google_chat_message_name("spaces/SP")); + assert!(!is_google_chat_message_name("spaces/SP/messages/")); + assert!(!is_google_chat_message_name("spaces/SP/messages/msg1/extra")); + assert!(!is_google_chat_message_name("spaces/SP/messages/msg1?x=1")); + assert!(!is_google_chat_message_name("spaces/SP/messages/msg1#fragment")); + assert!(!is_google_chat_message_name("spaces/SP/messages/msg%2F1")); + } + + #[tokio::test] + #[ignore = "integration: uses a local HTTP server"] + async fn adc_takes_precedence_over_static_access_token() { + use wiremock::matchers::{method, path, path_regex}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path( + "/computeMetadata/v1/instance/service-accounts/default/email", + )) + .respond_with( + ResponseTemplate::new(200) + .set_body_string("openab-host@dev-seba.iam.gserviceaccount.com"), + ) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path( + "/computeMetadata/v1/instance/service-accounts/default/token", + )) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "access_token": "base-tok", "expires_in": 3600 + }))) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path_regex( + r"/v1/projects/-/serviceAccounts/.*:generateAccessToken", + )) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "accessToken": "chat-bot-tok" + }))) + .mount(&server) + .await; + + // Adapter has BOTH an ADC source and a static token; ADC must win. + let mut adapter = GoogleChatAdapter::from_parts(GoogleChatParts { + access_token: Some("static-tok".into()), + use_adc: true, + adc_target_service_account: Some("chat-bot@project.iam.gserviceaccount.com".into()), + ..Default::default() + }); + // Repoint the ADC source at the mock server (bases are private now). + adapter.metadata_source = Some(MetadataTokenSource::with_bases( + "chat-bot@project.iam.gserviceaccount.com".into(), + server.uri(), + server.uri(), + )); + let token = adapter.get_token().await.expect("a token"); + assert_eq!(token, "chat-bot-tok", "ADC should win over static token"); + } + + #[tokio::test] + #[ignore = "integration: uses a local HTTP server"] + async fn metadata_token_source_rejects_blank_minted_token() { + use wiremock::matchers::{method, path, path_regex}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path( + "/computeMetadata/v1/instance/service-accounts/default/email", + )) + .respond_with( + ResponseTemplate::new(200) + .set_body_string("openab-host@dev-seba.iam.gserviceaccount.com"), + ) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path( + "/computeMetadata/v1/instance/service-accounts/default/token", + )) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "access_token": "base-tok", "expires_in": 3600 + }))) + .mount(&server) + .await; + // A malformed IAM response with an empty accessToken must surface as a + // mint error (and thus follow the static-token degradation path), not + // be cached as a "valid" credential. + Mock::given(method("POST")) + .and(path_regex( + r"/v1/projects/-/serviceAccounts/.*:generateAccessToken", + )) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "accessToken": "" + }))) + .mount(&server) + .await; + + let src = MetadataTokenSource::with_bases( + "chat-bot@project.iam.gserviceaccount.com".into(), + server.uri(), + server.uri(), + ); + let err = src.get_token().await.expect_err("blank token must be rejected"); + assert!( + err.contains("missing or empty accessToken"), + "unexpected error: {err}" + ); + } + + #[tokio::test] + #[ignore = "integration: uses a local HTTP server"] + async fn handle_reply_edit_message_ignores_synthetic_unified_id() { + use wiremock::matchers::method; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + // Expect ZERO requests: a synthetic `unified_` id is not a valid + // message resource name, so the edit must be refused locally instead + // of being sent to the API (which would 400 INVALID_ARGUMENT). + let mock_server = MockServer::start().await; + Mock::given(method("PATCH")) + .respond_with(ResponseTemplate::new(200)) + .expect(0) + .mount(&mock_server) + .await; + + let (event_tx, _event_rx) = tokio::sync::broadcast::channel::(16); + let mut adapter = GoogleChatAdapter::new(None, Some("fake-token".into()), None); + adapter.api_base = mock_server.uri(); + + let reply = GatewayReply { + schema: "openab.gateway.reply.v1".into(), + reply_to: "unified_a1b2c3d4e5f6".into(), + platform: "googlechat".into(), + channel: ReplyChannel { + id: "spaces/SP".into(), + thread_id: None, + }, + content: Content { + content_type: "text".into(), + attachments: Vec::new(), + text: "updated text".into(), + }, + command: Some("edit_message".into()), + request_id: None, + quote_message_id: None, + }; + + adapter.handle_reply(&reply, &event_tx).await; + // MockServer verifies the expect(0) on drop. + } + + #[tokio::test] + async fn handle_reply_delete_message_is_explicit_noop() { + let (event_tx, mut event_rx) = tokio::sync::broadcast::channel::(16); + let adapter = GoogleChatAdapter::new(None, Some("fake-token".into()), None); + let reply = GatewayReply { + schema: "openab.gateway.reply.v1".into(), + reply_to: "spaces/SP/messages/msg1".into(), + platform: "googlechat".into(), + channel: ReplyChannel { + id: "spaces/SP".into(), + thread_id: None, + }, + content: Content { + content_type: "text".into(), + attachments: Vec::new(), + text: String::new(), + }, + command: Some("delete_message".into()), + // Under the old fallthrough behavior this request id produced an + // "empty message" GatewayResponse. Explicit command routing emits + // no response and performs no token/network work. + request_id: Some("req_delete".into()), + quote_message_id: None, + }; + + adapter.handle_reply(&reply, &event_tx).await; + assert!( + event_rx.try_recv().is_err(), + "delete_message must return before the empty-send response path" + ); + } + + // --- Bot filtering logic test --- + + #[test] + fn bot_user_type_detected() { + let json = make_envelope("hello", None, "BOT", "DM", None); + let envelope: GoogleChatEnvelope = serde_json::from_str(&json).unwrap(); + let chat = envelope.chat.unwrap(); + let sender = chat + .message_payload + .as_ref() + .and_then(|p| p.message.as_ref()) + .and_then(|m| m.sender.as_ref()) + .or(chat.user.as_ref()); + let is_bot = sender.map(|s| s.user_type == "BOT").unwrap_or(false); + assert!(is_bot); + } + + // --- JWT verifier tests --- + + #[tokio::test] + async fn jwt_rejects_missing_bearer_prefix() { + let verifier = GoogleChatJwtVerifier::new("123456".into()); + let result = verifier.verify("NotBearer xyz").await; + assert!(result.is_err()); + assert!(result.unwrap_err().contains("Bearer")); + } + + #[tokio::test] + async fn jwt_rejects_invalid_token() { + let verifier = GoogleChatJwtVerifier::new("123456".into()); + let result = verifier.verify("Bearer not.a.valid.jwt").await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn jwt_rejects_empty_bearer() { let verifier = GoogleChatJwtVerifier::new("123456".into()); let result = verifier.verify("Bearer ").await; assert!(result.is_err()); @@ -2017,6 +3380,26 @@ mod tests { assert_eq!(name, Some("spaces/SP1/messages/msg123")); } + #[test] + fn emit_delivery_response_reports_missing_adapter_failure() { + let (event_tx, mut event_rx) = tokio::sync::broadcast::channel::(4); + emit_delivery_response( + &event_tx, + &Some("req_missing_adapter".into()), + Err("googlechat adapter is not configured".into()), + ); + + let response: GatewayResponse = + serde_json::from_str(&event_rx.try_recv().expect("expected failure response")).unwrap(); + assert_eq!(response.request_id, "req_missing_adapter"); + assert!(!response.success); + assert!(response.message_id.is_none()); + assert_eq!( + response.error.as_deref(), + Some("googlechat adapter is not configured") + ); + } + #[tokio::test] async fn handle_reply_sends_gateway_response_success() { use wiremock::{Mock, MockServer, ResponseTemplate}; @@ -2246,6 +3629,7 @@ mod tests { .respond_with(ResponseTemplate::new(200).set_body_json( serde_json::json!({"name": "spaces/SP/messages/msg1"}), )) + .expect(1) .mount(&mock_server) .await; @@ -2326,9 +3710,10 @@ mod tests { #[tokio::test] async fn handle_reply_multi_chunk_partial_failure_reports_failure() { - // Mixed success/failure: chunk 1 succeeds, subsequent chunks fail. - // Expect success=false (any chunk failure marks overall as failed), - // but message_id is still set so core has a reference. + // Mixed success/failure: chunk 1 succeeds, chunk 2 fails. The direct + // Result contract reports no overall message receipt, but the error + // retains explicit partial-delivery context so operators know one chunk + // already reached the space and should not blindly retry the whole turn. use wiremock::{Mock, MockServer, ResponseTemplate}; use wiremock::matchers::{method, path_regex}; @@ -2379,9 +3764,69 @@ mod tests { let resp: GatewayResponse = serde_json::from_str(&received.unwrap()).unwrap(); assert_eq!(resp.request_id, "req_partial"); assert!(!resp.success, "partial failure must report success=false"); - assert_eq!(resp.message_id, Some("spaces/TEST/messages/first_chunk".into())); + assert!(resp.message_id.is_none(), "partial delivery is not an overall receipt"); let err = resp.error.expect("partial failure should set error"); - assert!(err.contains("500")); + assert!(err.contains("chunk 2/2"), "{err}"); + assert!(err.contains("after 1 successful chunk(s)"), "{err}"); + assert!(err.contains("500"), "{err}"); + } + + #[tokio::test] + #[ignore = "integration: delayed local HTTP response validates partial-timeout context"] + async fn multi_chunk_timeout_reports_successful_chunk_count() { + use wiremock::matchers::{method, path_regex}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path_regex("/spaces/.*/messages")) + .respond_with(ResponseTemplate::new(200).set_body_json( + serde_json::json!({"name": "spaces/TEST/messages/first_chunk"}), + )) + .up_to_n_times(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path_regex("/spaces/.*/messages")) + .respond_with( + ResponseTemplate::new(200) + .set_delay(std::time::Duration::from_millis(250)) + .set_body_json( + serde_json::json!({"name": "spaces/TEST/messages/late_chunk"}), + ), + ) + .mount(&server) + .await; + + let mut adapter = GoogleChatAdapter::new(None, Some("fake-token".into()), None); + adapter.api_base = server.uri(); + let reply = GatewayReply { + schema: "openab.gateway.reply.v1".into(), + reply_to: "orig".into(), + platform: "googlechat".into(), + channel: ReplyChannel { + id: "spaces/TEST".into(), + thread_id: None, + }, + content: Content { + content_type: "text".into(), + attachments: vec![], + text: "x".repeat(5000), + }, + command: None, + request_id: Some("req_partial_timeout".into()), + quote_message_id: None, + }; + + let err = adapter + .deliver_message_before( + &reply, + tokio::time::Instant::now() + std::time::Duration::from_millis(75), + ) + .await + .expect_err("second chunk must exceed the absolute deadline"); + assert!(err.contains("chunk 2/2 timed out"), "{err}"); + assert!(err.contains("after 1 successful chunk(s)"), "{err}"); } // --- Attachment parsing tests --- diff --git a/crates/openab-gateway/src/lib.rs b/crates/openab-gateway/src/lib.rs index d2b257c7d..587cc15fe 100644 --- a/crates/openab-gateway/src/lib.rs +++ b/crates/openab-gateway/src/lib.rs @@ -195,10 +195,7 @@ impl AppState { .unwrap_or(false); if enabled { Some(adapters::googlechat::GoogleChatAdapter::from_parts( - std::env::var("GOOGLE_CHAT_SA_KEY_JSON").ok(), - std::env::var("GOOGLE_CHAT_SA_KEY_FILE").ok(), - std::env::var("GOOGLE_CHAT_ACCESS_TOKEN").ok(), - std::env::var("GOOGLE_CHAT_AUDIENCE").ok(), + adapters::googlechat::GoogleChatParts::from_env(), )) } else { None @@ -450,10 +447,14 @@ impl AppState { self.googlechat_webhook_path = cfg.webhook_path; self.google_chat = if cfg.enabled { Some(adapters::googlechat::GoogleChatAdapter::from_parts( - cfg.sa_key_json, - cfg.sa_key_file, - cfg.access_token, - cfg.audience, + adapters::googlechat::GoogleChatParts { + sa_key_json: cfg.sa_key_json, + sa_key_file: cfg.sa_key_file, + access_token: cfg.access_token, + audience: cfg.audience, + use_adc: cfg.use_adc, + adc_target_service_account: cfg.adc_target_service_account, + }, )) } else { None @@ -554,6 +555,8 @@ pub struct GatewayGoogleChatConfig { pub sa_key_file: Option, pub access_token: Option, pub audience: Option, + pub use_adc: bool, + pub adc_target_service_account: Option, pub webhook_path: String, } @@ -748,10 +751,7 @@ pub async fn serve(config: ServeConfig) -> anyhow::Result<()> { info!(path = %googlechat_webhook_path, "googlechat adapter enabled"); app = app.route(&googlechat_webhook_path, post(adapters::googlechat::webhook)); Some(adapters::googlechat::GoogleChatAdapter::from_parts( - std::env::var("GOOGLE_CHAT_SA_KEY_JSON").ok(), - std::env::var("GOOGLE_CHAT_SA_KEY_FILE").ok(), - std::env::var("GOOGLE_CHAT_ACCESS_TOKEN").ok(), - std::env::var("GOOGLE_CHAT_AUDIENCE").ok(), + adapters::googlechat::GoogleChatParts::from_env(), )) } else { None @@ -954,6 +954,25 @@ async fn ws_handler( ws.on_upgrade(move |socket| handle_oab_connection(state, socket)) } +#[cfg(feature = "googlechat")] +fn spawn_googlechat_reply( + state: Arc, + reply: schema::GatewayReply, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + if let Some(ref google_chat) = state.google_chat { + google_chat.handle_reply(&reply, &state.event_tx).await; + } else { + tracing::warn!("reply for googlechat but adapter not configured"); + adapters::googlechat::emit_delivery_response( + &state.event_tx, + &reply.request_id, + Err("googlechat adapter is not configured".into()), + ); + } + }) +} + async fn handle_oab_connection(state: Arc, socket: axum::extract::ws::WebSocket) { use axum::extract::ws::Message; use futures_util::{SinkExt, StreamExt}; @@ -1051,11 +1070,14 @@ async fn handle_oab_connection(state: Arc, socket: axum::extract::ws:: } #[cfg(feature = "googlechat")] "googlechat" => { - if let Some(ref gc) = state_for_recv.google_chat { - gc.handle_reply(&reply, &state_for_recv.event_tx).await; - } else { - warn!("reply for googlechat but adapter not configured"); - } + // Do not await delivery inline: core's ack timer is already + // running, and the receive loop must keep draining replies. + // Dropping the JoinHandle detaches the delivery task; its deadline + // includes any wait for the adapter's delivery lock. + std::mem::drop(spawn_googlechat_reply( + state_for_recv.clone(), + reply.clone(), + )); } #[cfg(feature = "wecom")] "wecom" => { @@ -1107,6 +1129,51 @@ async fn handle_oab_connection(state: Arc, socket: axum::extract::ws:: info!("OAB client disconnected"); } +#[cfg(all(test, feature = "googlechat"))] +mod googlechat_dispatch_tests { + use super::{schema, spawn_googlechat_reply, AppState}; + use std::sync::Arc; + use tokio::sync::broadcast; + + #[tokio::test] + async fn missing_adapter_dispatch_emits_correlated_failure() { + let (event_tx, mut event_rx) = broadcast::channel::(4); + let state = Arc::new(AppState::test_default(event_tx)); + let reply = schema::GatewayReply { + schema: "openab.gateway.reply.v1".into(), + reply_to: "original".into(), + platform: "googlechat".into(), + channel: schema::ReplyChannel { + id: "spaces/TEST".into(), + thread_id: None, + }, + content: schema::Content { + content_type: "text".into(), + text: "hello".into(), + attachments: vec![], + }, + command: None, + request_id: Some("req_route_missing".into()), + quote_message_id: None, + }; + + spawn_googlechat_reply(state, reply).await.unwrap(); + + let response: schema::GatewayResponse = serde_json::from_str( + &event_rx + .try_recv() + .expect("missing adapter must emit a failure response"), + ) + .unwrap(); + assert_eq!(response.request_id, "req_route_missing"); + assert!(!response.success); + assert_eq!( + response.error.as_deref(), + Some("googlechat adapter is not configured") + ); + } +} + async fn health() -> &'static str { "ok" } @@ -1249,6 +1316,8 @@ mod l1_audit_tests { sa_key_file: None, access_token: Some("tok".into()), audience: None, + use_adc: false, + adc_target_service_account: None, webhook_path: "/hook/gc".into(), }); assert!(s.google_chat.is_some()); @@ -1262,6 +1331,8 @@ mod l1_audit_tests { sa_key_file: None, access_token: Some("tok".into()), audience: Some("aud".into()), + use_adc: false, + adc_target_service_account: None, webhook_path: "/hook/gc".into(), }); assert!(flagged(&s).is_empty()); @@ -1273,6 +1344,8 @@ mod l1_audit_tests { sa_key_file: None, access_token: None, audience: None, + use_adc: false, + adc_target_service_account: None, webhook_path: "/hook/gc".into(), }); assert!(s.google_chat.is_none()); diff --git a/crates/openab-gateway/tests/config_first_conformance.rs b/crates/openab-gateway/tests/config_first_conformance.rs index eeefc8d8c..447b2216f 100644 --- a/crates/openab-gateway/tests/config_first_conformance.rs +++ b/crates/openab-gateway/tests/config_first_conformance.rs @@ -92,6 +92,8 @@ const COVERED: &[&str] = &[ "GOOGLE_CHAT_SA_KEY_JSON", "GOOGLE_CHAT_SA_KEY_FILE", "GOOGLE_CHAT_ACCESS_TOKEN", + "GOOGLE_CHAT_USE_ADC", + "GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT", "GOOGLE_CHAT_AUDIENCE", "GOOGLE_CHAT_WEBHOOK_PATH", "GOOGLE_CHAT_ALLOW_ALL_USERS", diff --git a/docs/config-reference.md b/docs/config-reference.md index 3e0a29af5..838ceacb6 100644 --- a/docs/config-reference.md +++ b/docs/config-reference.md @@ -225,11 +225,24 @@ Full first-class Google Chat section (config-first parity, #1379) — credential | `sa_key_json` | string | — | Inline service-account key JSON (wins over `sa_key_file`). Env: `GOOGLE_CHAT_SA_KEY_JSON`. | | `sa_key_file` | string | — | Path to a service-account key file. Env: `GOOGLE_CHAT_SA_KEY_FILE`. | | `access_token` | string | — | Static access token alternative. Env: `GOOGLE_CHAT_ACCESS_TOKEN`. | +| `use_adc` | bool | `false` | Enable keyless ADC: the attached runtime SA impersonates a distinct Chat-app target via IAM Credentials. Requires `roles/iam.serviceAccountTokenCreator` on the target + `iamcredentials.googleapis.com`. Self-impersonation is prohibited. Env: `GOOGLE_CHAT_USE_ADC`. | +| `adc_target_service_account` | string | — | Dedicated Chat-app SA email to impersonate. Required with `use_adc=true`; MUST differ from the runtime SA. Env: `GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT`. | | `audience` | string | — | JWT audience — enables webhook JWT verification (L1). Env: `GOOGLE_CHAT_AUDIENCE`. | | `webhook_path` | string | `/webhook/googlechat` | Env: `GOOGLE_CHAT_WEBHOOK_PATH`. | | `allow_all_users` | bool \| omit | `false` (deny-all) | Env: `GOOGLE_CHAT_ALLOW_ALL_USERS`. | | `allowed_users` | string[] | `[]` | User resource names (`users/`). Env: `GOOGLE_CHAT_ALLOWED_USERS`. | +Outbound auth precedence is: a successfully loaded service-account key, then +ADC, then the configured static `access_token`. ADC mint failures may fall back +to that static token and log a possible identity switch. + +For GKE Workload Identity, set Helm value +`agents..gateway.serviceAccountName` to an existing annotated Kubernetes +ServiceAccount for the runtime GSA. An empty value omits `serviceAccountName` +and preserves the Kubernetes default identity; it does not inherit per-agent or +chart-global ServiceAccount values. The chart does not create or annotate the +ServiceAccount. + --- ## `[teams]` diff --git a/docs/google-chat.md b/docs/google-chat.md index dcee33d0e..4cfb8a6f0 100644 --- a/docs/google-chat.md +++ b/docs/google-chat.md @@ -74,7 +74,7 @@ Google Chat uses a service account to authenticate outbound API calls (bot repli ## 3. Configure the Gateway -The gateway supports two authentication methods for sending replies: +The gateway supports three authentication methods for sending replies: ### Option A: Service Account Key (recommended — auto-refresh) @@ -112,6 +112,54 @@ docker run -d --name openab-gateway \ ghcr.io/openabdev/openab-gateway:latest ``` +### Option C: Keyless ADC (recommended on GCP — no key file) + +When the gateway runs on GCP (GKE / GCE / Cloud Run), its attached **runtime service account** can impersonate a separate **Google Chat service account** and mint a `chat.bot` token without any key file. The gateway reads the runtime identity and a base token from the GCE metadata server, then calls IAM Credentials `generateAccessToken` for the configured Chat-app target identity. + +The two identities **must be different**. Google prohibits using a service account's short-lived access token to generate another access token for that same service account (`FAILED_PRECONDITION`); see [Service account credentials — Self-impersonation](https://cloud.google.com/iam/docs/service-account-creds#self-impersonation). + +Prerequisites: + +- Attach a runtime SA to the workload (for example `openab-runtime@PROJECT.iam.gserviceaccount.com`). +- Use a distinct SA as the Google Chat app identity (for example `openab-chat@PROJECT.iam.gserviceaccount.com`); that target SA must be the app/space member that sends messages. +- Grant the runtime SA `roles/iam.serviceAccountTokenCreator` **on the target Chat SA**. +- Enable `iamcredentials.googleapis.com`. +- The metadata base token must carry `cloud-platform` (or `.../auth/iam`) scope. A default-scope GCE VM returns `403 PERMISSION_DENIED: "Request had insufficient authentication scopes."`; match *scopes* to distinguish it from a missing role. GCE access scopes cannot be changed while the VM is running: create the VM with `--scopes=cloud-platform`, or use `gcloud compute instances set-scopes` followed by a stop/start. + +`chat.bot` is a Workspace scope and is not a subset of `cloud-platform`, so the runtime metadata token cannot call Chat directly. The supported flow is runtime SA → distinct target Chat SA via `generateAccessToken`. + +```bash +export GOOGLE_CHAT_ENABLED=true +export GOOGLE_CHAT_USE_ADC=true +export GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT="openab-chat@PROJECT.iam.gserviceaccount.com" +``` + +For GKE + Helm, bind a Kubernetes ServiceAccount (KSA) to the **runtime** GSA +out of band, then attach that KSA to the gateway pod. The chart references an +existing KSA; it does not create or annotate one: + +```yaml +agents: + kiro: + gateway: + enabled: true + serviceAccountName: openab-googlechat-runtime + googleChat: + useAdc: true + adcTargetServiceAccount: openab-chat@PROJECT.iam.gserviceaccount.com +``` + +The `openab-googlechat-runtime` KSA must carry the usual +`iam.gke.io/gcp-service-account: openab-runtime@PROJECT.iam.gserviceaccount.com` +annotation and Workload Identity IAM binding. Set +`gateway.serviceAccountName` explicitly to attach that KSA. An empty value +preserves the Kubernetes default identity and does not inherit per-agent or +chart-global ServiceAccount values. + +Precedence: if a configured SA key loads successfully, it wins and ADC is ignored. If a key is configured but fails to load, the adapter uses the configured ADC target and logs the identity switch. `GOOGLE_CHAT_USE_ADC=true` without `GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT` fails closed (ADC is not installed). If ADC fails and a static token is explicitly configured, the adapter degrades to it with a warning that it may represent a different identity. + +> **Migrating an existing release from a SA key to ADC:** the chart renders the Google Chat Secret only when `saKeyJson` / `accessToken` is set, and that Secret carries `helm.sh/resource-policy: keep`. Switching to ADC-only stops Helm from managing it but **leaves the old key material in the cluster indefinitely**. Delete the orphaned Secret after the switch: it is the gateway Secret named by the chart's `openab.agentFullname` helper — `--gateway` by default (or the agent's `nameOverride`) — and it contains the `google-chat-sa-key-json` key. Find it with `kubectl get secrets -o name | grep gateway`, confirm with `kubectl get secret -o jsonpath='{.data}' | grep -o google-chat-sa-key-json`, then `kubectl delete secret `. Otherwise the "no key to mount or leak" benefit is undercut. + ### Local development ```bash @@ -199,7 +247,7 @@ Each field falls back to its `GOOGLE_CHAT_ALLOW_ALL_USERS` / `GOOGLE_CHAT_ALLOWE - Links: `[text](url)` → `` - Inline code, fenced code blocks: pass through unchanged - Tables and other unsupported syntax pass through as-is -- **Streaming (edit_message)** — when OAB streaming is enabled, the bot edits its initial reply in-place as tokens arrive (typewriter effect) +- **Send-once (no streaming)** — Google Chat is a request/response REST surface, so the adapter posts the full reply once with no in-place editing. It is in `NON_STREAMING_PLATFORMS`; see `docs/platforms/schema/googlechat.toml` for why (the unified adapter's synthetic message id is not a valid resource name → `patch` returns `400 INVALID_ARGUMENT`, and the API documents a 1 write/sec-per-space quota). - **Inbound attachments** — image, text file, and audio attachments are downloaded via Google Chat Media API and stored to `~/.openab/media/inbound/` (colocate filesystem store): - Images: resized to ≤1200px JPEG (q75); GIFs preserved. Max 10 MB. - Text files: only known text extensions (`.txt`, `.md`, `.json`, `.py`, `.rs`, etc.). Max 512 KB. @@ -221,6 +269,8 @@ Each field falls back to its `GOOGLE_CHAT_ALLOW_ALL_USERS` / `GOOGLE_CHAT_ALLOWE | `GOOGLE_CHAT_SA_KEY_JSON` | No | — | Service account key JSON string (enables auto-refresh) | | `GOOGLE_CHAT_SA_KEY_FILE` | No | — | Path to service account key JSON file (alternative to `SA_KEY_JSON`) | | `GOOGLE_CHAT_ACCESS_TOKEN` | No | — | Static OAuth2 access token (fallback, expires in 1 hour) | +| `GOOGLE_CHAT_USE_ADC` | No | `false` | Set to `true` or `1` to enable keyless ADC: attached runtime SA impersonates a distinct Chat-app target via IAM Credentials — see Option C | +| `GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT` | With ADC | — | Email of the dedicated Chat-app SA to impersonate. Required when `USE_ADC=true`; MUST differ from the runtime SA | | `GOOGLE_CHAT_WEBHOOK_PATH` | No | `/webhook/googlechat` | Webhook endpoint path | ## Security: Webhook Verification diff --git a/docs/platforms/schema/googlechat.toml b/docs/platforms/schema/googlechat.toml index 375e62663..c43489d18 100644 --- a/docs/platforms/schema/googlechat.toml +++ b/docs/platforms/schema/googlechat.toml @@ -131,9 +131,9 @@ pr = "" [[openab_features]] feature = "streaming" -status = "partial" -note = "No native streaming API. Gateway adapter leaves uses_native_streaming=false, so core streams via the post-then-edit_message loop; each edit sends the full accumulated text as a patch call." -source = ["crates/openab-gateway/src/adapters/googlechat.rs#edit_message", "crates/openab-core/src/adapter.rs#uses_native_streaming"] +status = "not_implemented" +note = "Send-once by design (no cosmetic streaming). The decisive reason is structural: the unified adapter returns a synthetic `unified_` message id that is not a valid resource name, so `spaces.messages.patch` rejects it with 400 INVALID_ARGUMENT ('Missing or malformed message resource name') before any content is applied — per-token post-then-edit cannot work at all. Separately, Google Chat documents a 1 write/sec-per-space quota (create+patch+delete combined; https://developers.google.com/workspace/chat/limits); treat that as a documented constraint on high-frequency editing rather than an observed hard failure, since enforcement is burst-tolerant in practice. googlechat is therefore in NON_STREAMING_PLATFORMS, and `resolve_streaming` forces send-once on BOTH the embedded dispatch and the WebSocket gateway paths — matching Google Chat's documented send-once default. (Previously 'partial': core attempted post-then-edit, which failed on every edit.)" +source = ["crates/openab-core/src/gateway.rs#NON_STREAMING_PLATFORMS", "crates/openab-core/src/adapter.rs#resolve_streaming", "crates/openab-core/src/adapter.rs#uses_native_streaming"] pr = "" [[openab_features]] @@ -153,7 +153,7 @@ pr = "" [[openab_features]] feature = "delete_message" status = "not_implemented" -note = "GatewayAdapter overrides delete_message to emit a fire-and-forget command:\"delete_message\" instead of the trait default (edit-to-zero-width). The googlechat adapter does not match delete_message in handle_reply (only add_reaction/remove_reaction/create_topic/edit_message), so it falls through to the send path with empty text → hits the empty-message short-circuit and sends nothing. Net: delete is a no-op on Google Chat." +note = "GatewayAdapter emits a fire-and-forget command:\"delete_message\" instead of the trait default (edit-to-zero-width). The googlechat adapter explicitly matches delete_message with the other unsupported commands and returns before token resolution, logging, or network I/O. Net: delete is an intentional no-op on Google Chat." source = ["crates/openab-gateway/src/adapters/googlechat.rs#handle_reply", "crates/openab-core/src/gateway.rs#delete_message", "crates/openab-core/src/adapter.rs#delete_message"] pr = "" @@ -245,6 +245,14 @@ kind = "openab_decision" source = "crates/openab-gateway/src/adapters/googlechat.rs#build_jwt" refs = [] +[[quirks]] +date = "2026-08-25" +title = "Keyless ADC outbound path (no SA key)" +note = "Outbound creds have a third option beside the SA-key JWT-bearer exchange and the static token: keyless ADC (use_adc / GOOGLE_CHAT_USE_ADC). MetadataTokenSource reads the attached runtime SA email + base token from GCE metadata, then calls IAM Credentials generateAccessToken for the distinct adc_target_service_account / GOOGLE_CHAT_ADC_TARGET_SERVICE_ACCOUNT and requests chat.bot. Google prohibits access-token self-impersonation, so runtime and target identities MUST differ; the code rejects equality before minting. The runtime SA needs roles/iam.serviceAccountTokenCreator on the target. Auth precedence: SA key > ADC target > explicitly configured static token (ADC-to-static degradation logs a possible identity switch)." +kind = "openab_decision" +source = "crates/openab-gateway/src/adapters/googlechat.rs#MetadataTokenSource" +refs = [] + [[quirks]] date = "2026-07-04" title = "Reactions are structurally impossible for the bot" @@ -287,8 +295,8 @@ refs = [] [[quirks]] date = "2026-07-04" -title = "Delete is a silent no-op (not even the edit fallback)" -note = "Unlike platforms where delete_message falls back to the trait's edit-to-zero-width, on Google Chat the delete_message command isn't matched in handle_reply, falls through to the send path with empty text, and hits the empty-message short-circuit — so nothing is sent and no edit occurs. Streaming-placeholder cleanup that relies on delete is therefore a no-op here." +title = "Delete is an explicit no-op (not even the edit fallback)" +note = "Unlike platforms where delete_message falls back to the trait's edit-to-zero-width, Google Chat matches the delete_message command in handle_reply and returns before token resolution, logging, or network I/O. Streaming-placeholder cleanup that relies on delete is therefore an intentional no-op here." kind = "openab_decision" source = "crates/openab-gateway/src/adapters/googlechat.rs#handle_reply" refs = [] diff --git a/docs/platforms/schema/lineworks.toml b/docs/platforms/schema/lineworks.toml index 066944446..9b6cd013e 100644 --- a/docs/platforms/schema/lineworks.toml +++ b/docs/platforms/schema/lineworks.toml @@ -132,8 +132,8 @@ pr = "" [[openab_features]] feature = "streaming" status = "n_a" -note = "No edit API to drive post+edit streaming. The platform is listed in NON_EDITABLE_PLATFORMS so the core forces streaming off and the cosmetic edit/delete commands are dropped by the dispatcher." -source = ["crates/openab-core/src/gateway.rs#NON_EDITABLE_PLATFORMS", "crates/openab-gateway/src/adapters/lineworks.rs#dispatch_lineworks_reply"] +note = "No edit API to drive post+edit streaming. The platform is listed in NON_STREAMING_PLATFORMS so the core forces streaming off and the cosmetic edit/delete commands are dropped by the dispatcher." +source = ["crates/openab-core/src/gateway.rs#NON_STREAMING_PLATFORMS", "crates/openab-gateway/src/adapters/lineworks.rs#dispatch_lineworks_reply"] pr = "" [[openab_features]] diff --git a/src/main.rs b/src/main.rs index a2ee786ac..f49879778 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1284,6 +1284,8 @@ async fn main() -> anyhow::Result<()> { sa_key_json: r.sa_key_json, sa_key_file: r.sa_key_file, access_token: r.access_token, + use_adc: r.use_adc, + adc_target_service_account: r.adc_target_service_account, audience: r.audience, webhook_path: r.webhook_path, }); diff --git a/src/unified_adapter.rs b/src/unified_adapter.rs index c943f4d0d..e34e5fb27 100644 --- a/src/unified_adapter.rs +++ b/src/unified_adapter.rs @@ -1,7 +1,7 @@ //! UnifiedGatewayAdapter — routes ChatAdapter calls through in-process gateway //! platform adapters based on the ChannelRef.platform field. -use anyhow::Result; +use anyhow::{anyhow, Result}; use async_trait::async_trait; use openab_core::adapter::{ChannelRef, ChatAdapter, MessageRef}; use openab_gateway::schema::{Content, GatewayReply, ReplyChannel}; @@ -10,6 +10,16 @@ use std::collections::HashMap; use std::sync::Arc; use tokio::sync::Mutex; +fn synthetic_unified_id() -> String { + format!( + "unified_{:x}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + ) +} + pub struct UnifiedGatewayAdapter { pub gw_state: Arc, /// Telegram reaction state (message_id -> emoji list) for add/remove_reaction @@ -24,8 +34,10 @@ impl UnifiedGatewayAdapter { } } - /// Dispatch a GatewayReply to the correct platform adapter. - async fn dispatch_reply(&self, reply: &GatewayReply) { + /// Dispatch a GatewayReply to the correct platform adapter. Platforms with + /// direct delivery receipts return the real message resource name; legacy + /// fire-and-forget adapters return `None`. + async fn dispatch_reply(&self, reply: &GatewayReply) -> Result> { let client = &self.gw_state.client; match reply.platform.as_str() { #[cfg(feature = "telegram")] @@ -69,7 +81,21 @@ impl UnifiedGatewayAdapter { #[cfg(feature = "googlechat")] "googlechat" => { if let Some(ref gc) = self.gw_state.google_chat { + if reply.command.is_none() { + return gc + .deliver_message(reply) + .await + .map(Some) + .map_err(anyhow::Error::msg); + } gc.handle_reply(reply, &self.gw_state.event_tx).await; + } else if reply.command.is_none() { + return Err(anyhow!("googlechat adapter is not configured")); + } else { + tracing::warn!( + command = ?reply.command.as_deref(), + "googlechat command dropped: adapter is not configured" + ); } } #[cfg(feature = "wecom")] @@ -112,9 +138,13 @@ impl UnifiedGatewayAdapter { } } other => { - tracing::warn!(platform = other, "unified adapter: unknown platform, cannot route reply"); + tracing::warn!( + platform = other, + "unified adapter: unknown platform, cannot route reply" + ); } } + Ok(None) } /// Build a GatewayReply from ChatAdapter parameters. @@ -157,11 +187,13 @@ impl ChatAdapter for UnifiedGatewayAdapter { async fn send_message(&self, channel: &ChannelRef, content: &str) -> Result { let reply = self.build_reply(channel, content, None, None); - self.dispatch_reply(&reply).await; + let message_id = self + .dispatch_reply(&reply) + .await? + .unwrap_or_else(synthetic_unified_id); Ok(MessageRef { channel: channel.clone(), - message_id: format!("unified_{:x}", std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_nanos()), + message_id, }) } @@ -172,7 +204,7 @@ impl ChatAdapter for UnifiedGatewayAdapter { title: &str, ) -> Result { let reply = self.build_reply(channel, title, Some("create_topic"), None); - self.dispatch_reply(&reply).await; + let _ = self.dispatch_reply(&reply).await?; // Return a thread channel ref with the trigger message as thread_id Ok(ChannelRef { platform: channel.platform.clone(), @@ -187,7 +219,7 @@ impl ChatAdapter for UnifiedGatewayAdapter { let mut reply = self.build_reply(&msg.channel, emoji, Some("add_reaction"), None); // Use the actual platform message_id (not origin_event_id which is a UUID) reply.reply_to = msg.message_id.clone(); - self.dispatch_reply(&reply).await; + let _ = self.dispatch_reply(&reply).await?; Ok(()) } @@ -195,7 +227,7 @@ impl ChatAdapter for UnifiedGatewayAdapter { let mut reply = self.build_reply(&msg.channel, emoji, Some("remove_reaction"), None); // Use the actual platform message_id (not origin_event_id which is a UUID) reply.reply_to = msg.message_id.clone(); - self.dispatch_reply(&reply).await; + let _ = self.dispatch_reply(&reply).await?; Ok(()) } @@ -203,7 +235,7 @@ impl ChatAdapter for UnifiedGatewayAdapter { let mut reply = self.build_reply(&msg.channel, content, Some("edit_message"), None); // Use the actual platform message_id (e.g. "draft" for streaming, or numeric for edits) reply.reply_to = msg.message_id.clone(); - self.dispatch_reply(&reply).await; + let _ = self.dispatch_reply(&reply).await?; Ok(()) } @@ -214,11 +246,13 @@ impl ChatAdapter for UnifiedGatewayAdapter { reply_to_message_id: &str, ) -> Result { let reply = self.build_reply(channel, content, None, Some(reply_to_message_id)); - self.dispatch_reply(&reply).await; + let message_id = self + .dispatch_reply(&reply) + .await? + .unwrap_or_else(synthetic_unified_id); Ok(MessageRef { channel: channel.clone(), - message_id: format!("unified_{:x}", std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_nanos()), + message_id, }) } @@ -247,3 +281,35 @@ impl ChatAdapter for UnifiedGatewayAdapter { platform == "telegram" && self.gw_state.telegram_rich_messages } } + +#[cfg(all(test, feature = "googlechat"))] +mod tests { + use super::*; + use openab_gateway::adapters::googlechat::GoogleChatAdapter; + use tokio::sync::broadcast; + + #[tokio::test] + async fn googlechat_send_failure_propagates_in_unified_mode() { + let (event_tx, _event_rx) = broadcast::channel(4); + let mut state = AppState::test_default(event_tx); + // No credential source: delivery fails before any network access. + state.google_chat = Some(GoogleChatAdapter::new(None, None, None)); + let adapter = UnifiedGatewayAdapter::new(Arc::new(state)); + let channel = ChannelRef { + platform: "googlechat".into(), + channel_id: "spaces/TEST".into(), + thread_id: None, + parent_id: None, + origin_event_id: Some("evt_test".into()), + }; + + let err = adapter + .send_message(&channel, "hello") + .await + .expect_err("unified mode must not synthesize success after delivery failure"); + assert!( + err.to_string().contains("no credentials configured"), + "{err}" + ); + } +}