# DEB-362: Gmail Agent RAG Cache Standalone LLD (Draft for Practical Delivery)

> **Jira:** [DEB-362](https://divami.atlassian.net/browse/DEB-362)  
- **Status:** Draft
- **Assignee:** arpit
---
**Description:**

This document expands DEB-362 into a standalone Gmail Agent low-level design.

The Gmail agent should:
- connect to Gmail using a real connector
- sync and persist Gmail data into Postgres
- maintain a RAG cache for previously answered Gmail questions
- check the RAG cache first for every incoming Gmail question
- fall back to the normal retrieval and answer-generation workflow when the cache does not contain a valid answer

Chanakya integration is intentionally a later step. The first delivery goal is a Gmail agent that works correctly on its own.

---

## 1. Goals

- Build the Gmail agent as an independently testable data source agent
- Replace the current mock-email path with real OAuth-authenticated Gmail API calls
- Persist Gmail messages and semantic chunks in Postgres with `pgvector`
- Add deterministic cache-first query behavior enforced in Python service code (not by the LLM)
- Preserve citations and source traceability for every answer

## 2. Non-Goals (First Delivery)

- No Chanakya orchestration changes
- No multi-source synthesis across Gmail and Jira
- No multi-mailbox / multi-tenant UI — single mailbox configured via environment
- No Gmail push notifications (Pub/Sub)
- No attachment parsing (PDF, XLSX)
- No generic connector framework

## 3. Current Baseline

- Gmail agent uses mock data and a single keyword `search_emails` tool in `backend/chanakya/gmail/agent.py`
- Jira agent is the reference pattern for real-API standalone agents
- Stack: `pydantic-ai`, `asyncpg`, `pgvector`, `google-api-python-client`, `sqlalchemy` (already in `pyproject.toml`)
- Auth framework: `authlib` already installed

---

## 4. Standalone Gmail Agent Architecture

```mermaid
flowchart LR
    U[User or Chanakya] --> A[Gmail Agent A2A / Chat Entry]
    A --> B[Query Orchestrator]
    B --> C[Cache Service]
    C --> D{Valid Cache Hit?}
    D -->|Yes| E[Return Cached Answer]
    D -->|No| F[Retrieval Service]
    F --> G[Postgres pgvector Search]
    G --> H[Message Expansion]
    H --> I[Answer Service]
    I --> J[Cache Write-back]
    J --> K[Return Answer]
    E --> K

    S[APScheduler] --> SS[Gmail Sync Service]
    SS --> GC[Gmail Connector]
    GC --> M[Gmail API]
    GC --> N[Normalizer + Chunker]
    N --> O[(Postgres: messages + chunks)]
    N --> P[Embedder]
    P --> O
    O --> G
    O --> C
```

---

## 5. Query-Time Flow

```mermaid
sequenceDiagram
    participant Q as Query Caller
    participant GA as Gmail Agent
    participant OR as Query Orchestrator
    participant CS as Cache Service
    participant DB as Postgres
    participant RS as Retrieval Service
    participant AS as Answer Service

    Q->>GA: answer_question(question)
    GA->>OR: handle(question, tenant_id)
    OR->>CS: lookup(question, tenant_id)
    CS->>DB: SELECT rag_cache_entries WHERE<br/>tenant_id = ? AND similarity > 0.92<br/>AND expires_at > now()<br/>AND dependency_max_history_id <= current_watermark
    DB-->>CS: candidate rows or empty

    alt valid cache hit (similarity >= 0.92, not expired, watermark valid)
        OR-->>GA: cached answer + citations
        GA-->>Q: response
    else cache miss or stale
        OR->>RS: retrieve(question, tenant_id)
        RS->>DB: pgvector cosine search over gmail_message_chunks
        DB-->>RS: top-k chunk rows
        RS->>DB: fetch parent gmail_messages by id
        DB-->>RS: full message payloads
        RS-->>OR: retrieval context + source message ids
        OR->>AS: generate_answer(question, context)
        AS-->>OR: answer + citations + dependency_message_ids
        OR->>CS: store(answer, query, dependency_message_ids, current_watermark)
        CS->>DB: INSERT rag_cache_entries
        OR-->>GA: final answer
        GA-->>Q: response
    end
```

---

## 6. Sync and Ingestion Flow

```mermaid
sequenceDiagram
    participant T as APScheduler
    participant SS as Gmail Sync Service
    participant GC as Gmail Connector
    participant API as Gmail API
    participant CH as Chunker / Embedder
    participant DB as Postgres

    T->>SS: incremental_sync(tenant_id)
    SS->>DB: SELECT stored_history_id FROM gmail_sync_state
    SS->>GC: create authenticated client (from encrypted token in DB)
    GC->>API: history.list(startHistoryId=stored_history_id)
    API-->>GC: history events OR 410 Gone

    alt 410 Gone
        GC-->>SS: raise HistoryExpiredError
        SS->>SS: full_sync(tenant_id)
        SS->>GC: messages.list(all pages)
    else success
        GC->>API: messages.get for each changed id
        API-->>GC: raw message payloads
    end

    GC-->>SS: normalized MessagePayload list
    SS->>CH: chunk(body_text, max_tokens=512, overlap=64)
    CH->>CH: strip HTML, split on semantic boundaries
    CH-->>SS: chunk list with token counts
    SS->>SS: embed(chunk_texts) via text-embedding-3-small
    SS->>DB: UPSERT gmail_messages ON CONFLICT (tenant_id, gmail_message_id)
    SS->>DB: DELETE + INSERT gmail_message_chunks for updated messages
    SS->>DB: UPDATE gmail_sync_state SET history_id = latest, synced_at = now()
    SS->>CS: invalidate_cache_entries(changed_message_ids)
```

---

## 7. Internal Module Layout

```
backend/chanakya/gmail/
├── agent.py            # EnterpriseAgent entrypoint, exposes answer_question + sync_mailbox tools
├── connector.py        # OAuth client creation, token refresh, Gmail API wrappers
├── sync_service.py     # full_sync / incremental_sync orchestration, 410 fallback
├── repository.py       # All Postgres reads/writes (messages, chunks, cache, sync state, tokens)
├── chunking.py         # HTML strip → sentence-aware chunking at 512 tokens / 64 overlap
├── embeddings.py       # Calls OpenAI text-embedding-3-small, batched, 1536-dim
├── retrieval.py        # pgvector cosine search → message expansion → grounded context
├── cache_service.py    # lookup, write-back, TTL expiry, watermark staleness check, invalidation
├── answer_service.py   # prompt construction, citation extraction, answer generation
└── schemas.py          # Pydantic DTOs: MessagePayload, ChunkRow, CacheEntry, RetrievalContext
```

### Component Responsibilities

- **`agent.py`** — Registers two tools with the `EnterpriseAgent`: `answer_question` and `sync_mailbox`. Contains no business logic.
- **`connector.py`** — Loads encrypted OAuth tokens from `repository.py`, handles refresh via `authlib`, wraps Gmail API list/get calls. Raises typed exceptions (`AuthError`, `HistoryExpiredError`).
- **`sync_service.py`** — Calls `connector.py` to fetch, calls `chunking.py` + `embeddings.py` to process, calls `repository.py` to persist. Triggers `cache_service.invalidate_cache_entries` after upsert.
- **`cache_service.py`** — `lookup()` performs a single pgvector query with all validity conditions in SQL (`similarity`, `expires_at`, `dependency_max_history_id`). `store()` constructs the cache row. `invalidate_entries_for_messages()` deletes cache rows whose `dependency_message_ids` overlap with the changed set.
- **`answer_service.py`** — LLM call with retrieval context injected into prompt. Must always return `citations` (list of `{message_id, subject, date, from_email}`). If citations cannot be formed, raises rather than returning an uncited answer.

---

## 8. Database Schema

### Migration Strategy

Alembic is the migration tool. Directory: `backend/alembic/`. Each schema change creates a new auto-generated revision. Run via `alembic upgrade head` in CI and on container start.

---

### 8.1 OAuth Token Storage

```sql
CREATE TABLE gmail_oauth_tokens (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id       TEXT NOT NULL,
    user_email      TEXT NOT NULL,
    -- Tokens encrypted at rest using Fernet (key from GMAIL_TOKEN_ENCRYPTION_KEY env var)
    access_token    BYTEA NOT NULL,
    refresh_token   BYTEA NOT NULL,
    token_expiry    TIMESTAMPTZ NOT NULL,
    scopes          TEXT[] NOT NULL,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE (tenant_id, user_email)
);
```

> **Security:** Tokens are encrypted with `cryptography.fernet` before INSERT and decrypted after SELECT. The encryption key is sourced from `GMAIL_TOKEN_ENCRYPTION_KEY` env var (32-byte base64 key). Never logged.

---

### 8.2 Sync State

```sql
CREATE TABLE gmail_sync_state (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id       TEXT NOT NULL UNIQUE,
    user_email      TEXT NOT NULL,
    stored_history_id TEXT,           -- NULL before first sync
    last_full_sync_at   TIMESTAMPTZ,
    last_incremental_sync_at TIMESTAMPTZ,
    sync_status     TEXT NOT NULL DEFAULT 'idle',  -- idle | running | error
    error_text      TEXT,
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);
```

---

### 8.3 Gmail Messages

```sql
CREATE TABLE gmail_messages (
    id                  UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id           TEXT NOT NULL,
    gmail_message_id    TEXT NOT NULL,
    thread_id           TEXT NOT NULL,
    subject             TEXT,
    from_email          TEXT,
    to_emails           JSONB,          -- ["a@b.com", ...]
    cc_emails           JSONB,
    label_ids           TEXT[],
    snippet             TEXT,
    body_text           TEXT,           -- plain text only; HTML stripped before storage
    internal_date       TIMESTAMPTZ,
    history_id          TEXT,
    raw_payload         JSONB,          -- full Gmail API payload retained for re-processing
    is_deleted          BOOLEAN NOT NULL DEFAULT FALSE,
    created_at          TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at          TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE (tenant_id, gmail_message_id)
);

CREATE INDEX ON gmail_messages (tenant_id, internal_date DESC);
CREATE INDEX ON gmail_messages (tenant_id, thread_id);
```

---

### 8.4 Gmail Message Chunks

```sql
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE gmail_message_chunks (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    message_id      UUID NOT NULL REFERENCES gmail_messages(id) ON DELETE CASCADE,
    tenant_id       TEXT NOT NULL,
    chunk_index     INT NOT NULL,
    chunk_text      TEXT NOT NULL,
    token_count     INT NOT NULL,
    -- text-embedding-3-small produces 1536-dim vectors
    embedding       vector(1536),
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX ON gmail_message_chunks USING ivfflat (embedding vector_cosine_ops)
    WITH (lists = 100);
CREATE INDEX ON gmail_message_chunks (tenant_id);
```

---

### 8.5 RAG Cache

```sql
CREATE TABLE rag_cache_entries (
    id                          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id                   TEXT NOT NULL,
    query_text                  TEXT NOT NULL,
    query_embedding             vector(1536) NOT NULL,
    answer_text                 TEXT NOT NULL,
    citations                   JSONB NOT NULL,  -- [{message_id, subject, date, from_email}]
    dependency_message_ids      JSONB NOT NULL,  -- [uuid, ...] of messages used to generate answer
    dependency_max_history_id   TEXT NOT NULL,   -- Gmail history_id watermark at generation time
    expires_at                  TIMESTAMPTZ NOT NULL,  -- TTL: created_at + 24h default
    hit_count                   INT NOT NULL DEFAULT 0,
    last_hit_at                 TIMESTAMPTZ,
    created_at                  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX ON rag_cache_entries USING ivfflat (query_embedding vector_cosine_ops)
    WITH (lists = 50);
CREATE INDEX ON rag_cache_entries (tenant_id, expires_at);
```

---

### 8.6 Entity Relationship Diagram

```mermaid
erDiagram
    gmail_oauth_tokens {
        uuid id PK
        text tenant_id
        text user_email
        bytea access_token
        bytea refresh_token
        timestamptz token_expiry
    }

    gmail_sync_state {
        uuid id PK
        text tenant_id
        text stored_history_id
        text sync_status
    }

    gmail_messages {
        uuid id PK
        text tenant_id
        text gmail_message_id
        text thread_id
        text subject
        text from_email
        jsonb to_emails
        text body_text
        jsonb raw_payload
    }

    gmail_message_chunks {
        uuid id PK
        uuid message_id FK
        text tenant_id
        int chunk_index
        text chunk_text
        vector_1536 embedding
    }

    rag_cache_entries {
        uuid id PK
        text tenant_id
        text query_text
        vector_1536 query_embedding
        text answer_text
        jsonb citations
        jsonb dependency_message_ids
        text dependency_max_history_id
        timestamptz expires_at
    }

    gmail_oauth_tokens ||--|| gmail_sync_state : "same tenant"
    gmail_messages ||--o{ gmail_message_chunks : split_into
    gmail_messages }o--o{ rag_cache_entries : "referenced_by dependency_message_ids"
```

---

## 9. Cache Validity Contract

A cache entry is a **valid hit** when ALL of the following are true:

1. `tenant_id` matches the current request
2. `cosine_similarity(query_embedding, stored_query_embedding) >= 0.92`
3. `expires_at > now()`
4. `dependency_max_history_id <= gmail_sync_state.stored_history_id` — the mailbox has not received new mail that postdates the cache entry's source data

A cache entry is **stale and skipped** when:
- `expires_at <= now()`, OR
- `gmail_sync_state.stored_history_id` has advanced past `dependency_max_history_id` for any of the dependency messages

**Cache write suppressed when:**
- Answer generation raised an exception
- Citations list is empty
- Retrieval returned zero chunks

**Cache invalidation on sync:**
After each incremental sync, `cache_service.invalidate_entries_for_messages(changed_message_ids)` deletes all cache rows where `dependency_message_ids` contains any of the updated message UUIDs. This is a direct DELETE by jsonb containment — no async queue needed at this scale.

```sql
DELETE FROM rag_cache_entries
WHERE tenant_id = $1
  AND dependency_message_ids ?| ARRAY[$2, $3, ...]::text[];
```

**Default TTL:** 24 hours. Configurable via `GMAIL_CACHE_TTL_HOURS` env var.

---

## 10. Chunking Specification

| Parameter | Value | Rationale |
|---|---|---|
| Tokenizer | `tiktoken` with `cl100k_base` | Same tokenizer as `text-embedding-3-small` |
| Chunk size | 512 tokens | Balances context width vs. retrieval precision |
| Chunk overlap | 64 tokens | Avoids split-sentence misses |
| HTML stripping | `BeautifulSoup` `get_text(separator=" ")` before any chunking | Already installed |
| Minimum chunk size | 20 tokens | Discard near-empty chunks (signatures, forwarding headers) |
| Boundary preference | Split on sentence boundaries using `\. |\n` before hard token limit | More coherent chunks |

---

## 11. Embedding Specification

| Parameter | Value |
|---|---|
| Model | `gemini-embedding-001` |
| Dimensions | 1536 |
| Batch size | 100 chunks per API call |
| Provider | OpenAI (direct) or via LiteLLM proxy using `LITELLM_PROVIDER_BASE_URL` |
| Index type | `ivfflat` with `vector_cosine_ops`, `lists=100` for chunks, `lists=50` for cache |

---

## 12. OAuth Flow

```mermaid
sequenceDiagram
    participant Admin as Admin / Setup
    participant App as Gmail Connector
    participant Google as Google OAuth
    participant DB as Postgres

    Admin->>App: POST /auth/gmail/initiate?tenant_id=X
    App->>Google: Redirect to OAuth consent URL<br/>(scopes: gmail.readonly)
    Google-->>Admin: Consent page
    Admin->>Google: Approve
    Google-->>App: Authorization code (via callback)
    App->>Google: Exchange code for tokens
    Google-->>App: access_token + refresh_token
    App->>App: Encrypt tokens with Fernet
    App->>DB: INSERT gmail_oauth_tokens
    App-->>Admin: "Gmail connected for tenant X"

    Note over App,DB: Normal operation — token refresh
    App->>DB: SELECT (decrypt) latest tokens
    App->>Google: API call with access_token
    alt token expired
        Google-->>App: 401
        App->>Google: POST /token with refresh_token
        Google-->>App: new access_token
        App->>App: Encrypt new access_token
        App->>DB: UPDATE gmail_oauth_tokens SET access_token, token_expiry
    end
```

**Required environment variables:**
- `GMAIL_OAUTH_CLIENT_ID`
- `GMAIL_OAUTH_CLIENT_SECRET`
- `GMAIL_OAUTH_REDIRECT_URI`
- `GMAIL_TOKEN_ENCRYPTION_KEY` — 32-byte Fernet key, base64-encoded

---

## 13. Agent API Contract

### User-Facing Tools (registered on `EnterpriseAgent`)

```python
@gmail_agent.tool
async def answer_question(ctx: RunContext[RequestContext], question: str) -> str:
    """Answer a question about emails. Cache-first; falls back to retrieval."""

@gmail_agent.tool
async def sync_mailbox(ctx: RunContext[RequestContext], mode: str = "incremental") -> str:
    """Trigger Gmail sync. mode: 'incremental' (default) or 'full'."""
```

### Internal Service Contracts (Python only, not exposed to LLM)

```python
# cache_service.py
async def lookup(question: str, tenant_id: str, current_history_id: str) -> CacheEntry | None
async def store(entry: CacheEntry) -> None
async def invalidate_entries_for_messages(tenant_id: str, message_ids: list[str]) -> int

# retrieval.py
async def retrieve(question: str, tenant_id: str, top_k: int = 10) -> RetrievalContext

# answer_service.py
async def generate_answer(question: str, context: RetrievalContext) -> AnswerResult

# sync_service.py
async def incremental_sync(tenant_id: str) -> SyncResult
async def full_sync(tenant_id: str) -> SyncResult
```

---

## 14. Sync Scheduler

Using `apscheduler` (already in `backend/pyproject.toml`):

```python
# Added to agent.py startup lifespan
from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler()
scheduler.add_job(
    sync_service.incremental_sync,
    "interval",
    minutes=int(os.environ.get("GMAIL_SYNC_INTERVAL_MINUTES", "15")),
    kwargs={"tenant_id": os.environ["GMAIL_TENANT_ID"]},
)
scheduler.start()
```

**Environment variable:** `GMAIL_SYNC_INTERVAL_MINUTES` (default: 15)

---

## 15. Failure Handling

| Failure | Behaviour |
|---|---|
| `AuthError` (OAuth expired, revoked) | Return operational error message to caller. Do not write cache. Log `ERROR`. |
| `HistoryExpiredError` (Gmail 410) | Automatically fall back to `full_sync`. Update `sync_state.error_text` temporarily. |
| Cache lookup failure (DB unreachable) | Log `WARNING`, continue with normal retrieval path. Never block the query. |
| Embedding API failure | Log `ERROR`, return "retrieval unavailable" message. Do not write cache. |
| Empty retrieval (no chunks match) | Return "no relevant emails found" message. Optionally cache with 1h TTL. |
| Answer generation failure | Log `ERROR`, return error message. Do not write cache. |
| Token encryption/decryption failure | Raise hard `AuthError`. Never proceed with plaintext tokens. |

---

## 16. Deferred Chanakya Integration

After the standalone Gmail agent is stable, Chanakya becomes a thin delegator:

```mermaid
flowchart LR
    C[Chanakya] --> G[Gmail Agent A2A]
    G --> O[Cache-first Query Orchestrator]
    O --> P[(Postgres)]
    O --> A[Gmail API]
```

No cache logic should be implemented in Chanakya. Chanakya calls `answer_question` over A2A and receives the final grounded answer — identical to how Chanakya currently delegates to the Jira agent.

---

## 17. Testing Strategy

### Unit Tests

| Test | What it verifies |
|---|---|
| `test_cache_hit_returns_without_retrieval` | `lookup()` returns a valid entry; retrieval and answer\_service are never called |
| `test_cache_miss_triggers_retrieval_and_write` | `lookup()` returns None; full retrieval → generate → store path executes |
| `test_stale_watermark_rejected` | Cache entry with `dependency_max_history_id` > current is classified as miss |
| `test_expired_ttl_rejected` | Cache entry with `expires_at` in the past is classified as miss |
| `test_empty_retrieval_no_cache_write` | Zero chunks returned → `store()` is never called |
| `test_missing_citations_no_cache_write` | Answer with empty citations → `store()` is never called |
| `test_sync_upsert_invalidates_cache` | After `incremental_sync`, changed message's cache entries are deleted |
| `test_history_410_fallback_to_full_sync` | `HistoryExpiredError` triggers `full_sync` |
| `test_token_encrypt_decrypt_roundtrip` | Fernet encrypt → store → decrypt produces original token |
| `test_chunking_strips_html` | HTML email body produces clean plain-text chunks |

### Integration Tests

| Test | What it verifies |
|---|---|
| `test_gmail_sync_persists_messages` | After sync with test fixture, `gmail_messages` rows exist with correct `tenant_id` |
| `test_chunk_search_returns_expected_messages` | pgvector search on a known question returns the seeded message |
| `test_answer_question_tool_end_to_end` | Full path via `answer_question` tool returns a cited answer |
| `test_answer_question_uses_cache_on_second_call` | Second identical question hits cache (confirmed by mock assertion on retrieval) |

### Test Infrastructure

- Use `pytest` with `pytest-asyncio`
- Postgres: `pytest-postgresql` or Docker Compose test service
- Gmail API: mock with `unittest.mock` (fixture payloads in `tests/fixtures/gmail/`)
- Embedding API: mock returning deterministic 1536-dim float vectors

---

## 18. Environment Variables Reference

| Variable | Required | Default | Description |
|---|---|---|---|
| `GMAIL_OAUTH_CLIENT_ID` | Yes | — | Google OAuth app client ID |
| `GMAIL_OAUTH_CLIENT_SECRET` | Yes | — | Google OAuth app client secret |
| `GMAIL_OAUTH_REDIRECT_URI` | Yes | — | Callback URI for consent flow |
| `GMAIL_TOKEN_ENCRYPTION_KEY` | Yes | — | 32-byte Fernet key (base64) |
| `GMAIL_TENANT_ID` | Yes | — | Tenant identifier for single-mailbox mode |
| `GMAIL_SYNC_INTERVAL_MINUTES` | No | `15` | APScheduler polling interval |
| `GMAIL_CACHE_TTL_HOURS` | No | `24` | Cache entry TTL |
| `GMAIL_CACHE_SIMILARITY_THRESHOLD` | No | `0.92` | Minimum cosine similarity for cache hit |
| `GMAIL_RETRIEVAL_TOP_K` | No | `10` | Number of chunks returned by pgvector search |
| `LITELLM_PROVIDER_BASE_URL` | Yes | — | LiteLLM proxy for LLM calls |
| `DATABASE_URL` | Yes | — | Postgres connection string |

---

## 19. Delivery Phases

| Phase | Deliverables | 
|---|---|
| **A — OAuth + DB** | `connector.py` (OAuth flow + token refresh), `repository.py`, Alembic migrations for all 5 tables 
| **B — Sync** | `sync_service.py` (full + incremental + 410 fallback), `chunking.py`, `embeddings.py` 
| **C — Retrieval + Answer** | `retrieval.py` (pgvector search + message expansion), `answer_service.py` (cited answers) 
| **D — Cache** | `cache_service.py` (lookup, write-back, invalidation), watermark staleness check 
| **E — Agent Wire-up** | Update `agent.py` with real tools, scheduler startup, remove mock data path 
| **F — Tests** | Unit + integration tests, test fixtures, CI Postgres service