"""FlowDeck — Agent tests (v4.10.0).

Covers the DB schema, LLM client (offline mock), permission manager, tool
registry (writes + undo snapshots + rollback), the ReAct engine (SSE events)
and the /api/agent/* router (agents, conversations, run, audit, skills).
"""
import asyncio
import json
import os
import tempfile

import pytest
from conftest import login_test_client
from fastapi.testclient import TestClient


@pytest.fixture
def client():
    db_file = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
    db_path = db_file.name
    db_file.close()

    os.environ["DATABASE_URL"] = f"sqlite:///{db_path}"
    os.environ["APP_SECRET_KEY"] = "test-secret-for-tests"
    os.environ["RATE_LIMIT_ENABLED"] = "false"
    os.environ["LLM_PROVIDER"] = "offline"  # deterministic, no network

    from app.config import settings
    from app.db import get_conn, init_db
    from app.main import app
    from app.password_utils import hash_password

    # Ensure settings singleton uses OUR temp DB (not one set by another test file).
    settings.database_url = f"sqlite:///{db_path}"
    settings.llm_provider = "offline"
    settings.agent_max_iterations = 12
    settings.agent_max_tokens_budget = 500_000
    settings.agent_run_timeout_seconds = 30

    init_db()
    with get_conn() as conn:
        conn.execute(
            "INSERT OR IGNORE INTO users (login, full_name, email, password_hash, is_admin) "
            "VALUES ('admin', 'Admin', '', ?, 1)",
            (hash_password("test"),),
        )
        conn.commit()

    yield login_test_client(TestClient(app))

    try:
        os.unlink(db_path)
    except FileNotFoundError:
        pass


def _admin_id() -> int:
    from app.db import get_conn
    with get_conn() as conn:
        row = conn.execute("SELECT id FROM users WHERE login='admin'").fetchone()
    return row["id"] if row else 1


def _collect(events):
    return [e for e in events]


async def _run_engine(engine, *args, **kwargs):
    return [e async for e in engine.run(*args, **kwargs)]


# ── LLM client (offline mock) ──

def test_llm_client_offline_mock(client):
    from app.services.llm_client import LLMClient
    llm = LLMClient(provider="offline")
    resp = asyncio.run(llm.complete(
        [{"role": "user", "content": "create collection Projets"}], tools=[],
    ))
    assert resp.tool_calls, "offline mock should detect a create-collection intent"
    assert resp.tool_calls[0]["name"] == "create_collection"
    assert resp.tool_calls[0]["arguments"]["name"] == "Projets"


def test_llm_client_providers_listed(client):
    from app.services.llm_client import PROVIDERS
    for name in ("openai", "anthropic", "google", "deepseek", "qwencloud",
                 "nvidia", "openrouter", "ollama", "offline"):
        assert name in PROVIDERS


def test_mistral_defaults_are_tier_safe(client):
    """La clé Mistral plan basique → 403 « tier_not_allowed » sur mistral-large/pixtral-large.
    Le modèle par défaut et le premier de la liste du test de connexion doivent
    être des modèles servis par tous les plans."""
    from app.services.llm_client import PROVIDER_MODELS, PROVIDERS
    tier_blocked = {"mistral-large-latest", "pixtral-large-latest"}
    assert PROVIDERS["mistral"][1] not in tier_blocked
    assert PROVIDER_MODELS["mistral"][0] not in tier_blocked
    # le fallback du modèle retiré (PROVIDERS default) est présent dans la liste
    assert PROVIDERS["mistral"][1] in PROVIDER_MODELS["mistral"]
    # et la liste des modèles Mistral est validée chat à chaque fetch (403 exclus)
    from app.services.llm_config import _CHAT_VALIDATED_PROVIDERS
    assert "mistral" in _CHAT_VALIDATED_PROVIDERS


def test_llm_client_openai_compatible_bases(client):
    """Providers that need an OpenAI-compatible surface must point at it.

    Anthropic/Google/Cohere native APIs are not chat-completions endpoints:
    the client always speaks `/chat/completions` with a Bearer token, so the
    base URLs must target the compatibility surfaces (validated live).
    """
    from app.services.llm_client import PROVIDERS
    assert PROVIDERS["cohere"][0] == "https://api.cohere.ai/compatibility/v1"
    assert PROVIDERS["google"][0] == (
        "https://generativelanguage.googleapis.com/v1beta/openai")
    # Perplexity serves /chat/completions at the host root (no /v1 prefix).
    assert PROVIDERS["perplexity"][0] == "https://api.perplexity.ai"
    # Chutes' OpenAI-compatible inference host.
    assert PROVIDERS["chutes"][0] == "https://llm.chutes.ai/v1"
    assert PROVIDERS["sensenova"][0].endswith("/compatible-mode/v1")


def test_user_llm_key_api_base_normalization(client):
    """A stored api_base equal to the provider default is not persisted as an
    override, so a corrected provider URL always takes effect."""
    from app.services.llm_config import get_user_llm_key, upsert_user_llm_key

    uid = _admin_id()
    # default base → stored as empty (no override)
    upsert_user_llm_key(uid, "mistral", api_key="sk-test",
                        api_base="https://api.mistral.ai/v1")
    assert get_user_llm_key(uid, "mistral")["api_base"] == ""
    # custom base → kept
    upsert_user_llm_key(uid, "mistral", api_base="https://proxy.example.com/v1")
    assert get_user_llm_key(uid, "mistral")["api_base"] == "https://proxy.example.com/v1"
    # api_base="" resets the override, api_base=None keeps it
    upsert_user_llm_key(uid, "mistral", api_base="")
    assert get_user_llm_key(uid, "mistral")["api_base"] == ""
    upsert_user_llm_key(uid, "mistral", api_base="https://proxy.example.com/v1")
    upsert_user_llm_key(uid, "mistral", default_model="mistral-large-latest")
    assert get_user_llm_key(uid, "mistral")["api_base"] == "https://proxy.example.com/v1"


