app/services/pipeline/core.py

"""pipeline/core.py — The 12-step RAG pipeline orchestrator.

This is the single entry point for ALL chatbot response generation.
Every chat message flows through all 12 steps in sequence.

RULE: No step may be skipped.
RULE: Every step failure must be handled gracefully — never crash the user's session.
RULE: Token usage is tracked and returned for budget accounting.

Pipeline Steps:
  1.  Boundary check (query)
  2.  Intent classification
  3.  Book resolution (select or show selector)
  4.  Query rewriting
  5.  Vector retrieval (ChromaDB)
  6.  Cross-encoder re-ranking
  6.5 Chunk deduplication
  7.  Context assembly (token-aware)
  8-10. LLM generation + faithfulness + safety  ← see pipeline/generation.py
  11. Upsell strategy injection
  12. Response formatting + link injection
"""

import time
from dataclasses import replace

import structlog
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import get_settings
from app.models.user import User
from app.repositories.book_repo import BookRepository
from app.services.context_builder import build_context, diversify_chunks
from app.services.formatter import ResponseFormatter
from app.services.guardrails import check_boundary, sanitize_user_input
from app.services.intent import classify_intent
from app.services.objection import count_prior_objections, detect_objection
from app.services.prompter import JAILBREAK_RESPONSE, OFF_TOPIC_RESPONSE
from app.services.reranker import rerank_chunks
from app.services.rewriter import rewrite_query
from app.services.session_core.context import effective_interest_score
from app.services.session_core.manager import SessionContext
from app.services.upsell_engine import UpsellEngine
from app.services.vector_store import retrieve_chunks

from app.services.pipeline.cache import CachedAnswer, cache_get, cache_key, cache_set
from app.services.pipeline.dedup import deduplicate_chunks
from app.services.pipeline.generation import generate_response
from app.services.pipeline.guards import (
    is_book_selection_turn,
    is_catalog_question,
    is_full_story_request,
    is_greeting,
    should_short_circuit_to_catalog,
)
from app.services.pipeline.handlers import (
    PipelineResult,
    book_selected_response,
    book_selector_response,
    books_list_response,
    boundary_response,
    catalog_response,
    full_story_response,
    greeting_response,
    no_books_response,
    no_context_response,
    piracy_response,
)
from app.services.pipeline.helpers import (
    check_custom_qa,
    find_book,
    get_book_links,
    resolve_book,
    selected_book_title,
)
from app.services.followup_service import generate_follow_ups

logger = structlog.get_logger(__name__)
cfg = get_settings()

_upsell_engine = UpsellEngine()
_formatter = ResponseFormatter()


def _effective_intent(query: str, intent: str) -> str:
    """Resolve intent used for upsell strategy and link gating."""
    if is_full_story_request(query):
        return "full_story_request"
    return intent


def _upsell_session_context(
    session_context: SessionContext,
    effective_intent: str,
) -> SessionContext:
    """Session context with interest score boosted for the current turn's intent."""
    score = effective_interest_score(
        session_context.turn_count,
        session_context.interest_tags,
        current_intent=effective_intent,
    )
    return replace(session_context, interest_score=score)


async def _apply_upsell_and_format(
    raw_response: str,
    effective_intent: str,
    session_context: SessionContext,
    author_id: str,
    search_book_id: str | None,
    top_book_ids: list[str],
    db: AsyncSession,
    *,
    strategy: str | None = None,
    max_chars: int | None = None,
) -> tuple[dict, str, bool]:
    """Run Steps 11–12: strategy, optional hook fallback, format with links."""
    upsell_ctx = _upsell_session_context(session_context, effective_intent)
    if strategy is None:
        strategy = _upsell_engine.select_strategy(effective_intent, upsell_ctx)

    show_link = _upsell_engine.should_include_link(effective_intent, upsell_ctx, strategy)
    top_book_id = search_book_id or (top_book_ids[0] if top_book_ids else None)
    purchase_url, preview_url = await get_book_links(top_book_id, author_id, db)

    upsell_hook = _upsell_engine.fallback_hook_if_needed(
        raw_response,
        strategy,
        upsell_ctx.interest_score,
    )

    formatted = _formatter.format(
        response_text=raw_response,
        upsell_hook=upsell_hook,
        purchase_url=purchase_url,
        preview_url=preview_url,
        show_link=show_link and bool(purchase_url),
        max_chars=max_chars,
    )
    _log_upsell_turn(author_id, strategy, formatted["has_links"], upsell_ctx.interest_score)
    return formatted, strategy, formatted["has_links"]


