activities.py

import asyncio
import re
import json
import os
import uuid
import time
import logging
import requests
from dataclasses import dataclass, field
from typing import List, Optional, Dict, Any, Set
from urllib.parse import urlparse, urljoin

from temporalio import activity
from temporalio.exceptions import ApplicationError

from bs4 import BeautifulSoup
from langchain_core.documents import Document
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_ollama import OllamaEmbeddings
from langchain_qdrant import QdrantVectorStore
from qdrant_client import QdrantClient
from qdrant_client.http.models import Distance, VectorParams, PointStruct
from langchain_groq import ChatGroq

from config import settings

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(name)s %(thread_id)s %(activity_name)s %(message)s'))
logger.addHandler(_handler)


class ContextualizationError(Exception):
    pass


class RetrievalError(Exception):
    pass


class GenerationError(Exception):
    pass


class EmbeddingError(Exception):
    pass


class IngestionError(Exception):
    pass


@dataclass
class ActivityLogContext:
    thread_id: str
    activity_name: str
    extra: Dict[str, Any] = field(default_factory=dict)


def _log_with_context(ctx: ActivityLogContext, level: int, message: str, **kwargs):
    extra = {"thread_id": ctx.thread_id, "activity_name": ctx.activity_name}
    extra.update(ctx.extra)
    extra.update(kwargs)
    adapter = logging.LoggerAdapter(logger, extra)
    adapter.log(level, message)


def _new_log_ctx(thread_id: str, activity_name: str) -> ActivityLogContext:
    return ActivityLogContext(thread_id=thread_id, activity_name=activity_name)


contextualize_q_system_prompt = (
    "Given a chat history and the latest user question "
    "which might reference context in the chat history, "
    "formulate a standalone question which can be understood "
    "without the chat history. Do NOT answer the question, "
    "just reformulate it if needed and otherwise return it as is."
)

contextualize_q_prompt = ChatPromptTemplate.from_messages([
    ("system", contextualize_q_system_prompt),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{question}"),
])

qa_system_prompt = (
    "You are an expert AI assistant. Answer the user's question using ONLY the provided context below.\n"
    "CRITICAL RULE: Do NOT include any '<think>' tags, internal reasoning, or chain-of-thought blocks in your output. "
    "Provide ONLY the direct, final answer to the user's question.\n\n"
    "If you do not know the answer or if it's not in the context, say 'I am sorry, but I do not have that information.'\n\n"
    "<context>\n{context}\n</context>"
)

qa_prompt = ChatPromptTemplate.from_messages([
    ("system", qa_system_prompt),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{question}"),
])

_reranker_cache = {}
_embeddings_cache = {}
_llm_cache = {}
_qdrant_client_cache: Optional[QdrantClient] = None
_vectorstore_cache: Optional[QdrantVectorStore] = None


def _get_embeddings():
    key = settings.models.embedding_model
    if key not in _embeddings_cache:
        _embeddings_cache[key] = OllamaEmbeddings(model=settings.models.embedding_model)
    return _embeddings_cache[key]


def _get_llm():
    key = (settings.models.llm_model, settings.models.llm_temperature)
    if key not in _llm_cache:
        _llm_cache[key] = ChatGroq(
            model_name=settings.models.llm_model,
            temperature=settings.models.llm_temperature,
            timeout=10,
            max_retries=2,
        )
    return _llm_cache[key]


def _get_qdrant_client() -> QdrantClient:
    global _qdrant_client_cache
    if _qdrant_client_cache is None:
        _qdrant_client_cache = QdrantClient(url=settings.qdrant.url)
    return _qdrant_client_cache


def _get_vectorstore() -> QdrantVectorStore:
    global _vectorstore_cache
    client = _get_qdrant_client()
    collections = [c.name for c in client.get_collections().collections]
    needs_rebuild = _vectorstore_cache is None or (settings.qdrant.collection_name not in collections)
    if needs_rebuild:
        if settings.qdrant.collection_name not in collections:
            client.create_collection(
                collection_name=settings.qdrant.collection_name,
                vectors_config=VectorParams(
                    size=settings.qdrant.vector_size,
                    distance=Distance.COSINE,
                ),
            )
        _vectorstore_cache = QdrantVectorStore(
            client=client,
            collection_name=settings.qdrant.collection_name,
            embedding=_get_embeddings(),
        )
    return _vectorstore_cache


