# Gmail Read-Only Agent Gateway — v1 Implementation Plan (Python)

**Audience:** an implementing agent (and its sub-agents). This document is self-contained: it defines the architecture, exact contracts, module boundaries, a non-overlapping task breakdown, and machine-verifiable success criteria for each task so a main agent can verify completion without human judgment.

---

## 1. Project summary

A personal, self-hosted service that exposes a Gmail mailbox to AI agents in a **strictly read-only** fashion. The service syncs new mail from Gmail into a local SQLite store using an OAuth token with only the `gmail.readonly` scope, and serves it to a single polling agent over a small authenticated HTTP API. **No write capability of any kind exists anywhere in the codebase** — no send, no delete, no modify, no label changes. Attachments are never stored or served.

### Security invariants (must hold in every task)
1. The only Google OAuth scope requested anywhere is `https://www.googleapis.com/auth/gmail.readonly`.
2. Google credentials (client secret, refresh token) are never readable via any HTTP endpoint and never logged.
3. The HTTP API registers only `GET` routes. No `POST`/`PUT`/`PATCH`/`DELETE` handlers exist (they must return 405).
4. Email bodies are stored and served as sanitized plain text only (see §7). Attachments and raw MIME are discarded at ingest.
5. API auth is a single static bearer token compared in constant time (`secrets.compare_digest`). The comparison lives in one function (`app/auth.py:verify_token`) so it can later be swapped for multi-key lookup.

---

## 2. Stack and constraints

| Concern | Decision |
|---|---|
| Language | Python 3.12+ |
| Web framework | FastAPI + uvicorn |
| Storage | SQLite via stdlib `sqlite3` (WAL mode). No ORM. |
| Gmail client | `google-api-python-client`, `google-auth`, `google-auth-oauthlib` |
| HTML→text | `beautifulsoup4` |
| Config | Environment variables via a single `app/config.py` (may use `pydantic-settings`) |
| Testing | `pytest`, `httpx` (ASGI test client). No network access in tests. |
| Packaging | `pyproject.toml`, `uv` for dependency management |
| Container | Single Dockerfile; one volume mounted at `/data` for DB + token |
| Lint/format | `ruff` (lint + format), config in `pyproject.toml` |

The sync worker runs as an `asyncio` background task inside the same FastAPI process (started via lifespan). No queue, no Postgres, no Redis, no rate limiting, no multi-key management — explicitly out of scope for v1.

---

## 3. Repository layout (create exactly this)

```
.
├── pyproject.toml
├── README.md
├── Dockerfile
├── .env.example
├── app/
│   ├── __init__.py
│   ├── config.py          # Settings: env parsing, paths, validation
│   ├── db.py              # connection factory, schema DDL, migrations
│   ├── models.py          # dataclasses: Message, HealthStatus, UpdatesPage
│   ├── auth.py            # verify_token(header_value) -> None | raises 401
│   ├── sanitize.py        # html_to_text, strip_dangerous_unicode, normalize
│   ├── cursor.py          # encode_cursor / decode_cursor (opaque, urlsafe)
│   ├── store.py           # insert_message, get_updates(cursor), get_since(ts), sync metadata
│   ├── gmail/
│   │   ├── __init__.py
│   │   ├── client.py      # GmailClient protocol (interface)
│   │   ├── real.py        # RealGmailClient: OAuth + history.list + messages.get
│   │   └── fake.py        # FakeGmailClient: reads JSON fixtures from a directory
│   ├── sync.py            # sync loop: cursor mgmt, fetch, normalize, persist
│   ├── api.py             # FastAPI app factory, routes, lifespan
│   └── main.py            # uvicorn entrypoint
├── scripts/
│   └── authorize.py       # one-time OAuth consent flow, writes token file
├── tests/
│   ├── conftest.py        # tmp DB fixture, app fixture with FakeGmailClient
│   ├── fixtures/emails/   # JSON fixture emails for fake mode
│   ├── test_sanitize.py
│   ├── test_cursor.py
│   ├── test_store.py
│   ├── test_sync.py
│   ├── test_api_auth.py
│   ├── test_api_updates.py
│   └── test_api_readonly.py
└── .github/ (none needed for v1)
```

---

## 4. Configuration contract (`app/config.py`)

All config from environment variables. `.env.example` must list every one with a comment.

