# Gmail Data Pipeline – Email Enrichment to Org Snapshot

> **Related Doc:** [data-flow-understaning.md](../data-flow-understaning.md)

---

## 1. Overview

This document describes the low-level design of the Gmail data pipeline — from collecting raw emails across all org users, through per-email enrichment and project-level grouping (both done by the Data Source Agent), through storage and handoff, and finally the RT Agent's cross-source snapshot update.

**Responsibility split:**

| Stage | Owner | What happens |
|---|---|---|
| Step 1 | Data Source Agent (Gmail) | Collect all org emails, deduplicate |
| Step 2 | Data Source Agent (Gmail) | Call Gmail API per email → LLM enrichment → extract enriched intelligence unit → store to `eb.emails` + `eb.email_bullets` |
| Step 3 | Data Source Agent (Gmail) | Send the `project_id`s of newly written data to the RT Agent |
| RT Agent | RT Decision Agent | Receive `project_id`s from all DSAs → group bullet references per project across all sources → diff against previous snapshot → write updated org snapshot (no modification to bullet data) |

Steps 1–3 are the responsibility of **each Data Source Agent independently** — Gmail does it for emails, Jira does it for tickets, and so on. The RT Agent never touches raw emails or tickets. It reads directly from `eb.email_bullets` (and the equivalent source tables) using the `project_id`s notified by each DSA.

---

## 2. Storage Model

Three tables total — one canonical registry plus two for the email layer — plus the org snapshot written by the RT Agent. No raw content is ever persisted.

| Layer | Table | Written by | Read by |
|---|---|---|---|
| Project registry | `eb.projects` | Admin / provisioning | All pipeline tables (FK source), ad-hoc queries |
| Email record | `eb.emails` | Gmail DSA (Step 2) | RT Agent, RAG search, ad-hoc queries |
| Email bullets | `eb.email_bullets` | Gmail DSA (Step 2) | RT Agent, RAG search, ad-hoc queries |
| Org project snapshot | `eb.org_project_snapshot` | RT Agent | Distribution Engine (Component 2), CXO dashboards |

For each email, the Gmail API provides the content that feeds the LLM enrichment call. Only the LLM-derived intelligence is stored — the email header and summary in `eb.emails`, and each extracted bullet point as an independent row in `eb.email_bullets`. If enrichment fails, the Gmail API is called again on the next sync run.

**What each table is for:**

- **`eb.projects`** — One row per registered project. The canonical project registry. `project_id` is the short key (e.g. `EB`, `IIRM`, `Optinn`) used as FK across `eb.emails` and all downstream pipeline tables.

- **`eb.emails`** — One row per unique email. Stores standard header fields (id, thread_id, user_id, from, subject, recipients, created_at, attachments, email_link) plus `project_id` (FK → `eb.projects`), an LLM-generated overall summary, and a semantic embedding. The parent record that all bullet rows reference via FK.

- **`eb.email_bullets`** — One row per bullet point extracted from an email. Each row is an independent intelligence unit with its own `sentiment_value`, `sentiment_score`, `type`, `category`, `role`, `hashtags`, `urgency_score`, and per-bullet `embeddings`. This is the primary unit for RAG search and RT Agent queries. FK → `eb.emails.id`.

- **`eb.org_project_snapshot`** — One row per project (upserted on every sync run). Written by the RT Agent after cross-source merge and diff. Read by the Distribution Engine (Component 2) to build per-user views and trigger alerts. Contains the org-level unified view of a project across all sources, plus diff signals (new vs. resolved items since last snapshot).

---

## 3. Scope

- Gmail as the data source (same pattern applies to Jira, GChat, and future connectors)
- Data Source Agent responsibilities: collection, deduplication, per-email enrichment, per-project grouping, and handoff to RT Agent
- RT Agent responsibilities: cross-source aggregation, snapshot diffing, and org snapshot update
- Storage model: four tables — `eb.projects` (canonical project registry) + `eb.emails` (email header + summary, FK → projects) + `eb.email_bullets` (per-bullet intelligence unit) → `eb.org_project_snapshot` (org-level project state, written by RT Agent after querying bullets directly)

