Building a Business Automation Agent with Antigravity × Google Workspace API — Complete Gmail, Drive, Sheets & Docs Integration

# requirements.txt
google-auth==2.29.0
google-auth-oauthlib==1.2.0
google-auth-httplib2==0.2.0
google-api-python-client==2.125.0
google-generativeai==0.5.3
 
# auth.py
import os
import json
from pathlib import Path
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
 
# Declare ALL required scopes upfront — adding scopes later requires re-authentication
SCOPES = [
    "https://www.googleapis.com/auth/gmail.modify",
    "https://www.googleapis.com/auth/drive",
    "https://www.googleapis.com/auth/spreadsheets",
    "https://www.googleapis.com/auth/documents",
]
 
TOKEN_PATH = Path("token.json")
CREDENTIALS_PATH = Path("credentials.json")  # Downloaded from Google Cloud Console
 
def get_credentials() -> Credentials:
    """
    Obtain valid OAuth 2.0 credentials.
    Reuses token.json when available; refreshes silently when expired.
    Only launches browser flow when no valid credentials exist.
    """
    creds = None
    
    if TOKEN_PATH.exists():
        creds = Credentials.from_authorized_user_file(str(TOKEN_PATH), SCOPES)
    
    # Refresh silently if expired (access tokens expire after 1 hour)
    if creds and creds.expired and creds.refresh_token:
        try:
            creds.refresh(Request())
        except Exception as e:
            print(f"Token refresh failed: {e} — Re-authentication required")
            creds = None
    
    # Only trigger browser flow if no valid credentials
    if not creds or not creds.valid:
        if not CREDENTIALS_PATH.exists():
            raise FileNotFoundError(
                "credentials.json not found. "
                "Download your OAuth 2.0 client ID from Google Cloud Console."
            )
        flow = InstalledAppFlow.from_client_secrets_file(str(CREDENTIALS_PATH), SCOPES)
        creds = flow.run_local_server(port=0)
        TOKEN_PATH.write_text(creds.to_json())
        print(f"✅ Authentication successful. Saved to {TOKEN_PATH}")
    
    return creds
 
def build_services() -> dict:
    """Build all four Workspace API service objects in one call."""
    creds = get_credentials()
    return {
        "gmail": build("gmail", "v1", credentials=creds),
        "drive": build("drive", "v3", credentials=creds),
        "sheets": build("sheets", "v4", credentials=creds),
        "docs": build("docs", "v1", credentials=creds),
    }

import base64
from email.mime.text import MIMEText
from typing import Optional
import google.generativeai as genai
from googleapiclient.errors import HttpError
 
