fix(fixtures): honor timeout and kill leftover processes

fixtures/ansi_capture.go

@@ -1,7 +1,9 @@
 package fixtures
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
 	osExec "os/exec"
@@ -10,6 +12,7 @@ import (
 	"time"
 
 	"github.com/creack/pty"
+	"github.com/shirou/gopsutil/v3/process"
 
 	"github.com/flanksource/gavel/fixtures/record"
 )
@@ -28,6 +31,7 @@ const (
 // by settled snapshots. SnapshotInterval controls how often the live viewport
 // is appended to the snapshot timeline.
 type CaptureOptions struct {
+	Context          context.Context
 	Width, Height    int
 	SnapshotInterval time.Duration
 	Command          []string
@@ -141,6 +145,13 @@ func CaptureANSI(opts CaptureOptions) (*Capture, error) {
 		interval = 100 * time.Millisecond
 	}
 
+	ctx := opts.Context
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, fmt.Errorf("ansi capture: %w", err)
+	}
 	cmd := osExec.Command(opts.Command[0], opts.Command[1:]...)
 	cmd.Dir = opts.Dir
 	cmd.Env = append(os.Environ(), opts.Env...)
@@ -152,6 +163,20 @@ func CaptureANSI(opts CaptureOptions) (*Capture, error) {
 		return nil, fmt.Errorf("ansi capture: start pty: %w", err)
 	}
 	defer ptmx.Close()
+	// Keep the watcher alive through cmd.Wait(), not just PTY EOF. A child can
+	// close its terminal then sleep; read returns while the process is still
+	// running, and the fixture timeout must still be able to kill it.
+	waitDone := make(chan struct{})
+	go func() {
+		select {
+		case <-ctx.Done():
+			// Kill before closing the PTY so descendants that created their own
+			// process groups remain attached long enough to be discovered.
+			_ = cancelCaptureProcess(cmd.Process)
+			_ = ptmx.Close()
+		case <-waitDone:
+		}
+	}()
 
 	var (
 		mu        sync.Mutex
@@ -223,7 +248,9 @@ func CaptureANSI(opts CaptureOptions) (*Capture, error) {
 	wg.Wait()
 
 	exitCode := 0
-	if werr := cmd.Wait(); werr != nil {
+	werr := cmd.Wait()
+	close(waitDone)
+	if werr != nil {
 		ee, ok := werr.(*osExec.ExitError)
 		if !ok {
 			return nil, fmt.Errorf("ansi capture: wait for %q: %w", opts.Command[0], werr)
@@ -252,6 +279,50 @@ func CaptureANSI(opts CaptureOptions) (*Capture, error) {
 	return capture, nil
 }
 
+func captureProcessDescendants(root int32) ([]*process.Process, error) {
+	all, err := process.Processes()
+	if err != nil {
+		return nil, err
+	}
+	children := make(map[int32][]*process.Process)
+	for _, candidate := range all {
+		parent, err := candidate.Ppid()
+		if err == nil {
+			children[parent] = append(children[parent], candidate)
+		}
+	}
+
+	var descendants []*process.Process
+	parents := []int32{root}
+	seen := map[int32]bool{root: true}
+	for len(parents) > 0 {
+		parent := parents[0]
+		parents = parents[1:]
+		for _, child := range children[parent] {
+			if seen[child.Pid] {
+				continue
+			}
+			seen[child.Pid] = true
+			descendants = append(descendants, child)
+			parents = append(parents, child.Pid)
+		}
+	}
+	return descendants, nil
+}
+
+func killCaptureDescendants(root int) error {
+	descendants, err := captureProcessDescendants(int32(root))
+	failures := []error{err}
+	// Leaves first prevents a killed parent from orphaning children before they
+	// receive the cancellation signal.
+	for i := len(descendants) - 1; i >= 0; i-- {
+		if err := descendants[i].Kill(); err != nil && !errors.Is(err, os.ErrProcessDone) {
+			failures = append(failures, err)
+		}
+	}
+	return errors.Join(failures...)
+}
+
 func ensureEnv(env []string, key, val string) []string {
 	prefix := key + "="
 	for _, e := range env {

fixtures/ansi_capture_cancel_other.go

@@ -0,0 +1,15 @@
+//go:build !unix
+
+package fixtures
+
+import (
+	"errors"
+	"os"
+)
+
+func cancelCaptureProcess(process *os.Process) error {
+	if process == nil {
+		return os.ErrProcessDone
+	}
+	return errors.Join(killCaptureDescendants(process.Pid), process.Kill())
+}

fixtures/ansi_capture_cancel_unix.go

@@ -0,0 +1,25 @@
+//go:build unix
+
+package fixtures
+
+import (
+	"errors"
+	"os"
+	"syscall"
+)
+
+func cancelCaptureProcess(process *os.Process) error {
+	if process == nil {
+		return os.ErrProcessDone
+	}
+	stopErr := syscall.Kill(-process.Pid, syscall.SIGSTOP)
+	if errors.Is(stopErr, syscall.ESRCH) {
+		stopErr = nil
+	}
+	descendantsErr := killCaptureDescendants(process.Pid)
+	groupErr := syscall.Kill(-process.Pid, syscall.SIGKILL)
+	if errors.Is(groupErr, syscall.ESRCH) {
+		groupErr = nil
+	}
+	return errors.Join(stopErr, descendantsErr, groupErr)
+}

fixtures/parser.go

@@ -7,6 +7,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/goccy/go-yaml"
 
@@ -144,6 +145,14 @@ func parseTableRow(headers, values []string) (*FixtureNode, error) {
 			fixture.Expected.Output = value
 		case "cel validation", "cel", "validation", "expr":
 			fixture.Expected.CEL = value
+		case "timeout":
+			if value != "" && value != "-" {
+				timeout, err := parseFixtureDuration(value)
+				if err != nil {
+					return nil, fmt.Errorf("fixture %q: invalid timeout %q: %w", fixture.Name, value, err)
+				}
+				fixture.Expected.Timeout = &timeout
+			}
 		default:
 			if value != "" {
 				if fixture.Expected.Properties == nil {
@@ -165,6 +174,13 @@ func parseTableRow(headers, values []string) (*FixtureNode, error) {
 	}, nil
 }
 
+func parseFixtureDuration(value string) (time.Duration, error) {
+	if seconds, err := strconv.Atoi(value); err == nil {
+		return time.Duration(seconds) * time.Second, nil
+	}
+	return time.ParseDuration(value)
+}
+
 // parseFrontMatter extracts YAML front-matter from a markdown file
 func parseFrontMatter(file *os.File) (*FrontMatter, string, error) {
 	scanner := bufio.NewScanner(file)

fixtures/parser_ast.go

@@ -490,7 +490,6 @@ func buildFixtureFromCommand(cmd *commandBlockBuilder, frontMatter *FrontMatter,
 			CWD      string         `yaml:"cwd"`
 			ExitCode *int           `yaml:"exitCode"`
 			Env      map[string]any `yaml:"env"`
-			Timeout  string         `yaml:"timeout"`
 			Terminal string         `yaml:"terminal"`
 			OS       string         `yaml:"os"`
 			Arch     string         `yaml:"arch"`
@@ -513,6 +512,21 @@ func buildFixtureFromCommand(cmd *commandBlockBuilder, frontMatter *FrontMatter,
 		if recordOnly.Record != nil {
 			fixture.Record = recordOnly.Record
 		}
+		// Timeout must not disappear when the legacy tolerant decode below
+		// rejects another field.
+		var timeoutOnly struct {
+			Timeout string `yaml:"timeout"`
+		}
+		if err := yaml.Unmarshal([]byte(cmd.frontmatter), &timeoutOnly); err != nil {
+			return nil, fmt.Errorf("%s: invalid timeout configuration: %w", cmd.name, err)
+		}
+		if timeoutOnly.Timeout != "" {
+			timeout, err := parseFixtureDuration(timeoutOnly.Timeout)
+			if err != nil {
+				return nil, fmt.Errorf("command %q: invalid timeout %q: %w", cmd.name, timeoutOnly.Timeout, err)
+			}
+			fixture.Expected.Timeout = &timeout
+		}
 
 		if err := yaml.Unmarshal([]byte(cmd.frontmatter), &cmdFrontMatter); err == nil {
 			if cmdFrontMatter.Setup != nil {

fixtures/runner.go

@@ -318,7 +318,7 @@ func (r *Runner) executeFixtures() (*FixtureGroup, error) {
 					return result, progressErr
 				}
 				return result, err
-			}, clicky.WithTaskTimeout(2*time.Minute))
+			}, clicky.WithTaskTimeout(fixtureTimeout(*node.Test)))
 		}
 	})
 
@@ -358,6 +358,16 @@ func (r *Runner) executeFixtures() (*FixtureGroup, error) {
 	return results, nil
 }
 
+func fixtureTimeout(fixture FixtureTest) time.Duration {
+	if fixture.Expected.Timeout != nil {
+		return *fixture.Expected.Timeout
+	}
+	if fixture.Timeout != nil {
+		return *fixture.Timeout
+	}
+	return 2 * time.Minute
+}
+
 func (r *Runner) executionSteps() []ExecutionStep {
 	var steps []ExecutionStep
 	if r.hasSetup() {