def test_migration_clears_stale_llm_api_bases(client):
    """Migration 14 clears legacy/incorrect api_base values so the corrected
    provider default applies (e.g. Mistral `/v2` → `/v1`)."""
    from app.db import get_conn
    from app.migrations import _migration_fix_llm_api_bases
    from app.services.llm_config import get_user_llm_key, upsert_user_llm_key

    uid = _admin_id()
    upsert_user_llm_key(uid, "mistral", api_key="sk-test")
    with get_conn() as conn:
        conn.execute(
            "UPDATE user_llm_keys SET api_base='https://api.mistral.ai/v2' "
            "WHERE user_id=? AND provider='mistral'",
            (uid,),
        )
        conn.commit()
    assert get_user_llm_key(uid, "mistral")["api_base"] == "https://api.mistral.ai/v2"
    with get_conn() as conn:
        _migration_fix_llm_api_bases(conn)
        conn.commit()
    assert get_user_llm_key(uid, "mistral")["api_base"] == ""


def test_llm_client_offline_no_credentials(client):
    from app.services.llm_client import LLMClient
    llm = LLMClient(provider="openai", api_key="")
    resp = asyncio.run(llm.complete(
        [{"role": "user", "content": "hello"}], tools=[],
    ))
    # No key → falls back to mock (a final text answer, no crash)
    assert resp.text != "" or resp.tool_calls


def test_llm_client_offline_mock_creates_document(client):
    """A document creation request must resolve to the create_document tool."""
    from app.services.llm_client import LLMClient
    llm = LLMClient(provider="offline")
    resp = asyncio.run(llm.complete(
        [{"role": "user", "content": 'Faire la création d\'un document "Projet" '
                                    'dans le workspace "TEST WS 1"'}],
        tools=[],
    ))
    assert resp.tool_calls
    assert resp.tool_calls[0]["name"] == "create_document"
    args = resp.tool_calls[0]["arguments"]
    assert args["title"] == "Projet"
    assert args["workspace_name"] == "TEST WS 1"


# ── DB schema ──

def test_agent_tables_exist(client):
    tables = {"agents", "agent_conversations", "agent_messages",
              "agent_actions", "agent_skills", "agent_triggers",
              "agent_feedback"}
    from app.db import get_conn
    with get_conn() as conn:
        rows = conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'"
        ).fetchall()
    names = {r["name"] for r in rows}
    assert tables.issubset(names)


# ── Tools ──

def test_tool_create_collection_and_search(client):
    from app.services.tool_registry import ToolRegistry
    reg = ToolRegistry()
    res = asyncio.run(reg.execute("create_collection", {"name": "Projets"}))
    assert res.status == "success"
    cid = res.data["collection_id"]
    assert res.undo == {"action": "delete", "table": "collections", "id": cid}

    search = asyncio.run(reg.execute("search_workspace", {"query": "Projets"}))
    assert search.status == "success"
    assert any(c["id"] == cid for c in search.data["collections"])


def test_tool_create_page_with_properties(client):
    from app.services.tool_registry import ToolRegistry
    reg = ToolRegistry()
    coll = asyncio.run(reg.execute("create_collection", {"name": "Tasks"}))
    cid = coll.data["collection_id"]
    asyncio.run(reg.execute("add_property", {"collection_id": cid, "name": "Status", "prop_type": "select", "options": ["Todo", "Done"]}))
    page = asyncio.run(reg.execute("create_page", {"collection_id": cid, "title": "Task 1", "properties": {"Status": "Todo"}}))
    assert page.status == "success"
    assert page.target_type == "page"
    assert page.data["title"] == "Task 1"


def test_tool_unknown_returns_error(client):
    from app.services.tool_registry import ToolRegistry
    res = asyncio.run(ToolRegistry().execute("nope", {}))
    assert res.status == "error"


def test_tool_workspace_document_crud_and_search(client):
    from app.db import get_conn
    from app.services.tool_registry import ToolRegistry
    with get_conn() as conn:
        conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('TEST WS 1', ?)",
                     (_admin_id(),))
        conn.commit()

    reg = ToolRegistry()

    ws = asyncio.run(reg.execute("read_workspaces", {}))
    assert ws.status == "success"
    assert any(w["name"] == "TEST WS 1" for w in ws.data["workspaces"])

    res = asyncio.run(reg.execute("create_document",
                                  {"title": "Projet", "workspace_name": "TEST WS 1"}))
    assert res.status == "success", res.message
    pid = res.data["document_id"]
    assert res.target_type == "document"
    assert res.undo == {"action": "delete", "table": "pages", "id": pid}

    rd = asyncio.run(reg.execute("read_document", {"page_id": pid}))
    assert rd.status == "success"
    assert rd.data["document"]["title"] == "Projet"

    ws2 = asyncio.run(reg.execute("read_workspaces", {}))
    row = next(w for w in ws2.data["workspaces"] if w["name"] == "TEST WS 1")
    assert row["document_count"] == 1

    search = asyncio.run(reg.execute("search_workspace", {"query": "Projet"}))
    assert any(d["id"] == pid for d in search.data["documents"])

    # write_blocks can also rename the document
    upd = asyncio.run(reg.execute("write_blocks", {"page_id": pid, "title": "Projet v2"}))
    assert upd.status == "success"
    rd2 = asyncio.run(reg.execute("read_document", {"page_id": pid}))
    assert rd2.data["document"]["title"] == "Projet v2"

    # unknown workspace name → clear error, no document created
    bad = asyncio.run(reg.execute("create_document",
                                  {"title": "X", "workspace_name": "NOPE"}))
    assert bad.status == "error"
    assert "introuvable" in bad.message


