Skip to content

# Task: RAG ingestion — Wave 1 (Chat messages, Docs, Files)

## Goal

Підключити першу хвилю RAG-ingestion подій до `rag-ingest-worker`, щоб агенти могли робити RAG по:

- чат-повідомленнях (`message.created`),
- документах/wiki (`doc.upserted`),
- файлах (`file.uploaded`),

з урахуванням режимів `public/confidential` та прапору `indexed`.

Wave 1 = MVP RAG: максимум корисного контексту при мінімальній кількості подій.

## Context

- Root: `microdao-daarion/`.
- Базовий воркер: `docs/cursor/rag_ingestion_worker_task.md`.
- Подробиці для перших подій: `docs/cursor/rag_ingestion_events_task.md` (message/doc → IngestChunk).
- Event Catalog: `docs/cursor/42_nats_event_streams_and_event_catalog.md`.
- Privacy/Confidential:
- `docs/cursor/47_messaging_channels_and_privacy_layers.md`
- `docs/cursor/48_teams_access_control_and_confidential_mode.md`

Ingestion-воркер читає події з NATS JetStream (streams типу `STREAM_CHAT`, `STREAM_PROJECT`, `STREAM_TASK` або `teams.*` outbox — згідно актуальної конфігурації).

## 1. Принципи для Wave 1

1. Тільки доменні події, не CRUD по БД:
2. `message.created`, `doc.upserted`, `file.uploaded`.
3. Поважати `mode` та `indexed`:
4. індексувати тільки якщо `indexed = true`;
5. plaintext зберігати тільки для `public` (для `confidential` — embeddings/summary без відкритого тексту, згідно політики).
6. Мінімальний, але стандартний payload:
7. `team_id`, `channel_id` або `project_id`,
8. `mode` (`public | confidential`),
9. `author_user_id` / `author_agent_id`,
10. `created_at` / `updated_at`,
11. `kind` / `doc_type`,
12. `indexed` (bool),
13. `source_ref` (ID оригінальної сутності).

Ці принципи мають бути відображені як у схемах подій, так і в нормалізації → IngestChunk.

## 2. Event contracts (Wave 1)

### 2.1. `message.created`

Джерело: Messaging service (`STREAM_CHAT` / outbox для командних просторів).

Використати Event Envelope з `42_nats_event_streams_and_event_catalog.md`, але уточнити payload для RAG:

- Subject/type (рекомендовано): `chat.message.created`.
- Envelope:
- `meta.team_id` — DAO / команда.
- `payload.message_id`.
- `payload.channel_id`.
- `payload.author_user_id` або `payload.author_agent_id`.
- `payload.mode`: `public | confidential`.
- `payload.kind`: `text | image | file | system`.
- `payload.thread_id` (optional).
- `payload.created_at`.
- `payload.indexed`: bool (derived: mode + налаштування каналу).
- `payload.text_summary` / `payload.text_plain` (залежно від політики збереження plaintext).

RAG-правила:

- індексувати тільки якщо `payload.indexed = true`;
- якщо `kind != "text"` — пропускати в Wave 1 (image/audio/pdf покриваються через `file.uploaded`);
- якщо `mode = "confidential"` — не зберігати plaintext в Milvus metadata, тільки embeddings + мінімальні метадані.

### 2.2. `doc.upserted`

Джерело: Docs/Wiki/Co-Memory сервіс (`STREAM_PROJECT` або окремий docs-stream).

Рекомендований payload для RAG:

- `payload.doc_id`
- `payload.team_id`
- `payload.project_id`
- `payload.path` (wiki path/tree)
- `payload.title`
- `payload.text` (може бути великий)
- `payload.mode`: `public | confidential`
- `payload.indexed`: bool
- `payload.labels` / `payload.tags` (optional)
- `payload.updated_at`

RAG-правила:

- індексувати тільки якщо `indexed = true`;
- для великих текстів — розбивати на чанки (512–1024 символів/токенів);
- `mode = "confidential"` → embeddings без відкритого тексту.

### 2.3. `file.uploaded`

Джерело: Files/Co-Memory (`files` таблиця, окремий стрім або частина STREAM_PROJECT/STREAM_CHAT).

Рекомендований payload:

- `payload.file_id`
- `payload.owner_team_id`
- `payload.size`
- `payload.mime`
- `payload.storage_key`
- `payload.mode`: `public | confidential`
- `payload.indexed`: bool
- `payload.enc`: bool (чи зашифрований в storage)
- `payload.linked_to`: `{message_id|project_id|doc_id}`
- `payload.extracted_text_ref` (ключ до вже пропаршеного тексту, якщо є)

RAG-правила:

- індексувати тільки якщо `indexed = true` та `mime` ∈ текстових/документних форматів (`text/*`, `application/pdf`, `markdown`, тощо);
- якщо текст ще не витягнутий — створити ingestion-джоб (черга/OCR) і не індексувати до появи `file.text_parsed`/`file.text_ready` (це може бути окремий event у Wave 1 або 1.5).

## 3. Зміни в `rag-ingest-worker`

### 3.1. Routing / підписки

У `services/rag-ingest-worker/events/consumer.py`:

1. Додати (або уточнити) підписки на subjects для Wave 1:
2. `chat.message.created`
3. `doc.upserted` (назву узгодити з фактичним стрімом — напр. `project.doc.upserted`)
4. `file.uploaded`

5. Ввести routing таблицю (може бути dict):

6. `"chat.message.created" → handle_message_created`

7. `"doc.upserted" → handle_doc_upserted`

8. `"file.uploaded" → handle_file_uploaded`

