25115074240

Committed 29 Apr 2026 02:30PM UTC coverage: 82.853% (+0.4%) from 82.426%

Build # 25115074240

Build Type

Pull #309

github

Committed by

bronislav

Commit Message

fix: address code review findings

- document {{TASK_HEADER_PATTERNS}} in CLAUDE.md and README.md variable lists
- update CLAUDE.md 'Plan format' key pattern to reflect configurable headers
  and the removal of the package-level taskHeaderPattern regex
- rename TestDefaultPatternsMatchLegacyRegex and rewrite its comment to be
  honest: the template-driven compiler is stricter than the legacy regex
  about whitespace, so the test only covers canonical inputs
- exclude gosec G704 (SSRF taint) in *_test.go files; false positive on
  httptest server URLs in pkg/web/watcher_test.go

Pull Request Pull Request #309: feat: configurable task header patterns

Coverage Stats

367 of 399 new or added lines in 13 files covered. (91.98%)

1 existing line in 1 file now uncovered.

7011 of 8462 relevant lines covered (82.85%)

227.55 hits per line

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

91.84

/pkg/web/session_progress.go

package web

import (
        "bufio"
        "errors"
        "fmt"
        "io"
        "log"
        "os"
        "strconv"
        "strings"
        "time"

        "github.com/umputun/ralphex/pkg/status"
)

// ParseProgressHeader reads the header section of a progress file and extracts metadata.
// the header format is:
//
//        # Ralphex Progress Log
//        Plan: path/to/plan.md
//        Branch: feature-branch
//        Mode: full
//        Started: 2026-01-22 10:30:00
//        ------------------------------------------------------------
//
// the second return value reports whether the terminating separator line was
// observed, which means the header is fully written. during a truncate+rewrite
// the header is emitted across several writes, so a mid-write read can return
// incomplete metadata (zero StartTime, missing fields); callers should use the
// complete flag to decide whether to trust the parsed metadata.
//
// after the header separator, the function also scans the remainder of the
// file for TaskHeaderPatterns: lines emitted by the progress writer next to
// a "--- restarted at ... ---" marker. retried runs with new patterns re-emit
// the key next to the restart marker so the dashboard picks up current
// patterns instead of the stale value stored in the original header. the
// last TaskHeaderPatterns occurrence in the file wins, so restarts that
// change patterns mid-file are reflected correctly.
func ParseProgressHeader(path string) (meta SessionMetadata, complete bool, err error) {
        f, err := os.Open(path) //nolint:gosec // path from user-controlled glob pattern, acceptable for session discovery
        if err != nil {
                return SessionMetadata{}, false, fmt.Errorf("open file: %w", err)
        }
        defer f.Close()

        reader := bufio.NewReader(f)

        for {
                line, readErr := reader.ReadString('\n')
                line = trimLineEnding(line)

                // stop at separator line; check readErr even when breaking
                if line != "" && strings.HasPrefix(line, "---") {
                        if readErr != nil && !errors.Is(readErr, io.EOF) {
                                return SessionMetadata{}, false, fmt.Errorf("read file: %w", readErr)
                        }
                        complete = true
                        break
                }

                // parse key-value pairs (process line before checking error,
                // as ReadString may return partial data alongside an error)
                applyHeaderField(line, &meta)

                if readErr != nil {
                        if !errors.Is(readErr, io.EOF) {
                                return SessionMetadata{}, false, fmt.Errorf("read file: %w", readErr)
                        }
                        return meta, complete, nil // EOF before separator
                }
        }

        // scan the remainder for restart-marker TaskHeaderPatterns overrides.
        // last occurrence wins so the freshest run's patterns are used.
        for {
                line, readErr := reader.ReadString('\n')
                line = trimLineEnding(line)
                if val, found := strings.CutPrefix(line, "TaskHeaderPatterns: "); found {
                        if patterns := parsePatternList(val); len(patterns) > 0 {
                                meta.TaskHeaderPatterns = patterns
                        }
                }
                if readErr != nil {
                        if !errors.Is(readErr, io.EOF) {
                                return meta, complete, fmt.Errorf("read file: %w", readErr)
                        }
                        return meta, complete, nil
                }
        }
}