def test_write_blocks_normalizes_llm_blocks(client):
    """LLM blocks are normalized: `text`→`content`, `heading`→`heading_1`,
    missing ids are generated, unknown types fall back to paragraph."""
    from app.services.tool_registry import ToolRegistry

    reg = ToolRegistry()
    doc = asyncio.run(reg.execute("create_document", {"title": "Script"}))
    pid = doc.data["document_id"]

    bad_blocks = [
        {"type": "heading", "text": "test.ps1 — Script de test PowerShell"},
        {"type": "paragraph", "text": "# Test script\nWrite-Host 'Hello, World!'"},
        {"type": "code", "text": "Write-Host 'hi'", "language": "powershell"},
        {"type": "made_up_type", "content": "inconnu"},
    ]
    res = asyncio.run(reg.execute("write_blocks", {"page_id": pid, "blocks": bad_blocks}))
    assert res.status == "success", res.message

    from app.db import get_conn
    with get_conn() as conn:
        row = conn.execute("SELECT content, content_format FROM pages WHERE id=?", (pid,)).fetchone()
    assert row["content_format"] == "blocks"
    stored = json.loads(row["content"])

    assert stored[0]["type"] == "heading_1", stored[0]
    assert stored[0]["content"] == "test.ps1 — Script de test PowerShell"
    assert stored[0]["id"], "id must be generated"

    assert stored[1]["type"] == "paragraph"
    assert stored[1]["content"] == "# Test script\nWrite-Host 'Hello, World!'"

    assert stored[2]["type"] == "code"
    assert stored[2]["content"] == "Write-Host 'hi'"
    assert stored[2]["language"] == "powershell"

    assert stored[3]["type"] == "paragraph"   # unknown → paragraph
    assert stored[3]["content"] == "inconnu"


def test_engine_titles_conversation_on_each_request(client):
    """The conversation title is refreshed after EVERY AI request (not only
    while the title is a default placeholder)."""
    from app.services.agent_engine import AgentEngine

    engine = AgentEngine(_admin_id(), workspace_id=None)
    conv = client.post("/api/agent/conversations",
                       json={"title": "Nouvelle conversation"}).json()["id"]
    first = "crée un document « Rapport Q3 »"
    asyncio.run(_run_engine(engine, conv, first))
    detail = client.get(f"/api/agent/conversations/{conv}").json()
    title1 = detail["conversation"]["title"]
    assert "Rapport Q3" in title1

    second = "crée une collection « Launches »"
    asyncio.run(_run_engine(engine, conv, second))
    detail = client.get(f"/api/agent/conversations/{conv}").json()
    title2 = detail["conversation"]["title"]
    assert "Launches" in title2, title2
    assert title2 != title1, "title must be updated on the second request too"


def test_suggest_title_strips_composer_prefixes(client):
    from app.services.agent_engine import AgentEngine
    title = AgentEngine._suggest_title(
        "Contexte « test.ps1 »\n\nÉcrire un script de test en powershell dans la page courante",
        None,
    )
    assert title != "Contexte", title
    assert "Contexte" not in title
    assert title


def test_tool_delete_document_and_undo(client):
    from app.db import get_conn
    from app.services.agent_engine import undo_action
    from app.services.tool_registry import ToolRegistry

    reg = ToolRegistry()
    res = asyncio.run(reg.execute("create_document", {"title": "À supprimer"}))
    pid = res.data["document_id"]

    dele = asyncio.run(reg.execute("delete_document", {"page_id": pid}))
    assert dele.status == "success"
    assert dele.undo == {"action": "softdelete", "table": "pages", "id": pid}

    with get_conn() as conn:
        row = conn.execute("SELECT deleted_at FROM pages WHERE id=?", (pid,)).fetchone()
        assert row is not None and row["deleted_at"] is not None

    # store an audit row so undo_action can restore the document
    conv_id = client.post("/api/agent/conversations", json={"title": "t"}).json()["id"]
    with get_conn() as conn:
        cur = conn.execute(
            """INSERT INTO agent_actions (conversation_id, tool_name, target_type,
               target_id, payload_json, result_json, status, undo_snapshot_json, executed_by)
               VALUES (?, 'delete_document', 'document', ?, '{}', '{}', 'success', ?, ?)""",
            (conv_id, str(pid), json.dumps(dele.undo), _admin_id()),
        )
        conn.commit()
        aid = cur.lastrowid

    assert undo_action(aid) is True
    with get_conn() as conn:
        row = conn.execute("SELECT deleted_at FROM pages WHERE id=?", (pid,)).fetchone()
        assert row["deleted_at"] is None


# ── Permissions ──

def test_permission_manager_roles(client):
    from app.db import get_conn
    from app.services.permission_manager import PermissionManager
    with get_conn() as conn:
        owner = conn.execute("SELECT id FROM users WHERE login='admin'").fetchone()
        conn.execute("INSERT INTO users (login, is_admin) VALUES ('viewer_user', 0)")
        viewer = conn.execute("SELECT id FROM users WHERE login='viewer_user'").fetchone()
        conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('ws', ?)", (owner["id"],))
        ws = conn.execute("SELECT id FROM workspaces WHERE name='ws'").fetchone()
        conn.execute("INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?,?,?)", (ws["id"], viewer["id"], "viewer"))
        conn.execute("INSERT INTO workspace_members (workspace_id, user_id, role) VALUES (?,?,?)", (ws["id"], owner["id"], "owner"))
        conn.commit()

    pm_owner = PermissionManager(owner["id"])
    pm_viewer = PermissionManager(viewer["id"])
    assert pm_owner.can_write(ws["id"]) is True
    assert pm_viewer.can_write(ws["id"]) is False

    from fastapi import HTTPException
    # viewer cannot write
    with pytest.raises(HTTPException):
        pm_viewer.assert_can("create_page", {}, ws["id"])
    # owner can write
    pm_owner.assert_can("create_page", {}, ws["id"])
    # destructive requires confirm mode
    with pytest.raises(HTTPException):
        pm_owner.assert_can("delete_page", {}, ws["id"], approval_mode="auto")
    pm_owner.assert_can("delete_page", {}, ws["id"], approval_mode="confirm")


# ── Engine (ReAct + audit + rollback) ──

def test_engine_run_creates_collection(client):
    from app.db import get_conn
    from app.services.agent_engine import AgentEngine

    engine = AgentEngine(_admin_id(), workspace_id=None)
    events = asyncio.run(_run_engine(engine, _make_conversation(client), "crée une collection Stats"))

    types = [e["type"] for e in events]
    assert "reasoning" in types
    assert "action" in types
    assert "final" in types

    actions = [e for e in events if e["type"] == "action"]
    assert any(a["tool"] == "create_collection" and a["status"] == "success" for a in actions)

    # Collection actually persisted
    with get_conn() as conn:
        row = conn.execute("SELECT id FROM collections WHERE name='Stats'").fetchone()
    assert row is not None


def test_engine_audit_and_undo(client):
    from app.db import get_conn
    from app.services.agent_engine import AgentEngine, undo_action

    conv = _make_conversation(client)
    engine = AgentEngine(_admin_id())
    asyncio.run(_run_engine(engine, conv, "créer une collection CRM"))

    with get_conn() as conn:
        coll = conn.execute("SELECT id FROM collections WHERE name='CRM'").fetchone()
        assert coll is not None
        action = conn.execute(
            "SELECT id FROM agent_actions WHERE tool_name='create_collection' AND target_id=?",
            (str(coll["id"]),),
        ).fetchone()
        assert action is not None

    # undo removes the collection and marks the action reverted
    undo_action(action["id"])
    from app.db import get_conn as gc
    with gc() as conn:
        assert conn.execute("SELECT id FROM collections WHERE name='CRM'").fetchone() is None
        a = conn.execute("SELECT status FROM agent_actions WHERE id=?", (action["id"],)).fetchone()
        assert a["status"] == "reverted"


def test_engine_run_creates_document_in_workspace_and_titles_conversation(client):
    from app.db import get_conn
    from app.services.agent_engine import AgentEngine

    with get_conn() as conn:
        conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('TEST WS 1', ?)",
                     (_admin_id(),))
        conn.commit()

    engine = AgentEngine(_admin_id(), workspace_id=None)
    conv = client.post("/api/agent/conversations",
                       json={"title": "Nouvelle conversation"}).json()["id"]
    objective = 'Faire la création d\'un document « Projet » dans le workspace « TEST WS 1 »'
    events = asyncio.run(_run_engine(engine, conv, objective))

    actions = [e for e in events if e["type"] == "action"]
    assert any(a["tool"] == "create_document" and a["status"] == "success" for a in actions), actions

    with get_conn() as conn:
        doc = conn.execute("SELECT * FROM pages WHERE title='Projet'").fetchone()
        ws = conn.execute("SELECT id FROM workspaces WHERE name='TEST WS 1'").fetchone()
        assert doc is not None
        assert doc["workspace_id"] == ws["id"]

    # the conversation is renamed with a descriptive title
    detail = client.get(f"/api/agent/conversations/{conv}").json()
    title = detail["conversation"]["title"]
    assert title.lower() != "nouvelle conversation"
    assert "Projet" in title


class _Resp:
    def __init__(self, text="", tool_calls=None):
        self.text = text
        self.tool_calls = tool_calls or []
        self.usage = {}


def test_engine_tool_protocol_messages(client):
    """Le message assistant qui annonce un outil porte ses `tool_calls`, et le
    résultat d'outil est renvoyé avec le `tool_call_id` correspondant — sans quoi
    le fournisseur refuse la passe suivante (et l'agent retombait sur le mock)."""
    from app.services.agent_engine import AgentEngine

    class FakeLLM:
        def __init__(self):
            self.calls = []

        async def complete(self, messages, *, model=None, tools=None, stream=False):
            self.calls.append([dict(m) for m in messages])
            if len(self.calls) == 1:
                return _Resp("Plan de création.",
                             [{"id": "call_abc", "name": "create_collection",
                               "arguments": {"name": "ProtoCol"}}])
            return _Resp("Collection ProtoCol créée.", [])

    fake = FakeLLM()
    engine = AgentEngine(_admin_id(), llm=fake)
    conv = client.post("/api/agent/conversations",
                       json={"title": "protocol"}).json()["id"]
    events = asyncio.run(_run_engine(engine, conv, "crée la collection ProtoCol"))

    finals = [e for e in events if e["type"] == "final"]
    assert any("ProtoCol" in e.get("content", "") for e in finals), events

    # la 2e passe reçoit un bloc assistant `tool_calls` + un résultat d'outil lié
    call2 = fake.calls[1]
    asm = next((m for m in call2 if m["role"] == "assistant" and m.get("tool_calls")), None)
    tmsg = next((m for m in call2 if m["role"] == "tool"), None)
    assert asm is not None and tmsg is not None
    assert asm["tool_calls"][0]["id"] == "call_abc"
    assert tmsg["tool_call_id"] == asm["tool_calls"][0]["id"]
    payload = json.loads(tmsg["content"])
    assert payload["status"] == "ok"


def test_engine_multiple_skills_applied(client):
    """Plusieurs skills (skill_ids) sont injectés ensemble dans les instructions."""
    from app.db import get_conn
    from app.services.agent_engine import AgentEngine

    with get_conn() as conn:
        c1 = conn.execute(
            "INSERT INTO agent_skills (workspace_id, name, description, prompt_template, allowed_tools_json, created_by) "
            "VALUES (NULL, 'Skill A', '', 'Template A', '[]', ?)", (_admin_id(),))
        c2 = conn.execute(
            "INSERT INTO agent_skills (workspace_id, name, description, prompt_template, allowed_tools_json, created_by) "
            "VALUES (NULL, 'Skill B', '', 'Template B', '[]', ?)", (_admin_id(),))
        conn.commit()
        id1, id2 = c1.lastrowid, c2.lastrowid

    class FakeLLM:
        def __init__(self):
            self.system = ""

        async def complete(self, messages, *, model=None, tools=None, stream=False):
            self.system = next((m["content"] for m in messages if m["role"] == "system"), "")
            return _Resp("ok", [])

    fake = FakeLLM()
    engine = AgentEngine(_admin_id(), llm=fake)
    conv = client.post("/api/agent/conversations", json={"title": "multi-skill"}).json()["id"]
    asyncio.run(_run_engine(engine, conv, "bonjour", skill_ids=[id1, id2]))

    assert "Skill A" in fake.system and "Template A" in fake.system
    assert "Skill B" in fake.system and "Template B" in fake.system


