fix(process): bound context command cleanup

internal/agenteval/agent_command.go

@@ -3,9 +3,10 @@ package agenteval
 import (
 	"bytes"
 	"context"
-	"errors"
 	"os/exec"
 	"strings"
+
+	"github.com/Gitlawb/zero/internal/execution"
 )
 
 type AgentRunInput struct {
@@ -67,7 +68,7 @@ func (runner CommandAgentRunner) Run(ctx context.Context, input AgentRunInput) A
 	cmd.Stdout = stdout
 	cmd.Stderr = stderr
 
-	err := cmd.Run()
+	err := execution.RunCommand(ctx, cmd)
 	result.Stdout = stdout.buf.String()
 	result.Stderr = stderr.buf.String()
 	result.Truncated = stdout.truncated || stderr.truncated
@@ -81,8 +82,7 @@ func (runner CommandAgentRunner) Run(ctx context.Context, input AgentRunInput) A
 		result.Error = ctxErr.Error()
 		return result
 	}
-	var exitErr *exec.ExitError
-	if errors.As(err, &exitErr) {
+	if exitErr, ok := execution.AsPureExitError(err); ok {
 		result.ExitCode = exitErr.ExitCode()
 		return result
 	}

internal/agenteval/materialize.go

@@ -10,6 +10,8 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+
+	"github.com/Gitlawb/zero/internal/execution"
 )
 
 type Materializer struct{}