| Variable | Required | Default | Meaning |
|---|---|---|---|
| `GATEWAY_API_KEY` | yes | — | Static bearer token agents present. Startup fails with a clear error if unset or < 32 chars. |
| `GATEWAY_DB_PATH` | no | `/data/gateway.db` | SQLite file path. |
| `GATEWAY_GMAIL_MODE` | no | `real` | `real` or `fake`. |
| `GATEWAY_FAKE_MAILDIR` | if fake | — | Directory of JSON fixture emails for fake mode. |
| `GATEWAY_GOOGLE_CREDENTIALS` | if real | `/data/credentials.json` | OAuth client file from Google Cloud console. |
| `GATEWAY_GOOGLE_TOKEN` | if real | `/data/token.json` | Refresh-token file written by `scripts/authorize.py`. |
| `GATEWAY_SYNC_INTERVAL_SECONDS` | no | `300` | Sync loop period. |
| `GATEWAY_HOST` / `GATEWAY_PORT` | no | `127.0.0.1` / `48291` | Bind address. Port is deliberately uncommon. |

---

## 5. Database schema (`app/db.py`)

```sql
CREATE TABLE IF NOT EXISTS messages (
  id            TEXT PRIMARY KEY,        -- Gmail message id
  thread_id     TEXT NOT NULL,
  sender_name   TEXT NOT NULL DEFAULT '',
  sender_email  TEXT NOT NULL,
  subject       TEXT NOT NULL DEFAULT '',
  body_text     TEXT NOT NULL,           -- sanitized plain text only
  internal_date_ms INTEGER NOT NULL,     -- Gmail internalDate (epoch ms)
  synced_at_ms  INTEGER NOT NULL         -- when THIS service ingested it (epoch ms, monotonic per §6)
);
CREATE INDEX IF NOT EXISTS idx_messages_synced ON messages (synced_at_ms, id);

CREATE TABLE IF NOT EXISTS sync_state (
  key   TEXT PRIMARY KEY,   -- 'history_id', 'last_sync_ok_ms', 'last_sync_error'
  value TEXT NOT NULL
);
```

Rules:
- Open connections with `PRAGMA journal_mode=WAL;` and `PRAGMA foreign_keys=ON;`.
- `synced_at_ms` must be strictly increasing across inserts within one process (if `time.time_ns()//1e6` ties, bump by 1ms) — this makes the cursor total-order well-defined together with `id` as tiebreaker.
- Inserts are idempotent: `INSERT OR IGNORE` on primary key (re-sync must not duplicate).

---

## 6. Cursor format (`app/cursor.py`)

Opaque, URL-safe token encoding `(synced_at_ms: int, last_id: str)`:
- `encode_cursor(synced_at_ms, last_id)` → `base64.urlsafe_b64encode(json.dumps({"v":1,"t":synced_at_ms,"i":last_id}))` without padding.
- `decode_cursor(s)` → tuple; raises `InvalidCursor` on any malformed input (bad base64, bad JSON, wrong version, missing keys, non-int t). API maps `InvalidCursor` to HTTP 400 with `{"error":"invalid_cursor"}`.
- Pagination predicate: `(synced_at_ms, id) > (t, i)` lexicographically, ordered by `synced_at_ms ASC, id ASC`.

---

## 7. Sanitization spec (`app/sanitize.py`)

Applied at ingest, before storage:
1. Prefer the `text/plain` MIME part; if absent, convert `text/html` via BeautifulSoup `get_text(separator="\n")`.
2. Strip zero-width characters: U+200B, U+200C, U+200D, U+2060, U+FEFF.
3. Strip bidirectional override/embedding controls: U+202A–U+202E, U+2066–U+2069.
4. Normalize to NFC. Collapse runs of 3+ newlines to 2. Strip leading/trailing whitespace.
5. Truncate stored body at 100,000 characters (append `\n[truncated]` when cut).

No attachments: at MIME walk time, skip any part with `filename` set or `Content-Disposition: attachment`; never store or return part bytes other than the chosen text body.

---

## 8. HTTP API contract (`app/api.py`)

All routes require `Authorization: Bearer <GATEWAY_API_KEY>`; failures return 401 `{"error":"unauthorized"}` (same response for missing and wrong key). Only GET routes exist.

