21343491481

Committed 26 Jan 2026 01:41AM UTC coverage: 78.697% (+0.3%) from 78.401%

Build # 21343491481

Build Type

push

github

Committed by

web-flow

Commit Message

Fix: kill entire process group on context cancellation (#21)

* fix: kill entire process group on context cancellation

When ralphex exits (max iterations, Ctrl+C, error), child processes spawned
by claude (like test commands) were being orphaned because exec.CommandContext
only kills the direct child process.

Changes:
- Use process groups (Setpgid: true) for spawned commands
- On context cancellation, kill entire process group with syscall.Kill(-pid)
- Add test that verifies child processes are killed with parent

This prevents orphaned processes that could block subsequent ralphex runs
due to file lock contention.

* refactor: extract shared process group cleanup helper

Address code review feedback:
- Extract duplicated process group logic into shared procgroup.go helper
- Make Wait() idempotent using sync.Once (prevents panic on double-call)
- Use SIGTERM before SIGKILL for graceful shutdown (100ms delay)
- Log kill errors instead of silently ignoring them
- Move Unix-specific tests to procgroup_test.go with build tag
- Replace time.Sleep with require.Eventually for less flaky tests
- Add test for Wait() idempotency

* fix: address code review feedback for process group cleanup

- Add PID validation before attempting process group kill
- Replace fmt.Printf with log.Printf for proper logging
- Extract gracefulShutdownDelay constant (100ms)
- Always attempt SIGKILL even if SIGTERM fails
- Improve comments: clarify Wait() idempotency and setupProcessGroup ordering
- Capitalize doc comments per Go conventions

* fix: address codex review findings

Add pre-start context check to both Claude and Codex runners to avoid
spawning processes when the context is already canceled. This addresses
the valid finding that switching from exec.CommandContext to exec.Command
lost the pre-start cancellation behavior.

Note: The "Windows build break risk" finding was rejected as Windows
support is explicitly excluded in CLAUDE.md - the project only targets
Linux/macOS... (continued)

Coverage Stats

58 of 71 new or added lines in 3 files covered. (81.69%)

3310 of 4206 relevant lines covered (78.7%)

66.69 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.71

/pkg/executor/executor.go

// Package executor provides CLI execution for Claude and Codex tools.
package executor

import (
        "bufio"
        "context"
        "encoding/json"
        "fmt"
        "io"
        "os"
        "os/exec"
        "strings"
)

//go:generate moq -out mocks/command_runner.go -pkg mocks -skip-ensure -fmt goimports . CommandRunner

// Result holds execution result with output and detected signal.
type Result struct {
        Output string // accumulated text output
        Signal string // detected signal (COMPLETED, FAILED, etc.) or empty
        Error  error  // execution error if any
}

// CommandRunner abstracts command execution for testing.
// Returns an io.Reader for streaming output and a wait function for completion.
type CommandRunner interface {
        Run(ctx context.Context, name string, args ...string) (output io.Reader, wait func() error, err error)
}

// execClaudeRunner is the default command runner using os/exec.
type execClaudeRunner struct{}

func (r *execClaudeRunner) Run(ctx context.Context, name string, args ...string) (io.Reader, func() error, error) {
        // check context before starting to avoid spawning a process that will be immediately killed
        if err := ctx.Err(); err != nil {
                return nil, nil, fmt.Errorf("context already canceled: %w", err)
        }

        // use exec.Command (not CommandContext) because we handle cancellation ourselves
        // to ensure the entire process group is killed, not just the direct child
        cmd := exec.Command(name, args...) //nolint:noctx // intentional: we handle context cancellation via process group kill

        // filter out ANTHROPIC_API_KEY from environment (claude uses different auth)
        cmd.Env = filterEnv(os.Environ(), "ANTHROPIC_API_KEY")

        // create new process group so we can kill all descendants on cleanup
        setupProcessGroup(cmd)

        stdout, err := cmd.StdoutPipe()
        if err != nil {
                return nil, nil, fmt.Errorf("create stdout pipe: %w", err)
        }
        // merge stderr into stdout like python's stderr=subprocess.STDOUT
        cmd.Stderr = cmd.Stdout
        if err := cmd.Start(); err != nil {
                return nil, nil, fmt.Errorf("start command: %w", err)
        }

        // setup process group cleanup with graceful shutdown on context cancellation
        cleanup := newProcessGroupCleanup(cmd, ctx.Done())

        return stdout, cleanup.Wait, nil
}

// splitArgs splits a space-separated argument string into a slice.
// handles quoted strings (both single and double quotes).
func splitArgs(s string) []string {
        var args []string
        var current strings.Builder
        var inQuote rune
        var escaped bool

        for _, r := range s {
                if escaped {
                        current.WriteRune(r)
                        escaped = false
                        continue
                }

                if r == '\\' {
                        escaped = true
                        continue
                }

                if r == '"' || r == '\'' {
                        switch { //nolint:staticcheck // cannot use tagged switch because we compare with both inQuote and r
                        case inQuote == 0:
                                inQuote = r
                        case inQuote == r:
                                inQuote = 0
                        default:
                                current.WriteRune(r)
                        }
                        continue
                }

                if r == ' ' && inQuote == 0 {
                        if current.Len() > 0 {
                                args = append(args, current.String())
                                current.Reset()
                        }
                        continue
                }

                current.WriteRune(r)
        }

        if current.Len() > 0 {
                args = append(args, current.String())
        }

        return args
}

// filterEnv returns a copy of env with specified keys removed.
func filterEnv(env []string, keysToRemove ...string) []string {
        result := make([]string, 0, len(env))
        for _, e := range env {
                skip := false
                for _, key := range keysToRemove {
                        if strings.HasPrefix(e, key+"=") {
                                skip = true
                                break
                        }
                }
                if !skip {
                        result = append(result, e)
                }
        }
        return result
}

// streamEvent represents a JSON event from claude CLI stream output.
type streamEvent struct {
        Type    string `json:"type"`
        Message struct {
                Content []struct {
                        Type string `json:"type"`
                        Text string `json:"text"`
                } `json:"content"`
        } `json:"message"`
        ContentBlock struct {
                Type string `json:"type"`
                Text string `json:"text"`
        } `json:"content_block"`
        Delta struct {
                Type string `json:"type"`
                Text string `json:"text"`
        } `json:"delta"`
        Result json.RawMessage `json:"result"` // can be string or object with "output" field
}

// ClaudeExecutor runs claude CLI commands with streaming JSON parsing.
type ClaudeExecutor struct {
        Command       string            // command to execute, defaults to "claude"
        Args          string            // additional arguments (space-separated), defaults to standard args
        OutputHandler func(text string) // called for each text chunk, can be nil
        Debug         bool              // enable debug output
        cmdRunner     CommandRunner     // for testing, nil uses default
}

// Run executes claude CLI with the given prompt and parses streaming JSON output.
func (e *ClaudeExecutor) Run(ctx context.Context, prompt string) Result {
        cmd := e.Command
        if cmd == "" {
                cmd = "claude"
        }

        // build args from configured string or use defaults
        var args []string
        if e.Args != "" {
                args = splitArgs(e.Args)
        } else {
                args = []string{
                        "--dangerously-skip-permissions",
                        "--output-format", "stream-json",
                        "--verbose",
                }
        }
        args = append(args, "-p", prompt)

        runner := e.cmdRunner
        if runner == nil {
                runner = &execClaudeRunner{}
        }

        stdout, wait, err := runner.Run(ctx, cmd, args...)
        if err != nil {
                return Result{Error: err}
        }

        result := e.parseStream(stdout)

        if err := wait(); err != nil {
                // check if it was context cancellation
                if ctx.Err() != nil {
                        return Result{Output: result.Output, Signal: result.Signal, Error: ctx.Err()}
                }
                // non-zero exit might still have useful output
                if result.Output == "" {
                        return Result{Error: fmt.Errorf("claude exited with error: %w", err)}
                }
        }

        return result
}

// parseStream reads and parses the JSON stream from claude CLI.
func (e *ClaudeExecutor) parseStream(r io.Reader) Result {
        var output strings.Builder
        var signal string

        scanner := bufio.NewScanner(r)
        // increase buffer size for large JSON lines (16MB max for large diffs with parallel agents)
        buf := make([]byte, 0, 64*1024)
        scanner.Buffer(buf, 16*1024*1024)

        for scanner.Scan() {
                line := scanner.Text()
                if line == "" {
                        continue
                }

                var event streamEvent
                if err := json.Unmarshal([]byte(line), &event); err != nil {
                        // print non-JSON lines as-is
                        if e.Debug {
                                fmt.Printf("[debug] non-JSON line: %s\n", line)
                        }
                        output.WriteString(line)
                        output.WriteString("\n")
                        if e.OutputHandler != nil {
                                e.OutputHandler(line + "\n")
                        }
                        continue
                }

                text := e.extractText(&event)
                if text != "" {
                        output.WriteString(text)
                        if e.OutputHandler != nil {
                                e.OutputHandler(text)
                        }

                        // check for signals in text
                        if sig := detectSignal(text); sig != "" {
                                signal = sig
                        }
                }
        }

        if err := scanner.Err(); err != nil {
                return Result{Output: output.String(), Signal: signal, Error: fmt.Errorf("stream read: %w", err)}
        }

        return Result{Output: output.String(), Signal: signal}
}

// extractText extracts text content from various event types.
func (e *ClaudeExecutor) extractText(event *streamEvent) string {
        switch event.Type {
        case "assistant":
                // assistant events contain message.content array with text blocks
                var texts []string
                for _, c := range event.Message.Content {
                        if c.Type == "text" && c.Text != "" {
                                texts = append(texts, c.Text)
                        }
                }
                return strings.Join(texts, "")
        case "content_block_delta":
                if event.Delta.Type == "text_delta" {
                        return event.Delta.Text
                }
        case "message_stop":
                // check final message content
                for _, c := range event.Message.Content {
                        if c.Type == "text" {
                                return c.Text
                        }
                }
        case "result":
                // result can be a string or object with "output" field
                if len(event.Result) == 0 {
                        return ""
                }
                // try as string first (session summary format)
                var resultStr string
                if err := json.Unmarshal(event.Result, &resultStr); err == nil {
                        return "" // skip session summary - content already streamed
                }
                // try as object with output field
                var resultObj struct {
                        Output string `json:"output"`
                }
                if err := json.Unmarshal(event.Result, &resultObj); err == nil {
                        return resultObj.Output
                }
        }
        return ""
}

// detectSignal checks text for completion signals.
// Looks for <<<RALPHEX:...>>> format signals.
func detectSignal(text string) string {
        signals := []string{
                "<<<RALPHEX:ALL_TASKS_DONE>>>",
                "<<<RALPHEX:TASK_FAILED>>>",
                "<<<RALPHEX:REVIEW_DONE>>>",
                "<<<RALPHEX:CODEX_REVIEW_DONE>>>",
                "<<<RALPHEX:PLAN_READY>>>",
        }
        for _, sig := range signals {
                if strings.Contains(text, sig) {
                        return sig
                }
        }
        return ""
}

1	// Package executor provides CLI execution for Claude and Codex tools.
2	package executor
3
4	import (
5	"bufio"
6	"context"
7	"encoding/json"
8	"fmt"
9	"io"
10	"os"
11	"os/exec"
12	"strings"
13	)
14
15	//go:generate moq -out mocks/command_runner.go -pkg mocks -skip-ensure -fmt goimports . CommandRunner
16
17	// Result holds execution result with output and detected signal.
18	type Result struct {
19	Output string // accumulated text output
20	Signal string // detected signal (COMPLETED, FAILED, etc.) or empty
21	Error error // execution error if any
22	}
23
24	// CommandRunner abstracts command execution for testing.
25	// Returns an io.Reader for streaming output and a wait function for completion.
26	type CommandRunner interface {
27	Run(ctx context.Context, name string, args ...string) (output io.Reader, wait func() error, err error)
28	}
29
30	// execClaudeRunner is the default command runner using os/exec.
31	type execClaudeRunner struct{}
32
33	func (r *execClaudeRunner) Run(ctx context.Context, name string, args ...string) (io.Reader, func() error, error) {	2✔
34	// check context before starting to avoid spawning a process that will be immediately killed	2✔
35	if err := ctx.Err(); err != nil {	2✔
NEW 36	return nil, nil, fmt.Errorf("context already canceled: %w", err)	×
NEW 37	}	×
38
39	// use exec.Command (not CommandContext) because we handle cancellation ourselves
40	// to ensure the entire process group is killed, not just the direct child
41	cmd := exec.Command(name, args...) //nolint:noctx // intentional: we handle context cancellation via process group kill	2✔
42		2✔
43	// filter out ANTHROPIC_API_KEY from environment (claude uses different auth)	2✔
44	cmd.Env = filterEnv(os.Environ(), "ANTHROPIC_API_KEY")	2✔
45		2✔
46	// create new process group so we can kill all descendants on cleanup	2✔
47	setupProcessGroup(cmd)	2✔
48		2✔
49	stdout, err := cmd.StdoutPipe()	2✔
50	if err != nil {	2✔
51	return nil, nil, fmt.Errorf("create stdout pipe: %w", err)	×
52	}	×
53	// merge stderr into stdout like python's stderr=subprocess.STDOUT
54	cmd.Stderr = cmd.Stdout	2✔
55	if err := cmd.Start(); err != nil {	2✔
56	return nil, nil, fmt.Errorf("start command: %w", err)	×
57	}	×
58
59	// setup process group cleanup with graceful shutdown on context cancellation
60	cleanup := newProcessGroupCleanup(cmd, ctx.Done())	2✔
61		2✔
62	return stdout, cleanup.Wait, nil	2✔
63	}
64
65	// splitArgs splits a space-separated argument string into a slice.
66	// handles quoted strings (both single and double quotes).
67	func splitArgs(s string) []string {	11✔
68	var args []string	11✔
69	var current strings.Builder	11✔
70	var inQuote rune	11✔
71	var escaped bool	11✔
72		11✔
73	for _, r := range s {	252✔
74	if escaped {	243✔
75	current.WriteRune(r)	2✔
76	escaped = false	2✔
77	continue	2✔
78	}
79
80	if r == '\\' {	241✔
81	escaped = true	2✔
82	continue	2✔
83	}
84
85	if r == '"' \|\| r == '\'' {	245✔
86	switch { //nolint:staticcheck // cannot use tagged switch because we compare with both inQuote and r	8✔
87	case inQuote == 0:	4✔
88	inQuote = r	4✔
89	case inQuote == r:	4✔
90	inQuote = 0	4✔
91	default:	×
92	current.WriteRune(r)	×
93	}
94	continue	8✔
95	}
96
97	if r == ' ' && inQuote == 0 {	249✔
98	if current.Len() > 0 {	35✔
99	args = append(args, current.String())	15✔
100	current.Reset()	15✔
101	}	15✔
102	continue	20✔
103	}
104
105	current.WriteRune(r)	209✔
106	}
107
108	if current.Len() > 0 {	20✔
109	args = append(args, current.String())	9✔
110	}	9✔
111
112	return args	11✔
113	}
114
115	// filterEnv returns a copy of env with specified keys removed.
116	func filterEnv(env []string, keysToRemove ...string) []string {	7✔
117	result := make([]string, 0, len(env))	7✔
118	for _, e := range env {	239✔
119	skip := false	232✔
120	for _, key := range keysToRemove {	466✔
121	if strings.HasPrefix(e, key+"=") {	238✔
122	skip = true	4✔
123	break	4✔
124	}
125	}
126	if !skip {	460✔
127	result = append(result, e)	228✔
128	}	228✔
129	}
130	return result	7✔
131	}
132
133	// streamEvent represents a JSON event from claude CLI stream output.
134	type streamEvent struct {
135	Type string `json:"type"`
136	Message struct {
137	Content []struct {
138	Type string `json:"type"`
139	Text string `json:"text"`
140	} `json:"content"`
141	} `json:"message"`
142	ContentBlock struct {
143	Type string `json:"type"`
144	Text string `json:"text"`
145	} `json:"content_block"`
146	Delta struct {
147	Type string `json:"type"`
148	Text string `json:"text"`
149	} `json:"delta"`
150	Result json.RawMessage `json:"result"` // can be string or object with "output" field
151	}
152
153	// ClaudeExecutor runs claude CLI commands with streaming JSON parsing.
154	type ClaudeExecutor struct {
155	Command string // command to execute, defaults to "claude"
156	Args string // additional arguments (space-separated), defaults to standard args
157	OutputHandler func(text string) // called for each text chunk, can be nil
158	Debug bool // enable debug output
159	cmdRunner CommandRunner // for testing, nil uses default
160	}
161
162	// Run executes claude CLI with the given prompt and parses streaming JSON output.
163	func (e *ClaudeExecutor) Run(ctx context.Context, prompt string) Result {	9✔
164	cmd := e.Command	9✔
165	if cmd == "" {	16✔
166	cmd = "claude"	7✔
167	}	7✔
168
169	// build args from configured string or use defaults
170	var args []string	9✔
171	if e.Args != "" {	11✔
172	args = splitArgs(e.Args)	2✔
173	} else {	9✔
174	args = []string{	7✔
175	"--dangerously-skip-permissions",	7✔
176	"--output-format", "stream-json",	7✔
177	"--verbose",	7✔
178	}	7✔
179	}	7✔
180	args = append(args, "-p", prompt)	9✔
181		9✔
182	runner := e.cmdRunner	9✔
183	if runner == nil {	9✔
184	runner = &execClaudeRunner{}	×
185	}	×
186
187	stdout, wait, err := runner.Run(ctx, cmd, args...)	9✔
188	if err != nil {	10✔
189	return Result{Error: err}	1✔
190	}	1✔
191
192	result := e.parseStream(stdout)	8✔
193		8✔
194	if err := wait(); err != nil {	11✔
195	// check if it was context cancellation	3✔
196	if ctx.Err() != nil {	4✔
197	return Result{Output: result.Output, Signal: result.Signal, Error: ctx.Err()}	1✔
198	}	1✔
199	// non-zero exit might still have useful output
200	if result.Output == "" {	3✔
201	return Result{Error: fmt.Errorf("claude exited with error: %w", err)}	1✔
202	}	1✔
203	}
204
205	return result	6✔
206	}
207
208	// parseStream reads and parses the JSON stream from claude CLI.
209	func (e *ClaudeExecutor) parseStream(r io.Reader) Result {	27✔
210	var output strings.Builder	27✔
211	var signal string	27✔
212		27✔
213	scanner := bufio.NewScanner(r)	27✔
214	// increase buffer size for large JSON lines (16MB max for large diffs with parallel agents)	27✔
215	buf := make([]byte, 0, 64*1024)	27✔
216	scanner.Buffer(buf, 1610241024)	27✔
217		27✔
218	for scanner.Scan() {	64✔
219	line := scanner.Text()	37✔
220	if line == "" {	40✔
221	continue	3✔
222	}
223
224	var event streamEvent	34✔
225	if err := json.Unmarshal([]byte(line), &event); err != nil {	36✔
226	// print non-JSON lines as-is	2✔
227	if e.Debug {	3✔
228	fmt.Printf("[debug] non-JSON line: %s\n", line)	1✔
229	}	1✔
230	output.WriteString(line)	2✔
231	output.WriteString("\n")	2✔
232	if e.OutputHandler != nil {	2✔
233	e.OutputHandler(line + "\n")	×
234	}	×
235	continue	2✔
236	}
237
238	text := e.extractText(&event)	32✔
239	if text != "" {	63✔
240	output.WriteString(text)	31✔
241	if e.OutputHandler != nil {	35✔
242	e.OutputHandler(text)	4✔
243	}	4✔
244
245	// check for signals in text
246	if sig := detectSignal(text); sig != "" {	37✔
247	signal = sig	6✔
248	}	6✔
249	}
250	}
251
252	if err := scanner.Err(); err != nil {	27✔
253	return Result{Output: output.String(), Signal: signal, Error: fmt.Errorf("stream read: %w", err)}	×
254	}	×
255
256	return Result{Output: output.String(), Signal: signal}	27✔
257	}
258
259	// extractText extracts text content from various event types.
260	func (e ClaudeExecutor) extractText(event streamEvent) string {	43✔
261	switch event.Type {	43✔
262	case "assistant":	4✔
263	// assistant events contain message.content array with text blocks	4✔
264	var texts []string	4✔
265	for _, c := range event.Message.Content {	8✔
266	if c.Type == "text" && c.Text != "" {	8✔
267	texts = append(texts, c.Text)	4✔
268	}	4✔
269	}
270	return strings.Join(texts, "")	4✔
271	case "content_block_delta":	31✔
272	if event.Delta.Type == "text_delta" {	61✔
273	return event.Delta.Text	30✔
274	}	30✔
275	case "message_stop":	3✔
276	// check final message content	3✔
277	for _, c := range event.Message.Content {	5✔
278	if c.Type == "text" {	3✔
279	return c.Text	1✔
280	}	1✔
281	}
282	case "result":	3✔
283	// result can be a string or object with "output" field	3✔
284	if len(event.Result) == 0 {	3✔
285	return ""	×
286	}	×
287	// try as string first (session summary format)
288	var resultStr string	3✔
289	if err := json.Unmarshal(event.Result, &resultStr); err == nil {	4✔
290	return "" // skip session summary - content already streamed	1✔
291	}	1✔
292	// try as object with output field
293	var resultObj struct {	2✔
294	Output string `json:"output"`	2✔
295	}	2✔
296	if err := json.Unmarshal(event.Result, &resultObj); err == nil {	4✔
297	return resultObj.Output	2✔
298	}	2✔
299	}
300	return ""	5✔
301	}
302
303	// detectSignal checks text for completion signals.
304	// Looks for <<<RALPHEX:...>>> format signals.
305	func detectSignal(text string) string {	47✔
306	signals := []string{	47✔
307	"<<<RALPHEX:ALL_TASKS_DONE>>>",	47✔
308	"<<<RALPHEX:TASK_FAILED>>>",	47✔
309	"<<<RALPHEX:REVIEW_DONE>>>",	47✔
310	"<<<RALPHEX:CODEX_REVIEW_DONE>>>",	47✔
311	"<<<RALPHEX:PLAN_READY>>>",	47✔
312	}	47✔
313	for _, sig := range signals {	256✔
314	if strings.Contains(text, sig) {	221✔
315	return sig	12✔
316	}	12✔
317	}
318	return ""	35✔
319	}

umputun / ralphex / 21343491481

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous