# DEB-362: Data Source Agent — Query-Plan Cache LLD

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

This document defines a **data-source-agnostic agent architecture** built around a **query-plan cache**.

The pattern applies uniformly to any structured or semi-structured data source (email, project management, databases, documents, etc.). No part of this design is specific to a single connector.

**Core architectural principle:**
- The **data source is always the ultimate target**. Every answer is grounded in a live query against the real data source.
- The **cache stores query plans** (structured, parameterized queries), not pre-computed answers.
- An **LLM-cache-agent** converts a natural-language question into a semantically indexed key and a parameterized query value.
- An **LLM-QA-agent** retrieves the top-k matching query plans from cache, executes them against the data source, and generates a grounded, cited answer.

---

## 1. Goals

**Stage 1 (this document):** Prove the Q→K→V→A pipeline end-to-end against a stub data source.
- Implement `plan_cache_service`, `cache_agent`, `query_executor`, and `qa_agent` as a self-contained, testable pipeline
- Use a **realistic stub connector** (in-memory or fixture-backed) that returns real query shapes without live API calls
- Validate plan generation, key embedding, top-k lookup, param filling, and answer generation
- Enforce all cache validity rules in Python service code (not by the LLM)
- Preserve citations and source traceability for every answer

**Stage 2 (separate LLD):** Replace the stub with a real connector.
- Add `connector.py`, `sync_service.py`, auth, and Alembic migrations
- Keep connector logic isolated so it can be swapped in without touching any pipeline code

## 2. Non-Goals (Stage 1)

- No real connector, no live API calls
- No auth or token management
- No sync scheduler or incremental/full sync
- No Alembic migrations (schema created directly in test setup)
- No Chanakya orchestration changes
- No multi-source synthesis
- No connector-specific push notifications or attachment parsing

## 3. Current Baseline

- Stack: `pydantic-ai`, `asyncpg`, `pgvector`, `sqlalchemy` (already in `pyproject.toml`)
- Existing agents use mock data; Stage 1 keeps a stub but wires it into the real pipeline
- Stage 1 stub must use **realistic query shapes** (e.g. actual JQL / SQL fragments with named params) so that the pipeline is exercised against the real template structure before the live connector lands

---

## 4. Two-Agent Architecture

The system is built around two distinct LLM agents with separate responsibilities:

| Agent | Role | What it stores / returns |
|---|---|---|
| **LLM-cache-agent** | Translates a natural-language question into a semantically addressable key and a parameterized data-source query | Writes `(key_embedding, parameterized_query, TTL)` into the query-plan cache |
| **LLM-QA-agent** | Receives a question, retrieves top-k matching query plans from cache, executes them against the live data source, and generates a grounded answer | Returns a cited answer |

### Cache Structure (Key → Value)

```
Key   : question intent, short canonical phrase — EMBEDDED (pgvector), no chunking needed
Value : parameterized query template — NOT embedded, stored as plain JSONB
TTL   : expiry timestamp per entry
```

**Runtime pipeline:** `Q → K → V → A`

| Step | What happens |
|---|---|
| **Q** | User asks a natural-language question |
| **→ K** | LLM-QA-Agent embeds the question and finds top-k matching intent keys in the plan cache |
| **→ V** | Matched keys return their parameterized query templates (Values) |
| **→ A** | Query Executor fills parameters and runs the query against the live data source; LLM generates the cited answer |

**Contrast with the naïve Q → A flow:**  
Without the cache, every question triggers a fresh LLM call to produce a query from scratch.  
With the plan cache, semantically equivalent questions (e.g. asking about project DEB vs. project RIT) reuse the same query template — only the parameter values differ.

**The cache is a plan cache, not an answer cache.** The data source is always queried at answer time.

---

## 5. Standalone Data Source Agent Architecture

```mermaid
flowchart LR
    U[User or Chanakya] --> QA[LLM-QA-Agent]
    QA --> PC[Plan Cache Service]
    PC --> DB[(Postgres: plan_cache_entries)]
    PC -->|top-k plan hits| QA
    QA --> EX[Query Executor]
    EX --> DS[(Data Source)]
    DS --> EX
    EX --> QA
    QA --> ANS[Answer + Citations]

    CA[LLM-Cache-Agent] --> PC
    CA --> DB

    S[APScheduler] --> SS[Sync Service]
    SS --> CN[Connector]
    CN --> DS
    CN --> N[Normalizer]
    N --> MSG[(Postgres: source_records)]
    N --> EMB[Embedder]
    EMB --> MSG
```

