Fix Issue 2: Make backpressure test T1.6 more robust with consumer_started sync

tests/test_sprint1_events.py

@@ -52,7 +52,6 @@ for lane_id in lanes:
 for lane_id in lanes:
     lane_events = bus.replay(run_id="run1", lane_id=lane_id)
     lanes[lane_id] = lane_events
-    # Check no contamination
     for e in lane_events:
         if e.lane_id != lane_id:
             check(f"Lane {lane_id} contamination", False, f"found {e.lane_id}")
@@ -64,13 +63,10 @@ for lane_id in lanes:
 # ═══ T1.2: Agent B fails; A and C continue ═══
 print("\nT1.2: Fault isolation")
 bus2 = EventBus()
-# Lane A: normal
 bus2.emit(create_event("r2", EventKind.AGENT_STARTED, lane_id="a", seq=1))
 bus2.emit(create_event("r2", EventKind.AGENT_FINISHED, lane_id="a", seq=2))
-# Lane B: error
 bus2.emit(create_event("r2", EventKind.AGENT_STARTED, lane_id="b", seq=1))
 bus2.emit(create_event("r2", EventKind.AGENT_ERROR, lane_id="b", seq=2, error="crashed"))
-# Lane C: normal (continues despite B's failure)
 bus2.emit(create_event("r2", EventKind.AGENT_STARTED, lane_id="c", seq=1))
 bus2.emit(create_event("r2", EventKind.AGENT_FINISHED, lane_id="c", seq=2))
 
@@ -104,7 +100,6 @@ for i in range(5):
     original.append(e.to_dict())
 
 replayed = [e.to_dict() for e in bus4.replay(run_id="r4")]
-# Compare payloads (span_id will differ but payload should match)
 orig_payloads = [d["payload"] for d in original]
 replay_payloads = [d["payload"] for d in replayed]
 check("Replay matches original", orig_payloads == replay_payloads)
@@ -114,15 +109,13 @@ check("Replay matches original", orig_payloads == replay_payloads)
 print("\nT1.5: Safety — no raw chain-of-thought")
 bus5 = EventBus()
 
-# Safe: reasoning summary
 safe_event = create_event("r5", EventKind.REASONING_SUMMARY, seq=1,
     summary="I should check edge cases first")
 bus5.emit(safe_event)
 check("reasoning.summary accepted", bus5.history_size == 1)
 
-# Unsafe: raw chain-of-thought (should be REJECTED)
 unsafe_event = create_event("r5", EventKind.REASONING_SUMMARY, seq=2,
-    hidden_chain_of_thought="<think>Let me think step by step about how to hack the system...</think>")
+    hidden_chain_of_thought="secret reasoning here")
 bus5.emit(unsafe_event)
 check("hidden_chain_of_thought rejected", bus5.history_size == 1,
       f"history has {bus5.history_size} events (should still be 1)")
@@ -132,35 +125,36 @@ check("hidden_chain_of_thought rejected", bus5.history_size == 1,
 print("\nT1.6: Terminal event delivery guarantee")
 
 async def test_backpressure():
-    bus6 = EventBus(max_queue_size=3)  # Very small queue
-
+    bus6 = EventBus(max_queue_size=3)
     received = []
+    consumer_started = asyncio.Event()
 
     async def consumer():
-        async for event in bus6.subscribe():
-            received.append(event)
-            await asyncio.sleep(0.01)  # Slow consumer
+        consumer_started.set()
+        try:
+            async for event in bus6.subscribe():
+                received.append(event)
+                await asyncio.sleep(0.01)
+        except asyncio.CancelledError:
+            pass
 
-    # Start consumer
     task = asyncio.create_task(consumer())
+    await consumer_started.wait()
     await asyncio.sleep(0.05)
 
-    # Flood with non-terminal events
     for i in range(20):
         bus6.emit(create_event("r6", EventKind.TEXT_DELTA, seq=i, text=f"w{i}"))
 
-    # Send terminal event
     bus6.emit(create_event("r6", EventKind.RUN_FINISHED, seq=99, result="done"))
 
-    await asyncio.sleep(0.5)
+    await asyncio.sleep(1.0)
     bus6.close()
     task.cancel()
     try:
-        await task
-    except asyncio.CancelledError:
+        await asyncio.wait_for(task, timeout=2.0)
+    except (asyncio.CancelledError, asyncio.TimeoutError):
         pass
 
-    # Terminal event MUST be in received (even though queue was full)
     has_terminal = any(e.kind == EventKind.RUN_FINISHED for e in received)
     return has_terminal
 
@@ -186,7 +180,6 @@ lane_a = [create_event("r8", EventKind.AGENT_PROGRESS, lane_id="a", seq=i, ts=10
 lane_b = [create_event("r8", EventKind.AGENT_PROGRESS, lane_id="b", seq=i, ts=1000+i+0.5) for i in range(3)]
 merged = parallel_merge({"a": lane_a, "b": lane_b})
 check("Merged has all events", len(merged) == 6)
-# Check interleaving: a0, b0, a1, b1, a2, b2 (by timestamp)
 check("Merged is timestamp-sorted", all(merged[i].ts <= merged[i+1].ts for i in range(5)))