9. Кожен handler повинен:

10. розпарсити envelope (`event`, `meta.team_id`, `payload`),
11. перевірити `indexed` та `mode`,
12. викликати відповідну функцію нормалізації з `pipeline/normalization.py`,
13. віддати chunks в embedding + Milvus + Neo4j.

### 3.2. Нормалізація у `pipeline/normalization.py`

Розширити/уточнити:

- `async def normalize_message_created(event: dict) -> list[IngestChunk]:`
- орієнтуватися на схему з `rag_ingestion_events_task.md` + тепер додати перевірку `indexed`/`mode`;

- повертати 0 чанків, якщо `indexed = false` або `kind != "text"`.

- `async def normalize_doc_upserted(event: dict) -> list[IngestChunk]:`

- аналогічно до `normalize_doc_upsert` з `rag_ingestion_events_task.md`, але з полями `indexed`, `mode`, `labels`;

- розбивати довгі тексти.

- `async def normalize_file_uploaded(event: dict) -> list[IngestChunk]:`

- якщо текст уже доступний (через `extracted_text_ref` або інший сервіс) — розбити на чанки;
- якщо ні — поки що повертати `[]` і логувати TODO (інтеграція з parser/Co-Memory).

У всіх нормалізаторах стежити, щоб:

- `chunk_id` був детермінованим (див. `rag_ingestion_worker_task.md`),
- `visibility` / `mode` коректно мапились (public/confidential),
- `source_type` ∈ {`"message"`, `"doc"`, `"file"`},
- метадані включали `team_id`, `channel_id`/`project_id`, `author_id`, `created_at`.

### 3.3. Embeddings + Milvus/Neo4j

У Wave 1 достатньо:

- використовувати вже існуючі пайплайни з `rag_ingestion_worker_task.md`:
- `embedding.embed_chunks(chunks)`
- `index_milvus.upsert_chunks_to_milvus(...)`
- `index_neo4j.update_graph_for_event(event, chunks)` (мінімальний граф: User–Message–Channel, Project–Doc, File–(Message|Doc|Project)).

Головне — ідемпотентний upsert по `chunk_id` (Milvus) та `MERGE` в Neo4j.

## 4. Узгодження з Meilisearch indexer

Хоча цей таск фокусується на RAG (Milvus/Neo4j), потрібно:

1. Переконатися, що логіка `indexed`/`mode` співпадає з існуючим search-indexer (Meilisearch) для:
2. `chat.message.created` / `chat.message.updated`,
3. `doc.upserted`,
4. `file.uploaded` (якщо вже індексується).
5. По можливості, винести спільну функцію/константу для визначення `indexed` (based on channel/project settings), щоб RAG та Meilisearch не роз’їхались.

## 5. Тестування

Мінімальний набір тестів (unit/integration):

1. Unit:
2. `normalize_message_created`:
   - `indexed=false` → `[]`;
   - `kind != "text"` → `[]`;
   - `mode=public/indexed=true` → валідні `IngestChunk` з текстом;
   - `mode=confidential/indexed=true` → валідні `IngestChunk` без plaintext у метаданих.
3. `normalize_doc_upserted`:
   - довгий текст → декілька чанків з коректними `chunk_id`;
   - `indexed=false` → `[]`.

4. `normalize_file_uploaded`:

   - текст доступний → чанки;
   - текст недоступний → `[]` + лог.

5. Integration (dev):

6. опублікувати test-event `chat.message.created` у dev-стрім;
7. перевірити по логах, що воркер:
   - спожив подію,
   - зробив N чанків,
   - відправив їх у embedding + Milvus;
8. повторно відправити ту ж саму подію і переконатися, що дублікатів у Milvus немає.

## Files to create/modify (suggested)

> Актуальні шляхи можуть трохи відрізнятися — орієнтуйся по існуючому `rag-ingest-worker`.

- `services/rag-ingest-worker/events/consumer.py`
- додати routing для `chat.message.created`, `doc.upserted`, `file.uploaded`;

- для кожної події — handler з перевіркою `indexed`/`mode` та викликом нормалізатора.

- `services/rag-ingest-worker/pipeline/normalization.py`

- реалізувати/оновити:

  - `normalize_message_created(event)`
  - `normalize_doc_upserted(event)`
  - `normalize_file_uploaded(event)`

- (за потреби) `services/rag-ingest-worker/pipeline/index_neo4j.py`

- оновити побудову графових вузлів/ребер для Message/Doc/File.

- Тести для нормалізаторів (якщо є тестовий пакет).

## Acceptance criteria

1. `rag-ingest-worker` підписаний на Wave 1 події (`chat.message.created`, `doc.upserted`, `file.uploaded`) у dev-конфігурації.
2. Для кожної події є нормалізатор, який:
3. поважає `mode` та `indexed`;
4. повертає коректні `IngestChunk` з потрібними полями.
5. Чанки успішно проходять через embedding-пайплайн і індексуються в Milvus з ідемпотентною семантикою (`chunk_id`).
6. Neo4j отримує хоча б базові вузли/ребра для Message/Doc/File.
7. Повторне програвання тих самих подій не створює дублікатів у Milvus/Neo4j.
8. Логіка `indexed`/`mode` для RAG узгоджена з Meilisearch search-indexer.
9. Цей файл (`docs/cursor/rag_ingestion_events_wave1_mvp_task.md`) можна виконати через Cursor:

`bash cursor task < docs/cursor/rag_ingestion_events_wave1_mvp_task.md`

і Cursor використовує його як джерело правди для реалізації Wave 1 RAG-ingestion.