from __future__ import annotations

import pytest

from app.llm.base import ChatResult, ProviderError, ToolCall
from app.models import ChatMessage, ChatSession
from app.orchestrator import (
    _PHOTO_PRODUCT_INTENT,
    _needs_grounding,
    ATTACHMENT_MARKER,
    PHOTO_GROUNDING_NOTE,
    run_turn,
)
from app.tools.registry import ToolContext


class FakeRouter:
    def __init__(self, results: list[ChatResult]):
        self._results = list(results)
        self.calls = 0

    async def chat(self, messages, tools, tier="large", temperature=None, order=None, tool_choice=None, max_tokens=None):
        self.calls += 1
        return self._results.pop(0)


class FakeShopify:
    async def execute(self, query, variables=None):
        return {"orders": {"edges": []}}


@pytest.fixture(autouse=True)
def no_real_search(monkeypatch):
    async def fake_search(session, query, k=4, *, tenant_id=None):
        return []

    from app.rag import index

    monkeypatch.setattr(index, "search", fake_search)


async def _session(db):
    s = ChatSession(shop="x")
    db.add(s)
    await db.flush()
    return s


async def test_run_turn_persists_detected_language_on_session(db_session):
    """Regression: ctx.session.lang was computed for the reply but never SAVED,
    so ChatResponse.lang (what the widget reads to match the handoff contact
    form's language to the conversation) was always None -- a visitor chatting
    in English could see a Spanish 'Tu nombre / Tu email' form."""
    router = FakeRouter([
        ChatResult(content="Sure, we ship for free.", tool_calls=[], finish_reason="stop"),
        ChatResult(content="Sure, we ship for free.", tool_calls=[], finish_reason="stop"),
    ])
    session = await _session(db_session)
    assert session.lang is None
    ctx = ToolContext(db=db_session, session=session)

    resp = await run_turn(router, ctx, "Do you ship for free?", brand_name="Tienda")

    assert resp.lang == "en"
    assert session.lang == "en"  # persisted, not just returned this once


async def test_info_turn_uses_knowledge_tool_then_answers(db_session):
    router = FakeRouter(
        [
            ChatResult(
                content=None,
                tool_calls=[ToolCall(id="t1", name="search_knowledge", arguments={"query": "envíos"})],
                finish_reason="tool_calls",
            ),
            ChatResult(content="Hacemos envíos gratis desde 199€.", tool_calls=[], finish_reason="stop"),
        ]
    )
    ctx = ToolContext(db=db_session, session=await _session(db_session))
    resp = await run_turn(router, ctx, "¿hacéis envíos gratis?", brand_name="Tienda")

    assert resp.reply == "Hacemos envíos gratis desde 199€."
    assert resp.used_tools == ["search_knowledge"]
    assert router.calls == 2

    # persisted user + assistant
    from sqlalchemy import select

    msgs = (await db_session.execute(select(ChatMessage))).scalars().all()
    roles = [m.role for m in msgs]
    assert "user" in roles and "assistant" in roles


async def test_order_turn_unverified_asks_for_credentials(db_session):
    router = FakeRouter(
        [
            ChatResult(
                content=None,
                tool_calls=[ToolCall(id="o1", name="lookup_order", arguments={})],
                finish_reason="tool_calls",
            ),
            ChatResult(
                content="Para localizar tu pedido necesito tu email y número de pedido.",
                tool_calls=[],
                finish_reason="stop",
            ),
        ]
    )
    ctx = ToolContext(db=db_session, session=await _session(db_session), shopify=FakeShopify())
    resp = await run_turn(router, ctx, "¿dónde está mi pedido?")
    assert "email" in resp.reply.lower()
    assert resp.used_tools == ["lookup_order"]


class _ToolFailRouter:
    """Fails (ProviderError) whenever tools are sent; answers when no tools."""

    def __init__(self, reply="Respuesta sin herramientas."):
        self.reply = reply
        self.calls_with_tools = 0

    async def chat(self, messages, tools, tier="large", temperature=None, order=None, tool_choice=None, max_tokens=None):
        if tools:
            self.calls_with_tools += 1
            raise ProviderError("tool_use_failed")
        return ChatResult(content=self.reply, tool_calls=[], finish_reason="stop")