def _make_conversation(client) -> int:
    resp = client.post("/api/agent/conversations", json={"title": "test"})
    return resp.json()["id"]
# ── Router: agents, conversations, run, skills ──

def test_router_list_agents(client):
    resp = client.get("/api/agent")
    assert resp.status_code == 200
    assert resp.json()["agents"]
    assert any(a["agent_type"] == "personal" for a in resp.json()["agents"])


def test_router_create_custom_agent(client):
    resp = client.post("/api/agent", json={"name": "Reporting", "agent_type": "custom",
                                            "system_instructions": "Produce weekly report"})
    assert resp.status_code == 200
    assert resp.json()["id"]
    # list now includes it
    agents = client.get("/api/agent").json()["agents"]
    assert any(a["name"] == "Reporting" for a in agents)


def test_router_conversation_crud(client):
    c = client.post("/api/agent/conversations", json={"title": "Hello"}).json()
    cid = c["id"]
    detail = client.get(f"/api/agent/conversations/{cid}")
    assert detail.status_code == 200
    assert detail.json()["conversation"]["id"] == cid
    assert client.delete(f"/api/agent/conversations/{cid}").status_code == 200
    assert client.get(f"/api/agent/conversations/{cid}").status_code == 404


def test_router_run_sse_stream(client):
    cid = _make_conversation(client)
    resp = client.post(f"/api/agent/conversations/{cid}/run",
                       json={"message": "create collection Roadmap"})
    assert resp.status_code == 200
    assert resp.headers["content-type"].startswith("text/event-stream")
    body = resp.text
    assert "data:" in body
    events = [_parse_sse(line) for line in body.strip().split("\n\n") if line.strip()]
    types = [e.get("type") for e in events if e]
    assert "final" in types


def test_router_run_malformed(client):
    cid = _make_conversation(client)
    resp = client.post(f"/api/agent/conversations/{cid}/run", json={})
    assert resp.status_code == 400


def test_router_actions_and_undo(client):
    cid = _make_conversation(client)
    client.post(f"/api/agent/conversations/{cid}/run", json={"message": "créer une collection Docs"})
    actions = client.get(f"/api/agent/conversations/{cid}/actions").json()["actions"]
    assert actions, "run should have logged actions"
    target = actions[0]
    assert target["tool_name"] == "create_collection"
    # undo via endpoint
    r = client.post(f"/api/agent/actions/{target['id']}/undo")
    assert r.status_code == 200
    assert r.json()["status"] == "reverted"


def test_router_skills(client):
    s = client.post("/api/agent/skills", json={"name": "Préparation de sprint",
                                               "prompt_template": "Analyse les issues et crée un sprint",
                                               "allowed_tools": ["read_gitea_issues", "create_collection"]})
    assert s.status_code == 200
    sid = s.json()["id"]
    skills = client.get("/api/agent/skills").json()["skills"]
    assert any(k["name"] == "Préparation de sprint" for k in skills)
    # apply → creates a conversation ready to run
    apply = client.post(f"/api/agent/skills/{sid}/apply")
    assert apply.status_code == 200
    assert apply.json()["conversation_id"]


def test_router_mentions(client):
    from app.db import get_conn
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO pages (workspace, title, content, content_format) "
            "VALUES ('', 'Compte rendu projet', '## Résumé\nDécision prise.', 'markdown')")
        doc_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
        conn.execute("INSERT INTO collections (name, icon) VALUES ('Roadmap', '🗺️')")
        coll_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
        conn.execute(
            "INSERT INTO collection_pages (collection_id, title, property_values_json) "
            "VALUES (?, 'Page Analyse', '{}')", (coll_id,))
        page_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
        conn.commit()

    r = client.get("/api/agent/mentions")
    assert r.status_code == 200
    data = r.json()
    sections = data["sections"]
    assert isinstance(sections, list) and sections
    tokens = {it["token"] for grp in sections for it in grp["items"]}
    assert f"document:{doc_id}" in tokens
    assert f"collection:{coll_id}" in tokens
    assert f"page:{page_id}" in tokens

    # search filter
    r2 = client.get("/api/agent/mentions", params={"q": "Compte rendu"})
    found = [it for grp in r2.json()["sections"] for it in grp["items"]]
    assert found and all(it["type"] == "document" and it["label"] == "Compte rendu projet" for it in found)

    # déduplication : deux fichiers avec le même titre n'apparaissent qu'une fois
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO pages (workspace, title, content, content_format) "
            "VALUES ('', 'Compte rendu projet', 'Bis', 'markdown')")
        conn.commit()
    r3 = client.get("/api/agent/mentions")
    dup_docs = [it for grp in r3.json()["sections"] for it in grp["items"]
                if it["type"] == "document" and it["label"] == "Compte rendu projet"]
    assert len(dup_docs) == 1, dup_docs

    # scoping : les fichiers de l'espace courant ouvert sont proposés en section dédiée
    with get_conn() as conn:
        conn.execute("INSERT INTO workspaces (name, owner_id) VALUES ('Espace Scope', ?)", (_admin_id(),))
        ws_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
        conn.execute(
            "INSERT INTO pages (workspace, workspace_id, title, content, content_format) "
            "VALUES ('Espace Scope', ?, 'Fichier du ws', '', 'markdown')", (ws_id,))
        conn.commit()
    r4 = client.get("/api/agent/mentions", params={"workspace_id": ws_id})
    secs = r4.json()["sections"]
    assert secs[0]["key"] == "files" and secs[0]["label"] == "Espace Scope"
    file_tokens = {it["token"] for it in secs[0]["items"]}
    assert any(t.startswith("document:") and (r4.json()["workspace_id"] == ws_id) for t in file_tokens)


def test_router_feedback(client):
    cid = _make_conversation(client)
    ok = client.post("/api/agent/feedback", json={
        "rating": "up", "conversation_id": cid, "content": "Merci, très utile",
    })
    assert ok.status_code == 200
    assert ok.json()["status"] == "recorded"

    from app.db import get_conn
    with get_conn() as conn:
        rows = conn.execute("SELECT rating, snippet FROM agent_feedback").fetchall()
    assert len(rows) == 1
    assert rows[0]["rating"] == "up"
    assert rows[0]["snippet"] == "Merci, très utile"

    bad = client.post("/api/agent/feedback", json={"rating": "meh"})
    assert bad.status_code == 400