def _get_reranker():
    if not settings.retrieval.enable_reranking:
        return None
    model_name = settings.retrieval.reranker_model
    if model_name not in _reranker_cache:
        from sentence_transformers import CrossEncoder
        _reranker_cache[model_name] = CrossEncoder(model_name)
    return _reranker_cache[model_name]


def _serialize_messages(messages: List[BaseMessage]) -> List[Dict[str, Any]]:
    result = []
    for msg in messages:
        msg_type = "human" if isinstance(msg, HumanMessage) else "ai" if isinstance(msg, AIMessage) else "system"
        result.append({"type": msg_type, "content": msg.content})
    return result


def _deserialize_messages(messages: List[Dict[str, Any]]) -> List[BaseMessage]:
    result = []
    for msg in messages:
        if msg["type"] == "human":
            result.append(HumanMessage(content=msg["content"]))
        elif msg["type"] == "ai":
            result.append(AIMessage(content=msg["content"]))
    return result


@dataclass
class ContextualizeInput:
    thread_id: str
    question: str
    chat_history: List[Dict[str, Any]]


@activity.defn(name="contextualize_activity")
async def contextualize_activity(input_data: ContextualizeInput) -> str:
    ctx = _new_log_ctx(input_data.thread_id, "contextualize_activity")
    start = time.perf_counter()
    _log_with_context(ctx, logging.INFO, "Starting contextualize activity",
                      question_length=len(input_data.question),
                      chat_history_len=len(input_data.chat_history))

    async def _body() -> str:
        debug_sleep_raw = os.environ.get("DEBUG_SLOW_CONTEXTUALIZE_SECONDS", "0")
        try:
            debug_sleep = float(debug_sleep_raw)
        except Exception:
            debug_sleep = 0.0
        if debug_sleep > 0:
            _log_with_context(ctx, logging.INFO, "DEBUG_SLOW_CONTEXTUALIZE sleeping", seconds=debug_sleep)
            await asyncio.sleep(debug_sleep)

        chat_history = _deserialize_messages(input_data.chat_history)
        if not chat_history:
            elapsed = time.perf_counter() - start
            _log_with_context(ctx, logging.INFO, "No chat history, returning raw question",
                              latency_sec=round(elapsed, 4))
            return input_data.question

        llm = _get_llm()
        chain = contextualize_q_prompt | llm
        response = await chain.ainvoke({
            "chat_history": chat_history,
            "question": input_data.question,
        })
        result = response.content

        def _looks_like_question(text: str) -> bool:
            t = text.strip()
            if not t:
                return False
            if t.endswith("?"):
                return True
            lower = t.lower()
            starts = tuple(
                w + " " for w in [
                    "what", "when", "where", "who", "whom", "whose", "why",
                    "how", "which", "is", "are", "was", "were", "do", "does",
                    "did", "has", "have", "had", "can", "could", "should",
                    "would", "will", "shall", "may", "might", "must",
                ]
            )
            return lower.startswith(starts)

        def _original_is_standalone(text: str) -> bool:
            """If the raw user question contains no pronouns / shorthand that
            clearly references prior chat, we can safely return it unchanged
            and skip any ambiguous LLM reformulation."""
            t = re.sub(r"[^\w\s]", " ", text.lower()).split()
            # Refers-to-prior signals: 3rd-person pronouns / anaphoric terms.
            ref_words = {
                "it", "its", "they", "them", "their", "theirs", "this", "that",
                "these", "those", "he", "she", "him", "her", "his", "hers",
            }
            return not ref_words.intersection(t)

        if not _looks_like_question(result):
            original = input_data.question
            if _original_is_standalone(original):
                result = original
            else:
                result = f"{original.rstrip().rstrip('?')}? (about the topic we were just discussing)".strip()

        token_usage = getattr(response, "usage_metadata", None)
        elapsed = time.perf_counter() - start
        _log_with_context(ctx, logging.INFO, "Contextualize activity completed",
                          latency_sec=round(elapsed, 4),
                          token_usage=json.dumps(token_usage) if token_usage else None)
        return result

    try:
        return await asyncio.wait_for(
            _body(), timeout=settings.timeouts.contextualize
        )
    except asyncio.TimeoutError as e:
        elapsed = time.perf_counter() - start
        _log_with_context(ctx, logging.WARNING,
                          f"Contextualize activity timed out after {settings.timeouts.contextualize}s",
                          latency_sec=round(elapsed, 4))
        raise ApplicationError(
            f"Contextualization timed out after {settings.timeouts.contextualize}s",
            type="ContextualizationError",
            non_retryable=True,
        ) from e
    except Exception as e:
        elapsed = time.perf_counter() - start
        _log_with_context(ctx, logging.WARNING, "Contextualize activity failed, raising to workflow",
                          latency_sec=round(elapsed, 4),
                          error=str(e))
        is_non_retryable = "authentication" in str(e).lower() or "invalid" in str(e).lower() or "401" in str(e)
        raise ApplicationError(
            f"Contextualization failed: {e}",
            type="ContextualizationError",
            non_retryable=is_non_retryable,
        ) from e