def _log_upsell_turn(author_id: str, strategy: str, link_shown: bool, interest_score: float) -> None:
    """Structured log line for upsell observability (R: strategy tuning via logs).

    Every turn's strategy + link-shown outcome is logged so authors/ops can
    aggregate strategy performance externally (e.g. via log search or the
    `/analytics/upsell` endpoint, which reads the same signal from ChatMessage
    rows) without building a full A/B testing system.
    """
    logger.info(
        "upsell_turn",
        author_id=author_id,
        upsell_strategy=strategy,
        link_shown=link_shown,
        interest_score=interest_score,
    )


async def run_pipeline(
    query: str,
    author: User,
    session_context: SessionContext,
    db: AsyncSession,
) -> PipelineResult:
    """Execute the full 12-step RAG pipeline for one chat turn.

    Args:
        query: The user's raw message text.
        author: The author whose catalog is being queried.
        session_context: Current session state (history, selected book, interest).
        db: Active database session.

    Returns:
        PipelineResult with formatted response and all metadata for logging.
    """
    start_ms = time.monotonic()
    log = logger.bind(author_id=author.id, turn=session_context.turn_count)

    # ── Step 0: Sanitize input ────────────────────────────────────────────────
    query = sanitize_user_input(query)
    if not query:
        return boundary_response(
            "I didn't catch that — try asking about one of the books!",
            start_ms, "empty_input",
        )

    # B2 fix: previously sanitize_user_input() silently truncated at 2000 chars,
    # causing nonsense LLM responses. Now we reject over-long input with clear feedback.
    if len(query) > cfg.RAG_MAX_INPUT_CHARS:
        return boundary_response(
            "That message is a bit long for me! Try asking one specific question at a time.",
            start_ms, "input_too_long",
        )

    # ── Step 1: Boundary Check ────────────────────────────────────────────────
    violation_type, _ = check_boundary(query)
    if violation_type == "jailbreak_attempt":
        return boundary_response(
            JAILBREAK_RESPONSE.format(
                bot_name=author.bot_name,
                author_name=author.full_name or "the author",
            ),
            start_ms, "jailbreak_attempt",
        )
    if violation_type == "piracy_request":
        return await piracy_response(author, session_context, db, start_ms)
    if violation_type == "off_topic":
        return boundary_response(
            OFF_TOPIC_RESPONSE.format(author_name=author.full_name or "the author"),
            start_ms, "off_topic",
        )
    if violation_type == "competitor_mention":
        return boundary_response(
            "I focus on this author's books only — happy to help you explore their work.",
            start_ms, "competitor_mention",
        )

    # ── Step 1.5: Custom Q&A short-circuit ───────────────────────────────────
    qa_match = await check_custom_qa(query, author.id, db)
    if qa_match:
        return PipelineResult(
            response={"text": qa_match["answer"], "links": [], "has_links": False},
            intent="custom_qa",
            intent_confidence=qa_match["score"],
            response_ms=int((time.monotonic() - start_ms) * 1000),
        )

    # ── Step 2: Intent Classification ─────────────────────────────────────────
    intent_result = await classify_intent(query, session_context.history)
    log.debug("Intent classified", intent=intent_result.intent, source=intent_result.source)

    if intent_result.intent == "jailbreak_attempt":
        return boundary_response(
            JAILBREAK_RESPONSE.format(
                bot_name=author.bot_name,
                author_name=author.full_name or "the author",
            ),
            start_ms, "jailbreak_attempt",
        )

    # ── Step 3: Book Resolution ───────────────────────────────────────────────
    active_books = await BookRepository(db).list_active_for_author(author.id)
    if not active_books:
        return no_books_response(start_ms)

    if intent_result.intent == "greeting" or is_greeting(query):
        return greeting_response(author, active_books, session_context, start_ms)

    if should_short_circuit_to_catalog(
        query, intent_result.intent, session_context.selected_book_id,
    ):
        return await catalog_response(author, active_books, session_context, db, start_ms)

    if is_book_selection_turn(
        query, session_context.selected_book_id, active_books, session_context.turn_count,
    ):
        book = find_book(active_books, session_context.selected_book_id)
        if book:
            return await book_selected_response(book, author.id, db, start_ms)

    if intent_result.intent == "full_story_request" or is_full_story_request(query):
        book = find_book(active_books, session_context.selected_book_id) or active_books[0]
        return await full_story_response(
            book, author.id, db, start_ms, session_context=session_context,
        )

    if (
        len(active_books) > 1
        and not session_context.selected_book_id
        and not session_context.is_cross_book
        and is_catalog_question(query)
    ):
        return books_list_response(
            "Select a book below to ask about it.", active_books, start_ms, intent="comparison",
        )

    target_book_id = await resolve_book(intent_result, session_context, active_books, author.id)
    cross_book_search = (
        session_context.is_cross_book
        or (
            intent_result.intent in ("comparison", "book_comparison")
            and not session_context.selected_book_id
            and not is_catalog_question(query)
        )
    )
    if (
        target_book_id is None
        and len(active_books) > 1
        and intent_result.book_confidence < cfg.RAG_BOOK_CONFIDENCE_THRESHOLD
        and session_context.selected_book_id is None
        and not cross_book_search
    ):
        return book_selector_response(active_books, start_ms)

    if cross_book_search:
        search_book_id = None
    else:
        search_book_id = target_book_id or session_context.selected_book_id
    effective_int = _effective_intent(query, intent_result.intent)

    # ── Price intelligence short-circuit (after book resolve; skip RAG/LLM) ──
    from app.services.price_catalog_service import PriceCatalogService, classify_price_query

    price_q = classify_price_query(query)
    price_kind = price_q.kind
    use_price_catalog = search_book_id and (
        intent_result.intent == "price_inquiry"
        or (
            intent_result.intent == "purchase_intent"
            and price_kind == "how_much"
        )
    )
    if use_price_catalog and price_kind != "soft_value":
        book_title = selected_book_title(active_books, search_book_id) or "this book"
        price_result = await PriceCatalogService(db).handle(
            query, author.id, search_book_id, book_title,
        )
        return PipelineResult(
            response=price_result,
            intent="price_inquiry" if intent_result.intent == "price_inquiry" else intent_result.intent,
            intent_confidence=intent_result.confidence,
            response_ms=int((time.monotonic() - start_ms) * 1000),
            link_shown=bool(price_result.get("has_links") or price_result.get("platform_offers")),
        )

    # ── Objection Detection (persuasion modifier, never blocks) ──────────────
    objection_type = detect_objection(query)
    prior_objections = (
        count_prior_objections(session_context.history) if objection_type else 0
    )
    if objection_type:
        log.info("Purchase objection detected", type=objection_type, prior=prior_objections)

    # ── Cache Lookup (objection replies are personalized — never cached) ─────
    if objection_type is None and intent_result.intent not in ("purchase_intent", "complaint", "greeting"):
        ck = cache_key(author.id, search_book_id, query)
        cached = cache_get(ck)
        if cached is not None:
            log.debug("Cache hit — re-applying session upsell", key=ck[:8])
            formatted, strategy, link_shown = await _apply_upsell_and_format(
                cached.raw_response,
                effective_int,
                session_context,
                author.id,
                search_book_id,
                cached.top_book_ids,
                db,
            )
            formatted["follow_ups"] = generate_follow_ups(
                intent=effective_int,
                interest_tags=session_context.interest_tags,
                has_multiple_books=len(active_books) > 1,
                book_title=selected_book_title(active_books, session_context.selected_book_id),
            )
            return PipelineResult(
                response=formatted,
                intent=cached.intent,
                intent_confidence=cached.intent_confidence,
                faithfulness_score=cached.faithfulness_score,
                hallucination_detected=cached.hallucination_detected,
                boundary_triggered=False,
                upsell_strategy=strategy,
                link_shown=link_shown,
                prompt_tokens=cached.prompt_tokens,
                completion_tokens=cached.completion_tokens,
                response_ms=int((time.monotonic() - start_ms) * 1000),
                top_book_ids=cached.top_book_ids,
            )

    # ── Step 4: Query Rewriting ───────────────────────────────────────────────
    query_variations = await rewrite_query(query, session_context.history)

    # ── Step 5: Vector Retrieval ──────────────────────────────────────────────
    raw_chunks = await retrieve_chunks(
        queries=query_variations,
        author_id=author.id,
        book_id=search_book_id,
        top_k=cfg.RAG_RETRIEVAL_TOP_K,
    )
    if not raw_chunks:
        log.warning("No chunks retrieved")
        return await no_context_response(query, author, active_books, session_context, db, start_ms)

    # ── Step 6: Re-ranking ────────────────────────────────────────────────────
    top_chunks = await rerank_chunks(
        query=query, chunks=raw_chunks,
        top_n=cfg.RAG_RERANK_TOP_N, min_score=cfg.RAG_RERANK_MIN_SCORE,
    )
    if not top_chunks:
        top_chunks = raw_chunks[: cfg.RAG_RERANK_TOP_N]
    if not top_chunks:
        return await no_context_response(query, author, active_books, session_context, db, start_ms)

    # ── Step 6.5: Chunk Deduplication + Diversity ────────────────────────────
    top_chunks = deduplicate_chunks(top_chunks)
    top_chunks = diversify_chunks(top_chunks)

    # ── Step 7: Context Assembly ──────────────────────────────────────────────
    context_str, _ = build_context(top_chunks)

    # ── Step 11 (pre-generation): Upsell Strategy ─────────────────────────────
    upsell_ctx = _upsell_session_context(session_context, effective_int)
    if objection_type:
        # An objection is active consideration — treat it as a warm signal.
        upsell_ctx = replace(
            upsell_ctx,
            interest_score=round(min(upsell_ctx.interest_score + 0.15, 1.0), 3),
        )
        strategy = "PAIN_SOLUTION" if objection_type in ("price", "time") else "STORY_BRIDGE"
    else:
        strategy = _upsell_engine.select_strategy(effective_int, upsell_ctx)

    # ── Steps 8–10: LLM Generation + Faithfulness + Safety ───────────────────
    price_facts = ""
    if objection_type == "price" and search_book_id:
        try:
            from app.services.price_catalog_service import (
                PriceCatalogService,
                format_price_facts,
            )
            _offers = await PriceCatalogService(db).catalog_for_book(
                author.id, search_book_id,
            )
            price_facts = format_price_facts(_offers)
        except Exception:
            log.warning("price_facts_load_failed", exc_info=True)
            price_facts = ""
    raw_response, faithfulness_score, hallucination_detected, prompt_tokens, completion_tokens, max_response_chars = (
        await generate_response(
            query=query,
            author=author,
            active_books=active_books,
            session_context=session_context,
            top_chunks=top_chunks,
            context_str=context_str,
            strategy=strategy,
            effective_interest_score=upsell_ctx.interest_score,
            objection_type=objection_type,
            prior_objections=prior_objections,
            intent=effective_int,
            price_facts=price_facts,
        )
    )

    top_book_ids = list({c.book_id for c in top_chunks})

    # ── Step 11.5: Follow-up Question Suggestions ────────────────────────────
    follow_ups = generate_follow_ups(
        intent=effective_int,
        interest_tags=session_context.interest_tags,
        has_multiple_books=len(active_books) > 1,
        book_title=selected_book_title(active_books, session_context.selected_book_id),
    )

    # ── Step 12: Format Response ──────────────────────────────────────────────
    formatted, strategy, link_shown = await _apply_upsell_and_format(
        raw_response,
        effective_int,
        session_context,
        author.id,
        search_book_id,
        top_book_ids,
        db,
        strategy=strategy,
        max_chars=max_response_chars,
    )
    formatted["follow_ups"] = follow_ups

    elapsed_ms = int((time.monotonic() - start_ms) * 1000)
    log.info("Pipeline complete", ms=elapsed_ms, faithfulness=faithfulness_score, strategy=strategy)

    result = PipelineResult(
        response=formatted,
        intent=intent_result.intent,
        intent_confidence=intent_result.confidence,
        faithfulness_score=faithfulness_score,
        hallucination_detected=hallucination_detected,
        boundary_triggered=False,
        upsell_strategy=strategy,
        link_shown=link_shown,
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        response_ms=elapsed_ms,
        top_book_ids=top_book_ids,
        follow_ups=follow_ups,
    )

    if objection_type is None and intent_result.intent not in ("purchase_intent", "complaint", "greeting"):
        ck = cache_key(author.id, search_book_id, query)
        cache_set(
            ck,
            CachedAnswer(
                raw_response=raw_response,
                faithfulness_score=faithfulness_score,
                hallucination_detected=hallucination_detected,
                prompt_tokens=prompt_tokens,
                completion_tokens=completion_tokens,
                top_book_ids=top_book_ids,
                intent=intent_result.intent,
                intent_confidence=intent_result.confidence,
            ),
        )
        log.debug("Answer cached", key=ck[:8])

    return result