---

## 5a. Query-Time Flow (LLM-QA-Agent)

```mermaid
sequenceDiagram
    participant Q as Query Caller
    participant QA as LLM-QA-Agent
    participant PC as Plan Cache Service
    participant DB as Postgres
    participant CA as LLM-Cache-Agent
    participant EX as Query Executor
    participant DS as Data Source

    Q->>QA: answer_question(question)
    QA->>PC: lookup(question, tenant_id, top_k=5)
    PC->>DB: embed(question) → cosine search plan_cache_entries<br/>WHERE tenant_id = ? AND expires_at > now()<br/>ORDER BY similarity DESC LIMIT top_k
    DB-->>PC: top-k (key_text, parameterized_query) rows

    alt plan cache hit(s) found
        PC-->>QA: list of parameterized queries
        QA->>EX: execute(parameterized_queries, question)
        EX->>DS: run queries against live data source
        DS-->>EX: result records
        EX-->>QA: grounded context + source record ids
        QA->>QA: generate_answer(question, context)
        QA-->>Q: answer + citations
    else no plan in cache
        QA->>CA: build_plan(question, tenant_id)
        CA->>CA: LLM call: question → key phrase + parameterized query
        CA->>PC: store(key_text, parameterized_query, TTL)
        PC->>DB: INSERT plan_cache_entries
        CA-->>QA: parameterized_query
        QA->>EX: execute(parameterized_query, question)
        EX->>DS: run query against live data source
        DS-->>EX: result records
        EX-->>QA: grounded context + source record ids
        QA->>QA: generate_answer(question, context)
        QA-->>Q: answer + citations
    end
```

---

## 5b. Plan Generation Flow (LLM-Cache-Agent)

The LLM-cache-agent is responsible for one task: translate a question into a reusable, parameterized query plan.

**Without cache — naïve Q → A:**
```
Q: "what is the update since yesterday for DEB project"

LLM-QA produces directly:
A: SELECT * FROM JIRA WHERE project = 'DEB' AND updated >= now() - interval '1 day'
```
This query is generated from scratch on every call and is not reusable for any other project.

---

**With plan cache — Q → K → V → A:**

```
Q: "what is the update since yesterday for DEB project"

LLM-Cache-Agent stores:

K (embedded):
  "when user asks for an update in a project for a time range"

V (not embedded, stored as JSONB):
  SELECT * FROM JIRA
  WHERE project = {project_name}
    AND updated > {start_datetime}
    AND updated < {end_datetime}

TTL: now() + PLAN_CACHE_TTL_HOURS
```

**Reuse on the next semantically similar question:**
```
Q: "what is the update since yesterday for RIT project"

→ embed(Q) matches K in cache
→ V retrieved: same query template
→ Executor fills: project_name=RIT, start_datetime=yesterday_00:00, end_datetime=now()
→ Query runs against live data source → answer generated
```

The key captures the **intent pattern** (semantic, embedded for similarity search).  
The value is a **structural query template with named parameters** — the QA-agent fills parameters at runtime from the question context.

**Rules enforced in code (not by LLM):**
- Keys are short (≤ 30 tokens). No chunking needed — embed whole.
- Values are never embedded. They are plain JSONB stored as-is.
- One question may resolve to multiple query plans (top-k) which are all executed and merged.

---

## 6. Sync and Ingestion Flow

> **Stage 2 only.** Sync, connector auth, and record persistence are deferred. In Stage 1 the `stub_connector.py` returns fixture records directly to the `query_executor` without any DB persistence or scheduling. The sync flow and schema for `source_records` / `sync_state` tables will be defined in the Stage 2 LLD.

---

---

## 7. Internal Module Layout

### Stage 1 — Pipeline modules (this LLD)

