Compare commits

...

19 Commits

Author SHA1 Message Date
Adam Weidman bb379102d6 fix: SSE keepalive, request timeout, NoOp task store caching, and Docker tooling
- Add SSE keepalive middleware (15s comment heartbeat) to prevent Cloud Run
  LB from closing idle SSE connections during long tool executions
- Increase A2A server request timeout from 300s to 3600s (1 hour)
- Cache loaded tasks in NoOpTaskStore to prevent redundant GCS workspace
  restores on every SDK event cycle (was restoring 5+ times per request)
- Add undici dispatcher with 10-min timeouts for bridge SSE connections
- Install gh CLI and gcloud CLI in Docker image for agent GitHub/GCP access
2026-02-20 10:02:13 -05:00
Adam Weidman 305a47e5b5 docs: update README known limitations for message splitting 2026-02-19 16:49:52 -05:00
Adam Weidman 9f0a2cf7b0 fix: streaming text extraction, YOLO timing, and message chunking
Streaming fixes:
- Fix A2UI text extraction preferring incremental chunks over accumulated
  text by swapping push order in extractFromStreamEvent
- Defer tool approval handling to post-stream — server YOLO mode briefly
  publishes awaiting_approval before auto-approving, which was breaking
  the stream loop prematurely
- Always update latestPendingApprovals to clear stale approvals after
  server auto-approves

Message chunking:
- Split long messages at paragraph/line boundaries to stay within Google
  Chat's 4096 character limit
- Cards attach to first chunk only

Streaming tool confirmations:
- Add sendToolConfirmationStream and sendBatchToolConfirmationsStream
  for SSE-based tool approval flow

README:
- Update bridge deployment to max-instances=1 (single instance for
  in-memory async guard consistency)
- Add Known Limitations section

Working:
- Server YOLO mode with streaming text extraction
- Chat API push with proper threading
- Session persistence across Cloud Run restarts
- Retry logic for agent concurrency exhaustion
- Text chunking for long responses
- Per-session /yolo and /safe commands

Not working:
- Tool confirmation streaming with GEMINI_YOLO_MODE=false (executor
  aborts on SSE disconnect — SDK-level issue)
- CARD_CLICKED button routing through Add-ons (text-based approval
  works as fallback)
2026-02-19 16:32:28 -05:00
Adam Weidman a9eb2ee5e3 fix: push all messages via Chat API for threading, add retry and model override
- All bot responses now pushed via Chat API instead of synchronous webhook
  response — Add-ons createMessageAction ignores thread info in Spaces,
  causing messages to appear as top-level instead of in-thread
- Webhook returns bare {} for empty responses to prevent Google Chat
  from retrying (wrapped empty createMessageAction caused retry storm)
- Add retryStream() with exponential backoff (3 retries, 5s/10s/20s)
  for transient A2A server 500/503 errors
- Add GEMINI_MODEL env var override in config (was hardcoded to
  gemini-3-pro-preview which hit capacity limits)
- Extract pushAndReturn() helper for fire-and-forget Chat API sends
2026-02-19 14:00:26 -05:00
Adam Weidman fc9623248d fix: session cancellation, YOLO auto-approval, threading, and approval UX
- Add session cancellation on /reset to stop in-flight async streams
- Implement per-session YOLO auto-approval with batch tool confirmations
  (sendBatchToolConfirmations sends all approvals in one A2A message to
  avoid hangs when agent needs ALL tools approved before proceeding)
- Fix threading: include thread info in Add-ons response wrapper so
  replies appear in the user's thread instead of top-level
- Make tool approval async: return immediate ack, process confirmation
  in background, push result via Chat API (fixes "Agent is processing..."
  empty response after approve)
- Replace text-based approval with clickable Approve/Always Allow/Reject
  buttons on compact Cards V2
- Wire CARD_CLICKED handler to async approval flow (fire-and-forget
  with UPDATE_MESSAGE ack)

Tested via Cloud Run proxy curl suite:
  /reset, simple messages, async guard, /yolo, /safe, CARD_CLICKED
  (approve + reject), ADDED_TO_SPACE, empty message, cancellation.
2026-02-19 11:42:06 -05:00
Adam Weidman 45c9545f78 fix: Cloud Run service-to-service auth and agent card URL
- Add identity token auth in A2ABridgeClient for Cloud Run (K_SERVICE)
- Support CODER_AGENT_PUBLIC_URL env var for agent card URL on Cloud Run
- Strip @Bot mention prefix before slash command detection (Add-ons)
- Grant bridge SA roles/run.invoker on A2A server via IAM
2026-02-15 00:22:08 -07:00
Adam Weidman 6872805274 feat: split chat bridge into standalone microservice
Separate the Google Chat bridge from the A2A agent server so each can
scale independently on Cloud Run. The bridge is a lightweight proxy
(concurrency=80) while the agent needs concurrency=1.

- Add standalone entry point (src/chat-bridge/server.ts)
- Add Dockerfile.chat-bridge and cloudbuild-chat-bridge.yaml
- Remove chat bridge setup from app.ts
- Inline A2UI constants in a2a-bridge-client.ts (no agent deps)
- Update README for two-service architecture
2026-02-14 20:04:32 -07:00
Adam Weidman 4f3ffc8959 feat: async streaming chat bridge with Chat API push
Replace blocking A2A calls with streaming: webhook returns
immediate "Processing..." response, then streams results from
A2A agent and pushes them to Google Chat via REST API.

- Add ChatApiClient for proactive messaging via Chat REST API
- Add sendMessageStream() to A2ABridgeClient for SSE streaming
- Add extractFromStreamEvent() for parsing individual stream events
- Refactor handler to fire-and-forget async processing
- Fix isTerminal logic to use stream state instead of taskId presence
- Add asyncProcessing guard to prevent overlapping requests
- Add comprehensive README with deployment and setup guide
2026-02-14 19:54:47 -07:00
Adam Weidman 03ef280120 fix: reply in user's thread and prevent git credential hangs
Pass thread name through Add-ons response wrapper so bot replies stay
in the user's thread instead of posting top-level messages. Add
GIT_TERMINAL_PROMPT=0 to Dockerfile to prevent git from hanging on
credential prompts, which was blocking all requests under concurrency=1.
2026-02-12 17:36:58 -05:00
Adam Weidman 60b01761b8 feat: add GCS-backed session and conversation persistence
Enable session resumability across Cloud Run restarts:

- executor.ts: Save conversation history in task metadata during
  toSDKTask(), restore via setHistory() in reconstruct()
- gcs.ts: Persist conversation history as separate GCS object
  (conversation.tar.gz) alongside metadata and workspace
- session-store.ts: Add optional GCS-backed persistence with periodic
  flush and restore-on-startup for thread→session mappings
- handler.ts: Restore persisted sessions on initialize()
- types.ts: Add gcsBucket to ChatBridgeConfig
- app.ts: Pass GCS_BUCKET_NAME to chat bridge config

Validated end-to-end: message persists, Cloud Run restarts, follow-up
message in same thread correctly recalls prior context. Different
threads maintain isolation.
2026-02-12 17:17:08 -05:00
Adam Weidman df81bfe1f2 feat: Google Chat bridge with YOLO mode, text-based approvals, and Add-ons support
- Add normalizeEvent() to convert Workspace Add-ons event format to legacy ChatEvent
- Add wrapAddOnsResponse() for Add-ons response wrapping
- Mount chat bridge routes BEFORE A2A SDK catch-all handler
- Replace card button approvals with text-based approve/reject/always allow
- Add /reset, /yolo, /safe slash commands for session control
- Add YOLO mode tool approval dedup (filter auto-approved surfaces)
- Add extractCommandSummary() for concise tool card display
- Delegate auth to Cloud Run IAM when K_SERVICE env var detected
- Add JWT debug logging for token claim inspection
2026-02-12 16:12:18 -05:00
Adam Weidman 1c926921ed fix: delegate auth to Cloud Run IAM when K_SERVICE is set
Cloud Run strips the Authorization header after using it for IAM
validation, so our JWT middleware can never see the token. When
running on Cloud Run (detected via K_SERVICE env var), skip
app-level JWT verification since Cloud Run IAM already ensures
only authorized service accounts (chat@system.gserviceaccount.com)
can reach the container.
2026-02-12 12:19:35 -05:00
Adam Weidman d1ca874dd1 fix: deduplicate agent response text from task history
Task history contains multiple status-update messages that may
reference the same A2UI surface. Use only the last non-empty
agent response to avoid duplicate text in Chat output.
2026-02-12 11:20:32 -05:00
Adam Weidman 470228e7a0 fix: extract A2UI content from task history in blocking mode
The blocking DefaultRequestHandler accumulates intermediate status-update
events into task.history. The A2UI response content from "working" events
lives there, while the final "input-required" status has no message.
Updated extractAllParts to check history. Reverted to blocking mode
since streaming had transport issues.
2026-02-12 11:10:50 -05:00
Adam Weidman 9d12980baa feat: add JWT verification middleware for Google Chat webhook
Verifies Bearer tokens from Google Chat using google-auth-library.
Checks issuer (chat@system.gserviceaccount.com) and audience
(CHAT_PROJECT_NUMBER). Verification is skipped when project number
is not configured, allowing local testing without tokens.
2026-02-12 10:56:42 -05:00
Adam Weidman b85a3bafe5 feat: switch chat bridge to streaming for full A2UI content capture
Blocking mode only returns the final task state, missing intermediate
A2UI response content from working status-update events. Streaming
captures all events and aggregates parts into the response.
2026-02-12 10:36:58 -05:00
Adam Weidman 72ea38b306 fix: resolve agent card URL for A2A client initialization
ClientFactory.createFromUrl expects the full agent card URL
(/.well-known/agent-card.json), not just the base server URL.
Also adds CHAT_BRIDGE_A2A_URL to k8s deployment and test script.
2026-02-12 10:23:47 -05:00
Adam Weidman 57f3c9ca1a feat: add Google Chat bridge for A2UI integration
Implements a Google Chat HTTP webhook bridge that connects Google Chat
to the A2A server. Each Chat thread maps to an A2A contextId/taskId
pair. The bridge converts A2UI tool approval surfaces to Google Chat
Cards V2 with Approve/Always Allow/Reject buttons, and handles
CARD_CLICKED events to forward tool confirmations back to the A2A server.

Components:
- chat-bridge/types.ts: Google Chat event/response types
- chat-bridge/session-store.ts: Thread -> A2A session mapping
- chat-bridge/a2a-bridge-client.ts: A2A SDK client wrapper
- chat-bridge/response-renderer.ts: A2UI -> Google Chat Cards V2
- chat-bridge/handler.ts: Event handler (MESSAGE, CARD_CLICKED)
- chat-bridge/routes.ts: Express routes mounted at /chat/webhook
2026-02-12 10:11:58 -05:00
Adam Weidman 5ea957c84b feat: add A2UI extension support and GKE deployment infrastructure
Implement A2UI v0.10 protocol compliance for the a2a-server:
- Add a2ui-extension.ts with constants, Part helpers, extension detection
- Add a2ui-components.ts with standard catalog component builders
- Add a2ui-surface-manager.ts for tool approval, response, thought surfaces
- Integrate A2UI into task.ts and executor.ts
- Add configurable bind host (CODER_AGENT_HOST) for k8s compatibility