class GmailAgent:
    """Analyzes incoming emails with Gemini and handles replies automatically."""
    
    # Confidence threshold for automatic sending (below this → save as Draft)
    AUTO_REPLY_THRESHOLD = 0.85
    
    def __init__(self, services: dict, gemini_api_key: str):
        self.gmail = services["gmail"]
        genai.configure(api_key=gemini_api_key)
        self.model = genai.GenerativeModel("gemini-2.0-flash")
    
    def fetch_unread_emails(
        self,
        query: str = "is:unread is:inbox newer_than:1d",
        max_results: int = 10
    ) -> list[dict]:
        """
        Fetch unread emails matching the query filter.
        Default query limits to today's inbox to stay within API quota.
        """
        try:
            result = self.gmail.users().messages().list(
                userId="me",
                q=query,
                maxResults=max_results
            ).execute()
            
            messages = result.get("messages", [])
            if not messages:
                return []
            
            emails = []
            for msg in messages:
                detail = self.gmail.users().messages().get(
                    userId="me",
                    id=msg["id"],
                    format="full"
                ).execute()
                
                headers = {h["name"]: h["value"] for h in detail["payload"]["headers"]}
                body = self._extract_body(detail["payload"])
                
                emails.append({
                    "id": msg["id"],
                    "from": headers.get("From", ""),
                    "subject": headers.get("Subject", ""),
                    "body": body[:3000],  # Cap at 3000 chars to stay within token limits
                    "thread_id": detail["threadId"],
                })
            
            return emails
            
        except HttpError as e:
            print(f"❌ Gmail fetch error: {e}")
            return []
    
    def _extract_body(self, payload: dict) -> str:
        """Extract plain text body from email payload."""
        body = ""
        if "parts" in payload:
            for part in payload["parts"]:
                if part["mimeType"] == "text/plain":
                    data = part["body"].get("data", "")
                    body = base64.urlsafe_b64decode(data).decode("utf-8", errors="ignore")
                    break
        elif "body" in payload:
            data = payload["body"].get("data", "")
            if data:
                body = base64.urlsafe_b64decode(data).decode("utf-8", errors="ignore")
        return body
    
    def analyze_and_reply(self, email: dict) -> dict:
        """
        Uses Gemini to analyze email intent and generate a contextual reply.
        Returns: {"action": "send"|"draft"|"skip", "reply": str, "confidence": float, "category": str}
        """
        prompt = f"""
You are an email assistant. Analyze the following email and respond in JSON format only (no code blocks):
 
From: {email['from']}
Subject: {email['subject']}
Body:
{email['body']}
 
Response format:
{{
  "category": "inquiry|support|spam|notification|other",
  "confidence": number between 0.0 and 1.0,
  "should_reply": true or false,
  "reply_draft": "reply text in English. Empty string if should_reply is false",
  "reason": "one-sentence explanation of your decision"
}}
"""
        try:
            response = self.model.generate_content(prompt)
            import json
            result = json.loads(response.text.strip())
            
            # Determine action based on confidence
            if result["should_reply"] and result["confidence"] >= self.AUTO_REPLY_THRESHOLD:
                action = "send"
            elif result["should_reply"]:
                action = "draft"  # Save for human review
            else:
                action = "skip"
            
            return {
                "action": action,
                "reply": result.get("reply_draft", ""),
                "confidence": result["confidence"],
                "category": result["category"],
                "reason": result.get("reason", ""),
            }
            
        except Exception as e:
            print(f"❌ Gemini API error: {e}")
            # Always fail safe: never auto-send on error
            return {"action": "draft", "reply": "", "confidence": 0.0, "category": "error", "reason": str(e)}
    
    def send_or_save_draft(self, original_email: dict, analysis: dict) -> Optional[str]:
        """Send reply or save as Draft based on analysis result."""
        if analysis["action"] == "skip" or not analysis["reply"]:
            print(f"  ↳ Skipped (category: {analysis['category']})")
            return None
        
        message = MIMEText(analysis["reply"], "plain", "utf-8")
        message["to"] = original_email["from"]
        message["subject"] = f"Re: {original_email['subject']}"
        message["In-Reply-To"] = original_email["id"]
        
        raw = base64.urlsafe_b64encode(message.as_bytes()).decode("utf-8")
        body = {"raw": raw, "threadId": original_email["thread_id"]}
        
        try:
            if analysis["action"] == "send":
                sent = self.gmail.users().messages().send(userId="me", body=body).execute()
                print(f"  ✅ Sent (confidence: {analysis['confidence']:.2f})")
                return sent["id"]
            else:
                draft = self.gmail.users().drafts().create(
                    userId="me", body={"message": body}
                ).execute()
                print(f"  📝 Saved as Draft (confidence: {analysis['confidence']:.2f} — needs review)")
                return draft["id"]
        except HttpError as e:
            print(f"  ❌ Send error: {e}")
            return None

from googleapiclient.errors import HttpError
import os
 