class _AlwaysFailRouter:
    async def chat(self, messages, tools, tier="large", temperature=None, order=None, tool_choice=None, max_tokens=None):
        raise ProviderError("down")


async def test_degrades_to_no_tools_when_tool_calling_fails(db_session):
    router = _ToolFailRouter()
    ctx = ToolContext(db=db_session, session=await _session(db_session))
    resp = await run_turn(router, ctx, "¿qué vendéis?")
    assert resp.reply == "Respuesta sin herramientas."
    assert router.calls_with_tools >= 1  # it tried tools first, then degraded


async def test_total_provider_failure_returns_fallback_not_500(db_session):
    from app.orchestrator import FALLBACK_REPLY

    ctx = ToolContext(db=db_session, session=await _session(db_session))
    resp = await run_turn(_AlwaysFailRouter(), ctx, "hola")
    assert resp.reply == FALLBACK_REPLY


# --- deterministic grounding gates (regex level) ----------------------------


@pytest.mark.parametrize(
    "msg",
    [
        # ES similarity / availability
        "¿tenéis algo como esto?",
        "teneis algo como esto",  # unaccented typing
        "¿tenéis algo parecido a esto?",
        "busco algo parecido",
        "quiero uno similar",
        "¿tenéis esto?",
        "¿tenéis algo así?",
        "¿lo tenéis en tienda?",
        "¿la vendéis?",
        # EN similarity / availability
        "do you have this?",
        "do you have something like this?",
        "do you have anything similar?",
        "I want something like this",
        "got one similar to this?",
        "do you sell anything like this one?",
    ],
)
def test_ground_intent_fires_on_similarity_phrases(msg):
    assert _needs_grounding(msg)


@pytest.mark.parametrize(
    "msg",
    [
        # OPEN grounding: ANY real question grounds, whatever its wording — not a
        # fixed intent list. (Regression: these used to free-wheel from memory.)
        "si pido una srx-100 cuanto tarda en llegarme",
        "necesito saber las medidas de la asx-90",
        "es compatible con discos de 28mm?",
        "mis discos de 28 valen en la asx-2000?",
        "cuanto tarda el envio a canarias",
        "que material es",
        "how long does shipping take?",
    ],
)
def test_grounding_fires_on_any_real_question(msg):
    assert _needs_grounding(msg)


@pytest.mark.parametrize(
    "msg",
    [
        "hola, buenos días",
        "gracias por todo",
        "ok perfecto",
        "¿qué tal?",
        "¿qué pone en el documento?",
        "resume el pdf adjunto",
        "what does the document say?",
    ],
)
def test_grounding_silent_on_smalltalk_and_document_questions(msg):
    assert not _needs_grounding(msg)


@pytest.mark.parametrize(
    "msg",
    [
        "¿tenéis algo como esto?",
        "¿tenéis algo parecido?",
        "¿vendéis esto?",
        "¿cuánto cuesta esto?",
        "quiero comprarlo",
        "do you have this in stock?",
        "is something like this available?",
        "how much is it? price please",
    ],
)
def test_photo_product_intent_fires_on_product_seeking(msg):
    assert _PHOTO_PRODUCT_INTENT.search(msg)


@pytest.mark.parametrize(
    "msg",
    [
        "¿qué pone en el documento?",
        "¿qué pone en el pdf?",
        "resume el adjunto",
        "what does the pdf say?",
    ],
)
def test_photo_product_intent_silent_on_document_questions(msg):
    assert not _PHOTO_PRODUCT_INTENT.search(msg)


async def test_photo_plus_product_intent_injects_photo_grounding_note(db_session):
    """Attachment marker + product-seeking message -> the forced search_products
    note is REALLY in the prompt the router receives."""

    class Capture:
        def __init__(self):
            self.prompts = []

        async def chat(self, messages, tools, tier="large", temperature=None, order=None, tool_choice=None, max_tokens=None):
            self.prompts.append([dict(m) for m in messages])
            return ChatResult(content="Claro, busco.", tool_calls=[], finish_reason="stop")

    router = Capture()
    ctx = ToolContext(db=db_session, session=await _session(db_session))
    msg = (
        ATTACHMENT_MARKER + " foto.png]:\nmanguera de jardín verde, 25 metros\n\n"
        "¿tenéis algo como esto?"
    )
    resp = await run_turn(router, ctx, msg)
    assert resp.reply == "Claro, busco."  # the turn really flowed
    systems = [m["content"] for m in router.prompts[0] if m["role"] == "system"]
    assert any(PHOTO_GROUNDING_NOTE in s for s in systems)