// applyHeaderField parses a single header line (e.g. "Plan: ...") and updates
// the metadata in place. unknown or unmatched lines are ignored.
func applyHeaderField(line string, meta *SessionMetadata) {
        if val, found := strings.CutPrefix(line, "Plan: "); found {
                meta.PlanPath = val
                return
        }
        if val, found := strings.CutPrefix(line, "Branch: "); found {
                meta.Branch = val
                return
        }
        if val, found := strings.CutPrefix(line, "Mode: "); found {
                meta.Mode = val
                return
        }
        if val, found := strings.CutPrefix(line, "Started: "); found {
                // header timestamps are written in local time without a zone offset
                if t, parseErr := time.ParseInLocation("2006-01-02 15:04:05", val, time.Local); parseErr == nil {
                        meta.StartTime = t
                }
                return
        }
        if val, found := strings.CutPrefix(line, "TaskHeaderPatterns: "); found {
                if patterns := parsePatternList(val); len(patterns) > 0 {
                        meta.TaskHeaderPatterns = patterns
                }
        }
}

// parsePatternList splits a comma-separated header value into a trimmed,
// non-empty template list. the writer (pkg/progress writeHeader) rejects
// templates containing commas, so this split is lossless.
func parsePatternList(val string) []string {
        parts := strings.Split(val, ",")
        out := make([]string, 0, len(parts))
        for _, p := range parts {
                if trimmed := strings.TrimSpace(p); trimmed != "" {
                        out = append(out, trimmed)
                }
        }
        return out
}

// loadProgressFileIntoSession reads a progress file and publishes events to the session's SSE server.
// used for completed sessions that were discovered after they finished.
// errors are silently ignored since this is best-effort loading.
// records the total bytes consumed into session.lastOffset so a later Reactivate()
// can resume tailing from after the loaded content instead of re-emitting it.
func (m *SessionManager) loadProgressFileIntoSession(path string, session *Session) {
        f, err := os.Open(path) //nolint:gosec // path from user-controlled glob pattern, acceptable for session discovery
        if err != nil {
                return
        }
        defer f.Close()

        reader := bufio.NewReader(f)
        inHeader := true
        phase := status.PhaseTask
        var pendingSection string // section header waiting for first timestamped event
        var bytesRead int64

        for {
                line, readErr := reader.ReadString('\n')
                // a partial trailing line (no newline delimiter) means the writer is
                // mid-write — the flock-race recovery path is the realistic case where
                // loader is invoked on a still-being-written file. counting and
                // publishing the partial would advance lastOffset past the partial
                // bytes; a later Reactivate would then resume reading the suffix of
                // the writer's eventual completed line as if it were a separate event,
                // reintroducing the corruption this PR is meant to eliminate.
                if readErr != nil && line != "" {
                        break
                }
                // count raw bytes including the delimiter before trimming, so LF, CRLF,
                // and the final empty read all count correctly
                bytesRead += int64(len(line))
                line = trimLineEnding(line)

                if line != "" {
                        var parsed ParsedLine
                        parsed, inHeader = parseProgressLine(line, inHeader)
                        phase, pendingSection = m.processProgressLine(session, parsed, phase, pendingSection)
                }

                if readErr != nil {
                        break // EOF or read error; best-effort loading, errors silently ignored
                }
        }

        if pendingSection != "" {
                m.emitPendingSection(session, pendingSection, phase, time.Now())
        }

        session.setLastOffset(bytesRead)
        // record the phase the parser ended on so a later Reactivate resumes with
        // the correct phase rather than the tailer's default (PhaseTask).
        session.setLastPhase(phase)
}