---

## 4. Pipeline Flow

```mermaid
sequenceDiagram
    participant RM as Refresh Mechanism
    participant GDSA as Gmail DSA
    participant JDSA as Jira DSA
    participant GM as Gmail API
    participant JR as Jira API
    participant DB as Summary DB
    participant RT as RT Decision Agent
    participant ORG as Org Snapshot

    RM->>GDSA: Trigger refresh (Gmail)
    RM->>JDSA: Trigger refresh (Jira)

    Note over GDSA,GM: Step 1 — Collect all org emails (flatten + deduplicate)
    loop for each user in org
        GDSA->>GM: Fetch emails for user
        GM-->>GDSA: Emails (in memory)
    end
    GDSA->>GDSA: Deduplicate by message_id

    Note over GDSA,DB: Step 2 — Enrich each email individually
    loop for each unique email
        GDSA->>GDSA: LLM: generate summary + extract bullets (per-bullet: type, category, role, sentiment, hashtags, urgency)
    end
    GDSA->>DB: Write enriched records → eb.emails + eb.email_bullets
    DB-->>GDSA: Confirm write

    Note over GDSA,RT: Step 3 — Notify RT Agent
    GDSA->>RT: Send project_ids of newly written data [EB, IIRM, Optinn]

    Note over JDSA,JR: Jira DSA runs same Steps 1–3 independently
    JDSA->>JR: Fetch tickets
    JR-->>JDSA: Tickets
    JDSA->>JDSA: Enrich + write to jira equivalent tables
    JDSA->>RT: Send project_ids of newly written data [EB, IIRM]

    Note over RT,ORG: RT Agent — Cross-source query + snapshot update
    RT->>DB: Query eb.email_bullets WHERE project_id IN received list
    DB-->>RT: Bullets for each project (all sources)
    RT->>RT: Merge per project across sources\n(EB: Gmail bullets + Jira bullets → unified EB view)
    RT->>ORG: Read previous snapshot for each project
    ORG-->>RT: Previous snapshots
    RT->>RT: Diff: what is new vs. already in snapshot
    RT->>ORG: Write updated org snapshot (replace previous)
```

---

## 5. Step-by-Step Design

### Step 1 — Collect All Org Emails (Flatten + Deduplicate)

The Data Source Agent fetches emails for every connected user in the org and builds a single flat set, deduplicated by `message_id`. User inbox boundaries are dropped — the pipeline works on org-level email content, not individual inboxes.

```
all_mails = set()
for each user in org:
    for each email in gmail_api.fetch_new_emails(user):
        all_mails.add(email)   # set deduplicates by message_id
```

**Output:** A flat, deduplicated collection of raw email objects held entirely in memory. Each object carries its `from`, `to`, `cc`, `bcc`, `subject`, and `attachments` metadata from the Gmail API response. Nothing is written to the DB at this stage.

```
Email 1 — project: Optinn   (in memory)
Email 2 — project: EB       (in memory)
Email 3 — project: EB       (in memory)
Email 4 — project: IIRM     (in memory)
Email 5 — project: EB       (in memory)
Email 6 — project: IIRM     (in memory)
```

**Key principle:** We do not care which inbox an email came from. We only care what emails exist in the org.

---

### Step 2 — Enrich Each Email Individually

Each email is processed independently to become a **self-contained intelligence unit**. The Gmail API fetch provides the email content, which feeds the LLM enrichment call. Only the LLM-derived intelligence is persisted — it is never written raw to the DB.

For each email, the LLM derives:

**Stored in `eb.emails` (one row per email):**

| Field | Source | Method |
|---|---|---|
| `recipients` | `from` + `to` + `cc` + `bcc` | Direct field merge |
| `project_id` | email content + participants | LLM inference — matched against `eb.projects` controlled vocabulary; stored as FK |
| `summary` | email content | LLM — overall plain-text summary of the email |
| `embeddings` | summary text | Embedding model — `vector(1536)` email-level semantic embedding |

