Add research digest: POST /research sends Gemini-powered email on any topic

README.md

@@ -1,6 +1,6 @@
 # Email Assistant
 
-Watches your Gmail inbox, uses Gemini to detect schedulable events and to draft replies, then creates Google Calendar events and Gmail drafts automatically. Also generates a daily morning standup email draft addressed to your key stakeholders, and keeps a shareable availability calendar in sync with your busy time. Runs on Google Cloud Run, triggered by Cloud Scheduler.
+Watches your Gmail inbox, uses Gemini to detect schedulable events and to draft replies, then creates Google Calendar events and Gmail drafts automatically. Also generates a daily morning standup email draft addressed to your key stakeholders, keeps a shareable availability calendar in sync with your busy time, and sends recurring research digest emails on any topic you configure using Gemini with live Google Search grounding. Runs on Google Cloud Run, triggered by Cloud Scheduler.
 
 ## How it works
 
@@ -39,6 +39,15 @@ The endpoint responds with `{"draft_url": "https://mail.google.com/mail/#drafts/
 
 Share the availability calendar with consulting clients (read-only, free/busy) so they can see when you're open without seeing what you're doing. The endpoint responds with a summary like `{"busy_blocks": 14, "created": 3, "updated": 1, "deleted": 2, "skipped_free": 5, "window_days": 60}`.
 
+### Research digest (configurable schedule)
+
+1. Cloud Scheduler sends a `POST /research` request on your configured schedule
+2. Gemini researches your configured topic using live Google Search grounding
+3. The summary is sent directly to your Gmail inbox
+4. An atomic GCS marker under `research_runs/` reserves each topic and date before generation, so overlapping Scheduler retries cannot send the same digest twice. Changing `RESEARCH_TOPIC` creates a different marker and sends a fresh digest immediately
+
+The endpoint responds with `{"sent": true, "topic": "...", "date": "YYYY-MM-DD"}`, or `{"skipped": true, "reason": "already_sent_today"}` if today's digest for this topic already went out.
+
 ## AI Engineering
 
 ### LLM call design
@@ -100,14 +109,16 @@ Several layers prevent runaway billing:
 Cloud Scheduler (*/5 * * * *)     → POST /                  → email processing pipeline
 Cloud Scheduler (0 8 * * 1-5)    → POST /standup           → daily standup draft
 Cloud Scheduler (*/30 * * * *)   → POST /availability-sync → mirror busy time to availability calendar
+Cloud Scheduler (your schedule)  → POST /research          → research digest email
 
 Cloud Run (Flask)
-  ├── Gmail API      (read emails, apply labels, create drafts)
-  ├── Gemini API     (extract event details, draft replies, compose standup)
+  ├── Gmail API      (read emails, apply labels, create drafts, send emails)
+  ├── Gemini API     (extract event details, draft replies, compose standup,
+  │                   research topics with Google Search grounding)
   ├── Calendar API   (create events, read schedule, mirror busy blocks)
   ├── Tasks API      (read tasks for standup; create action items when ENABLE_ACTION_ITEMS=true)
   └── GCS            (token.json, processed_ids.json, user_context.json,
-                       last_standup_date.json)
+                       last_standup_date.json, last_research.json)
 ```
 
 ## Setup
@@ -250,6 +261,22 @@ gcloud scheduler jobs create http availability-sync \
   --http-method=POST \
   --oidc-service-account-email=cloud-scheduler-invoker@YOUR_PROJECT.iam.gserviceaccount.com \
   --location=us-central1