// processProgressLine handles a single parsed progress line,
// updating phase and pendingSection state and publishing events as needed.
func (m *SessionManager) processProgressLine(session *Session, parsed ParsedLine,
        phase status.Phase, pendingSection string) (status.Phase, string) {
        switch parsed.Type {
        case ParsedLineSkip:
                return phase, pendingSection
        case ParsedLineSection:
                if pendingSection != "" {
                        m.emitPendingSection(session, pendingSection, phase, time.Now())
                }
                phase = parsed.Phase
                // defer emitting section until we see a timestamped event
                return phase, parsed.Section
        case ParsedLineTimestamp:
                // emit pending section with this event's timestamp (for accurate durations)
                if pendingSection != "" {
                        m.emitPendingSection(session, pendingSection, phase, parsed.Timestamp)
                        pendingSection = ""
                }
                event := Event{
                        Type:      parsed.EventType,
                        Phase:     phase,
                        Text:      parsed.Text,
                        Timestamp: parsed.Timestamp,
                        Signal:    parsed.Signal,
                }
                if event.Type == EventTypeOutput {
                        if stats, ok := parseDiffStats(event.Text); ok {
                                session.SetDiffStats(stats)
                        }
                }
                _ = session.Publish(event)
        case ParsedLinePlain:
                _ = session.Publish(Event{
                        Type:      EventTypeOutput,
                        Phase:     phase,
                        Text:      parsed.Text,
                        Timestamp: time.Now(),
                })
        }
        return phase, pendingSection
}

// emitPendingSection publishes section and task_start events for a pending section.
// task_start is emitted before section for task iteration sections.
func (m *SessionManager) emitPendingSection(session *Session, sectionName string, phase status.Phase, ts time.Time) {
        // emit task_start event for task iteration sections
        if matches := taskIterationRegex.FindStringSubmatch(sectionName); matches != nil {
                taskNum, err := strconv.Atoi(matches[1])
                if err != nil {
                        // log parse error but continue - section will still be emitted
                        log.Printf("[WARN] failed to parse task number from section %q: %v", sectionName, err)
                } else {
                        if err := session.Publish(Event{
                                Type:      EventTypeTaskStart,
                                Phase:     phase,
                                TaskNum:   taskNum,
                                Text:      sectionName,
                                Timestamp: ts,
                        }); err != nil {
                                log.Printf("[WARN] failed to publish task_start event: %v", err)
                        }
                }
        }

        if err := session.Publish(Event{
                Type:      EventTypeSection,
                Phase:     phase,
                Section:   sectionName,
                Text:      sectionName,
                Timestamp: ts,
        }); err != nil {
                log.Printf("[WARN] failed to publish section event: %v", err)
        }
}

// phaseFromSection determines the phase from a section name.
// checks "codex"/"custom" before "review" because external review sections should be PhaseCodex.
func phaseFromSection(name string) status.Phase {
        nameLower := strings.ToLower(name)
        switch {
        case strings.Contains(nameLower, "task"):
                return status.PhaseTask
        case strings.Contains(nameLower, "codex"), strings.Contains(nameLower, "custom"):
                return status.PhaseCodex
        case strings.Contains(nameLower, "review"):
                return status.PhaseReview
        case strings.Contains(nameLower, "claude-eval"), strings.Contains(nameLower, "claude eval"):
                return status.PhaseClaudeEval
        default:
                return status.PhaseTask
        }
}

// trimLineEnding removes trailing line ending to match bufio.ScanLines semantics:
// strips \n, \r\n, or a bare trailing \r (which ScanLines drops via dropCR at EOF).
// unlike strings.TrimRight("\r\n"), this preserves embedded \r characters in content.
func trimLineEnding(line string) string {
        n := len(line)
        if n > 0 && line[n-1] == '\n' {
                n--
        }
        if n > 0 && line[n-1] == '\r' {
                n--
        }
        return line[:n]
}