@dataclass
class RetrieveInput:
    thread_id: str
    question: str


@dataclass
class RetrieveResult:
    docs: List[Dict[str, Any]]
    skipped: bool = False
    skip_reason: Optional[str] = None


@activity.defn(name="retrieve_activity")
async def retrieve_activity(input_data: RetrieveInput) -> RetrieveResult:
    ctx = _new_log_ctx(input_data.thread_id, "retrieve_activity")
    start = time.perf_counter()
    _log_with_context(ctx, logging.INFO, "Starting retrieve activity",
                      question_length=len(input_data.question))

    async def _body() -> RetrieveResult:
        vectorstore = _get_vectorstore()
        client = _get_qdrant_client()
        embeddings = _get_embeddings()

        query_embedding = await embeddings.aembed_query(input_data.question)

        search_k = settings.retrieval.top_k_initial if settings.retrieval.enable_reranking else settings.retrieval.top_k_final
        query_response = client.query_points(
            collection_name=settings.qdrant.collection_name,
            query=query_embedding,
            limit=search_k,
            with_payload=True,
        )
        search_result = query_response.points

        if not search_result:
            elapsed = time.perf_counter() - start
            _log_with_context(ctx, logging.INFO, "No documents retrieved", latency_sec=round(elapsed, 4))
            return RetrieveResult(docs=[], skipped=False)

        top_score = search_result[0].score
        _log_with_context(ctx, logging.INFO, f"Initial retrieval top score: {top_score}",
                          initial_count=len(search_result))

        similarity_threshold = settings.retrieval.similarity_threshold
        try:
            override = float(os.environ.get("DEBUG_RETRIEVAL_SIMILARITY_THRESHOLD", ""))
            similarity_threshold = override
            _log_with_context(ctx, logging.INFO, "DEBUG override similarity_threshold", thr=similarity_threshold)
        except Exception:
            pass

        if top_score < similarity_threshold:
            elapsed = time.perf_counter() - start
            _log_with_context(ctx, logging.INFO,
                              f"Top score {top_score} below threshold {similarity_threshold}, skipping generation",
                              latency_sec=round(elapsed, 4))
            return RetrieveResult(
                docs=[],
                skipped=True,
                skip_reason=f"Top similarity score {top_score:.4f} below threshold {similarity_threshold}",
            )

        doc_payloads = []
        for point in search_result:
            doc_payloads.append({
                "page_content": point.payload.get("page_content", ""),
                "metadata": point.payload.get("metadata", {}),
                "score": point.score,
            })

        if settings.retrieval.enable_reranking and len(doc_payloads) > 1:
            reranker = _get_reranker()
            if reranker is not None:
                pairs = [[input_data.question, doc["page_content"]] for doc in doc_payloads]
                rerank_scores = reranker.predict(pairs)
                for doc, score in zip(doc_payloads, rerank_scores):
                    doc["rerank_score"] = float(score)
                doc_payloads.sort(key=lambda d: d.get("rerank_score", 0.0), reverse=True)
                doc_payloads = doc_payloads[:settings.retrieval.top_k_final]
                _log_with_context(ctx, logging.INFO, "Reranking completed",
                                  final_count=len(doc_payloads),
                                  top_rerank_score=doc_payloads[0].get("rerank_score"))

        docs_out = [
            {"page_content": d["page_content"], "metadata": d["metadata"]}
            for d in doc_payloads
        ]

        elapsed = time.perf_counter() - start
        _log_with_context(ctx, logging.INFO, "Retrieve activity completed",
                          latency_sec=round(elapsed, 4),
                          doc_count=len(docs_out))
        return RetrieveResult(docs=docs_out, skipped=False)

    try:
        return await asyncio.wait_for(
            _body(), timeout=settings.timeouts.retrieve
        )
    except asyncio.TimeoutError as e:
        elapsed = time.perf_counter() - start
        _log_with_context(ctx, logging.ERROR,
                          f"Retrieve activity timed out after {settings.timeouts.retrieve}s",
                          latency_sec=round(elapsed, 4))
        raise ApplicationError(
            f"Retrieval timed out after {settings.timeouts.retrieve}s",
            type="RetrievalError",
            non_retryable=True,
        ) from e
    except Exception as e:
        elapsed = time.perf_counter() - start
        _log_with_context(ctx, logging.ERROR, "Retrieve activity failed",
                          latency_sec=round(elapsed, 4),
                          error=str(e))
        is_non_retryable = "not found" in str(e).lower() or "collection" in str(e).lower()
        raise ApplicationError(
            f"Retrieval failed: {e}",
            type="RetrievalError",
            non_retryable=is_non_retryable,
        ) from e