class DocumentAgent:
    """Generates Google Docs from templates using Drive API and Gemini."""
    
    def __init__(self, services: dict, gemini_api_key: str):
        self.drive = services["drive"]
        self.docs = services["docs"]
        genai.configure(api_key=gemini_api_key)
        self.model = genai.GenerativeModel("gemini-2.0-flash")
    
    def create_from_template(
        self,
        template_id: str,
        replacements: dict[str, str],
        output_folder_id: str,
        title: str
    ) -> str:
        """
        Copy a Google Doc template and substitute all placeholders.
        
        Args:
            template_id: ID of the template document
            replacements: {"{{VARIABLE}}": "replacement value"}
            output_folder_id: Destination folder ID
            title: Name for the new document
        Returns:
            ID of the created document
        """
        try:
            copied = self.drive.files().copy(
                fileId=template_id,
                body={"name": title, "parents": [output_folder_id]}
            ).execute()
            doc_id = copied["id"]
            print(f"  📄 Template copied: {title}")
        except HttpError as e:
            raise RuntimeError(f"Failed to copy template: {e}")
        
        requests = [
            {
                "replaceAllText": {
                    "containsText": {"text": placeholder, "matchCase": True},
                    "replaceText": value
                }
            }
            for placeholder, value in replacements.items()
        ]
        
        if requests:
            try:
                self.docs.documents().batchUpdate(
                    documentId=doc_id,
                    body={"requests": requests}
                ).execute()
                print(f"  ✅ Replaced {len(requests)} variables")
            except HttpError as e:
                # Keep the file even if replacement fails — partial output is better than nothing
                print(f"  ⚠️ Replacement error: {e}")
        
        return doc_id
    
    def generate_release_notes(
        self,
        app_name: str,
        version: str,
        changes: list[str]
    ) -> str:
        """
        Generate App Store / Google Play release notes with Gemini,
        then create a Google Doc with the result.
        Returns: URL of the created document
        """
        prompt = f"""
Write App Store release notes for the following update.
Keep it user-friendly, avoid technical jargon, under 200 characters.
 
App: {app_name}
Version: {version}
Changes:
{chr(10).join(f'- {c}' for c in changes)}
"""
        response = self.model.generate_content(prompt)
        release_notes = response.text.strip()
        
        replacements = {
            "{{APP_NAME}}": app_name,
            "{{VERSION}}": version,
            "{{RELEASE_DATE}}": "2026-04-20",
            "{{RELEASE_NOTES}}": release_notes,
            "{{RAW_CHANGES}}": "\n".join(f"• {c}" for c in changes),
        }
        
        template_id = os.environ.get("RELEASE_NOTE_TEMPLATE_ID", "")
        folder_id = os.environ.get("RELEASE_NOTES_FOLDER_ID", "")
        
        if not template_id or not folder_id:
            # Fallback: create a fresh document without a template
            doc = self.docs.documents().create(
                body={"title": f"{app_name} v{version} Release Notes"}
            ).execute()
            doc_id = doc["documentId"]
            self.docs.documents().batchUpdate(
                documentId=doc_id,
                body={"requests": [{
                    "insertText": {
                        "location": {"index": 1},
                        "text": f"{app_name} v{version}\n\n{release_notes}\n\nChanges:\n{chr(10).join(f'• {c}' for c in changes)}"
                    }
                }]}
            ).execute()
        else:
            doc_id = self.create_from_template(
                template_id=template_id,
                replacements=replacements,
                output_folder_id=folder_id,
                title=f"{app_name} v{version} Release Notes"
            )
        
        url = f"https://docs.google.com/document/d/{doc_id}/edit"
        print(f"  ✅ Release notes created: {url}")
        return url

from datetime import datetime, timezone, timedelta
from typing import Any
 
class SheetsAgent:
    """Appends daily metrics to Sheets and generates weekly AI-written summaries."""
    
    JST = timezone(timedelta(hours=9))
    
    def __init__(self, services: dict, gemini_api_key: str):
        self.sheets = services["sheets"]
        genai.configure(api_key=gemini_api_key)
        self.model = genai.GenerativeModel("gemini-2.0-flash")
    
    def append_daily_metrics(
        self,
        spreadsheet_id: str,
        sheet_name: str,
        metrics: dict[str, Any]
    ) -> None:
        """Append a row of daily metrics to the specified sheet."""
        today = datetime.now(self.JST).strftime("%Y-%m-%d")
        row = [today] + list(metrics.values())
        
        try:
            self.sheets.spreadsheets().values().append(
                spreadsheetId=spreadsheet_id,
                range=f"{sheet_name}\!A:Z",
                valueInputOption="USER_ENTERED",
                insertDataOption="INSERT_ROWS",
                body={"values": [row]}
            ).execute()
            print(f"  ✅ Appended {today} ({len(row)-1} metrics)")
        except HttpError as e:
            print(f"  ❌ Sheet write error: {e}")
            raise
    
    def generate_weekly_summary(
        self,
        spreadsheet_id: str,
        data_sheet: str,
        summary_sheet: str
    ) -> str:
        """
        Read the last 7 rows of data, generate a Gemini-written weekly summary,
        and write it to the summary sheet.
        Returns: Generated insight text
        """
        try:
            result = self.sheets.spreadsheets().values().get(
                spreadsheetId=spreadsheet_id,
                range=f"{data_sheet}\!A:Z"
            ).execute()
            all_rows = result.get("values", [])
        except HttpError as e:
            print(f"  ❌ Data read error: {e}")
            return ""
        
        if len(all_rows) < 2:
            print("  ⚠️ Insufficient data (need at least 1 row after header)")
            return ""
        
        header = all_rows[0]
        recent_rows = all_rows[-7:]
        
        data_text = "\n".join(
            ", ".join(f"{h}: {v}" for h, v in zip(header, row))
            for row in recent_rows
        )
        
        prompt = f"""
Analyze the following 7-day app performance data and provide:
1. This week's highlight (best result)
2. A trend worth watching
3. One specific action to try next week
 
Keep it under 250 words and be direct and practical.
 
Data:
{data_text}
"""
        response = self.model.generate_content(prompt)
        insight = response.text.strip()
        
        today = datetime.now(self.JST).strftime("%Y-%m-%d")
        try:
            self.sheets.spreadsheets().values().append(
                spreadsheetId=spreadsheet_id,
                range=f"{summary_sheet}\!A:B",
                valueInputOption="USER_ENTERED",
                insertDataOption="INSERT_ROWS",
                body={"values": [[today, insight]]}
            ).execute()
            print(f"  ✅ Weekly summary written to {summary_sheet}")
        except HttpError as e:
            print(f"  ❌ Summary write error: {e}")
        
        return insight

