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