+
+# Research digest: runs daily at 7 am (adjust schedule and topic as desired)
+gcloud scheduler jobs create http research-digest \
+  --schedule="0 7 * * *" \
+  --uri="$SERVICE_URL/research" \
+  --http-method=POST \
+  --oidc-service-account-email=cloud-scheduler-invoker@YOUR_PROJECT.iam.gserviceaccount.com \
+  --location=us-central1 \
+  --time-zone="America/Los_Angeles"
+```
+
+Set the research topic as an environment variable on the Cloud Run service:
+```bash
+gcloud run services update assistant \
+  --region us-central1 \
+  --update-env-vars RESEARCH_TOPIC="AI and LLM news this week"
 ```
 
 For the availability sync, first create a dedicated Google Calendar (in Google
@@ -307,7 +334,7 @@ curl -X POST http://localhost:8080/availability-sync  # mirror busy time
 | Variable | Description |
 |---|---|
 | `GEMINI_API_KEY` | API key from [aistudio.google.com](https://aistudio.google.com) |
-| `GCS_BUCKET_NAME` | GCS bucket for `token.json`, `processed_ids.json`, `user_context.json`, `last_standup_date.json` |
+| `GCS_BUCKET_NAME` | GCS bucket for `token.json`, `processed_ids.json`, `user_context.json`, `last_standup_date.json`, and `research_runs/` markers |
 | `CALENDAR_ID` | Calendar to read/write events (`primary` = default) |
 | `TASKS_LIST_ID` | Google Tasks list for action items (`@default` = My Tasks). Only used when `ENABLE_ACTION_ITEMS` is enabled |
 | `ENABLE_ACTION_ITEMS` | Extract action items and create Google Tasks (default `true`). Set to `false` to disable and reduce Gemini API cost |
@@ -319,6 +346,7 @@ curl -X POST http://localhost:8080/availability-sync  # mirror busy time
 | `AVAILABILITY_FREEBUSY_CALENDAR_IDS` | Comma-separated calendars shared as free/busy-only (e.g. a corporate work calendar), read via the FreeBusy API — busy time only, never details (default empty) |
 | `AVAILABILITY_SYNC_DAYS` | How many days ahead to keep mirrored (default `60`) |
 | `AVAILABILITY_BUSY_TITLE` | Title used for every mirrored block (default `Busy`) |
+| `RESEARCH_TOPIC` | Topic for the research digest, e.g. `"AI and LLM news this week"` (required for `/research`) |
 | `GOOGLE_CREDENTIALS_FILE` | Path to OAuth client secrets (local only) |
 | `GOOGLE_TOKEN_FILE` | Path to OAuth token file (local only) |
 | `BROWSER_PATH` | Optional path to a specific browser for OAuth consent |

assistant/email_reader.py

@@ -199,6 +199,19 @@ def create_new_draft(service, to: list[str], subject: str, body: str) -> str:
     return f"https://mail.google.com/mail/#drafts/{draft['id']}"
 
 
+def send_message(service, to: str, subject: str, body: str) -> str:
+    """Send an email message immediately. Returns the sent message ID."""
+    mime_msg = email.mime.text.MIMEText(body, "plain", "utf-8")
+    mime_msg["To"] = to
+    mime_msg["Subject"] = subject
+    raw = base64.urlsafe_b64encode(mime_msg.as_bytes()).decode("utf-8")
+    sent = service.users().messages().send(
+        userId="me",
+        body={"raw": raw},
+    ).execute()
+    return sent["id"]
+
+
 def _extract_text(payload: dict) -> str:
     """Recursively extract plain text from a MIME payload."""
     mime_type = payload.get("mimeType", "")

assistant/researcher.py

@@ -0,0 +1,31 @@
+import logging
+from datetime import date
+
+from google import genai
+from google.genai import types
+
+logger = logging.getLogger(__name__)
+
+def research_topic(client: genai.Client, topic: str, model: str) -> str:
+    """Research a topic using Gemini with Google Search grounding.
+
+    Returns the research summary as plain text suitable for an email body.
+    """
+    today = date.today().strftime("%B %d, %Y")
+    prompt = (
+        f"Today is {today}. Research the following topic and write a clear, "
+        f"well-organized summary covering key recent developments, insights, "
+        f"and anything notable. Write only the email body — no greeting or sign-off.\n\n"
+        f"Topic: {topic}"
+    )
+    response = client.models.generate_content(
+        model=model,
+        contents=prompt,
+        config=types.GenerateContentConfig(
+            tools=[types.Tool(google_search=types.GoogleSearch())],
+            max_output_tokens=2048,
+        ),
+    )
+    if not response.text:
+        raise ValueError("Gemini returned an empty response or the response was blocked.")
+    return response.text

assistant/state.py

@@ -1,9 +1,11 @@
 import json
+import hashlib
 import logging
 import os
 import tempfile
 
 from google.cloud import storage
+from google.api_core import exceptions as google_exceptions
 
 logger = logging.getLogger(__name__)
 
@@ -77,6 +79,41 @@ def save_last_standup_date(bucket_name: str, date_str: str) -> None:
     )
 
 
+_RESEARCH_PREFIX = "research_runs"
+
+
+def claim_research(bucket_name: str, topic: str, date_str: str) -> str | None:
+    """Atomically reserve one digest for a topic and date.
+
+    Returns the marker name when this caller acquired the reservation, or
+    ``None`` when another request already owns/completed it. GCS's generation
+    precondition makes concurrent Scheduler retries race safely: exactly one
+    upload can create the marker.
+    """
+    topic_hash = hashlib.sha256(topic.encode("utf-8")).hexdigest()[:24]
+    marker = f"{_RESEARCH_PREFIX}/{date_str}-{topic_hash}.json"
+    gcs = storage.Client()
+    blob = gcs.bucket(bucket_name).blob(marker)
+    try:
+        blob.upload_from_string(
+            json.dumps({"topic": topic, "date": date_str}),
+            content_type="application/json",
+            if_generation_match=0,
+        )
+    except google_exceptions.PreconditionFailed:
+        return None
+    return marker
+
+
+def release_research(bucket_name: str, marker: str) -> None:
+    """Release a reservation after a definite failure before email delivery."""
+    gcs = storage.Client()
+    try:
+        gcs.bucket(bucket_name).blob(marker).delete()
+    except google_exceptions.NotFound:
+        pass
+
+
 def load_processed_ids(filepath: str) -> set[str]:
     """Load the set of processed Gmail message IDs from a JSON file.
 

main.py

@@ -19,7 +19,9 @@
     get_or_create_label,
     get_user_email,
     list_unread_message_ids,
+    send_message,
 )
+from assistant.researcher import research_topic
 from assistant.event_extractor import analyze_email, make_gemini_client, update_user_context
 from assistant.standup_generator import generate_standup_draft
 from assistant.state import (
@@ -29,6 +31,8 @@
     save_user_context,
     load_last_standup_date,
     save_last_standup_date,
+    claim_research,
+    release_research,
 )
 
 load_dotenv()
@@ -291,5 +295,50 @@ def availability_sync():
         return jsonify({"error": str(e)}), 500
 
 
+@app.route("/research", methods=["POST"])
+def research():
+    bucket_name = os.environ["GCS_BUCKET_NAME"]
+    topic = os.getenv("RESEARCH_TOPIC")
+    if not topic:
+        return jsonify({"error": "RESEARCH_TOPIC environment variable is not set"}), 400
+    gemini_api_key = os.environ["GEMINI_API_KEY"]
+    gemini_model = os.getenv("GEMINI_MODEL", "gemini-3.5-flash")
+
+    creds = _get_credentials()
+    gmail_service = get_gmail_service(creds)
+    gemini_client = make_gemini_client(gemini_api_key)
+
+    # Atomically reserve this topic/date immediately before the slow model call.
+    # A simple read-then-write check allows overlapping Scheduler retries to
+    # both send; setup failures above do not leave a reservation behind.
+    today_str = date.today().isoformat()
+    marker = claim_research(bucket_name, topic, today_str)
+    if marker is None:
+        logger.info("Research digest already sent for '%s' on %s, skipping.", topic[:60], today_str)
+        return jsonify({"skipped": True, "reason": "already_sent_today"}), 200
+
+    logger.info("Researching topic: %s", topic[:80])
+    try:
+        content = research_topic(gemini_client, topic, gemini_model)
+    except Exception as e:
+        release_research(bucket_name, marker)
+        logger.error("Research failed for topic '%s': %s", topic[:60], e, exc_info=True)
+        return jsonify({"error": str(e)}), 500
+
+    user_email = get_user_email(gmail_service)
+    if not user_email:
+        release_research(bucket_name, marker)
+        logger.error("Could not resolve the account email; cannot send research digest.")
+        return jsonify({"error": "could not resolve account email"}), 500
+
+    subject = f"Research digest: {topic} ({today_str})"
+    # Keep the reservation if Gmail raises: delivery may have succeeded before
+    # the response was lost, so retrying could produce the duplicate we avoid.
+    send_message(gmail_service, user_email, subject, content)
+    logger.info("Research email sent to %s for topic: %s", user_email, topic[:60])
+
+    return jsonify({"sent": True, "topic": topic, "date": today_str}), 200
+
+
 if __name__ == "__main__":
     app.run(host="0.0.0.0", port=int(os.getenv("PORT", "8080")))

tests/test_research_state.py

@@ -0,0 +1,57 @@
+"""Tests for atomic research-digest reservations."""
+
+from google.api_core import exceptions as google_exceptions
+
+from assistant import state
+
+
+class _Blob:
+    def __init__(self, name, objects):
+        self.name = name
+        self.objects = objects
+
+    def upload_from_string(self, value, **kwargs):
+        assert kwargs["if_generation_match"] == 0
+        if self.name in self.objects:
+            raise google_exceptions.PreconditionFailed("already exists")
+        self.objects[self.name] = value
+
+    def delete(self):
+        if self.name not in self.objects:
+            raise google_exceptions.NotFound("missing")
+        del self.objects[self.name]
+
+
+class _Bucket:
+    def __init__(self, objects):
+        self.objects = objects
+
+    def blob(self, name):
+        return _Blob(name, self.objects)
+
+
+class _Client:
+    def __init__(self, objects):
+        self.objects = objects
+
+    def bucket(self, _name):
+        return _Bucket(self.objects)
+
+
+def test_claim_is_atomic_and_topic_scoped(monkeypatch):
+    objects = {}
+    monkeypatch.setattr(state.storage, "Client", lambda: _Client(objects))
+
+    marker = state.claim_research("bucket", "AI policy", "2026-07-17")
+    assert marker
+    assert state.claim_research("bucket", "AI policy", "2026-07-17") is None
+    assert state.claim_research("bucket", "different topic", "2026-07-17")
+
+
+def test_release_allows_retry_after_definite_failure(monkeypatch):
+    objects = {}
+    monkeypatch.setattr(state.storage, "Client", lambda: _Client(objects))
+
+    marker = state.claim_research("bucket", "AI policy", "2026-07-17")
+    state.release_research("bucket", marker)
+    assert state.claim_research("bucket", "AI policy", "2026-07-17") == marker

tests/test_researcher.py

@@ -0,0 +1,30 @@
+from types import SimpleNamespace
+
+import pytest
+
+from assistant.researcher import research_topic
+
+
+class FakeModels:
+    def __init__(self, text):
+        self.text = text
+        self.calls = []
+
+    def generate_content(self, **kwargs):
+        self.calls.append(kwargs)
+        return SimpleNamespace(text=self.text)
+
+
+def test_research_uses_configured_model():
+    models = FakeModels("Digest")
+    client = SimpleNamespace(models=models)
+
+    assert research_topic(client, "AI policy", "configured-model") == "Digest"
+    assert models.calls[0]["model"] == "configured-model"
+
+
+def test_research_rejects_empty_response():
+    client = SimpleNamespace(models=FakeModels(""))
+
+    with pytest.raises(ValueError, match="empty response"):
+        research_topic(client, "AI policy", "configured-model")