```
backend/chanakya/<source_name>/
├── agent.py               # EnterpriseAgent entrypoint, exposes answer_question tool
├── stub_connector.py      # Fixture-backed stub returning realistic query-shaped records (Stage 1 only)
├── repository.py          # Postgres reads/writes for plan_cache_entries
├── embeddings.py          # Calls gemini-embedding-001, batched, 1536-dim
├── query_executor.py      # Fills parameterized query templates, calls connector, returns records
├── plan_cache_service.py  # lookup (top-k), store, TTL expiry, invalidation
├── cache_agent.py         # LLM-cache-agent: question → key phrase + parameterized query template
├── qa_agent.py            # LLM-QA-agent: question + plans → execute → generate cited answer
└── schemas.py             # Pydantic DTOs: SourceRecord, PlanCacheEntry, QueryPlan, AnswerResult
```

### Stage 2 — Connector modules (separate LLD, added later)

```
backend/chanakya/<source_name>/
├── connector.py           # Authenticated client for the real data source, API wrappers
├── sync_service.py        # full_sync / incremental_sync, cursor-expiry fallback
└── (stub_connector.py removed)
```

---

### Component Responsibilities (Stage 1)

- **`agent.py`** — Registers `answer_question` tool with the `EnterpriseAgent`. Contains no business logic.
- **`stub_connector.py`** — Returns fixture records with realistic query shapes. Replaced by `connector.py` in Stage 2.
- **`cache_agent.py`** — LLM call that produces a `(key_text, parameterized_query)` pair. Key is a short intent phrase (≤ 30 tokens). Value is a structured query template with named `{params}`. Calls `plan_cache_service.store()` after generation.
- **`plan_cache_service.py`** — `lookup()` embeds the incoming question and does a pgvector top-k search over `plan_cache_entries.key_embedding`. Returns list of `QueryPlan`. `store()` embeds only the key phrase. `invalidate_plans_for_records()` deletes plans whose `dependency_record_ids` overlap with specified record ids.
- **`query_executor.py`** — Receives a `QueryPlan` and the current question context. Fills named parameters in the query template (LLM-assisted extraction of param values). Calls the connector (stub in Stage 1, real in Stage 2). Returns raw records.
- **`qa_agent.py`** — Orchestrates: lookup → execute → generate. Always hits the connector. LLM call only for final answer generation and param extraction.
- **`embeddings.py`** — Whole-record embedding (no chunking). Batches API calls. Used by `plan_cache_service` for key phrases; also used by Stage 2 sync for persisting record embeddings.

---

## 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.

**Stage 1 migration:** `plan_cache_entries` only.  
**Stage 2 migrations:** `sync_state`, `source_records`.

---

### 8.1 Sync State — *Stage 2*

```sql
CREATE TABLE sync_state (
    id                       UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id                TEXT NOT NULL,
    source_type              TEXT NOT NULL,         -- identifies the connector, e.g. 'gmail', 'jira'
    sync_cursor              TEXT,                  -- NULL before first sync; connector-specific cursor/token
    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(),
    UNIQUE (tenant_id, source_type)
);
```

---

### 8.2 Source Records — *Stage 2*

> No chunks table. Each record is embedded whole.

```sql
CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE source_records (
    id               UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id        TEXT NOT NULL,
    source_type      TEXT NOT NULL,          -- connector identifier
    source_record_id TEXT NOT NULL,          -- native ID from the data source
    title            TEXT,                   -- subject, ticket title, document name, etc.
    body_text        TEXT,                   -- normalized plain text; markup stripped before storage
    -- whole-record embedding (no chunking); gemini-embedding-001, 1536-dim
    embedding        vector(1536),
    record_date      TIMESTAMPTZ,            -- creation or last-update timestamp from source
    metadata         JSONB,                  -- connector-specific fields (labels, status, assignee, etc.)
    raw_payload      JSONB,                  -- full source 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, source_type, source_record_id)
);

CREATE INDEX ON source_records USING ivfflat (embedding vector_cosine_ops)
    WITH (lists = 100);
CREATE INDEX ON source_records (tenant_id, source_type, record_date DESC);
```

---

### 8.3 Query-Plan Cache — *Stage 1*