1	package web
2
3	import (
4	"bufio"
5	"errors"
6	"fmt"
7	"io"
8	"log"
9	"os"
10	"strconv"
11	"strings"
12	"time"
13
14	"github.com/umputun/ralphex/pkg/status"
15	)
16
17	// ParseProgressHeader reads the header section of a progress file and extracts metadata.
18	// the header format is:
19	//
20	// # Ralphex Progress Log
21	// Plan: path/to/plan.md
22	// Branch: feature-branch
23	// Mode: full
24	// Started: 2026-01-22 10:30:00
25	// ------------------------------------------------------------
26	//
27	// the second return value reports whether the terminating separator line was
28	// observed, which means the header is fully written. during a truncate+rewrite
29	// the header is emitted across several writes, so a mid-write read can return
30	// incomplete metadata (zero StartTime, missing fields); callers should use the
31	// complete flag to decide whether to trust the parsed metadata.
32	//
33	// after the header separator, the function also scans the remainder of the
34	// file for TaskHeaderPatterns: lines emitted by the progress writer next to
35	// a "--- restarted at ... ---" marker. retried runs with new patterns re-emit
36	// the key next to the restart marker so the dashboard picks up current
37	// patterns instead of the stale value stored in the original header. the
38	// last TaskHeaderPatterns occurrence in the file wins, so restarts that
39	// change patterns mid-file are reflected correctly.
40	func ParseProgressHeader(path string) (meta SessionMetadata, complete bool, err error) {	178✔
41	f, err := os.Open(path) //nolint:gosec // path from user-controlled glob pattern, acceptable for session discovery	178✔
42	if err != nil {	180✔
43	return SessionMetadata{}, false, fmt.Errorf("open file: %w", err)	2✔
44	}	2✔
45	defer f.Close()	176✔
46		176✔
47	reader := bufio.NewReader(f)	176✔
48		176✔
49	for {	1,228✔
50	line, readErr := reader.ReadString('\n')	1,052✔
51	line = trimLineEnding(line)	1,052✔
52		1,052✔
53	// stop at separator line; check readErr even when breaking	1,052✔
54	if line != "" && strings.HasPrefix(line, "---") {	1,225✔
55	if readErr != nil && !errors.Is(readErr, io.EOF) {	173✔
56	return SessionMetadata{}, false, fmt.Errorf("read file: %w", readErr)	×
57	}	×
58	complete = true	173✔
59	break	173✔
60	}
61
62	// parse key-value pairs (process line before checking error,
63	// as ReadString may return partial data alongside an error)
64	applyHeaderField(line, &meta)	879✔
65		879✔
66	if readErr != nil {	882✔
67	if !errors.Is(readErr, io.EOF) {	3✔
NEW 68	return SessionMetadata{}, false, fmt.Errorf("read file: %w", readErr)	×
UNCOV 69	}	×
70	return meta, complete, nil // EOF before separator	3✔
71	}
72	}
73
74	// scan the remainder for restart-marker TaskHeaderPatterns overrides.
75	// last occurrence wins so the freshest run's patterns are used.
76	for {	464✔
77	line, readErr := reader.ReadString('\n')	291✔
78	line = trimLineEnding(line)	291✔
79	if val, found := strings.CutPrefix(line, "TaskHeaderPatterns: "); found {	294✔
80	if patterns := parsePatternList(val); len(patterns) > 0 {	6✔
81	meta.TaskHeaderPatterns = patterns	3✔
82	}	3✔
83	}
84	if readErr != nil {	464✔
85	if !errors.Is(readErr, io.EOF) {	173✔
NEW 86	return meta, complete, fmt.Errorf("read file: %w", readErr)	×
87	}	×
88	return meta, complete, nil	173✔
89	}
90	}
91	}
92
93	// applyHeaderField parses a single header line (e.g. "Plan: ...") and updates
94	// the metadata in place. unknown or unmatched lines are ignored.
95	func applyHeaderField(line string, meta *SessionMetadata) {	879✔
96	if val, found := strings.CutPrefix(line, "Plan: "); found {	1,053✔
97	meta.PlanPath = val	174✔
98	return	174✔
99	}	174✔
100	if val, found := strings.CutPrefix(line, "Branch: "); found {	881✔
101	meta.Branch = val	176✔
102	return	176✔
103	}	176✔
104	if val, found := strings.CutPrefix(line, "Mode: "); found {	704✔
105	meta.Mode = val	175✔
106	return	175✔
107	}	175✔
108	if val, found := strings.CutPrefix(line, "Started: "); found {	526✔
109	// header timestamps are written in local time without a zone offset	172✔
110	if t, parseErr := time.ParseInLocation("2006-01-02 15:04:05", val, time.Local); parseErr == nil {	344✔
111	meta.StartTime = t	172✔
112	}	172✔
113	return	172✔
114	}
115	if val, found := strings.CutPrefix(line, "TaskHeaderPatterns: "); found {	185✔
116	if patterns := parsePatternList(val); len(patterns) > 0 {	6✔
117	meta.TaskHeaderPatterns = patterns	3✔
118	}	3✔
119	}
120	}
121
122	// parsePatternList splits a comma-separated header value into a trimmed,
123	// non-empty template list. the writer (pkg/progress writeHeader) rejects
124	// templates containing commas, so this split is lossless.
125	func parsePatternList(val string) []string {	6✔
126	parts := strings.Split(val, ",")	6✔
127	out := make([]string, 0, len(parts))	6✔
128	for _, p := range parts {	13✔
129	if trimmed := strings.TrimSpace(p); trimmed != "" {	14✔
130	out = append(out, trimmed)	7✔
131	}	7✔
132	}
133	return out	6✔
134	}
135
136	// loadProgressFileIntoSession reads a progress file and publishes events to the session's SSE server.
137	// used for completed sessions that were discovered after they finished.
138	// errors are silently ignored since this is best-effort loading.
139	// records the total bytes consumed into session.lastOffset so a later Reactivate()
140	// can resume tailing from after the loaded content instead of re-emitting it.
141	func (m SessionManager) loadProgressFileIntoSession(path string, session Session) {	165✔
142	f, err := os.Open(path) //nolint:gosec // path from user-controlled glob pattern, acceptable for session discovery	165✔
143	if err != nil {	167✔
144	return	2✔
145	}	2✔
146	defer f.Close()	163✔
147		163✔
148	reader := bufio.NewReader(f)	163✔
149	inHeader := true	163✔
150	phase := status.PhaseTask	163✔
151	var pendingSection string // section header waiting for first timestamped event	163✔
152	var bytesRead int64	163✔
153		163✔
154	for {	1,416✔
155	line, readErr := reader.ReadString('\n')	1,253✔
156	// a partial trailing line (no newline delimiter) means the writer is	1,253✔
157	// mid-write — the flock-race recovery path is the realistic case where	1,253✔
158	// loader is invoked on a still-being-written file. counting and	1,253✔
159	// publishing the partial would advance lastOffset past the partial	1,253✔
160	// bytes; a later Reactivate would then resume reading the suffix of	1,253✔
161	// the writer's eventual completed line as if it were a separate event,	1,253✔
162	// reintroducing the corruption this PR is meant to eliminate.	1,253✔
163	if readErr != nil && line != "" {	1,254✔
164	break	1✔
165	}
166	// count raw bytes including the delimiter before trimming, so LF, CRLF,
167	// and the final empty read all count correctly
168	bytesRead += int64(len(line))	1,252✔
169	line = trimLineEnding(line)	1,252✔
170		1,252✔
171	if line != "" {	2,298✔
172	var parsed ParsedLine	1,046✔
173	parsed, inHeader = parseProgressLine(line, inHeader)	1,046✔
174	phase, pendingSection = m.processProgressLine(session, parsed, phase, pendingSection)	1,046✔
175	}	1,046✔
176
177	if readErr != nil {	1,414✔
178	break // EOF or read error; best-effort loading, errors silently ignored	162✔
179	}
180	}
181
182	if pendingSection != "" {	163✔
183	m.emitPendingSection(session, pendingSection, phase, time.Now())	×
184	}	×
185
186	session.setLastOffset(bytesRead)	163✔
187	// record the phase the parser ended on so a later Reactivate resumes with	163✔
188	// the correct phase rather than the tailer's default (PhaseTask).	163✔
189	session.setLastPhase(phase)	163✔
190	}
191
192	// processProgressLine handles a single parsed progress line,
193	// updating phase and pendingSection state and publishing events as needed.
194	func (m SessionManager) processProgressLine(session Session, parsed ParsedLine,
195	phase status.Phase, pendingSection string) (status.Phase, string) {	1,046✔
196	switch parsed.Type {	1,046✔
197	case ParsedLineSkip:	972✔
198	return phase, pendingSection	972✔
199	case ParsedLineSection:	15✔
200	if pendingSection != "" {	15✔
201	m.emitPendingSection(session, pendingSection, phase, time.Now())	×
202	}	×
203	phase = parsed.Phase	15✔
204	// defer emitting section until we see a timestamped event	15✔
205	return phase, parsed.Section	15✔
206	case ParsedLineTimestamp:	58✔
207	// emit pending section with this event's timestamp (for accurate durations)	58✔
208	if pendingSection != "" {	73✔
209	m.emitPendingSection(session, pendingSection, phase, parsed.Timestamp)	15✔
210	pendingSection = ""	15✔
211	}	15✔
212	event := Event{	58✔
213	Type: parsed.EventType,	58✔
214	Phase: phase,	58✔
215	Text: parsed.Text,	58✔
216	Timestamp: parsed.Timestamp,	58✔
217	Signal: parsed.Signal,	58✔
218	}	58✔
219	if event.Type == EventTypeOutput {	114✔
220	if stats, ok := parseDiffStats(event.Text); ok {	57✔
221	session.SetDiffStats(stats)	1✔
222	}	1✔
223	}
224	_ = session.Publish(event)	58✔
225	case ParsedLinePlain:	1✔
226	_ = session.Publish(Event{	1✔
227	Type: EventTypeOutput,	1✔
228	Phase: phase,	1✔
229	Text: parsed.Text,	1✔
230	Timestamp: time.Now(),	1✔
231	})	1✔
232	}
233	return phase, pendingSection	59✔
234	}
235
236	// emitPendingSection publishes section and task_start events for a pending section.
237	// task_start is emitted before section for task iteration sections.
238	func (m SessionManager) emitPendingSection(session Session, sectionName string, phase status.Phase, ts time.Time) {	15✔
239	// emit task_start event for task iteration sections	15✔
240	if matches := taskIterationRegex.FindStringSubmatch(sectionName); matches != nil {	21✔
241	taskNum, err := strconv.Atoi(matches[1])	6✔
242	if err != nil {	6✔
243	// log parse error but continue - section will still be emitted	×
244	log.Printf("[WARN] failed to parse task number from section %q: %v", sectionName, err)	×
245	} else {	6✔
246	if err := session.Publish(Event{	6✔
247	Type: EventTypeTaskStart,	6✔
248	Phase: phase,	6✔
249	TaskNum: taskNum,	6✔
250	Text: sectionName,	6✔
251	Timestamp: ts,	6✔
252	}); err != nil {	6✔
253	log.Printf("[WARN] failed to publish task_start event: %v", err)	×
254	}	×
255	}
256	}
257
258	if err := session.Publish(Event{	15✔
259	Type: EventTypeSection,	15✔
260	Phase: phase,	15✔
261	Section: sectionName,	15✔
262	Text: sectionName,	15✔
263	Timestamp: ts,	15✔
264	}); err != nil {	15✔
265	log.Printf("[WARN] failed to publish section event: %v", err)	×
266	}	×
267	}
268
269	// phaseFromSection determines the phase from a section name.
270	// checks "codex"/"custom" before "review" because external review sections should be PhaseCodex.
271	func phaseFromSection(name string) status.Phase {	60✔
272	nameLower := strings.ToLower(name)	60✔
273	switch {	60✔
274	case strings.Contains(nameLower, "task"):	26✔
275	return status.PhaseTask	26✔
276	case strings.Contains(nameLower, "codex"), strings.Contains(nameLower, "custom"):	11✔
277	return status.PhaseCodex	11✔
278	case strings.Contains(nameLower, "review"):	13✔
279	return status.PhaseReview	13✔
280	case strings.Contains(nameLower, "claude-eval"), strings.Contains(nameLower, "claude eval"):	8✔
281	return status.PhaseClaudeEval	8✔
282	default:	2✔
283	return status.PhaseTask	2✔
284	}
285	}
286
287	// trimLineEnding removes trailing line ending to match bufio.ScanLines semantics:
288	// strips \n, \r\n, or a bare trailing \r (which ScanLines drops via dropCR at EOF).
289	// unlike strings.TrimRight("\r\n"), this preserves embedded \r characters in content.
290	func trimLineEnding(line string) string {	2,604✔
291	n := len(line)	2,604✔
292	if n > 0 && line[n-1] == '\n' {	4,866✔
293	n--	2,262✔
294	}	2,262✔
295	if n > 0 && line[n-1] == '\r' {	2,618✔
296	n--	14✔
297	}	14✔
298	return line[:n]	2,604✔
299	}

umputun / ralphex / 25115074240

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