**Stored in `eb.email_bullets` (one row per bullet):**

| Field | Source | Method |
|---|---|---|
| `text` | email content | LLM: one bullet per key insight extracted |
| `hashtags` | bullet text | LLM: content hashtags specific to this bullet |
| `sentiment_value` | bullet text | LLM: `positive / neutral / negative` per bullet |
| `sentiment_score` | bullet text | LLM: float `-1.0` to `1.0` — per-bullet polarity |
| `type` | bullet text | LLM: `information / alert / insight / escalation` per bullet |
| `category` | bullet text | LLM: `bug-report / release-update / meeting-notes / status-update / announcement / general` per bullet |
| `role` | bullet text | LLM: who this update is relevant for (e.g. `project-manager`, `engineering-lead`, `finance-lead`) |
| `user_id` | sync context | Inherited — org user whose inbox provided the parent email |
| `recipients` | parent email | Inherited — participants from the parent email |
| `embeddings` | bullet text | Embedding model — `vector(1536)` per-bullet embedding for fine-grained RAG |

**Critical detail — per-bullet independence:**
Each bullet in `eb.email_bullets` is a fully independent intelligence unit. The parent `eb.emails` record holds only the header and summary — all intelligence signals (sentiment, type, category, role, hashtags) live on the bullet row.

```
Email 2 — subject: "Sprint 1 Status — Enterprise Brain"
  emails →         summary: "Sprint delivery on track; Safari login bug filed; budget approval pending"
  email_bullets →  row 1: "Sprint 1 delivery on track for March 28"    hashtags:[#deadline]  sentiment_value: positive   score: +0.62  role: project-manager   urgency: 2
  email_bullets →  row 2: "Login flow bug reported on Safari"            hashtags:[#bug]       sentiment_value: negative   score: -0.45  role: engineering-lead  urgency: 4
  email_bullets →  row 3: "Budget approval needed for cloud credits"     hashtags:[#finance]   sentiment_value: neutral    score:  0.02  role: finance-lead      urgency: 2

Email 3 — subject: "FW: Sprint 1 Release Risk"
  emails →         summary: "QA sign-off blocked; second Safari bug occurrence flagged as release risk"
  email_bullets →  row 1: "QA sign-off blocked, release at risk"         hashtags:[#deadline]  sentiment_value: negative   score: -0.81  role: project-manager   urgency: 5
  email_bullets →  row 2: "Second occurrence of Safari login bug"        hashtags:[#bug]       sentiment_value: negative   score: -0.63  role: engineering-lead  urgency: 4
```

If enrichment needs to be re-run (e.g., prompt version update), the Gmail API is called again for that email and all bullet rows for that `email_id` are deleted and re-inserted.

---

### Step 3 — Notify RT Agent

Once all records are committed to `eb.emails` and `eb.email_bullets`, the Data Source Agent sends the **`project_id`s** of the projects that received new data in this sync run to the RT Agent. No summary records are created — the RT Agent queries the bullet tables directly.

```
rt_agent.notify(
    source="gmail",
    project_ids=["Optinn", "EB", "IIRM"]
)
```

The Jira DSA (and every other source) does the same independently:

```
rt_agent.notify(
    source="jira",
    project_ids=["EB", "IIRM"]
)
```

**Why `project_id`s only:**
- All bullet data is already committed to `eb.email_bullets` with `project_id` on the parent `eb.emails` row
- The RT Agent queries directly: `SELECT * FROM eb.email_bullets JOIN eb.emails USING (id) WHERE project_id = ?`
- Avoids duplicating data into an intermediate summary table
- RT Agent always reads the latest committed state from the DB — the bullet rows are the single source of truth

---

### RT Agent — Group, Diff, and Snapshot