```sql
CREATE TABLE plan_cache_entries (
    id                      UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id               TEXT NOT NULL,
    source_type             TEXT NOT NULL,  -- connector identifier, e.g. 'gmail', 'jira', 'gdrive'
    -- Key: short intent phrase, EMBEDDED for similarity search
    key_text                TEXT NOT NULL,
    key_embedding           vector(1536) NOT NULL,
    -- Value: parameterized query template, NOT embedded — plain structured text
    parameterized_query     JSONB NOT NULL,  -- { "query": "assignee:{user} ...", "params": ["user", ...] }
    dependency_record_ids   JSONB,           -- [uuid, ...] of source_records sampled when plan was built (nullable)
    expires_at              TIMESTAMPTZ NOT NULL,
    hit_count               INT NOT NULL DEFAULT 0,
    last_hit_at             TIMESTAMPTZ,
    created_at              TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Index for key similarity search
CREATE INDEX ON plan_cache_entries USING ivfflat (key_embedding vector_cosine_ops)
    WITH (lists = 50);
CREATE INDEX ON plan_cache_entries (tenant_id, source_type, expires_at);
```

**Key constraint:** `key_text` is always ≤ 30 tokens. The LLM-cache-agent is instructed to produce short, canonical intent phrases (e.g. `"records assigned to person about topic in time range"`).  
**Value constraint:** `parameterized_query` is never embedded. It is retrieved by key match and executed at runtime.

---

### 8.4 Entity Relationship Diagram

> Stage 1 only involves `plan_cache_entries`. `sync_state` and `source_records` are Stage 2.

```mermaid
erDiagram
    sync_state {
        uuid id PK
        text tenant_id
        text source_type
        text sync_cursor
        text sync_status
    }

    source_records {
        uuid id PK
        text tenant_id
        text source_type
        text source_record_id
        text title
        text body_text
        vector_1536 embedding
        jsonb metadata
        jsonb raw_payload
    }

    plan_cache_entries {
        uuid id PK
        text tenant_id
        text source_type
        text key_text
        vector_1536 key_embedding
        jsonb parameterized_query
        jsonb dependency_record_ids
        timestamptz expires_at
    }

    source_records }o--o{ plan_cache_entries : "sampled_when_plan_built"
```

---

## 9. Plan Cache Validity Contract

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

1. `tenant_id` matches the current request
2. `source_type` matches the data source being queried
3. `cosine_similarity(embed(question), key_embedding) >= PLAN_CACHE_SIMILARITY_THRESHOLD` (default: `0.88`)
4. `expires_at > now()`

A plan cache entry is **stale and skipped** when:
- `expires_at <= now()`

**Plan cache write suppressed when:**
- LLM-cache-agent call raised an exception
- Generated `parameterized_query` is empty or malformed
- `key_text` exceeds 30 tokens (plan rejected as too broad)

**Plan cache invalidation on sync:**  
After each incremental sync, `plan_cache_service.invalidate_plans_for_records(changed_record_ids)` deletes all plan entries where `dependency_record_ids` contains any of the changed UUIDs. This handles cases where the plan was built using sample records that are now stale.

```sql
DELETE FROM plan_cache_entries
WHERE tenant_id = $1
  AND source_type = $2
  AND dependency_record_ids ?| ARRAY[$3, $4, ...]::text[];
```

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

> **Key design invariant:** The cache never stores answers. On every cache hit the real data source is always queried. The cache only accelerates query planning — not result retrieval.

---

## 10. Embedding Specification

| Parameter | Value |
|---|---|
| Model | `gemini-embedding-001` |
| Dimensions | 1536 |
| What is embedded | (1) Full record `body_text` at sync time &nbsp; (2) `key_text` (short intent phrase, ≤30 tokens) at plan-cache-write time |
| What is NOT embedded | `parameterized_query` values — never embedded, stored as plain `JSONB` |
| Batch size | 100 items per API call |
| Provider | Via LiteLLM proxy using `LITELLM_PROVIDER_BASE_URL` |
| Index type | `ivfflat` with `vector_cosine_ops`, `lists=100` for source records, `lists=50` for plan cache keys |

> **No chunking.** Source records are short by nature. Embedding the full normalized body provides sufficient semantic coverage without the complexity of chunk management, chunk-to-record joins, or chunk invalidation.