@dataclass
class GenerateInput:
    thread_id: str
    # The question the LLM should answer, possibly rephrased (contextualized)
    # from the user's raw question to resolve pronouns against chat_history.
    question: str
    chat_history: List[Dict[str, Any]]
    docs: List[Dict[str, Any]]
    original_question: Optional[str] = None


@dataclass
class GenerateResult:
    answer: str
    human_message: Dict[str, Any]
    ai_message: Dict[str, Any]


@activity.defn(name="generate_activity")
async def generate_activity(input_data: GenerateInput) -> GenerateResult:
    ctx = _new_log_ctx(input_data.thread_id, "generate_activity")
    start = time.perf_counter()
    _log_with_context(ctx, logging.INFO, "Starting generate activity",
                      question_length=len(input_data.question),
                      doc_count=len(input_data.docs),
                      chat_history_len=len(input_data.chat_history))

    async def _body() -> GenerateResult:
        debug_sleep_raw = os.environ.get("DEBUG_SLOW_GENERATE_SECONDS", "0")
        try:
            debug_sleep = float(debug_sleep_raw)
        except Exception:
            debug_sleep = 0.0
        if debug_sleep > 0:
            _log_with_context(ctx, logging.INFO, "DEBUG_SLOW_GENERATE sleeping", seconds=debug_sleep)
            await asyncio.sleep(debug_sleep)

        llm = _get_llm()
        chat_history = _deserialize_messages(input_data.chat_history)
        context_text = "\n\n".join([doc["page_content"] for doc in input_data.docs])

        chain = qa_prompt | llm
        response = await chain.ainvoke({
            "context": context_text,
            "chat_history": chat_history,
            "question": input_data.question,
        })

        clean_answer = re.sub(r'<think>.*?</think>', '', response.content, flags=re.DOTALL).strip()
        token_usage = getattr(response, "usage_metadata", None)

        elapsed = time.perf_counter() - start
        _log_with_context(ctx, logging.INFO, "Generate activity completed",
                          latency_sec=round(elapsed, 4),
                          answer_length=len(clean_answer),
                          token_usage=json.dumps(token_usage) if token_usage else None)

        return GenerateResult(
            answer=clean_answer,
            human_message={"type": "human", "content": (input_data.original_question or input_data.question)},
            ai_message={"type": "ai", "content": clean_answer},
        )

    try:
        return await asyncio.wait_for(
            _body(), timeout=settings.timeouts.generate
        )
    except asyncio.TimeoutError as e:
        elapsed = time.perf_counter() - start
        _log_with_context(ctx, logging.ERROR,
                          f"Generate activity timed out after {settings.timeouts.generate}s",
                          latency_sec=round(elapsed, 4))
        raise ApplicationError(
            f"Generation timed out after {settings.timeouts.generate}s",
            type="GenerationError",
            non_retryable=True,
        ) from e
    except Exception as e:
        elapsed = time.perf_counter() - start
        _log_with_context(ctx, logging.ERROR, "Generate activity failed",
                          latency_sec=round(elapsed, 4),
                          error=str(e))
        is_non_retryable = "authentication" in str(e).lower() or "invalid" in str(e).lower() or "401" in str(e) or "quota" in str(e).lower()
        raise ApplicationError(
            f"Generation failed: {e}",
            type="GenerationError",
            non_retryable=is_non_retryable,
        ) from e