def test_context_builder_document_mention(client):
    import json as _json

    from app.db import get_conn
    from app.services.context_builder import ContextBuilder

    with get_conn() as conn:
        conn.execute(
            "INSERT INTO pages (workspace, title, content, content_format) "
            "VALUES ('ws', 'Note brainstorming', 'Contenu ultra secret du doc.', 'markdown')")
        mid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
        conn.execute(
            "INSERT INTO pages (workspace, title, content, content_format) "
            "VALUES ('ws', 'Doc blocs', ?, 'blocks')",
            (_json.dumps([
                {"id": "b1", "type": "heading", "content": "Titre du bloc"},
                {"id": "b2", "type": "paragraph", "content": "Paragraphe du bloc.",
                 "children": [{"id": "b3", "type": "paragraph", "content": "Enfant aussi."}]},
            ]),))
        bid = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
        conn.commit()

    ctx = ContextBuilder(_admin_id()).build(mentions=[f"document:{mid}"])
    assert "Note brainstorming" in ctx
    assert "Contenu ultra secret du doc." in ctx

    ctx2 = ContextBuilder(_admin_id()).build(mentions=[f"document:{bid}"])
    assert "Titre du bloc" in ctx2
    assert "Paragraphe du bloc." in ctx2
    assert "Enfant aussi." in ctx2


def test_router_tools_and_providers(client):
    tools = client.get("/api/agent/tools").json()["tools"]
    assert len(tools) >= 10
    names = {t["name"] for t in tools}
    for expected in ("search_workspace", "create_collection", "create_page",
                     "update_page", "add_property", "create_view", "read_page",
                     "read_workspaces", "read_document", "create_document"):
        assert expected in names

    prov = client.get("/api/agent/providers").json()
    assert prov["provider"] == "offline"
    assert prov["available"] is False
    assert "providers" in prov


def test_router_providers_enriched(client):
    prov = client.get("/api/agent/providers").json()
    assert "providers" in prov
    ids = [p["id"] for p in prov["providers"]]
    for expected in ("openai", "anthropic", "google", "ollama", "offline"):
        assert expected in ids
    openai = next(p for p in prov["providers"] if p["id"] == "openai")
    assert openai["default_model"] == "gpt-4o"
    assert openai["models"]
    assert openai["requires_key"] is True
    # offline requires no key and exposes no models
    offline = next(p for p in prov["providers"] if p["id"] == "offline")
    assert offline["requires_key"] is False


def test_router_patch_providers_updates_config(client):
    from app.auth.session import SessionManager
    from app.db import get_conn

    with get_conn() as conn:
        admin = conn.execute("SELECT * FROM users WHERE login='admin'").fetchone()
        sess = SessionManager.create_session(dict(admin))

    client.cookies.set("flowdeck_session", sess)
    r = client.patch("/api/agent/providers",
                     json={"provider": "openai", "model": "gpt-4o-mini",
                           "api_key": "sk-test", "api_base": ""})
    assert r.status_code == 200
    body = r.json()
    assert body["provider"] == "openai"
    assert body["model"] == "gpt-4o-mini"
    assert body["has_api_key"] is True

    # persisted in llm_config (single row id=1)
    with get_conn() as conn:
        row = conn.execute("SELECT * FROM llm_config WHERE id=1").fetchone()
    assert row is not None
    assert row["provider"] == "openai"
    assert row["model"] == "gpt-4o-mini"
    assert row["api_key"] == "sk-test"

    # GET /providers reflects the new config
    prov = client.get("/api/agent/providers").json()
    assert prov["provider"] == "openai"
    assert prov["model"] == "gpt-4o-mini"
    assert prov["has_api_key"] is True
    # a real provider with a key is marked available (no network call made)
    assert prov["available"] is True


def test_router_patch_providers_requires_admin(client):
    from app.auth.session import SessionManager
    from app.db import get_conn

    with get_conn() as conn:
        conn.execute("INSERT INTO users (login, full_name, is_admin) VALUES ('user_basic', 'Basic', 0)")
        user = conn.execute("SELECT * FROM users WHERE login='user_basic'").fetchone()
        conn.commit()
        sess = SessionManager.create_session(dict(user))
    client.cookies.set("flowdeck_session", sess)

    r = client.patch("/api/agent/providers", json={"provider": "openai"})
    assert r.status_code == 403


def test_llm_client_default_model_provider_scoped(client):
    # Le modèle global ne doit pas fuir vers un autre provider testé (ex. nvidia
    # alors que deepseek est actif) sinon le provider reçoit un modèle inconnu
    # (« model not found », HTTP 404).
    from app.services.llm_client import LLMClient
    from app.services.llm_config import set_llm_config

    set_llm_config(provider="deepseek", model="deepseek-v4-flash",
                   api_key="sk-global", api_base="https://api.deepseek.com/v1")

    llm = LLMClient(provider="nvidia", api_key="nvapi-xxx")
    assert llm.provider == "nvidia"
    assert llm.default_model == "nvidia/nemotron-3-super-120b-a12b"
    assert llm.default_model != "deepseek-v4-flash"

    # le provider global conserve bien son modèle
    llm2 = LLMClient()
    assert llm2.provider == "deepseek"
    assert llm2.default_model == "deepseek-v4-flash"


def test_router_providers_test_offline(client):
    r = client.post("/api/agent/providers/test", json={"provider": "offline"})
    assert r.status_code == 200
    assert r.json()["ok"] is True


def test_router_providers_test_missing_key(client):
    # openai without a key must report failure (no mock fallback for the ping)
    r = client.post("/api/agent/providers/test",
                    json={"provider": "openai", "api_key": ""})
    assert r.status_code == 200
    assert r.json()["ok"] is False
    assert "clé" in r.json()["error"].lower()