### `GET /v1/updates?cursor=<opaque>&limit=<1..500, default 100>`
- Without `cursor`: returns **no messages**, only a `next_cursor` positioned at the current end (bootstrap call — agents start from "now").
- With `cursor`: returns messages after the cursor position ordered by `(synced_at_ms, id)`, up to `limit`.
- Response:
```json
{
  "messages": [{
    "id": "...", "thread_id": "...",
    "sender_name": "...", "sender_email": "...",
    "subject": "...",
    "body_untrusted": "...",
    "received_at": "2026-08-28T11:00:00Z",
    "synced_at": "2026-08-28T11:02:11Z"
  }],
  "next_cursor": "...",
  "has_more": false
}
```
- `next_cursor` always present (equals input cursor position advanced past returned rows; unchanged when zero rows). The body field is deliberately named `body_untrusted`.

### `GET /v1/messages?since=<ISO8601>&limit=<1..500, default 100>`
- Timestamp convenience variant filtering on `synced_at_ms > since`. Missing/invalid `since` → 400 `{"error":"invalid_since"}`. Same message shape, no cursor in response.

### `GET /v1/health`
- Returns 200 `{"status":"ok"|"degraded","last_sync_ok":"<ISO8601>|null","last_sync_error":"<string>|null","message_count":<int>,"mode":"real"|"fake"}`. `degraded` when last successful sync is older than 3× the sync interval. Health requires auth too (it leaks mailbox metadata).

---

## 9. Gmail client boundary (`app/gmail/`)

`client.py` defines the protocol both implementations satisfy:
```python
class GmailClient(Protocol):
    def list_new_message_ids(self, history_id: str | None) -> tuple[list[str], str]:
        """Returns (new_message_ids, new_history_id). history_id None => bootstrap:
        return ([], current_history_id) — v1 starts from 'now', no backfill.
        Raises HistoryExpired if the stored history_id is too old."""
    def fetch_message(self, message_id: str) -> RawMessage:
        """RawMessage: dataclass with headers dict, mime parts (text/plain, text/html), internal_date_ms."""
```

- **`real.py`**: loads token via `google.oauth2.credentials.Credentials.from_authorized_user_file`, refreshes if expired, builds the service with `cache_discovery=False`. Uses `users.history.list` (with `startHistoryId`, `historyTypes=messageAdded`) and `users.messages.get(format='full')`. On `HistoryExpired` (HTTP 404 from history.list), sync resets: store the profile's current `historyId` and continue (accepting a gap; log a warning).
- **`fake.py`**: reads `*.json` files from `GATEWAY_FAKE_MAILDIR`; each file is `{"id","thread_id","headers":{"From","Subject","Date"},"text_plain":...,"text_html":...,"internal_date_ms":...}`. `list_new_message_ids` returns files not yet reported this process lifetime, sorted by filename, so dropping a new JSON file simulates mail arrival. This makes the entire system runnable and testable with zero Google setup.
- **`scripts/authorize.py`**: standalone `InstalledAppFlow.run_local_server` consent flow requesting only the readonly scope; writes `GATEWAY_GOOGLE_TOKEN`. Never imported by the app.

---

## 10. Sync loop (`app/sync.py`)

`async def run_sync_loop(client, store, interval)`: every `interval` seconds → read `history_id` from `sync_state` → `list_new_message_ids` → for each id not already in DB: `fetch_message`, extract sender via `email.utils.parseaddr(headers['From'])`, sanitize body per §7, insert → persist new `history_id` and `last_sync_ok_ms`. On any exception: record `last_sync_error`, log, sleep, continue (loop never dies). Also expose `async def sync_once(...)` used by tests and by lifespan for an immediate first sync.

---

## 11. Task breakdown for sub-agents

Tasks are designed to be **non-overlapping by file**: no two tasks in the same wave touch the same file. Interfaces above are the contract; implement to them exactly.

### Wave 1 (parallel, no dependencies)

**T1 — Skeleton & config.** Files: `pyproject.toml`, `.env.example`, `README.md`, `Dockerfile`, `app/__init__.py`, `app/config.py`, `app/models.py`, `app/main.py` (stub that imports app factory lazily). 
*Success criteria:* `uv sync` succeeds; `uv run python -c "from app.config import Settings"` works; instantiating settings without `GATEWAY_API_KEY` raises a validation error whose message names the variable; `ruff check .` clean.

**T2 — Storage core.** Files: `app/db.py`, `app/cursor.py`, `app/store.py`, `tests/test_cursor.py`, `tests/test_store.py`. 
*Success criteria:* `uv run pytest tests/test_cursor.py tests/test_store.py` green. Tests must cover: cursor round-trip; `decode_cursor` raising `InvalidCursor` on ≥4 malformed inputs; idempotent insert (same id twice → 1 row); monotonic `synced_at_ms` under two same-millisecond inserts; `get_updates` pagination returning correct pages and `has_more` across a 250-row dataset with limit 100.