Once the DSAs have committed all bullet data and sent their `project_id` notifications, the RT Agent takes over. Its first task is to **group the bullet data from all sources together per project** — for example, placing Gmail bullets and Jira bullets for project `EB` side by side under the same project context. From that unified per-project view, it computes the snapshot.

**What the RT Agent does, in order:**

1. **Group** — For each `project_id` notified, query all bullet rows from all source tables (`eb.email_bullets` for Gmail, Jira equivalent table, etc.). Group the resulting bullet IDs by source under that project.

2. **Diff** — Read the previous snapshot for that project from `eb.org_project_snapshot`. Compare the current set of bullet IDs (per source) against what was in the previous snapshot to identify:
   - `new_bullet_ids` — bullets present now that were not in the previous snapshot
   - `resolved_bullet_ids` — bullets that were in the previous snapshot but are no longer in the current query window

3. **Aggregate scalars** — Compute summary signals from bullet metadata (no text generation):
   - `dominant_type` = most severe `type` present (`escalation > alert > insight > information`)
   - `all_hashtags` = union of all `hashtags` arrays across all bullets
   - `all_participants` = union of all `recipients` arrays across all bullets

4. **Write snapshot** — Upsert the result into `eb.org_project_snapshot`. The snapshot stores **only bullet ID references and computed scalars** — no bullet text is copied. All intelligence (text, role, sentiment, urgency) remains on the original bullet rows and is fetched by the Distribution Engine on demand.

---

## 6. LLM Extraction Coverage

### 6.1 Full Extraction Map

Every field the DSA extracts per email. Fields marked **Direct** come from Gmail API headers with no LLM needed.

| Field | Stored in | Source | Method |
|---|---|---|---|
| `id` | `emails` | Gmail API | Direct — Gmail `message_id` |
| `thread_id` | `emails` | Gmail header | Direct |
| `from` | `emails` | Gmail header | Direct |
| `to / cc / bcc` | *(not stored)* | Gmail header | Direct — used only to build `recipients` list |
| `recipients` | `emails` | from + to + cc + bcc | Direct — merge + deduplicate all email addresses |
| `subject` | `emails` | Gmail header | Direct |
| `created_at` | `emails` | Gmail header (`Date`) | Direct — the email send timestamp |
| `email_link` | `emails` | Gmail API metadata | Direct — constructed from `message_id` |
| `attachments` | `emails` | MIME part headers | Direct — filename + MIME type only, no file download |
| `user_id` | `emails` + `email_bullets` | sync context | Direct — org user whose inbox provided this email (from the Step 1 loop) |
| `project_id` | `emails` | email content + participants | LLM — matched against `eb.projects` controlled vocabulary; FK stored on the email row |
| `summary` | `emails` | email content | LLM — overall plain-text summary of the email |
| `embeddings` | `emails` | summary text | Embedding model — `vector(1536)` email-level semantic embedding |
| `text` | `email_bullets` | email content | LLM: one row per key insight bullet extracted from the email |
| `hashtags` | `email_bullets` | bullet text | LLM: content hashtags specific to this bullet |
| `sentiment_value` | `email_bullets` | bullet text | LLM: `positive / neutral / negative` per bullet |
| `sentiment_score` | `email_bullets` | bullet text | LLM: float `-1.0` to `1.0` — per-bullet polarity |
| `type` | `email_bullets` | bullet text | LLM: `information / alert / insight / escalation` per bullet |
| `category` | `email_bullets` | bullet text | LLM: `bug-report / release-update / meeting-notes / status-update / announcement / general` per bullet |
| `role` | `email_bullets` | bullet text | LLM: who this update is relevant for (e.g. `project-manager`, `engineering-lead`, `finance-lead`) |
| `recipients` | `email_bullets` | parent email | Inherited — participants from the parent email |
| `embeddings` | `email_bullets` | bullet text | Embedding model — `vector(1536)` per-bullet embedding for fine-grained RAG |

---

### 6.2 Why These Fields — Nothing Missing?