Add GKE deployment infrastructure:
- Dockerfile, cloudbuild.yaml, k8s deployment manifest, .dockerignore
2026-02-11 19:23:56 -05:00
26 changed files with 4759 additions and 17 deletions
+15
View File
@@ -0,0 +1,15 @@
.git
.github
.gcp
bundle
evals
integration-tests
docs
packages/cli
packages/vscode-ide-companion
packages/test-utils
**/*.test.ts
**/*.test.js
**/src/**/*.ts
!packages/a2a-server/dist/**
!packages/core/dist/**
+25 -1
View File
@@ -2255,6 +2255,7 @@
"integrity": "sha512-t54CUOsFMappY1Jbzb7fetWeO0n6K0k/4+/ZpkS+3Joz8I4VcvY9OiEBFRYISqaI2fq5sCiPtAjRDOzVYG8m+Q==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"@octokit/auth-token": "^6.0.0",
"@octokit/graphql": "^9.0.2",
@@ -2435,6 +2436,7 @@
"resolved": "https://registry.npmjs.org/@opentelemetry/api/-/api-1.9.0.tgz",
"integrity": "sha512-3giAOQvZiH5F9bMlMiv8+GSPMeqg0dbaeo58/0SlA9sxSqZhnUtxzX9/2FzyhS9sWQf5S0GJE0AKBrFqjpeYcg==",
"license": "Apache-2.0",
"peer": true,
"engines": {
"node": ">=8.0.0"
}
@@ -2468,6 +2470,7 @@
"resolved": "https://registry.npmjs.org/@opentelemetry/core/-/core-2.0.1.tgz",
"integrity": "sha512-MaZk9SJIDgo1peKevlbhP6+IwIiNPNmswNL4AF0WaQJLbHXjr9SrZMgS12+iqr9ToV4ZVosCcc0f8Rg67LXjxw==",
"license": "Apache-2.0",
"peer": true,
"dependencies": {
"@opentelemetry/semantic-conventions": "^1.29.0"
},
@@ -2836,6 +2839,7 @@
"resolved": "https://registry.npmjs.org/@opentelemetry/resources/-/resources-2.0.1.tgz",
"integrity": "sha512-dZOB3R6zvBwDKnHDTB4X1xtMArB/d324VsbiPkX/Yu0Q8T2xceRthoIVFhJdvgVM2QhGVUyX9tzwiNxGtoBJUw==",
"license": "Apache-2.0",
"peer": true,
"dependencies": {
"@opentelemetry/core": "2.0.1",
"@opentelemetry/semantic-conventions": "^1.29.0"
@@ -2869,6 +2873,7 @@
"resolved": "https://registry.npmjs.org/@opentelemetry/sdk-metrics/-/sdk-metrics-2.0.1.tgz",
"integrity": "sha512-wf8OaJoSnujMAHWR3g+/hGvNcsC16rf9s1So4JlMiFaFHiE4HpIA3oUh+uWZQ7CNuK8gVW/pQSkgoa5HkkOl0g==",
"license": "Apache-2.0",
"peer": true,
"dependencies": {
"@opentelemetry/core": "2.0.1",
"@opentelemetry/resources": "2.0.1"
@@ -2921,6 +2926,7 @@
"resolved": "https://registry.npmjs.org/@opentelemetry/sdk-trace-base/-/sdk-trace-base-2.0.1.tgz",
"integrity": "sha512-xYLlvk/xdScGx1aEqvxLwf6sXQLXCjk3/1SQT9X9AoN5rXRhkdvIFShuNNmtTEPRBqcsMbS4p/gJLNI2wXaDuQ==",
"license": "Apache-2.0",
"peer": true,
"dependencies": {
"@opentelemetry/core": "2.0.1",
"@opentelemetry/resources": "2.0.1",
@@ -4136,6 +4142,7 @@
"integrity": "sha512-6mDvHUFSjyT2B2yeNx2nUgMxh9LtOWvkhIU3uePn2I2oyNymUAX1NIsdgviM4CH+JSrp2D2hsMvJOkxY+0wNRA==",
"devOptional": true,
"license": "MIT",
"peer": true,
"dependencies": {
"csstype": "^3.0.2"
}
@@ -4430,6 +4437,7 @@
"integrity": "sha512-6sMvZePQrnZH2/cJkwRpkT7DxoAWh+g6+GFRK6bV3YQo7ogi3SX5rgF6099r5Q53Ma5qeT7LGmOmuIutF4t3lA==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"@typescript-eslint/scope-manager": "8.35.0",
"@typescript-eslint/types": "8.35.0",
@@ -5422,6 +5430,7 @@
"resolved": "https://registry.npmjs.org/acorn/-/acorn-8.15.0.tgz",
"integrity": "sha512-NZyJarBfL7nWwIq+FDL6Zp/yHEhePMNnnJ0y3qfieCrmNvYct8uvtiV41UvlSe6apAfk0fY1FbWx+NwfmpvtTg==",
"license": "MIT",
"peer": true,
"bin": {
"acorn": "bin/acorn"
},
@@ -8431,6 +8440,7 @@
"integrity": "sha512-GsGizj2Y1rCWDu6XoEekL3RLilp0voSePurjZIkxL3wlm5o5EC9VpgaP7lrCvjnkuLvzFBQWB3vWB3K5KQTveQ==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"@eslint-community/eslint-utils": "^4.2.0",
"@eslint-community/regexpp": "^4.12.1",
@@ -8971,6 +8981,7 @@
"resolved": "https://registry.npmjs.org/express/-/express-5.2.1.tgz",
"integrity": "sha512-hIS4idWWai69NezIdRt2xFVofaF4j+6INOpJlVOLDO8zXGpUVEVzIYk12UUi2JzjEzWL3IOAxcTubgz9Po0yXw==",
"license": "MIT",
"peer": true,
"dependencies": {
"accepts": "^2.0.0",
"body-parser": "^2.2.1",
@@ -10584,6 +10595,7 @@
"resolved": "https://registry.npmjs.org/@jrichman/ink/-/ink-6.4.8.tgz",
"integrity": "sha512-v0thcXIKl9hqF/1w4HqA6MKxIcMoWSP3YtEZIAA+eeJngXpN5lGnMkb6rllB7FnOdwyEyYaFTcu1ZVr4/JZpWQ==",
"license": "MIT",
"peer": true,
"dependencies": {
"@alcalzone/ansi-tokenize": "^0.2.1",
"ansi-escapes": "^7.0.0",
@@ -14368,6 +14380,7 @@
"resolved": "https://registry.npmjs.org/react/-/react-19.2.0.tgz",
"integrity": "sha512-tmbWg6W31tQLeB5cdIBOicJDJRR2KzXsV7uSK9iNfLWQ5bIZfxuPEHp7M8wiHyHnn0DD1i7w3Zmin0FtkrwoCQ==",
"license": "MIT",
"peer": true,
"engines": {
"node": ">=0.10.0"
}
@@ -14378,6 +14391,7 @@
"integrity": "sha512-ePrwPfxAnB+7hgnEr8vpKxL9cmnp7F322t8oqcPshbIQQhDKgFDW4tjhF2wjVbdXF9O/nyuy3sQWd9JGpiLPvA==",
"devOptional": true,
"license": "MIT",
"peer": true,
"dependencies": {
"shell-quote": "^1.6.1",
"ws": "^7"
@@ -16614,6 +16628,7 @@
"resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz",
"integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==",
"license": "MIT",
"peer": true,
"engines": {
"node": ">=12"
},
@@ -16837,7 +16852,8 @@
"resolved": "https://registry.npmjs.org/tslib/-/tslib-2.8.1.tgz",
"integrity": "sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==",
"dev": true,
"license": "0BSD"
"license": "0BSD",
"peer": true
},
"node_modules/tsx": {
"version": "4.20.3",
@@ -16845,6 +16861,7 @@
"integrity": "sha512-qjbnuR9Tr+FJOMBqJCW5ehvIo/buZq7vH7qD7JziU98h6l3qGy0a/yPFjwO+y0/T7GFpNgNAvEcPPVfyT8rrPQ==",
"devOptional": true,
"license": "MIT",
"peer": true,
"dependencies": {
"esbuild": "~0.25.0",
"get-tsconfig": "^4.7.5"
@@ -17017,6 +17034,7 @@
"integrity": "sha512-p1diW6TqL9L07nNxvRMM7hMMw4c5XOo/1ibL4aAIGmSAt9slTE1Xgw5KWuof2uTOvCg9BY7ZRi+GaF+7sfgPeQ==",
"devOptional": true,
"license": "Apache-2.0",
"peer": true,
"bin": {
"tsc": "bin/tsc",
"tsserver": "bin/tsserver"
@@ -17224,6 +17242,7 @@
"resolved": "https://registry.npmjs.org/vite/-/vite-7.2.2.tgz",
"integrity": "sha512-BxAKBWmIbrDgrokdGZH1IgkIk/5mMHDreLDmCJ0qpyJaAteP8NvMhkwr/ZCQNqNH97bw/dANTE9PDzqwJghfMQ==",
"license": "MIT",
"peer": true,
"dependencies": {
"esbuild": "^0.25.0",
"fdir": "^6.5.0",
@@ -17337,6 +17356,7 @@
"resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz",
"integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==",
"license": "MIT",
"peer": true,
"engines": {
"node": ">=12"
},
@@ -17349,6 +17369,7 @@
"resolved": "https://registry.npmjs.org/vitest/-/vitest-3.2.4.tgz",
"integrity": "sha512-LUCP5ev3GURDysTWiP47wRRUpLKMOfPh+yKTx3kVIEiu5KOMeqzpnYNsKyOoVrULivR8tLcks4+lga33Whn90A==",
"license": "MIT",
"peer": true,
"dependencies": {
"@types/chai": "^5.2.2",
"@vitest/expect": "3.2.4",
@@ -18053,6 +18074,7 @@
"resolved": "https://registry.npmjs.org/zod/-/zod-3.25.76.tgz",
"integrity": "sha512-gzUt/qt81nXsFGKIFcC3YnfEAx5NkunCfnDlvuBSSFS02bcXu4Lmea0AFIUwbLWxWPx3d9p8S5QoaujKcNQxcQ==",
"license": "MIT",
"peer": true,
"funding": {
"url": "https://github.com/sponsors/colinhacks"
}
@@ -18075,6 +18097,7 @@
"@google/gemini-cli-core": "file:../core",
"express": "^5.1.0",
"fs-extra": "^11.3.0",
"google-auth-library": "^9.11.0",
"tar": "^7.5.2",
"uuid": "^13.0.0",
"winston": "^3.17.0"
@@ -18351,6 +18374,7 @@
"resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz",
"integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==",
"license": "MIT",
"peer": true,
"engines": {
"node": ">=12"
},
+40
View File
@@ -0,0 +1,40 @@
# Pre-built production image for a2a-server
# Used with Cloud Build: npm install + build runs in step 1, then Docker copies artifacts
FROM docker.io/library/node:20-slim
RUN apt-get update && apt-get install -y --no-install-recommends \
python3 curl git jq ripgrep ca-certificates gpg apt-transport-https \
&& curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg \
| gpg --dearmor -o /usr/share/keyrings/githubcli-archive-keyring.gpg \
&& echo "deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main" \
> /etc/apt/sources.list.d/github-cli.list \
&& curl -fsSL https://packages.cloud.google.com/apt/doc/apt-key.gpg \
| gpg --dearmor -o /usr/share/keyrings/cloud.google.gpg \
&& echo "deb [signed-by=/usr/share/keyrings/cloud.google.gpg] https://packages.cloud.google.com/apt cloud-sdk main" \
> /etc/apt/sources.list.d/google-cloud-sdk.list \
&& apt-get update && apt-get install -y --no-install-recommends gh google-cloud-cli \
&& apt-get clean && rm -rf /var/lib/apt/lists/*
WORKDIR /app
# Copy everything including pre-installed node_modules and pre-built dist
COPY package.json package-lock.json ./
COPY node_modules/ node_modules/
COPY packages/core/ packages/core/
COPY packages/a2a-server/ packages/a2a-server/
# Create workspace directory for agent operations
RUN mkdir -p /workspace && chown -R node:node /workspace
USER node
ENV CODER_AGENT_WORKSPACE_PATH=/workspace
ENV CODER_AGENT_PORT=8080
ENV NODE_ENV=production
# Prevent git from prompting for credentials interactively — fails fast instead of hanging
ENV GIT_TERMINAL_PROMPT=0
ENV CODER_AGENT_HOST=0.0.0.0
EXPOSE 8080
CMD ["node", "packages/a2a-server/dist/src/http/server.js"]
@@ -0,0 +1,23 @@
# Standalone Google Chat bridge server.
# Connects to the A2A agent server over HTTP — no agent dependencies needed.
FROM docker.io/library/node:20-slim
RUN apt-get update && apt-get install -y --no-install-recommends \
ca-certificates \
&& apt-get clean && rm -rf /var/lib/apt/lists/*
WORKDIR /app
# Copy pre-installed node_modules and pre-built dist
COPY package.json package-lock.json ./
COPY node_modules/ node_modules/
COPY packages/a2a-server/ packages/a2a-server/
USER node
ENV PORT=8080
ENV NODE_ENV=production
EXPOSE 8080
CMD ["node", "packages/a2a-server/dist/src/chat-bridge/server.js"]
+353 -2
View File
@@ -1,5 +1,356 @@
# Gemini CLI A2A Server
## All code in this package is experimental and under active development
> **Experimental** - This package is under active development.
This package contains the A2A server implementation for the Gemini CLI.
An [A2A (Agent-to-Agent)](https://google.github.io/A2A/) server that wraps the
Gemini CLI agent, enabling remote interaction via the A2A protocol. Includes a
Google Chat bridge for using the agent directly from Google Chat.
## Architecture
```
Google Chat ──webhook──> Chat Bridge ──A2A──> A2A Server ──> Gemini CLI Agent
└── Chat REST API (push responses back to Chat)
```
This package contains two independently deployable services:
1. **A2A Server** (`src/http/server.ts`) - Standard A2A protocol endpoint
(JSON-RPC + SSE streaming) that wraps the Gemini CLI agent. Heavy workload —
deploy with `concurrency=1`.
2. **Chat Bridge** (`src/chat-bridge/server.ts`) - Lightweight proxy that
translates Google Chat webhooks into A2A protocol calls. Connects to the A2A
server over HTTP. Deploy with high concurrency (`concurrency=80`).
The Chat Bridge responds immediately to webhooks with "Processing..." (avoiding
Google Chat's 30s timeout), then streams results from the A2A agent and pushes
them to Chat via the REST API.
## Prerequisites
- **GCP project** with the following APIs enabled:
- Cloud Run API
- Cloud Build API
- Artifact Registry API
- Google Chat API
- Cloud Storage API (for session persistence)
- **gcloud CLI** authenticated with your project
- **Node.js 20+** for local development
- **Gemini API key** from [Google AI Studio](https://aistudio.google.com/)
## Environment Variables
### A2A Server
| Variable | Required | Description |
| ---------------------------- | -------- | -------------------------------------------------------------- |
| `GEMINI_API_KEY` | Yes | Gemini API key for the agent |
| `CODER_AGENT_PORT` | No | Server port (default: `8080`) |
| `CODER_AGENT_HOST` | No | Bind host (default: `localhost`, set `0.0.0.0` for containers) |
| `CODER_AGENT_WORKSPACE_PATH` | No | Agent workspace directory (default: `/workspace`) |
| `GCS_BUCKET_NAME` | No | GCS bucket for task persistence |
| `GEMINI_YOLO_MODE` | No | Set `true` to auto-approve all tool calls |
| `GIT_TERMINAL_PROMPT` | No | Set `0` to prevent git credential prompts in headless env |
### Chat Bridge
| Variable | Required | Description |
| --------------------- | -------- | -------------------------------------------------------------- |
| `A2A_SERVER_URL` | Yes | URL of the A2A agent server (e.g. `http://localhost:8080`) |
| `PORT` | No | Server port (default: `8080`) |
| `CHAT_PROJECT_NUMBER` | No | Google Chat project number for JWT verification |
| `CHAT_SA_KEY_PATH` | No | Path to service account key for Chat API (uses ADC if not set) |
| `GCS_BUCKET_NAME` | No | GCS bucket for session persistence |
| `CHAT_BRIDGE_DEBUG` | No | Set `true` for verbose bridge logging |
## Local Development
### Build
From the repo root:
```bash
npm install
npm run build
```
### Run the A2A Server
```bash
export GEMINI_API_KEY="your-api-key"
export CODER_AGENT_PORT=8080
node packages/a2a-server/dist/src/http/server.js
```
### Run the Chat Bridge (separate terminal)
```bash
export A2A_SERVER_URL=http://localhost:8080
export PORT=8090
node packages/a2a-server/dist/src/chat-bridge/server.js
```
### Test the A2A endpoint
```bash
# Check the agent card
curl http://localhost:8080/.well-known/agent-card.json | jq .
# Send a message (JSON-RPC)
curl -X POST http://localhost:8080 \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "message/send",
"params": {
"message": {
"kind": "message",
"role": "user",
"messageId": "test-1",
"parts": [{"kind": "text", "text": "Hello, what can you do?"}]
},
"configuration": {"blocking": true}
}
}'
```
### Test the Chat Bridge
```bash
# Health check
curl http://localhost:8080/chat/health | jq .
# Simulate a Google Chat MESSAGE event
curl -X POST http://localhost:8080/chat/webhook \
-H "Content-Type: application/json" \
-d '{
"type": "MESSAGE",
"eventTime": "2026-01-01T00:00:00Z",
"message": {
"name": "spaces/test/messages/1",
"text": "Hello agent",
"thread": {"name": "spaces/test/threads/abc"},
"sender": {"name": "users/1", "displayName": "Test User"},
"space": {"name": "spaces/test", "type": "DM"}
},
"space": {"name": "spaces/test", "type": "DM"},
"user": {"name": "users/1", "displayName": "Test User"}
}'
```
## Cloud Run Deployment
### 1. Create Artifact Registry repository
```bash
export PROJECT_ID=your-project-id
export REGION=us-central1
gcloud artifacts repositories create gemini-a2a \
--repository-format=docker \
--location=$REGION \
--project=$PROJECT_ID
```
### 2. Create GCS bucket (optional, for session persistence)
```bash
gsutil mb -l $REGION gs://gemini-a2a-sessions-$PROJECT_ID
```
### 3. Build both images
```bash
# Build A2A agent server
gcloud builds submit \
--config=packages/a2a-server/cloudbuild.yaml \
--project=$PROJECT_ID
# Build Chat bridge
gcloud builds submit \
--config=packages/a2a-server/cloudbuild-chat-bridge.yaml \
--project=$PROJECT_ID
```
### 4. Deploy A2A agent server
```bash
export AGENT_IMAGE=us-central1-docker.pkg.dev/$PROJECT_ID/gemini-a2a/a2a-server:latest
gcloud run deploy gemini-a2a-server \
--image=$AGENT_IMAGE \
--region=$REGION \
--project=$PROJECT_ID \
--platform=managed \
--allow-unauthenticated \
--memory=2Gi \
--cpu=2 \
--timeout=300 \
--concurrency=1 \
--max-instances=1 \
--set-env-vars="GEMINI_YOLO_MODE=true,GCS_BUCKET_NAME=gemini-a2a-sessions-$PROJECT_ID" \
--set-secrets="GEMINI_API_KEY=gemini-api-key:latest"
```
### 5. Deploy Chat bridge
Get the A2A server URL first:
```bash
export A2A_URL=$(gcloud run services describe gemini-a2a-server \
--region=$REGION --project=$PROJECT_ID --format='value(status.url)')
export BRIDGE_IMAGE=us-central1-docker.pkg.dev/$PROJECT_ID/gemini-a2a/chat-bridge:latest
gcloud run deploy gemini-chat-bridge \
--image=$BRIDGE_IMAGE \
--region=$REGION \
--project=$PROJECT_ID \
--platform=managed \
--allow-unauthenticated \
--memory=512Mi \
--cpu=1 \
--timeout=60 \
--concurrency=80 \
--max-instances=1 \
--set-env-vars="A2A_SERVER_URL=$A2A_URL,GCS_BUCKET_NAME=gemini-a2a-sessions-$PROJECT_ID"
```
> **Important**: After initial deployment, always use `--update-env-vars`
> instead of `--set-env-vars` to avoid wiping existing environment variables.
### 6. Update an existing deployment
```bash
# Update env vars without replacing existing ones
gcloud run services update gemini-a2a-server \
--region=$REGION \
--project=$PROJECT_ID \
--update-env-vars="NEW_VAR=value"
# Deploy a new image
gcloud run services update gemini-a2a-server \
--region=$REGION \
--project=$PROJECT_ID \
--image=$IMAGE
```
## Google Chat App Configuration
### 1. Create a service account for Chat API
The Chat bridge needs a service account with the Chat API scope to push messages
proactively.
```bash
# Create service account
gcloud iam service-accounts create gemini-chat-bot \
--display-name="Gemini Chat Bot" \
--project=$PROJECT_ID
# Download key (for local dev)
gcloud iam service-accounts keys create chat-sa-key.json \
--iam-account=gemini-chat-bot@$PROJECT_ID.iam.gserviceaccount.com
```
On Cloud Run, use Application Default Credentials (ADC) instead of a key file.
Grant the Cloud Run service account the `chat.bot` scope by configuring it as
the Chat app's service account.
### 2. Configure the Google Chat app
1. Go to
[Google Cloud Console > APIs & Services > Google Chat API > Configuration](https://console.cloud.google.com/apis/api/chat.googleapis.com/hangouts-chat)
2. Set **App name** and **Description**
3. Under **Connection settings**, select **HTTP endpoint URL**
4. Set the URL to your **Chat bridge** Cloud Run service URL + `/chat/webhook`:
```
https://gemini-chat-bridge-HASH-uc.a.run.app/chat/webhook
```
5. Under **Authentication Audience**, select **HTTP endpoint URL**
6. Under **Visibility**, choose who can use the app
7. Under **Permissions**, configure who can install it
8. Click **Save**
### 3. Grant Cloud Run invoker permission
If your Cloud Run service requires authentication (recommended):
```bash
# Get the Chat service account
# It's usually chat@system.gserviceaccount.com
gcloud run services add-iam-policy-binding gemini-chat-bridge \
--region=$REGION \
--project=$PROJECT_ID \
--member="serviceAccount:chat@system.gserviceaccount.com" \
--role="roles/run.invoker"
```
## Chat Bridge Commands
When messaging the bot in Google Chat:
| Command | Description |
| ----------------------- | --------------------------------------------------- |
| `/reset` or `reset` | Clear the current session and start fresh |
| `/yolo` | Enable YOLO mode - auto-approve all tool calls |
| `/safe` | Disable YOLO mode - require approval for tool calls |
| `approve` / `yes` / `y` | Approve a pending tool call |
| `reject` / `no` / `n` | Reject a pending tool call |
| `always allow` | Approve and always allow this tool |
## Troubleshooting
### "Gemini CLI Agent is not responding" in Google Chat
This usually means the bridge couldn't return "Processing..." within Google
Chat's 30-second timeout. Check Cloud Run logs for both services:
```bash
# Chat bridge logs
gcloud run services logs read gemini-chat-bridge \
--region=$REGION --project=$PROJECT_ID --limit=50
# A2A agent logs
gcloud run services logs read gemini-a2a-server \
--region=$REGION --project=$PROJECT_ID --limit=50
```
### Tool approvals appearing in YOLO mode
Ensure `GEMINI_YOLO_MODE=true` is set. If you used `--set-env-vars` during a
deployment, it may have wiped this variable. Use `--update-env-vars` instead.
### Agent hangs on git operations
The `GIT_TERMINAL_PROMPT=0` env var (set in the Dockerfile) prevents git from
prompting for credentials. If git operations require authentication, configure a
credential helper or use `gh auth` with a token.
### Session state lost after restart
Enable GCS persistence by setting `GCS_BUCKET_NAME`. Sessions are automatically
flushed to GCS every 30 seconds and restored on startup.
### Chat responses appear as top-level messages instead of thread replies
The Chat bridge includes `thread.name` in all responses. If replies still appear
at the top level, ensure the webhook event includes thread information. DM
conversations always thread correctly; spaces may need threading enabled.
## Known Limitations
- **Google Chat 4096 character limit**: Long agent responses are automatically
split into multiple messages at paragraph/line boundaries.
- **Single bridge instance**: The bridge uses `max-instances=1` so that the
in-memory async processing guard works correctly. This means no redundancy
during deploys (brief downtime during revision rollover).
- **Tool confirmation in streaming mode**: When the A2A server has
`GEMINI_YOLO_MODE=false`, tool confirmations via streaming may not return text
due to an SDK-level issue (executor aborts on SSE disconnect). Server YOLO
mode works correctly.
@@ -0,0 +1,35 @@
steps:
# Step 1: Install all dependencies and build
- name: 'node:20-slim'
entrypoint: 'bash'
args:
- '-c'
- |
apt-get update && apt-get install -y python3 make g++ git
npm pkg delete scripts.prepare
npm install
npm run build
env:
- 'HUSKY=0'
# Step 2: Build Docker image for Chat bridge
- name: 'gcr.io/cloud-builders/docker'
args:
- 'build'
- '-t'
- 'us-central1-docker.pkg.dev/$PROJECT_ID/gemini-a2a/chat-bridge:latest'
- '-f'
- 'packages/a2a-server/Dockerfile.chat-bridge'
- '.'
# Step 3: Push to Artifact Registry
- name: 'gcr.io/cloud-builders/docker'
args:
- 'push'
- 'us-central1-docker.pkg.dev/$PROJECT_ID/gemini-a2a/chat-bridge:latest'
images:
- 'us-central1-docker.pkg.dev/$PROJECT_ID/gemini-a2a/chat-bridge:latest'
timeout: '1800s'
options:
machineType: 'E2_HIGHCPU_8'
+35
View File
@@ -0,0 +1,35 @@
steps:
# Step 1: Install all dependencies and build
- name: 'node:20-slim'
entrypoint: 'bash'
args:
- '-c'
- |
apt-get update && apt-get install -y python3 make g++ git
npm pkg delete scripts.prepare
npm install
npm run build
env:
- 'HUSKY=0'
# Step 2: Build Docker image (using pre-built dist/ from step 1)
- name: 'gcr.io/cloud-builders/docker'
args:
- 'build'
- '-t'
- 'us-central1-docker.pkg.dev/$PROJECT_ID/gemini-a2a/a2a-server:latest'
- '-f'
- 'packages/a2a-server/Dockerfile'
- '.'
# Step 3: Push to Artifact Registry
- name: 'gcr.io/cloud-builders/docker'
args:
- 'push'
- 'us-central1-docker.pkg.dev/$PROJECT_ID/gemini-a2a/a2a-server:latest'
images:
- 'us-central1-docker.pkg.dev/$PROJECT_ID/gemini-a2a/a2a-server:latest'
timeout: '1800s'
options:
machineType: 'E2_HIGHCPU_8'
+74
View File
@@ -0,0 +1,74 @@
apiVersion: apps/v1
kind: Deployment
metadata:
name: gemini-a2a-server
labels:
app: gemini-a2a-server
spec:
replicas: 1
selector:
matchLabels:
app: gemini-a2a-server
template:
metadata:
labels:
app: gemini-a2a-server
spec:
containers:
- name: a2a-server
image: us-central1-docker.pkg.dev/adamfweidman-test/gemini-a2a/a2a-server:latest
ports:
- containerPort: 8080
protocol: TCP
env:
- name: CODER_AGENT_PORT
value: "8080"
- name: CODER_AGENT_HOST
value: "0.0.0.0"
- name: CODER_AGENT_WORKSPACE_PATH
value: "/workspace"
- name: GEMINI_API_KEY
valueFrom:
secretKeyRef:
name: gemini-secrets
key: api-key
- name: GEMINI_YOLO_MODE
value: "true"
- name: CHAT_BRIDGE_A2A_URL
value: "http://localhost:8080"
- name: NODE_ENV
value: "production"
resources:
requests:
cpu: "500m"
memory: "512Mi"
limits:
cpu: "2000m"
memory: "2Gi"
readinessProbe:
httpGet:
path: /.well-known/agent-card.json
port: 8080
initialDelaySeconds: 10
periodSeconds: 10
livenessProbe:
httpGet:
path: /.well-known/agent-card.json
port: 8080
initialDelaySeconds: 15
periodSeconds: 30
---
apiVersion: v1
kind: Service
metadata:
name: gemini-a2a-server
labels:
app: gemini-a2a-server
spec:
type: ClusterIP
selector:
app: gemini-a2a-server
ports:
- port: 80
targetPort: 8080
protocol: TCP
+1
View File
@@ -29,6 +29,7 @@
"@google-cloud/storage": "^7.16.0",
"@google/gemini-cli-core": "file:../core",
"express": "^5.1.0",
"google-auth-library": "^9.11.0",
"fs-extra": "^11.3.0",
"tar": "^7.5.2",
"uuid": "^13.0.0",
@@ -0,0 +1,157 @@
/**
* @license
* Copyright 2026 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
/**
* Builder functions for A2UI standard catalog components.
* These create the component objects that go into updateComponents messages.
*/
import type { A2UIComponent } from './a2ui-extension.js';
// Layout components
export function column(
id: string,
children: string[],
opts?: { align?: string; justify?: string; weight?: number },
): A2UIComponent {
return {
id,
component: 'Column',
children,
...opts,
};
}
export function row(
id: string,
children: string[],
opts?: { align?: string; justify?: string },
): A2UIComponent {
return {
id,
component: 'Row',
children,
...opts,
};
}
export function card(
id: string,
child: string,
opts?: Record<string, unknown>,
): A2UIComponent {
return {
id,
component: 'Card',
child,
...opts,
};
}
// Content components
export function text(
id: string,
textContent: string | { path: string },
opts?: { variant?: string },
): A2UIComponent {
return {
id,
component: 'Text',
text: textContent,
...opts,
};
}
export function icon(id: string, name: string): A2UIComponent {
return {
id,
component: 'Icon',
name,
};
}
export function divider(
id: string,
axis: 'horizontal' | 'vertical' = 'horizontal',
): A2UIComponent {
return {
id,
component: 'Divider',
axis,
};
}
// Interactive components
export function button(
id: string,
child: string,
action: {
event?: { name: string; context: Record<string, unknown> };
functionCall?: { call: string; args: Record<string, unknown> };
},
opts?: { variant?: 'primary' | 'borderless' },
): A2UIComponent {
return {
id,
component: 'Button',
child,
action,
...opts,
};
}
export function textField(
id: string,
label: string,
valuePath: string,
opts?: {
variant?: 'shortText' | 'longText';
checks?: Array<{
call: string;
args: Record<string, unknown>;
message: string;
}>;
},
): A2UIComponent {
return {
id,
component: 'TextField',
label,
value: { path: valuePath },
...opts,
};
}
export function checkBox(
id: string,
label: string,
valuePath: string,
): A2UIComponent {
return {
id,
component: 'CheckBox',
label,
value: { path: valuePath },
};
}
export function choicePicker(
id: string,
options: Array<{ label: string; value: string }>,
valuePath: string,
opts?: { variant?: 'mutuallyExclusive' | 'multiSelect' },
): A2UIComponent {
return {
id,
component: 'ChoicePicker',
options,
value: { path: valuePath },
...opts,
};
}
@@ -0,0 +1,194 @@
/**
* @license
* Copyright 2026 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
/**
* A2UI (Agent-to-UI) Extension for A2A protocol.
* Implements the A2UI v0.10 specification for generating declarative UI
* messages that clients can render natively.
*
* @see https://a2ui.org/specification/v0_10/docs/a2ui_protocol.md
* @see https://a2ui.org/specification/v0_10/docs/a2ui_extension_specification.md
*/
import type { Part } from '@a2a-js/sdk';
// Extension constants
export const A2UI_EXTENSION_URI = 'https://a2ui.org/a2a-extension/a2ui/v0.10';
export const A2UI_MIME_TYPE = 'application/json+a2ui';
export const A2UI_VERSION = 'v0.10';
export const STANDARD_CATALOG_ID =
'https://a2ui.org/specification/v0_10/standard_catalog.json';
// Metadata keys
export const MIME_TYPE_KEY = 'mimeType';
export const A2UI_CLIENT_CAPABILITIES_KEY = 'a2uiClientCapabilities';
export const A2UI_CLIENT_DATA_MODEL_KEY = 'a2uiClientDataModel';
/**
* A2UI message types (server-to-client).
*/
export interface CreateSurfaceMessage {
version: typeof A2UI_VERSION;
createSurface: {
surfaceId: string;
catalogId: string;
theme?: Record<string, unknown>;
sendDataModel?: boolean;
};
}
export interface UpdateComponentsMessage {
version: typeof A2UI_VERSION;
updateComponents: {
surfaceId: string;
components: A2UIComponent[];
};
}
export interface UpdateDataModelMessage {
version: typeof A2UI_VERSION;
updateDataModel: {
surfaceId: string;
path?: string;
value?: unknown;
};
}
export interface DeleteSurfaceMessage {
version: typeof A2UI_VERSION;
deleteSurface: {
surfaceId: string;
};
}
export type A2UIServerMessage =
| CreateSurfaceMessage
| UpdateComponentsMessage
| UpdateDataModelMessage
| DeleteSurfaceMessage;
/**
* A2UI component definition.
*/
export interface A2UIComponent {
id: string;
component: string;
[key: string]: unknown;
}
/**
* A2UI client-to-server action message.
*/
export interface A2UIActionMessage {
version: typeof A2UI_VERSION;
action: {
name: string;
surfaceId: string;
sourceComponentId: string;
timestamp: string;
context: Record<string, unknown>;
};
}
/**
* A2UI client capabilities sent in metadata.
*/
export interface A2UIClientCapabilities {
supportedCatalogIds: string[];
inlineCatalogs?: unknown[];
}
/**
* Creates an A2A DataPart containing A2UI messages.
* Per the spec, the data field contains an ARRAY of A2UI messages.
*/
export function createA2UIPart(messages: A2UIServerMessage[]): Part {
return {
kind: 'data',
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
data: messages as unknown as Record<string, unknown>,
metadata: {
[MIME_TYPE_KEY]: A2UI_MIME_TYPE,
},
} as Part;
}
/**
* Creates a single A2A DataPart from one A2UI message.
*/
export function createA2UISinglePart(message: A2UIServerMessage): Part {
return createA2UIPart([message]);
}
/**
* Checks if an A2A Part contains A2UI data.
*/
export function isA2UIPart(part: Part): boolean {
return (
part.kind === 'data' &&
part.metadata != null &&
part.metadata[MIME_TYPE_KEY] === A2UI_MIME_TYPE
);
}
/**
* Extracts A2UI action messages from an A2A Part.
*/
export function extractA2UIActions(part: Part): A2UIActionMessage[] {
if (!isA2UIPart(part)) return [];
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
const data = (part as unknown as { data?: unknown[] }).data;
if (!Array.isArray(data)) return [];
return data.filter(
(msg): msg is A2UIActionMessage =>
typeof msg === 'object' &&
msg !== null &&
'action' in msg &&
'version' in msg,
);
}
/**
* Creates the A2UI AgentExtension configuration for the AgentCard.
*/
export function getA2UIAgentExtension(
supportedCatalogIds: string[] = [STANDARD_CATALOG_ID],
acceptsInlineCatalogs = false,
): {
uri: string;
description: string;
required: boolean;
params: Record<string, unknown>;
} {
const params: Record<string, unknown> = {};
if (supportedCatalogIds.length > 0) {
params['supportedCatalogIds'] = supportedCatalogIds;
}
if (acceptsInlineCatalogs) {
params['acceptsInlineCatalogs'] = true;
}
return {
uri: A2UI_EXTENSION_URI,
description: 'Provides agent driven UI using the A2UI JSON format.',
required: false,
params,
};
}
/**
* Checks if the A2UI extension was requested via extension headers or message.
*/
export function isA2UIRequested(
requestedExtensions?: string[],
messageExtensions?: string[],
): boolean {
return (
(requestedExtensions?.includes(A2UI_EXTENSION_URI) ?? false) ||
(messageExtensions?.includes(A2UI_EXTENSION_URI) ?? false)
);
}
@@ -0,0 +1,468 @@
/**
* @license
* Copyright 2026 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
/**
* Manages A2UI surfaces for the Gemini CLI A2A server.
* Creates and updates surfaces for:
* - Tool call approval UIs
* - Agent text/thought streaming displays
* - Task status indicators
*/
import type { Part } from '@a2a-js/sdk';
import { logger } from '../utils/logger.js';
import {
A2UI_VERSION,
STANDARD_CATALOG_ID,
createA2UIPart,
type A2UIServerMessage,
type A2UIComponent,
} from './a2ui-extension.js';
import {
column,
row,
text,
button,
card,
icon,
divider,
} from './a2ui-components.js';
/**
* Generates A2UI parts for tool call approval surfaces.
*/
export function createToolCallApprovalSurface(
taskId: string,
toolCall: {
callId: string;
name: string;
displayName?: string;
description?: string;
args?: Record<string, unknown>;
kind?: string;
},
): Part {
const surfaceId = `tool_approval_${taskId}_${toolCall.callId}`;
const toolDisplayName = toolCall.displayName || toolCall.name;
const argsPreview = toolCall.args
? JSON.stringify(toolCall.args, null, 2).substring(0, 500)
: 'No arguments';
logger.info(
`[A2UI] Creating tool approval surface: ${surfaceId} for tool: ${toolDisplayName}`,
);
const messages: A2UIServerMessage[] = [
// 1. Create the surface
{
version: A2UI_VERSION,
createSurface: {
surfaceId,
catalogId: STANDARD_CATALOG_ID,
theme: {
primaryColor: '#1a73e8',
agentDisplayName: 'Gemini CLI Agent',
},
sendDataModel: true,
},
},
// 2. Define the components
{
version: A2UI_VERSION,
updateComponents: {
surfaceId,
components: buildToolApprovalComponents(
taskId,
toolCall.callId,
toolDisplayName,
toolCall.description || '',
argsPreview,
toolCall.kind || 'tool',
),
},
},
// 3. Populate the data model
{
version: A2UI_VERSION,
updateDataModel: {
surfaceId,
value: {
tool: {
callId: toolCall.callId,
name: toolCall.name,
displayName: toolDisplayName,
description: toolCall.description || '',
args: argsPreview,
kind: toolCall.kind || 'tool',
status: 'awaiting_approval',
},
taskId,
},
},
},
];
return createA2UIPart(messages);
}
function buildToolApprovalComponents(
taskId: string,
callId: string,
toolName: string,
description: string,
argsPreview: string,
kind: string,
): A2UIComponent[] {
return [
// Root card
card('root', 'main_column'),
// Main vertical layout
column(
'main_column',
[
'header_row',
'description_text',
'divider_1',
'args_label',
'args_text',
'divider_2',
'action_row',
],
{ align: 'stretch' },
),
// Header with icon and tool name
row('header_row', ['tool_icon', 'tool_name_text'], {
align: 'center',
}),
icon('tool_icon', kind === 'shell' ? 'terminal' : 'build'),
text('tool_name_text', `**${toolName}** requires approval`, {
variant: 'h3',
}),
// Description
text(
'description_text',
description || 'This tool needs your permission to execute.',
),
divider('divider_1'),
// Arguments preview
text('args_label', '**Arguments:**', { variant: 'caption' }),
text('args_text', `\`\`\`\n${argsPreview}\n\`\`\``),
divider('divider_2'),
// Action buttons row
row(
'action_row',
['approve_button', 'approve_always_button', 'reject_button'],
{ justify: 'spaceBetween' },
),
// Approve button
text('approve_label', 'Approve'),
button(
'approve_button',
'approve_label',
{
event: {
name: 'tool_confirmation',
context: {
taskId,
callId,
outcome: 'proceed_once',
},
},
},
{ variant: 'primary' },
),
// Approve always button
text('approve_always_label', 'Always Allow'),
button('approve_always_button', 'approve_always_label', {
event: {
name: 'tool_confirmation',
context: {
taskId,
callId,
outcome: 'proceed_always_tool',
},
},
}),
// Reject button
text('reject_label', 'Reject'),
button('reject_button', 'reject_label', {
event: {
name: 'tool_confirmation',
context: {
taskId,
callId,
outcome: 'cancel',
},
},
}),
];
}
/**
* Creates an A2UI surface update for tool execution status.
*/
export function updateToolCallStatus(
taskId: string,
callId: string,
status: string,
output?: string,
): Part {
const surfaceId = `tool_approval_${taskId}_${callId}`;
logger.info(
`[A2UI] Updating tool status surface: ${surfaceId} status: ${status}`,
);
const messages: A2UIServerMessage[] = [
{
version: A2UI_VERSION,
updateDataModel: {
surfaceId,
path: '/tool/status',
value: status,
},
},
];
// If tool completed, update the UI to show result
if (['success', 'error', 'cancelled'].includes(status)) {
messages.push({
version: A2UI_VERSION,
updateComponents: {
surfaceId,
components: [
// Replace action row with status indicator
row('action_row', ['status_icon', 'status_text'], {
align: 'center',
}),
icon(
'status_icon',
status === 'success'
? 'check_circle'
: status === 'error'
? 'error'
: 'cancel',
),
text(
'status_text',
status === 'success'
? 'Tool executed successfully'
: status === 'error'
? 'Tool execution failed'
: 'Tool execution cancelled',
),
],
},
});
if (output) {
messages.push({
version: A2UI_VERSION,
updateDataModel: {
surfaceId,
path: '/tool/output',
value: output,
},
});
}
}
return createA2UIPart(messages);
}
/**
* Creates an A2UI text content surface for agent messages.
*/
export function createTextContentPart(
taskId: string,
content: string,
surfaceId?: string,
): Part {
const sid = surfaceId || `agent_text_${taskId}`;
const messages: A2UIServerMessage[] = [
{
version: A2UI_VERSION,
updateDataModel: {
surfaceId: sid,
path: '/content/text',
value: content,
},
},
];
return createA2UIPart(messages);
}
/**
* Creates the initial agent response surface.
*/
export function createAgentResponseSurface(taskId: string): Part {
const surfaceId = `agent_response_${taskId}`;
logger.info(`[A2UI] Creating agent response surface: ${surfaceId}`);
const messages: A2UIServerMessage[] = [
{
version: A2UI_VERSION,
createSurface: {
surfaceId,
catalogId: STANDARD_CATALOG_ID,
theme: {
primaryColor: '#1a73e8',
agentDisplayName: 'Gemini CLI Agent',
},
},
},
{
version: A2UI_VERSION,
updateComponents: {
surfaceId,
components: [
card('root', 'response_column'),
column('response_column', ['response_text', 'status_text'], {
align: 'stretch',
}),
text('response_text', { path: '/response/text' }),
text(
'status_text',
{ path: '/response/status' },
{
variant: 'caption',
},
),
],
},
},
{
version: A2UI_VERSION,
updateDataModel: {
surfaceId,
value: {
response: {
text: '',
status: 'Working...',
},
},
},
},
];
return createA2UIPart(messages);
}
/**
* Updates the agent response surface with new text content.
*/
export function updateAgentResponseText(
taskId: string,
content: string,
status?: string,
): Part {
const surfaceId = `agent_response_${taskId}`;
const messages: A2UIServerMessage[] = [
{
version: A2UI_VERSION,
updateDataModel: {
surfaceId,
path: '/response/text',
value: content,
},
},
];
if (status) {
messages.push({
version: A2UI_VERSION,
updateDataModel: {
surfaceId,
path: '/response/status',
value: status,
},
});
}
return createA2UIPart(messages);
}
/**
* Creates an A2UI thought surface.
*/
export function createThoughtPart(
taskId: string,
subject: string,
description: string,
): Part {
const surfaceId = `thought_${taskId}_${Date.now()}`;
const messages: A2UIServerMessage[] = [
{
version: A2UI_VERSION,
createSurface: {
surfaceId,
catalogId: STANDARD_CATALOG_ID,
theme: {
primaryColor: '#7c4dff',
agentDisplayName: 'Gemini CLI Agent',
},
},
},
{
version: A2UI_VERSION,
updateComponents: {
surfaceId,
components: [
card('root', 'thought_column'),
column('thought_column', ['thought_icon_row', 'thought_desc'], {
align: 'stretch',
}),
row('thought_icon_row', ['thought_icon', 'thought_subject'], {
align: 'center',
}),
icon('thought_icon', 'psychology'),
text('thought_subject', `*${subject}*`, { variant: 'h4' }),
text('thought_desc', description),
],
},
},
];
return createA2UIPart(messages);
}
/**
* Deletes a tool approval surface after resolution.
*/
export function deleteToolApprovalSurface(
taskId: string,
callId: string,
): Part {
const surfaceId = `tool_approval_${taskId}_${callId}`;
logger.info(`[A2UI] Deleting tool approval surface: ${surfaceId}`);
const messages: A2UIServerMessage[] = [
{
version: A2UI_VERSION,
deleteSurface: {
surfaceId,
},
},
];
return createA2UIPart(messages);
}
+60 -1
View File
@@ -36,6 +36,10 @@ import { loadExtensions } from '../config/extension.js';
import { Task } from './task.js';
import { requestStorage } from '../http/requestStorage.js';
import { pushTaskStateFailed } from '../utils/executor_utils.js';
import {
A2UI_CLIENT_CAPABILITIES_KEY,
A2UI_EXTENSION_URI,
} from '../a2ui/a2ui-extension.js';
/**
* Provides a wrapper for Task. Passes data from Task to SDKTask.
@@ -73,6 +77,24 @@ class TaskWrapper {
artifacts: [],
};
sdkTask.metadata!['_contextId'] = this.task.contextId;
// Persist conversation history for session resumability.
// GCSTaskStore saves this as a separate object and restores it on load.
try {
const conversationHistory = this.task.geminiClient.getHistory();
if (conversationHistory.length > 0) {
sdkTask.metadata!['_conversationHistory'] = conversationHistory;
logger.info(
`Task ${this.task.id}: Persisting ${conversationHistory.length} conversation history entries.`,
);
}
} catch {
// GeminiClient may not be initialized yet
logger.warn(
`Task ${this.task.id}: Could not get conversation history for persistence.`,
);
}
return sdkTask;
}
}
@@ -127,7 +149,25 @@ export class CoderAgentExecutor implements AgentExecutor {
agentSettings.autoExecute,
);
runtimeTask.taskState = persistedState._taskState;
await runtimeTask.geminiClient.initialize();
// Restore conversation history if available from the TaskStore.
// This enables session resumability — the LLM gets full context of
// prior interactions rather than starting with a blank slate.
const conversationHistory = metadata['_conversationHistory'];
if (Array.isArray(conversationHistory) && conversationHistory.length > 0) {
logger.info(
`Task ${sdkTask.id}: Resuming with ${conversationHistory.length} conversation history entries.`,
);
// History was serialized from GeminiClient.getHistory() which returns
// Content[]. After JSON round-trip it's structurally identical.
await runtimeTask.geminiClient.initialize();
runtimeTask.geminiClient.setHistory(
conversationHistory,
);
} else {
await runtimeTask.geminiClient.initialize();
}
const wrapper = new TaskWrapper(runtimeTask, agentSettings);
this.tasks.set(sdkTask.id, wrapper);
@@ -435,6 +475,22 @@ export class CoderAgentExecutor implements AgentExecutor {
const currentTask = wrapper.task;
// Detect A2UI extension activation from the request
// Check if user message metadata contains A2UI client capabilities
// or if the extensions header includes the A2UI URI
const messageMetadata = userMessage.metadata;
const hasA2UICapabilities =
messageMetadata?.[A2UI_CLIENT_CAPABILITIES_KEY] != null;
// Also check if extension URI is referenced in message extensions
const messageExtensions = messageMetadata?.['extensions'];
const hasA2UIExtension =
Array.isArray(messageExtensions) &&
messageExtensions.includes(A2UI_EXTENSION_URI);
if (hasA2UICapabilities || hasA2UIExtension) {
currentTask.a2uiEnabled = true;
logger.info(`[CoderAgentExecutor] A2UI enabled for task ${taskId}`);
}
if (['canceled', 'failed', 'completed'].includes(currentTask.taskState)) {
logger.warn(
`[CoderAgentExecutor] Attempted to execute task ${taskId} which is already in state ${currentTask.taskState}. Ignoring.`,
@@ -552,6 +608,9 @@ export class CoderAgentExecutor implements AgentExecutor {
logger.info(
`[CoderAgentExecutor] Task ${taskId}: Agent turn finished, setting to input-required.`,
);
// Finalize A2UI surfaces before marking complete
currentTask.finalizeA2UISurfaces();
const stateChange: StateChange = {
kind: CoderAgentEvent.StateChangeEvent,
};
+201 -6
View File
@@ -56,6 +56,15 @@ import type {
Citation,
} from '../types.js';
import type { PartUnion, Part as genAiPart } from '@google/genai';
import {
createToolCallApprovalSurface,
updateToolCallStatus,
createAgentResponseSurface,
updateAgentResponseText,
createThoughtPart as createA2UIThoughtPart,
deleteToolApprovalSurface,
} from '../a2ui/a2ui-surface-manager.js';
import { isA2UIPart, extractA2UIActions } from '../a2ui/a2ui-extension.js';
type UnionKeys<T> = T extends T ? keyof T : never;
@@ -75,6 +84,11 @@ export class Task {
promptCount = 0;
autoExecute: boolean;
// A2UI support
a2uiEnabled = false;
private accumulatedText = '';
private a2uiResponseSurfaceCreated = false;
// For tool waiting logic
private pendingToolCalls: Map<string, string> = new Map(); //toolCallId --> status
private toolCompletionPromise?: Promise<void>;
@@ -391,6 +405,44 @@ export class Task {
: { kind: CoderAgentEvent.ToolCallUpdateEvent };
const message = this.toolStatusMessage(tc, this.id, this.contextId);
// Add A2UI parts for tool call updates if A2UI is enabled
if (this.a2uiEnabled) {
try {
if (tc.status === 'awaiting_approval') {
const a2uiPart = createToolCallApprovalSurface(this.id, {
callId: tc.request.callId,
name: tc.request.name,
displayName: tc.tool?.displayName || tc.tool?.name,
description: tc.tool?.description,
args: tc.request.args as Record<string, unknown> | undefined,
kind: tc.tool?.kind,
});
message.parts.push(a2uiPart);
logger.info(
`[Task] A2UI: Added tool approval surface for ${tc.request.callId}`,
);
} else if (['success', 'error', 'cancelled'].includes(tc.status)) {
const output =
'liveOutput' in tc ? String(tc.liveOutput) : undefined;
const a2uiPart = updateToolCallStatus(
this.id,
tc.request.callId,
tc.status,
output,
);
message.parts.push(a2uiPart);
logger.info(
`[Task] A2UI: Updated tool status for ${tc.request.callId}: ${tc.status}`,
);
}
} catch (a2uiError) {
logger.error(
'[Task] A2UI: Error generating tool call surface:',
a2uiError,
);
}
}
const event = this._createStatusUpdateEvent(
this.taskState,
coderAgentMessage,
@@ -954,7 +1006,66 @@ export class Task {
let anyConfirmationHandled = false;
let hasContentForLlm = false;
// Reset A2UI accumulated text for new user turn
if (this.a2uiEnabled) {
this.accumulatedText = '';
this.a2uiResponseSurfaceCreated = false;
}
for (const part of userMessage.parts) {
// Handle A2UI action messages (e.g., button clicks for tool approval)
if (this.a2uiEnabled && isA2UIPart(part)) {
const actions = extractA2UIActions(part);
for (const action of actions) {
if (action.action.name === 'tool_confirmation') {
const ctx = action.action.context;
// Convert A2UI action to a tool confirmation data part
const syntheticPart: Part = {
kind: 'data',
data: {
callId: ctx['callId'],
outcome: ctx['outcome'],
},
} as Part;
const handled =
await this._handleToolConfirmationPart(syntheticPart);
if (handled) {
anyConfirmationHandled = true;
// Emit a delete surface part for the approval UI
try {
const deletePart = deleteToolApprovalSurface(
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
(ctx['taskId'] as string) || this.id,
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
ctx['callId'] as string,
);
const deleteMessage: Message = {
kind: 'message',
role: 'agent',
parts: [deletePart],
messageId: uuidv4(),
taskId: this.id,
contextId: this.contextId,
};
const event = this._createStatusUpdateEvent(
this.taskState,
{ kind: CoderAgentEvent.ToolCallUpdateEvent },
deleteMessage,
false,
);
this.eventBus?.publish(event);
} catch (a2uiError) {
logger.error(
'[Task] A2UI: Error deleting approval surface:',
a2uiError,
);
}
}
}
}
continue;
}
const confirmationHandled = await this._handleToolConfirmationPart(part);
if (confirmationHandled) {
anyConfirmationHandled = true;
@@ -1020,6 +1131,33 @@ export class Task {
}
logger.info('[Task] Sending text content to event bus.');
const message = this._createTextMessage(content);
// Add A2UI response surface parts if A2UI is enabled
if (this.a2uiEnabled) {
try {
this.accumulatedText += content;
if (!this.a2uiResponseSurfaceCreated) {
const surfacePart = createAgentResponseSurface(this.id);
message.parts.push(surfacePart);
this.a2uiResponseSurfaceCreated = true;
logger.info(
`[Task] A2UI: Created agent response surface for task ${this.id}`,
);
}
const updatePart = updateAgentResponseText(
this.id,
this.accumulatedText,
'Working...',
);
message.parts.push(updatePart);
} catch (a2uiError) {
logger.error(
'[Task] A2UI: Error generating text content surface:',
a2uiError,
);
}
}
const textContent: TextContent = {
kind: CoderAgentEvent.TextContentEvent,
};
@@ -1041,15 +1179,35 @@ export class Task {
return;
}
logger.info('[Task] Sending thought to event bus.');
const parts: Part[] = [
{
kind: 'data',
data: content,
} as Part,
];
// Add A2UI thought surface if A2UI is enabled
if (this.a2uiEnabled) {
try {
const a2uiPart = createA2UIThoughtPart(
this.id,
content.subject || 'Thinking...',
content.description || '',
);
parts.push(a2uiPart);
logger.info(`[Task] A2UI: Added thought surface for task ${this.id}`);
} catch (a2uiError) {
logger.error(
'[Task] A2UI: Error generating thought surface:',
a2uiError,
);
}
}
const message: Message = {
kind: 'message',
role: 'agent',
parts: [
{
kind: 'data',
data: content,
} as Part,
],
parts,
messageId: uuidv4(),
taskId: this.id,
contextId: this.contextId,
@@ -1070,6 +1228,43 @@ export class Task {
);
}
/**
* Finalizes A2UI surfaces when the agent turn is complete.
* Updates the response surface status to "Done".
*/
finalizeA2UISurfaces(): void {
if (!this.a2uiEnabled || !this.a2uiResponseSurfaceCreated) {
return;
}
try {
const finalPart = updateAgentResponseText(
this.id,
this.accumulatedText,
'Done',
);
const message: Message = {
kind: 'message',
role: 'agent',
parts: [finalPart],
messageId: uuidv4(),
taskId: this.id,
contextId: this.contextId,
};
const event = this._createStatusUpdateEvent(
this.taskState,
{ kind: CoderAgentEvent.TextContentEvent },
message,
false,
);
this.eventBus?.publish(event);
logger.info(
`[Task] A2UI: Finalized response surface for task ${this.id}`,
);
} catch (a2uiError) {
logger.error('[Task] A2UI: Error finalizing surfaces:', a2uiError);
}
}
_sendCitation(citation: string) {
if (!citation || citation.trim() === '') {
return;
@@ -0,0 +1,527 @@
/**
* @license
* Copyright 2026 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
/**
* A2A client wrapper for the Google Chat bridge.
* Connects to the A2A server (local or remote) and sends/receives messages.
* Follows the patterns from core/agents/a2a-client-manager.ts and
* core/agents/remote-invocation.ts.
*/
import type {
Message,
Task,
Part,
MessageSendParams,
TaskStatusUpdateEvent,
TaskArtifactUpdateEvent,
} from '@a2a-js/sdk';
import {
type Client,
ClientFactory,
ClientFactoryOptions,
DefaultAgentCardResolver,
RestTransportFactory,
JsonRpcTransportFactory,
} from '@a2a-js/sdk/client';
import { GoogleAuth } from 'google-auth-library';
import { Agent } from 'undici';
import { v4 as uuidv4 } from 'uuid';
import { logger } from '../utils/logger.js';
/**
* Undici agent with long timeouts for SSE streaming.
* Default body/headers timeouts are ~30s which kills idle SSE connections
* when the agent runs long tools (npm install, tsc builds, etc.).
*/
const sseDispatcher = new Agent({
bodyTimeout: 10 * 60 * 1000, // 10 minutes
headersTimeout: 10 * 60 * 1000,
keepAliveTimeout: 10 * 60 * 1000,
});
// Inline A2UI constants so the chat bridge has no dependency on ../a2ui/
const A2UI_EXTENSION_URI = 'https://a2ui.org/a2a-extension/a2ui/v0.10';
const A2UI_MIME_TYPE = 'application/json+a2ui';
export type A2AResponse = Message | Task;
export type A2AStreamEventData =
| Message
| Task
| TaskStatusUpdateEvent
| TaskArtifactUpdateEvent;
/**
* Extracts contextId and taskId from an A2A response.
* Follows extractIdsFromResponse pattern from a2aUtils.ts.
*/
export function extractIdsFromResponse(result: A2AResponse): {
contextId?: string;
taskId?: string;
} {
if (result.kind === 'message') {
return {
contextId: result.contextId,
taskId: result.taskId,
};
}
if (result.kind === 'task') {
const contextId = result.contextId;
let taskId: string | undefined = result.id;
// Clear taskId on terminal states so next interaction starts a fresh task
const state = result.status?.state;
if (state === 'completed' || state === 'failed' || state === 'canceled') {
taskId = undefined;
}
return { contextId, taskId };
}
return {};
}
/**
* Extracts all parts from an A2A response.
* For Tasks, checks history (accumulated from intermediate status-update events),
* the final status message, and artifacts. The blocking DefaultRequestHandler
* accumulates intermediate events into task.history, so the A2UI response content
* from "working" events lives there even if the final status message is empty.
*/
export function extractAllParts(result: A2AResponse): Part[] {
const parts: Part[] = [];
if (result.kind === 'message') {
parts.push(...(result.parts ?? []));
} else if (result.kind === 'task') {
// Parts from task history (accumulated intermediate status-update messages)
if (result.history) {
for (const msg of result.history) {
if (msg.parts) {
parts.push(...msg.parts);
}
}
}
// Parts from the final status message
if (result.status?.message?.parts) {
parts.push(...result.status.message.parts);
}
// Parts from artifacts
if (result.artifacts) {
for (const artifact of result.artifacts) {
parts.push(...(artifact.parts ?? []));
}
}
}
return parts;
}
/**
* Extracts plain text content from response parts.
*/
export function extractTextFromParts(parts: Part[]): string {
return parts
.filter((p) => p.kind === 'text')
.map(
(p) =>
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
(p as unknown as { text: string }).text,
)
.filter(Boolean)
.join('\n');
}
/**
* Extracts A2UI data parts from response parts.
* A2UI parts are DataParts with metadata.mimeType === 'application/json+a2ui'.
*/
export function extractA2UIParts(parts: Part[]): unknown[][] {
const a2uiMessages: unknown[][] = [];
for (const part of parts) {
if (
part.kind === 'data' &&
part.metadata != null &&
part.metadata['mimeType'] === A2UI_MIME_TYPE
) {
// The data field is an array of A2UI messages
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
const data = (part as unknown as { data: unknown }).data;
if (Array.isArray(data)) {
a2uiMessages.push(data);
}
}
}
return a2uiMessages;
}
/**
* A2A client for the chat bridge.
* Manages connection to the A2A server and provides message send/receive.
*/
export class A2ABridgeClient {
private client: Client | null = null;
private agentUrl: string;
constructor(agentUrl: string) {
this.agentUrl = agentUrl;
}
/**
* Initializes the client connection to the A2A server.
* On Cloud Run (K_SERVICE is set), wraps fetch with an identity token
* for service-to-service authentication.
*/
async initialize(): Promise<void> {
if (this.client) return;
// Create fetch wrapper with long SSE timeouts.
// On Cloud Run, also add identity tokens for service-to-service auth.
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const baseFetch = (input: any, init?: any) =>
fetch(input, { ...init, dispatcher: sseDispatcher });
let fetchImpl: typeof fetch = baseFetch;
if (process.env['K_SERVICE']) {
const auth = new GoogleAuth();
const idTokenClient = await auth.getIdTokenClient(this.agentUrl);
fetchImpl = async (input, init?) => {
const authHeaders = await idTokenClient.getRequestHeaders();
const merged = new Headers(init?.headers);
for (const [key, value] of Object.entries(authHeaders)) {
merged.set(key, value);
}
return baseFetch(input, { ...init, headers: merged });
};
logger.info(
'[ChatBridge] Using Cloud Run identity token for A2A server auth',
);
}
const resolver = new DefaultAgentCardResolver({ fetchImpl });
const options = ClientFactoryOptions.createFrom(
ClientFactoryOptions.default,
{
transports: [
new RestTransportFactory({ fetchImpl }),
new JsonRpcTransportFactory({ fetchImpl }),
],
cardResolver: resolver,
},
);
const factory = new ClientFactory(options);
// createFromUrl expects the agent card URL, not just the base URL
const agentCardUrl =
this.agentUrl.replace(/\/$/, '') + '/.well-known/agent-card.json';
this.client = await factory.createFromUrl(agentCardUrl, '');
const card = await this.client.getAgentCard();
logger.info(
`[ChatBridge] Connected to A2A agent: ${card.name} (${card.url})`,
);
}
/**
* Sends a text message to the A2A server using blocking mode.
* The blocking DefaultRequestHandler accumulates all intermediate events
* (including A2UI content from "working" status updates) into the Task's
* history array, so extractAllParts can find them.
*/
async sendMessage(
text: string,
options: { contextId?: string; taskId?: string },
): Promise<A2AResponse> {
if (!this.client) {
throw new Error('A2A client not initialized. Call initialize() first.');
}
const params: MessageSendParams = {
message: {
kind: 'message',
role: 'user',
messageId: uuidv4(),
parts: [{ kind: 'text', text }],
contextId: options.contextId,
taskId: options.taskId,
// Signal A2UI support in message metadata
metadata: {
extensions: [A2UI_EXTENSION_URI],
},
},
configuration: {
blocking: true,
},
};
return this.client.sendMessage(params);
}
/**
* Sends a text message and returns a streaming async generator.
* Each yielded event is a Message, Task, TaskStatusUpdateEvent,
* or TaskArtifactUpdateEvent.
*/
sendMessageStream(
text: string,
options: { contextId?: string; taskId?: string },
): AsyncGenerator<A2AStreamEventData, void, undefined> {
if (!this.client) {
throw new Error('A2A client not initialized. Call initialize() first.');
}
const params: MessageSendParams = {
message: {
kind: 'message',
role: 'user',
messageId: uuidv4(),
parts: [{ kind: 'text', text }],
contextId: options.contextId,
taskId: options.taskId,
metadata: {
extensions: [A2UI_EXTENSION_URI],
},
},
};
return this.client.sendMessageStream(params);
}
/**
* Sends a tool confirmation action back to the A2A server.
* The action is sent as a DataPart containing the A2UI action message.
*/
async sendToolConfirmation(
callId: string,
outcome: string,
taskId: string,
options: { contextId?: string },
): Promise<A2AResponse> {
if (!this.client) {
throw new Error('A2A client not initialized. Call initialize() first.');
}
// Build the A2UI action message as a DataPart
const actionPart: Part = {
kind: 'data',
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
data: [
{
version: 'v0.10',
action: {
name: 'tool_confirmation',
surfaceId: `tool_approval_${taskId}_${callId}`,
sourceComponentId:
outcome === 'cancel' ? 'reject_button' : 'approve_button',
timestamp: new Date().toISOString(),
context: { callId, outcome, taskId },
},
},
] as unknown as Record<string, unknown>,
metadata: {
mimeType: A2UI_MIME_TYPE,
},
} as Part;
const params: MessageSendParams = {
message: {
kind: 'message',
role: 'user',
messageId: uuidv4(),
parts: [actionPart],
contextId: options.contextId,
taskId,
metadata: {
extensions: [A2UI_EXTENSION_URI],
},
},
configuration: {
blocking: true,
},
};
return this.client.sendMessage(params);
}
/**
* Sends a tool confirmation via streaming (SSE) so the caller can
* follow the full task lifecycle after approval.
*/
sendToolConfirmationStream(
callId: string,
outcome: string,
taskId: string,
options: { contextId?: string },
): AsyncGenerator<A2AStreamEventData, void, undefined> {
if (!this.client) {
throw new Error('A2A client not initialized. Call initialize() first.');
}
const actionPart: Part = {
kind: 'data',
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
data: [
{
version: 'v0.10',
action: {
name: 'tool_confirmation',
surfaceId: `tool_approval_${taskId}_${callId}`,
sourceComponentId:
outcome === 'cancel' ? 'reject_button' : 'approve_button',
timestamp: new Date().toISOString(),
context: { callId, outcome, taskId },
},
},
] as unknown as Record<string, unknown>,
metadata: {
mimeType: A2UI_MIME_TYPE,
},
} as Part;
const params: MessageSendParams = {
message: {
kind: 'message',
role: 'user',
messageId: uuidv4(),
parts: [actionPart],
contextId: options.contextId,
taskId,
metadata: {
extensions: [A2UI_EXTENSION_URI],
},
},
};
return this.client.sendMessageStream(params);
}
/**
* Sends multiple tool confirmations in a single A2A message.
* Needed when the agent requests multiple tool approvals at once —
* sending them one at a time with blocking mode would hang because
* the agent waits for ALL approvals before proceeding.
*/
async sendBatchToolConfirmations(
approvals: Array<{ callId: string; outcome: string; taskId: string }>,
options: { contextId?: string },
): Promise<A2AResponse> {
if (!this.client) {
throw new Error('A2A client not initialized. Call initialize() first.');
}
const parts: Part[] = approvals.map(
(approval) =>
({
kind: 'data',
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
data: [
{
version: 'v0.10',
action: {
name: 'tool_confirmation',
surfaceId: `tool_approval_${approval.taskId}_${approval.callId}`,
sourceComponentId:
approval.outcome === 'cancel'
? 'reject_button'
: 'approve_button',
timestamp: new Date().toISOString(),
context: {
callId: approval.callId,
outcome: approval.outcome,
taskId: approval.taskId,
},
},
},
] as unknown as Record<string, unknown>,
metadata: {
mimeType: A2UI_MIME_TYPE,
},
}) as Part,
);
const params: MessageSendParams = {
message: {
kind: 'message',
role: 'user',
messageId: uuidv4(),
parts,
contextId: options.contextId,
taskId: approvals[0]?.taskId,
metadata: {
extensions: [A2UI_EXTENSION_URI],
},
},
configuration: {
blocking: true,
},
};
return this.client.sendMessage(params);
}
/**
* Sends batch tool confirmations via streaming (SSE) so the caller can
* follow the full task lifecycle after approval. Returns an async generator
* that yields events until the task reaches a terminal state.
*/
sendBatchToolConfirmationsStream(
approvals: Array<{ callId: string; outcome: string; taskId: string }>,
options: { contextId?: string },
): AsyncGenerator<A2AStreamEventData, void, undefined> {
if (!this.client) {
throw new Error('A2A client not initialized. Call initialize() first.');
}
const parts: Part[] = approvals.map(
(approval) =>
({
kind: 'data',
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
data: [
{
version: 'v0.10',
action: {
name: 'tool_confirmation',
surfaceId: `tool_approval_${approval.taskId}_${approval.callId}`,
sourceComponentId:
approval.outcome === 'cancel'
? 'reject_button'
: 'approve_button',
timestamp: new Date().toISOString(),
context: {
callId: approval.callId,
outcome: approval.outcome,
taskId: approval.taskId,
},
},
},
] as unknown as Record<string, unknown>,
metadata: {
mimeType: A2UI_MIME_TYPE,
},
}) as Part,
);
const params: MessageSendParams = {
message: {
kind: 'message',
role: 'user',
messageId: uuidv4(),
parts,
contextId: options.contextId,
taskId: approvals[0]?.taskId,
metadata: {
extensions: [A2UI_EXTENSION_URI],
},
},
};
return this.client.sendMessageStream(params);
}
}
@@ -0,0 +1,212 @@
/**
* @license
* Copyright 2026 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
/**
* Google Chat REST API client for sending proactive messages.
* Used to push agent responses back to Google Chat after the webhook
* has already returned an immediate acknowledgment.
*/
import { GoogleAuth } from 'google-auth-library';
import type { ChatCardV2 } from './types.js';
import { logger } from '../utils/logger.js';
const CHAT_API_BASE = 'https://chat.googleapis.com/v1';
/** Google Chat max text length. Leave margin for formatting overhead. */
const MAX_TEXT_LENGTH = 4000;
export interface ChatApiClientConfig {
/** Path to service account key JSON file. If not set, uses ADC. */
serviceAccountKeyPath?: string;
}
export class ChatApiClient {
private auth: GoogleAuth;
private initialized = false;
constructor(config?: ChatApiClientConfig) {
this.auth = new GoogleAuth({
keyFile: config?.serviceAccountKeyPath,
scopes: ['https://www.googleapis.com/auth/chat.bot'],
});
}
async initialize(): Promise<void> {
if (this.initialized) return;
await this.auth.getClient();
this.initialized = true;
logger.info('[ChatApiClient] Initialized with chat.bot scope');
}
/**
* Sends a new message to a Google Chat space in a specific thread.
* Automatically splits text longer than 4000 chars into multiple messages.
*/
async sendMessage(
spaceName: string,
threadName: string,
options: { text?: string; cardsV2?: ChatCardV2[] },
): Promise<string | undefined> {
if (!this.initialized) await this.initialize();
const chunks = options.text ? splitText(options.text) : [''];
// First chunk gets the cards (if any). Subsequent chunks are text-only.
let lastMessageName: string | undefined;
for (let i = 0; i < chunks.length; i++) {
const message: Record<string, unknown> = {};
if (chunks[i]) message['text'] = chunks[i];
if (i === 0 && options.cardsV2) message['cardsV2'] = options.cardsV2;
message['thread'] = { name: threadName };
const name = await this.postMessage(spaceName, message);
if (name) lastMessageName = name;
}
return lastMessageName;
}
/** Posts a single message to the Chat API. */
private async postMessage(
spaceName: string,
message: Record<string, unknown>,
): Promise<string | undefined> {
try {
const url =
`${CHAT_API_BASE}/${spaceName}/messages` +
`?messageReplyOption=REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD`;
const client = await this.auth.getClient();
const headers = await client.getRequestHeaders();
const response = await fetch(url, {
method: 'POST',
headers: {
...headers,
'Content-Type': 'application/json',
},
body: JSON.stringify(message),
});
if (!response.ok) {
const body = await response.text();
logger.error(
`[ChatApiClient] sendMessage failed: ${response.status} ${body}`,
);
return undefined;
}
const result: unknown = await response.json();
let messageName: string | undefined;
if (typeof result === 'object' && result !== null && 'name' in result) {
const rec = result as Record<string, unknown>;
if (typeof rec['name'] === 'string') {
messageName = rec['name'];
}
}
logger.info(
`[ChatApiClient] Message sent to ${spaceName}: ${messageName ?? 'unknown'}`,
);
return messageName;
} catch (error) {
const msg = error instanceof Error ? error.message : 'Unknown error';
logger.error(`[ChatApiClient] sendMessage error: ${msg}`, error);
return undefined;
}
}
/**
* Updates an existing message in Google Chat.
*/
async updateMessage(
messageName: string,
options: { text?: string; cardsV2?: ChatCardV2[] },
): Promise<void> {
try {
if (!this.initialized) await this.initialize();
const message: Record<string, unknown> = {};
const updateMasks: string[] = [];
if (options.text) {
message['text'] = options.text;
updateMasks.push('text');
}
if (options.cardsV2) {
message['cardsV2'] = options.cardsV2;
updateMasks.push('cardsV2');
}
const url = `${CHAT_API_BASE}/${messageName}?updateMask=${updateMasks.join(',')}`;
const client = await this.auth.getClient();
const headers = await client.getRequestHeaders();
const response = await fetch(url, {
method: 'PATCH',
headers: {
...headers,
'Content-Type': 'application/json',
},
body: JSON.stringify(message),
});
if (!response.ok) {
const body = await response.text();
logger.error(
`[ChatApiClient] updateMessage failed: ${response.status} ${body}`,
);
} else {
logger.info(`[ChatApiClient] Message updated: ${messageName}`);
}
} catch (error) {
const msg = error instanceof Error ? error.message : 'Unknown error';
logger.error(`[ChatApiClient] updateMessage error: ${msg}`, error);
}
}
}
/**
* Splits text into chunks that fit within Google Chat's character limit.
* Splits on paragraph boundaries (double newline) first, then single
* newlines, then hard-splits as a last resort.
*/
function splitText(text: string): string[] {
if (text.length <= MAX_TEXT_LENGTH) return [text];
const chunks: string[] = [];
let remaining = text;
while (remaining.length > MAX_TEXT_LENGTH) {
let splitAt = -1;
// Try splitting at a paragraph boundary
const paraIdx = remaining.lastIndexOf('\n\n', MAX_TEXT_LENGTH);
if (paraIdx > MAX_TEXT_LENGTH * 0.3) {
splitAt = paraIdx + 2; // include the double newline in the first chunk
}
// Fall back to single newline
if (splitAt < 0) {
const lineIdx = remaining.lastIndexOf('\n', MAX_TEXT_LENGTH);
if (lineIdx > MAX_TEXT_LENGTH * 0.3) {
splitAt = lineIdx + 1;
}
}
// Hard split as last resort
if (splitAt < 0) {
splitAt = MAX_TEXT_LENGTH;
}
chunks.push(remaining.substring(0, splitAt));
remaining = remaining.substring(splitAt);
}
if (remaining) chunks.push(remaining);
return chunks;
}
@@ -0,0 +1,841 @@
/**
* @license
* Copyright 2026 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
/**
* Google Chat webhook handler.
* Responds immediately with "Processing..." and streams results
* from the A2A server, pushing updates to Chat via the REST API.
*/
import type { ChatEvent, ChatResponse, ChatBridgeConfig } from './types.js';
import type { SessionInfo } from './session-store.js';
import { SessionStore } from './session-store.js';
import {
A2ABridgeClient,
type A2AStreamEventData,
} from './a2a-bridge-client.js';
import { ChatApiClient } from './chat-api-client.js';
import { renderResponse, extractFromStreamEvent } from './response-renderer.js';
import { logger } from '../utils/logger.js';
const TERMINAL_STATES = new Set([
'completed',
'failed',
'canceled',
'rejected',
]);
export class ChatBridgeHandler {
private sessionStore: SessionStore;
private a2aClient: A2ABridgeClient;
private chatApiClient: ChatApiClient;
private initialized = false;
/** Full webhook URL for card button actions (HTTP Add-ons need a URL, not a function name). */
private webhookUrl: string | undefined;
constructor(
private config: ChatBridgeConfig,
chatApiClient?: ChatApiClient,
) {
this.sessionStore = new SessionStore(config.gcsBucket);
this.a2aClient = new A2ABridgeClient(config.a2aServerUrl);
this.chatApiClient =
chatApiClient ??
new ChatApiClient({
serviceAccountKeyPath: config.serviceAccountKeyPath,
});
// For HTTP Add-ons, card button action.function must be a full HTTPS URL.
// Set CHAT_WEBHOOK_URL env var to the bridge's public webhook endpoint.
this.webhookUrl = process.env['CHAT_WEBHOOK_URL'] || undefined;
if (this.webhookUrl) {
logger.info(`[ChatBridge] Button action URL: ${this.webhookUrl}`);
}
}
/**
* Initializes the A2A client connection, Chat API client,
* and restores persisted sessions. Must be called before handling events.
*/
async initialize(): Promise<void> {
if (this.initialized) return;
await this.a2aClient.initialize();
await this.chatApiClient.initialize();
await this.sessionStore.restore();
this.initialized = true;
logger.info(
`[ChatBridge] Handler initialized, connected to ${this.config.a2aServerUrl}`,
);
}
/**
* Main entry point for handling Google Chat webhook events.
*/
async handleEvent(event: ChatEvent): Promise<ChatResponse> {
if (!this.initialized) {
await this.initialize();
}
logger.info(
`[ChatBridge] Received event: type=${event.type}, space=${event.space.name}`,
);
switch (event.type) {
case 'MESSAGE':
return this.handleMessage(event);
case 'CARD_CLICKED':
return this.handleCardClicked(event);
case 'ADDED_TO_SPACE':
return this.handleAddedToSpace(event);
case 'REMOVED_FROM_SPACE':
return this.handleRemovedFromSpace(event);
default:
logger.warn(`[ChatBridge] Unknown event type: ${event.type}`);
return { text: 'Unknown event type.' };
}
}
/**
* Pushes a text message via Chat API (properly threaded) and returns
* an empty webhook response. Add-ons createMessageAction ignores
* thread info and always creates a top-level message in Spaces,
* so ALL user-visible messages must go through the Chat API.
*/
private pushAndReturn(
spaceName: string,
threadName: string,
text: string,
): ChatResponse {
this.chatApiClient
.sendMessage(spaceName, threadName, { text })
.catch((err) => {
const msg = err instanceof Error ? err.message : 'Unknown error';
logger.warn(`[ChatBridge] Failed to push message: ${msg}`);
});
return {};
}
/**
* Handles a MESSAGE event: user sent a text message in Chat.
* All responses are pushed via Chat API for proper threading in Spaces.
* The webhook always returns an empty response.
*/
private async handleMessage(event: ChatEvent): Promise<ChatResponse> {
const message = event.message;
if (!message?.thread?.name) {
return { text: 'Error: Missing thread information.' };
}
// argumentText has bot mentions stripped (legacy format only).
// For Add-ons format, strip leading @mention manually.
const rawText = message.argumentText || message.text || '';
const text = rawText.replace(/^@\S+\s*/, '');
if (!text.trim()) {
return { text: "I didn't receive any text. Please try again." };
}
const threadName = message.thread.name;
const spaceName = event.space.name;
// Handle slash commands — push response via Chat API for threading
const trimmed = text.trim().toLowerCase();
if (
trimmed === '/reset' ||
trimmed === '/clear' ||
trimmed === 'reset' ||
trimmed === 'clear'
) {
this.sessionStore.remove(threadName);
logger.info(`[ChatBridge] Session cleared for thread ${threadName}`);
return this.pushAndReturn(
spaceName,
threadName,
'Session cleared. Send a new message to start fresh.',
);
}
const session = this.sessionStore.getOrCreate(threadName, spaceName);
if (trimmed === '/yolo') {
session.yoloMode = true;
logger.info(`[ChatBridge] YOLO mode enabled for thread ${threadName}`);
return this.pushAndReturn(
spaceName,
threadName,
'YOLO mode enabled. All tool calls will be auto-approved.',
);
}
if (trimmed === '/safe') {
session.yoloMode = false;
logger.info(`[ChatBridge] YOLO mode disabled for thread ${threadName}`);
return this.pushAndReturn(
spaceName,
threadName,
'Safe mode enabled. Tool calls will require approval.',
);
}
logger.info(
`[ChatBridge] MESSAGE from ${event.user.displayName}: "${text.substring(0, 100)}"`,
);
// Handle text-based tool approval responses
if (session.pendingToolApproval && this.isToolApprovalText(trimmed)) {
return this.handleToolApprovalText(event, session, trimmed);
}
// Guard against overlapping async requests
if (session.asyncProcessing) {
return this.pushAndReturn(
spaceName,
threadName,
'Still processing your previous request. Please wait...',
);
}
// Fire-and-forget async processing
this.processMessageAsync(event, session, text).catch((err) => {
const msg = err instanceof Error ? err.message : 'Unknown error';
logger.error(`[ChatBridge] Async processing failed: ${msg}`, err);
});
// Push "Processing..." via Chat API for proper threading
return this.pushAndReturn(
spaceName,
threadName,
'_Processing your request..._',
);
}
/**
* Checks if text is a tool approval command.
*/
private isToolApprovalText(text: string): boolean {
return (
text === 'approve' ||
text === 'yes' ||
text === 'y' ||
text === 'reject' ||
text === 'no' ||
text === 'n' ||
text === 'always allow'
);
}
/**
* Handles text-based tool approval responses.
* Returns an immediate acknowledgment and processes the confirmation
* asynchronously, pushing the agent's response via Chat API.
*/
private handleToolApprovalText(
event: ChatEvent,
session: SessionInfo,
trimmed: string,
): ChatResponse {
const message = event.message!;
const threadName = message.thread.name;
const approval = session.pendingToolApproval!;
const isReject =
trimmed === 'reject' || trimmed === 'no' || trimmed === 'n';
const isAlwaysAllow = trimmed === 'always allow';
const outcome = isReject
? 'cancel'
: isAlwaysAllow
? 'proceed_always_tool'
: 'proceed_once';
logger.info(
`[ChatBridge] Text-based tool ${outcome}: callId=${approval.callId}, taskId=${approval.taskId}`,
);
session.pendingToolApproval = undefined;
// Fire-and-forget async processing of the tool confirmation
this.processToolApprovalAsync(event, session, approval, outcome).catch(
(err) => {
const msg = err instanceof Error ? err.message : 'Unknown error';
logger.error(
`[ChatBridge] Tool approval async processing failed: ${msg}`,
err,
);
},
);
const ackText = isReject
? '_Tool rejected._'
: '_Tool approved, processing..._';
return this.pushAndReturn(event.space.name, threadName, ackText);
}
/**
* Processes a tool confirmation asynchronously.
* Sends the confirmation to the A2A server, handles the response,
* and pushes results to Google Chat via the REST API.
*/
private async processToolApprovalAsync(
event: ChatEvent,
session: SessionInfo,
approval: { callId: string; taskId: string; toolName: string },
outcome: string,
): Promise<void> {
const message = event.message!;
const threadName = message.thread.name;
const spaceName = event.space.name;
session.asyncProcessing = true;
try {
// Stream the tool confirmation to collect text as it arrives
const stream = this.a2aClient.sendToolConfirmationStream(
approval.callId,
outcome,
approval.taskId,
{ contextId: session.contextId },
);
let lastText = '';
let lastTaskId: string | undefined;
let lastContextId: string | undefined;
let lastState: string | undefined;
let sentResponse = false;
for await (const streamEvent of stream) {
if (session.cancelled) break;
const extracted = extractFromStreamEvent(streamEvent);
if (extracted.taskId) lastTaskId = extracted.taskId;
if (extracted.contextId) lastContextId = extracted.contextId;
if (extracted.state) lastState = extracted.state;
if (extracted.text) lastText = extracted.text;
// Check for new tool approvals
const pending = extracted.toolApprovals.filter(
(a) => a.status === 'awaiting_approval',
);
if (pending.length > 0) {
if (session.yoloMode) {
// YOLO: auto-approve via streaming
const autoResult = await this.autoApproveTools(
session,
pending,
lastContextId,
);
if (autoResult.lastContextId)
lastContextId = autoResult.lastContextId;
if (autoResult.lastTaskId) lastTaskId = autoResult.lastTaskId;
if (autoResult.lastState) lastState = autoResult.lastState;
if (autoResult.text) lastText = autoResult.text;
} else {
// Non-YOLO: push approval card
session.pendingToolApproval = {
callId: pending[0].callId,
taskId: pending[0].taskId,
toolName: pending[0].displayName || pending[0].name,
};
// Build a minimal task response for the card renderer
const cardResponse = renderResponse(
{
kind: 'task',
id: pending[0].taskId,
contextId: lastContextId ?? session.contextId,
status: {
state: 'input-required',
timestamp: new Date().toISOString(),
message:
streamEvent.kind === 'status-update'
? streamEvent.status?.message
: undefined,
},
history: [],
artifacts: [],
},
message.thread.threadKey || threadName,
threadName,
this.webhookUrl,
);
await this.chatApiClient.sendMessage(spaceName, threadName, {
text: cardResponse.text,
cardsV2: cardResponse.cardsV2,
});
sentResponse = true;
logger.info(
`[ChatBridge] Pushed approval card after confirmation: ${pending[0].displayName || pending[0].name}`,
);
}
break;
}
}
if (session.cancelled) return;
// Update session IDs
if (lastContextId) session.contextId = lastContextId;
const isTerminal = lastState ? TERMINAL_STATES.has(lastState) : false;
this.sessionStore.updateTaskId(
threadName,
isTerminal ? undefined : lastTaskId,
);
// Push final text if we haven't already pushed a card
if (lastText && !sentResponse) {
await this.chatApiClient.sendMessage(spaceName, threadName, {
text: lastText,
});
logger.info(
`[ChatBridge] Pushed post-approval response (${lastText.length} chars): "${lastText.substring(0, 200)}"`,
);
}
} catch (error) {
if (session.cancelled) return;
const errorMsg = error instanceof Error ? error.message : 'Unknown error';
logger.error(
`[ChatBridge] Error in tool approval async: ${errorMsg}`,
error,
);
await this.chatApiClient.sendMessage(spaceName, threadName, {
text: `Error processing tool confirmation: ${errorMsg}`,
});
} finally {
session.asyncProcessing = false;
}
}
/**
* Processes a message asynchronously using A2A streaming.
* Pushes results to Google Chat via the REST API.
*/
private async processMessageAsync(
event: ChatEvent,
session: SessionInfo,
text: string,
): Promise<void> {
const message = event.message!;
const threadName = message.thread.name;
const spaceName = event.space.name;
session.asyncProcessing = true;
try {
// Retry streaming if the A2A server returns 500 (no available instance).
// With concurrency=1, this happens when another request is in-flight.
const stream = await this.retryStream(
() =>
this.a2aClient.sendMessageStream(text, {
contextId: session.contextId,
taskId: session.taskId,
}),
session,
);
let lastText = '';
let lastTaskId: string | undefined;
let lastContextId: string | undefined;
let lastState: string | undefined;
let sentFinalResponse = false;
let eventCount = 0;
// Track the latest pending approvals across events — only act on them
// when the server signals input-required (meaning it actually needs input).
// In server YOLO mode, tools are auto-approved so the stream continues
// past the brief 'awaiting_approval' status without hitting input-required.
let latestPendingApprovals: Array<{
callId: string;
taskId: string;
name: string;
displayName: string;
}> = [];
let approvalStatusMessage: unknown;
for await (const streamEvent of stream) {
// Check if session was cancelled (e.g. by /reset)
if (session.cancelled) {
logger.info(
`[ChatBridge] Session cancelled, stopping stream for ${threadName}`,
);
break;
}
eventCount++;
const extracted = extractFromStreamEvent(streamEvent);
if (extracted.taskId) lastTaskId = extracted.taskId;
if (extracted.contextId) lastContextId = extracted.contextId;
if (extracted.state) lastState = extracted.state;
// Log each event for debugging
logger.info(
`[ChatBridge] Stream event #${eventCount}: kind=${streamEvent.kind}, ` +
`state=${extracted.state ?? 'n/a'}, text=${extracted.text.length} chars, ` +
`approvals=${extracted.toolApprovals.filter((a) => a.status === 'awaiting_approval').length}`,
);
// Track latest text content
if (extracted.text) {
lastText = extracted.text;
}
// Track tool approvals — always update with latest state so
// stale approvals are cleared when the server auto-approves.
const pending = extracted.toolApprovals.filter(
(a) => a.status === 'awaiting_approval',
);
latestPendingApprovals = pending;
if (pending.length > 0) {
approvalStatusMessage =
streamEvent.kind === 'status-update'
? streamEvent.status?.message
: undefined;
}
// On terminal or input-required state, stop streaming.
// input-required means the server is asking for user action
// (tool approval or follow-up message).
if (
extracted.state &&
(TERMINAL_STATES.has(extracted.state) ||
extracted.state === 'input-required')
) {
break;
}
}
logger.info(
`[ChatBridge] Stream complete: ${eventCount} events, ` +
`state=${lastState ?? 'none'}, text=${lastText.length} chars, ` +
`pendingApprovals=${latestPendingApprovals.length}`,
);
// Handle pending approvals (only relevant when server sent input-required)
if (latestPendingApprovals.length > 0 && lastState === 'input-required') {
if (session.yoloMode) {
// Bridge YOLO mode: auto-approve all tools
const autoApproved = await this.autoApproveTools(
session,
latestPendingApprovals,
lastContextId,
);
if (autoApproved.lastContextId)
lastContextId = autoApproved.lastContextId;
if (autoApproved.lastTaskId) lastTaskId = autoApproved.lastTaskId;
if (autoApproved.lastState) lastState = autoApproved.lastState;
if (autoApproved.text) lastText = autoApproved.text;
} else {
// Non-YOLO: push approval card and wait for user input
const firstApproval = latestPendingApprovals[0];
session.pendingToolApproval = {
callId: firstApproval.callId,
taskId: firstApproval.taskId,
toolName: firstApproval.displayName || firstApproval.name,
};
const approvalResponse = renderResponse(
{
kind: 'task',
id: firstApproval.taskId,
contextId: lastContextId ?? session.contextId,
status: {
state: 'input-required',
timestamp: new Date().toISOString(),
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
message: approvalStatusMessage as
| import('@a2a-js/sdk').Message
| undefined,
},
history: [],
artifacts: [],
},
message.thread.threadKey || threadName,
threadName,
this.webhookUrl,
);
await this.chatApiClient.sendMessage(spaceName, threadName, {
text: approvalResponse.text,
cardsV2: approvalResponse.cardsV2,
});
sentFinalResponse = true;
logger.info(
`[ChatBridge] Pushed tool approval card: ${firstApproval.displayName || firstApproval.name}`,
);
}
}
// If session was cancelled, don't push any messages
if (session.cancelled) {
logger.info(
`[ChatBridge] Skipping response push for cancelled session ${threadName}`,
);
return;
}
// Update session IDs
if (lastContextId) session.contextId = lastContextId;
// Clear taskId on terminal states so next message starts a fresh task
const isTerminal = lastState ? TERMINAL_STATES.has(lastState) : false;
this.sessionStore.updateTaskId(
threadName,
isTerminal ? undefined : lastTaskId,
);
// Push final text response if we haven't already pushed a tool approval
if (lastText && !sentFinalResponse) {
await this.chatApiClient.sendMessage(spaceName, threadName, {
text: lastText,
});
logger.info(
`[ChatBridge] Pushed final response (${lastText.length} chars): "${lastText.substring(0, 200)}"`,
);
} else if (!lastText && !sentFinalResponse) {
await this.chatApiClient.sendMessage(spaceName, threadName, {
text: '_Agent completed without generating a response._',
});
}
} catch (error) {
if (session.cancelled) return; // Don't push errors for cancelled sessions
const errorMsg = error instanceof Error ? error.message : 'Unknown error';
logger.error(`[ChatBridge] Async processing error: ${errorMsg}`, error);
await this.chatApiClient.sendMessage(spaceName, threadName, {
text: `Sorry, I encountered an error: ${errorMsg}`,
});
} finally {
session.asyncProcessing = false;
}
}
/**
* Retries creating a stream when the A2A server returns 500.
* Cloud Run returns 500 "no available instance" when concurrency is
* exhausted. We retry with exponential backoff up to 3 times.
*/
private async retryStream(
createStream: () => AsyncGenerator<A2AStreamEventData, void, undefined>,
session: SessionInfo,
): Promise<AsyncGenerator<A2AStreamEventData, void, undefined>> {
const MAX_RETRIES = 3;
const BASE_DELAY_MS = 5000;
for (let attempt = 0; attempt <= MAX_RETRIES; attempt++) {
if (session.cancelled) return createStream(); // will be caught by caller
try {
const stream = createStream();
// Try to get the first value to verify the stream connects
const iter = stream[Symbol.asyncIterator]();
const first = await iter.next();
// Re-wrap into an async generator that yields the first value
// then delegates to the rest of the iterator
async function* replayStream(): AsyncGenerator<
A2AStreamEventData,
void,
undefined
> {
if (!first.done) {
yield first.value;
yield* { [Symbol.asyncIterator]: () => iter };
}
}
return replayStream();
} catch (error) {
const msg = error instanceof Error ? error.message : '';
const isRetryable = msg.includes('500') || msg.includes('503');
if (!isRetryable || attempt === MAX_RETRIES) throw error;
const delay = BASE_DELAY_MS * Math.pow(2, attempt);
logger.warn(
`[ChatBridge] A2A server unavailable, retrying in ${delay}ms (attempt ${attempt + 1}/${MAX_RETRIES})`,
);
await new Promise((r) => setTimeout(r, delay));
}
}
// Should not reach here, but just in case
return createStream();
}
/**
* Auto-approves tool calls in YOLO mode using streaming.
* Sends approvals and collects streamed text from the SSE response.
* The A2A server streams text incrementally and closes with final:true
* at input-required (more tools) or completed (done).
*/
private async autoApproveTools(
session: SessionInfo,
initialApprovals: Array<{
callId: string;
taskId: string;
name: string;
displayName: string;
}>,
contextId: string | undefined,
): Promise<{
lastContextId?: string;
lastTaskId?: string;
lastState?: string;
text?: string;
}> {
let approvalsToProcess = initialApprovals;
let lastContextId = contextId;
let lastTaskId: string | undefined;
let lastState: string | undefined;
let lastText: string | undefined;
const approvedNames: string[] = [];
const MAX_ROUNDS = 20;
for (let round = 0; round < MAX_ROUNDS && !session.cancelled; round++) {
if (approvalsToProcess.length === 0) break;
for (const a of approvalsToProcess) {
const label = a.displayName || a.name;
logger.info(`[ChatBridge] YOLO auto-approving: ${label}`);
approvedNames.push(label);
}
// Stream tool confirmations — text arrives incrementally via SSE
const stream = this.a2aClient.sendBatchToolConfirmationsStream(
approvalsToProcess.map((a) => ({
callId: a.callId,
outcome: 'proceed_once',
taskId: a.taskId,
})),
{ contextId: lastContextId ?? session.contextId },
);
approvalsToProcess = [];
// Consume the stream, collecting text and detecting new tool approvals
let eventCount = 0;
for await (const event of stream) {
if (session.cancelled) break;
eventCount++;
const extracted = extractFromStreamEvent(event);
if (extracted.taskId) lastTaskId = extracted.taskId;
if (extracted.contextId) lastContextId = extracted.contextId;
if (extracted.state) lastState = extracted.state;
if (extracted.text) lastText = extracted.text;
logger.info(
`[ChatBridge] YOLO event #${eventCount}: kind=${event.kind}, ` +
`state=${extracted.state ?? 'n/a'}, text=${extracted.text.length} chars`,
);
// New tool approvals → break to send them in next round
const pending = extracted.toolApprovals.filter(
(a) => a.status === 'awaiting_approval',
);
if (pending.length > 0) {
approvalsToProcess = pending;
break;
}
}
logger.info(
`[ChatBridge] YOLO round ${round}: state=${lastState ?? 'none'}, ` +
`text=${lastText?.length ?? 0} chars, newApprovals=${approvalsToProcess.length}`,
);
if (lastState && TERMINAL_STATES.has(lastState)) break;
}
logger.info(
`[ChatBridge] YOLO auto-approved ${approvedNames.length} tools: ${approvedNames.join(', ')}`,
);
return { lastContextId, lastTaskId, lastState, text: lastText };
}
/**
* Handles a CARD_CLICKED event: user clicked a button on a card.
* Fires async processing and returns an immediate UPDATE_MESSAGE ack.
*/
private handleCardClicked(event: ChatEvent): ChatResponse {
const action = event.action;
if (!action) {
return { text: 'Error: Missing action data.' };
}
const threadName = event.message?.thread?.name;
if (!threadName) {
return { text: 'Error: Missing thread information.' };
}
const session = this.sessionStore.get(threadName);
if (!session) {
return { text: 'Error: No active session found for this thread.' };
}
logger.info(
`[ChatBridge] CARD_CLICKED: function=${action.actionMethodName}`,
);
if (
action.actionMethodName === 'tool_confirmation' ||
action.actionMethodName === this.webhookUrl
) {
const params = action.parameters || [];
const paramMap = new Map(params.map((p) => [p.key, p.value]));
const callId = paramMap.get('callId');
const outcome = paramMap.get('outcome');
const taskId = paramMap.get('taskId');
if (!callId || !outcome || !taskId) {
return { text: 'Error: Missing tool confirmation parameters.' };
}
const isReject = outcome === 'cancel';
const toolName = session.pendingToolApproval?.toolName ?? 'Tool';
// Clear pending approval tracked for text-based flow
session.pendingToolApproval = undefined;
// Fire-and-forget async processing
this.processToolApprovalAsync(
event,
session,
{ callId, taskId, toolName },
outcome,
).catch((err) => {
const msg = err instanceof Error ? err.message : 'Unknown error';
logger.error(`[ChatBridge] Card click async failed: ${msg}`, err);
});
// Update the card in-place with an acknowledgment
return {
actionResponse: { type: 'UPDATE_MESSAGE' },
text: isReject
? `*${toolName} — Rejected*`
: `*${toolName} — Approved, processing...*`,
};
}
return { text: `Unknown action: ${action.actionMethodName}` };
}
/**
* Handles ADDED_TO_SPACE event: bot was added to a space or DM.
*/
private handleAddedToSpace(event: ChatEvent): ChatResponse {
const spaceType = event.space.type === 'DM' ? 'DM' : 'space';
logger.info(`[ChatBridge] Bot added to ${spaceType}: ${event.space.name}`);
return {
text:
`Hello! I'm the Gemini CLI Agent. Send me a message to get started with code generation and development tasks.\n\n` +
`I can:\n` +
`- Generate code from natural language\n` +
`- Edit files and run commands\n` +
`- Answer questions about code\n\n` +
`I'll ask for your approval before executing tools.`,
};
}
/**
* Handles REMOVED_FROM_SPACE event: bot was removed from a space.
*/
private handleRemovedFromSpace(event: ChatEvent): ChatResponse {
logger.info(`[ChatBridge] Bot removed from space: ${event.space.name}`);
return {};
}
}
@@ -0,0 +1,543 @@
/**
* @license
* Copyright 2026 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
/**
* Converts A2A/A2UI responses into Google Chat messages and Cards V2.
*
* This renderer understands the A2UI v0.10 surface structures produced by our
* a2a-server (tool approval surfaces, agent response surfaces, thought surfaces)
* and converts them to Google Chat's Cards V2 format.
*
* Inspired by the A2UI web_core message processor pattern but simplified for
* server-side rendering to a constrained card format.
*/
import type { ChatResponse, ChatCardV2, ChatWidget } from './types.js';
import type { Part } from '@a2a-js/sdk';
import {
type A2AResponse,
type A2AStreamEventData,
extractAllParts,
extractTextFromParts,
extractA2UIParts,
} from './a2a-bridge-client.js';
export interface ToolApprovalInfo {
taskId: string;
callId: string;
name: string;
displayName: string;
description: string;
args: string;
kind: string;
status: string;
}
interface AgentResponseInfo {
text: string;
status: string;
}
/**
* Extracts tool approval info from an A2A response.
* Used by the handler to track pending approvals for text-based confirmation.
*/
export function extractToolApprovals(
response: A2AResponse,
): ToolApprovalInfo[] {
const parts = extractAllParts(response);
const a2uiMessageGroups = extractA2UIParts(parts);
const toolApprovals: ToolApprovalInfo[] = [];
const agentResponses: AgentResponseInfo[] = [];
const thoughts: Array<{ subject: string; description: string }> = [];
for (const messages of a2uiMessageGroups) {
parseA2UIMessages(messages, toolApprovals, agentResponses, thoughts);
}
return deduplicateToolApprovals(toolApprovals);
}
/**
* Renders an A2A response as a Google Chat response.
* Extracts text content and A2UI surfaces, converting them to Chat format.
*/
export function renderResponse(
response: A2AResponse,
threadKey?: string,
threadName?: string,
webhookUrl?: string,
): ChatResponse {
const parts = extractAllParts(response);
const textContent = extractTextFromParts(parts);
const a2uiMessageGroups = extractA2UIParts(parts);
// Parse A2UI surfaces for known types
const toolApprovals: ToolApprovalInfo[] = [];
const agentResponses: AgentResponseInfo[] = [];
const thoughts: Array<{ subject: string; description: string }> = [];
for (const messages of a2uiMessageGroups) {
parseA2UIMessages(messages, toolApprovals, agentResponses, thoughts);
}
// Deduplicate tool approvals by surfaceId — A2UI history contains both
// initial 'awaiting_approval' and later 'success' events for auto-approved tools.
const dedupedApprovals = deduplicateToolApprovals(toolApprovals);
const cards: ChatCardV2[] = [];
// Only render tool approval cards for tools still awaiting approval.
// In YOLO mode, tools are auto-approved and their status becomes "success"
// so we skip rendering approval cards for those.
for (const approval of dedupedApprovals) {
if (approval.status === 'awaiting_approval') {
cards.push(renderToolApprovalCard(approval, webhookUrl));
}
}
// Build text response from agent responses and plain text
const responseTexts: string[] = [];
// Add thought summaries
for (const thought of thoughts) {
responseTexts.push(`_${thought.subject}_: ${thought.description}`);
}
// Add agent response text (from A2UI surfaces).
// Use only the last non-empty response since later updates supersede earlier
// ones for the same surface (history contains multiple status-update messages).
for (let i = agentResponses.length - 1; i >= 0; i--) {
if (agentResponses[i].text) {
responseTexts.push(agentResponses[i].text);
break;
}
}
// Fall back to plain text content if no A2UI response text
if (responseTexts.length === 0 && textContent) {
responseTexts.push(textContent);
}
// Add task state info
if (response.kind === 'task' && response.status) {
const state = response.status.state;
if (state === 'input-required' && cards.length > 0) {
responseTexts.push('*Waiting for your approval to continue...*');
} else if (state === 'failed') {
responseTexts.push('*Task failed.*');
} else if (state === 'canceled') {
responseTexts.push('*Task was cancelled.*');
}
}
const chatResponse: ChatResponse = {};
if (responseTexts.length > 0) {
chatResponse.text = responseTexts.join('\n\n');
}
if (cards.length > 0) {
chatResponse.cardsV2 = cards;
}
if (threadKey || threadName) {
chatResponse.thread = {};
if (threadKey) chatResponse.thread.threadKey = threadKey;
if (threadName) chatResponse.thread.name = threadName;
}
// Ensure we always return something
if (!chatResponse.text && !chatResponse.cardsV2) {
chatResponse.text = '_Agent is processing..._';
}
return chatResponse;
}
/**
* Renders a CARD_CLICKED acknowledgment response.
*/
export function renderActionAcknowledgment(
action: string,
outcome: string,
): ChatResponse {
const emoji =
outcome === 'cancel'
? 'Rejected'
: outcome === 'proceed_always_tool'
? 'Always Allowed'
: 'Approved';
return {
actionResponse: { type: 'UPDATE_MESSAGE' },
text: `*Tool ${emoji}* - Processing...`,
};
}
/** Safely extracts a string property from an unknown object. */
function str(obj: Record<string, unknown>, key: string): string {
const v = obj[key];
return typeof v === 'string' ? v : '';
}
/** Safely checks if an unknown value is a record. */
function isRecord(v: unknown): v is Record<string, unknown> {
return typeof v === 'object' && v !== null && !Array.isArray(v);
}
/** Safely extracts a nested object property. */
function obj(
parent: Record<string, unknown>,
key: string,
): Record<string, unknown> | undefined {
const v = parent[key];
return isRecord(v) ? v : undefined;
}
/**
* Deduplicates tool approvals by surfaceId, keeping the last entry per surface.
* In blocking mode, A2UI history accumulates ALL intermediate events — a tool
* surface may appear first as 'awaiting_approval' then as 'success' (YOLO mode).
* By keeping only the last entry per surfaceId, auto-approved tools show 'success'.
*/
function deduplicateToolApprovals(
approvals: ToolApprovalInfo[],
): ToolApprovalInfo[] {
const byId = new Map<string, ToolApprovalInfo>();
for (const a of approvals) {
const key = `${a.taskId}_${a.callId}`;
byId.set(key, a);
}
return [...byId.values()];
}
/**
* Parses A2UI v0.10 messages to extract known surface types.
* Our server produces specific surfaces: tool approval, agent response, thought.
*/
function parseA2UIMessages(
messages: unknown[],
toolApprovals: ToolApprovalInfo[],
agentResponses: AgentResponseInfo[],
thoughts: Array<{ subject: string; description: string }>,
): void {
for (const msg of messages) {
if (!isRecord(msg)) continue;
// Look for updateDataModel messages that contain tool approval or response data
const updateDM = obj(msg, 'updateDataModel');
if (updateDM) {
const surfaceId = str(updateDM, 'surfaceId');
const value = obj(updateDM, 'value');
const path = str(updateDM, 'path');
if (value && !path) {
// Full data model update (initial) - check for known structures
const tool = obj(value, 'tool');
if (surfaceId.startsWith('tool_approval_') && tool) {
toolApprovals.push({
taskId: str(value, 'taskId'),
callId: str(tool, 'callId'),
name: str(tool, 'name'),
displayName: str(tool, 'displayName'),
description: str(tool, 'description'),
args: str(tool, 'args'),
kind: str(tool, 'kind') || 'tool',
status: str(tool, 'status') || 'unknown',
});
}
const resp = obj(value, 'response');
if (surfaceId.startsWith('agent_response_') && resp) {
agentResponses.push({
text: str(resp, 'text'),
status: str(resp, 'status'),
});
}
}
// Partial data model updates (path-based)
if (path === '/response/text' && updateDM['value'] != null) {
agentResponses.push({
text: String(updateDM['value']),
status: '',
});
}
// Tool status updates (e.g., YOLO mode changes status to 'success')
if (
surfaceId.startsWith('tool_approval_') &&
path === '/tool/status' &&
typeof updateDM['value'] === 'string'
) {
// Find existing tool approval for this surface and update its status
const existing = toolApprovals.find(
(a) => `tool_approval_${a.taskId}_${a.callId}` === surfaceId,
);
if (existing) {
existing.status = updateDM['value'];
}
}
}
// Look for updateComponents to extract thought text
const updateComp = obj(msg, 'updateComponents');
if (updateComp) {
const surfaceId = str(updateComp, 'surfaceId');
const components = updateComp['components'];
if (surfaceId.startsWith('thought_') && Array.isArray(components)) {
const subject = extractComponentText(components, 'thought_subject');
const desc = extractComponentText(components, 'thought_desc');
if (subject || desc) {
thoughts.push({
subject: subject || 'Thinking',
description: desc || '',
});
}
}
}
}
}
/**
* Extracts the text content from a named component in a component array.
* Components use our a2ui-components.ts builder format.
*/
function extractComponentText(
components: unknown[],
componentId: string,
): string {
for (const comp of components) {
if (!isRecord(comp)) continue;
if (comp['id'] === componentId && comp['component'] === 'text') {
return str(comp, 'text');
}
}
return '';
}
/**
* Extracts a concise command summary from tool approval args.
* For shell tools, returns just the command string.
* For file tools, returns the file path.
*/
function extractCommandSummary(approval: ToolApprovalInfo): string {
if (!approval.args || approval.args === 'No arguments') return '';
try {
const parsed: unknown = JSON.parse(approval.args);
if (isRecord(parsed)) {
// Shell tool: {"command": "ls -F"}
if (typeof parsed['command'] === 'string') {
return parsed['command'];
}
// File tools: {"file_path": "/path/to/file", ...}
if (typeof parsed['file_path'] === 'string') {
const action =
approval.name || approval.displayName || 'File operation';
return `${action}: ${parsed['file_path']}`;
}
}
} catch {
// Not JSON, return as-is if short enough
if (approval.args.length <= 200) return approval.args;
}
return '';
}
/**
* Renders a tool approval surface as a compact Google Chat Card V2
* with clickable Approve/Reject buttons.
*/
function renderToolApprovalCard(
approval: ToolApprovalInfo,
webhookUrl?: string,
): ChatCardV2 {
const widgets: ChatWidget[] = [];
const toolLabel = approval.displayName || approval.name;
// Show a concise summary of what the tool will do.
const commandSummary = extractCommandSummary(approval);
if (commandSummary) {
widgets.push({
decoratedText: {
text: `\`${commandSummary}\``,
topLabel: toolLabel,
startIcon: { knownIcon: 'DESCRIPTION' },
wrapText: true,
},
});
} else if (approval.args && approval.args !== 'No arguments') {
const truncatedArgs =
approval.args.length > 200
? approval.args.substring(0, 200) + '...'
: approval.args;
widgets.push({
decoratedText: {
text: truncatedArgs,
topLabel: toolLabel,
startIcon: { knownIcon: 'DESCRIPTION' },
wrapText: true,
},
});
}
// Clickable buttons for approve/reject
widgets.push({
buttonList: {
buttons: [
{
text: 'Approve',
onClick: {
action: {
function: webhookUrl ?? 'tool_confirmation',
parameters: [
{ key: 'callId', value: approval.callId },
{ key: 'outcome', value: 'proceed_once' },
{ key: 'taskId', value: approval.taskId },
],
},
},
},
{
text: 'Always Allow',
onClick: {
action: {
function: webhookUrl ?? 'tool_confirmation',
parameters: [
{ key: 'callId', value: approval.callId },
{ key: 'outcome', value: 'proceed_always_tool' },
{ key: 'taskId', value: approval.taskId },
],
},
},
},
{
text: 'Reject',
onClick: {
action: {
function: webhookUrl ?? 'tool_confirmation',
parameters: [
{ key: 'callId', value: approval.callId },
{ key: 'outcome', value: 'cancel' },
{ key: 'taskId', value: approval.taskId },
],
},
},
color: { red: 0.8, green: 0.2, blue: 0.2 },
},
],
},
});
return {
cardId: `tool_approval_${approval.callId}`,
card: {
header: {
title: toolLabel,
subtitle: 'Approval Required',
},
sections: [{ widgets }],
},
};
}
/**
* Extracts text and tool approval info from a single streaming event.
* Works with TaskStatusUpdateEvent, Task, and Message events.
*/
export function extractFromStreamEvent(event: A2AStreamEventData): {
text: string;
toolApprovals: ToolApprovalInfo[];
state?: string;
taskId?: string;
contextId?: string;
} {
const toolApprovals: ToolApprovalInfo[] = [];
const agentResponses: AgentResponseInfo[] = [];
const thoughts: Array<{ subject: string; description: string }> = [];
let state: string | undefined;
let taskId: string | undefined;
let contextId: string | undefined;
if (event.kind === 'status-update') {
state = event.status?.state;
taskId = event.taskId;
contextId = event.contextId;
const parts: Part[] = event.status?.message?.parts ?? [];
// Extract plain text FIRST (incremental chunks) so A2UI accumulated
// text is added AFTER — backward iteration will prefer A2UI.
const plainText = extractTextFromParts(parts);
if (plainText) {
agentResponses.push({ text: plainText, status: '' });
}
const a2uiGroups = extractA2UIParts(parts);
for (const messages of a2uiGroups) {
parseA2UIMessages(messages, toolApprovals, agentResponses, thoughts);
}
} else if (event.kind === 'task') {
state = event.status?.state;
taskId = event.id;
contextId = event.contextId;
const parts = extractAllParts(event);
const plainText = extractTextFromParts(parts);
if (plainText) {
agentResponses.push({ text: plainText, status: '' });
}
const a2uiGroups = extractA2UIParts(parts);
for (const messages of a2uiGroups) {
parseA2UIMessages(messages, toolApprovals, agentResponses, thoughts);
}
} else if (event.kind === 'message') {
contextId = event.contextId;
taskId = event.taskId;
const parts: Part[] = event.parts ?? [];
const plainText = extractTextFromParts(parts);
if (plainText) {
agentResponses.push({ text: plainText, status: '' });
}
const a2uiGroups = extractA2UIParts(parts);
for (const messages of a2uiGroups) {
parseA2UIMessages(messages, toolApprovals, agentResponses, thoughts);
}
}
// Build text from the last non-empty agent response
let text = '';
for (let i = agentResponses.length - 1; i >= 0; i--) {
if (agentResponses[i].text) {
text = agentResponses[i].text;
break;
}
}
// Add thought summaries
if (thoughts.length > 0) {
const thoughtText = thoughts
.map((t) => `_${t.subject}_: ${t.description}`)
.join('\n');
text = text ? `${thoughtText}\n\n${text}` : thoughtText;
}
return {
text,
toolApprovals: deduplicateToolApprovals(toolApprovals),
state,
taskId,
contextId,
};
}
@@ -0,0 +1,365 @@
/**
* @license
* Copyright 2026 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
/**
* Express routes for the Google Chat bridge webhook.
* Adds a POST /chat/webhook endpoint to the existing Express app.
* Includes JWT verification for Google Chat requests when configured.
*/
import type { Router, Request, Response, NextFunction } from 'express';
import { Router as createRouter } from 'express';
import { OAuth2Client } from 'google-auth-library';
import type { ChatEvent, ChatBridgeConfig, ChatResponse } from './types.js';
import { ChatBridgeHandler } from './handler.js';
import { logger } from '../utils/logger.js';
const CHAT_ISSUER = 'chat@system.gserviceaccount.com';
/**
* Creates middleware that verifies Google Chat JWT tokens.
*
* On Cloud Run (detected via K_SERVICE env var), authentication is handled by
* Cloud Run's IAM layer — only principals with roles/run.invoker can reach the
* container. Cloud Run strips the Authorization header after validation, so our
* middleware cannot re-verify the token. We trust Cloud Run's IAM instead.
*
* When NOT on Cloud Run and projectNumber is set, requests must include a valid
* Bearer token signed by Google Chat with the correct audience.
*
* When neither condition applies, verification is skipped (local testing).
*/
function createAuthMiddleware(
projectNumber: string | undefined,
): (req: Request, res: Response, next: NextFunction) => void {
// On Cloud Run, IAM handles auth — the Authorization header is stripped
// before reaching the container, so we cannot verify it ourselves.
if (process.env['K_SERVICE']) {
logger.info(
'[ChatBridge] Running on Cloud Run — auth delegated to Cloud Run IAM.',
);
return (_req: Request, _res: Response, next: NextFunction) => {
next();
};
}
if (!projectNumber) {
logger.warn(
'[ChatBridge] CHAT_PROJECT_NUMBER not set — JWT verification disabled. ' +
'Set it in production to verify requests come from Google Chat.',
);
return (_req: Request, _res: Response, next: NextFunction) => {
next();
};
}
const authClient = new OAuth2Client();
return (req: Request, res: Response, next: NextFunction) => {
const authHeader = req.headers['authorization'];
if (!authHeader || !authHeader.startsWith('Bearer ')) {
logger.warn('[ChatBridge] Missing or invalid Authorization header');
res.status(401).json({ error: 'Unauthorized: missing Bearer token' });
return;
}
const token = authHeader.substring(7);
// Debug: decode token payload without verification to inspect claims
try {
const payloadB64 = token.split('.')[1];
if (payloadB64) {
const decoded = JSON.parse(
Buffer.from(payloadB64, 'base64').toString(),
);
logger.info(
`[ChatBridge] Token claims: iss=${String(decoded.iss ?? 'none')} ` +
`aud=${String(decoded.aud ?? 'none')} ` +
`email=${String(decoded.email ?? 'none')} ` +
`sub=${String(decoded.sub ?? 'none')}`,
);
}
} catch {
logger.warn('[ChatBridge] Could not decode token for debug logging');
}
authClient
.verifyIdToken({
idToken: token,
audience: projectNumber,
})
.then((ticket) => {
const payload = ticket.getPayload();
if (payload?.iss !== CHAT_ISSUER) {
logger.warn(
`[ChatBridge] Invalid token issuer: ${payload?.iss ?? 'unknown'}`,
);
res.status(403).json({ error: 'Forbidden: invalid token issuer' });
return;
}
next();
})
.catch((err: unknown) => {
const msg = err instanceof Error ? err.message : 'Unknown error';
logger.warn(`[ChatBridge] Token verification failed: ${msg}`);
res.status(401).json({ error: 'Unauthorized: invalid token' });
});
};
}
/** Safely extract a string from an unknown record. */
function str(obj: Record<string, unknown>, key: string): string {
const v = obj[key];
return typeof v === 'string' ? v : '';
}
/** Safely check if a value is a plain object. */
function isObj(v: unknown): v is Record<string, unknown> {
return typeof v === 'object' && v !== null && !Array.isArray(v);
}
/**
* Normalizes a Google Chat event to the legacy ChatEvent format.
* Workspace Add-ons send: {chat: {messagePayload, user, ...}, commonEventObject}
* Legacy format: {type: "MESSAGE", message: {...}, space: {...}, user: {...}}
*/
function normalizeEvent(raw: Record<string, unknown>): ChatEvent | null {
// Already in legacy format
if (typeof raw['type'] === 'string') {
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
return raw as unknown as ChatEvent;
}
// Workspace Add-ons format
const chat = raw['chat'];
if (!isObj(chat)) return null;
const user = isObj(chat['user']) ? chat['user'] : {};
const eventTime = str(chat, 'eventTime');
// Check for card click actions (button clicks) via commonEventObject
const common = raw['commonEventObject'];
if (isObj(common) && typeof common['invokedFunction'] === 'string') {
const invokedFunction = common['invokedFunction'];
const params = isObj(common['parameters']) ? common['parameters'] : {};
// Build action parameters array from commonEventObject.parameters
const actionParams = Object.entries(params)
.filter(([, v]) => typeof v === 'string')
.map(([key, value]) => ({ key, value: String(value) }));
// Extract message/thread/space from chat object
const message = isObj(chat['message']) ? chat['message'] : {};
const thread = isObj(message['thread']) ? message['thread'] : {};
const space = isObj(chat['space'])
? chat['space']
: isObj(message['space'])
? message['space']
: {};
logger.info(
`[ChatBridge] Add-ons CARD_CLICKED: function=${invokedFunction} ` +
`params=${JSON.stringify(params)} thread=${str(thread, 'name')}`,
);
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
return {
type: 'CARD_CLICKED',
eventTime,
message: { ...message, thread, space },
space,
user,
action: {
actionMethodName: invokedFunction,
parameters: actionParams,
},
} as unknown as ChatEvent;
}
// Determine event type from which payload field is present
if (isObj(chat['messagePayload'])) {
const payload = chat['messagePayload'];
const message = isObj(payload['message']) ? payload['message'] : {};
const space = isObj(payload['space'])
? payload['space']
: isObj(message['space'])
? message['space']
: {};
const thread = isObj(message['thread']) ? message['thread'] : {};
logger.info(
`[ChatBridge] Add-ons MESSAGE: text="${str(message, 'text')}" ` +
`space=${str(space, 'name')} thread=${str(thread, 'name')}`,
);
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
return {
type: 'MESSAGE',
eventTime,
message: {
...message,
sender: message['sender'] ?? user,
thread,
space,
},
space,
user,
} as unknown as ChatEvent;
}
if (isObj(chat['addedToSpacePayload'])) {
const payload = chat['addedToSpacePayload'];
const space = isObj(payload['space']) ? payload['space'] : {};
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
return {
type: 'ADDED_TO_SPACE',
eventTime,
space,
user,
} as unknown as ChatEvent;
}
if (isObj(chat['removedFromSpacePayload'])) {
const payload = chat['removedFromSpacePayload'];
const space = isObj(payload['space']) ? payload['space'] : {};
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
return {
type: 'REMOVED_FROM_SPACE',
eventTime,
space,
user,
} as unknown as ChatEvent;
}
logger.warn(
`[ChatBridge] Unknown Add-ons event, chat keys: ${Object.keys(chat).join(',')}`,
);
return null;
}
/**
* Wraps a legacy ChatResponse in the Workspace Add-ons response format.
* Add-ons expects: {hostAppDataAction: {chatDataAction: {createMessageAction: {message}}}}
*/
function wrapAddOnsResponse(response: ChatResponse): Record<string, unknown> {
// Build the message object for the Add-ons format.
// Include thread info so replies appear in the same thread as the user's
// message. Without it, createMessageAction creates a top-level message.
const message: Record<string, unknown> = {};
if (response.text) {
message['text'] = response.text;
}
if (response.cardsV2) {
message['cardsV2'] = response.cardsV2;
}
if (response.thread) {
message['thread'] = response.thread;
}
// For action responses (like CARD_CLICKED acknowledgments), use updateMessageAction
if (response.actionResponse?.type === 'UPDATE_MESSAGE') {
return {
hostAppDataAction: {
chatDataAction: {
updateMessageAction: { message },
},
},
};
}
return {
hostAppDataAction: {
chatDataAction: {
createMessageAction: { message },
},
},
};
}
/**
* Creates Express routes for the Google Chat bridge.
*/
export function createChatBridgeRoutes(config: ChatBridgeConfig): Router {
const router = createRouter();
const handler = new ChatBridgeHandler(config);
const authMiddleware = createAuthMiddleware(config.projectNumber);
// Google Chat sends webhook events as POST requests
router.post(
'/chat/webhook',
authMiddleware,
async (req: Request, res: Response) => {
try {
// eslint-disable-next-line @typescript-eslint/no-unsafe-type-assertion
const rawBody = req.body as Record<string, unknown>;
// Normalize to legacy ChatEvent format. Google Chat HTTP endpoints
// configured as Workspace Add-ons send a different event structure:
// {chat: {messagePayload, user, eventTime}, commonEventObject: {...}}
// We convert to the legacy format our handler expects:
// {type: "MESSAGE", message: {...}, space: {...}, user: {...}}
const event = normalizeEvent(rawBody);
if (!event || !event.type) {
logger.warn(
`[ChatBridge] Could not parse event. Keys: ${Object.keys(rawBody).join(',')}`,
);
res.status(400).json({ error: 'Invalid event: missing type field' });
return;
}
logger.info(`[ChatBridge] Webhook received: type=${event.type}`);
// Detect if the request came in Add-ons format
const isAddOnsFormat = Boolean(rawBody['chat'] && !rawBody['type']);
const response = await handler.handleEvent(event);
// For CARD_CLICKED events, force UPDATE_MESSAGE so the card is
// replaced in-place rather than posting a new message.
if (event.type === 'CARD_CLICKED' && !response.actionResponse) {
response.actionResponse = { type: 'UPDATE_MESSAGE' };
}
if (isAddOnsFormat) {
// If the handler returned an empty response (messages pushed via
// Chat API), return a bare {} so Add-ons doesn't try to create
// an empty message — which causes Google Chat to retry the webhook.
if (!response.text && !response.cardsV2 && !response.actionResponse) {
logger.info(`[ChatBridge] Add-ons response: {} (empty ack)`);
res.json({});
} else {
const addOnsResponse = wrapAddOnsResponse(response);
logger.info(
`[ChatBridge] Add-ons response: ${JSON.stringify(addOnsResponse).substring(0, 200)}`,
);
res.json(addOnsResponse);
}
} else {
res.json(response);
}
} catch (error) {
const errorMsg =
error instanceof Error ? error.message : 'Unknown error';
logger.error(`[ChatBridge] Webhook error: ${errorMsg}`, error);
res.status(500).json({
text: `Internal error: ${errorMsg}`,
});
}
},
);
// Health check endpoint for the chat bridge (no auth required)
router.get('/chat/health', (_req: Request, res: Response) => {
res.json({
status: 'ok',
bridge: 'google-chat',
a2aServerUrl: config.a2aServerUrl,
});
});
return router;
}
@@ -0,0 +1,62 @@
/**
* @license
* Copyright 2026 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
/**
* Standalone Google Chat bridge server.
* Runs independently from the A2A agent server — connects to it
* via the A2A protocol over HTTP. Deploy as a separate Cloud Run
* service for independent scaling.
*/
import express from 'express';
import { createChatBridgeRoutes } from './routes.js';
import { logger } from '../utils/logger.js';
function main() {
const a2aServerUrl = process.env['A2A_SERVER_URL'];
if (!a2aServerUrl) {
logger.error(
'[ChatBridge] A2A_SERVER_URL is required. Set it to the A2A agent server URL.',
);
process.exit(1);
}
const app = express();
app.use(express.json());
const chatRoutes = createChatBridgeRoutes({
a2aServerUrl,
projectNumber: process.env['CHAT_PROJECT_NUMBER'],
debug: process.env['CHAT_BRIDGE_DEBUG'] === 'true',
gcsBucket: process.env['GCS_BUCKET_NAME'],
serviceAccountKeyPath: process.env['CHAT_SA_KEY_PATH'],
});
app.use(chatRoutes);
// Root health check
app.get('/', (_req, res) => {
res.json({
service: 'gemini-chat-bridge',
status: 'ok',
a2aServerUrl,
});
});
const port = Number(process.env['PORT'] || 8080);
const host = process.env['HOST'] || '0.0.0.0';
app.listen(port, host, () => {
logger.info(`[ChatBridge] Server started on http://${host}:${port}`);
logger.info(`[ChatBridge] Connected to A2A agent at ${a2aServerUrl}`);
});
}
process.on('uncaughtException', (error) => {
logger.error('[ChatBridge] Unhandled exception:', error);
process.exit(1);
});
main();
@@ -0,0 +1,240 @@
/**
* @license
* Copyright 2026 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
/**
* Manages mapping between Google Chat threads and A2A sessions.
* Each Google Chat thread maintains a persistent contextId (conversation)
* and a transient taskId (active task within that conversation).
*
* Supports optional GCS persistence so session mappings survive
* Cloud Run instance restarts.
*/
import { v4 as uuidv4 } from 'uuid';
import { logger } from '../utils/logger.js';
export interface PendingToolApproval {
callId: string;
taskId: string;
toolName: string;
}
export interface SessionInfo {
/** A2A contextId - persists for the lifetime of the Chat thread. */
contextId: string;
/** A2A taskId - cleared on terminal states, reused on input-required. */
taskId?: string;
/** Space name for async messaging. */
spaceName: string;
/** Thread name for async messaging. */
threadName: string;
/** Last activity timestamp. */
lastActivity: number;
/** Pending tool approval waiting for text-based response. */
pendingToolApproval?: PendingToolApproval;
/** When true, all tool calls are auto-approved. */
yoloMode?: boolean;
/** When true, an async task is currently processing. */
asyncProcessing?: boolean;
/** When true, session has been cancelled (e.g. by /reset). Signals async processing to stop. */
cancelled?: boolean;
}
/** Serializable subset of SessionInfo for GCS persistence. */
interface PersistedSession {
contextId: string;
taskId?: string;
spaceName: string;
threadName: string;
lastActivity: number;
yoloMode?: boolean;
}
/**
* Session store mapping Google Chat thread names to A2A sessions.
* Optionally backed by GCS for persistence across restarts.
*/
export class SessionStore {
private sessions = new Map<string, SessionInfo>();
private gcsBucket?: string;
private gcsObjectPath = 'chat-bridge/sessions.json';
private dirty = false;
private flushTimer?: ReturnType<typeof setInterval>;
constructor(gcsBucket?: string) {
this.gcsBucket = gcsBucket;
if (gcsBucket) {
// Flush to GCS every 30 seconds if dirty
this.flushTimer = setInterval(() => {
if (this.dirty) {
this.persistToGCS().catch((err) =>
logger.warn(`[ChatBridge] GCS session flush failed:`, err),
);
}
}, 30000);
}
}
/**
* Restores sessions from GCS on startup.
*/
async restore(): Promise<void> {
if (!this.gcsBucket) return;
try {
const { Storage } = await import('@google-cloud/storage');
const storage = new Storage();
const file = storage.bucket(this.gcsBucket).file(this.gcsObjectPath);
const [exists] = await file.exists();
if (!exists) {
logger.info('[ChatBridge] No persisted sessions found in GCS.');
return;
}
const [contents] = await file.download();
const persisted: PersistedSession[] = JSON.parse(contents.toString());
for (const s of persisted) {
this.sessions.set(s.threadName, {
contextId: s.contextId,
taskId: s.taskId,
spaceName: s.spaceName,
threadName: s.threadName,
lastActivity: s.lastActivity,
yoloMode: s.yoloMode,
});
}
logger.info(
`[ChatBridge] Restored ${persisted.length} sessions from GCS.`,
);
} catch (err) {
logger.warn(`[ChatBridge] Could not restore sessions from GCS:`, err);
}
}
/**
* Persists current sessions to GCS.
*/
private async persistToGCS(): Promise<void> {
if (!this.gcsBucket) return;
try {
const { Storage } = await import('@google-cloud/storage');
const storage = new Storage();
const file = storage.bucket(this.gcsBucket).file(this.gcsObjectPath);
const persisted: PersistedSession[] = [];
for (const session of this.sessions.values()) {
persisted.push({
contextId: session.contextId,
taskId: session.taskId,
spaceName: session.spaceName,
threadName: session.threadName,
lastActivity: session.lastActivity,
yoloMode: session.yoloMode,
});
}
await file.save(JSON.stringify(persisted), {
contentType: 'application/json',
});
this.dirty = false;
logger.info(
`[ChatBridge] Persisted ${persisted.length} sessions to GCS.`,
);
} catch (err) {
logger.warn(`[ChatBridge] Failed to persist sessions to GCS:`, err);
}
}
/**
* Gets or creates a session for a Google Chat thread.
*/
getOrCreate(threadName: string, spaceName: string): SessionInfo {
let session = this.sessions.get(threadName);
if (!session) {
session = {
contextId: uuidv4(),
spaceName,
threadName,
lastActivity: Date.now(),
};
this.sessions.set(threadName, session);
this.dirty = true;
logger.info(
`[ChatBridge] New session for thread ${threadName}: contextId=${session.contextId}`,
);
}
session.lastActivity = Date.now();
return session;
}
/**
* Gets an existing session by thread name.
*/
get(threadName: string): SessionInfo | undefined {
return this.sessions.get(threadName);
}
/**
* Updates the taskId for a session.
*/
updateTaskId(threadName: string, taskId: string | undefined): void {
const session = this.sessions.get(threadName);
if (session) {
session.taskId = taskId;
this.dirty = true;
logger.info(
`[ChatBridge] Session ${threadName}: taskId=${taskId ?? 'cleared'}`,
);
}
}
/**
* Removes a session (e.g. when bot is removed from space).
*/
remove(threadName: string): void {
const session = this.sessions.get(threadName);
if (session) {
// Signal any in-flight async processing to stop
session.cancelled = true;
}
this.sessions.delete(threadName);
this.dirty = true;
}
/**
* Cleans up stale sessions older than the given max age (ms).
*/
cleanup(maxAgeMs: number = 24 * 60 * 60 * 1000): void {
const now = Date.now();
for (const [threadName, session] of this.sessions.entries()) {
if (now - session.lastActivity > maxAgeMs) {
this.sessions.delete(threadName);
this.dirty = true;
logger.info(`[ChatBridge] Cleaned up stale session: ${threadName}`);
}
}
}
/**
* Forces an immediate flush to GCS.
*/
async flush(): Promise<void> {
if (this.dirty) {
await this.persistToGCS();
}
}
/**
* Stops the periodic flush timer.
*/
dispose(): void {
if (this.flushTimer) {
clearInterval(this.flushTimer);
this.flushTimer = undefined;
}
}
}
@@ -0,0 +1,143 @@
/**
* @license
* Copyright 2026 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
/**
* Google Chat HTTP endpoint event types.
* @see https://developers.google.com/workspace/chat/api/reference/rest/v1/Event
*/
export interface ChatUser {
name: string;
displayName: string;
type?: 'HUMAN' | 'BOT';
}
export interface ChatThread {
name: string;
threadKey?: string;
}
export interface ChatSpace {
name: string;
type: 'DM' | 'ROOM' | 'SPACE';
displayName?: string;
}
export interface ChatMessage {
name: string;
sender: ChatUser;
createTime: string;
text?: string;
argumentText?: string;
thread: ChatThread;
space: ChatSpace;
cardsV2?: ChatCardV2[];
}
export interface ChatActionParameter {
key: string;
value: string;
}
export interface ChatAction {
actionMethodName: string;
parameters: ChatActionParameter[];
}
export type ChatEventType =
| 'MESSAGE'
| 'CARD_CLICKED'
| 'ADDED_TO_SPACE'
| 'REMOVED_FROM_SPACE';
export interface ChatEvent {
type: ChatEventType;
eventTime: string;
message?: ChatMessage;
space: ChatSpace;
user: ChatUser;
action?: ChatAction;
common?: Record<string, unknown>;
threadKey?: string;
}
// Google Chat Cards V2 response types
export interface ChatCardV2 {
cardId: string;
card: ChatCard;
}
export interface ChatCard {
header?: ChatCardHeader;
sections: ChatCardSection[];
}
export interface ChatCardHeader {
title: string;
subtitle?: string;
imageUrl?: string;
imageType?: 'CIRCLE' | 'SQUARE';
}
export interface ChatCardSection {
header?: string;
widgets: ChatWidget[];
collapsible?: boolean;
uncollapsibleWidgetsCount?: number;
}
export type ChatWidget =
| { textParagraph: { text: string } }
| { decoratedText: ChatDecoratedText }
| { buttonList: { buttons: ChatButton[] } }
| { divider: Record<string, never> };
export interface ChatDecoratedText {
text: string;
topLabel?: string;
bottomLabel?: string;
startIcon?: { knownIcon: string };
wrapText?: boolean;
}
export interface ChatButton {
text: string;
onClick: ChatOnClick;
color?: { red: number; green: number; blue: number; alpha?: number };
disabled?: boolean;
}
export interface ChatOnClick {
action: {
function: string;
parameters: ChatActionParameter[];
};
}
export interface ChatResponse {
text?: string;
cardsV2?: ChatCardV2[];
thread?: { threadKey?: string; name?: string };
actionResponse?: {
type: 'NEW_MESSAGE' | 'UPDATE_MESSAGE' | 'REQUEST_CONFIG';
};
}
// Bridge configuration
export interface ChatBridgeConfig {
/** URL of the A2A server to connect to (e.g. http://localhost:8080) */
a2aServerUrl: string;
/** Google Chat project number for verification (optional) */
projectNumber?: string;
/** Whether to enable debug logging */
debug?: boolean;
/** GCS bucket name for session persistence (optional) */
gcsBucket?: string;
/** Path to service account key for Chat API auth (optional, uses ADC if not set) */
serviceAccountKeyPath?: string;
}
+1 -1
View File
@@ -59,7 +59,7 @@ export async function loadConfig(
const configParams: ConfigParameters = {
sessionId: taskId,
model: PREVIEW_GEMINI_MODEL,
model: process.env['GEMINI_MODEL'] || PREVIEW_GEMINI_MODEL,
embeddingModel: DEFAULT_GEMINI_EMBEDDING_MODEL,
sandbox: undefined, // Sandbox might not be relevant for a server-side agent
targetDir: workspaceDir, // Or a specific directory the agent operates on
+36 -4
View File
@@ -28,6 +28,7 @@ import { commandRegistry } from '../commands/command-registry.js';
import { debugLogger, SimpleExtensionLoader } from '@google/gemini-cli-core';
import type { Command, CommandArgument } from '../commands/types.js';
import { GitService } from '@google/gemini-cli-core';
import { getA2UIAgentExtension } from '../a2ui/a2ui-extension.js';
type CommandResponse = {
name: string;
@@ -46,11 +47,12 @@ const coderAgentCard: AgentCard = {
url: 'https://google.com',
},
protocolVersion: '0.3.0',
version: '0.0.2', // Incremented version
version: '0.1.0', // A2UI-enabled version
capabilities: {
streaming: true,
pushNotifications: false,
pushNotifications: true,
stateTransitionHistory: true,
extensions: [getA2UIAgentExtension()],
},
securitySchemes: undefined,
security: undefined,
@@ -75,7 +77,11 @@ const coderAgentCard: AgentCard = {
};
export function updateCoderAgentCardUrl(port: number) {
coderAgentCard.url = `http://localhost:${port}/`;
// On Cloud Run, use the public service URL so remote clients can reach us
const publicUrl = process.env['CODER_AGENT_PUBLIC_URL'];
coderAgentCard.url = publicUrl
? publicUrl.replace(/\/$/, '') + '/'
: `http://localhost:${port}/`;
}
async function handleExecuteCommand(
@@ -200,6 +206,31 @@ export async function createApp() {
requestStorage.run({ req }, next);
});
// SSE keepalive — sends periodic comment lines to prevent Cloud Run's
// load balancer from closing idle SSE connections during long tool
// executions (npm install, tsc builds, etc.). SSE comments (`: ...`)
// are ignored by conformant parsers per the spec.
expressApp.use((req, res, next) => {
const origFlush = res.flushHeaders;
res.flushHeaders = function (this: express.Response) {
origFlush.call(this);
const ct = this.getHeader('content-type');
if (ct && String(ct).includes('text/event-stream')) {
const timer = setInterval(() => {
if (!res.writableEnded) {
res.write(': keepalive\n\n');
} else {
clearInterval(timer);
}
}, 15_000);
res.on('close', () => clearInterval(timer));
}
};
next();
});
// Google Chat bridge runs as a separate service (src/chat-bridge/server.ts).
// It connects to this A2A server over HTTP.
const appBuilder = new A2AExpressApp(requestHandler);
expressApp = appBuilder.setupRoutes(expressApp, '');
expressApp.use(express.json());
@@ -330,7 +361,8 @@ export async function main() {
const expressApp = await createApp();
const port = Number(process.env['CODER_AGENT_PORT'] || 0);
const server = expressApp.listen(port, 'localhost', () => {
const host = process.env['CODER_AGENT_HOST'] || 'localhost';
const server = expressApp.listen(port, host, () => {
const address = server.address();
let actualPort;
if (process.env['CODER_AGENT_PORT']) {
+60 -2
View File
@@ -18,7 +18,7 @@ import { setTargetDir } from '../config/config.js';
import { getPersistedState, type PersistedTaskMetadata } from '../types.js';
import { v4 as uuidv4 } from 'uuid';
type ObjectType = 'metadata' | 'workspace';
type ObjectType = 'metadata' | 'workspace' | 'conversation';
const getTmpArchiveFilename = (taskId: string): string =>
`task-${taskId}-workspace-${uuidv4()}.tar.gz`;
@@ -224,6 +224,28 @@ export class GCSTaskStore implements TaskStore {
`Workspace directory ${workDir} not found, skipping workspace save for task ${taskId}.`,
);
}
// Save conversation history if present in metadata
const rawHistory = dataToStore?.['_conversationHistory'];
const conversationHistory = Array.isArray(rawHistory)
? rawHistory
: undefined;
if (conversationHistory && conversationHistory.length > 0) {
const conversationObjectPath = this.getObjectPath(
taskId,
'conversation',
);
const historyJson = JSON.stringify(conversationHistory);
const compressedHistory = gzipSync(Buffer.from(historyJson));
const conversationFile = this.storage
.bucket(this.bucketName)
.file(conversationObjectPath);
await conversationFile.save(compressedHistory, {
contentType: 'application/gzip',
});
logger.info(
`Task ${taskId} conversation history saved to GCS: gs://${this.bucketName}/${conversationObjectPath} (${conversationHistory.length} entries)`,
);
}
} catch (error) {
logger.error(`Failed to save task ${taskId} to GCS:`, error);
throw error;
@@ -280,6 +302,29 @@ export class GCSTaskStore implements TaskStore {
logger.info(`Task ${taskId} workspace archive not found in GCS.`);
}
// Restore conversation history if available
const conversationObjectPath = this.getObjectPath(taskId, 'conversation');
const conversationFile = this.storage
.bucket(this.bucketName)
.file(conversationObjectPath);
const [conversationExists] = await conversationFile.exists();
if (conversationExists) {
try {
const [compressedHistory] = await conversationFile.download();
const historyJson = gunzipSync(compressedHistory).toString();
const conversationHistory: unknown[] = JSON.parse(historyJson);
loadedMetadata['_conversationHistory'] = conversationHistory;
logger.info(
`Task ${taskId} conversation history restored from GCS (${conversationHistory.length} entries)`,
);
} catch (historyError) {
logger.warn(
`Task ${taskId} conversation history could not be restored:`,
historyError,
);
}
}
return {
id: taskId,
contextId: loadedMetadata._contextId || uuidv4(),
@@ -300,6 +345,8 @@ export class GCSTaskStore implements TaskStore {
}
export class NoOpTaskStore implements TaskStore {
private cache = new Map<string, SDKTask>();
constructor(private realStore: TaskStore) {}
async save(task: SDKTask): Promise<void> {
@@ -308,9 +355,20 @@ export class NoOpTaskStore implements TaskStore {
}
async load(taskId: string): Promise<SDKTask | undefined> {
const cached = this.cache.get(taskId);
if (cached) {
logger.info(
`[NoOpTaskStore] load called for task ${taskId}, returning cached.`,
);
return cached;
}
logger.info(
`[NoOpTaskStore] load called for task ${taskId}, delegating to real store.`,
);
return this.realStore.load(taskId);
const result = await this.realStore.load(taskId);
if (result) {
this.cache.set(taskId, result);
}
return result;
}
}
+48
View File
@@ -0,0 +1,48 @@
#!/bin/bash
# Test script for the Google Chat bridge webhook endpoint.
# Simulates Google Chat events to verify the bridge works.
#
# Usage: ./test-chat-bridge.sh [PORT]
# Default port: 9090 (for kubectl port-forward)
PORT=${1:-9090}
BASE_URL="http://localhost:${PORT}"
echo "Testing chat bridge at ${BASE_URL}..."
# 1. Test health endpoint
echo -e "\n--- Health Check ---"
curl -s "${BASE_URL}/chat/health" | jq .
# 2. Test ADDED_TO_SPACE event
echo -e "\n--- ADDED_TO_SPACE ---"
curl -s -X POST "${BASE_URL}/chat/webhook" \
-H "Content-Type: application/json" \
-d '{
"type": "ADDED_TO_SPACE",
"eventTime": "2026-01-01T00:00:00Z",
"space": { "name": "spaces/test123", "type": "DM" },
"user": { "name": "users/123", "displayName": "Test User" }
}' | jq .
# 3. Test MESSAGE event
echo -e "\n--- MESSAGE (Hello) ---"
curl -s -X POST "${BASE_URL}/chat/webhook" \
-H "Content-Type: application/json" \
-d '{
"type": "MESSAGE",
"eventTime": "2026-01-01T00:01:00Z",
"message": {
"name": "spaces/test123/messages/msg1",
"sender": { "name": "users/123", "displayName": "Test User" },
"createTime": "2026-01-01T00:01:00Z",
"text": "Hello, write me a python hello world",
"argumentText": "Hello, write me a python hello world",
"thread": { "name": "spaces/test123/threads/thread1" },
"space": { "name": "spaces/test123", "type": "DM" }
},
"space": { "name": "spaces/test123", "type": "DM" },
"user": { "name": "users/123", "displayName": "Test User" }
}' | jq .
echo -e "\nDone."