fix: handle HTTP errors gracefully in gws_bridge token refresh

Instead of crashing with a raw urllib traceback on refresh failure,
print a clean error message and suggest re-running setup.py.

skills/productivity/google-workspace/SKILL.md

@@ -1,7 +1,7 @@
 ---
 name: google-workspace
-description: Gmail, Calendar, Drive, Contacts, Sheets, and Docs integration via Python. Uses OAuth2 with automatic token refresh. No external binaries needed — runs entirely with Google's Python client libraries in the Hermes venv.
-version: 1.0.0
+description: Gmail, Calendar, Drive, Contacts, Sheets, and Docs integration via gws CLI (googleworkspace/cli). Uses OAuth2 with automatic token refresh via bridge script. Requires gws binary.
+version: 2.0.0
 author: Nous Research
 license: MIT
 required_credential_files:
@@ -11,14 +11,25 @@ required_credential_files:
     description: Google OAuth2 client credentials (downloaded from Google Cloud Console)
 metadata:
   hermes:
-    tags: [Google, Gmail, Calendar, Drive, Sheets, Docs, Contacts, Email, OAuth]
+    tags: [Google, Gmail, Calendar, Drive, Sheets, Docs, Contacts, Email, OAuth, gws]
     homepage: https://github.com/NousResearch/hermes-agent
     related_skills: [himalaya]
 ---
 
 # Google Workspace
 
-Gmail, Calendar, Drive, Contacts, Sheets, and Docs — all through Python scripts in this skill. No external binaries to install.
+Gmail, Calendar, Drive, Contacts, Sheets, and Docs — powered by `gws` (Google's official Rust CLI). The skill provides a backward-compatible Python wrapper that handles OAuth token refresh and delegates to `gws`.
+
+## Architecture
+
+```
+google_api.py  →  gws_bridge.py  →  gws CLI
+(argparse compat)  (token refresh)    (Google APIs)
+```
+
+- `setup.py` handles OAuth2 (headless-compatible, works on CLI/Telegram/Discord)
+- `gws_bridge.py` refreshes the Hermes token and injects it into `gws` via `GOOGLE_WORKSPACE_CLI_TOKEN`
+- `google_api.py` provides the same CLI interface as v1 but delegates to `gws`
 
 ## References
 
@@ -27,7 +38,22 @@ Gmail, Calendar, Drive, Contacts, Sheets, and Docs — all through Python script
 ## Scripts
 
 - `scripts/setup.py` — OAuth2 setup (run once to authorize)
-- `scripts/google_api.py` — API wrapper CLI (agent uses this for all operations)
+- `scripts/gws_bridge.py` — Token refresh bridge to gws CLI
+- `scripts/google_api.py` — Backward-compatible API wrapper (delegates to gws)
+
+## Prerequisites
+
+Install `gws`:
+
+```bash
+cargo install google-workspace-cli
+# or via npm (recommended, downloads prebuilt binary):
+npm install -g @googleworkspace/cli
+# or via Homebrew:
+brew install googleworkspace-cli
+```
+
+Verify: `gws --version`
 
 ## First-Time Setup
 
@@ -56,42 +82,29 @@ If it prints `AUTHENTICATED`, skip to Usage — setup is already done.
 
 ### Step 1: Triage — ask the user what they need
 
-Before starting OAuth setup, ask the user TWO questions:
-
 **Question 1: "What Google services do you need? Just email, or also
 Calendar/Drive/Sheets/Docs?"**
 
-- **Email only** → They don't need this skill at all. Use the `himalaya` skill
-  instead — it works with a Gmail App Password (Settings → Security → App
-  Passwords) and takes 2 minutes to set up. No Google Cloud project needed.
-  Load the himalaya skill and follow its setup instructions.
+- **Email only** → Use the `himalaya` skill instead — simpler setup.
+- **Calendar, Drive, Sheets, Docs (or email + these)** → Continue below.
 
-- **Calendar, Drive, Sheets, Docs (or email + these)** → Continue with this
-  skill's OAuth setup below.
+**Partial scopes**: Users can authorize only a subset of services. The setup
+script accepts partial scopes and warns about missing ones.
 
-**Question 2: "Does your Google account use Advanced Protection (hardware
-security keys required to sign in)? If you're not sure, you probably don't
-— it's something you would have explicitly enrolled in."**
+**Question 2: "Does your Google account use Advanced Protection?"**
 
-- **No / Not sure** → Normal setup. Continue below.
-- **Yes** → Their Workspace admin must add the OAuth client ID to the org's
-  allowed apps list before Step 4 will work. Let them know upfront.
+- **No / Not sure** → Normal setup.
+- **Yes** → Workspace admin must add the OAuth client ID to allowed apps first.
 
 ### Step 2: Create OAuth credentials (one-time, ~5 minutes)
 
 Tell the user:
 
-> You need a Google Cloud OAuth client. This is a one-time setup:
->
 > 1. Go to https://console.cloud.google.com/apis/credentials
 > 2. Create a project (or use an existing one)
-> 3. Click "Enable APIs" and enable: Gmail API, Google Calendar API,
->    Google Drive API, Google Sheets API, Google Docs API, People API
-> 4. Go to Credentials → Create Credentials → OAuth 2.0 Client ID
-> 5. Application type: "Desktop app" → Create
-> 6. Click "Download JSON" and tell me the file path
-
-Once they provide the path:
+> 3. Enable the APIs you need (Gmail, Calendar, Drive, Sheets, Docs, People)
+> 4. Credentials → Create Credentials → OAuth 2.0 Client ID → Desktop app
+> 5. Download JSON and tell me the file path
 
 ```bash
 $GSETUP --client-secret /path/to/client_secret.json
@@ -103,20 +116,10 @@ $GSETUP --client-secret /path/to/client_secret.json
 $GSETUP --auth-url
 ```
 
-This prints a URL. **Send the URL to the user** and tell them:
-
-> Open this link in your browser, sign in with your Google account, and
-> authorize access. After authorizing, you'll be redirected to a page that
-> may show an error — that's expected. Copy the ENTIRE URL from your
-> browser's address bar and paste it back to me.
+Send the URL to the user. After authorizing, they paste back the redirect URL or code.
 
 ### Step 4: Exchange the code
 
-The user will paste back either a URL like `http://localhost:1/?code=4/0A...&scope=...`
-or just the code string. Either works. The `--auth-url` step stores a temporary
-pending OAuth session locally so `--auth-code` can complete the PKCE exchange
-later, even on headless systems:
-
 ```bash
 $GSETUP --auth-code "THE_URL_OR_CODE_THE_USER_PASTED"
 ```