# Business Automation Agents
 
## Gmail Agent
Role: Monitor inbox and handle incoming email automatically
Triggers:
  - Scheduled poll every 30 minutes
  - subject_keywords: ["inquiry", "support", "help"]
Actions:
  - fetch_unread_emails(query="is:unread is:inbox newer_than:1d")
  - analyze_and_reply(email)
  - send_or_save_draft(email, analysis)
Error handling:
  - API error → retry 3 times with 30s backoff
  - Gemini timeout → save as draft, log to error.log
 
## Document Agent
Role: Generate release notes and reports automatically
Triggers:
  - Git tag push (via GitHub webhook)
  - New file in monitored Drive folder
Actions:
  - generate_release_notes(app_name, version, changes)
Dependencies: Gmail Agent (send completion notification)
 
## Sheets Agent
Role: Daily metric collection and weekly reporting
Triggers:
  - Daily at 09:00 JST
  - Weekly Monday at 08:00 JST
Actions:
  - append_daily_metrics(spreadsheet_id, "Daily", metrics)
  - generate_weekly_summary(spreadsheet_id, "Daily", "Weekly Summary")

def setup_push_notification(self, pubsub_topic: str) -> None:
    """
    Configure Gmail to push new message notifications to a Pub/Sub topic.
    Eliminates polling — you only process emails when they arrive.
    Note: This setting expires after 7 days and must be renewed.
    """
    try:
        self.gmail.users().watch(
            userId="me",
            body={
                "topicName": pubsub_topic,
                "labelIds": ["INBOX"],
                "labelFilterAction": "include"
            }
        ).execute()
        print(f"✅ Push notifications enabled: {pubsub_topic}")
        print("⚠️ Renew this setting every 7 days — it expires automatically.")
    except HttpError as e:
        print(f"❌ Push notification setup failed: {e}")

def refresh_if_needed(creds: Credentials) -> Credentials:
    """Silently refresh the access token if expired. Call at the top of each batch loop."""
    if creds.expired and creds.refresh_token:
        creds.refresh(Request())
        TOKEN_PATH.write_text(creds.to_json())
    return creds

# .github/workflows/workspace-automation.yml
name: Google Workspace Automation
 
on:
  schedule:
    - cron: "0 0 * * *"   # Daily at 09:00 JST (UTC 00:00)
    - cron: "0 23 * * 0"  # Weekly Monday 08:00 JST (UTC Sunday 23:00)
  workflow_dispatch:      # Allow manual trigger
 
