e2e/test_memory_source_of_truth_regression.py

from __future__ import annotations

import inspect
import os

from playwright.sync_api import expect, sync_playwright


BASE_URL = os.getenv("MATCHDAY_E2E_BASE_URL", "http://127.0.0.1:3100")
PROOF_TIMEOUT_MS = int(os.getenv("MATCHDAY_E2E_PROOF_TIMEOUT_MS", "30000"))
E2E_BROWSER = os.getenv("MATCHDAY_E2E_BROWSER", "chromium").strip().lower()
SUPPORTED_E2E_BROWSERS = {"chromium", "firefox", "webkit"}

if E2E_BROWSER not in SUPPORTED_E2E_BROWSERS:
    raise RuntimeError(
        "MATCHDAY_E2E_BROWSER must be one of chromium, firefox, or webkit"
    )


def launch_browser(playwright):
    return getattr(playwright, E2E_BROWSER).launch(headless=True)


def test_manual_profile_memory_is_recalled_in_chat_across_browser_reload() -> None:
    # Regression: the profile UI, chat, and decision trace must share one
    # durable-memory source of truth across reloads.
    with sync_playwright() as playwright:
        browser = launch_browser(playwright)
        context = browser.new_context(viewport={"width": 1280, "height": 900})
        page = context.new_page()

        page.goto(BASE_URL, wait_until="domcontentloaded")
        add = page.get_by_role("button", name="Add preference")
        expect(add).to_be_enabled(timeout=PROOF_TIMEOUT_MS)
        add.click()
        page.get_by_label("Preference type").select_option("home_airport")
        page.get_by_label("Preference value").fill("YUL")
        page.get_by_role("button", name="Save preference").click()
        home = page.locator('[data-memory-key="home_airport"]')
        expect(home.locator("strong")).to_have_text("YUL", timeout=PROOF_TIMEOUT_MS)

        page.reload(wait_until="domcontentloaded")
        expect(page.locator('[data-memory-key="home_airport"] strong')).to_have_text(
            "YUL",
            timeout=PROOF_TIMEOUT_MS,
        )
        page.get_by_label("Trip request").fill("what are my preferences?")
        page.get_by_role("button", name="Plan this trip").click()

        expect(page.get_by_text("Home airport: YUL", exact=False).first).to_be_visible(
            timeout=PROOF_TIMEOUT_MS,
        )
        page.get_by_text("Full run log", exact=True).click()
        expect(
            page.get_by_text(
                "Recalled 1 relevant confirmed preferences.",
                exact=True,
            )
        ).to_be_visible(
            timeout=PROOF_TIMEOUT_MS,
        )
        expect(page.get_by_text("FAILED", exact=True)).to_have_count(0)

        context.close()
        browser.close()


def test_saved_home_airport_conflict_is_clarified_then_replans_in_place() -> None:
    with sync_playwright() as playwright:
        browser = launch_browser(playwright)
        context = browser.new_context(viewport={"width": 1280, "height": 900})
        page = context.new_page()

        page.goto(BASE_URL, wait_until="domcontentloaded")
        page.get_by_role("button", name="Add preference").click()
        page.get_by_label("Preference type").select_option("home_airport")
        page.get_by_label("Preference value").fill("YUL")
        page.get_by_role("button", name="Save preference").click()
        expect(page.locator('[data-memory-key="home_airport"] strong')).to_have_text(
            "YUL",
            timeout=PROOF_TIMEOUT_MS,
        )

        composer = page.get_by_label("Trip request")
        composer.fill("Plan the 2026 World Cup Final from YVR under C$2,200.")
        page.get_by_role("button", name="Plan this trip").click()
        expect(
            page.locator('[data-proof-step="verified"][data-proof-state="done"]')
        ).to_be_visible(timeout=PROOF_TIMEOUT_MS)
        parent_run_id = page.evaluate(
            "localStorage.getItem('matchday-v3-active-run')"
        )
        assert parent_run_id

        composer.fill("book me another package basedon my new preference")
        page.get_by_role("button", name="Send follow-up").click()
        page.wait_for_function(
            "parent => localStorage.getItem('matchday-v3-active-run') !== parent",
            arg=parent_run_id,
            timeout=PROOF_TIMEOUT_MS,
        )
        approval = page.locator(".approval-panel")
        expect(approval).to_be_visible(timeout=PROOF_TIMEOUT_MS)
        expect(approval).to_contain_text("saved home airport is YUL")
        expect(approval).to_contain_text("currently starts at YVR")

        held = page.evaluate(
            """async () => {
              const id = localStorage.getItem('matchday-v3-active-run');
              return await fetch(`/api/backend/v1/runs/${id}`, {cache: 'no-store'}).then(r => r.json());
            }"""
        )
        assert held["status"] == "awaiting_approval"
        recalled = next(
            event for event in held["events"] if event["type"] == "memory.recalled"
        )
        assert "home_airport" in {
            memory["key"] for memory in recalled["payload"]["memories"]
        }

        expect(approval).to_contain_text("Which origin should I use")
        approval.get_by_label("Clarification answer").fill("YUL")
        approval.get_by_role("button", name="Continue this run").click()
        expect(
            page.locator('[data-proof-step="verified"][data-proof-state="done"]')
        ).to_be_visible(timeout=PROOF_TIMEOUT_MS)
        expect(page.locator(".trip-hero")).to_contain_text(
            "YUL",
            timeout=PROOF_TIMEOUT_MS,
        )
        completed = page.evaluate(
            """async () => {
              const id = localStorage.getItem('matchday-v3-active-run');
              return await fetch(`/api/backend/v1/runs/${id}`, {cache: 'no-store'}).then(r => r.json());
            }"""
        )
        assert completed["id"] == held["id"]
        assert completed["status"] == "completed"
        assert completed["result"]["itinerary"]["intent"]["origin_airport"] == "YUL"

        context.close()
        browser.close()


def test_memory_journeys_do_not_bypass_the_configured_browser_launcher() -> None:
    for journey in (
        test_manual_profile_memory_is_recalled_in_chat_across_browser_reload,
        test_saved_home_airport_conflict_is_clarified_then_replans_in_place,
    ):
        source = inspect.getsource(journey)
        assert "launch_browser(playwright)" in source
        assert "playwright.chromium" not in source