def _content_point_id(content: str) -> str:
    return str(uuid.uuid5(uuid.NAMESPACE_DNS, content))


def _fetch_and_extract(url: str) -> Optional[Document]:
    try:
        response = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=30)
        response.raise_for_status()
    except requests.RequestException:
        return None
    soup = BeautifulSoup(response.text, "html.parser")
    for tag in soup(["script", "style", "nav", "footer", "header"]):
        tag.decompose()
    clean_text = soup.get_text(separator="\n", strip=True)
    if len(clean_text) < 60:
        return None
    return Document(page_content=clean_text, metadata={"source": url, "title": (soup.title.string.strip() if soup.title and soup.title.string else url)})


def web_base_loader(url: str, max_pages: int = 30) -> List[Document]:
    """Load docs from ``url``. Strategy (in order, falls back through each):

    1. Try ``{url.rstrip('/')}/sitemap.xml``. Only trust the response if it is
       actually XML (starts with ``<?xml`` / ``<urlset`` / ``<sitemapindex`` —
       Mintlify + similar SPAs return 200 HTML catch-alls for every URL, so
       HTTP 200 alone is a false positive). If valid XML sitemap found, parse
       up to ``max_pages`` same-host URLs from it.
    2. Otherwise, fetch the landing page itself and extract *all* unique
       same-host ``<a href>`` URLs (plus allow a slight netloc relaxation —
       e.g. ``docs.smith.langchain.com`` redirects to ``docs.langchain.com``
       and the navigation links on the landing page point to the latter;
       since the *landing page itself* surfaced those URLs, we accept them as
       the canonical doc pages even though the netloc differs from the
       user-supplied input). Prefer URLs whose path shares a common prefix
       with the landing page so we stay scoped to one doc area.
    3. If neither produced any pages, fall back to the single-page scrape we
       used to do.

    This fixes the prior "ingestion returns 3 chunks even though 30 doc pages
    exist" bug seen for ``https://docs.smith.langchain.com/``.
    """
    if max_pages <= 1:
        doc = _fetch_and_extract(url)
        if doc is None:
            raise IngestionError(f"Failed to fetch or extract text from URL {url}")
        return [doc]

    try:
        root_resp = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=30)
        root_resp.raise_for_status()
    except requests.RequestException as e:
        raise IngestionError(f"Failed to fetch URL {url}: {e}") from e
    final_root_url = root_resp.url.rstrip("/") or url.rstrip("/")
    parsed_root = urlparse(final_root_url)

    candidates: List[str] = []

    sitemap_url = f"{final_root_url}/sitemap.xml"
    try:
        resp = requests.get(
            sitemap_url,
            headers={"User-Agent": "Mozilla/5.0", "Accept": "application/xml,text/xml;q=0.9,*/*;q=0.1"},
            timeout=15,
            allow_redirects=True,
        )
        if resp.status_code == 200:
            head = (resp.content[:512].lstrip() or b"").decode("utf-8", errors="ignore").lower()
            if head.startswith("<?xml") or head.startswith("<urlset") or head.startswith("<sitemapindex"):
                import xml.etree.ElementTree as ET
                root = ET.fromstring(resp.content)
                ns = ""
                if root.tag.startswith("{"):
                    ns = root.tag.split("}")[0].lstrip("{")
                loc_tag = f"{{{ns}}}loc" if ns else "loc"
                for loc in root.iter(loc_tag):
                    if loc.text:
                        candidates.append(loc.text.strip())
    except Exception:
        pass

    def _url_path_prefix_ok(candidate: str) -> bool:
        """Prefer candidate URLs whose path shares a prefix with the landing
        page (e.g. landing page is on `/langsmith/observability`, keep
        `/langsmith/**` pages but drop `/use-these-docs` cross-site meta
        links).  If we can't find any, we'll fall back to all links."""
        try:
            p = urlparse(candidate)
            root_path = parsed_root.path.rstrip("/")
            if not root_path or root_path == "/":
                return True
            # Common prefix of at least the first two path segments.
            root_parts = [seg for seg in root_path.split("/") if seg]
            cand_parts = [seg for seg in p.path.split("/") if seg]
            if not root_parts or not cand_parts:
                return True
            n_match = 0
            for a, b in zip(root_parts, cand_parts):
                if a == b:
                    n_match += 1
                else:
                    break
            # At least first part matches, or for short root (1 segment),
            # accept any same-topic page.
            return n_match >= min(1, len(root_parts))
        except Exception:
            return False

    # Extract <a href> links from the landing page as secondary source (and
    # primary source for SPAs whose sitemap URL returns HTML).
    a_candidates: List[str] = []
    try:
        soup = BeautifulSoup(root_resp.text, "html.parser")
        for a in soup.find_all("a", href=True):
            href = a["href"].strip()
            if not href or href.startswith(("#", "mailto:", "tel:", "javascript:")):
                continue
            abs_url = urljoin(final_root_url, href)
            p = urlparse(abs_url)
            if p.scheme not in ("http", "https"):
                continue
            # Accept both:
            #   - links to the *canonical* host (post-redirect host)
            #   - links to the *original* user-input host
            user_parsed = urlparse(url.rstrip("/"))
            netloc_ok = (
                p.netloc.lower() == parsed_root.netloc.lower()
                or p.netloc.lower() == user_parsed.netloc.lower()
            )
            if not netloc_ok:
                continue
            a_candidates.append(abs_url.split("#", 1)[0].rstrip("/"))
    except Exception:
        pass

    # Build ordered list: (a) sitemap URLs first (if any valid + matching),
    # then (b) <a href> from landing page (prefixed), then (c) <a href>
    # relaxed (no path-prefix) to fill up to max_pages.
    def _norm(u: str) -> str:
        return u.split("#", 1)[0].rstrip("/")

    seen: Set[str] = set()
    ordered: List[str] = []
    ordered.append(_norm(final_root_url))
    seen.add(_norm(final_root_url))

    # --- (a) sitemap URLs (strict same-host + prefix OK) ---
    for u in candidates:
        try:
            p = urlparse(u)
            if p.netloc.lower() != parsed_root.netloc.lower():
                continue
        except Exception:
            continue
        n = _norm(u)
        if n in seen:
            continue
        if not _url_path_prefix_ok(u):
            continue
        seen.add(n)
        ordered.append(n)
        if len(ordered) >= max_pages:
            break

    # --- (b) landing-page <a href> with prefix match ---
    if len(ordered) < max_pages:
        for u in a_candidates:
            n = _norm(u)
            if n in seen:
                continue
            if not _url_path_prefix_ok(u):
                continue
            seen.add(n)
            ordered.append(n)
            if len(ordered) >= max_pages:
                break

    # --- (c) landing-page <a href> relaxed (no prefix check) ---
    if len(ordered) < max_pages:
        for u in a_candidates:
            n = _norm(u)
            if n in seen:
                continue
            seen.add(n)
            ordered.append(n)
            if len(ordered) >= max_pages:
                break

    ordered = ordered[:max_pages]

    docs: List[Document] = []
    for u in ordered:
        doc = _fetch_and_extract(u)
        if doc is not None:
            docs.append(doc)

    if not docs:
        raise IngestionError(f"Failed to extract any text from {url} or its linked pages (tried {len(ordered)} urls)")
    return docs