@@ -198,7 +200,7 @@ func initGitBaseline(ctx context.Context, workspace string) error {
 		var output bytes.Buffer
 		cmd.Stdout = &output
 		cmd.Stderr = &output
-		if err := cmd.Run(); err != nil {
+		if err := execution.RunCommand(ctx, cmd); err != nil {
 			if ctxErr := ctx.Err(); ctxErr != nil {
 				return ctxErr
 			}

internal/agenteval/run.go

@@ -10,6 +10,8 @@ import (
 	"path/filepath"
 	"strings"
 	"time"
+
+	"github.com/Gitlawb/zero/internal/execution"
 )
 
 // defaultCommandTimeout bounds a single verification command so a hung command
@@ -144,15 +146,14 @@ func execCommand(ctx context.Context, workspace string, command Command) Command
 	var stderr bytes.Buffer
 	cmd.Stdout = &stdout
 	cmd.Stderr = &stderr
-	err := cmd.Run()
+	err := execution.RunCommand(ctx, cmd)
 	result.Stdout = stdout.String()
 	result.Stderr = stderr.String()
 	if err == nil {
 		result.ExitCode = 0
 		return result
 	}
-	var exitErr *exec.ExitError
-	if errors.As(err, &exitErr) {
+	if exitErr, ok := execution.AsPureExitError(err); ok {
 		result.ExitCode = exitErr.ExitCode()
 		return result
 	}
@@ -167,14 +168,16 @@ func execCommand(ctx context.Context, workspace string, command Command) Command
 func defaultRunGit(ctx context.Context, workspace string, args ...string) ([]byte, error) {
 	allArgs := append([]string{"-C", workspace}, args...)
 	cmd := exec.CommandContext(ctx, "git", allArgs...)
-	output, err := cmd.Output()
+	var output bytes.Buffer
+	cmd.Stdout = &output
+	err := execution.RunCommand(ctx, cmd)
 	if err != nil {
 		if ctxErr := ctx.Err(); ctxErr != nil {
 			return nil, ctxErr
 		}
 		return nil, err
 	}
-	return output, nil
+	return output.Bytes(), nil
 }
 
 func parseGitStatusPorcelain(output []byte) []string {

internal/dictation/runner.go

@@ -7,6 +7,8 @@ import (
 	"os"
 	"os/exec"
 	"time"
+
+	"github.com/Gitlawb/zero/internal/execution"
 )
 
 // commandSpec describes one capture-process invocation. Argv is always
@@ -107,7 +109,7 @@ func runCommandOutput(ctx context.Context, name string, args ...string) ([]byte,
 	var out bytes.Buffer
 	cmd.Stdout = &out
 	cmd.Stderr = &out
-	err := cmd.Run()
+	err := execution.RunCommand(ctx, cmd)
 	return out.Bytes(), err
 }
 

internal/execution/command_context.go

@@ -0,0 +1,152 @@
+package execution
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"reflect"
+	"sync"
+)
+
+// RunCommand runs a context-bound command in a retained process tree and
+// prevents inherited output handles from blocking Wait indefinitely.
+func RunCommand(ctx context.Context, command *exec.Cmd) (err error) {
+	if command == nil {
+		return errors.New("execution: nil command")
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	drains := observeOutputDrains(command)
+	tree, err := prepareCommandTree(command)
+	if err != nil {
+		return err
+	}
+	defer func() { err = errors.Join(err, tree.close()) }()
+
+	command.WaitDelay = processWaitDelay
+	// Adapters may return exec.Command, which rejects a non-nil Cancel.
+	// The watcher below also owns cancellation for commands without that hook.
+	if command.Cancel != nil {
+		command.Cancel = tree.cancel
+	}
+	if err := command.Start(); err != nil {
+		_ = tree.attach(nil)
+		return err
+	}
+	if err := tree.attach(command.Process); err != nil {
+		killErr := command.Process.Kill()
+		waitErr := command.Wait()
+		return errors.Join(fmt.Errorf("execution: attach process tree: %w", err), killErr, waitErr)
+	}
+	waitComplete := make(chan struct{})
+	type cancellation struct {
+		err      error
+		canceled bool
+	}
+	cancelResult := make(chan cancellation, 1)
+	go func() {
+		select {
+		case <-ctx.Done():
+			cancelResult <- cancellation{err: tree.cancel(), canceled: true}
+		case <-waitComplete:
+			cancelResult <- cancellation{}
+		}
+	}()
+	waitErr := command.Wait()
+	close(waitComplete)
+	canceled := <-cancelResult
+	if canceled.canceled {
+		return errors.Join(waitErr, ctx.Err(), canceled.err)
+	}
+	if waitErr != nil && drains.err() != nil {
+		// Cmd.Wait intentionally prefers an ExitError over a copying error. When
+		// WaitDelay forcibly closes an inherited output pipe after a nonzero root
+		// exit, retain that cleanup failure so result consumers cannot reconcile it
+		// as an ordinary command exit.
+		return errors.Join(waitErr, exec.ErrWaitDelay, tree.cancel())
+	}
+	if waitErr != nil {
+		return errors.Join(waitErr, tree.cancel())
+	}
+	return waitErr
+}
+
+type outputDrains struct {
+	stdout *drainObserver
+	stderr *drainObserver
+}
+
+func observeOutputDrains(command *exec.Cmd) outputDrains {
+	drains := outputDrains{}
+	sharedOutput := sameWriter(command.Stderr, command.Stdout)
+	if !isFile(command.Stdout) && command.Stdout != nil {
+		drains.stdout = &drainObserver{writer: command.Stdout}
+		command.Stdout = drains.stdout
+	}
+	if sharedOutput && drains.stdout != nil {
+		command.Stderr = drains.stdout
+		drains.stderr = drains.stdout
+	} else if !isFile(command.Stderr) && command.Stderr != nil {
+		drains.stderr = &drainObserver{writer: command.Stderr}
+		command.Stderr = drains.stderr
+	}
+	return drains
+}
+
+func (drains outputDrains) err() error {
+	if drains.stdout != nil && drains.stdout.err() != nil {
+		return drains.stdout.err()
+	}
+	if drains.stderr != nil {
+		return drains.stderr.err()
+	}
+	return nil
+}
+
+func isFile(writer io.Writer) bool {
+	_, ok := writer.(*os.File)
+	return ok
+}
+
+func sameWriter(left io.Writer, right io.Writer) bool {
+	if left == nil || right == nil {
+		return false
+	}
+	leftType := reflect.TypeOf(left)
+	return leftType == reflect.TypeOf(right) && leftType.Comparable() && left == right
+}
+
+// drainObserver records an error from os/exec's output-copying goroutine.
+// Cmd.Wait drops that error when process exit itself fails.
+type drainObserver struct {
+	writer  io.Writer
+	mu      sync.Mutex
+	copyErr error
+}
+
+func (observer *drainObserver) Write(data []byte) (int, error) {
+	return observer.writer.Write(data)
+}
+
+func (observer *drainObserver) ReadFrom(reader io.Reader) (int64, error) {
+	written, err := io.Copy(observer.writer, reader)
+	if err != nil {
+		observer.mu.Lock()
+		observer.copyErr = err
+		observer.mu.Unlock()
+	}
+	return written, err
+}
+
+func (observer *drainObserver) err() error {
+	observer.mu.Lock()
+	defer observer.mu.Unlock()
+	return observer.copyErr
+}

internal/execution/command_context_process_unix_test.go

@@ -0,0 +1,100 @@
+//go:build !windows
+
+package execution
+
+import (
+	"errors"
+	"os"
+	"strconv"
+	"strings"
+	"syscall"
+	"testing"
+	"time"
+)
+
+type helperProcessOwner struct {
+	pidFile  string
+	stopFile string
+	pid      int
+	exited   bool
+}
+
+func ownHelperProcess(t *testing.T, pidFile, stopFile string) *helperProcessOwner {
+	t.Helper()
+	owner := &helperProcessOwner{pidFile: pidFile, stopFile: stopFile}
+	t.Cleanup(func() { owner.cleanup(t) })
+	return owner
+}
+
+func (owner *helperProcessOwner) waitReady(t *testing.T, timeout time.Duration) int {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	for {
+		data, err := os.ReadFile(owner.pidFile)
+		if err == nil {
+			pid, parseErr := strconv.Atoi(strings.TrimSpace(string(data)))
+			if parseErr == nil && pid > 0 {
+				owner.pid = pid
+				return pid
+			}
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("helper did not hand off a valid PID within %s", timeout)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+func (owner *helperProcessOwner) cleanup(t *testing.T) {
+	t.Helper()
+	if err := os.WriteFile(owner.stopFile, nil, 0o600); err != nil {
+		t.Errorf("request helper process stop: %v", err)
+	}
+	if owner.exited {
+		return
+	}
+	if owner.pid == 0 {
+		data, err := os.ReadFile(owner.pidFile)
+		if err == nil {
+			owner.pid, _ = strconv.Atoi(strings.TrimSpace(string(data)))
+		}
+	}
+	if owner.pid <= 0 {
+		return
+	}
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		err := syscall.Kill(owner.pid, syscall.Signal(0))
+		if errors.Is(err, syscall.ESRCH) {
+			owner.pid = 0
+			owner.exited = true
+			return
+		}
+		if err != nil {
+			t.Errorf("check helper process %d after cleanup: %v", owner.pid, err)
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Errorf("helper process %d survived cleanup", owner.pid)
+}
+
+func (owner *helperProcessOwner) awaitExit(t *testing.T) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for {
+		err := syscall.Kill(owner.pid, syscall.Signal(0))
+		if errors.Is(err, syscall.ESRCH) {
+			owner.pid = 0
+			owner.exited = true
+			return
+		}
+		if err != nil {
+			t.Fatalf("check helper process %d: %v", owner.pid, err)
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("helper process %d is still running after command cancellation", owner.pid)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}