---

## 11. Agent API Contract

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

```python
@source_agent.tool
async def answer_question(ctx: RunContext[RequestContext], question: str) -> str:
    """Answer a question about the data source. Always queries live data. Plan cache accelerates query planning."""

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

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

```python
# plan_cache_service.py
async def lookup(question: str, tenant_id: str, source_type: str, top_k: int = 5) -> list[QueryPlan]
async def store(key_text: str, parameterized_query: dict, tenant_id: str, source_type: str, dependency_record_ids: list[str] | None = None) -> None
async def invalidate_plans_for_records(tenant_id: str, source_type: str, record_ids: list[str]) -> int

# cache_agent.py  (LLM-cache-agent)
async def build_plan(question: str, tenant_id: str, source_type: str) -> QueryPlan
# QueryPlan = { key_text: str, parameterized_query: dict }

# query_executor.py
async def execute(plans: list[QueryPlan], question: str, tenant_id: str) -> list[SourceRecord]

# qa_agent.py  (LLM-QA-agent)
async def answer(question: str, tenant_id: str, source_type: str) -> AnswerResult
# AnswerResult = { answer_text: str, citations: list[Citation] }

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

---

## 12. Sync Scheduler — *Stage 2*

> Deferred. The scheduler, `sync_service`, and `connector` are all Stage 2 concerns. Stage 1 has no background jobs.

---

## 13. Failure Handling

> Rows marked *Stage 2* are not applicable until the real connector is introduced.

| Failure | Stage | Behaviour |
|---|---|---|
| Plan cache lookup failure (DB unreachable) | 1 | Log `WARNING`, continue — call LLM-cache-agent to build a fresh plan. Never block the query. |
| LLM-cache-agent failure | 1 | Log `ERROR`, attempt direct query using a default broad filter. Do not write plan cache. |
| `key_text` > 30 tokens from LLM-cache-agent | 1 | Reject plan, log `WARNING`. Do not write cache. Attempt query with broad unparameterized fallback. |
| Query executor failure (connector error) | 1 | Log `ERROR`, return "data source unavailable" message. |
| Empty results from connector | 1 | Return "no relevant records found" message. Do not write plan cache. |
| Answer generation failure | 1 | Log `ERROR`, return error message. |
| `AuthError` (credentials expired or revoked) | 2 | Return operational error message to caller. Do not write plan cache. Log `ERROR`. |
| `SyncCursorExpiredError` (source cursor stale) | 2 | Automatically fall back to `full_sync`. Update `sync_state.error_text` temporarily. |


---

## 14. Deferred Chanakya Integration

After a standalone data source agent is stable, Chanakya becomes a thin delegator:

```mermaid
flowchart LR
    C[Chanakya] --> G[Data Source Agent A2A]
    G --> QA[LLM-QA-Agent]
    QA --> PC[Plan Cache]
    QA --> DS[(Data Source)]
    PC --> CA[LLM-Cache-Agent]
```

No plan cache logic and no query execution logic should be implemented in Chanakya. Chanakya calls `answer_question` over A2A and receives the final grounded answer.

---

## 15. Testing Strategy

> Stage 1 tests cover the full pipeline against the stub connector. Stage 2 tests (connector, sync, record persistence) are defined in the Stage 2 LLD.

### Unit Tests (Stage 1)

| Test | What it verifies |
|---|---|
| `test_plan_cache_hit_skips_cache_agent` | `plan_cache_service.lookup()` returns plans; `cache_agent.build_plan` is never called |
| `test_plan_cache_miss_triggers_build_plan` | `lookup()` returns empty; `cache_agent.build_plan` is called, plan is stored |
| `test_expired_plan_rejected` | Plan entry with `expires_at` in the past is not returned by `lookup()` |
| `test_key_too_long_rejected` | `key_text` > 30 tokens causes `store()` to raise `PlanKeyTooLongError` |
| `test_query_executor_fills_params` | Executor correctly fills `{user}`, `{topic}`, `{date}` params from question context |
| `test_empty_source_results_no_plan_write` | Zero records from data source → `store()` is never called |
| `test_sync_upsert_invalidates_plans` | After `incremental_sync`, plans referencing changed records are deleted |
| `test_cursor_expiry_fallback_to_full_sync` | `SyncCursorExpiredError` triggers `full_sync` — **Stage 2** |
| `test_whole_record_embedding_no_chunks` | Sync stores embedding on `source_records.embedding`; no chunks table written — **Stage 2** |