def test_router_run_persists_provider_model(client):
    cid = _make_conversation(client)
    resp = client.post(f"/api/agent/conversations/{cid}/run",
                       json={"message": "crée une collection Bilan",
                             "provider": "openai", "model": "gpt-4o-mini"})
    assert resp.status_code == 200
    detail = client.get(f"/api/agent/conversations/{cid}").json()
    assert detail["conversation"]["provider"] == "openai"
    assert detail["conversation"]["model"] == "gpt-4o-mini"


def _parse_sse(line: str) -> dict:
    for part in line.split("\n"):
        if part.startswith("data:"):
            try:
                return json.loads(part[len("data:"):].strip())
            except json.JSONDecodeError:
                return {}


# ── Per-user provider keys (v4.10.2) ──


def _session_for_login(login: str) -> str:
    from app.auth.session import SessionManager
    from app.db import get_conn
    with get_conn() as conn:
        user = conn.execute("SELECT * FROM users WHERE login=?", (login,)).fetchone()
    return SessionManager.create_session(dict(user))


def test_router_user_keys_crud(client):
    # no session → falls back to the single admin user
    r = client.put("/api/agent/keys/openai",
                   json={"api_key": "sk-secret-1", "api_base": "", "default_model": "gpt-4o-mini"})
    assert r.status_code == 200
    assert r.json()["has_key"] is True

    keys = client.get("/api/agent/keys").json()["keys"]
    entry = [k for k in keys if k["provider"] == "openai"][0]
    assert entry["has_key"] is True
    assert entry["default_model"] == "gpt-4o-mini"
    assert "sk-secret-1" not in json.dumps(entry), "raw api_key must never be exposed"

    # empty key keeps the stored one
    r = client.put("/api/agent/keys/openai",
                   json={"api_key": "", "models": ["gpt-4o", "gpt-4o-custom"]})
    assert r.status_code == 200
    keys = client.get("/api/agent/keys").json()["keys"]
    entry = [k for k in keys if k["provider"] == "openai"][0]
    assert entry["has_key"] is True
    assert "gpt-4o-custom" in entry["models"]

    # providers endpoint merges saved models + has_key flag
    prov = client.get("/api/agent/providers").json()
    p = [x for x in prov["providers"] if x["id"] == "openai"][0]
    assert p["has_key"] is True
    assert "gpt-4o-custom" in p["models"]

    r = client.delete("/api/agent/keys/openai")
    assert r.status_code == 200
    keys = client.get("/api/agent/keys").json()["keys"]
    assert all(k["provider"] != "openai" for k in keys)


def test_router_user_keys_scoped_per_user(client):
    from app.db import get_conn

    with get_conn() as conn:
        conn.execute("INSERT INTO users (login, full_name, is_admin) VALUES ('alice', 'Alice', 0)")
        conn.commit()

    # admin saves a key
    client.put("/api/agent/keys/deepseek", json={"api_key": "sk-admin-deepseek"})
    assert client.get("/api/agent/keys").json()["keys"]

    # alice saves her own key and must NOT see the admin's
    client.cookies.set("flowdeck_session", _session_for_login("alice"))
    client.put("/api/agent/keys/deepseek", json={"api_key": "sk-alice-deepseek"})
    keys = client.get("/api/agent/keys").json()["keys"]
    entry = [k for k in keys if k["provider"] == "deepseek"][0]
    assert entry["has_key"] is True

    from app.db import get_conn as _gc
    with _gc() as conn:
        rows = conn.execute(
            "SELECT user_id, api_key FROM user_llm_keys WHERE provider='deepseek'"
        ).fetchall()
    by_user = {r["user_id"]: r["api_key"] for r in rows}
    assert len(by_user) == 2
    assert "sk-admin-deepseek" in by_user.values()
    assert "sk-alice-deepseek" in by_user.values()
    # masked API never leaks a raw key
    assert "sk-alice-deepseek" not in json.dumps(keys)


def test_router_user_keys_invalid_provider(client):
    assert client.put("/api/agent/keys/bogus", json={"api_key": "x"}).status_code == 400
    assert client.post("/api/agent/keys/bogus/models").status_code == 400
    assert client.delete("/api/agent/keys/bogus").status_code == 400


def test_router_fetch_models_uses_stored_key(client, monkeypatch):
    from app.routers import agent as agent_router

    calls = []

    async def fake_fetch(provider, *, api_key="", api_base="", timeout=20):
        calls.append({"provider": provider, "api_key": api_key, "api_base": api_base})
        return ["stub-a", "stub-b"]

    monkeypatch.setattr(agent_router, "fetch_provider_models", fake_fetch)

    # no stored key yet → empty key is sent
    r = client.post("/api/agent/keys/openai/models", json={})
    assert r.status_code == 200
    assert r.json()["ok"] is True
    assert r.json()["models"] == ["stub-a", "stub-b"]
    assert calls[-1]["api_key"] == ""

    # stored key is used as fallback when the body carries none
    client.put("/api/agent/keys/openai", json={"api_key": "sk-stored"})
    r = client.post("/api/agent/keys/openai/models", json={"api_base": "https://proxy.home/v1"})
    assert r.status_code == 200
    assert calls[-1]["api_key"] == "sk-stored"
    assert calls[-1]["api_base"] == "https://proxy.home/v1"

    # body-provided key wins over the stored one
    r = client.post("/api/agent/keys/openai/models", json={"api_key": "sk-typed"})
    assert calls[-1]["api_key"] == "sk-typed"


def test_router_fetch_models_error_surfaced(client, monkeypatch):
    from app.routers import agent as agent_router

    async def fake_fetch(provider, *, api_key="", api_base="", timeout=20):
        raise ConnectionError("injoignable")

    monkeypatch.setattr(agent_router, "fetch_provider_models", fake_fetch)
    r = client.post("/api/agent/keys/openai/models", json={"api_key": "sk-x"})
    assert r.status_code == 200
    body = r.json()
    assert body["ok"] is False
    assert "injoignable" in body["error"]


