microdao-daarion/docs/cursor/rag_ingestion_events_wave1_mvp_task.md

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 по БД:
   • message.created, doc.upserted, file.uploaded.
2. Поважати mode та indexed:
   • індексувати тільки якщо indexed = true;
   • plaintext зберігати тільки для public (для confidential — embeddings/summary без відкритого тексту, згідно політики).
3. Мінімальний, але стандартний payload:
   • team_id, channel_id або project_id,
   • mode (public | confidential),
   • author_user_id / author_agent_id,
   • created_at / updated_at,
   • kind / doc_type,
   • indexed (bool),
   • 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:

   • chat.message.created
   • doc.upserted (назву узгодити з фактичним стрімом — напр. project.doc.upserted)
   • file.uploaded

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

   • "chat.message.created" → handle_message_created
   • "doc.upserted" → handle_doc_upserted
   • "file.uploaded" → handle_file_uploaded

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

   • розпарсити envelope (event, meta.team_id, payload),
   • перевірити indexed та mode,
   • викликати відповідну функцію нормалізації з pipeline/normalization.py,
   • віддати 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) для:
   • chat.message.created / chat.message.updated,
   • doc.upserted,
   • file.uploaded (якщо вже індексується).
2. По можливості, винести спільну функцію/константу для визначення indexed (based on channel/project settings), щоб RAG та Meilisearch не роз’їхались.

---

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

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

1. Unit:

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

2. Integration (dev):

   • опублікувати test-event chat.message.created у dev-стрім;
   • перевірити по логах, що воркер:
     • спожив подію,
     • зробив N чанків,
     • відправив їх у embedding + Milvus;
   • повторно відправити ту ж саму подію і переконатися, що дублікатів у 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. Для кожної події є нормалізатор, який:

   • поважає mode та indexed;
   • повертає коректні IngestChunk з потрібними полями.

3. Чанки успішно проходять через embedding-пайплайн і індексуються в Milvus з ідемпотентною семантикою (chunk_id).

4. Neo4j отримує хоча б базові вузли/ребра для Message/Doc/File.

5. Повторне програвання тих самих подій не створює дублікатів у Milvus/Neo4j.

6. Логіка indexed/mode для RAG узгоджена з Meilisearch search-indexer.

7. Цей файл (docs/cursor/rag_ingestion_events_wave1_mvp_task.md) можна виконати через Cursor:

   cursor task < docs/cursor/rag_ingestion_events_wave1_mvp_task.md
   

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