@@ -127,18 +130,11 @@ $GSETUP --auth-code "THE_URL_OR_CODE_THE_USER_PASTED"
 $GSETUP --check
 ```
 
-Should print `AUTHENTICATED`. Setup is complete — token refreshes automatically from now on.
-
-### Notes
-
-- Token is stored at `google_token.json` under the active profile's `HERMES_HOME` and auto-refreshes.
-- Pending OAuth session state/verifier are stored temporarily at `google_oauth_pending.json` under the active profile's `HERMES_HOME` until exchange completes.
-- Hermes now refuses to overwrite a full Google Workspace token with a narrower re-auth token missing Gmail scopes, so one profile's partial consent cannot silently break email actions later.
-- To revoke: `$GSETUP --revoke`
+Should print `AUTHENTICATED`. Token refreshes automatically from now on.
 
 ## Usage
 
-All commands go through the API script. Set `GAPI` as a shorthand:
+All commands go through the API script:
 
 ```bash
 HERMES_HOME="${HERMES_HOME:-$HOME/.hermes}"
@@ -153,40 +149,21 @@ GAPI="$PYTHON_BIN $GWORKSPACE_SKILL_DIR/scripts/google_api.py"
 ### Gmail
 
 ```bash
-# Search (returns JSON array with id, from, subject, date, snippet)
 $GAPI gmail search "is:unread" --max 10
-$GAPI gmail search "from:boss@company.com newer_than:1d"
-$GAPI gmail search "has:attachment filename:pdf newer_than:7d"
-
-# Read full message (returns JSON with body text)
 $GAPI gmail get MESSAGE_ID
-
-# Send
 $GAPI gmail send --to user@example.com --subject "Hello" --body "Message text"
-$GAPI gmail send --to user@example.com --subject "Report" --body "<h1>Q4</h1><p>Details...</p>" --html
-
-# Reply (automatically threads and sets In-Reply-To)
+$GAPI gmail send --to user@example.com --subject "Report" --body "<h1>Q4</h1>" --html
 $GAPI gmail reply MESSAGE_ID --body "Thanks, that works for me."
-
-# Labels
 $GAPI gmail labels
 $GAPI gmail modify MESSAGE_ID --add-labels LABEL_ID
-$GAPI gmail modify MESSAGE_ID --remove-labels UNREAD
 ```
 
 ### Calendar
 
 ```bash
-# List events (defaults to next 7 days)
 $GAPI calendar list
-$GAPI calendar list --start 2026-03-01T00:00:00Z --end 2026-03-07T23:59:59Z
-
-# Create event (ISO 8601 with timezone required)
-$GAPI calendar create --summary "Team Standup" --start 2026-03-01T10:00:00-06:00 --end 2026-03-01T10:30:00-06:00
-$GAPI calendar create --summary "Lunch" --start 2026-03-01T12:00:00Z --end 2026-03-01T13:00:00Z --location "Cafe"
-$GAPI calendar create --summary "Review" --start 2026-03-01T14:00:00Z --end 2026-03-01T15:00:00Z --attendees "alice@co.com,bob@co.com"
-
-# Delete event
+$GAPI calendar create --summary "Standup" --start 2026-03-01T10:00:00+01:00 --end 2026-03-01T10:30:00+01:00
+$GAPI calendar create --summary "Review" --start ... --end ... --attendees "alice@co.com,bob@co.com"
 $GAPI calendar delete EVENT_ID
 ```
 
@@ -206,13 +183,8 @@ $GAPI contacts list --max 20
 ### Sheets
 
 ```bash
-# Read
 $GAPI sheets get SHEET_ID "Sheet1!A1:D10"
-
-# Write
 $GAPI sheets update SHEET_ID "Sheet1!A1:B2" --values '[["Name","Score"],["Alice","95"]]'
-
-# Append rows
 $GAPI sheets append SHEET_ID "Sheet1!A:C" --values '[["new","row","data"]]'
 ```
 
