Use active event-wait pattern in todos_changed E2E tests

Replaces the post-hoc events.filter() pattern with the existing getNextEventOfType helper (Node/Python) and a CompletableFuture event subscription (Java). This matches the pattern used by Go/.NET/Rust and removes a race window where the event could arrive after sendAndWait resolves but before the assertion is checked.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: SteveSandersonMS

java/src/test/java/com/github/copilot/SessionTodosChangedTest.java

@@ -6,16 +6,13 @@
 
 import static org.junit.jupiter.api.Assertions.*;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
-import com.github.copilot.generated.SessionEvent;
 import com.github.copilot.generated.SessionTodosChangedEvent;
 import com.github.copilot.generated.rpc.PlanSqlTodoDependency;
 import com.github.copilot.rpc.MessageOptions;
@@ -46,8 +43,12 @@ void firesSessionTodosChangedAndExposesRowsAndDependencies() throws Exception {
             CopilotSession session = client
                     .createSession(new SessionConfig().setOnPermissionRequest(PermissionHandler.APPROVE_ALL)).get();
 
-            List<SessionEvent> events = Collections.synchronizedList(new ArrayList<>());
-            session.on(events::add);
+            CompletableFuture<SessionTodosChangedEvent> todosChanged = new CompletableFuture<>();
+            session.on(event -> {
+                if (event instanceof SessionTodosChangedEvent todosEvent && !todosChanged.isDone()) {
+                    todosChanged.complete(todosEvent);
+                }
+            });
 
             session.sendAndWait(new MessageOptions()
                     .setPrompt("Use the sql tool to execute exactly these statements, in order, with no extra rows:\n"
@@ -57,14 +58,14 @@ void firesSessionTodosChangedAndExposesRowsAndDependencies() throws Exception {
                             + "Then stop. Do not insert any other rows or create any other tables."))
                     .get(120, TimeUnit.SECONDS);
 
-            assertTrue(events.stream().anyMatch(SessionTodosChangedEvent.class::isInstance),
+            assertNotNull(todosChanged.get(15, TimeUnit.SECONDS),
                     "Should have received at least one session.todos_changed event");
 
             var result = session.getRpc().plan.readSqlTodosWithDependencies().get(15, TimeUnit.SECONDS);
             assertEquals(2, result.rows().size());
             var ids = result.rows().stream().map(row -> row.id()).filter(id -> id != null).sorted().toList();
 
-            assertEquals(List.of("alpha", "beta"), ids);
+            assertEquals(java.util.List.of("alpha", "beta"), ids);
             assertTrue(result.dependencies().stream().anyMatch(SessionTodosChangedTest::isBetaDependsOnAlpha),
                     "Should contain beta -> alpha dependency");
 

nodejs/test/e2e/session_todos_changed.e2e.test.ts

@@ -6,8 +6,9 @@ import fs, { realpathSync } from "node:fs";
 import os from "node:os";
 import { join } from "node:path";
 import { describe, expect, it } from "vitest";
-import { approveAll, type SessionEvent } from "../../src/index.js";
+import { approveAll } from "../../src/index.js";
 import { createSdkTestContext } from "./harness/sdkTestContext.js";
+import { getNextEventOfType } from "./harness/sdkTestHelper.js";
 
 /**
  * E2E coverage for the runtime's `session.todos_changed` event and
@@ -28,10 +29,7 @@ describe("Todos changed event + readSqlTodosWithDependencies", async () => {
         async () => {
             const session = await client.createSession({ onPermissionRequest: approveAll });
 
-            const events: SessionEvent[] = [];
-            session.on((event) => {
-                events.push(event);
-            });
+            const todosChanged = getNextEventOfType(session, "session.todos_changed");
 
             await session.sendAndWait({
                 prompt:
@@ -42,8 +40,7 @@ describe("Todos changed event + readSqlTodosWithDependencies", async () => {
                     "Then stop. Do not insert any other rows or create any other tables.",
             });
 
-            const todosEvents = events.filter((e) => e.type === "session.todos_changed");
-            expect(todosEvents.length).toBeGreaterThanOrEqual(1);
+            await todosChanged;
 
             const result = await session.rpc.plan.readSqlTodosWithDependencies();
             const ids = result.rows

python/e2e/test_session_todos_changed_e2e.py

@@ -2,11 +2,13 @@
 
 from __future__ import annotations
 
+import asyncio
+
 import pytest
 
 from copilot.session import PermissionHandler
 
-from .testharness import E2ETestContext
+from .testharness import E2ETestContext, get_next_event_of_type
 
 pytestmark = pytest.mark.asyncio(loop_scope="module")
 
@@ -18,28 +20,18 @@
 Then stop. Do not insert any other rows or create any other tables."""
 
 
-def _event_type_value(event) -> str:
-    return getattr(event.type, "value", event.type)
-
-
 class TestSessionTodosChanged:
     async def test_fires_session_todos_changed_and_exposes_rows_and_dependencies(
         self, ctx: E2ETestContext
     ):
         async with await ctx.client.create_session(
             on_permission_request=PermissionHandler.approve_all,
         ) as session:
-            events = []
-            unsubscribe = session.on(events.append)
-            try:
-                await session.send_and_wait(PROMPT, timeout=120.0)
-            finally:
-                unsubscribe()
-
-            todos_events = [
-                event for event in events if _event_type_value(event) == "session.todos_changed"
-            ]
-            assert len(todos_events) >= 1
+            todos_changed = asyncio.create_task(
+                get_next_event_of_type(session, "session.todos_changed", timeout=120.0)
+            )
+            await session.send_and_wait(PROMPT, timeout=120.0)
+            await todos_changed
 
             result = await session.rpc.plan.read_sql_todos_with_dependencies()
             ids = sorted(row.id for row in result.rows if row.id)