async def test_attachment_marker_alone_does_not_force_product_search(db_session):
    """A pure document question with an attachment must NOT inject the photo
    note: the answer is already in the attached text."""

    class Capture:
        def __init__(self):
            self.prompts = []

        async def chat(self, messages, tools, tier="large", temperature=None, order=None, tool_choice=None, max_tokens=None):
            self.prompts.append([dict(m) for m in messages])
            return ChatResult(content="Pone 78901.", tool_calls=[], finish_reason="stop")

    router = Capture()
    ctx = ToolContext(db=db_session, session=await _session(db_session))
    msg = (
        ATTACHMENT_MARKER + " factura.pdf]:\nFactura 78901 garantia dos anios\n\n"
        "¿qué pone en el documento?"
    )
    resp = await run_turn(router, ctx, msg)
    assert resp.reply == "Pone 78901."  # the turn really flowed
    systems = [m["content"] for m in router.prompts[0] if m["role"] == "system"]
    assert all(PHOTO_GROUNDING_NOTE not in s for s in systems)


async def test_intent_gates_ignore_trigger_words_inside_attachment_text(db_session):
    """ADVERSARIAL: the attachment text is whatever the vision model / PDF
    extractor produced — it may well contain 'precio', 'producto', 'similar'
    or 'stock' (an invoice, a photo with a visible price tag). With
    intent_text (the customer's own words, exactly as the /chat route passes
    it), those words must NEVER force a grounding note onto a pure document
    question."""

    class Capture:
        def __init__(self):
            self.prompts = []

        async def chat(self, messages, tools, tier="large", temperature=None, order=None, tool_choice=None, max_tokens=None):
            self.prompts.append([dict(m) for m in messages])
            return ChatResult(content="Es una factura de 49,90.", tool_calls=[], finish_reason="stop")

    router = Capture()
    ctx = ToolContext(db=db_session, session=await _session(db_session))
    question = "¿qué pone en el documento?"
    msg = (
        ATTACHMENT_MARKER + " factura.pdf]:\n"
        "Factura 78901 precio total 49,90 producto similar en stock\n\n" + question
    )
    resp = await run_turn(router, ctx, msg, intent_text=question)
    assert resp.reply == "Es una factura de 49,90."  # the turn really flowed
    systems = [m["content"] for m in router.prompts[0] if m["role"] == "system"]
    # only the base system prompt survives: neither the photo note nor the
    # generic grounding note was triggered by the attachment's own words
    assert len(systems) == 1
    assert PHOTO_GROUNDING_NOTE not in systems[0]


async def test_tool_budget_exhausted_forces_final_answer(db_session):
    # Always returns tool_calls; orchestrator must stop and force a final answer.
    loop_result = ChatResult(
        content=None,
        tool_calls=[ToolCall(id="t", name="search_knowledge", arguments={"query": "x"})],
        finish_reason="tool_calls",
    )
    router = FakeRouter(
        [loop_result, loop_result, loop_result, loop_result,
         ChatResult(content="Respuesta final.", tool_calls=[], finish_reason="stop")]
    )
    ctx = ToolContext(db=db_session, session=await _session(db_session))
    resp = await run_turn(router, ctx, "bucle")
    assert resp.reply == "Respuesta final."
    assert router.calls == 5  # 4 tool iters + 1 forced final


# --- verification: the OPEN grounding actually reaches the model -------------

class CapturingRouter:
    """Records the messages sent to the LLM on the first call, then answers."""

    def __init__(self, reply="ok"):
        self.reply = reply
        self.seen = None

    async def chat(self, messages, tools, tier="large", temperature=None, order=None, tool_choice=None, max_tokens=None):
        if self.seen is None:
            self.seen = list(messages)
        return ChatResult(content=self.reply, tool_calls=[], finish_reason="stop")

    def system_text(self):
        return "\n".join(m.get("content", "") for m in (self.seen or []) if m.get("role") == "system").lower()


@pytest.mark.parametrize(
    "q",
    [
        "si pido una srx-100 cuanto tarda en llegarme",   # delivery time
        "necesito las medidas de la asx-90",               # dimensions
        "es compatible con discos de 28mm?",               # compatibility
        "mis discos de 28 valen en la asx-2000?",          # a DIFFERENT model
        "que material es la barra",                         # specs
    ],
)
async def test_real_question_forces_grounding_note(db_session, q):
    """Every real store question must reach the model WITH the grounding note that
    forces a tool search first and forbids inventing / reusing other products /
    redirecting to third parties — whatever the wording."""
    r = CapturingRouter(reply="...")
    ctx = ToolContext(db=db_session, session=await _session(db_session))
    await run_turn(r, ctx, q, brand_name="Tienda")
    sys = r.system_text()
    assert "consulta siempre primero tus herramientas" in sys   # search-first
    assert "no reutilices datos de otro" in sys                 # no product conflation
    assert "tercero" in sys and "prohibido inventar" in sys     # no redirect / no invent


async def test_smalltalk_does_not_force_grounding_note(db_session):
    r = CapturingRouter(reply="¡Hola! ¿En qué te ayudo?")
    ctx = ToolContext(db=db_session, session=await _session(db_session))
    await run_turn(r, ctx, "hola, buenos días", brand_name="Tienda")
    assert "consulta siempre primero tus herramientas" not in r.system_text()


async def test_delivery_question_runs_knowledge_search_and_answers(db_session, monkeypatch):
    """The exact production failure: a delivery-time question now triggers a
    knowledge search and answers from it (instead of free-wheeling 'no info')."""
    from app.models import KnowledgeChunk
    from app.rag import index

    async def fake_search(session, query, k=4, *, tenant_id=None):
        chunk = KnowledgeChunk(
            source_id=999, tenant_id=tenant_id, ordinal=0,
            text="Envíos a la península en 24-48h.", embedding=[], meta={"source_name": "envios"},
        )
        return [(chunk, 0.92)]

    monkeypatch.setattr(index, "search", fake_search)
    router = FakeRouter([
        ChatResult(content=None, tool_calls=[ToolCall(id="t1", name="search_knowledge", arguments={"query": "plazo de entrega peninsula"})], finish_reason="tool_calls"),
        ChatResult(content="Llega en 24-48h a la península.", tool_calls=[], finish_reason="stop"),
    ])
    ctx = ToolContext(db=db_session, session=await _session(db_session))
    resp = await run_turn(router, ctx, "cuanto tarda en llegarme la srx-100", brand_name="Tienda")
    assert resp.used_tools == ["search_knowledge"]   # it consulted the brain
    assert "24-48" in resp.reply


async def test_cart_email_auto_attached_deterministically(db_session):
    """Recovery is on, the bot already built a cart this session, and the customer
    types their email — but the (weak) model forgets to call save_cart. The email must
    still be attached to the open cart IN CODE, so the recovery nudge can fire."""
    from sqlalchemy import select

    from app import carts
    from app.models import AbandonedCart

    sess = await _session(db_session)
    await carts.capture(db_session, 1, session_id=sess.id, checkout_url="https://x/cart/1:1")
    # plain replies (no save_cart) — twice, since the grounding gate nudges once
    router = FakeRouter([
        ChatResult(content="Vale, te lo guardo.", tool_calls=[], finish_reason="stop"),
        ChatResult(content="Vale, te lo guardo.", tool_calls=[], finish_reason="stop"),
    ])
    ctx = ToolContext(db=db_session, session=sess, tenant_id=1, allow_cart_recovery=True)
    await run_turn(router, ctx, "guárdame el carrito y avísame a Buyer@Gmail.com")
    cart = (await db_session.execute(
        select(AbandonedCart).where(AbandonedCart.session_id == sess.id))).scalar_one()
    assert cart.email == "buyer@gmail.com"  # attached deterministically, normalized