@dataclass
class IngestInput:
    url: str
    thread_id: str = "ingestion"


@activity.defn(name="ingest_documents_activity")
async def ingest_documents_activity(input_data: IngestInput) -> Dict[str, Any]:
    ctx = _new_log_ctx(input_data.thread_id, "ingest_documents_activity")
    start = time.perf_counter()
    _log_with_context(ctx, logging.INFO, "Starting document ingestion", url=input_data.url)

    async def _body() -> Dict[str, Any]:
        docs = web_base_loader(input_data.url)
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=settings.splitter.chunk_size,
            chunk_overlap=settings.splitter.chunk_overlap,
        )
        split_docs = splitter.split_documents(docs)
        _log_with_context(ctx, logging.INFO, f"Split into {len(split_docs)} chunks")

        embeddings = _get_embeddings()
        client = _get_qdrant_client()

        texts = [doc.page_content for doc in split_docs]
        metadatas = [doc.metadata for doc in split_docs]
        embedded_vectors = await embeddings.aembed_documents(texts)

        existing_ids = set()
        batch_size = 100
        for i in range(0, len(texts), batch_size):
            batch_texts = texts[i:i + batch_size]
            for text in batch_texts:
                pid = _content_point_id(text)
                existing_ids.add(pid)

        collection_info = client.count(settings.qdrant.collection_name)
        _log_with_context(ctx, logging.INFO, f"Collection has {collection_info.count} points before ingestion")

        scroll_result = client.scroll(
            collection_name=settings.qdrant.collection_name,
            limit=10000,
            with_payload=False,
            with_vectors=False,
        )
        existing_point_ids = {p.id for p in scroll_result[0]}

        points_to_upsert = []
        duplicate_count = 0
        for text, metadata, vector in zip(texts, metadatas, embedded_vectors):
            pid = _content_point_id(text)
            if pid in existing_point_ids:
                duplicate_count += 1
                continue
            points_to_upsert.append(PointStruct(
                id=pid,
                vector=vector,
                payload={
                    "page_content": text,
                    "metadata": metadata,
                },
            ))

        if points_to_upsert:
            batch_size_upsert = 100
            for i in range(0, len(points_to_upsert), batch_size_upsert):
                batch = points_to_upsert[i:i + batch_size_upsert]
                client.upsert(
                    collection_name=settings.qdrant.collection_name,
                    points=batch,
                )

        new_count = client.count(settings.qdrant.collection_name).count
        elapsed = time.perf_counter() - start
        _log_with_context(ctx, logging.INFO, "Ingestion completed",
                          latency_sec=round(elapsed, 4),
                          total_chunks=len(split_docs),
                          duplicates_skipped=duplicate_count,
                          new_points_upserted=len(points_to_upsert),
                          total_collection_points=new_count)

        return {
            "total_chunks": len(split_docs),
            "duplicates_skipped": duplicate_count,
            "new_points_upserted": len(points_to_upsert),
            "total_collection_points": new_count,
        }

    # Ingestion timeout: 5 minutes by default (crawls + embedding can be slow)
    INGEST_TIMEOUT_SECONDS = 300
    try:
        return await asyncio.wait_for(_body(), timeout=INGEST_TIMEOUT_SECONDS)
    except asyncio.TimeoutError as e:
        elapsed = time.perf_counter() - start
        _log_with_context(ctx, logging.ERROR,
                          f"Ingestion timed out after {INGEST_TIMEOUT_SECONDS}s",
                          latency_sec=round(elapsed, 4))
        raise ApplicationError(
            f"Ingestion timed out after {INGEST_TIMEOUT_SECONDS}s",
            type="IngestionError",
            non_retryable=True,
        ) from e
    except Exception as e:
        elapsed = time.perf_counter() - start
        _log_with_context(ctx, logging.ERROR, "Ingestion failed",
                          latency_sec=round(elapsed, 4),
                          error=str(e))
        raise ApplicationError(
            f"Ingestion failed: {e}",
            type="IngestionError",
            non_retryable=False,
        ) from e