| Field | Why it cannot be left out |
|---|---|
| `summary` (email-level) | Provides a coarse-grained search surface and a human-readable digest at the email level. Without it, `eb.emails` has no LLM intelligence — it is only a header index. |
| `role` (per-bullet) | The Distribution Engine (Component 2) routes bullets to the right users. Without `role`, every bullet reaches everyone — targeted signals become noise. |
| `sentiment_value` + `sentiment_score` (per-bullet) | Sentiment varies within a single email — one bullet can be positive (milestone hit) while another is negative (blocker raised). Per-bullet sentiment preserves signal fidelity. |
| `hashtags` (per-bullet) | Enables tag-based filtering at bullet granularity for RAG queries and org-level pattern detection across sync runs. |
| `type` (per-bullet) | Distinguishes actionable (`alert`, `escalation`) from informational (`information`, `insight`) bullets — critical for displaying the correct signals on CXO dashboards. |
| `category` (per-bullet) | Enables role-filtered views: a PM wants `status-update` bullets; an engineering lead wants `bug-report`. Without category, views cannot be filtered meaningfully. |
| `attachments` (email-level) | A contract, invoice, or design doc is meaningful context even without file content being read. Captured as filename + MIME type only. |
| `embeddings` (both levels) | Email-level embeddings support broad semantic queries; bullet-level embeddings support precise RAG retrieval. Without both, search depth is limited to one granularity. |

---

## 7. Database Schemas

PostgreSQL 15+ with `pgvector` (embeddings). All tables live in the `eb` schema.

```sql
CREATE EXTENSION IF NOT EXISTS "vector";
```

---

### 7.0 `eb.projects`

Canonical project registry. Written by admin / provisioning. The `project_id` is the short canonical key (e.g. `EB`, `IIRM`, `Optinn`) used as FK across `eb.emails` and all downstream pipeline tables. The Step 2 LLM enrichment prompt receives the current list of valid `project_id` values to constrain its inference.

```sql
CREATE TABLE eb.projects (
    project_id      TEXT        PRIMARY KEY,
    -- Short canonical label, e.g. 'EB', 'IIRM', 'Optinn'

    project_name    TEXT        NOT NULL,
    -- Full display name, e.g. 'Enterprise Brain', 'IIRM Platform', 'Optinn'

    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX idx_projects_name ON eb.projects (project_name);
```

---

### 7.1 `eb.emails`

Written during Step 2. One row per unique email. Stores standard header fields, `project_id` (FK → `eb.projects`), an LLM-generated summary, and a semantic embedding. Parent record for all bullet rows.

```sql
CREATE TYPE eb.thread_position AS ENUM ('original', 'reply', 'forward');

CREATE TABLE eb.emails (
    -- Identity
    id                  TEXT        PRIMARY KEY,       -- Gmail message_id
    thread_id           TEXT,
    email_link          TEXT,                          -- Deep link to Gmail message

    -- Headers (direct from API)
    user_id             TEXT        NOT NULL,          -- org user whose inbox provided this email
    "from"              TEXT        NOT NULL,
    recipients          TEXT[]      NOT NULL DEFAULT '{}',   -- merged To + CC + BCC
    subject             TEXT,
    created_at          TIMESTAMPTZ NOT NULL,          -- email send timestamp

    -- Project mapping
    project_id          TEXT        NOT NULL REFERENCES eb.projects (project_id),
    -- FK to canonical project registry; resolved by LLM inference in Step 2

    -- Pipeline metadata
    enriched_at         TIMESTAMPTZ NOT NULL DEFAULT now(),

    -- Additional columns
    attachments         JSONB       NOT NULL DEFAULT '[]',
    -- [{ "filename": "q2-quote.pdf", "mime_type": "application/pdf", "size_bytes": 120000 }]

    summary             TEXT,                          -- LLM-generated overall email summary

    embeddings          vector(1536)
    -- generated from summary text; email-level semantic embedding
);

CREATE INDEX idx_emails_project_id      ON eb.emails (project_id);
CREATE INDEX idx_emails_user_id         ON eb.emails (user_id);
CREATE INDEX idx_emails_thread_id       ON eb.emails (thread_id);
CREATE INDEX idx_emails_created_at      ON eb.emails (created_at DESC);
CREATE INDEX idx_emails_recipients_gin  ON eb.emails USING GIN (recipients);

-- Semantic search (IVFFlat — tune lists param after data volume is known)
CREATE INDEX idx_emails_embeddings_ivf  ON eb.emails
    USING ivfflat (embeddings vector_cosine_ops) WITH (lists = 100);
```