### Integration Tests (Stage 1)

| Test | What it verifies |
|---|---|
| `test_plan_cache_lookup_returns_similar_plan` | A stored plan is returned for a semantically similar (not identical) question |
| `test_answer_question_always_hits_connector` | Full path via `answer_question` tool calls query executor against stub connector fixture |
| `test_second_question_uses_cached_plan` | Second similar question returns same plan from cache (confirmed by mock assertion on `build_plan`) |
| `test_plan_invalidated_explicitly` | Plan is absent from `lookup()` after `invalidate_plans_for_records()` is called directly |

### Test Infrastructure

- Use `pytest` with `pytest-asyncio`
- Postgres: `pytest-postgresql` or Docker Compose test service
- Connector: `stub_connector.py` with fixture payloads in `tests/fixtures/<source_name>/` (realistic query shapes, not random data)
- Embedding API: mock returning deterministic 1536-dim float vectors
- LLM calls (cache-agent + qa-agent): mock with fixed response fixtures

---

## 16. Environment Variables Reference

### Stage 1

| Variable | Required | Default | Description |
|---|---|---|---|
| `TENANT_ID` | Yes | — | Tenant identifier |
| `SOURCE_TYPE` | Yes | — | Connector identifier (e.g. `jira`, `gmail`) |
| `PLAN_CACHE_TTL_HOURS` | No | `24` | Plan cache entry TTL |
| `PLAN_CACHE_SIMILARITY_THRESHOLD` | No | `0.88` | Minimum cosine similarity for a plan cache hit |
| `PLAN_CACHE_TOP_K` | No | `5` | Number of top plan candidates retrieved per question |
| `LITELLM_PROVIDER_BASE_URL` | Yes | — | LiteLLM proxy for LLM calls (QA-agent + cache-agent) |
| `DATABASE_URL` | Yes | — | Postgres connection string |

### Stage 2 (added with connector)

| Variable | Required | Default | Description |
|---|---|---|---|
| `SYNC_INTERVAL_MINUTES` | No | `15` | APScheduler polling interval |

---

## 17. Delivery Phases

### Stage 1 — Pipeline (this LLD)

Goal: prove the full Q→K→V→A loop works correctly before any real connector exists.

| Phase | Deliverables |
|---|---|
| **1A — Schema + Stub** | Alembic migration for `plan_cache_entries` only. `stub_connector.py` returning fixture records with realistic query shapes (e.g. JQL / SQL with named params). `repository.py` for plan cache reads/writes. |
| **1B — Cache Agent** | `cache_agent.py`: LLM call: question → `key_text` + `parameterized_query`. `plan_cache_service.py`: `store()`, `lookup()` (top-k pgvector search), `invalidate()`. |
| **1C — QA Agent + Executor** | `query_executor.py`: fills named params from question context, calls stub connector. `qa_agent.py`: orchestrates lookup → execute → generate cited answer. |
| **1D — Agent Wire-up** | `agent.py` with `answer_question` tool backed by real pipeline. End-to-end smoke test. |
| **1E — Tests** | Unit tests for all pipeline components. Integration tests: plan cache hit/miss, param filling, answer generation, cache invalidation. All against stub. |

### Stage 2 — Connector + Sync (separate LLD)

Goal: replace the stub with a real data source connector. No pipeline code changes expected.

| Phase | Deliverables |
|---|---|
| **2A — DB + Connector** | `connector.py` (authenticated API wrappers), `repository.py` extended for `source_records` + `sync_state`. Alembic migrations for both tables. |
| **2B — Sync** | `sync_service.py` (full + incremental + cursor-expiry fallback), body normalization, `embeddings.py` (whole-record embed, no chunking). |
| **2C — Wire-up + Tests** | Swap stub connector for real connector in `agent.py`. Sync scheduler. Integration tests against real connector fixtures. CI Postgres service. |