jobs:
  daily-automation:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: "pip"
      
      - run: pip install -r requirements.txt
      
      - name: Restore OAuth token from secrets
        env:
          GOOGLE_TOKEN_JSON: ${{ secrets.GOOGLE_TOKEN_JSON }}
        run: echo "$GOOGLE_TOKEN_JSON" > token.json
      
      - name: Run automation
        env:
          GOOGLE_CREDENTIALS_JSON: ${{ secrets.GOOGLE_CREDENTIALS_JSON }}
          GEMINI_API_KEY: ${{ secrets.GEMINI_API_KEY }}
          SPREADSHEET_ID: ${{ secrets.SPREADSHEET_ID }}
        run: |
          echo "$GOOGLE_CREDENTIALS_JSON" > credentials.json
          python scripts/run_automation.py
      
      - name: Write back refreshed token to secrets
        # Handles the case where the access token was refreshed during the run
        env:
          GH_TOKEN: ${{ secrets.GH_TOKEN }}
          REPO: ${{ github.repository }}
        run: |
          NEW_TOKEN=$(cat token.json)
          gh secret set GOOGLE_TOKEN_JSON --body "$NEW_TOKEN" --repo "$REPO"

# scripts/run_automation.py
import os
import sys
from pathlib import Path
 
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))
 
from auth import build_services, refresh_if_needed, get_credentials
from agents.gmail_agent import GmailAgent
from agents.document_agent import DocumentAgent
from agents.sheets_agent import SheetsAgent
from datetime import datetime, timezone, timedelta
 
def main():
    JST = timezone(timedelta(hours=9))
    now = datetime.now(JST)
    print(f"\n🤖 Workspace Automation — {now.strftime('%Y-%m-%d %H:%M JST')}")
    print("=" * 60)
    
    # Initialize services
    try:
        creds = get_credentials()
        services = build_services()
    except Exception as e:
        print(f"❌ Authentication failed: {e}")
        sys.exit(1)
    
    GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "")
    SPREADSHEET_ID = os.environ.get("SPREADSHEET_ID", "")
    
    if not GEMINI_API_KEY:
        print("❌ GEMINI_API_KEY is not set")
        sys.exit(1)
    
    results = {
        "emails_processed": 0,
        "emails_sent": 0,
        "emails_drafted": 0,
        "emails_skipped": 0,
        "errors": [],
    }
    
    # --- Gmail Agent ---
    print("\n📧 Gmail Agent")
    print("-" * 40)
    try:
        creds = refresh_if_needed(creds)
        gmail_agent = GmailAgent(services, GEMINI_API_KEY)
        emails = gmail_agent.fetch_unread_emails(max_results=20)
        
        print(f"Found {len(emails)} unread email(s)")
        
        for email in emails:
            print(f"\n  From: {email['from'][:50]}")
            print(f"  Subject: {email['subject'][:60]}")
            
            analysis = gmail_agent.analyze_and_reply(email)
            message_id = gmail_agent.send_or_save_draft(email, analysis)
            
            results["emails_processed"] += 1
            if analysis["action"] == "send":
                results["emails_sent"] += 1
            elif analysis["action"] == "draft":
                results["emails_drafted"] += 1
            else:
                results["emails_skipped"] += 1
                
    except Exception as e:
        error_msg = f"Gmail agent error: {e}"
        print(f"❌ {error_msg}")
        results["errors"].append(error_msg)
    
    # --- Sheets Agent (daily metrics) ---
    print("\n📊 Sheets Agent — Daily Metrics")
    print("-" * 40)
    if SPREADSHEET_ID:
        try:
            creds = refresh_if_needed(creds)
            sheets_agent = SheetsAgent(services, GEMINI_API_KEY)
            
            # Collect your metrics here (replace with your actual data sources)
            metrics = {
                "downloads": fetch_app_downloads(),   # Implement per your data source
                "revenue_jpy": fetch_admob_revenue(),
                "active_users": fetch_dau(),
                "ratings": fetch_average_rating(),
            }
            
            sheets_agent.append_daily_metrics(SPREADSHEET_ID, "Daily", metrics)
            
            # Run weekly summary on Mondays
            if now.weekday() == 0:
                print("\n📝 Generating weekly summary...")
                summary = sheets_agent.generate_weekly_summary(
                    SPREADSHEET_ID, "Daily", "Weekly Summary"
                )
                print(f"\nInsight: {summary[:200]}...")
                
        except Exception as e:
            error_msg = f"Sheets agent error: {e}"
            print(f"❌ {error_msg}")
            results["errors"].append(error_msg)
    else:
        print("⏭️ Skipped (SPREADSHEET_ID not configured)")
    
    # --- Final Summary ---
    print("\n" + "=" * 60)
    print("✅ Automation Complete")
    print(f"  Emails: {results['emails_processed']} processed, "
          f"{results['emails_sent']} sent, "
          f"{results['emails_drafted']} drafted, "
          f"{results['emails_skipped']} skipped")
    if results["errors"]:
        print(f"  Errors: {len(results['errors'])}")
        for err in results["errors"]:
            print(f"    - {err}")
    print("=" * 60)
 
