It was 2 a.m. when the shrieking of PagerDuty yanked me out of sleep. The alert read: “Users report the AI assistant suddenly lost memory – conversation context gone.” Groggy, I pulled up monitoring and saw that one-third of the memory write operations in production were timing out, even though the test environment had been all green just ten minutes earlier. Damn. Another async timing issue that end-to-end tests never caught.
Breaking Down the Problem: Why AI Memory Tests Slip Through So Easily
Our AI app is an LLM-based customer service bot. It writes conversation history and user preferences to Redis as memory storage. For example, if a user says “My name is Zhang San, I like dark mode,” every subsequent interaction must recall that fact. The most feared scenario is a “successful write on the surface, but no actual persistence” – the API returns 200, but Redis is still flushing or a background async task hasn’t finished.
How are typical tests done? Unit tests mock Redis and can’t capture real latency. Integration tests resort to time.sleep(2) to wait blindly – slow and unreliable. Manual clicking looks intuitive but can’t possibly cover all timing combinations. An even more subtle issue: Playwright’s waitForLoadState('networkidle') only cares about the browser’s network stack being idle; it’s completely blind to backend async tasks. That leaves a huge testing blind spot: if the backend write lags by just 100ms, the assertion will fire before the data truly takes effect and you’ll get a false failure.
Designing a Solution: Teaching Playwright to Wait for Async Persistence
The core of the problem boils down to one thing: how to precisely align the timing of browser-side operations with the completion of backend async storage.
I considered three approaches:
- Expose a health-check endpoint on the server and have the test poll until the memory key exists – feasible but too intrusive; you can’t open that backdoor in production.
- Use Playwright to directly read Redis – that’s crossing layers, and the test machine might not even have access.
- Intercept memory-related API calls from the frontend, wait for the response to complete, and supplement with application-level proof that the memory has taken effect – for instance, wait until the AI’s next reply actually contains the previously stored information.
I chose the third. The reason is simple: an end-to-end test must verify from the user’s perspective that the AI actually “remembers.” The architecture: pytest + Playwright driving the browser directly, a FastAPI app serving the chat interface, and Redis for memory storage. In tests, we use page.route() to intercept network requests, knowing precisely when the memory-save API returns, then assert on the AI’s next response to prove the memory is active. No sleep, no polling – purely event-driven.
Core Implementation
1. Launching the browser and intercepting memory save requests
This code sets up a network intercept test environment that captures all requests to /api/memory and returns a Future, allowing subsequent assertions to await the moment the write succeeds (at least the API returns success).
import pytest
import asyncio
from playwright.async_api import async_playwright, Page, Route
@pytest.fixture
async def page_with_memory_tracker():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
context = await browser.new_context()
page = await context.new_page()
# 存储每个记忆保存操作对应的Future,用于精确等待
memory_write_events = []
async def track_memory_request(route: Route):
req = route.request
if '/api/memory' in req.url and req.method == 'POST':
# 为这个写入创建一个Future,响应完成后set
future = asyncio.get_event_loop().create_future()
memory_write_events.append(future)
response = await route.fetch()
await route.fulfill(response=response)
future.set_result(True) # 只有API响应完成才通知
else:
await route.continue_()
await page.route('**/*', track_memory_request)
yield page, memory_write_events
await browser.close()
2. Simulating multi-turn conversation and verifying memory
This test walks through the complete chain: set memory → confirm write → new conversation references memory, without any hard waits.
@pytest.mark.asyncio
async def test_ai_remembers_user_name(page_with_memory_tracker):
page, memory_events = page_with_memory_tracker
await page.goto('https://chat.example.com')
# 第一轮:告诉AI我的名字
await page.fill('#chat-input', '我叫张三,喜欢极简模式')
await page.click('#send-btn')
# 等待记忆保存API完成(避免下轮对话时记忆尚未生效)
if memory_events:
await memory_events[-1] # 阻塞直到最近的记忆写入future完成
# 第二轮:问一个需要记忆才能回答的问题
await page.fill('#chat-input', '我刚才说我叫什么?')
await page.click('#send-btn')
# 等待AI回复中出现"张三",超时10秒,足够从Redis加载记忆
await page.wait_for_function(
"""() => {
const msgs = document.querySelectorAll('.assistant-message');
const last = msgs[msgs.length - 1];
return last && last.textContent.includes('张三');
}""",
timeout=10000
)
Reflecting on that experience, if I had this mechanism in place, the midnight PagerDuty would never have gone off. A truly reliable end-to-end test for AI applications can’t just check “what the page shows right now” – it must ensure the underlying state machine has completed its transitions. Whether it’s Redis writes, Kafka offset commits, or sidecar index syncing, as long as the browser can’t see it, you have to proactively capture it. Use API interception + application-level state verification as a double lock. Only when asynchronous state changes can be precisely aligned can tests truly serve as a safety net.
Top comments (1)
The distinction between
networkidleand a completed Redis-backed state transition is the important one here: the browser can be quiet while the memory write is still in flight. Pairing aFuturearound/api/memorywith a later response containing "" gives the test both a transport boundary and a user-visible outcome, although the latter still needs a deterministic test fixture so the model cannot simply hallucinate the answer. For a founder or engineering lead, this is also a useful contract-design lesson: expose observability at async boundaries without exposing production-only backdoors, or timing bugs will keep appearing as.