**T3 — Sanitization.** Files: `app/sanitize.py`, `tests/test_sanitize.py`. 
*Success criteria:* `uv run pytest tests/test_sanitize.py` green. Tests must cover: HTML→text of a nested HTML email; removal of every codepoint listed in §7 items 2–3 (assert none remain in output); NFC normalization; newline collapsing; 100k truncation with marker; plain-text passthrough unchanged apart from whitespace rules.

**T4 — Gmail clients.** Files: `app/gmail/__init__.py`, `app/gmail/client.py`, `app/gmail/real.py`, `app/gmail/fake.py`, `scripts/authorize.py`, `tests/fixtures/emails/` (≥5 varied fixtures: plain-text, HTML-only, both-parts, attachment-bearing, unicode-tricks). 
*Success criteria:* `uv run python -c "import app.gmail.real"` succeeds (imports must not require credentials); FakeGmailClient over the fixtures returns all ids once and then `([], hid)` on the next call; grep gate: `grep -r "gmail" app/ scripts/ | grep -i scope` shows only `gmail.readonly`; no other Gmail scope string appears anywhere in the repo.

### Wave 2 (parallel, depends on Wave 1)

**T5 — Sync engine.** Files: `app/sync.py`, `tests/test_sync.py`, `tests/conftest.py` (owns conftest). 
*Success criteria:* `uv run pytest tests/test_sync.py` green. Tests must cover: `sync_once` with FakeGmailClient ingests all fixtures with sanitized bodies (assert the unicode-tricks fixture is cleaned and the attachment fixture stored without attachment data); second `sync_once` inserts nothing; a client whose `fetch_message` raises records `last_sync_error` without raising out of `sync_once`'s loop wrapper `run_sync_loop` iteration; history_id persisted.

**T6 — API layer.** Files: `app/auth.py`, `app/api.py`, `tests/test_api_auth.py`, `tests/test_api_updates.py`, `tests/test_api_readonly.py`; finalize `app/main.py`. 
*Success criteria:* `uv run pytest tests/test_api_auth.py tests/test_api_updates.py tests/test_api_readonly.py` green. Tests must cover: 401 on missing/wrong bearer for all three routes (identical body); `verify_token` uses `secrets.compare_digest` (assert via source inspection or monkeypatch); bootstrap `/v1/updates` without cursor returns 0 messages + usable `next_cursor`; full poll cycle: bootstrap → insert 3 rows → poll returns exactly 3 with correct field names including `body_untrusted` → next poll returns 0; invalid cursor → 400; `/v1/messages?since=` valid & invalid; `/v1/health` shape and `degraded` transition; **read-only gate:** parametrized test asserting `POST/PUT/PATCH/DELETE` on every route path returns 405, and a route-table introspection test asserting every registered route allows only GET (and HEAD).

### Wave 3 (single agent, integration)

**T7 — Integration & docs polish.** May touch any file. Run the full suite, run the service end-to-end in fake mode: start server, bootstrap cursor via curl, drop a new JSON fixture into the maildir, wait one sync interval, poll and confirm the new message appears; verify Docker build; finalize README (setup incl. Google Cloud OAuth steps, fake-mode quickstart, API reference, security notes). 
*Success criteria:* `uv run pytest` fully green; `ruff check .` clean; the curl end-to-end transcript shows the dropped fixture returned by `/v1/updates`; `docker build .` succeeds; README contains a copy-pasteable fake-mode quickstart that works.

### Main-agent verification checklist (after all waves)
1. `uv run pytest -q` → exit 0, zero skipped security tests.
2. `ruff check .` → exit 0.
3. Grep gates: only `gmail.readonly` scope string in repo; no `users().messages().send`, `modify`, `trash`, or `delete` calls anywhere; `body_untrusted` present in API responses.
4. Live fake-mode smoke test per T7 transcript.
5. Confirm `.env.example` documents every variable in §4.

---

## 12. Explicitly out of scope for v1
Attachments, HTML bodies in responses, multiple API keys, rate limiting, MCP adapter, Gmail push (Pub/Sub), backfill of pre-existing mail, label filtering, TLS termination (run behind a reverse proxy or on localhost/tailnet).

## 13. Future-proofing notes for the implementer
Keep `store.get_updates` and `sync_once` free of FastAPI imports — they are the durable core. The HTTP layer and any future MCP adapter are skins over those functions. `app/auth.py:verify_token` is the single seam for a future multi-key upgrade.