21493195690

Committed 29 Jan 2026 08:13PM UTC coverage: 79.83% (+0.2%) from 79.632%

Build # 21493195690

Build Type

push

github

Committed by

umputun

Commit Message

docs: document error pattern detection feature

Update CLAUDE.md and README.md with error pattern detection configuration.
Move completed plan to docs/plans/completed/.

Run Details

4049 of 5072 relevant lines covered (79.83%)

123.57 hits per line

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

92.59

/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

// maxScannerBuffer is the maximum buffer size for bufio.Scanner.
// set to 64MB to handle large outputs (e.g., diffs of large JSON files).
const maxScannerBuffer = 64 * 1024 * 1024

// 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
}

// PatternMatchError is returned when a configured error pattern is detected in output.
type PatternMatchError struct {
        Pattern string // the pattern that matched
        HelpCmd string // command to run for more information (e.g., "claude /usage")
}

func (e *PatternMatchError) Error() string {
        return fmt.Sprintf("detected error pattern: %q", e.Pattern)
}

// 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
        ErrorPatterns []string          // patterns to detect in output (e.g., rate limit messages)
        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)}
                }
        }

        // check for error patterns in output
        if pattern := checkErrorPatterns(result.Output, e.ErrorPatterns); pattern != "" {
                return Result{
                        Output: result.Output,
                        Signal: result.Signal,
                        Error:  &PatternMatchError{Pattern: pattern, HelpCmd: "claude /usage"},
                }
        }

        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 (large diffs with parallel agents)
        buf := make([]byte, 0, 64*1024)
        scanner.Buffer(buf, maxScannerBuffer)

        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 ""
}

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

umputun / ralphex / 21493195690

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