flowdeck/tests/test_v754_memory.py

"""v7.54.0 — Mémoire de l'agent : toggle par conversation, injection au contexte."""

import asyncio
from pathlib import Path

from app.db import get_conn
from app.services import agent_memory

ROOT = Path(__file__).resolve().parent.parent
JS_PATH = ROOT / "static" / "js" / "agent_panel_2.js"
ENGINE = ROOT / "app" / "services" / "agent_engine.py"


def _conv(client) -> tuple[int, dict]:
    r = client.post("/api/agent/conversations", json={"title": "mémoire"})
    assert r.status_code == 200, r.text
    return r.json()["id"], r.json()


def test_migration_created_column_and_table(client):
    with get_conn() as conn:
        cols = {r[1] for r in conn.execute("PRAGMA table_info(agent_conversations)")}
        assert "memory_enabled" in cols
        tbl = conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table' AND name='agent_memory'"
        ).fetchone()
    assert tbl is not None


def test_create_conversation_defaults_to_memory_on(client):
    _, created = _conv(client)
    assert created["memory_enabled"] == 1  # settings.agent_memory_default


def test_patch_conversation_toggles_memory_persisted(client):
    cid, _ = _conv(client)
    assert client.patch(f"/api/agent/conversations/{cid}", json={"memory_enabled": 0}).status_code == 200
    rows = {c["id"]: c for c in client.get("/api/agent/conversations").json()["conversations"]}
    assert rows[cid]["memory_enabled"] == 0

    client.patch(f"/api/agent/conversations/{cid}", json={"memory_enabled": 1})
    rows = {c["id"]: c for c in client.get("/api/agent/conversations").json()["conversations"]}
    assert rows[cid]["memory_enabled"] == 1


def test_context_for_on_off_and_persistence(client):
    cid, _ = _conv(client)
    agent_memory.remember(cid, "Où en est le design system ?", "Les tokens sont dans design-tokens.css.")
    ctx = agent_memory.context_for(cid)
    assert "Mémoire de la conversation" in ctx
    assert "design-tokens.css" in ctx

    # OFF → plus rien n'est lu ni écrit
    client.patch(f"/api/agent/conversations/{cid}", json={"memory_enabled": 0})
    assert agent_memory.context_for(cid) == ""
    agent_memory.remember(cid, "Effacé ?", "Ne doit pas apparaître.")
    assert agent_memory.context_for(cid) == ""

    # ON à nouveau → la mémoire écrite avant le OFF est toujours là (persistée)
    client.patch(f"/api/agent/conversations/{cid}", json={"memory_enabled": 1})
    assert "design-tokens.css" in agent_memory.context_for(cid)
    assert "Ne doit pas apparaître" not in agent_memory.context_for(cid)


def test_memory_budget_and_note_lines_bounded(client):
    cid, _ = _conv(client)
    for i in range(30):
        agent_memory.remember(cid, f"question {i} " + "x" * 300, "réponse " + "y" * 900)
    ctx = agent_memory.context_for(cid)
    header = "## Mémoire de la conversation (échanges précédents)"
    assert ctx.startswith(header)
    assert len(ctx) <= agent_memory.INJECT_BUDGET + len(header) + 60
    with get_conn() as conn:
        content = conn.execute(
            "SELECT content FROM agent_memory WHERE conversation_id=?", (cid,)
        ).fetchone()["content"]
    assert content.count("\n") + 1 <= agent_memory.NOTE_LINES


async def _drain(engine, cid, objective):
    """Consomme le générateur SSE de engine.run()."""
    return [ev async for ev in engine.run(cid, objective)]


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


class _RecorderLLM:
    def __init__(self):
        self.last_user = ""

    async def complete(self, messages, *, model=None, tools=None, stream=False):
        self.last_user = next(m["content"] for m in messages if m["role"] == "user")
        return _Resp("C'est noté.")


def test_engine_run_injects_memory_when_on(client):
    """Le run réel pousse la mémoire dans le prompt utilisateur (toggle ON)."""
    from app.services.agent_engine import AgentEngine

    cid, _ = _conv(client)
    agent_memory.remember(cid, "Priorité du sprint", "Livrer le menu + en priorité.")
    fake = _RecorderLLM()
    engine = AgentEngine(1, llm=fake)
    asyncio.run(_drain(engine, cid, "Quelle est la priorité ?"))
    assert "Mémoire de la conversation" in fake.last_user
    assert "menu +" in fake.last_user


def test_engine_run_omits_memory_when_off(client):
    from app.services.agent_engine import AgentEngine

    cid, _ = _conv(client)
    agent_memory.remember(cid, "Secret", "ne pas injecter")
    client.patch(f"/api/agent/conversations/{cid}", json={"memory_enabled": 0})
    fake = _RecorderLLM()
    engine = AgentEngine(1, llm=fake)
    asyncio.run(_drain(engine, cid, "Bonjour"))
    assert "Mémoire de la conversation" not in fake.last_user
    assert "ne pas injecter" not in fake.last_user


def test_memory_menu_toggle_wired():
    src = JS_PATH.read_text(encoding="utf-8")
    root = src.split("var FD_PLUS_MENU = [", 1)[1].split("];", 1)[0]
    assert "action:'memory'" in root
    assert "bientôt (phase 4)" not in root
    # phases 1-5 livrées → seul Plugins reste désactivé
    assert root.count("disabled:true") == 1
    assert "toggleMemory()" in src
    assert "memory_enabled: next" in src
    engine = ENGINE.read_text(encoding="utf-8")
    assert "agent_memory.context_for(" in engine
    assert "agent_memory.remember(" in engine