From 330010cc660dfa1c1369f4bcb5ea1cdab6f541e9 Mon Sep 17 00:00:00 2001 From: Devika Verma Date: Wed, 29 Jul 2026 18:00:38 +0530 Subject: [PATCH] compaction: harden the smoke against per-turn event loops (OPE-27) The live smoke exposed a harness trap: driving each turn through its own asyncio.run() binds the engine asyncio primitives to the first loop, and every later stream silently takes the interrupted path - full provider replies persisted as empty assistant messages. The scripted smoke had the same latent artifact and did not assert reply content, so it stayed green. Now the whole scenario runs on ONE loop (like the real server) and every turn asserts a real reply. --- tests/test_compaction_smoke.py | 44 ++++++++++++++++++++++------------ 1 file changed, 29 insertions(+), 15 deletions(-) diff --git a/tests/test_compaction_smoke.py b/tests/test_compaction_smoke.py index 0fdd6001..dd5f2b48 100644 --- a/tests/test_compaction_smoke.py +++ b/tests/test_compaction_smoke.py @@ -56,22 +56,36 @@ def test_long_session_survives_repeated_compaction(tmp_path): mgr._prefs["compaction_cap_tokens"] = 3_000 sid = "smoke-long" - async def drive(engine, text): - async for _ in engine.run(text): - pass - - engine = mgr.get_engine(sid, agent="cowork", workspace=str(tmp_path)) boundaries = [] - for i in range(8): - asyncio.run(drive(engine, f"user step {i}: keep drafting the Q3 report")) - if engine.compaction_state is not None: - if not boundaries or engine.compaction_state.boundary_index != boundaries[-1]: - boundaries.append(engine.compaction_state.boundary_index) - mgr.save(sid, engine) - if i == 4: # mid-conversation restart: state must survive the rebuild - mgr._engines.pop(sid) - engine = mgr.get_engine(sid, agent="cowork", workspace=str(tmp_path)) - assert engine.compaction_state is not None + + # ONE event loop for the whole session, like the real server — the engine's asyncio + # primitives bind to the loop they first run on, so a per-turn asyncio.run() would + # silently drop every stream after the first (found the hard way in the live smoke). + async def scenario(): + engine = mgr.get_engine(sid, agent="cowork", workspace=str(tmp_path)) + for i in range(8): + async for _ in engine.run(f"user step {i}: keep drafting the Q3 report"): + pass + # Every turn must produce a real reply — an empty assistant message means + # the stream got dropped, not answered. + last = next( + m for m in reversed(engine.messages) if m.get("role") == "assistant" + ) + assert f"turn {i + 1}:" in str(last.get("content", "")) + if engine.compaction_state is not None: + if ( + not boundaries + or engine.compaction_state.boundary_index != boundaries[-1] + ): + boundaries.append(engine.compaction_state.boundary_index) + mgr.save(sid, engine) + if i == 4: # mid-conversation restart: state must survive the rebuild + mgr._engines.pop(sid) + engine = mgr.get_engine(sid, agent="cowork", workspace=str(tmp_path)) + assert engine.compaction_state is not None + return engine + + engine = asyncio.run(scenario()) # Repeated compaction actually happened, moving forward each time. assert len(boundaries) >= 2