def test_router_patch_conversation(client):
    cid = _make_conversation(client)
    r = client.patch(f"/api/agent/conversations/{cid}",
                     json={"provider": "openai", "model": "gpt-4o-mini", "title": "Sprint préparé"})
    assert r.status_code == 200
    detail = client.get(f"/api/agent/conversations/{cid}").json()
    conv = detail["conversation"]
    assert conv["provider"] == "openai"
    assert conv["model"] == "gpt-4o-mini"
    assert conv["title"] == "Sprint préparé"


def test_router_run_uses_user_key(client, monkeypatch):
    from app.routers import agent as agent_router

    created = {}

    class FakeLLM(agent_router.LLMClient):
        def __init__(self, *args, **kwargs):
            created.update(kwargs)
            super().__init__(*args, **kwargs)

        async def complete(self, messages, *, model=None, tools=None, stream=False):
            return await self._mock_complete(messages, model, tools)

    monkeypatch.setattr(agent_router, "LLMClient", FakeLLM)

    # user key saved → run must create the LLM client with that key
    client.put("/api/agent/keys/openai", json={"api_key": "sk-user-secret", "api_base": "https://proxy.home/v1"})
    cid = _make_conversation(client)
    resp = client.post(f"/api/agent/conversations/{cid}/run",
                       json={"message": "crée une collection Bilan",
                             "provider": "openai", "model": "gpt-4o-mini"})
    assert resp.status_code == 200
    assert created.get("api_key") == "sk-user-secret"
    assert created.get("api_base") == "https://proxy.home/v1"

    # without a saved key the user key is not injected (global config/environment wins)
    assert client.delete("/api/agent/keys/openai").status_code == 200
    created.clear()
    cid2 = _make_conversation(client)
    resp = client.post(f"/api/agent/conversations/{cid2}/run",
                       json={"message": "crée une collection Bilan",
                             "provider": "openai", "model": "gpt-4o-mini"})
    assert resp.status_code == 200
    assert created.get("api_key", "") == ""
    return {}


def test_router_user_key_test_verifies(client, monkeypatch):
    """A successful user-key test flags the provider as functional (verified)."""
    from app.routers import agent as agent_router
    from app.services.llm_client import LLMResponse

    seen = {}

    class FakeLLM(agent_router.LLMClient):
        def __init__(self, *args, **kwargs):
            seen.update(kwargs)
            super().__init__(*args, **kwargs)

        async def ping(self, *, model=None):
            return LLMResponse(text="PONG", model=model or "gpt-4o")

    monkeypatch.setattr(agent_router, "LLMClient", FakeLLM)

    client.put("/api/agent/keys/openai", json={"api_key": "sk-user"})
    r = client.post("/api/agent/keys/openai/test", json={"model": "gpt-4o-mini"})
    assert r.status_code == 200
    body = r.json()
    assert body["ok"] is True
    assert body["verified"] is True
    assert seen.get("api_key") == "sk-user"

    # the stored key row now carries verified=True
    entry = [k for k in client.get("/api/agent/keys").json()["keys"]
             if k["provider"] == "openai"][0]
    assert entry["verified"] is True

    # /providers exposes openai as configured + verified + functional
    prov = client.get("/api/agent/providers").json()
    p = [x for x in prov["providers"] if x["id"] == "openai"][0]
    assert p["has_key"] is True
    assert p["configured"] is True
    assert p["verified"] is True
    assert p["functional"] is True


def test_router_user_key_test_failure(client, monkeypatch):
    """A failing test keeps the key but marks the provider non-functional."""
    from app.routers import agent as agent_router

    class FakeLLM(agent_router.LLMClient):
        def __init__(self, *args, **kwargs):
            super().__init__(*args, **kwargs)

        async def ping(self, *, model=None):
            raise PermissionError("Clé API manquante pour le provider « openai »")

    monkeypatch.setattr(agent_router, "LLMClient", FakeLLM)

    client.put("/api/agent/keys/openai", json={"api_key": "sk-user"})
    r = client.post("/api/agent/keys/openai/test", json={})
    assert r.status_code == 200
    body = r.json()
    assert body["ok"] is False
    assert body["verified"] is False

    keys = client.get("/api/agent/keys").json()["keys"]
    entry = [k for k in keys if k["provider"] == "openai"][0]
    assert entry["verified"] is False
    assert entry["last_error"]

    prov = client.get("/api/agent/providers").json()
    p = [x for x in prov["providers"] if x["id"] == "openai"][0]
    assert p["functional"] is False


def test_generate_endpoint_returns_text(client, monkeypatch):
    """/api/agent/generate performs headless text generation (no tools)."""
    from app.routers import agent as agent_router
    from app.services.llm_client import LLMResponse

    class FakeLLM(agent_router.LLMClient):
        def __init__(self, *args, **kwargs):
            super().__init__(*args, **kwargs)

        async def complete(self, messages, *, model=None, tools=None, stream=False):
            # The tool schema must never be offered for content generation.
            assert tools is None
            return LLMResponse(text="Voici le document traduit en anglais.", model="gpt-4o-mini")

    monkeypatch.setattr(agent_router, "LLMClient", FakeLLM)

    r = client.post("/api/agent/generate", json={
        "prompt": "Traduis ce document en anglais.",
        "context": "Titre du document : Doc\nContenu en français.",
        "model": "gpt-4o-mini",
    })
    assert r.status_code == 200
    body = r.json()
    assert body["ok"] is True
    assert "traduit en anglais" in body["text"]


def test_generate_endpoint_requires_prompt(client):
    r = client.post("/api/agent/generate", json={"prompt": "  "})
    assert r.status_code == 400


def test_mock_planner_ignores_context_keywords(client):
    """Document context words (ex: 'cherche', 'collection') must not trigger
    tool intents — the planner only inspects the user objective."""
    import asyncio

    from app.services.llm_client import LLMClient

    llm = LLMClient(provider="offline")
    resp = asyncio.run(llm.complete(
        [{"role": "system", "content": "x"},
         {"role": "user", "content": "Résume ce document.\n\n# Contexte\n"
                                     "Le rapport mentionne une recherche sur la collection Z."}],
        model="gpt-4o", tools=[{"type": "function"}],
    ))
    assert resp.tool_calls == []
    assert resp.text