From fee40bd667a1b90a133d72bc0ce6c0e12c627363 Mon Sep 17 00:00:00 2001 From: zzwong <6979793+zzwong@users.noreply.github.com> Date: Thu, 8 Oct 2026 16:39:04 -0400 Subject: [PATCH 1/2] test(llm): release detached pipe regression worker --- internal/llm/subprocess_launch_unix_test.go | 126 ++++++++++++++++++-- 1 file changed, 115 insertions(+), 11 deletions(-) diff --git a/internal/llm/subprocess_launch_unix_test.go b/internal/llm/subprocess_launch_unix_test.go index 91a38009..edb96d61 100644 --- a/internal/llm/subprocess_launch_unix_test.go +++ b/internal/llm/subprocess_launch_unix_test.go @@ -6,9 +6,10 @@ import ( "bufio" "context" "io" + "net" + "os" "os/exec" "path/filepath" - "strings" "testing" "time" ) @@ -27,22 +28,114 @@ func TestLaunchProcessUnblocksReadOnCanceledContext(t *testing.T) { subprocessWaitDelay = 100 * time.Millisecond defer func() { subprocessWaitDelay = prev }() + // Keep the Unix socket path short, including on macOS. This private control + // channel survives cancellation and never shares the pipe under test. + controlDir, err := os.MkdirTemp("", "llm-control-") + if err != nil { + t.Fatal(err) + } + defer func() { _ = os.RemoveAll(controlDir) }() + controlPath := filepath.Join(controlDir, "worker.sock") + listener, err := net.ListenUnix("unix", &net.UnixAddr{Name: controlPath, Net: "unix"}) + if err != nil { + t.Fatal(err) + } + defer func() { _ = listener.Close() }() + if err := listener.SetDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) - // The worker os.setsid()s out of the process group, prints READY on the - // inherited stdout, then holds it open. Reading READY proves it has escaped - // and owns the pipe. The parent sleeps so it (and the group kill) can't - // close stdout on its own. - script := `python3 -c 'import os,sys,time; os.setsid(); print("READY"); sys.stdout.flush(); time.sleep(30)' & sleep 20` + defer cancel() + // The worker detaches before signaling readiness and holds both inherited + // pipes until explicitly released. EOF on the test-owned control socket also + // releases it if the test exits early; the timeout is a final safety bound. + // The shell waits instead of using a timing-only sleep. + script := `python3 -c ' +import os, socket, sys, time +os.setsid() +control = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) +control.settimeout(30) +control.connect(sys.argv[1]) +deadline = time.monotonic() + 30 +try: + print("READY", flush=True) + command = b"" + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + control.settimeout(remaining) + command = control.recv(1) + if command != b"P": + break + control.sendall(b"P") +finally: + os.close(1) + os.close(2) +if command == b"Q": + control.sendall(b"Q") +control.close() +' "$1" & wait` logPath := filepath.Join(t.TempDir(), "subprocess.log") - p, err := LaunchProcess(ctx, "/bin/sh", []string{"-c", script}, "", nil, 0, logPath, func() error { return nil }, false) + p, err := LaunchProcess(ctx, "/bin/sh", []string{"-c", script, "worker", controlPath}, "", nil, 0, logPath, func() error { return nil }, false) if err != nil { t.Fatalf("LaunchProcess: %v", err) } + var control *net.UnixConn + defer func() { + // Release only after the pipe assertion (or an earlier fatal failure). + // Q acknowledges that both inherited pipe descriptors have been closed. + if control != nil { + if err := control.SetDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Error(err) + } else if _, err := control.Write([]byte("Q")); err != nil { + t.Errorf("release worker: %v", err) + } else { + var reply [1]byte + if _, err := io.ReadFull(control, reply[:]); err != nil || reply[0] != 'Q' { + t.Errorf("worker release acknowledgment = %q, err=%v", reply, err) + } + if n, err := control.Read(reply[:]); n != 0 || err != io.EOF { + t.Errorf("worker control did not close: n=%d err=%v", n, err) + } + } + _ = control.Close() + } + cancel() + _ = p.Stdout().Close() + _ = p.Stderr().Close() + waited := make(chan struct{}) + go func() { + _ = p.Command().Wait() + close(waited) + }() + select { + case <-waited: + case <-time.After(5 * time.Second): + t.Error("launcher did not exit during cleanup") + } + closeSubprocessLog(p.logFile) + _ = p.processGroup.close() + }() + control, err = listener.AcceptUnix() + if err != nil { + t.Fatalf("accept worker control: %v", err) + } rd := bufio.NewReader(p.Stdout()) - line, err := rd.ReadString('\n') - if err != nil || !strings.Contains(line, "READY") { - t.Fatalf("worker did not signal READY: %q err=%v", line, err) + ready := make(chan bool, 1) + go func() { + line, err := rd.ReadString('\n') + ready <- err == nil && line == "READY\n" + }() + select { + case ok := <-ready: + if !ok { + t.Fatal("worker did not signal READY") + } + case <-time.After(5 * time.Second): + t.Fatal("worker readiness timed out") } cancel() // simulate task-timeout / abort; the escaped worker still holds stdout @@ -58,5 +151,16 @@ func TestLaunchProcessUnblocksReadOnCanceledContext(t *testing.T) { case <-time.After(5 * time.Second): t.Fatal("stdout read wedged after context cancel: pipes were not force-closed") } - _ = p.Command().Wait() + // A live response rules out helper exit as the reason the read unblocked. + // Do not release the worker or call Wait before this assertion. + if err := control.SetDeadline(time.Now().Add(5 * time.Second)); err != nil { + t.Fatal(err) + } + if _, err := control.Write([]byte("P")); err != nil { + t.Fatalf("probe escaped worker: %v", err) + } + var reply [1]byte + if _, err := io.ReadFull(control, reply[:]); err != nil || reply[0] != 'P' { + t.Fatalf("escaped worker is not holding the pipes: reply=%q err=%v", reply, err) + } } From 2d48d94caf01118c18cbd07b282287fa9dd2d51c Mon Sep 17 00:00:00 2001 From: zzwong <6979793+zzwong@users.noreply.github.com> Date: Thu, 8 Oct 2026 16:43:29 -0400 Subject: [PATCH 2/2] test(llm): check worker control EOF with errors.Is --- internal/llm/subprocess_launch_unix_test.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/internal/llm/subprocess_launch_unix_test.go b/internal/llm/subprocess_launch_unix_test.go index edb96d61..e70afc69 100644 --- a/internal/llm/subprocess_launch_unix_test.go +++ b/internal/llm/subprocess_launch_unix_test.go @@ -5,6 +5,7 @@ package llm import ( "bufio" "context" + "errors" "io" "net" "os" @@ -96,7 +97,7 @@ control.close() if _, err := io.ReadFull(control, reply[:]); err != nil || reply[0] != 'Q' { t.Errorf("worker release acknowledgment = %q, err=%v", reply, err) } - if n, err := control.Read(reply[:]); n != 0 || err != io.EOF { + if n, err := control.Read(reply[:]); n != 0 || !errors.Is(err, io.EOF) { t.Errorf("worker control did not close: n=%d err=%v", n, err) } }