Task: Channel-agnostic document workflow (PDF + RAG)

Goal

Make the document (PDF) parsing + RAG workflow channel-agnostic, so it can be reused by:

• Telegram bots (DAARWIZZ, Helion)
• Web applications
• Mobile apps
• Any other client via HTTP API

This task defines a shared doc_service, HTTP endpoints for non-Telegram clients, and integration of Telegram handlers with this shared layer.

NOTE: If this task is re-run on a repo where it is already implemented, it should be treated as a validation/refinement task. Existing structures (services, endpoints) SHOULD NOT be removed, only improved if necessary.

---

Context

Existing components (expected state)

• Repo root: microdao-daarion/
• Gateway service: gateway-bot/

Key files:

• gateway-bot/http_api.py
• Telegram handlers for DAARWIZZ (/telegram/webhook) and Helion (/helion/telegram/webhook).
• Voice → STT flow (Whisper via STT_SERVICE_URL).
• Discord handler.

• Helper functions: get_telegram_file_path, send_telegram_message.

• gateway-bot/memory_client.py

• MemoryClient with methods:

  • get_context, save_chat_turn, create_dialog_summary, upsert_fact.

• gateway-bot/app.py

• FastAPI app, includes http_api.router as gateway_router.
• CORS configuration.

Router + parser (already implemented in router project):

• DAGI Router supports:
• mode: "doc_parse" with provider parser → OCRProvider → parser-service (DotsOCR).
• mode: "rag_query" for RAG questions.
• parser-service is available at http://parser-service:9400.

The goal of this task is to:

1. Add channel-agnostic document service into gateway-bot.
2. Add /api/doc/* HTTP endpoints for web/mobile.
3. Refactor Telegram handlers to use this service for PDF, /ingest, and RAG follow-ups.
4. Store document context in Memory Service via fact_key = "doc_context:{session_id}".

---

Changes to implement

1. Create service: gateway-bot/services/doc_service.py

Create a new directory and file:

• gateway-bot/services/__init__.py
• gateway-bot/services/doc_service.py

1.1. Pydantic models

Define models:

• QAItem — single Q&A pair
• ParsedResult — result of document parsing
• IngestResult — result of ingestion into RAG
• QAResult — result of RAG query about a document
• DocContext — stored document context

Example fields (can be extended as needed):

• QAItem: question: str, answer: str
• ParsedResult:
• success: bool
• doc_id: Optional[str]
• qa_pairs: Optional[List[QAItem]]
• markdown: Optional[str]
• chunks_meta: Optional[Dict[str, Any]] (e.g., {"count": int, "chunks": [...]})
• raw: Optional[Dict[str, Any]] (full payload from router)
• error: Optional[str]
• IngestResult:
• success: bool
• doc_id: Optional[str]
• ingested_chunks: int
• status: str
• error: Optional[str]
• QAResult:
• success: bool
• answer: Optional[str]
• doc_id: Optional[str]
• sources: Optional[List[Dict[str, Any]]]
• error: Optional[str]
• DocContext:
• doc_id: str
• dao_id: Optional[str]
• user_id: Optional[str]
• doc_url: Optional[str]
• file_name: Optional[str]
• saved_at: Optional[str]

1.2. DocumentService class

Implement DocumentService using router_client.send_to_router and memory_client:

Methods:

• async def save_doc_context(session_id, doc_id, doc_url=None, file_name=None, dao_id=None) -> bool
• Uses memory_client.upsert_fact with:
  • fact_key = f"doc_context:{session_id}"
  • fact_value_json = {"doc_id", "doc_url", "file_name", "dao_id", "saved_at"}.

• Extract user_id from session_id (e.g., telegram:123 → user_id="123").

• async def get_doc_context(session_id) -> Optional[DocContext]

• Uses memory_client.get_fact(user_id, fact_key).

• If fact_value_json exists, return DocContext(**fact_value_json).

• async def parse_document(session_id, doc_url, file_name, dao_id, user_id, output_mode="qa_pairs", metadata=None) -> ParsedResult

• Builds router request:
  • mode: "doc_parse"
  • agent: "parser"
  • metadata: includes source (derived from session_id), dao_id, user_id, session_id and optional metadata.
  • payload: includes doc_url, file_name, output_mode, dao_id, user_id.
• Calls send_to_router.

• On success:

  • Extract doc_id from response.
  • Call save_doc_context.
  • Map qa_pairs, markdown, chunks into ParsedResult.

• async def ingest_document(session_id, doc_id=None, doc_url=None, file_name=None, dao_id=None, user_id=None) -> IngestResult

• If doc_id is None, load from get_doc_context.
• Build router request with mode: "doc_parse", payload.output_mode="chunks", payload.ingest=True and doc_url / doc_id.

• Return IngestResult with ingested_chunks based on chunks length.

• async def ask_about_document(session_id, question, doc_id=None, dao_id=None, user_id=None) -> QAResult

• If doc_id is None, load from get_doc_context.
• Build router request with mode: "rag_query" and payload containing question, dao_id, user_id, doc_id.
• Return QAResult with answer and optional sources.

Provide small helper method:

• _extract_source(session_id: str) -> str → returns first segment before : (e.g. "telegram", "web").

At bottom of the file, export convenience functions:

• doc_service = DocumentService()
• Top-level async wrappers:
• parse_document(...), ingest_document(...), ask_about_document(...), save_doc_context(...), get_doc_context(...).

IMPORTANT: No Telegram-specific logic (emoji, message length, /ingest hints) in this file.

---

2. Extend MemoryClient: gateway-bot/memory_client.py

Add method:

async def get_fact(self, user_id: str, fact_key: str, team_id: Optional[str] = None) -> Optional[Dict[str, Any]]:
    """Get single fact by key"""

• Use Memory Service HTTP API, e.g.:
• GET {base_url}/facts/{fact_key} with user_id and optional team_id in query params.
• Return response.json() on 200, else None.

This method will be used by doc_service.get_doc_context.

Do not change existing public methods.

---

3. HTTP API for web/mobile: gateway-bot/http_api_doc.py

Create gateway-bot/http_api_doc.py with:

• APIRouter() named router.
• Import from services.doc_service:
• parse_document, ingest_document, ask_about_document, get_doc_context, and models.

Endpoints:

1. POST /api/doc/parse

Request (JSON body, Pydantic model ParseDocumentRequest):

• session_id: str
• doc_url: str
• file_name: str
• dao_id: str
• user_id: str
• output_mode: str = "qa_pairs"
• metadata: Optional[Dict[str, Any]]

Behaviour:

• Call parse_document(...) from doc_service.
• On failure → HTTPException(status_code=400, detail=result.error).

• On success → JSON with doc_id, qa_pairs (as list of dict), markdown, chunks_meta, raw.

• POST /api/doc/ingest

Request (IngestDocumentRequest):

• session_id: str
• doc_id: Optional[str]
• doc_url: Optional[str]
• file_name: Optional[str]
• dao_id: Optional[str]
• user_id: Optional[str]

Behaviour:

• If doc_id is missing, use get_doc_context(session_id).
• Call ingest_document(...).

• Return doc_id, ingested_chunks, status.

• POST /api/doc/ask

Request (AskDocumentRequest):

• session_id: str
• question: str
• doc_id: Optional[str]
• dao_id: Optional[str]
• user_id: Optional[str]

Behaviour:

• If doc_id is missing, use get_doc_context(session_id).
• Call ask_about_document(...).

• Return answer, doc_id, and sources (if any).

• GET /api/doc/context/{session_id}

Behaviour:

• Use get_doc_context(session_id).
• If missing → 404.
• Else return doc_id, dao_id, user_id, doc_url, file_name, saved_at.

Optional: POST /api/doc/parse/upload stub for future file-upload handling (currently can return 501 with note to use doc_url).

---

4. Wire API into app: gateway-bot/app.py

Update app.py:

• Import both routers:

python from http_api import router as gateway_router from http_api_doc import router as doc_router

• Include them:

python app.include_router(gateway_router, prefix="", tags=["gateway"]) app.include_router(doc_router, prefix="", tags=["docs"])

• Update root endpoint / to list new endpoints:

• "POST /api/doc/parse"

• "POST /api/doc/ingest"
• "POST /api/doc/ask"
• "GET /api/doc/context/{session_id}"

---

5. Refactor Telegram handlers: gateway-bot/http_api.py

Update http_api.py so Telegram uses doc_service for PDF/ingest/RAG, keeping existing chat/voice flows.

5.1. Imports and constants

• Add imports:

python from services.doc_service import ( parse_document, ingest_document, ask_about_document, get_doc_context, )

• Define Telegram length limits:

python TELEGRAM_MAX_MESSAGE_LENGTH = 4096 TELEGRAM_SAFE_LENGTH = 3500

5.2. DAARWIZZ /telegram/webhook

Inside telegram_webhook:

1. /ingest command

2. Check text from message: if starts with /ingest:

   • session_id = f"telegram:{chat_id}".
   • If message also contains a PDF document:
   • Use get_telegram_file_path(file_id) and correct bot token to build file_url.
   • await send_telegram_message(chat_id, "📥 Імпортую документ у RAG...").
   • Call ingest_document(session_id, doc_url=file_url, file_name=file_name, dao_id, user_id=f"tg:{user_id}").
   • Else:
   • Call ingest_document(session_id, dao_id=dao_id, user_id=f"tg:{user_id}") and rely on stored context.
   • Send success/failure message.

3. PDF detection

4. Check document = update.message.get("document").

5. Determine is_pdf via mime_type and/or file_name.endswith(".pdf").

6. If PDF:

   • Log file info.
   • Get file_path via get_telegram_file_path(file_id) + correct token → file_url.
   • Send "📄 Обробляю PDF-документ...".
   • session_id = f"telegram:{chat_id}".
   • Call parse_document(session_id, doc_url=file_url, file_name=file_name, dao_id, user_id=f"tg:{user_id}", output_mode="qa_pairs", metadata={"username": username, "chat_id": chat_id}).
   • On success, format:
   • Prefer Q&A (result.qa_pairs) → format_qa_response(...).
   • Else markdown → format_markdown_response(...).
   • Else chunks → format_chunks_response(...).
   • Append hint: "\n\n💡 _Використай /ingest для імпорту документа у RAG_".
   • Send response via send_telegram_message.

7. RAG follow-up questions

8. After computing text (from voice or direct text), before regular chat routing:

   • session_id = f"telegram:{chat_id}".
   • Load doc_context = await get_doc_context(session_id).
   • If doc_context.doc_id exists and text looks like a question (contains ? or Ukrainian question words):
   • Call ask_about_document(session_id, question=text, doc_id=doc_context.doc_id, dao_id=dao_id or doc_context.dao_id, user_id=f"tg:{user_id}").
   • If success, truncate answer to TELEGRAM_SAFE_LENGTH and send as Telegram message.
   • If RAG fails → fall back to normal chat routing.

9. Keep voice + normal chat flows

10. Existing STT flow and chat→router logic should remain as fallback for non-PDF / non-ingest / non-RAG messages.

5.3. Helion /helion/telegram/webhook

Mirror the same behaviours for Helion handler:

• /ingest command support.
• PDF detection and parse_document usage.
• RAG follow-up via ask_about_document.
• Use HELION_TELEGRAM_BOT_TOKEN for file download and message sending.
• Preserve existing chat→router behaviour when doc flow does not apply.

5.4. Formatting helpers

Add helper functions at the bottom of http_api.py (Telegram-specific):

• format_qa_response(qa_pairs: list, max_pairs: int = 5) -> str
• Adds header, enumerates Q&A pairs, truncates long answers, respects TELEGRAM_SAFE_LENGTH.
• format_markdown_response(markdown: str) -> str
• Wraps markdown with header; truncates to TELEGRAM_SAFE_LENGTH and appends hint about /ingest if truncated.
• format_chunks_response(chunks: list) -> str
• Shows summary about number of chunks and previews first ~3.

IMPORTANT: These helpers handle Telegram-specific constraints and SHOULD NOT be moved into doc_service.

---

Acceptance criteria

1. gateway-bot/services/doc_service.py exists and provides:
2. parse_document, ingest_document, ask_about_document, save_doc_context, get_doc_context.

3. Uses DAGI Router and Memory Service, with session_id-based context.

4. gateway-bot/http_api_doc.py exists and defines:

5. POST /api/doc/parse
6. POST /api/doc/ingest
7. POST /api/doc/ask

8. GET /api/doc/context/{session_id}

9. gateway-bot/app.py:

10. Includes both http_api.router and http_api_doc.router.

11. Root / lists new /api/doc/* endpoints.

12. gateway-bot/memory_client.py:

13. Includes get_fact(...) and existing methods still work.

14. doc_service uses upsert_fact + get_fact for doc_context:{session_id}.

15. gateway-bot/http_api.py:

16. Telegram handlers use doc_service for:
    • PDF parsing,
    • /ingest command,
    • RAG follow-up questions.

17. Continue to support existing voice→STT→chat flow and regular chat routing when doc flow isnt triggered.

18. Web/mobile clients can call /api/doc/* to:

19. Parse documents via doc_url.
20. Ingest into RAG.
21. Ask questions about the last parsed document for given session_id.

---

How to run this task with Cursor

From repo root (microdao-daarion):

cursor task < docs/cursor/channel_agnostic_doc_flow_task.md

Cursor should then:

• Create/modify the files listed above.
• Ensure implementation matches the described architecture and acceptance criteria.