---

### 7.2 `eb.email_bullets`

Written during Step 2 in the same transaction as `eb.emails`. One row per bullet point. Each row is a fully independent intelligence unit. FK → `eb.emails.id`.

```sql
CREATE TYPE eb.bullet_type      AS ENUM ('information', 'alert', 'insight', 'escalation');
CREATE TYPE eb.bullet_category  AS ENUM (
    'bug-report', 'release-update', 'meeting-notes',
    'status-update', 'announcement', 'general'
);
CREATE TYPE eb.sentiment_value  AS ENUM ('positive', 'neutral', 'negative');

CREATE TABLE eb.email_bullets (
    -- Identity
    bullet_id           UUID        PRIMARY KEY DEFAULT gen_random_uuid(),
    email_id            TEXT        NOT NULL REFERENCES eb.emails (id) ON DELETE CASCADE,

    -- Bullet content
    text                TEXT        NOT NULL,
    hashtags            TEXT[]      NOT NULL DEFAULT '{}',

    -- Per-bullet LLM classification
    sentiment_value     eb.sentiment_value  NOT NULL,
    sentiment_score     NUMERIC(4,2) NOT NULL CHECK (sentiment_score BETWEEN -1.00 AND 1.00),
    type                eb.bullet_type      NOT NULL,
    category            eb.bullet_category  NOT NULL,
    role                TEXT        NOT NULL,          -- who this update is relevant for

    -- Inherited context (denormalised for query convenience)
    project_id          TEXT        NOT NULL REFERENCES eb.projects (project_id),
    -- denormalized from parent email; avoids JOIN on hot query paths
    user_id             TEXT        NOT NULL,          -- from parent email
    recipients          TEXT[]      NOT NULL DEFAULT '{}',   -- from parent email

    -- Additional columns
    embeddings          vector(1536)
    -- per-bullet semantic embedding for fine-grained RAG search
);

CREATE INDEX idx_bullets_email_id        ON eb.email_bullets (email_id);
CREATE INDEX idx_bullets_position        ON eb.email_bullets (email_id, position);
CREATE INDEX idx_bullets_sentiment       ON eb.email_bullets (sentiment_value);
CREATE INDEX idx_bullets_type            ON eb.email_bullets (type);
CREATE INDEX idx_bullets_urgency         ON eb.email_bullets (urgency_score DESC);
CREATE INDEX idx_bullets_hashtags_gin    ON eb.email_bullets USING GIN (hashtags);
CREATE INDEX idx_bullets_recipients_gin  ON eb.email_bullets USING GIN (recipients);

-- Per-bullet semantic search
CREATE INDEX idx_bullets_embeddings_ivf  ON eb.email_bullets
    USING ivfflat (embeddings vector_cosine_ops) WITH (lists = 100);
```

---

### 7.3 `eb.org_project_snapshot`

Written by the RT Agent. **One row per project per source** — one row for Gmail, a separate row for Jira, etc. When a new sync run comes in for a source, the RT Agent updates the existing row for that `(project_id, source)` pair rather than inserting a new one.

**No bullet text is stored here** — only `bullet_id` references back to the source bullet tables. All per-bullet intelligence (`role`, `recipients`, `user_id`, `sentiment_value`, `type`) stays on the original rows; the Distribution Engine fetches those directly for per-user/per-role filtering.