@@ -222,37 +194,52 @@ $GAPI sheets append SHEET_ID "Sheet1!A:C" --values '[["new","row","data"]]'
 $GAPI docs get DOC_ID
 ```
 
+### Direct gws access (advanced)
+
+For operations not covered by the wrapper, use `gws_bridge.py` directly:
+
+```bash
+GBRIDGE="$PYTHON_BIN $GWORKSPACE_SKILL_DIR/scripts/gws_bridge.py"
+$GBRIDGE calendar +agenda --today --format table
+$GBRIDGE gmail +triage --labels --format json
+$GBRIDGE drive +upload ./report.pdf
+$GBRIDGE sheets +read --spreadsheet SHEET_ID --range "Sheet1!A1:D10"
+```
+
 ## Output Format
 
-All commands return JSON. Parse with `jq` or read directly. Key fields:
+All commands return JSON via `gws --format json`. Key output shapes:
 
-- **Gmail search**: `[{id, threadId, from, to, subject, date, snippet, labels}]`
-- **Gmail get**: `{id, threadId, from, to, subject, date, labels, body}`
-- **Gmail send/reply**: `{status: "sent", id, threadId}`
-- **Calendar list**: `[{id, summary, start, end, location, description, htmlLink}]`
-- **Calendar create**: `{status: "created", id, summary, htmlLink}`
-- **Drive search**: `[{id, name, mimeType, modifiedTime, webViewLink}]`
-- **Contacts list**: `[{name, emails: [...], phones: [...]}]`
-- **Sheets get**: `[[cell, cell, ...], ...]`
+- **Gmail search/triage**: Array of message summaries (sender, subject, date, snippet)
+- **Gmail get/read**: Message object with headers and body text
+- **Gmail send/reply**: Confirmation with message ID
+- **Calendar list/agenda**: Array of event objects (summary, start, end, location)
+- **Calendar create**: Confirmation with event ID and htmlLink
+- **Drive search**: Array of file objects (id, name, mimeType, webViewLink)
+- **Sheets get/read**: 2D array of cell values
+- **Docs get**: Full document JSON (use `body.content` for text extraction)
+- **Contacts list**: Array of person objects with names, emails, phones
+
+Parse output with `jq` or read JSON directly.
 
 ## Rules
 
-1. **Never send email or create/delete events without confirming with the user first.** Show the draft content and ask for approval.
-2. **Check auth before first use** — run `setup.py --check`. If it fails, guide the user through setup.
-3. **Use the Gmail search syntax reference** for complex queries — load it with `skill_view("google-workspace", file_path="references/gmail-search-syntax.md")`.
-4. **Calendar times must include timezone** — always use ISO 8601 with offset (e.g., `2026-03-01T10:00:00-06:00`) or UTC (`Z`).
-5. **Respect rate limits** — avoid rapid-fire sequential API calls. Batch reads when possible.
+1. **Never send email or create/delete events without confirming with the user first.**
+2. **Check auth before first use** — run `setup.py --check`.
+3. **Use the Gmail search syntax reference** for complex queries.
+4. **Calendar times must include timezone** — ISO 8601 with offset or UTC.
+5. **Respect rate limits** — avoid rapid-fire sequential API calls.
 
 ## Troubleshooting
 
 | Problem | Fix |
 |---------|-----|
-| `NOT_AUTHENTICATED` | Run setup Steps 2-5 above |
-| `REFRESH_FAILED` | Token revoked or expired — redo Steps 3-5 |
-| `HttpError 403: Insufficient Permission` | Missing API scope — `$GSETUP --revoke` then redo Steps 3-5 |
-| `HttpError 403: Access Not Configured` | API not enabled — user needs to enable it in Google Cloud Console |
-| `ModuleNotFoundError` | Run `$GSETUP --install-deps` |
-| Advanced Protection blocks auth | Workspace admin must allowlist the OAuth client ID |
+| `NOT_AUTHENTICATED` | Run setup Steps 2-5 |
+| `REFRESH_FAILED` | Token revoked — redo Steps 3-5 |
+| `gws: command not found` | Install: `npm install -g @googleworkspace/cli` |
+| `HttpError 403` | Missing scope — `$GSETUP --revoke` then redo Steps 3-5 |
+| `HttpError 403: Access Not Configured` | Enable API in Google Cloud Console |
+| Advanced Protection blocks auth | Admin must allowlist the OAuth client ID |
 
 ## Revoking Access
 

skills/productivity/google-workspace/scripts/google_api.py

@@ -1,16 +1,17 @@
 #!/usr/bin/env python3
 """Google Workspace API CLI for Hermes Agent.
 
-A thin CLI wrapper around Google's Python client libraries.
-Authenticates using the token stored by setup.py.
+Thin wrapper that delegates to gws (googleworkspace/cli) via gws_bridge.py.
+Maintains the same CLI interface for backward compatibility with Hermes skills.
 
 Usage:
   python google_api.py gmail search "is:unread" [--max 10]
   python google_api.py gmail get MESSAGE_ID
   python google_api.py gmail send --to user@example.com --subject "Hi" --body "Hello"
   python google_api.py gmail reply MESSAGE_ID --body "Thanks"
-  python google_api.py calendar list [--from DATE] [--to DATE] [--calendar primary]
+  python google_api.py calendar list [--start DATE] [--end DATE] [--calendar primary]
   python google_api.py calendar create --summary "Meeting" --start DATETIME --end DATETIME
+  python google_api.py calendar delete EVENT_ID
   python google_api.py drive search "budget report" [--max 10]
   python google_api.py contacts list [--max 20]
   python google_api.py sheets get SHEET_ID RANGE
@@ -20,386 +21,193 @@ Usage:
 """
 
 import argparse
-import base64
 import json
+import os
+import subprocess
 import sys
-from datetime import datetime, timedelta, timezone
-from email.mime.text import MIMEText
 from pathlib import Path
 
-try:
-    from hermes_constants import display_hermes_home, get_hermes_home
-except ModuleNotFoundError:
-    HERMES_AGENT_ROOT = Path(__file__).resolve().parents[4]
-    if HERMES_AGENT_ROOT.exists():
-        sys.path.insert(0, str(HERMES_AGENT_ROOT))
-    from hermes_constants import display_hermes_home, get_hermes_home
-
-HERMES_HOME = get_hermes_home()
-TOKEN_PATH = HERMES_HOME / "google_token.json"
-
-SCOPES = [
-    "https://www.googleapis.com/auth/gmail.readonly",
-    "https://www.googleapis.com/auth/gmail.send",
-    "https://www.googleapis.com/auth/gmail.modify",
-    "https://www.googleapis.com/auth/calendar",
-    "https://www.googleapis.com/auth/drive.readonly",
-    "https://www.googleapis.com/auth/contacts.readonly",
-    "https://www.googleapis.com/auth/spreadsheets",
-    "https://www.googleapis.com/auth/documents.readonly",
-]
+BRIDGE = Path(__file__).parent / "gws_bridge.py"
+PYTHON = sys.executable
 
 
-def _missing_scopes() -> list[str]:
-    try:
-        payload = json.loads(TOKEN_PATH.read_text())
-    except Exception:
-        return []
-    raw = payload.get("scopes") or payload.get("scope")
-    if not raw:
-        return []
-    granted = {s.strip() for s in (raw.split() if isinstance(raw, str) else raw) if s.strip()}
-    return sorted(scope for scope in SCOPES if scope not in granted)
+def gws(*args: str) -> None:
+    """Call gws via the bridge and exit with its return code."""
+    result = subprocess.run(
+        [PYTHON, str(BRIDGE)] + list(args),
+        env={**os.environ, "HERMES_HOME": os.environ.get("HERMES_HOME", str(Path.home() / ".hermes"))},
+    )
+    sys.exit(result.returncode)
 
 
-def get_credentials():
-    """Load and refresh credentials from token file."""
-    if not TOKEN_PATH.exists():
-        print("Not authenticated. Run the setup script first:", file=sys.stderr)
-        print(f"  python {Path(__file__).parent / 'setup.py'}", file=sys.stderr)
-        sys.exit(1)
-
-    from google.oauth2.credentials import Credentials
-    from google.auth.transport.requests import Request
-
-    creds = Credentials.from_authorized_user_file(str(TOKEN_PATH), SCOPES)
-    if creds.expired and creds.refresh_token:
-        creds.refresh(Request())
-        TOKEN_PATH.write_text(creds.to_json())
-    if not creds.valid:
-        print("Token is invalid. Re-run setup.", file=sys.stderr)
-        sys.exit(1)
-
-    missing_scopes = _missing_scopes()
-    if missing_scopes:
-        print(
-            "Token is valid but missing Google Workspace scopes required by this skill.",
-            file=sys.stderr,
-        )
-        for scope in missing_scopes:
-            print(f"  - {scope}", file=sys.stderr)
-        print(
-            f"Re-run setup.py from the active Hermes profile ({display_hermes_home()}) to restore full access.",
-            file=sys.stderr,
-        )
-        sys.exit(1)
-    return creds
-
-
-def build_service(api, version):
-    from googleapiclient.discovery import build
-    return build(api, version, credentials=get_credentials())
-
-
-# =========================================================================
-# Gmail
-# =========================================================================
+# -- Gmail --
 
 def gmail_search(args):
-    service = build_service("gmail", "v1")
-    results = service.users().messages().list(
-        userId="me", q=args.query, maxResults=args.max
-    ).execute()
-    messages = results.get("messages", [])
-    if not messages:
-        print("No messages found.")
-        return
-
-    output = []
-    for msg_meta in messages:
-        msg = service.users().messages().get(
-            userId="me", id=msg_meta["id"], format="metadata",
-            metadataHeaders=["From", "To", "Subject", "Date"],
-        ).execute()
-        headers = {h["name"]: h["value"] for h in msg.get("payload", {}).get("headers", [])}
-        output.append({
-            "id": msg["id"],
-            "threadId": msg["threadId"],
-            "from": headers.get("From", ""),
-            "to": headers.get("To", ""),
-            "subject": headers.get("Subject", ""),
-            "date": headers.get("Date", ""),
-            "snippet": msg.get("snippet", ""),
-            "labels": msg.get("labelIds", []),
-        })
-    print(json.dumps(output, indent=2, ensure_ascii=False))
-
+    cmd = ["gmail", "+triage", "--query", args.query, "--max", str(args.max), "--format", "json"]
+    gws(*cmd)
 
 def gmail_get(args):
-    service = build_service("gmail", "v1")
-    msg = service.users().messages().get(
-        userId="me", id=args.message_id, format="full"
-    ).execute()
-
-    headers = {h["name"]: h["value"] for h in msg.get("payload", {}).get("headers", [])}
-
-    # Extract body text
-    body = ""
-    payload = msg.get("payload", {})
-    if payload.get("body", {}).get("data"):
-        body = base64.urlsafe_b64decode(payload["body"]["data"]).decode("utf-8", errors="replace")
-    elif payload.get("parts"):
-        for part in payload["parts"]:
-            if part.get("mimeType") == "text/plain" and part.get("body", {}).get("data"):
-                body = base64.urlsafe_b64decode(part["body"]["data"]).decode("utf-8", errors="replace")
-                break
-        if not body:
-            for part in payload["parts"]:
-                if part.get("mimeType") == "text/html" and part.get("body", {}).get("data"):
-                    body = base64.urlsafe_b64decode(part["body"]["data"]).decode("utf-8", errors="replace")
-                    break
-
-    result = {
-        "id": msg["id"],
-        "threadId": msg["threadId"],
-        "from": headers.get("From", ""),
-        "to": headers.get("To", ""),
-        "subject": headers.get("Subject", ""),
-        "date": headers.get("Date", ""),
-        "labels": msg.get("labelIds", []),
-        "body": body,
-    }
-    print(json.dumps(result, indent=2, ensure_ascii=False))
-
+    gws("gmail", "+read", "--id", args.message_id, "--headers", "--format", "json")
 
 def gmail_send(args):
-    service = build_service("gmail", "v1")
-    message = MIMEText(args.body, "html" if args.html else "plain")
-    message["to"] = args.to
-    message["subject"] = args.subject
+    cmd = ["gmail", "+send", "--to", args.to, "--subject", args.subject, "--body", args.body, "--format", "json"]
     if args.cc:
-        message["cc"] = args.cc
-
-    raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
-    body = {"raw": raw}
-
-    if args.thread_id:
-        body["threadId"] = args.thread_id
-
-    result = service.users().messages().send(userId="me", body=body).execute()
-    print(json.dumps({"status": "sent", "id": result["id"], "threadId": result.get("threadId", "")}, indent=2))
-
+        cmd += ["--cc", args.cc]
+    if args.html:
+        cmd.append("--html")
+    gws(*cmd)
 
 def gmail_reply(args):
-    service = build_service("gmail", "v1")
-    # Fetch original to get thread ID and headers
-    original = service.users().messages().get(
-        userId="me", id=args.message_id, format="metadata",
-        metadataHeaders=["From", "Subject", "Message-ID"],
-    ).execute()
-    headers = {h["name"]: h["value"] for h in original.get("payload", {}).get("headers", [])}
-
-    subject = headers.get("Subject", "")
-    if not subject.startswith("Re:"):
-        subject = f"Re: {subject}"
-
-    message = MIMEText(args.body)
-    message["to"] = headers.get("From", "")
-    message["subject"] = subject
-    if headers.get("Message-ID"):
-        message["In-Reply-To"] = headers["Message-ID"]
-        message["References"] = headers["Message-ID"]
-
-    raw = base64.urlsafe_b64encode(message.as_bytes()).decode()
-    body = {"raw": raw, "threadId": original["threadId"]}
-
-    result = service.users().messages().send(userId="me", body=body).execute()
-    print(json.dumps({"status": "sent", "id": result["id"], "threadId": result.get("threadId", "")}, indent=2))
-
+    gws("gmail", "+reply", "--message-id", args.message_id, "--body", args.body, "--format", "json")
 
 def gmail_labels(args):
-    service = build_service("gmail", "v1")
-    results = service.users().labels().list(userId="me").execute()
-    labels = [{"id": l["id"], "name": l["name"], "type": l.get("type", "")} for l in results.get("labels", [])]
-    print(json.dumps(labels, indent=2))
-
+    gws("gmail", "users", "labels", "list", "--params", json.dumps({"userId": "me"}), "--format", "json")
 
 def gmail_modify(args):
-    service = build_service("gmail", "v1")
     body = {}
     if args.add_labels:
         body["addLabelIds"] = args.add_labels.split(",")
     if args.remove_labels:
         body["removeLabelIds"] = args.remove_labels.split(",")
-    result = service.users().messages().modify(userId="me", id=args.message_id, body=body).execute()
-    print(json.dumps({"id": result["id"], "labels": result.get("labelIds", [])}, indent=2))
+    gws(
+        "gmail", "users", "messages", "modify",
+        "--params", json.dumps({"userId": "me", "id": args.message_id}),
+        "--json", json.dumps(body),
+        "--format", "json",
+    )
 
 
-# =========================================================================
-# Calendar
-# =========================================================================
+# -- Calendar --
 
 def calendar_list(args):
-    service = build_service("calendar", "v3")
-    now = datetime.now(timezone.utc)
-    time_min = args.start or now.isoformat()
-    time_max = args.end or (now + timedelta(days=7)).isoformat()
-
-    # Ensure timezone info
-    for val in [time_min, time_max]:
-        if "T" in val and "Z" not in val and "+" not in val and "-" not in val[11:]:
-            val += "Z"
-
-    results = service.events().list(
-        calendarId=args.calendar, timeMin=time_min, timeMax=time_max,
-        maxResults=args.max, singleEvents=True, orderBy="startTime",
-    ).execute()
-
-    events = []
-    for e in results.get("items", []):
-        events.append({
-            "id": e["id"],
-            "summary": e.get("summary", "(no title)"),
-            "start": e.get("start", {}).get("dateTime", e.get("start", {}).get("date", "")),
-            "end": e.get("end", {}).get("dateTime", e.get("end", {}).get("date", "")),
-            "location": e.get("location", ""),
-            "description": e.get("description", ""),
-            "status": e.get("status", ""),
-            "htmlLink": e.get("htmlLink", ""),
-        })
-    print(json.dumps(events, indent=2, ensure_ascii=False))
-
+    if args.start or args.end:
+        # Specific date range — use raw Calendar API for precise timeMin/timeMax
+        from datetime import datetime, timedelta, timezone as tz
+        now = datetime.now(tz.utc)
+        time_min = args.start or now.isoformat()
+        time_max = args.end or (now + timedelta(days=7)).isoformat()
+        gws(
+            "calendar", "events", "list",
+            "--params", json.dumps({
+                "calendarId": args.calendar,
+                "timeMin": time_min,
+                "timeMax": time_max,
+                "maxResults": args.max,
+                "singleEvents": True,
+                "orderBy": "startTime",
+            }),
+            "--format", "json",
+        )
+    else:
+        # No date range — use +agenda helper (defaults to 7 days)
+        cmd = ["calendar", "+agenda", "--days", "7", "--format", "json"]
+        if args.calendar != "primary":
+            cmd += ["--calendar", args.calendar]
+        gws(*cmd)
 
 def calendar_create(args):
-    service = build_service("calendar", "v3")
-    event = {
-        "summary": args.summary,
-        "start": {"dateTime": args.start},
-        "end": {"dateTime": args.end},
-    }
+    cmd = [
+        "calendar", "+insert",
+        "--summary", args.summary,
+        "--start", args.start,
+        "--end", args.end,
+        "--format", "json",
+    ]
     if args.location:
-        event["location"] = args.location
+        cmd += ["--location", args.location]
     if args.description:
-        event["description"] = args.description
+        cmd += ["--description", args.description]
     if args.attendees:
-        event["attendees"] = [{"email": e.strip()} for e in args.attendees.split(",")]
-
-    result = service.events().insert(calendarId=args.calendar, body=event).execute()
-    print(json.dumps({
-        "status": "created",
-        "id": result["id"],
-        "summary": result.get("summary", ""),
-        "htmlLink": result.get("htmlLink", ""),
-    }, indent=2))
-
+        for email in args.attendees.split(","):
+            cmd += ["--attendee", email.strip()]
+    if args.calendar != "primary":
+        cmd += ["--calendar", args.calendar]
+    gws(*cmd)
 
 def calendar_delete(args):
-    service = build_service("calendar", "v3")
-    service.events().delete(calendarId=args.calendar, eventId=args.event_id).execute()
-    print(json.dumps({"status": "deleted", "eventId": args.event_id}))
+    gws(
+        "calendar", "events", "delete",
+        "--params", json.dumps({"calendarId": args.calendar, "eventId": args.event_id}),
+        "--format", "json",
+    )
 
 
-# =========================================================================
-# Drive
-# =========================================================================
+# -- Drive --
 
 def drive_search(args):
-    service = build_service("drive", "v3")
-    query = f"fullText contains '{args.query}'" if not args.raw_query else args.query
-    results = service.files().list(
-        q=query, pageSize=args.max, fields="files(id, name, mimeType, modifiedTime, webViewLink)",
-    ).execute()
-    files = results.get("files", [])
-    print(json.dumps(files, indent=2, ensure_ascii=False))
+    query = args.query if args.raw_query else f"fullText contains '{args.query}'"
+    gws(
+        "drive", "files", "list",
+        "--params", json.dumps({
+            "q": query,
+            "pageSize": args.max,
+            "fields": "files(id,name,mimeType,modifiedTime,webViewLink)",
+        }),
+        "--format", "json",
+    )
 
 
-# =========================================================================
-# Contacts
-# =========================================================================
+# -- Contacts --
 
 def contacts_list(args):
-    service = build_service("people", "v1")
-    results = service.people().connections().list(
-        resourceName="people/me",
-        pageSize=args.max,
-        personFields="names,emailAddresses,phoneNumbers",
-    ).execute()
-    contacts = []
-    for person in results.get("connections", []):
-        names = person.get("names", [{}])
-        emails = person.get("emailAddresses", [])
-        phones = person.get("phoneNumbers", [])
-        contacts.append({
-            "name": names[0].get("displayName", "") if names else "",
-            "emails": [e.get("value", "") for e in emails],
-            "phones": [p.get("value", "") for p in phones],
-        })
-    print(json.dumps(contacts, indent=2, ensure_ascii=False))
+    gws(
+        "people", "people", "connections", "list",
+        "--params", json.dumps({
+            "resourceName": "people/me",
+            "pageSize": args.max,
+            "personFields": "names,emailAddresses,phoneNumbers",
+        }),
+        "--format", "json",
+    )
 
 
-# =========================================================================
-# Sheets
-# =========================================================================
+# -- Sheets --
 
 def sheets_get(args):
-    service = build_service("sheets", "v4")
-    result = service.spreadsheets().values().get(
-        spreadsheetId=args.sheet_id, range=args.range,
-    ).execute()
-    print(json.dumps(result.get("values", []), indent=2, ensure_ascii=False))
-
+    gws(
+        "sheets", "+read",
+        "--spreadsheet", args.sheet_id,
+        "--range", args.range,
+        "--format", "json",
+    )
 
 def sheets_update(args):
-    service = build_service("sheets", "v4")
     values = json.loads(args.values)
-    body = {"values": values}
-    result = service.spreadsheets().values().update(
-        spreadsheetId=args.sheet_id, range=args.range,
-        valueInputOption="USER_ENTERED", body=body,
-    ).execute()
-    print(json.dumps({"updatedCells": result.get("updatedCells", 0), "updatedRange": result.get("updatedRange", "")}, indent=2))
-
+    gws(
+        "sheets", "spreadsheets", "values", "update",
+        "--params", json.dumps({
+            "spreadsheetId": args.sheet_id,
+            "range": args.range,
+            "valueInputOption": "USER_ENTERED",
+        }),
+        "--json", json.dumps({"values": values}),
+        "--format", "json",
+    )
 
 def sheets_append(args):
-    service = build_service("sheets", "v4")
     values = json.loads(args.values)
-    body = {"values": values}
-    result = service.spreadsheets().values().append(
-        spreadsheetId=args.sheet_id, range=args.range,
-        valueInputOption="USER_ENTERED", insertDataOption="INSERT_ROWS", body=body,
-    ).execute()
-    print(json.dumps({"updatedCells": result.get("updates", {}).get("updatedCells", 0)}, indent=2))
+    gws(
+        "sheets", "+append",
+        "--spreadsheet", args.sheet_id,
+        "--json-values", json.dumps(values),
+        "--format", "json",
+    )
 
 
-# =========================================================================
-# Docs
-# =========================================================================
+# -- Docs --
 
 def docs_get(args):
-    service = build_service("docs", "v1")
-    doc = service.documents().get(documentId=args.doc_id).execute()
-    # Extract plain text from the document structure
-    text_parts = []
-    for element in doc.get("body", {}).get("content", []):
-        paragraph = element.get("paragraph", {})
-        for pe in paragraph.get("elements", []):
-            text_run = pe.get("textRun", {})
-            if text_run.get("content"):
-                text_parts.append(text_run["content"])
-    result = {
-        "title": doc.get("title", ""),
-        "documentId": doc.get("documentId", ""),
-        "body": "".join(text_parts),
-    }
-    print(json.dumps(result, indent=2, ensure_ascii=False))
+    gws(
+        "docs", "documents", "get",
+        "--params", json.dumps({"documentId": args.doc_id}),
+        "--format", "json",
+    )
 
 
-# =========================================================================
-# CLI parser
-# =========================================================================
+# -- CLI parser (backward-compatible interface) --
 
 def main():
-    parser = argparse.ArgumentParser(description="Google Workspace API for Hermes Agent")
+    parser = argparse.ArgumentParser(description="Google Workspace API for Hermes Agent (gws backend)")
     sub = parser.add_subparsers(dest="service", required=True)
 
     # --- Gmail ---
@@ -421,7 +229,7 @@ def main():
     p.add_argument("--body", required=True)
     p.add_argument("--cc", default="")
     p.add_argument("--html", action="store_true", help="Send body as HTML")
-    p.add_argument("--thread-id", default="", help="Thread ID for threading")
+    p.add_argument("--thread-id", default="", help="Thread ID (unused with gws, kept for compat)")
     p.set_defaults(func=gmail_send)
 
     p = gmail_sub.add_parser("reply")

skills/productivity/google-workspace/scripts/gws_bridge.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+"""Bridge between Hermes OAuth token and gws CLI.
+
+Refreshes the token if expired, then executes gws with the valid access token.
+"""
+import json
+import os
+import subprocess
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+def get_hermes_home() -> Path:
+    return Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes"))
+
+
+def get_token_path() -> Path:
+    return get_hermes_home() / "google_token.json"
+
+
+def refresh_token(token_data: dict) -> dict:
+    """Refresh the access token using the refresh token."""
+    import urllib.error
+    import urllib.parse
+    import urllib.request
+
+    params = urllib.parse.urlencode({
+        "client_id": token_data["client_id"],
+        "client_secret": token_data["client_secret"],
+        "refresh_token": token_data["refresh_token"],
+        "grant_type": "refresh_token",
+    }).encode()
+
+    req = urllib.request.Request(token_data["token_uri"], data=params)
+    try:
+        with urllib.request.urlopen(req) as resp:
+            result = json.loads(resp.read())
+    except urllib.error.HTTPError as e:
+        body = e.read().decode("utf-8", errors="replace")
+        print(f"ERROR: Token refresh failed (HTTP {e.code}): {body}", file=sys.stderr)
+        print("Re-run setup.py to re-authenticate.", file=sys.stderr)
+        sys.exit(1)
+
+    token_data["token"] = result["access_token"]
+    token_data["expiry"] = datetime.fromtimestamp(
+        datetime.now(timezone.utc).timestamp() + result["expires_in"],
+        tz=timezone.utc,
+    ).isoformat()
+
+    get_token_path().write_text(json.dumps(token_data, indent=2))
+    return token_data
+
+
+def get_valid_token() -> str:
+    """Return a valid access token, refreshing if needed."""
+    token_path = get_token_path()
+    if not token_path.exists():
+        print("ERROR: No Google token found. Run setup.py --auth-url first.", file=sys.stderr)
+        sys.exit(1)
+
+    token_data = json.loads(token_path.read_text())
+
+    expiry = token_data.get("expiry", "")
+    if expiry:
+        exp_dt = datetime.fromisoformat(expiry.replace("Z", "+00:00"))
+        now = datetime.now(timezone.utc)
+        if now >= exp_dt:
+            token_data = refresh_token(token_data)
+
+    return token_data["token"]
+
+
+def main():
+    """Refresh token if needed, then exec gws with remaining args."""
+    if len(sys.argv) < 2:
+        print("Usage: gws_bridge.py <gws args...>", file=sys.stderr)
+        sys.exit(1)
+
+    access_token = get_valid_token()
+    env = os.environ.copy()
+    env["GOOGLE_WORKSPACE_CLI_TOKEN"] = access_token
+
+    result = subprocess.run(["gws"] + sys.argv[1:], env=env)
+    sys.exit(result.returncode)
+
+
+if __name__ == "__main__":
+    main()

skills/productivity/google-workspace/scripts/setup.py

@@ -23,6 +23,7 @@ Agent workflow:
 
 import argparse
 import json
+import os
 import subprocess
 import sys
 from pathlib import Path
@@ -128,7 +129,11 @@ def check_auth():
     from google.auth.transport.requests import Request
 
     try:
-        creds = Credentials.from_authorized_user_file(str(TOKEN_PATH), SCOPES)
+        # Don't pass scopes — user may have authorized only a subset.
+        # Passing scopes forces google-auth to validate them on refresh,
+        # which fails with invalid_scope if the token has fewer scopes
+        # than requested.
+        creds = Credentials.from_authorized_user_file(str(TOKEN_PATH))
     except Exception as e:
         print(f"TOKEN_CORRUPT: {e}")
         return False
@@ -137,8 +142,9 @@ def check_auth():
     if creds.valid:
         missing_scopes = _missing_scopes_from_payload(payload)
         if missing_scopes:
-            print(f"AUTH_SCOPE_MISMATCH: {_format_missing_scopes(missing_scopes)}")
-            return False
+            print(f"AUTHENTICATED (partial): Token valid but missing {len(missing_scopes)} scopes:")
+            for s in missing_scopes:
+                print(f"  - {s}")
         print(f"AUTHENTICATED: Token valid at {TOKEN_PATH}")
         return True
 
@@ -148,8 +154,9 @@ def check_auth():
             TOKEN_PATH.write_text(creds.to_json())
             missing_scopes = _missing_scopes_from_payload(_load_token_payload(TOKEN_PATH))
             if missing_scopes:
-                print(f"AUTH_SCOPE_MISMATCH: {_format_missing_scopes(missing_scopes)}")
-                return False
+                print(f"AUTHENTICATED (partial): Token refreshed but missing {len(missing_scopes)} scopes:")
+                for s in missing_scopes:
+                    print(f"  - {s}")
             print(f"AUTHENTICATED: Token refreshed at {TOKEN_PATH}")
             return True
         except Exception as e:
@@ -272,16 +279,33 @@ def exchange_auth_code(code: str):
 
     _ensure_deps()
     from google_auth_oauthlib.flow import Flow
+    from urllib.parse import parse_qs, urlparse
+
+    # Extract granted scopes from the callback URL if present
+    if returned_state and "scope" in parse_qs(urlparse(code).query if isinstance(code, str) and code.startswith("http") else {}):
+        granted_scopes = parse_qs(urlparse(code).query)["scope"][0].split()
+    else:
+        # Try to extract from code_or_url parameter
+        if isinstance(code, str) and code.startswith("http"):
+            params = parse_qs(urlparse(code).query)
+            if "scope" in params:
+                granted_scopes = params["scope"][0].split()
+            else:
+                granted_scopes = SCOPES
+        else:
+            granted_scopes = SCOPES
 
     flow = Flow.from_client_secrets_file(
         str(CLIENT_SECRET_PATH),
-        scopes=SCOPES,
+        scopes=granted_scopes,
         redirect_uri=pending_auth.get("redirect_uri", REDIRECT_URI),
         state=pending_auth["state"],
         code_verifier=pending_auth["code_verifier"],
     )
 
     try:
+        # Accept partial scopes — user may deselect some permissions in the consent screen
+        os.environ["OAUTHLIB_RELAX_TOKEN_SCOPE"] = "1"
         flow.fetch_token(code=code)
     except Exception as e:
         print(f"ERROR: Token exchange failed: {e}")
@@ -290,11 +314,21 @@ def exchange_auth_code(code: str):
 
     creds = flow.credentials
     token_payload = json.loads(creds.to_json())
+
+    # Store only the scopes actually granted by the user, not what was requested.
+    # creds.to_json() writes the requested scopes, which causes refresh to fail
+    # with invalid_scope if the user only authorized a subset.
+    actually_granted = list(creds.granted_scopes or []) if hasattr(creds, "granted_scopes") and creds.granted_scopes else []
+    if actually_granted:
+        token_payload["scopes"] = actually_granted
+    elif granted_scopes != SCOPES:
+        # granted_scopes was extracted from the callback URL
+        token_payload["scopes"] = granted_scopes
+
     missing_scopes = _missing_scopes_from_payload(token_payload)
     if missing_scopes:
-        print(f"ERROR: Refusing to save incomplete Google Workspace token. {_format_missing_scopes(missing_scopes)}")
-        print(f"Existing token at {TOKEN_PATH} was left unchanged.")
-        sys.exit(1)
+        print(f"WARNING: Token missing some Google Workspace scopes: {', '.join(missing_scopes)}")
+        print("Some services may not be available.")
 
     TOKEN_PATH.write_text(json.dumps(token_payload, indent=2))
     PENDING_AUTH_PATH.unlink(missing_ok=True)

tests/skills/test_google_oauth_setup.py

@@ -211,14 +211,15 @@ class TestExchangeAuthCode:
         assert setup_module.PENDING_AUTH_PATH.exists()
         assert not setup_module.TOKEN_PATH.exists()
 
-    def test_refuses_to_overwrite_existing_token_with_narrower_scopes(self, setup_module, capsys):
+    def test_accepts_narrower_scopes_with_warning(self, setup_module, capsys):
+        """Partial scopes are accepted with a warning (gws migration: v2.0)."""
         setup_module.PENDING_AUTH_PATH.write_text(
             json.dumps({"state": "saved-state", "code_verifier": "saved-verifier"})
         )
-        setup_module.TOKEN_PATH.write_text(json.dumps({"token": "existing-token", "scopes": setup_module.SCOPES}))
+        setup_module.TOKEN_PATH.write_text(json.dumps({"token": "***", "scopes": setup_module.SCOPES}))
         FakeFlow.credentials_payload = {
-            "token": "narrow-token",
-            "refresh_token": "refresh-token",
+            "token": "***",
+            "refresh_token": "***",
             "token_uri": "https://oauth2.googleapis.com/token",
             "client_id": "client-id",
             "client_secret": "client-secret",
@@ -228,10 +229,12 @@ class TestExchangeAuthCode:
             ],
         }
 
-        with pytest.raises(SystemExit):
-            setup_module.exchange_auth_code("4/test-auth-code")
+        setup_module.exchange_auth_code("4/test-auth-code")
 
         out = capsys.readouterr().out
-        assert "refusing to save incomplete google workspace token" in out.lower()
-        assert json.loads(setup_module.TOKEN_PATH.read_text())["token"] == "existing-token"
-        assert setup_module.PENDING_AUTH_PATH.exists()
+        assert "warning" in out.lower()
+        assert "missing" in out.lower()
+        # Token is saved (partial scopes accepted)
+        assert setup_module.TOKEN_PATH.exists()
+        # Pending auth is cleaned up
+        assert not setup_module.PENDING_AUTH_PATH.exists()

tests/skills/test_google_workspace_api.py

@@ -1,117 +1,175 @@
-"""Regression tests for Google Workspace API credential validation."""
+"""Tests for Google Workspace gws bridge and CLI wrapper."""
 
 import importlib.util
 import json
+import os
+import subprocess
 import sys
 import types
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
+from unittest.mock import MagicMock, patch
 
 import pytest
 
 
-SCRIPT_PATH = (
+BRIDGE_PATH = (
+    Path(__file__).resolve().parents[2]
+    / "skills/productivity/google-workspace/scripts/gws_bridge.py"
+)
+API_PATH = (
     Path(__file__).resolve().parents[2]
     / "skills/productivity/google-workspace/scripts/google_api.py"
 )
 
 
-class FakeAuthorizedCredentials:
-    def __init__(self, *, valid=True, expired=False, refresh_token="refresh-token"):
-        self.valid = valid
-        self.expired = expired
-        self.refresh_token = refresh_token
-        self.refresh_calls = 0
-
-    def refresh(self, _request):
-        self.refresh_calls += 1
-        self.valid = True
-        self.expired = False
-
-    def to_json(self):
-        return json.dumps({
-            "token": "refreshed-token",
-            "refresh_token": self.refresh_token,
-            "token_uri": "https://oauth2.googleapis.com/token",
-            "client_id": "client-id",
-            "client_secret": "client-secret",
-            "scopes": [
-                "https://www.googleapis.com/auth/gmail.readonly",
-                "https://www.googleapis.com/auth/gmail.send",
-                "https://www.googleapis.com/auth/gmail.modify",
-                "https://www.googleapis.com/auth/calendar",
-                "https://www.googleapis.com/auth/drive.readonly",
-                "https://www.googleapis.com/auth/contacts.readonly",
-                "https://www.googleapis.com/auth/spreadsheets",
-                "https://www.googleapis.com/auth/documents.readonly",
-            ],
-        })
-
-
-class FakeCredentialsFactory:
-    creds = FakeAuthorizedCredentials()
-
-    @classmethod
-    def from_authorized_user_file(cls, _path, _scopes):
-        return cls.creds
-
-
 @pytest.fixture
-def google_api_module(monkeypatch, tmp_path):
-    google_module = types.ModuleType("google")
-    oauth2_module = types.ModuleType("google.oauth2")
-    credentials_module = types.ModuleType("google.oauth2.credentials")
-    credentials_module.Credentials = FakeCredentialsFactory
-    auth_module = types.ModuleType("google.auth")
-    transport_module = types.ModuleType("google.auth.transport")
-    requests_module = types.ModuleType("google.auth.transport.requests")
-    requests_module.Request = object
+def bridge_module(monkeypatch, tmp_path):
+    hermes_home = tmp_path / ".hermes"
+    hermes_home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
 
-    monkeypatch.setitem(sys.modules, "google", google_module)
-    monkeypatch.setitem(sys.modules, "google.oauth2", oauth2_module)
-    monkeypatch.setitem(sys.modules, "google.oauth2.credentials", credentials_module)
-    monkeypatch.setitem(sys.modules, "google.auth", auth_module)
-    monkeypatch.setitem(sys.modules, "google.auth.transport", transport_module)
-    monkeypatch.setitem(sys.modules, "google.auth.transport.requests", requests_module)
-
-    spec = importlib.util.spec_from_file_location("google_workspace_api_test", SCRIPT_PATH)
+    spec = importlib.util.spec_from_file_location("gws_bridge_test", BRIDGE_PATH)
     module = importlib.util.module_from_spec(spec)
     assert spec.loader is not None
     spec.loader.exec_module(module)
-
-    monkeypatch.setattr(module, "TOKEN_PATH", tmp_path / "google_token.json")
     return module
 
 
-def _write_token(path: Path, scopes):
-    path.write_text(json.dumps({
-        "token": "access-token",
-        "refresh_token": "refresh-token",
+@pytest.fixture
+def api_module(monkeypatch, tmp_path):
+    hermes_home = tmp_path / ".hermes"
+    hermes_home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+    spec = importlib.util.spec_from_file_location("gws_api_test", API_PATH)
+    module = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+def _write_token(path: Path, *, token="ya29.test", expiry=None, **extra):
+    data = {
+        "token": token,
+        "refresh_token": "1//refresh",
+        "client_id": "123.apps.googleusercontent.com",
+        "client_secret": "secret",
         "token_uri": "https://oauth2.googleapis.com/token",
-        "client_id": "client-id",
-        "client_secret": "client-secret",
-        "scopes": scopes,
-    }))
+        **extra,
+    }
+    if expiry is not None:
+        data["expiry"] = expiry
+    path.write_text(json.dumps(data))
 
 
-def test_get_credentials_rejects_missing_scopes(google_api_module, capsys):
-    FakeCredentialsFactory.creds = FakeAuthorizedCredentials(valid=True)
-    _write_token(google_api_module.TOKEN_PATH, [
-        "https://www.googleapis.com/auth/drive.readonly",
-        "https://www.googleapis.com/auth/spreadsheets",
-    ])
+def test_bridge_returns_valid_token(bridge_module, tmp_path):
+    """Non-expired token is returned without refresh."""
+    future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
+    token_path = bridge_module.get_token_path()
+    _write_token(token_path, token="ya29.valid", expiry=future)
 
+    result = bridge_module.get_valid_token()
+    assert result == "ya29.valid"
+
+
+def test_bridge_refreshes_expired_token(bridge_module, tmp_path):
+    """Expired token triggers a refresh via token_uri."""
+    past = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
+    token_path = bridge_module.get_token_path()
+    _write_token(token_path, token="ya29.old", expiry=past)
+
+    mock_resp = MagicMock()
+    mock_resp.read.return_value = json.dumps({
+        "access_token": "ya29.refreshed",
+        "expires_in": 3600,
+    }).encode()
+    mock_resp.__enter__ = lambda s: s
+    mock_resp.__exit__ = MagicMock(return_value=False)
+
+    with patch("urllib.request.urlopen", return_value=mock_resp):
+        result = bridge_module.get_valid_token()
+
+    assert result == "ya29.refreshed"
+    # Verify persisted
+    saved = json.loads(token_path.read_text())
+    assert saved["token"] == "ya29.refreshed"
+
+
+def test_bridge_exits_on_missing_token(bridge_module):
+    """Missing token file causes exit with code 1."""
     with pytest.raises(SystemExit):
-        google_api_module.get_credentials()
-
-    err = capsys.readouterr().err
-    assert "missing google workspace scopes" in err.lower()
-    assert "gmail.send" in err
+        bridge_module.get_valid_token()
 
 
-def test_get_credentials_accepts_full_scope_token(google_api_module):
-    FakeCredentialsFactory.creds = FakeAuthorizedCredentials(valid=True)
-    _write_token(google_api_module.TOKEN_PATH, list(google_api_module.SCOPES))
+def test_bridge_main_injects_token_env(bridge_module, tmp_path):
+    """main() sets GOOGLE_WORKSPACE_CLI_TOKEN in subprocess env."""
+    future = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
+    token_path = bridge_module.get_token_path()
+    _write_token(token_path, token="ya29.injected", expiry=future)
 
-    creds = google_api_module.get_credentials()
+    captured = {}
 
-    assert creds is FakeCredentialsFactory.creds
+    def capture_run(cmd, **kwargs):
+        captured["cmd"] = cmd
+        captured["env"] = kwargs.get("env", {})
+        return MagicMock(returncode=0)
+
+    with patch.object(sys, "argv", ["gws_bridge.py", "gmail", "+triage"]):
+        with patch.object(subprocess, "run", side_effect=capture_run):
+            with pytest.raises(SystemExit):
+                bridge_module.main()
+
+    assert captured["env"]["GOOGLE_WORKSPACE_CLI_TOKEN"] == "ya29.injected"
+    assert captured["cmd"] == ["gws", "gmail", "+triage"]
+
+
+def test_api_calendar_list_uses_agenda_by_default(api_module):
+    """calendar list without dates uses +agenda helper."""
+    captured = {}
+
+    def capture_run(cmd, **kwargs):
+        captured["cmd"] = cmd
+        return MagicMock(returncode=0)
+
+    args = api_module.argparse.Namespace(
+        start="", end="", max=25, calendar="primary", func=api_module.calendar_list,
+    )
+
+    with patch.object(subprocess, "run", side_effect=capture_run):
+        with pytest.raises(SystemExit):
+            api_module.calendar_list(args)
+
+    gws_args = captured["cmd"][2:]  # skip python + bridge path
+    assert "calendar" in gws_args
+    assert "+agenda" in gws_args
+    assert "--days" in gws_args
+
+
+def test_api_calendar_list_respects_date_range(api_module):
+    """calendar list with --start/--end uses raw events list API."""
+    captured = {}
+
+    def capture_run(cmd, **kwargs):
+        captured["cmd"] = cmd
+        return MagicMock(returncode=0)
+
+    args = api_module.argparse.Namespace(
+        start="2026-04-01T00:00:00Z",
+        end="2026-04-07T23:59:59Z",
+        max=25,
+        calendar="primary",
+        func=api_module.calendar_list,
+    )
+
+    with patch.object(subprocess, "run", side_effect=capture_run):
+        with pytest.raises(SystemExit):
+            api_module.calendar_list(args)
+
+    gws_args = captured["cmd"][2:]
+    assert "events" in gws_args
+    assert "list" in gws_args
+    params_idx = gws_args.index("--params")
+    params = json.loads(gws_args[params_idx + 1])
+    assert params["timeMin"] == "2026-04-01T00:00:00Z"
+    assert params["timeMax"] == "2026-04-07T23:59:59Z"