async def test_third_party_email_not_auto_attached(db_session):
    """A business/manufacturer email the visitor merely QUOTES must never become the
    recovery target — only the shopper's own (freemail / known) address is attached."""
    from sqlalchemy import select

    from app import carts
    from app.models import AbandonedCart

    sess = await _session(db_session)
    await carts.capture(db_session, 1, session_id=sess.id, checkout_url="https://x/cart/1:1")
    router = FakeRouter([
        ChatResult(content="Eso lo gestiona la tienda.", tool_calls=[], finish_reason="stop"),
        ChatResult(content="Eso lo gestiona la tienda.", tool_calls=[], finish_reason="stop"),
    ])
    ctx = ToolContext(db=db_session, session=sess, tenant_id=1, allow_cart_recovery=True)
    await run_turn(router, ctx, "vi en vuestra web el correo soporte@marca-fabricante.com, ¿es de garantía?")
    cart = (await db_session.execute(
        select(AbandonedCart).where(AbandonedCart.session_id == sess.id))).scalar_one()
    assert cart.email is None  # the third-party address was NOT captured


async def test_no_cart_email_attach_when_recovery_off(db_session):
    """With recovery OFF, the deterministic attach must NOT run (no cart, no capture)."""
    from sqlalchemy import select

    from app.models import AbandonedCart

    sess = await _session(db_session)
    router = FakeRouter([
        ChatResult(content="ok", tool_calls=[], finish_reason="stop"),
        ChatResult(content="ok", tool_calls=[], finish_reason="stop"),
    ])
    ctx = ToolContext(db=db_session, session=sess, tenant_id=1, allow_cart_recovery=False)
    await run_turn(router, ctx, "mi email es buyer@mail.com guárdame algo")
    rows = (await db_session.execute(
        select(AbandonedCart).where(AbandonedCart.session_id == sess.id))).scalars().all()
    assert rows == []


class _CapturingRouter:
    def __init__(self, results):
        self._results = list(results)
        self.seen = []

    async def chat(self, messages, tools, tier="large", temperature=None, order=None,
                   tool_choice=None, max_tokens=None):
        self.seen.append([m.get("content", "") for m in messages])
        return self._results.pop(0)


async def test_bare_photo_forces_product_search(db_session):
    """A photo uploaded with NO substantive text ('do you have this?') deterministically
    forces a search_products call even if the model would have just chatted."""
    sess = await _session(db_session)
    router = _CapturingRouter([
        ChatResult(content="Bonita foto.", tool_calls=[], finish_reason="stop"),   # model dodges
        ChatResult(content="Esto es lo que tenemos.", tool_calls=[], finish_reason="stop"),
    ])
    ctx = ToolContext(db=db_session, session=sess, tenant_id=1)
    await run_turn(
        router, ctx, ATTACHMENT_MARKER + " foto.jpg]:\nuna zapatilla roja de running",
        intent_text="", image_uploaded=True,
    )
    flat = " ".join(c for call in router.seen for c in call)
    assert "search_products" in flat  # the deterministic force kicked in


async def test_pdf_or_plain_text_does_not_force_product_search(db_session):
    """No image uploaded + a normal question must NOT trigger the photo->product force."""
    sess = await _session(db_session)
    router = _CapturingRouter([
        ChatResult(content="Te ayudo con eso.", tool_calls=[], finish_reason="stop"),
        ChatResult(content="Aquí tienes.", tool_calls=[], finish_reason="stop"),
    ])
    ctx = ToolContext(db=db_session, session=sess, tenant_id=1)
    await run_turn(router, ctx, "¿cuánto tarda el envío?", intent_text="¿cuánto tarda el envío?",
                   image_uploaded=False)
    flat = " ".join(c for call in router.seen for c in call)
    # it may force search_knowledge (grounding), but NOT the image->search_products note
    assert "atributos de la imagen" not in flat


async def test_tool_exception_degrades_not_500(db_session):
    """A tool raising must NOT crash the turn — it degrades to a model-readable error."""
    router = FakeRouter([
        ChatResult(content=None, tool_calls=[
            ToolCall(id="t1", name="search_products", arguments={"query": "x"})],
            finish_reason="tool_calls"),
        ChatResult(content="Lo siento, ahora no puedo consultarlo.", tool_calls=[], finish_reason="stop"),
    ])
    ctx = ToolContext(db=db_session, session=await _session(db_session), tenant_id=1)

    import app.tools.registry as reg

    async def boom(name, args, c):
        raise RuntimeError("tool blew up")

    # monkeypatch dispatch to raise
    orig = reg.dispatch
    reg.dispatch = boom
    try:
        resp = await run_turn(router, ctx, "enséñame algo")
    finally:
        reg.dispatch = orig
    assert resp.reply  # a real reply, no exception propagated


