From d2b4169ab4e6def1180374c21481f39a8eca9615 Mon Sep 17 00:00:00 2001 From: Harshit Wandhare Date: Mon, 24 Aug 2026 12:00:24 -0500 Subject: [PATCH] test(remoteagent): name the awaited event instead of panicking on a closed channel The A2A cleanup-propagation tests took the first status update with a bare receive: taskID := (<-statusUpdateEventChan).TaskInfo().TaskID The producing goroutine defers close on that channel and returns early when SendStreamingMessage fails, after recording the error with t.Errorf. The receive then yields the zero value of a2a.Event, which is a nil interface, and TaskInfo dereferences it. A streaming error therefore surfaced as a nil-pointer panic that killed the test binary and buried the error that had just been recorded. Add testutil.AwaitValue, the one-value counterpart to AwaitN. It treats a closed channel as a failure rather than a receive, since callers use the value, and it reports which value never arrived. Use it for the three ad-hoc receives in these two tests. That also brings the v2 variant in line with the compat one, which already guarded remoteTaskIDChan with its own select, and replaces that ad-hoc 1s bound with the shared deadline so a loaded machine cannot trip it. Also correct the call-site comments and document the deadline: AwaitN arms one timer for all n receives, so it is a budget for the whole call, not per wait as the comments claimed. This does not fix the flake in #1298. It makes a failure name the event it was waiting for instead of panicking or hanging anonymously, which is step 1 of what that issue asks for. Part of #1298. --- agent/remoteagent/a2a_agent_compat_test.go | 12 +++---- agent/remoteagent/v2/a2a_e2e_test.go | 7 ++-- internal/testutil/test_agent_runner.go | 39 ++++++++++++++++++++-- 3 files changed, 44 insertions(+), 14 deletions(-) diff --git a/agent/remoteagent/a2a_agent_compat_test.go b/agent/remoteagent/a2a_agent_compat_test.go index 5616430a1..4b2b4e805 100644 --- a/agent/remoteagent/a2a_agent_compat_test.go +++ b/agent/remoteagent/a2a_agent_compat_test.go @@ -768,7 +768,8 @@ func TestCompat_A2ACleanupPropagation(t *testing.T) { // Cancel only after the subagent's output reaches the client: before that the // parent doesn't know the subagent task ID, so cancellation can't propagate. - taskID := (<-statusUpdateEventChan).TaskInfo().TaskID + firstUpdate := testutil.AwaitValue(t, statusUpdateEventChan, "first status update") + taskID := firstUpdate.TaskInfo().TaskID testutil.AwaitN(t, remoteStreamingChan, 1, "remote subagent streaming") cancelResultChan := make(chan *legacyA2A.Task, 1) wg.Add(1) @@ -798,14 +799,9 @@ func TestCompat_A2ACleanupPropagation(t *testing.T) { // Check subagent task got cancelled when the parent task was cancelled. // Subagent cleanup fires twice: once for cancelation, once for execution. - // A generous per-wait deadline avoids flaking under CPU contention. + // A generous deadline avoids flaking under CPU contention. testutil.AwaitN(t, remoteCleanupCalledChan, 2, "remote cleanup") - var remoteTaskID legacyA2A.TaskID - select { - case remoteTaskID = <-remoteTaskIDChan: - case <-time.After(1 * time.Second): - t.Fatal("server B was never reached; remoteTaskIDChan is empty") - } + remoteTaskID := testutil.AwaitValue(t, remoteTaskIDChan, "server B remote task ID") testutil.AwaitN(t, executorCleanupCalledChan, 2, "executor cleanup") remoteClient := newLegacyA2AClient(t, serverB) diff --git a/agent/remoteagent/v2/a2a_e2e_test.go b/agent/remoteagent/v2/a2a_e2e_test.go index a8fd9b6e7..1b829798d 100644 --- a/agent/remoteagent/v2/a2a_e2e_test.go +++ b/agent/remoteagent/v2/a2a_e2e_test.go @@ -550,7 +550,8 @@ func TestA2ACleanupPropagation(t *testing.T) { // Cancel only after the subagent's output reaches the client: before that the // parent doesn't know the subagent task ID, so cancellation can't propagate. - taskID := (<-statusUpdateEventChan).TaskInfo().TaskID + firstUpdate := testutil.AwaitValue(t, statusUpdateEventChan, "first status update") + taskID := firstUpdate.TaskInfo().TaskID testutil.AwaitN(t, remoteStreamingChan, 1, "remote subagent streaming") cancelResultChan := make(chan *a2a.Task, 1) wg.Add(1) @@ -579,9 +580,9 @@ func TestA2ACleanupPropagation(t *testing.T) { } // Subagent cleanup fires twice: once for cancelation, once for execution. - // A generous per-wait deadline avoids flaking under CPU contention. + // A generous deadline avoids flaking under CPU contention. testutil.AwaitN(t, remoteCleanupCalledChan, 2, "remote cleanup") - remoteTaskID := <-remoteTaskIDChan + remoteTaskID := testutil.AwaitValue(t, remoteTaskIDChan, "server B remote task ID") testutil.AwaitN(t, executorCleanupCalledChan, 2, "executor cleanup") remoteClient := newA2AClient(t, serverB) diff --git a/internal/testutil/test_agent_runner.go b/internal/testutil/test_agent_runner.go index 4e85015e9..44a89d44e 100644 --- a/internal/testutil/test_agent_runner.go +++ b/internal/testutil/test_agent_runner.go @@ -260,20 +260,53 @@ func CollectTextParts(stream iter.Seq2[*session.Event, error]) ([]string, error) return texts, nil } +// awaitDeadline bounds a single Await call. It is generous so that a loaded CI +// machine does not fail a test that is merely slow. Note that AwaitN arms one +// timer for all n receives, so this is the budget for the whole call rather +// than for each value. +const awaitDeadline = 30 * time.Second + // AwaitN receives n values from ch, or fails the test via t.Fatalf if they do // not all arrive within a generous, contention-tolerant deadline. A closed // channel counts as a receive, so AwaitN also joins a goroutine that closed ch // without sending. func AwaitN[T any](t *testing.T, ch <-chan T, n int, what string) { t.Helper() - const deadline = 30 * time.Second - timer := time.NewTimer(deadline) + timer := time.NewTimer(awaitDeadline) defer timer.Stop() for i := range n { select { case <-ch: case <-timer.C: - t.Fatalf("%s: got %d of %d within %v", what, i, n, deadline) + t.Fatalf("%s: got %d of %d within %v", what, i, n, awaitDeadline) + } + } +} + +// AwaitValue receives one value from ch and returns it, failing the test via +// t.Fatalf if none arrives within the same deadline AwaitN uses. +// +// Unlike AwaitN, a closed channel is a failure rather than a receive. Callers +// use the value, and a producer that closed ch without sending yields only the +// zero value, which for a channel of interface type is nil and panics on the +// first method call. Reporting which value never arrived is more useful than +// that panic, especially when the producer has already recorded the underlying +// error with t.Errorf: the panic would bury it. +func AwaitValue[T any](t *testing.T, ch <-chan T, what string) T { + t.Helper() + timer := time.NewTimer(awaitDeadline) + defer timer.Stop() + select { + case v, ok := <-ch: + if !ok { + var zero T + t.Fatalf("%s: channel closed before a value arrived", what) + return zero } + return v + case <-timer.C: + var zero T + t.Fatalf("%s: no value within %v", what, awaitDeadline) + return zero } }