From 6d9dd4f4a2d72db59ad453c9f12ea0caf0d38f11 Mon Sep 17 00:00:00 2001 From: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> Date: Mon, 9 Nov 2026 13:21:19 -0700 Subject: [PATCH 1/5] :white_check_mark: Update flaky tests Signed-off-by: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> --- cmd/agentop/tui/sessions_title_test.go | 15 +++-- core/listener/forwardproxy/server_test.go | 20 ++++-- .../forwardproxy/websocket_review_test.go | 18 ++---- .../script/internal/worker/worker_test.go | 36 +++++++---- core/reloader/reloader_test.go | 64 +++++++++---------- core/session/archive/archive_test.go | 22 +++---- core/sessionapi/usage_test.go | 7 +- 7 files changed, 99 insertions(+), 83 deletions(-) diff --git a/cmd/agentop/tui/sessions_title_test.go b/cmd/agentop/tui/sessions_title_test.go index 9e674788c..b4778b200 100644 --- a/cmd/agentop/tui/sessions_title_test.go +++ b/cmd/agentop/tui/sessions_title_test.go @@ -1218,18 +1218,23 @@ func TestLooksLikePath_AnyUnicodeSpaceSeparates(t *testing.T) { // // Asserted as a RATIO rather than a wall-clock bound, so it says what it means on a loaded CI machine: // quadratic growth shows up as ~4x per doubling, linear as ~2x, and the ceiling sits between them. -// Both ends are timed inside one test so the comparison is against the same machine at the same moment. +// Both ends are timed inside one test so the comparison is against the same machine at the same moment, +// and each is the best of three runs, so one preempted run does not set the ratio. func TestTrunc_ScalesLinearly(t *testing.T) { const budget = 40 measure := func(f func(string, int) string, runes int) time.Duration { s := "/" + strings.Repeat("a", runes-1) // Warm, so the first-call cost of anything lazy is not charged to the small input. f(s, budget) - start := time.Now() - for i := 0; i < 20; i++ { - f(s, budget) + best := time.Duration(1<<63 - 1) + for range 3 { + start := time.Now() + for i := 0; i < 20; i++ { + f(s, budget) + } + best = min(best, time.Since(start)) } - return time.Since(start) + return best } for _, tc := range []struct { name string diff --git a/core/listener/forwardproxy/server_test.go b/core/listener/forwardproxy/server_test.go index fc445fba8..9b1968d22 100644 --- a/core/listener/forwardproxy/server_test.go +++ b/core/listener/forwardproxy/server_test.go @@ -925,12 +925,20 @@ func TestForwardProxy_ResponseEventOmitsTtfbOnFailedExchange(t *testing.T) { store := session.New(5*time.Minute, 100, 0) defer store.Close() - // A server that is already closed: its port answers with a refusal, which - // is the fast half of the failure family. The 502 the proxy synthesizes is - // the same row shape a timed-out (black-holed) dial produces. - dead := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {})) - deadURL := dead.URL - dead.Close() + // An upstream that hangs up on every connection before answering. It keeps + // its port, so no other listener can take it. The 502 the proxy synthesizes + // is the same row shape a refused or black-holed dial produces. + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + go func() { + for c, err := ln.Accept(); err == nil; c, err = ln.Accept() { + _ = c.Close() + } + }() + deadURL := "http://" + ln.Addr().String() p, err := pipeline.New([]pipeline.Plugin{}) if err != nil { diff --git a/core/listener/forwardproxy/websocket_review_test.go b/core/listener/forwardproxy/websocket_review_test.go index bb2527f85..fea3b7de3 100644 --- a/core/listener/forwardproxy/websocket_review_test.go +++ b/core/listener/forwardproxy/websocket_review_test.go @@ -355,22 +355,16 @@ func TestResponsesWebSocket_UnansweredPrewarmIsNotFinishedTwice(t *testing.T) { count := &finishCounter{} logs := captureLogs(t) h := newWSHarness(t, config.ResponsesWebSocketsParse, count) + received := make(chan struct{}) scriptedOrigin(h, func(read func() *wsframe.Frame, write func(string)) { - read() // the prewarm, never answered + if read() != nil { // the prewarm, never answered + close(received) + } }) c := h.dial(t, nil) c.send(`{"type":"response.create","generate":false,"model":"gpt-6-luna","input":[]}`) - // Wait until the origin has the prewarm, so the close lands while it is queued. - deadline := time.Now().Add(10 * time.Second) - for { - h.origin.mu.Lock() - n := len(h.origin.creates) - h.origin.mu.Unlock() - if n > 0 || time.Now().After(deadline) { - break - } - time.Sleep(10 * time.Millisecond) - } + // The close lands once the origin has the prewarm, while it is still queued. + <-received c.close() events := h.events(t, codexSID, true) diff --git a/core/plugins/script/internal/worker/worker_test.go b/core/plugins/script/internal/worker/worker_test.go index 63ea36080..cb8de5173 100644 --- a/core/plugins/script/internal/worker/worker_test.go +++ b/core/plugins/script/internal/worker/worker_test.go @@ -647,32 +647,44 @@ func TestEnviron(t *testing.T) { } // A shell whose profile leaves a background process holding stdout still answers -// within the probe's budget: with its PATH when it printed one, empty when it hung. +// while that process lives: with its PATH when it printed one, empty when it hung. +// The process waits on a FIFO the test releases on cleanup, so the exited shell's +// probe can end only through loginPathWaitDelay, within its generous budget. func TestLoginPath_BackgroundChildHoldingStdout(t *testing.T) { sh, err := exec.LookPath("sh") if err != nil { t.Skip("no sh") } - for name, tc := range map[string]struct{ body, want string }{ - "printed then exited": {"sleep 5 &\nshift\nPATH=/from/login exec " + sh + " \"$@\"\n", "/from/login"}, - "hung": {"sleep 5 &\nsleep 5\n", ""}, + for name, tc := range map[string]struct { + body, want string + budget time.Duration + }{ + "printed then exited": {"read _ < \"$HOLD\" &\nshift\nPATH=/from/login exec " + sh + " \"$@\"\n", "/from/login", time.Minute}, + "hung": {"read _ < \"$HOLD\" &\nread _ < \"$HOLD\"\n", "", 500 * time.Millisecond}, } { t.Run(name, func(t *testing.T) { - shell := filepath.Join(t.TempDir(), "shell") + dir := t.TempDir() + hold := filepath.Join(dir, "hold") + if err := syscall.Mkfifo(hold, 0o600); err != nil { + t.Fatal(err) + } + // Opened read-write so it neither blocks nor sends EOF; closing it does. + release, err := os.OpenFile(hold, os.O_RDWR, 0) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { release.Close() }) + shell := filepath.Join(dir, "shell") os.WriteFile(shell, []byte("#!"+sh+"\n"+tc.body), 0o700) t.Setenv("SHELL", shell) + t.Setenv("HOLD", hold) loginPathMu.Lock() loginPathVal, loginPathTried = "", time.Time{} old := loginPathTimeout - loginPathTimeout = 500 * time.Millisecond + loginPathTimeout = tc.budget loginPathMu.Unlock() t.Cleanup(func() { loginPathVal, loginPathTried, loginPathTimeout = "", time.Now(), old }) - t0 := time.Now() - got := loginPath() - if d := time.Since(t0); d > 2*time.Second { - t.Errorf("loginPath took %v against a %v budget", d, loginPathTimeout) - } - if got != tc.want { + if got := loginPath(); got != tc.want { t.Errorf("loginPath() = %q, want %q", got, tc.want) } }) diff --git a/core/reloader/reloader_test.go b/core/reloader/reloader_test.go index ed8c5b4c9..4b7cc98e0 100644 --- a/core/reloader/reloader_test.go +++ b/core/reloader/reloader_test.go @@ -98,6 +98,22 @@ func waitFor(t *testing.T, timeout time.Duration, cond func() bool, msg string) t.Fatalf("timed out waiting for %s", msg) } +// start starts r, and on cleanup stops its watch loop and waits for it to +// close its watcher, so the watcher's descriptors close before the next test +// opens its own. +func start(t *testing.T, r *Reloader) { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + if err := r.Start(ctx); err != nil { + cancel() + t.Fatalf("Start: %v", err) + } + t.Cleanup(func() { + cancel() + <-r.loopDone + }) +} + func setup(t *testing.T) (*Reloader, *fakeBuilder, string, *pipeline.Holder, *pipeline.Holder) { t.Helper() return setupWith(t) @@ -133,11 +149,7 @@ func setupWith(t *testing.T, extra ...Option) (*Reloader, *fakeBuilder, string, WithStartTimeout(5 * time.Second), }, extra...)..., ) - ctx, cancel := context.WithCancel(context.Background()) - t.Cleanup(cancel) - if err := r.Start(ctx); err != nil { - t.Fatalf("Start: %v", err) - } + start(t, r) return r, b, cfgPath, inH, outH } @@ -162,20 +174,20 @@ func TestReloader_PlainRewrite(t *testing.T) { } } -// Two writes within the debounce window coalesce into a single reload. +// Two writes within the debounce window coalesce into a single reload. The +// window is wide so a loaded runner still lands both writes inside it. func TestReloader_DebouncesBurst(t *testing.T) { - r, b, cfgPath, _, _ := setup(t) + r, b, cfgPath, _, _ := setupWith(t, WithDebounce(500*time.Millisecond)) b.set(builderResult{inbound: emptyPipeline(t), outbound: emptyPipeline(t), cfg: &config.Config{Mode: "envoy-sidecar"}}) writeConfig(t, cfgPath, "mode: envoy-sidecar\n# edit 1\n") - writeConfig(t, cfgPath, "mode: envoy-sidecar\n# edit 2\n") - - waitFor(t, 2*time.Second, func() bool { return r.Status().ReloadsOK >= 1 }, "first reload") - // Give the debounce a chance to fire a second time if it's going to. - time.Sleep(100 * time.Millisecond) + last := "mode: envoy-sidecar\n# edit 2\n" + writeConfig(t, cfgPath, last) - if got := r.Status().ReloadsOK; got > 1 { - t.Errorf("expected coalesced single reload, got %d", got) + want := r.hash([]byte(last)) + waitFor(t, 5*time.Second, func() bool { return r.Status().ActiveConfigSHA256 == want }, "the last edit to go live") + if got := r.Status().ReloadsOK; got != 1 { + t.Errorf("ReloadsOK = %d, want the burst coalesced into 1", got) } } @@ -290,11 +302,7 @@ func TestReloader_RefusesCostLedgerChange(t *testing.T) { r := New(cfgPath, inH, outH, build, initialCfg, WithDebounce(20*time.Millisecond), WithDrainWindow(0), WithStartTimeout(5*time.Second)) - ctx, cancel := context.WithCancel(context.Background()) - t.Cleanup(cancel) - if err := r.Start(ctx); err != nil { - t.Fatalf("Start: %v", err) - } + start(t, r) shaBefore := r.Status().ActiveConfigSHA256 // The edit an operator makes to stop writing cost history. @@ -465,11 +473,7 @@ func setupAt(t *testing.T, cfgPath string) (*Reloader, *fakeBuilder, *pipeline.H WithDrainWindow(0), WithStartTimeout(5*time.Second), ) - ctx, cancel := context.WithCancel(context.Background()) - t.Cleanup(cancel) - if err := r.Start(ctx); err != nil { - t.Fatalf("Start: %v", err) - } + start(t, r) return r, b, inH, outH } @@ -899,11 +903,7 @@ func TestReloader_AnEditDuringStartIsReloaded(t *testing.T) { r := New(cfgPath, pipeline.NewHolder(emptyPipeline(t)), pipeline.NewHolder(emptyPipeline(t)), b.build, initialCfg, WithDebounce(20*time.Millisecond), WithDrainWindow(0), WithStartTimeout(5*time.Second), WithExtraWatch(dirs, func([]string) string { return "" })) - ctx, cancel := context.WithCancel(context.Background()) - t.Cleanup(cancel) - if err := r.Start(ctx); err != nil { - t.Fatalf("Start: %v", err) - } + start(t, r) waitFor(t, 2*time.Second, func() bool { return r.Status().ReloadsOK == 1 }, "the edit made during Start to reload") } @@ -1006,11 +1006,7 @@ func TestReloader_RefusesStatsAddressChange(t *testing.T) { outH := pipeline.NewHolder(emptyPipeline(t)) r := New(cfgPath, inH, outH, build, initialCfg, WithDebounce(20*time.Millisecond), WithDrainWindow(0), WithStartTimeout(5*time.Second)) - ctx, cancel := context.WithCancel(context.Background()) - t.Cleanup(cancel) - if err := r.Start(ctx); err != nil { - t.Fatalf("Start: %v", err) - } + start(t, r) writeConfig(t, cfgPath, "mode: envoy-sidecar\nstats:\n address: 127.0.0.1:19094\n") diff --git a/core/session/archive/archive_test.go b/core/session/archive/archive_test.go index aa7db3e34..457843896 100644 --- a/core/session/archive/archive_test.go +++ b/core/session/archive/archive_test.go @@ -4,6 +4,7 @@ import ( "errors" "os" "path/filepath" + "runtime" "slices" "sync" "syscall" @@ -166,19 +167,16 @@ func TestArchive_ANewProcessOpensANewSegment(t *testing.T) { func TestRecord_NeverBlocksAndCountsDropsWhenFull(t *testing.T) { a := openTest(t, t.TempDir(), newClock(), WithQueueDepth(reserve+6)) a.gate = make(chan struct{}) + // The writer takes a rename and is held at the gate with it, so the events + // below see the queue empty. + a.Rekeyed("filler", "filler2") + for len(a.ops) > 0 { + runtime.Gosched() + } evs := synthSession(45, 34, 512) // 102 events - done := make(chan struct{}) - go func() { - newRecorder(a).record("s1", evs...) - close(done) - }() - select { - case <-done: - case <-time.After(5 * time.Second): - t.Fatal("Record blocked with the writer stalled") - } - // The writer took one op and is held at the gate; six more fit below the reserve. - if got, want := a.Stats().DroppedEvents, uint64(len(evs)-7); got != want { + newRecorder(a).record("s1", evs...) + // Six fit below the reserve. + if got, want := a.Stats().DroppedEvents, uint64(len(evs)-6); got != want { t.Fatalf("dropped %d, want %d", got, want) } close(a.gate) diff --git a/core/sessionapi/usage_test.go b/core/sessionapi/usage_test.go index a5b4e43cc..10e4c92a7 100644 --- a/core/sessionapi/usage_test.go +++ b/core/sessionapi/usage_test.go @@ -555,8 +555,11 @@ func recordCostedMinute(t *testing.T, led *ledger.Writer, at time.Time, host, mo func TestHandleUsage_TodayIsServedFromTheLedger(t *testing.T) { // A closed minute on disk, plus an empty ring, so a non-zero total can only // have come from the ledger. - led := ledgerWithOneCostedMinute(t, insideToday(t, 2*time.Minute), "gw", "opus", 0.25) - ts, _ := newTestServer(t, WithUsage(usage.New()), WithCostLedger(led)) + // At noon, so today spans more than one bucket whatever time the suite runs. + now := time.Date(2026, time.June, 15, 12, 0, 0, 0, time.Local) + led := ledgerWithOneCostedMinute(t, now.Add(-2*time.Minute), "gw", "opus", 0.25) + ts, _ := newTestServer(t, WithUsage(usage.New()), WithCostLedger(led), + WithClock(func() time.Time { return now })) status, body := fetchUsage(t, ts.URL, "?window=today") if status != http.StatusOK { From 30bd2071c9f584c857585c04ee924e2b4ae9e8ba Mon Sep 17 00:00:00 2001 From: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> Date: Mon, 9 Nov 2026 16:00:23 -0700 Subject: [PATCH 2/5] :white_check_mark: Test updates Signed-off-by: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> --- cmd/agentop/tui/sessions_title_test.go | 15 +++++---------- core/reloader/reloader.go | 3 +++ core/reloader/reloader_test.go | 7 +++---- 3 files changed, 11 insertions(+), 14 deletions(-) diff --git a/cmd/agentop/tui/sessions_title_test.go b/cmd/agentop/tui/sessions_title_test.go index b4778b200..9e674788c 100644 --- a/cmd/agentop/tui/sessions_title_test.go +++ b/cmd/agentop/tui/sessions_title_test.go @@ -1218,23 +1218,18 @@ func TestLooksLikePath_AnyUnicodeSpaceSeparates(t *testing.T) { // // Asserted as a RATIO rather than a wall-clock bound, so it says what it means on a loaded CI machine: // quadratic growth shows up as ~4x per doubling, linear as ~2x, and the ceiling sits between them. -// Both ends are timed inside one test so the comparison is against the same machine at the same moment, -// and each is the best of three runs, so one preempted run does not set the ratio. +// Both ends are timed inside one test so the comparison is against the same machine at the same moment. func TestTrunc_ScalesLinearly(t *testing.T) { const budget = 40 measure := func(f func(string, int) string, runes int) time.Duration { s := "/" + strings.Repeat("a", runes-1) // Warm, so the first-call cost of anything lazy is not charged to the small input. f(s, budget) - best := time.Duration(1<<63 - 1) - for range 3 { - start := time.Now() - for i := 0; i < 20; i++ { - f(s, budget) - } - best = min(best, time.Since(start)) + start := time.Now() + for i := 0; i < 20; i++ { + f(s, budget) } - return best + return time.Since(start) } for _, tc := range []struct { name string diff --git a/core/reloader/reloader.go b/core/reloader/reloader.go index f93f0c03a..c55bfebc0 100644 --- a/core/reloader/reloader.go +++ b/core/reloader/reloader.go @@ -112,6 +112,8 @@ type Reloader struct { drainWindow time.Duration debounce time.Duration startTimeout time.Duration + // afterFunc arms the debounce timer: time.AfterFunc, or a test's own. + afterFunc func(time.Duration, func()) debounceTimer // onCommit is run after a reload is accepted; see WithOnCommit. onCommit func(*config.Config) // extraDirs and fingerprint widen what a reload depends on; see WithExtraWatch. @@ -160,6 +162,7 @@ func New(configPath string, inbound, outbound *pipeline.Holder, build PipelineBu drainWindow: 10 * time.Minute, debounce: 250 * time.Millisecond, startTimeout: 60 * time.Second, + afterFunc: func(d time.Duration, f func()) debounceTimer { return time.AfterFunc(d, f) }, closing: make(chan struct{}), } for _, opt := range opts { diff --git a/core/reloader/reloader_test.go b/core/reloader/reloader_test.go index 4b7cc98e0..bb88c8294 100644 --- a/core/reloader/reloader_test.go +++ b/core/reloader/reloader_test.go @@ -174,10 +174,9 @@ func TestReloader_PlainRewrite(t *testing.T) { } } -// Two writes within the debounce window coalesce into a single reload. The -// window is wide so a loaded runner still lands both writes inside it. +// Two writes within the debounce window coalesce into a single reload. func TestReloader_DebouncesBurst(t *testing.T) { - r, b, cfgPath, _, _ := setupWith(t, WithDebounce(500*time.Millisecond)) + r, b, cfgPath, _, _ := setup(t) b.set(builderResult{inbound: emptyPipeline(t), outbound: emptyPipeline(t), cfg: &config.Config{Mode: "envoy-sidecar"}}) writeConfig(t, cfgPath, "mode: envoy-sidecar\n# edit 1\n") @@ -185,7 +184,7 @@ func TestReloader_DebouncesBurst(t *testing.T) { writeConfig(t, cfgPath, last) want := r.hash([]byte(last)) - waitFor(t, 5*time.Second, func() bool { return r.Status().ActiveConfigSHA256 == want }, "the last edit to go live") + waitFor(t, 2*time.Second, func() bool { return r.Status().ActiveConfigSHA256 == want }, "the last edit to go live") if got := r.Status().ReloadsOK; got != 1 { t.Errorf("ReloadsOK = %d, want the burst coalesced into 1", got) } From 547a6773328373ce75cc05b68f55aa13c886299b Mon Sep 17 00:00:00 2001 From: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> Date: Mon, 9 Nov 2026 17:12:15 -0700 Subject: [PATCH 3/5] :recycle: Minimize source code changes Signed-off-by: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> --- cmd/agentop/tui/sessions_title_test.go | 27 +++------- core/plugins/script/internal/worker/env.go | 7 ++- .../script/internal/worker/worker_test.go | 53 ++++++++++++------- core/reloader/reloader.go | 10 +++- core/reloader/reloader_test.go | 35 +++++++++--- 5 files changed, 86 insertions(+), 46 deletions(-) diff --git a/cmd/agentop/tui/sessions_title_test.go b/cmd/agentop/tui/sessions_title_test.go index 9e674788c..fa68b1eaa 100644 --- a/cmd/agentop/tui/sessions_title_test.go +++ b/cmd/agentop/tui/sessions_title_test.go @@ -1216,20 +1216,14 @@ func TestLooksLikePath_AnyUnicodeSpaceSeparates(t *testing.T) { // 2.33s — four times the input for sixteen times the work. Reachable because the cwd tier was // uncapped, and the renderer redraws on every poll. // -// Asserted as a RATIO rather than a wall-clock bound, so it says what it means on a loaded CI machine: -// quadratic growth shows up as ~4x per doubling, linear as ~2x, and the ceiling sits between them. -// Both ends are timed inside one test so the comparison is against the same machine at the same moment. +// The cost is counted in allocations, so the assertion is about the work done and not about the +// machine: the skip-ahead tries at most budget candidates whatever the input length, and the +// one-at-a-time search allocates a candidate per dropped rune. func TestTrunc_ScalesLinearly(t *testing.T) { const budget = 40 - measure := func(f func(string, int) string, runes int) time.Duration { + measure := func(f func(string, int) string, runes int) float64 { s := "/" + strings.Repeat("a", runes-1) - // Warm, so the first-call cost of anything lazy is not charged to the small input. - f(s, budget) - start := time.Now() - for i := 0; i < 20; i++ { - f(s, budget) - } - return time.Since(start) + return testing.AllocsPerRun(3, func() { f(s, budget) }) } for _, tc := range []struct { name string @@ -1241,15 +1235,10 @@ func TestTrunc_ScalesLinearly(t *testing.T) { t.Run(tc.name, func(t *testing.T) { small := measure(tc.f, 4000) large := measure(tc.f, 16000) - if small <= 0 { - t.Skip("timer resolution too coarse to compare") - } - // 4x the input. Linear predicts ~4x the time; quadratic predicts ~16x. A ceiling of 8x - // separates them with room for scheduling noise. - if ratio := float64(large) / float64(small); ratio > 8 { - t.Errorf("4x the input took %.1fx the time (%v -> %v) — that is the quadratic shape "+ + if large > small { + t.Errorf("4x the input made %.0f allocations, up from %.0f — that is the quadratic shape "+ "back: the per-rune search must skip to the last/first n runes before measuring", - ratio, small, large) + large, small) } }) } diff --git a/core/plugins/script/internal/worker/env.go b/core/plugins/script/internal/worker/env.go index 31f36e308..ee46feffa 100644 --- a/core/plugins/script/internal/worker/env.go +++ b/core/plugins/script/internal/worker/env.go @@ -90,6 +90,11 @@ var ( // exited or been killed. A process the profile backgrounds inherits stdout, and // without this the probe would last as long as that process does. loginPathWaitDelay = 100 * time.Millisecond + // loginPathContext is one probe's context, bounded by loginPathTimeout. A var + // so a test can end a probe itself. + loginPathContext = func() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), loginPathTimeout) + } ) // loginPath is the PATH a login shell of the user's own sets up. A successful @@ -107,7 +112,7 @@ func loginPath() string { if shell == "" { return "" } - ctx, cancel := context.WithTimeout(context.Background(), loginPathTimeout) + ctx, cancel := loginPathContext() defer cancel() cmd := exec.CommandContext(ctx, shell, "-l", "-c", `printf '\n'"$0"'%s'"$0"'\n' "$PATH"`, pathMarker) cmd.WaitDelay = loginPathWaitDelay diff --git a/core/plugins/script/internal/worker/worker_test.go b/core/plugins/script/internal/worker/worker_test.go index cb8de5173..d333a15f8 100644 --- a/core/plugins/script/internal/worker/worker_test.go +++ b/core/plugins/script/internal/worker/worker_test.go @@ -647,9 +647,9 @@ func TestEnviron(t *testing.T) { } // A shell whose profile leaves a background process holding stdout still answers -// while that process lives: with its PATH when it printed one, empty when it hung. -// The process waits on a FIFO the test releases on cleanup, so the exited shell's -// probe can end only through loginPathWaitDelay, within its generous budget. +// while that process lives: with its PATH when it printed one, empty when it hung +// and its probe was ended. The process waits on a FIFO the test releases on +// cleanup, so the exited shell's probe ends only through loginPathWaitDelay. func TestLoginPath_BackgroundChildHoldingStdout(t *testing.T) { sh, err := exec.LookPath("sh") if err != nil { @@ -657,33 +657,50 @@ func TestLoginPath_BackgroundChildHoldingStdout(t *testing.T) { } for name, tc := range map[string]struct { body, want string - budget time.Duration + hangs bool }{ - "printed then exited": {"read _ < \"$HOLD\" &\nshift\nPATH=/from/login exec " + sh + " \"$@\"\n", "/from/login", time.Minute}, - "hung": {"read _ < \"$HOLD\" &\nread _ < \"$HOLD\"\n", "", 500 * time.Millisecond}, + "printed then exited": {"read _ < \"$HOLD\" &\nshift\nPATH=/from/login exec " + sh + " \"$@\"\n", "/from/login", false}, + "hung": {"read _ < \"$HOLD\" &\necho > \"$STARTED\"\nread _ < \"$HOLD\"\n", "", true}, } { t.Run(name, func(t *testing.T) { dir := t.TempDir() - hold := filepath.Join(dir, "hold") - if err := syscall.Mkfifo(hold, 0o600); err != nil { - t.Fatal(err) - } - // Opened read-write so it neither blocks nor sends EOF; closing it does. - release, err := os.OpenFile(hold, os.O_RDWR, 0) - if err != nil { - t.Fatal(err) + // fifo makes a FIFO, opened read-write so opening it neither blocks nor + // sends EOF; closing it on cleanup does. + fifo := func(name string) (string, *os.File) { + path := filepath.Join(dir, name) + if err := syscall.Mkfifo(path, 0o600); err != nil { + t.Fatal(err) + } + f, err := os.OpenFile(path, os.O_RDWR, 0) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { f.Close() }) + return path, f } - t.Cleanup(func() { release.Close() }) + hold, _ := fifo("hold") + started, startedR := fifo("started") shell := filepath.Join(dir, "shell") os.WriteFile(shell, []byte("#!"+sh+"\n"+tc.body), 0o700) t.Setenv("SHELL", shell) t.Setenv("HOLD", hold) + t.Setenv("STARTED", started) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + if tc.hangs { + // The probe ends once the shell is running and hung. + go func() { + if _, err := startedR.Read(make([]byte, 1)); err == nil { + cancel() + } + }() + } loginPathMu.Lock() loginPathVal, loginPathTried = "", time.Time{} - old := loginPathTimeout - loginPathTimeout = tc.budget + old := loginPathContext + loginPathContext = func() (context.Context, context.CancelFunc) { return ctx, func() {} } loginPathMu.Unlock() - t.Cleanup(func() { loginPathVal, loginPathTried, loginPathTimeout = "", time.Now(), old }) + t.Cleanup(func() { loginPathVal, loginPathTried, loginPathContext = "", time.Now(), old }) if got := loginPath(); got != tc.want { t.Errorf("loginPath() = %q, want %q", got, tc.want) } diff --git a/core/reloader/reloader.go b/core/reloader/reloader.go index c55bfebc0..68df9e8ce 100644 --- a/core/reloader/reloader.go +++ b/core/reloader/reloader.go @@ -308,6 +308,12 @@ func (r *Reloader) ConfigProvider() func() *config.Config { return func() *config.Config { return r.activeCfg.Load() } } +// debounceTimer is the part of *time.Timer the debounce uses. +type debounceTimer interface { + Reset(time.Duration) bool + Stop() bool +} + // watchLoop drains fsnotify events, debounces them, and dispatches // each debounced burst to reloadOnce. Exits when ctx is cancelled or // the watcher channel closes. @@ -315,7 +321,7 @@ func (r *Reloader) watchLoop(ctx context.Context, watcher *fsnotify.Watcher) { defer watcher.Close() base := filepath.Base(r.configPath) - var timer *time.Timer + var timer debounceTimer defer func() { if timer != nil { timer.Stop() @@ -358,7 +364,7 @@ func (r *Reloader) watchLoop(ctx context.Context, watcher *fsnotify.Watcher) { // Debounce: reset the timer on every event. Fire once when // the burst quiesces for `debounce`. if timer == nil { - timer = time.AfterFunc(r.debounce, func() { + timer = r.afterFunc(r.debounce, func() { select { case trigger <- struct{}{}: default: diff --git a/core/reloader/reloader_test.go b/core/reloader/reloader_test.go index bb88c8294..907414d62 100644 --- a/core/reloader/reloader_test.go +++ b/core/reloader/reloader_test.go @@ -174,19 +174,42 @@ func TestReloader_PlainRewrite(t *testing.T) { } } -// Two writes within the debounce window coalesce into a single reload. +// manualTimer is a debounce timer that only the test fires. +type manualTimer struct{} + +func (manualTimer) Reset(time.Duration) bool { return true } +func (manualTimer) Stop() bool { return true } + +// A burst of writes lands on one debounce timer, and its one firing reloads +// the last write. func TestReloader_DebouncesBurst(t *testing.T) { - r, b, cfgPath, _, _ := setup(t) + var arms atomic.Int32 + armed := make(chan func(), 1) + manual := func(r *Reloader) { + r.afterFunc = func(_ time.Duration, fire func()) debounceTimer { + if arms.Add(1) == 1 { + armed <- fire + } + return manualTimer{} + } + } + committed := make(chan struct{}, 2) + r, b, cfgPath, _, _ := setupWith(t, manual, WithOnCommit(func(*config.Config) { committed <- struct{}{} })) b.set(builderResult{inbound: emptyPipeline(t), outbound: emptyPipeline(t), cfg: &config.Config{Mode: "envoy-sidecar"}}) writeConfig(t, cfgPath, "mode: envoy-sidecar\n# edit 1\n") last := "mode: envoy-sidecar\n# edit 2\n" writeConfig(t, cfgPath, last) - want := r.hash([]byte(last)) - waitFor(t, 2*time.Second, func() bool { return r.Status().ActiveConfigSHA256 == want }, "the last edit to go live") - if got := r.Status().ReloadsOK; got != 1 { - t.Errorf("ReloadsOK = %d, want the burst coalesced into 1", got) + (<-armed)() + <-committed + r.Close(context.Background()) // returns once the watch loop has, so Status is final + st := r.Status() + if st.ReloadsOK != 1 || st.ActiveConfigSHA256 != r.hash([]byte(last)) { + t.Errorf("ReloadsOK = %d, active %s; want 1 reload of the last edit", st.ReloadsOK, st.ActiveConfigSHA256) + } + if n := arms.Load(); n != 1 { + t.Errorf("the burst armed %d debounce timers, want 1", n) } } From 046e328d212289ba317251b2cda0b3d78bb5aed7 Mon Sep 17 00:00:00 2001 From: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> Date: Mon, 9 Nov 2026 22:35:21 -0700 Subject: [PATCH 4/5] :art: Address review comments Signed-off-by: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> --- cmd/agentop/tui/sessions_title_test.go | 4 ++- .../forwardproxy/websocket_review_test.go | 10 +++--- .../script/internal/worker/worker_test.go | 6 +++- core/reloader/reloader_test.go | 32 +++++++++++++------ 4 files changed, 36 insertions(+), 16 deletions(-) diff --git a/cmd/agentop/tui/sessions_title_test.go b/cmd/agentop/tui/sessions_title_test.go index fa68b1eaa..7771caa26 100644 --- a/cmd/agentop/tui/sessions_title_test.go +++ b/cmd/agentop/tui/sessions_title_test.go @@ -1235,7 +1235,9 @@ func TestTrunc_ScalesLinearly(t *testing.T) { t.Run(tc.name, func(t *testing.T) { small := measure(tc.f, 4000) large := measure(tc.f, 16000) - if large > small { + // The allocation count is process-wide, so the ceiling leaves room for a stray + // allocation; the quadratic search makes thousands. + if large > 2*small { t.Errorf("4x the input made %.0f allocations, up from %.0f — that is the quadratic shape "+ "back: the per-rune search must skip to the last/first n runes before measuring", large, small) diff --git a/core/listener/forwardproxy/websocket_review_test.go b/core/listener/forwardproxy/websocket_review_test.go index fea3b7de3..04bc9450d 100644 --- a/core/listener/forwardproxy/websocket_review_test.go +++ b/core/listener/forwardproxy/websocket_review_test.go @@ -355,16 +355,16 @@ func TestResponsesWebSocket_UnansweredPrewarmIsNotFinishedTwice(t *testing.T) { count := &finishCounter{} logs := captureLogs(t) h := newWSHarness(t, config.ResponsesWebSocketsParse, count) - received := make(chan struct{}) + received := make(chan bool, 1) scriptedOrigin(h, func(read func() *wsframe.Frame, write func(string)) { - if read() != nil { // the prewarm, never answered - close(received) - } + received <- read() != nil // the prewarm, never answered }) c := h.dial(t, nil) c.send(`{"type":"response.create","generate":false,"model":"gpt-6-luna","input":[]}`) // The close lands once the origin has the prewarm, while it is still queued. - <-received + if !<-received { + t.Fatal("the origin's socket closed before the prewarm reached it") + } c.close() events := h.events(t, codexSID, true) diff --git a/core/plugins/script/internal/worker/worker_test.go b/core/plugins/script/internal/worker/worker_test.go index d333a15f8..1f8e966c2 100644 --- a/core/plugins/script/internal/worker/worker_test.go +++ b/core/plugins/script/internal/worker/worker_test.go @@ -700,7 +700,11 @@ func TestLoginPath_BackgroundChildHoldingStdout(t *testing.T) { old := loginPathContext loginPathContext = func() (context.Context, context.CancelFunc) { return ctx, func() {} } loginPathMu.Unlock() - t.Cleanup(func() { loginPathVal, loginPathTried, loginPathContext = "", time.Now(), old }) + t.Cleanup(func() { + loginPathMu.Lock() + defer loginPathMu.Unlock() + loginPathVal, loginPathTried, loginPathContext = "", time.Now(), old + }) if got := loginPath(); got != tc.want { t.Errorf("loginPath() = %q, want %q", got, tc.want) } diff --git a/core/reloader/reloader_test.go b/core/reloader/reloader_test.go index 907414d62..606785f57 100644 --- a/core/reloader/reloader_test.go +++ b/core/reloader/reloader_test.go @@ -174,23 +174,36 @@ func TestReloader_PlainRewrite(t *testing.T) { } } -// manualTimer is a debounce timer that only the test fires. -type manualTimer struct{} +// manualTimer is a debounce timer that only the test fires. Each Reset runs +// onReset. +type manualTimer struct{ onReset func() } -func (manualTimer) Reset(time.Duration) bool { return true } -func (manualTimer) Stop() bool { return true } +func (m manualTimer) Reset(time.Duration) bool { m.onReset(); return true } +func (manualTimer) Stop() bool { return true } -// A burst of writes lands on one debounce timer, and its one firing reloads -// the last write. +// A burst of writes lands on one debounce timer, which the last write resets, +// and its one firing reloads the last write. func TestReloader_DebouncesBurst(t *testing.T) { var arms atomic.Int32 armed := make(chan func(), 1) + last := "mode: envoy-sidecar\n# edit 2\n" + reset := make(chan struct{}, 1) manual := func(r *Reloader) { + // One write can raise several events, so a reset counts once the file + // holds the last write. + onReset := func() { + if b, _ := os.ReadFile(r.configPath); string(b) == last { + select { + case reset <- struct{}{}: + default: + } + } + } r.afterFunc = func(_ time.Duration, fire func()) debounceTimer { if arms.Add(1) == 1 { armed <- fire } - return manualTimer{} + return manualTimer{onReset: onReset} } } committed := make(chan struct{}, 2) @@ -198,10 +211,11 @@ func TestReloader_DebouncesBurst(t *testing.T) { b.set(builderResult{inbound: emptyPipeline(t), outbound: emptyPipeline(t), cfg: &config.Config{Mode: "envoy-sidecar"}}) writeConfig(t, cfgPath, "mode: envoy-sidecar\n# edit 1\n") - last := "mode: envoy-sidecar\n# edit 2\n" + fire := <-armed writeConfig(t, cfgPath, last) + <-reset - (<-armed)() + fire() <-committed r.Close(context.Background()) // returns once the watch loop has, so Status is final st := r.Status() From a4cfe22c972b2118c1da90376d8d64ea8af76f9d Mon Sep 17 00:00:00 2001 From: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> Date: Tue, 10 Nov 2026 08:47:41 -0700 Subject: [PATCH 5/5] :white_check_mark: Check probe ending Signed-off-by: Evaline Ju <69598118+evaline-ju@users.noreply.github.com> --- core/plugins/script/internal/worker/worker_test.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/core/plugins/script/internal/worker/worker_test.go b/core/plugins/script/internal/worker/worker_test.go index 1f8e966c2..63313efe4 100644 --- a/core/plugins/script/internal/worker/worker_test.go +++ b/core/plugins/script/internal/worker/worker_test.go @@ -705,7 +705,11 @@ func TestLoginPath_BackgroundChildHoldingStdout(t *testing.T) { defer loginPathMu.Unlock() loginPathVal, loginPathTried, loginPathContext = "", time.Now(), old }) - if got := loginPath(); got != tc.want { + got := loginPath() + if tc.hangs && ctx.Err() == nil { + t.Error("loginPath returned before the shell signalled it was hung") + } + if got != tc.want { t.Errorf("loginPath() = %q, want %q", got, tc.want) } })