test(llm): release detached pipe regression worker

internal/llm/subprocess_launch_unix_test.go

@@ -5,10 +5,12 @@ package llm
 import (
 	"bufio"
 	"context"
+	"errors"
 	"io"
+	"net"
+	"os"
 	"os/exec"
 	"path/filepath"
-	"strings"
 	"testing"
 	"time"
 )
@@ -27,22 +29,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 || !errors.Is(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 +152,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)
+	}
 }