async def test_photo_visual_search_backstop_runs_when_model_dodges(db_session, monkeypatch):
    """A photo + product intent, but the model never calls search_products (dodges with
    a clarification). The deterministic backstop must run search_products in code so the
    shopper still gets product matches."""
    from app.models import ProductChunk
    import app.tools.products_tool as pt

    sess = await _session(db_session)
    db_session.add(ProductChunk(tenant_id=1, shopify_product_id="SB1", title="The Minimal Snowboard",
                                text="snowboard", embedding=[0.0, 0.0, 0.0]))
    await db_session.flush()

    async def fake_kw(client, query, max_products=50):
        return []

    async def fake_sem(session, query, k=8, *, tenant_id=None):
        # the text arm matches the snowboard by the vision description
        return [(db_session.identity_map.get((ProductChunk, (1,))) or
                 (await session.execute(__import__("sqlalchemy").select(ProductChunk))).scalars().first(), 0.9)]

    async def fake_hydrate(client, ids):
        return [{"product_id": "SB1", "title": "The Minimal Snowboard", "price": "885", "available": True,
                 "image": "", "url": "u", "variants": [], "tags": [], "description": "", "product_type": ""}]

    monkeypatch.setattr(pt, "search_products", fake_kw)
    monkeypatch.setattr(pt, "fetch_products_by_ids", fake_hydrate)
    monkeypatch.setattr(pt.product_index, "search", fake_sem)

    # model dodges every time (plain replies, never a tool call)
    router = FakeRouter([
        ChatResult(content="¿Qué producto buscas?", tool_calls=[], finish_reason="stop"),
        ChatResult(content="¿Qué producto buscas?", tool_calls=[], finish_reason="stop"),
        ChatResult(content="Esto es lo más parecido a tu foto.", tool_calls=[], finish_reason="stop"),
    ])

    class _Shop:
        async def execute(self, q, variables=None):
            return {"products": {"edges": []}}

    ctx = ToolContext(db=db_session, session=sess, tenant_id=1, shopify=_Shop(),
                      shop="x.myshopify.com", image_uploaded=True)
    resp = await run_turn(
        router, ctx, ATTACHMENT_MARKER + " foto.jpg]:\nLa tabla de snowboard roja.\n\n¿Tenéis algo parecido a esto?",
        intent_text="¿Tenéis algo parecido a esto?", image_uploaded=True,
    )
    assert "search_products" in resp.used_tools  # the backstop fired in code
    assert any("Snowboard" in (c.get("title") or "") for c in resp.products)  # cards shown


async def test_returning_customer_last_topic_is_written(db_session):
    """Feature-2 completeness: the customer's last SUBSTANTIVE topic is recorded so the
    next session's greeting can pick up the thread; a greeting must not overwrite it."""
    from app.models import Customer

    cust = Customer(tenant_id=1, name="Ana", chats=1)
    db_session.add(cust)
    await db_session.flush()
    ctx = ToolContext(db=db_session, session=await _session(db_session), tenant_id=1, customer=cust)
    router = FakeRouter([
        ChatResult(content=None, finish_reason="tool_calls",
                   tool_calls=[ToolCall(id="t1", name="search_knowledge", arguments={"query": "envíos"})]),
        ChatResult(content="Los envíos tardan 48h.", tool_calls=[], finish_reason="stop"),
    ])
    await run_turn(router, ctx, "¿cuánto tardan los envíos?", intent_text="¿cuánto tardan los envíos?")
    assert "envíos" in (cust.last_topic or "")  # substantive topic recorded

    # a later pure greeting must NOT clobber the remembered topic
    await run_turn(FakeRouter([ChatResult(content="¡Hola!", tool_calls=[], finish_reason="stop")]),
                   ctx, "hola", intent_text="hola")
    assert "envíos" in (cust.last_topic or "")