def fetch_app_downloads() -> int:
    """
    Placeholder: implement to pull download data from App Store Connect API,
    Google Play Developer API, or your analytics provider.
    """
    return 0
 
def fetch_admob_revenue() -> float:
    """
    Placeholder: implement to pull AdMob estimated revenue from
    Google Ad Manager API or a manually maintained source.
    """
    return 0.0
 
def fetch_dau() -> int:
    """Placeholder: fetch daily active users from Firebase Analytics."""
    return 0
 
def fetch_average_rating() -> float:
    """Placeholder: fetch current app store rating."""
    return 0.0
 
if __name__ == "__main__":
    main()

def route_by_label(self, email: dict) -> str:
    """
    Pre-classify emails by subject keywords before Gemini analysis.
    This saves API calls and improves routing accuracy for predictable patterns.
    Returns: 'support' | 'billing' | 'partnership' | 'generic'
    """
    subject = email["subject"].lower()
    
    routing_rules = {
        "support": ["bug", "crash", "error", "help", "not working", "broken"],
        "billing": ["payment", "refund", "invoice", "charge", "subscription"],
        "partnership": ["collab", "partnership", "sponsor", "business"],
    }
    
    for category, keywords in routing_rules.items():
        if any(kw in subject for kw in keywords):
            return category
    
    return "generic"

def watch_for_trigger(
    self,
    spreadsheet_id: str,
    sheet_name: str,
    trigger_column_index: int,
    trigger_value: str
) -> list[int]:
    """
    Scan a spreadsheet for rows where a specific column matches trigger_value.
    Returns row indices that need document generation.
    Useful for: "generate invoice when status = 'approved'"
    """
    result = self.sheets.spreadsheets().values().get(
        spreadsheetId=spreadsheet_id,
        range=f"{sheet_name}\!A:Z"
    ).execute()
    
    rows = result.get("values", [])
    triggered_rows = []
    
    for i, row in enumerate(rows[1:], start=2):  # Skip header
        if len(row) > trigger_column_index:
            cell_value = row[trigger_column_index].strip()
            if cell_value == trigger_value:
                triggered_rows.append(i)
    
    return triggered_rows

import time
from functools import wraps
 
def with_retry(max_retries: int = 3, base_delay: float = 30.0):
    """
    Decorator that adds exponential backoff retry logic to API calls.
    Handles 429 (quota exceeded) and 503 (service unavailable) gracefully.
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except HttpError as e:
                    if e.resp.status in (429, 503) and attempt < max_retries - 1:
                        delay = base_delay * (2 ** attempt)  # Exponential backoff
                        print(f"  ⚠️ Rate limit hit. Waiting {delay:.0f}s before retry {attempt + 2}/{max_retries}...")
                        time.sleep(delay)
                    else:
                        raise
            return None
        return wrapper
    return decorator
 
# Usage example:
class GmailAgentWithRetry(GmailAgent):
    @with_retry(max_retries=3, base_delay=30.0)
    def send_or_save_draft(self, original_email: dict, analysis: dict):
        return super().send_or_save_draft(original_email, analysis)

def validate_reply_content(reply: str) -> bool:
    """
    Basic sanity check before auto-sending a Gemini-generated reply.
    Returns False if the reply looks broken or suspicious.
    """
    # Too short to be meaningful
    if len(reply.strip()) < 20:
        return False
    
    # Contains placeholder text that wasn't replaced
    if "{{" in reply or "}}" in reply:
        return False
    
    # Looks like raw JSON was leaked
    if reply.strip().startswith("{") and reply.strip().endswith("}"):
        return False
    
    # Contains markdown that would look broken in plain email
    suspicious_patterns = ["```", "###", "**"]
    if any(p in reply for p in suspicious_patterns):
        return False
    
    return True