```sql
CREATE TABLE eb.org_project_snapshot (
    -- Identity
    id                      UUID        PRIMARY KEY DEFAULT gen_random_uuid(),

    project_id              TEXT        NOT NULL REFERENCES eb.projects (project_id),
    source                  TEXT        NOT NULL,
    -- e.g. 'gmail', 'jira', 'gchat' — one row per source per project

    UNIQUE (project_id, source),
    -- RT Agent does UPDATE WHERE id = ? when a source syncs again

    -- Timestamps
    last_updated            TIMESTAMPTZ NOT NULL DEFAULT now(),
    previous_snapshot_at    TIMESTAMPTZ,
    -- NULL on first run; set to last_updated of the snapshot this replaced

    -- Bullet references for this source
    content                 JSONB       NOT NULL DEFAULT '{}'
    -- {
    --   "bullet_ids":          ["uuid1", "uuid2", "uuid3"],
    --   "new_bullet_ids":      ["uuid3"],        -- new since previous_snapshot_at
    --   "resolved_bullet_ids": ["uuid-old-1"]    -- in previous snapshot, absent now
    -- }
);

CREATE INDEX idx_ops_project_id  ON eb.org_project_snapshot (project_id);
CREATE INDEX idx_ops_source      ON eb.org_project_snapshot (source);
CREATE INDEX idx_ops_last_updated ON eb.org_project_snapshot (last_updated DESC);
```

**What a populated snapshot looks like** — after the RT Agent runs for project `EB`:

```
Row 1:
  id:                   "a1b2c3d4-..."
  project_id:           "EB"
  source:               "gmail"
  last_updated:         "2026-03-24T11:00:00Z"
  previous_snapshot_at: "2026-03-23T11:00:00Z"
  content: {
    bullet_ids:          ["uuid1", "uuid2", "uuid3", "uuid4", "uuid5"],
    new_bullet_ids:      ["uuid4", "uuid5"],
    resolved_bullet_ids: []
  }

Row 2:
  id:                   "e5f6g7h8-..."
  project_id:           "EB"
  source:               "jira"
  last_updated:         "2026-03-24T11:00:00Z"
  previous_snapshot_at: "2026-03-23T11:00:00Z"
  content: {
    bullet_ids:          ["uuid6", "uuid7", "uuid8"],
    new_bullet_ids:      ["uuid8"],
    resolved_bullet_ids: ["uuid-old-1"]
  }
```

The Distribution Engine queries `WHERE project_id = 'EB'` to get all source rows, then fetches the actual bullet rows using the `bullet_ids` from each source's `content`.

---

## 8. Open Items

1. **Project label canonicalisation** — The LLM must infer `project_id` values that match `eb.projects.project_id` exactly. The Step 2 enrichment prompt must receive the current list of valid `project_id` values so it selects the closest match rather than generating a free-form label. A fallback strategy for unrecognised project signals (e.g., a quarantine bucket or an `unknown` placeholder) needs to be defined before the pipeline is production-ready.

2. **Hashtag taxonomy** — Whether per-bullet content hashtags (`#bug`, `#finance`, `#deadline`) are a fixed enum or LLM-generated open strings needs to be decided. A fixed taxonomy makes RT Agent insight derivation deterministic and prevents tag drift across sync runs.


6. ~~**RT Agent: aggregate-and-diff vs. summarise**~~ — **Resolved.** The RT Agent does **aggregate-and-diff only** — no LLM call, no text generation. Bullet data is never modified or summarised by the RT Agent. All intelligence stays on the original bullet rows. The Distribution Engine is responsible for per-user/per-role filtering by querying bullets directly.

7. ~~**`eb.org_project_snapshot` schema**~~ — **Resolved.** One row per `(project_id, source)` — Gmail and Jira each get their own row. RT Agent updates the existing row on each sync run (`UPDATE WHERE id = ?`). Scalar signals (`urgency_max`, `dominant_type`, hashtags, participants) sit on each source row; `content` JSONB holds the bullet_id references and diff. Schema finalised in Section 7.3.