Sprint 1: event runtime tests (T1.1-T1.7)

tests/test_sprint1_events.py

@@ -0,0 +1,198 @@
+#!/usr/bin/env python3
+"""
+Sprint 1 Tests — Canonical Event Runtime.
+
+T1.1  Three agents stream in parallel; no lane contamination
+T1.2  Agent B fails; Agent A and C continue
+T1.3  Events in each lane are strictly seq-monotonic
+T1.4  Stream replay reproduces same visible event transcript
+T1.5  reasoning.summary allowed; raw hidden_chain_of_thought rejected
+T1.6  Backpressure does not drop terminal events
+T1.7  Legacy StreamEvent consumers still work
+"""
+import sys
+import os
+import asyncio
+import time
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+PASS = 0
+FAIL = 0
+
+def check(name, condition, detail=""):
+    global PASS, FAIL
+    if condition:
+        PASS += 1
+        print(f"  ✓ {name}")
+    else:
+        FAIL += 1
+        print(f"  ✗ {name}" + (f": {detail}" if detail else ""))
+
+
+from purpose_agent.runtime.events import PAEvent, EventKind, Visibility, create_event
+from purpose_agent.runtime.event_bus import EventBus, parallel_merge
+
+
+# ═══ T1.1: Three parallel lanes, no contamination ═══
+print("T1.1: Parallel lane isolation")
+bus = EventBus(max_queue_size=100)
+lanes = {"lane_a": [], "lane_b": [], "lane_c": []}
+
+for lane_id in lanes:
+    for i in range(5):
+        event = create_event(
+            run_id="run1", kind=EventKind.AGENT_PROGRESS,
+            lane_id=lane_id, seq=i+1,
+            message=f"{lane_id} step {i+1}",
+        )
+        bus.emit(event)
+
+# Replay per lane
+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}")
+            break
+    else:
+        check(f"Lane {lane_id} isolated ({len(lane_events)} events)", len(lane_events) == 5)
+
+
+# ═══ 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))
+
+a_events = bus2.replay(run_id="r2", lane_id="a")
+b_events = bus2.replay(run_id="r2", lane_id="b")
+c_events = bus2.replay(run_id="r2", lane_id="c")
+check("Lane A completed", any(e.kind == EventKind.AGENT_FINISHED for e in a_events))
+check("Lane B errored", any(e.kind == EventKind.AGENT_ERROR for e in b_events))
+check("Lane C completed despite B failure", any(e.kind == EventKind.AGENT_FINISHED for e in c_events))
+
+
+# ═══ T1.3: Seq-monotonic per lane ═══
+print("\nT1.3: Sequence monotonicity")
+bus3 = EventBus()
+for i in range(10):
+    bus3.emit(create_event("r3", EventKind.AGENT_PROGRESS, lane_id="main", seq=i+1))
+
+events = bus3.replay(run_id="r3")
+seqs = [e.seq for e in events]
+check("Seq monotonic", seqs == sorted(seqs), f"seqs={seqs}")
+check("No gaps", seqs == list(range(1, 11)))
+
+
+# ═══ T1.4: Replay reproduces transcript ═══
+print("\nT1.4: Replay fidelity")
+bus4 = EventBus()
+original = []
+for i in range(5):
+    e = create_event("r4", EventKind.TEXT_DELTA, seq=i+1, text=f"word_{i}")
+    bus4.emit(e)
+    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)
+
+
+# ═══ T1.5: reasoning.summary OK; hidden_chain_of_thought rejected ═══
+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>")
+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)")
+
+
+# ═══ T1.6: Backpressure doesn't drop terminal events ═══
+print("\nT1.6: Terminal event delivery guarantee")
+
+async def test_backpressure():
+    bus6 = EventBus(max_queue_size=3)  # Very small queue
+
+    received = []
+
+    async def consumer():
+        async for event in bus6.subscribe():
+            received.append(event)
+            await asyncio.sleep(0.01)  # Slow consumer
+
+    # Start consumer
+    task = asyncio.create_task(consumer())
+    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)
+    bus6.close()
+    task.cancel()
+    try:
+        await task
+    except asyncio.CancelledError:
+        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
+
+has_terminal = asyncio.run(test_backpressure())
+check("Terminal event delivered under backpressure", has_terminal)
+
+
+# ═══ T1.7: Legacy StreamEvent compat ═══
+print("\nT1.7: Legacy StreamEvent compatibility")
+from purpose_agent.streaming_v3 import pa_event_to_stream_event
+from purpose_agent.streaming import StreamEvent
+
+event = create_event("r7", EventKind.RUN_STARTED, seq=1, purpose="test")
+legacy = pa_event_to_stream_event(event)
+check("PAEvent → StreamEvent", legacy is not None)
+check("StreamEvent has event_type", hasattr(legacy, "event_type"))
+check("StreamEvent type is task_start", legacy.event_type == "task_start")
+
+
+# ═══ parallel_merge ═══
+print("\nT1.x: parallel_merge determinism")
+lane_a = [create_event("r8", EventKind.AGENT_PROGRESS, lane_id="a", seq=i, ts=1000+i) for i in range(3)]
+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)))
+
+
+# ═══ REPORT ═══
+print(f"\n{'='*50}")
+print(f"  Sprint 1 Tests: {PASS} pass, {FAIL} fail")
+print(f"  {'ALL PASS ✓' if FAIL == 0 else f'{FAIL} FAILURES'}")
+print(f"{'='*50}")
+sys.exit(0 if FAIL == 0 else 1)