From a4548e0b74cbd8ecf9da4ec9a2118c1e4d9f1b24 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 20 Jun 2026 01:39:40 +0000 Subject: [PATCH 1/5] Add research digest: POST /research sends a Gemini-powered email on any topic Uses Gemini 2.0 Flash with Google Search grounding to research a configurable topic and sends the summary directly to the authenticated Gmail account. Triggered by Cloud Scheduler on whatever cadence the user configures. New: assistant/researcher.py, send_message() in email_reader.py, /research route in main.py. README updated with setup instructions and env var docs. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CSnZNqUEh8uDyMCPDmzBTa --- README.md | 35 +++++++++++++++++++++++++++++++---- assistant/email_reader.py | 13 +++++++++++++ assistant/researcher.py | 32 ++++++++++++++++++++++++++++++++ main.py | 29 +++++++++++++++++++++++++++++ 4 files changed, 105 insertions(+), 4 deletions(-) create mode 100644 assistant/researcher.py diff --git a/README.md b/README.md index 4c8a6dd..eb9d91a 100644 --- a/README.md +++ b/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,14 @@ 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 + +The endpoint responds with `{"sent": true, "topic": "...", "date": "YYYY-MM-DD"}`. + ## AI Engineering ### LLM call design @@ -100,14 +108,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 +260,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 @@ -319,6 +345,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 | diff --git a/assistant/email_reader.py b/assistant/email_reader.py index 79fd37b..26782ee 100644 --- a/assistant/email_reader.py +++ b/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) + 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", "") diff --git a/assistant/researcher.py b/assistant/researcher.py new file mode 100644 index 0000000..7adcf14 --- /dev/null +++ b/assistant/researcher.py @@ -0,0 +1,32 @@ +import logging +from datetime import date + +from google import genai +from google.genai import types + +logger = logging.getLogger(__name__) + +RESEARCH_MODEL = "gemini-2.0-flash" + + +def research_topic(client: genai.Client, topic: 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=RESEARCH_MODEL, + contents=prompt, + config=types.GenerateContentConfig( + tools=[types.Tool(google_search=types.GoogleSearch())], + max_output_tokens=2048, + ), + ) + return response.text diff --git a/main.py b/main.py index 550f2be..d40defc 100644 --- a/main.py +++ b/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 ( @@ -291,5 +293,32 @@ def availability_sync(): return jsonify({"error": str(e)}), 500 +@app.route("/research", methods=["POST"]) +def research(): + topic = os.environ["RESEARCH_TOPIC"] + gemini_api_key = os.environ["GEMINI_API_KEY"] + + creds = _get_credentials() + gmail_service = get_gmail_service(creds) + gemini_client = make_gemini_client(gemini_api_key) + + logger.info("Researching topic: %s", topic[:80]) + try: + content = research_topic(gemini_client, topic) + except Exception as e: + logger.error("Research failed for topic '%s': %s", topic[:60], e, exc_info=True) + return jsonify({"error": str(e)}), 500 + + profile = gmail_service.users().getProfile(userId="me").execute() + user_email = profile["emailAddress"] + + today_str = date.today().isoformat() + subject = f"Research digest: {topic} ({today_str})" + 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"))) From 16d65bc1ed78b718de8ae1ee2a8ce15710b39c3a Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 20 Jun 2026 01:41:33 +0000 Subject: [PATCH 2/5] Address review: 400 on missing topic, guard empty response, UTF-8 MIME - Return 400 Bad Request when RESEARCH_TOPIC env var is not set - Raise ValueError on empty/blocked Gemini response in research_topic() - Use MIMEText(..., "plain", "utf-8") in send_message() to handle non-ASCII Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CSnZNqUEh8uDyMCPDmzBTa --- assistant/email_reader.py | 2 +- assistant/researcher.py | 2 ++ main.py | 4 +++- 3 files changed, 6 insertions(+), 2 deletions(-) diff --git a/assistant/email_reader.py b/assistant/email_reader.py index 26782ee..60e89f4 100644 --- a/assistant/email_reader.py +++ b/assistant/email_reader.py @@ -201,7 +201,7 @@ def create_new_draft(service, to: list[str], subject: str, body: str) -> str: 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) + 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") diff --git a/assistant/researcher.py b/assistant/researcher.py index 7adcf14..da8e94c 100644 --- a/assistant/researcher.py +++ b/assistant/researcher.py @@ -29,4 +29,6 @@ def research_topic(client: genai.Client, topic: str) -> str: max_output_tokens=2048, ), ) + if not response.text: + raise ValueError("Gemini returned an empty response or the response was blocked.") return response.text diff --git a/main.py b/main.py index d40defc..36a67eb 100644 --- a/main.py +++ b/main.py @@ -295,7 +295,9 @@ def availability_sync(): @app.route("/research", methods=["POST"]) def research(): - topic = os.environ["RESEARCH_TOPIC"] + 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"] creds = _get_credentials() From 79528a0f29b7b7178bbaee4e2362753934a0279f Mon Sep 17 00:00:00 2001 From: Kieran Maynard Date: Fri, 17 Jul 2026 17:13:41 -0700 Subject: [PATCH 3/5] Use configured model for research digests --- assistant/researcher.py | 7 ++----- main.py | 3 ++- tests/test_researcher.py | 30 ++++++++++++++++++++++++++++++ 3 files changed, 34 insertions(+), 6 deletions(-) create mode 100644 tests/test_researcher.py diff --git a/assistant/researcher.py b/assistant/researcher.py index da8e94c..687eef6 100644 --- a/assistant/researcher.py +++ b/assistant/researcher.py @@ -6,10 +6,7 @@ logger = logging.getLogger(__name__) -RESEARCH_MODEL = "gemini-2.0-flash" - - -def research_topic(client: genai.Client, topic: str) -> str: +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. @@ -22,7 +19,7 @@ def research_topic(client: genai.Client, topic: str) -> str: f"Topic: {topic}" ) response = client.models.generate_content( - model=RESEARCH_MODEL, + model=model, contents=prompt, config=types.GenerateContentConfig( tools=[types.Tool(google_search=types.GoogleSearch())], diff --git a/main.py b/main.py index 36a67eb..e86dcda 100644 --- a/main.py +++ b/main.py @@ -299,6 +299,7 @@ def research(): 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) @@ -306,7 +307,7 @@ def research(): logger.info("Researching topic: %s", topic[:80]) try: - content = research_topic(gemini_client, topic) + content = research_topic(gemini_client, topic, gemini_model) except Exception as e: logger.error("Research failed for topic '%s': %s", topic[:60], e, exc_info=True) return jsonify({"error": str(e)}), 500 diff --git a/tests/test_researcher.py b/tests/test_researcher.py new file mode 100644 index 0000000..71b26d2 --- /dev/null +++ b/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") From 4b93cf5650ab5105868628a81f775bbd9e95bb58 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 18 Jul 2026 00:23:54 +0000 Subject: [PATCH 4/5] Make /research idempotent to prevent duplicate digests on Scheduler retry POST /research sent immediately but recorded no completed run, so a Cloud Scheduler retry after a timeout could send the same dated digest twice. Mirror the /standup dedup pattern: persist the last sent (topic, date) to last_research.json in GCS and skip if today's digest for the current topic already went out. A changed RESEARCH_TOPIC still sends a fresh digest. Also use the get_user_email() helper (with its HttpError handling) instead of a raw getProfile call. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01CSnZNqUEh8uDyMCPDmzBTa --- README.md | 5 +++-- assistant/state.py | 27 +++++++++++++++++++++++++++ main.py | 20 +++++++++++++++++--- 3 files changed, 47 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index eb9d91a..36850d5 100644 --- a/README.md +++ b/README.md @@ -44,8 +44,9 @@ Share the availability calendar with consulting clients (read-only, free/busy) s 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. A GCS record (`last_research.json`) stores the topic and date of the last digest, so a Scheduler retry after a timeout won't send the same digest twice on the same day. Changing `RESEARCH_TOPIC` sends a fresh digest immediately -The endpoint responds with `{"sent": true, "topic": "...", "date": "YYYY-MM-DD"}`. +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 @@ -333,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`, `last_research.json` | | `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 | diff --git a/assistant/state.py b/assistant/state.py index 0f3dd2a..f6f9166 100644 --- a/assistant/state.py +++ b/assistant/state.py @@ -77,6 +77,33 @@ def save_last_standup_date(bucket_name: str, date_str: str) -> None: ) +_RESEARCH_BLOB = "last_research.json" + + +def load_last_research(bucket_name: str) -> dict: + """Return the last research digest record ({"topic": ..., "date": ...}). + + Returns an empty dict if a digest has never been sent. + """ + gcs = storage.Client() + blob = gcs.bucket(bucket_name).blob(_RESEARCH_BLOB) + if not blob.exists(): + return {} + try: + return json.loads(blob.download_as_text()) + except Exception: + logger.warning("Could not load %s from GCS.", _RESEARCH_BLOB) + return {} + + +def save_last_research(bucket_name: str, topic: str, date_str: str) -> None: + """Persist the topic and ISO date of the last sent research digest to GCS.""" + gcs = storage.Client() + gcs.bucket(bucket_name).blob(_RESEARCH_BLOB).upload_from_string( + json.dumps({"topic": topic, "date": date_str}), content_type="application/json" + ) + + def load_processed_ids(filepath: str) -> set[str]: """Load the set of processed Gmail message IDs from a JSON file. diff --git a/main.py b/main.py index e86dcda..27cf30a 100644 --- a/main.py +++ b/main.py @@ -31,6 +31,8 @@ save_user_context, load_last_standup_date, save_last_standup_date, + load_last_research, + save_last_research, ) load_dotenv() @@ -295,12 +297,22 @@ def availability_sync(): @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") + # Idempotency: a Cloud Scheduler retry (e.g. after a timeout on a slow run) + # must not send the same dated digest twice. Skip if we already sent this + # topic today. A changed topic still sends, matching the standup pattern. + today_str = date.today().isoformat() + last = load_last_research(bucket_name) + if last.get("date") == today_str and last.get("topic") == topic: + logger.info("Research digest already sent for '%s' on %s, skipping.", topic[:60], today_str) + return jsonify({"skipped": True, "reason": "already_sent_today"}), 200 + creds = _get_credentials() gmail_service = get_gmail_service(creds) gemini_client = make_gemini_client(gemini_api_key) @@ -312,12 +324,14 @@ def research(): logger.error("Research failed for topic '%s': %s", topic[:60], e, exc_info=True) return jsonify({"error": str(e)}), 500 - profile = gmail_service.users().getProfile(userId="me").execute() - user_email = profile["emailAddress"] + user_email = get_user_email(gmail_service) + if not user_email: + logger.error("Could not resolve the account email; cannot send research digest.") + return jsonify({"error": "could not resolve account email"}), 500 - today_str = date.today().isoformat() subject = f"Research digest: {topic} ({today_str})" send_message(gmail_service, user_email, subject, content) + save_last_research(bucket_name, topic, today_str) logger.info("Research email sent to %s for topic: %s", user_email, topic[:60]) return jsonify({"sent": True, "topic": topic, "date": today_str}), 200 From 2657ba0cfcf686abe7dfd866a59e823a72aeeae8 Mon Sep 17 00:00:00 2001 From: Kieran Maynard Date: Fri, 17 Jul 2026 17:31:36 -0700 Subject: [PATCH 5/5] Make research digest reservation atomic --- README.md | 4 +-- assistant/state.py | 42 ++++++++++++++++---------- main.py | 27 +++++++++-------- tests/test_research_state.py | 57 ++++++++++++++++++++++++++++++++++++ 4 files changed, 100 insertions(+), 30 deletions(-) create mode 100644 tests/test_research_state.py diff --git a/README.md b/README.md index 36850d5..bf063c4 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,7 @@ Share the availability calendar with consulting clients (read-only, free/busy) s 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. A GCS record (`last_research.json`) stores the topic and date of the last digest, so a Scheduler retry after a timeout won't send the same digest twice on the same day. Changing `RESEARCH_TOPIC` sends a fresh digest immediately +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. @@ -334,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`, `last_research.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 | diff --git a/assistant/state.py b/assistant/state.py index f6f9166..f3b20d2 100644 --- a/assistant/state.py +++ b/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,31 +79,39 @@ def save_last_standup_date(bucket_name: str, date_str: str) -> None: ) -_RESEARCH_BLOB = "last_research.json" +_RESEARCH_PREFIX = "research_runs" -def load_last_research(bucket_name: str) -> dict: - """Return the last research digest record ({"topic": ..., "date": ...}). +def claim_research(bucket_name: str, topic: str, date_str: str) -> str | None: + """Atomically reserve one digest for a topic and date. - Returns an empty dict if a digest has never been sent. + 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(_RESEARCH_BLOB) - if not blob.exists(): - return {} + blob = gcs.bucket(bucket_name).blob(marker) try: - return json.loads(blob.download_as_text()) - except Exception: - logger.warning("Could not load %s from GCS.", _RESEARCH_BLOB) - return {} + 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 save_last_research(bucket_name: str, topic: str, date_str: str) -> None: - """Persist the topic and ISO date of the last sent research digest to GCS.""" +def release_research(bucket_name: str, marker: str) -> None: + """Release a reservation after a definite failure before email delivery.""" gcs = storage.Client() - gcs.bucket(bucket_name).blob(_RESEARCH_BLOB).upload_from_string( - json.dumps({"topic": topic, "date": date_str}), content_type="application/json" - ) + try: + gcs.bucket(bucket_name).blob(marker).delete() + except google_exceptions.NotFound: + pass def load_processed_ids(filepath: str) -> set[str]: diff --git a/main.py b/main.py index 27cf30a..8f3b39a 100644 --- a/main.py +++ b/main.py @@ -31,8 +31,8 @@ save_user_context, load_last_standup_date, save_last_standup_date, - load_last_research, - save_last_research, + claim_research, + release_research, ) load_dotenv() @@ -304,34 +304,37 @@ def research(): gemini_api_key = os.environ["GEMINI_API_KEY"] gemini_model = os.getenv("GEMINI_MODEL", "gemini-3.5-flash") - # Idempotency: a Cloud Scheduler retry (e.g. after a timeout on a slow run) - # must not send the same dated digest twice. Skip if we already sent this - # topic today. A changed topic still sends, matching the standup pattern. - today_str = date.today().isoformat() - last = load_last_research(bucket_name) - if last.get("date") == today_str and last.get("topic") == topic: - logger.info("Research digest already sent for '%s' on %s, skipping.", topic[:60], today_str) - return jsonify({"skipped": True, "reason": "already_sent_today"}), 200 - 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) - save_last_research(bucket_name, topic, today_str) logger.info("Research email sent to %s for topic: %s", user_email, topic[:60]) return jsonify({"sent": True, "topic": topic, "date": today_str}), 200 diff --git a/tests/test_research_state.py b/tests/test_research_state.py new file mode 100644 index 0000000..e46079e --- /dev/null +++ b/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