20696400416

Committed 04 Jan 2026 05:15PM UTC coverage: 39.293% (-1.7%) from 41.026%

Build # 20696400416

Build Type

push

github

Committed by

web-flow

Commit Message

feat: Implement skip-if-exists flag for incremental processing (ADR-0… (#86)

* feat: Implement skip-if-exists flag for incremental processing (ADR-0006 Phase 3)

This commit implements the --skip-if-exists CLI flag to enable incremental
processing and recovery from failures. The flag checks if anchor output
exists before processing each document, allowing expensive LLM operations
to be skipped when results already exist.

Implementation Details:

1. CLI Flag Integration (cmd/opts.go):
   - Added skipIfExists field to optsAgent struct
   - Registered --skip-if-exists flag with description
   - Integrated skip logic into agent build pipeline
   - Validates that --output-dir is specified when using skip-if-exists

2. Anchor Key Computation (internal/blueprint/compiler/anchor.go):
   - Created AnchorKeyComputer to calculate expected output keys
   - Supports all step types: AgentStep, RouterStep, ForeachStep, RunStep
   - Handles emit attribute to compute prefixed output paths
   - For foreach steps, anchor is the array file (not individual elements)
   - Defaults to input key when no emit is specified

3. Skip Checking Logic (internal/blueprint/compiler/skip.go):
   - Created SkipChecker using Storage interface to check file existence
   - Uses AnchorKeyComputer to determine expected output location
   - Reports skipped documents via progress reporter
   - Returns early if anchor file already exists

4. Pipeline Integration (internal/service/worker/worker.go):
   - Added workflow field to Builder to store compiled workflow
   - Created SkipIfExists() builder method
   - Integrates storage, anchor computer, and progress reporter
   - Adds skip processor to pipeline before agent execution

5. Processor Implementation (internal/iosystem/processor/skipifexists.go):
   - Created SkipIfExists processor implementing Processor interface
   - Filters documents based on anchor existence check
   - Passes through EOF markers and empty document sets
   - Implements... (continued)

Run Details

15 of 216 new or added lines in 7 files covered. (6.94%)

1 existing line in 1 file now uncovered.

1578 of 4016 relevant lines covered (39.29%)

0.43 hits per line

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

28.61

/internal/progress/progress.go

//
// Copyright (C) 2025 Dmitry Kolesnikov
//
// This file may be modified and distributed under the terms
// of the MIT license.  See the LICENSE file for details.
// https://github.com/fogfish/iq
//

package progress

import (
        "fmt"
        "io"
        "os"
        "strings"
        "sync"
        "time"

        "github.com/nickwells/twrap.mod/twrap"
        "golang.org/x/term"
)

// TokenUsage represents token usage statistics
type TokenUsage struct {
        InputTokens int
        ReplyTokens int
}

// ProfileTokenUsage represents token usage for a specific profile
type ProfileTokenUsage struct {
        Name  string
        Usage TokenUsage
}

// TokenSource interface for accessing token usage from LLM routers
type TokenSource interface {
        Usage() TokenUsage
        ProfileUsage() []ProfileTokenUsage
}

// Icons used for progress reporting - centralized for easy customization
const (
        IconWorkflowLoad     = "📋" // Loading workflow file
        IconWorkflowCompiled = "✅" // Workflow successfully compiled
        IconWorkflowError    = "❌" // Workflow error

        IconDocumentStart    = "📄"  // Document processing started
        IconDocumentComplete = "✅"  // Document processed successfully
        IconDocumentError    = "❌"  // Document processing error
        IconDocumentSkipped  = "⏭️" // Document skipped

        IconStepProcessing = "⚙️ " // Step is processing (mutable line)
        IconStepComplete   = "✅"   // Step completed successfully
        IconStepError      = "❌"   // Step failed with error

        IconRouterEvaluating = "🧭" // Router evaluating conditions
        IconRouterMatched    = "↳" // Router matched a route (text, not emoji)

        IconForeachStart = "🔁" // Foreach loop started
        IconForeachItem  = "✅" // Foreach item completed

        IconRetryAttempt   = "🔄" // Retry attempt
        IconRetrySuccess   = "✅" // Retry succeeded
        IconRetryExhausted = "❌" // All retries exhausted

        IconBatchProcessing = "📂"  // Batch processing
        IconChunking        = "✂️" // Document chunking

        IconThinking = "💭" // LLM thinking/reasoning content

        IconSummary = "📊"  // Pipeline summary
        IconWarning = "⚠️" // Warning message
)

// Reporter handles all progress output to stderr with educational messages
type Reporter struct {
        w               io.Writer
        mu              sync.Mutex
        quiet           bool
        lastLine        string
        hasThinking     bool // Set to true when thinking content is shown
        isInForeachMode bool // Set to true when inside a foreach loop
        startTime       time.Time
        stats           Stats
        tokenSource     TokenSource // Source for final token usage reporting
}

// Stats tracks processing metrics
type Stats struct {
        DocsProcessed int
        DocsSkipped   int
        DocsErrors    int
        TokensInput   int
        TokensOutput  int
        Errors        []error
}

// New creates a new progress reporter
func New(quiet bool) *Reporter {
        return &Reporter{
                w:         os.Stderr,
                quiet:     quiet,
                startTime: time.Now(),
        }
}

// NewWithWriter creates a reporter with a custom writer (for testing)
func NewWithWriter(w io.Writer, quiet bool) *Reporter {
        return &Reporter{
                w:         w,
                quiet:     quiet,
                startTime: time.Now(),
        }
}

// SetTokenSource sets the token source for final usage reporting
func (r *Reporter) SetTokenSource(source TokenSource) {
        r.mu.Lock()
        defer r.mu.Unlock()
        r.tokenSource = source
}

// SetForeachMode sets whether we're inside a foreach loop
func (r *Reporter) SetForeachMode(inForeach bool) {
        r.mu.Lock()
        defer r.mu.Unlock()
        r.isInForeachMode = inForeach
}

// Workflow lifecycle events

// WorkflowLoading indicates workflow file is being loaded
func (r *Reporter) WorkflowLoading(path string) {
        r.println(fmt.Sprintf("%s Loading workflow from %s", IconWorkflowLoad, path))
}

// WorkflowCompiled indicates successful workflow compilation
func (r *Reporter) WorkflowCompiled(name string, jobCount, stepCount int) {
        if name == "" {
                name = "(unnamed)"
        }
        r.println(fmt.Sprintf("%s Workflow compiled: \"%s\" (%d jobs, %d steps)", IconWorkflowCompiled, name, jobCount, stepCount))
        r.println("")
}

// WorkflowError reports a workflow-level error
func (r *Reporter) WorkflowError(err error) {
        r.println(fmt.Sprintf("%s Workflow error: %v", IconWorkflowError, err))
}

// Document processing events

// DocumentStart indicates a document is starting to process
func (r *Reporter) DocumentStart(path string, sizeKB float64) {
        if sizeKB > 0 {
                r.println(fmt.Sprintf("%s Processing: %s (%.1f KB)", IconDocumentStart, path, sizeKB))
        } else {
                r.println(fmt.Sprintf("%s Processing: %s", IconDocumentStart, path))
        }
}

// DocumentComplete indicates successful document completion
func (r *Reporter) DocumentComplete(path string, duration time.Duration) {
        r.stats.DocsProcessed++
        r.println(fmt.Sprintf("%s Completed: %s (%.1fs total)", IconDocumentComplete, path, duration.Seconds()))
        r.println("")
}

// DocumentError reports a document processing error
func (r *Reporter) DocumentError(path string, err error) {
        r.stats.DocsErrors++
        r.stats.Errors = append(r.stats.Errors, err)
        r.println(fmt.Sprintf("%s Failed: %s - %v", IconDocumentError, path, err))
}

// DocumentSkipped indicates a document was skipped
func (r *Reporter) DocumentSkipped(path string, reason string) {
        r.stats.DocsSkipped++
        r.println(fmt.Sprintf("%s Skipped: %s (%s)", IconDocumentSkipped, path, reason))
}

// Step execution events

// StepStart indicates a workflow step is starting
func (r *Reporter) StepStart(jobName, stepName string, stepNum, totalSteps int) {
        r.mu.Lock()
        r.hasThinking = false // Reset thinking flag for new step
        r.mu.Unlock()

        if stepName != "" {
                r.updateLine(fmt.Sprintf("   %s Step %d/%d: %s.%s → Processing...",
                        IconStepProcessing, stepNum, totalSteps, jobName, stepName))
        } else {
                r.updateLine(fmt.Sprintf("   %s Step %d/%d: %s → Processing...",
                        IconStepProcessing, stepNum, totalSteps, jobName))
        }
}

// StepProgress updates the current step status (mutable)
func (r *Reporter) StepProgress(message string) {
        r.updateLine(fmt.Sprintf("   %s %s", IconStepProcessing, message))
}

// StepComplete indicates step completion
func (r *Reporter) StepComplete(jobName, stepName string, stepNum, totalSteps int, duration time.Duration, tokens int) {
        r.stats.TokensInput += tokens

        // Don't show step completion messages when inside a foreach loop
        // The foreach will handle its own reporting
        if r.isInForeachMode {
                return
        }

        if tokens > 0 {
                if stepName != "" {
                        r.println(fmt.Sprintf("   %s Step %d/%d: %s.%s completed in %.1fs (%d tokens)",
                                IconStepComplete, stepNum, totalSteps, jobName, stepName, duration.Seconds(), tokens))
                } else {
                        r.println(fmt.Sprintf("   %s Step %d/%d: %s completed in %.1fs (%d tokens)",
                                IconStepComplete, stepNum, totalSteps, jobName, duration.Seconds(), tokens))
                }
        } else {
                if stepName != "" {
                        r.println(fmt.Sprintf("   %s Step %d/%d: %s.%s completed in %.1fs",
                                IconStepComplete, stepNum, totalSteps, jobName, stepName, duration.Seconds()))
                } else {
                        r.println(fmt.Sprintf("   %s Step %d/%d: %s completed in %.1fs",
                                IconStepComplete, stepNum, totalSteps, jobName, duration.Seconds()))
                }
        }
}

// StepSkipped indicates step was skipped (e.g., cache hit)
func (r *Reporter) StepSkipped(jobName, stepName string, stepNum, totalSteps int, reason string) {
        // Don't show step skip messages when inside a foreach loop
        if r.isInForeachMode {
                return
        }

        if stepName != "" {
                r.println(fmt.Sprintf("   ⏭️  Step %d/%d: %s.%s skipped (%s)",
                        stepNum, totalSteps, jobName, stepName, reason))
        } else {
                r.println(fmt.Sprintf("   ⏭️  Step %d/%d: %s skipped (%s)",
                        stepNum, totalSteps, jobName, reason))
        }
}

// StepError reports a step error
func (r *Reporter) StepError(jobName, stepName string, stepNum, totalSteps int, err error) {
        if stepName != "" {
                r.println(fmt.Sprintf("   %s Step %d/%d: %s.%s failed - %v",
                        IconStepError, stepNum, totalSteps, jobName, stepName, err))
        } else {
                r.println(fmt.Sprintf("   %s Step %d/%d: %s failed - %v",
                        IconStepError, stepNum, totalSteps, jobName, err))
        }
}

// Router and control flow events

// RouterEvaluating indicates router is evaluating conditions
func (r *Reporter) RouterEvaluating() {
        r.updateLine(fmt.Sprintf("   %s Router evaluating conditions...", IconRouterEvaluating))
}

// RouterMatched indicates router matched a route
func (r *Reporter) RouterMatched(routeName string, targetJob string) {
        r.println(fmt.Sprintf("   %s Routing decision:", IconRouterEvaluating))
        r.println(fmt.Sprintf("      %s Matched route: %s → %s", IconRouterMatched, routeName, targetJob))
        r.println("")
}

// RouterDefault indicates router is using default route
func (r *Reporter) RouterDefault(targetJob string) {
        r.println(fmt.Sprintf("   %s Routing decision:", IconRouterEvaluating))
        r.println(fmt.Sprintf("      %s Using default route → %s", IconRouterMatched, targetJob))
        r.println("")
}

// RouterNoMatch indicates no route was matched
func (r *Reporter) RouterNoMatch() {
        r.println(fmt.Sprintf("   %s Router: No matching route found", IconWarning))
}

// Foreach iteration events

// ForeachStart indicates foreach loop is starting
func (r *Reporter) ForeachStart(itemCount int) {
        r.println(fmt.Sprintf("   %s Foreach: Processing %d items", IconForeachStart, itemCount))
}

// ForeachItem reports progress on a foreach item
func (r *Reporter) ForeachItem(index, total int, status string) {
        r.println(fmt.Sprintf("      [%d/%d] %s", index, total, status))
}

// ForeachComplete indicates foreach loop completed
func (r *Reporter) ForeachComplete(successCount, totalCount int, duration time.Duration) {
        r.println(fmt.Sprintf("   %s Foreach completed: %d/%d items succeeded (%.1fs)",
                IconForeachItem, successCount, totalCount, duration.Seconds()))
        r.println("")
}

// Retry and recovery events

// RetryAttempt indicates a retry is being attempted
func (r *Reporter) RetryAttempt(attempt, maxAttempts int, delay time.Duration) {
        if attempt == 1 {
                r.println(fmt.Sprintf("   %s Step failed (attempt %d/%d)", IconWarning, attempt, maxAttempts))
                if delay > 0 {
                        r.println(fmt.Sprintf("      %s Retrying in %.0fs...", IconRouterMatched, delay.Seconds()))
                }
        } else {
                r.updateLine(fmt.Sprintf("   %s Retry attempt %d/%d...", IconRetryAttempt, attempt, maxAttempts))
        }
}

// RetrySuccess indicates retry succeeded
func (r *Reporter) RetrySuccess(attempt int) {
        r.println(fmt.Sprintf("   %s Retry succeeded on attempt %d", IconRetrySuccess, attempt))
}

// RetryExhausted indicates all retry attempts failed
func (r *Reporter) RetryExhausted(maxAttempts int) {
        r.println(fmt.Sprintf("   %s All %d retry attempts exhausted", IconRetryExhausted, maxAttempts))
}

// LLM thinking events

// ThinkingContent displays LLM thinking/reasoning content
// This is used when --llm-think is enabled to show intermediate reasoning
func (r *Reporter) ThinkingContent(text string) {
        r.mu.Lock()
        defer r.mu.Unlock()

        if r.quiet {
                return
        }

        // Mark that we've shown thinking content
        r.hasThinking = true

        // Store the current processing line to restore it after thinking
        processingLine := r.lastLine

        // If there's a mutable line (e.g., "   ⚙️  Step 1/2: main → Processing..."),
        // finalize it before showing thinking header
        if r.lastLine != "" {
                fmt.Fprintln(r.w) // Move to new line, keep the processing line visible
                r.lastLine = ""
        }

        // Print thinking header (e.g., "   💭 Step 1/2: main → Thinking")
        if processingLine != "" {
                // Extract step info from processing line and create thinking header
                thinkingHeader := strings.Replace(processingLine, "Processing", "Thinking", 1)
                thinkingHeader = strings.Replace(thinkingHeader, IconStepProcessing, IconThinking, 1)
                fmt.Fprintln(r.w, thinkingHeader)
        }

        // Get terminal width for wrapping (default to 80 if not a terminal)
        width := 80
        if fd, ok := r.w.(*os.File); ok {
                if w, _, err := term.GetSize(int(fd.Fd())); err == nil && w > 0 {
                        width = w
                }
        }

        // Print thinking content with proper indentation and wrapping
        // Format: "   💭 <first line text>"
        //         "      <continuation lines>"

        // Use a buffer to capture wrapped output
        var buf strings.Builder
        tw, err := twrap.NewTWConf(
                twrap.SetWriter(&buf),
                twrap.SetTargetLineLen(width-6), // Reserve space for indent (6 chars: 3 spaces + "💭 ")
        )
        if err != nil {
                // Fallback: print without wrapping if configuration fails
                fmt.Fprintf(r.w, "   %s %s\n", IconThinking, text)
                return
        }

        // Wrap text without indent first
        tw.Wrap(text, 0)

        // Now add indentation to each line (thinking content is indented)
        lines := strings.Split(strings.TrimRight(buf.String(), "\n"), "\n")
        for _, line := range lines {
                fmt.Fprintf(r.w, "      %s\n", line) // All lines indented (no emoji on content lines)
        }

        // Restore the processing line as mutable after thinking
        if processingLine != "" {
                r.lastLine = processingLine
                fmt.Fprint(r.w, processingLine) // Print without newline (mutable)
        } else {
                fmt.Fprintln(r.w) // Just add blank line if no processing line to restore
        }
}

// Batch processing events

// BatchStart indicates batch processing is starting
func (r *Reporter) BatchStart(inputPath, outputPath string, mutable bool) {
        if mutable {
                r.println(fmt.Sprintf("%s Batch processing: %s → %s (mutable mode)", IconBatchProcessing, inputPath, outputPath))
        } else {
                r.println(fmt.Sprintf("%s Batch processing: %s → %s", IconBatchProcessing, inputPath, outputPath))
        }
}

// BatchProgress reports batch processing progress
func (r *Reporter) BatchProgress(processed, total int) {
        if total > 0 {
                percent := int(float64(processed) / float64(total) * 100)
                bars := percent / 5 // 20 bars total
                progress := strings.Repeat("█", bars) + strings.Repeat("░", 20-bars)
                r.updateLine(fmt.Sprintf("Processing: %s %d/%d files (%d%%)",
                        progress, processed, total, percent))
        } else {
                r.updateLine(fmt.Sprintf("Processing: %d files...", processed))
        }
}

// BatchComplete finalizes batch progress
func (r *Reporter) BatchComplete() {
        if r.lastLine != "" {
                r.println("") // Move to new line
        }
}

// Chunking events

// ChunkingStart indicates document is being split
func (r *Reporter) ChunkingStart(strategy string, chunkCount int) {
        r.println(fmt.Sprintf("   %s Splitting into %d chunks (strategy: %s)", IconChunking, chunkCount, strategy))
}

// Summary and statistics

// Summary prints final execution summary
func (r *Reporter) Summary() {
        duration := time.Since(r.startTime)

        r.println("")
        r.println(fmt.Sprintf("%s Pipeline Summary:", IconSummary))

        if r.stats.DocsProcessed > 0 {
                r.println(fmt.Sprintf("   %s Processed: %d documents", IconDocumentComplete, r.stats.DocsProcessed))
        }

        if r.stats.DocsSkipped > 0 {
                r.println(fmt.Sprintf("   %s Skipped: %d documents", IconDocumentSkipped, r.stats.DocsSkipped))
        }

        if r.stats.DocsErrors > 0 {
                r.println(fmt.Sprintf("   %s Errors: %d documents", IconWarning, r.stats.DocsErrors))
        }

        // Use token source for authoritative usage if available, otherwise fall back to step-level stats
        r.mu.Lock()
        tokenSource := r.tokenSource
        r.mu.Unlock()

        if tokenSource != nil {
                // Use Router's authoritative token data with per-profile breakdown
                usage := tokenSource.Usage()
                profiles := tokenSource.ProfileUsage()

                if usage.InputTokens > 0 || usage.ReplyTokens > 0 {
                        total := usage.InputTokens + usage.ReplyTokens
                        r.println(fmt.Sprintf("   📈 Tokens: %s total",
                                formatNumber(total)))

                        // Show per-profile breakdown
                        for _, profile := range profiles {
                                if profile.Usage.InputTokens > 0 || profile.Usage.ReplyTokens > 0 {
                                        profileTotal := profile.Usage.InputTokens + profile.Usage.ReplyTokens
                                        r.println(fmt.Sprintf("      └─ %s: %s (input: %s | output: %s)",
                                                profile.Name,
                                                formatNumber(profileTotal),
                                                formatNumber(profile.Usage.InputTokens),
                                                formatNumber(profile.Usage.ReplyTokens)))
                                }
                        }
                }
        } else if r.stats.TokensInput > 0 || r.stats.TokensOutput > 0 {
                // Fall back to step-level token tracking (legacy)
                total := r.stats.TokensInput + r.stats.TokensOutput
                r.println(fmt.Sprintf("   📈 Tokens: %s (input: %s | output: %s)",
                        formatNumber(total),
                        formatNumber(r.stats.TokensInput),
                        formatNumber(r.stats.TokensOutput)))
        }

        r.println(fmt.Sprintf("   ⏱️  Duration: %s", formatDuration(duration)))

        // Add visual separator before output
        r.println("")
        r.println("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
        r.println("")
}

// SummaryQuick prints a compact one-line summary
func (r *Reporter) SummaryQuick(docCount int, duration time.Duration) {
        tokens := r.stats.TokensInput + r.stats.TokensOutput
        if tokens > 0 {
                r.println(fmt.Sprintf("\n📊 Summary: %d document(s), %s tokens, %s",
                        docCount, formatNumber(tokens), formatDuration(duration)))
        } else {
                r.println(fmt.Sprintf("\n📊 Summary: %d document(s), %s",
                        docCount, formatDuration(duration)))
        }
}

// Helper methods

// UpdateTokens updates token statistics
func (r *Reporter) UpdateTokens(inputTokens, outputTokens int) {
        r.mu.Lock()
        defer r.mu.Unlock()
        r.stats.TokensInput += inputTokens
        r.stats.TokensOutput += outputTokens
}

// GetStats returns a copy of current statistics
func (r *Reporter) GetStats() Stats {
        r.mu.Lock()
        defer r.mu.Unlock()
        return r.stats
}

// updateLine updates the current line (mutable output)
func (r *Reporter) updateLine(text string) {
        if r.quiet {
                return
        }
        r.mu.Lock()
        defer r.mu.Unlock()

        // Clear previous line
        if r.lastLine != "" {
                fmt.Fprintf(r.w, "\r\033[K")
        }

        fmt.Fprint(r.w, text)
        r.lastLine = text
}

// println writes a new line (finalize any mutable line first)
func (r *Reporter) println(text string) {
        if r.quiet {
                return
        }
        r.mu.Lock()
        defer r.mu.Unlock()

        if r.lastLine != "" {
                if r.hasThinking {
                        // In thinking mode: keep the previous line, just move to next line
                        fmt.Fprintln(r.w)
                } else {
                        // Without thinking: clear the mutable line and replace it
                        fmt.Fprint(r.w, "\r\033[K")
                }
                r.lastLine = ""
        }
        fmt.Fprintln(r.w, text)
}

// Formatting helpers

func formatNumber(n int) string {
        if n >= 1000000 {
                return fmt.Sprintf("%.1fM", float64(n)/1000000)
        }
        if n >= 1000 {
                return fmt.Sprintf("%.1fK", float64(n)/1000)
        }
        return fmt.Sprintf("%d", n)
}

func formatDuration(d time.Duration) string {
        if d < time.Second {
                return fmt.Sprintf("%.0fms", d.Seconds()*1000)
        }
        if d < time.Minute {
                return fmt.Sprintf("%.1fs", d.Seconds())
        }
        minutes := int(d.Minutes())
        seconds := int(d.Seconds()) % 60
        return fmt.Sprintf("%dm %ds", minutes, seconds)
}

1	//
2	// Copyright (C) 2025 Dmitry Kolesnikov
3	//
4	// This file may be modified and distributed under the terms
5	// of the MIT license. See the LICENSE file for details.
6	// https://github.com/fogfish/iq
7	//
8
9	package progress
10
11	import (
12	"fmt"
13	"io"
14	"os"
15	"strings"
16	"sync"
17	"time"
18
19	"github.com/nickwells/twrap.mod/twrap"
20	"golang.org/x/term"
21	)
22
23	// TokenUsage represents token usage statistics
24	type TokenUsage struct {
25	InputTokens int
26	ReplyTokens int
27	}
28
29	// ProfileTokenUsage represents token usage for a specific profile
30	type ProfileTokenUsage struct {
31	Name string
32	Usage TokenUsage
33	}
34
35	// TokenSource interface for accessing token usage from LLM routers
36	type TokenSource interface {
37	Usage() TokenUsage
38	ProfileUsage() []ProfileTokenUsage
39	}
40
41	// Icons used for progress reporting - centralized for easy customization
42	const (
43	IconWorkflowLoad = "📋" // Loading workflow file
44	IconWorkflowCompiled = "✅" // Workflow successfully compiled
45	IconWorkflowError = "❌" // Workflow error
46
47	IconDocumentStart = "📄" // Document processing started
48	IconDocumentComplete = "✅" // Document processed successfully
49	IconDocumentError = "❌" // Document processing error
50	IconDocumentSkipped = "⏭️" // Document skipped
51
52	IconStepProcessing = "⚙️ " // Step is processing (mutable line)
53	IconStepComplete = "✅" // Step completed successfully
54	IconStepError = "❌" // Step failed with error
55
56	IconRouterEvaluating = "🧭" // Router evaluating conditions
57	IconRouterMatched = "↳" // Router matched a route (text, not emoji)
58
59	IconForeachStart = "🔁" // Foreach loop started
60	IconForeachItem = "✅" // Foreach item completed
61
62	IconRetryAttempt = "🔄" // Retry attempt
63	IconRetrySuccess = "✅" // Retry succeeded
64	IconRetryExhausted = "❌" // All retries exhausted
65
66	IconBatchProcessing = "📂" // Batch processing
67	IconChunking = "✂️" // Document chunking
68
69	IconThinking = "💭" // LLM thinking/reasoning content
70
71	IconSummary = "📊" // Pipeline summary
72	IconWarning = "⚠️" // Warning message
73	)
74
75	// Reporter handles all progress output to stderr with educational messages
76	type Reporter struct {
77	w io.Writer
78	mu sync.Mutex
79	quiet bool
80	lastLine string
81	hasThinking bool // Set to true when thinking content is shown
82	isInForeachMode bool // Set to true when inside a foreach loop
83	startTime time.Time
84	stats Stats
85	tokenSource TokenSource // Source for final token usage reporting
86	}
87
88	// Stats tracks processing metrics
89	type Stats struct {
90	DocsProcessed int
91	DocsSkipped int
92	DocsErrors int
93	TokensInput int
94	TokensOutput int
95	Errors []error
96	}
97
98	// New creates a new progress reporter
99	func New(quiet bool) *Reporter {	1✔
100	return &Reporter{	1✔
101	w: os.Stderr,	1✔
102	quiet: quiet,	1✔
103	startTime: time.Now(),	1✔
104	}	1✔
105	}	1✔
106
107	// NewWithWriter creates a reporter with a custom writer (for testing)
108	func NewWithWriter(w io.Writer, quiet bool) *Reporter {	1✔
109	return &Reporter{	1✔
110	w: w,	1✔
111	quiet: quiet,	1✔
112	startTime: time.Now(),	1✔
113	}	1✔
114	}	1✔
115
116	// SetTokenSource sets the token source for final usage reporting
117	func (r *Reporter) SetTokenSource(source TokenSource) {	×
118	r.mu.Lock()	×
119	defer r.mu.Unlock()	×
120	r.tokenSource = source	×
121	}	×
122
123	// SetForeachMode sets whether we're inside a foreach loop
124	func (r *Reporter) SetForeachMode(inForeach bool) {	×
125	r.mu.Lock()	×
126	defer r.mu.Unlock()	×
127	r.isInForeachMode = inForeach	×
128	}	×
129
130	// Workflow lifecycle events
131
132	// WorkflowLoading indicates workflow file is being loaded
133	func (r *Reporter) WorkflowLoading(path string) {	1✔
134	r.println(fmt.Sprintf("%s Loading workflow from %s", IconWorkflowLoad, path))	1✔
135	}	1✔
136
137	// WorkflowCompiled indicates successful workflow compilation
138	func (r *Reporter) WorkflowCompiled(name string, jobCount, stepCount int) {	×
139	if name == "" {	×
140	name = "(unnamed)"	×
141	}	×
142	r.println(fmt.Sprintf("%s Workflow compiled: \"%s\" (%d jobs, %d steps)", IconWorkflowCompiled, name, jobCount, stepCount))	×
143	r.println("")	×
144	}
145
146	// WorkflowError reports a workflow-level error
147	func (r *Reporter) WorkflowError(err error) {	×
148	r.println(fmt.Sprintf("%s Workflow error: %v", IconWorkflowError, err))	×
149	}	×
150
151	// Document processing events
152
153	// DocumentStart indicates a document is starting to process
154	func (r *Reporter) DocumentStart(path string, sizeKB float64) {	1✔
155	if sizeKB > 0 {	2✔
156	r.println(fmt.Sprintf("%s Processing: %s (%.1f KB)", IconDocumentStart, path, sizeKB))	1✔
157	} else {	1✔
158	r.println(fmt.Sprintf("%s Processing: %s", IconDocumentStart, path))	×
159	}	×
160	}
161
162	// DocumentComplete indicates successful document completion
163	func (r *Reporter) DocumentComplete(path string, duration time.Duration) {	1✔
164	r.stats.DocsProcessed++	1✔
165	r.println(fmt.Sprintf("%s Completed: %s (%.1fs total)", IconDocumentComplete, path, duration.Seconds()))	1✔
166	r.println("")	1✔
167	}	1✔
168
169	// DocumentError reports a document processing error
170	func (r *Reporter) DocumentError(path string, err error) {	×
171	r.stats.DocsErrors++	×
172	r.stats.Errors = append(r.stats.Errors, err)	×
173	r.println(fmt.Sprintf("%s Failed: %s - %v", IconDocumentError, path, err))	×
174	}	×
175
176	// DocumentSkipped indicates a document was skipped
177	func (r *Reporter) DocumentSkipped(path string, reason string) {	×
178	r.stats.DocsSkipped++	×
179	r.println(fmt.Sprintf("%s Skipped: %s (%s)", IconDocumentSkipped, path, reason))	×
180	}	×
181
182	// Step execution events
183
184	// StepStart indicates a workflow step is starting
185	func (r *Reporter) StepStart(jobName, stepName string, stepNum, totalSteps int) {	1✔
186	r.mu.Lock()	1✔
187	r.hasThinking = false // Reset thinking flag for new step	1✔
188	r.mu.Unlock()	1✔
189		1✔
190	if stepName != "" {	2✔
191	r.updateLine(fmt.Sprintf(" %s Step %d/%d: %s.%s → Processing...",	1✔
192	IconStepProcessing, stepNum, totalSteps, jobName, stepName))	1✔
193	} else {	1✔
194	r.updateLine(fmt.Sprintf(" %s Step %d/%d: %s → Processing...",	×
195	IconStepProcessing, stepNum, totalSteps, jobName))	×
196	}	×
197	}
198
199	// StepProgress updates the current step status (mutable)
200	func (r *Reporter) StepProgress(message string) {	×
201	r.updateLine(fmt.Sprintf(" %s %s", IconStepProcessing, message))	×
202	}	×
203
204	// StepComplete indicates step completion
205	func (r *Reporter) StepComplete(jobName, stepName string, stepNum, totalSteps int, duration time.Duration, tokens int) {	1✔
206	r.stats.TokensInput += tokens	1✔
207		1✔
208	// Don't show step completion messages when inside a foreach loop	1✔
209	// The foreach will handle its own reporting	1✔
210	if r.isInForeachMode {	1✔
211	return	×
212	}	×
213
214	if tokens > 0 {	2✔
215	if stepName != "" {	2✔
216	r.println(fmt.Sprintf(" %s Step %d/%d: %s.%s completed in %.1fs (%d tokens)",	1✔
217	IconStepComplete, stepNum, totalSteps, jobName, stepName, duration.Seconds(), tokens))	1✔
218	} else {	1✔
219	r.println(fmt.Sprintf(" %s Step %d/%d: %s completed in %.1fs (%d tokens)",	×
220	IconStepComplete, stepNum, totalSteps, jobName, duration.Seconds(), tokens))	×
221	}	×
222	} else {	×
223	if stepName != "" {	×
224	r.println(fmt.Sprintf(" %s Step %d/%d: %s.%s completed in %.1fs",	×
225	IconStepComplete, stepNum, totalSteps, jobName, stepName, duration.Seconds()))	×
226	} else {	×
227	r.println(fmt.Sprintf(" %s Step %d/%d: %s completed in %.1fs",	×
228	IconStepComplete, stepNum, totalSteps, jobName, duration.Seconds()))	×
229	}	×
230	}
231	}
232
233	// StepSkipped indicates step was skipped (e.g., cache hit)
NEW 234	func (r *Reporter) StepSkipped(jobName, stepName string, stepNum, totalSteps int, reason string) {	×
NEW 235	// Don't show step skip messages when inside a foreach loop	×
NEW 236	if r.isInForeachMode {	×
NEW 237	return	×
NEW 238	}	×
239
NEW 240	if stepName != "" {	×
NEW 241	r.println(fmt.Sprintf(" ⏭️ Step %d/%d: %s.%s skipped (%s)",	×
NEW 242	stepNum, totalSteps, jobName, stepName, reason))	×
NEW 243	} else {	×
NEW 244	r.println(fmt.Sprintf(" ⏭️ Step %d/%d: %s skipped (%s)",	×
NEW 245	stepNum, totalSteps, jobName, reason))	×
NEW 246	}	×
247	}
248
249	// StepError reports a step error
250	func (r *Reporter) StepError(jobName, stepName string, stepNum, totalSteps int, err error) {	×
251	if stepName != "" {	×
252	r.println(fmt.Sprintf(" %s Step %d/%d: %s.%s failed - %v",	×
253	IconStepError, stepNum, totalSteps, jobName, stepName, err))	×
254	} else {	×
255	r.println(fmt.Sprintf(" %s Step %d/%d: %s failed - %v",	×
256	IconStepError, stepNum, totalSteps, jobName, err))	×
257	}	×
258	}
259
260	// Router and control flow events
261
262	// RouterEvaluating indicates router is evaluating conditions
263	func (r *Reporter) RouterEvaluating() {	×
264	r.updateLine(fmt.Sprintf(" %s Router evaluating conditions...", IconRouterEvaluating))	×
265	}	×
266
267	// RouterMatched indicates router matched a route
268	func (r *Reporter) RouterMatched(routeName string, targetJob string) {	×
269	r.println(fmt.Sprintf(" %s Routing decision:", IconRouterEvaluating))	×
270	r.println(fmt.Sprintf(" %s Matched route: %s → %s", IconRouterMatched, routeName, targetJob))	×
271	r.println("")	×
272	}	×
273
274	// RouterDefault indicates router is using default route
275	func (r *Reporter) RouterDefault(targetJob string) {	×
276	r.println(fmt.Sprintf(" %s Routing decision:", IconRouterEvaluating))	×
277	r.println(fmt.Sprintf(" %s Using default route → %s", IconRouterMatched, targetJob))	×
278	r.println("")	×
279	}	×
280
281	// RouterNoMatch indicates no route was matched
282	func (r *Reporter) RouterNoMatch() {	×
283	r.println(fmt.Sprintf(" %s Router: No matching route found", IconWarning))	×
284	}	×
285
286	// Foreach iteration events
287
288	// ForeachStart indicates foreach loop is starting
289	func (r *Reporter) ForeachStart(itemCount int) {	×
290	r.println(fmt.Sprintf(" %s Foreach: Processing %d items", IconForeachStart, itemCount))	×
291	}	×
292
293	// ForeachItem reports progress on a foreach item
294	func (r *Reporter) ForeachItem(index, total int, status string) {	×
295	r.println(fmt.Sprintf(" [%d/%d] %s", index, total, status))	×
296	}	×
297
298	// ForeachComplete indicates foreach loop completed
299	func (r *Reporter) ForeachComplete(successCount, totalCount int, duration time.Duration) {	×
300	r.println(fmt.Sprintf(" %s Foreach completed: %d/%d items succeeded (%.1fs)",	×
301	IconForeachItem, successCount, totalCount, duration.Seconds()))	×
302	r.println("")	×
303	}	×
304
305	// Retry and recovery events
306
307	// RetryAttempt indicates a retry is being attempted
308	func (r *Reporter) RetryAttempt(attempt, maxAttempts int, delay time.Duration) {	×
309	if attempt == 1 {	×
310	r.println(fmt.Sprintf(" %s Step failed (attempt %d/%d)", IconWarning, attempt, maxAttempts))	×
311	if delay > 0 {	×
312	r.println(fmt.Sprintf(" %s Retrying in %.0fs...", IconRouterMatched, delay.Seconds()))	×
313	}	×
314	} else {	×
315	r.updateLine(fmt.Sprintf(" %s Retry attempt %d/%d...", IconRetryAttempt, attempt, maxAttempts))	×
316	}	×
317	}
318
319	// RetrySuccess indicates retry succeeded
320	func (r *Reporter) RetrySuccess(attempt int) {	×
321	r.println(fmt.Sprintf(" %s Retry succeeded on attempt %d", IconRetrySuccess, attempt))	×
322	}	×
323
324	// RetryExhausted indicates all retry attempts failed
325	func (r *Reporter) RetryExhausted(maxAttempts int) {	×
326	r.println(fmt.Sprintf(" %s All %d retry attempts exhausted", IconRetryExhausted, maxAttempts))	×
327	}	×
328
329	// LLM thinking events
330
331	// ThinkingContent displays LLM thinking/reasoning content
332	// This is used when --llm-think is enabled to show intermediate reasoning
333	func (r *Reporter) ThinkingContent(text string) {	×
334	r.mu.Lock()	×
335	defer r.mu.Unlock()	×
336		×
337	if r.quiet {	×
338	return	×
339	}	×
340
341	// Mark that we've shown thinking content
342	r.hasThinking = true	×
343		×
344	// Store the current processing line to restore it after thinking	×
345	processingLine := r.lastLine	×
346		×
347	// If there's a mutable line (e.g., " ⚙️ Step 1/2: main → Processing..."),	×
348	// finalize it before showing thinking header	×
349	if r.lastLine != "" {	×
350	fmt.Fprintln(r.w) // Move to new line, keep the processing line visible	×
351	r.lastLine = ""	×
352	}	×
353
354	// Print thinking header (e.g., " 💭 Step 1/2: main → Thinking")
355	if processingLine != "" {	×
356	// Extract step info from processing line and create thinking header	×
357	thinkingHeader := strings.Replace(processingLine, "Processing", "Thinking", 1)	×
358	thinkingHeader = strings.Replace(thinkingHeader, IconStepProcessing, IconThinking, 1)	×
359	fmt.Fprintln(r.w, thinkingHeader)	×
360	}	×
361
362	// Get terminal width for wrapping (default to 80 if not a terminal)
363	width := 80	×
364	if fd, ok := r.w.(*os.File); ok {	×
365	if w, _, err := term.GetSize(int(fd.Fd())); err == nil && w > 0 {	×
366	width = w	×
367	}	×
368	}
369
370	// Print thinking content with proper indentation and wrapping
371	// Format: " 💭 <first line text>"
372	// " <continuation lines>"
373
374	// Use a buffer to capture wrapped output
375	var buf strings.Builder	×
376	tw, err := twrap.NewTWConf(	×
377	twrap.SetWriter(&buf),	×
378	twrap.SetTargetLineLen(width-6), // Reserve space for indent (6 chars: 3 spaces + "💭 ")	×
379	)	×
380	if err != nil {	×
381	// Fallback: print without wrapping if configuration fails	×
382	fmt.Fprintf(r.w, " %s %s\n", IconThinking, text)	×
383	return	×
384	}	×
385
386	// Wrap text without indent first
387	tw.Wrap(text, 0)	×
388		×
389	// Now add indentation to each line (thinking content is indented)	×
390	lines := strings.Split(strings.TrimRight(buf.String(), "\n"), "\n")	×
391	for _, line := range lines {	×
392	fmt.Fprintf(r.w, " %s\n", line) // All lines indented (no emoji on content lines)	×
393	}	×
394
395	// Restore the processing line as mutable after thinking
396	if processingLine != "" {	×
397	r.lastLine = processingLine	×
398	fmt.Fprint(r.w, processingLine) // Print without newline (mutable)	×
399	} else {	×
400	fmt.Fprintln(r.w) // Just add blank line if no processing line to restore	×
401	}	×
402	}
403
404	// Batch processing events
405
406	// BatchStart indicates batch processing is starting
407	func (r *Reporter) BatchStart(inputPath, outputPath string, mutable bool) {	×
408	if mutable {	×
409	r.println(fmt.Sprintf("%s Batch processing: %s → %s (mutable mode)", IconBatchProcessing, inputPath, outputPath))	×
410	} else {	×
411	r.println(fmt.Sprintf("%s Batch processing: %s → %s", IconBatchProcessing, inputPath, outputPath))	×
412	}	×
413	}
414
415	// BatchProgress reports batch processing progress
416	func (r *Reporter) BatchProgress(processed, total int) {	×
417	if total > 0 {	×
418	percent := int(float64(processed) / float64(total) * 100)	×
419	bars := percent / 5 // 20 bars total	×
420	progress := strings.Repeat("█", bars) + strings.Repeat("░", 20-bars)	×
421	r.updateLine(fmt.Sprintf("Processing: %s %d/%d files (%d%%)",	×
422	progress, processed, total, percent))	×
423	} else {	×
424	r.updateLine(fmt.Sprintf("Processing: %d files...", processed))	×
425	}	×
426	}
427
428	// BatchComplete finalizes batch progress
429	func (r *Reporter) BatchComplete() {	×
430	if r.lastLine != "" {	×
431	r.println("") // Move to new line	×
432	}	×
433	}
434
435	// Chunking events
436
437	// ChunkingStart indicates document is being split
438	func (r *Reporter) ChunkingStart(strategy string, chunkCount int) {	×
439	r.println(fmt.Sprintf(" %s Splitting into %d chunks (strategy: %s)", IconChunking, chunkCount, strategy))	×
440	}	×
441
442	// Summary and statistics
443
444	// Summary prints final execution summary
445	func (r *Reporter) Summary() {	×
446	duration := time.Since(r.startTime)	×
447		×
448	r.println("")	×
449	r.println(fmt.Sprintf("%s Pipeline Summary:", IconSummary))	×
450		×
451	if r.stats.DocsProcessed > 0 {	×
452	r.println(fmt.Sprintf(" %s Processed: %d documents", IconDocumentComplete, r.stats.DocsProcessed))	×
453	}	×
454
455	if r.stats.DocsSkipped > 0 {	×
456	r.println(fmt.Sprintf(" %s Skipped: %d documents", IconDocumentSkipped, r.stats.DocsSkipped))	×
457	}	×
458
459	if r.stats.DocsErrors > 0 {	×
460	r.println(fmt.Sprintf(" %s Errors: %d documents", IconWarning, r.stats.DocsErrors))	×
461	}	×
462
463	// Use token source for authoritative usage if available, otherwise fall back to step-level stats
464	r.mu.Lock()	×
465	tokenSource := r.tokenSource	×
466	r.mu.Unlock()	×
467		×
468	if tokenSource != nil {	×
469	// Use Router's authoritative token data with per-profile breakdown	×
470	usage := tokenSource.Usage()	×
471	profiles := tokenSource.ProfileUsage()	×
472		×
473	if usage.InputTokens > 0 \|\| usage.ReplyTokens > 0 {	×
474	total := usage.InputTokens + usage.ReplyTokens	×
475	r.println(fmt.Sprintf(" 📈 Tokens: %s total",	×
476	formatNumber(total)))	×
477		×
478	// Show per-profile breakdown	×
479	for _, profile := range profiles {	×
480	if profile.Usage.InputTokens > 0 \|\| profile.Usage.ReplyTokens > 0 {	×
481	profileTotal := profile.Usage.InputTokens + profile.Usage.ReplyTokens	×
482	r.println(fmt.Sprintf(" └─ %s: %s (input: %s \| output: %s)",	×
483	profile.Name,	×
484	formatNumber(profileTotal),	×
485	formatNumber(profile.Usage.InputTokens),	×
486	formatNumber(profile.Usage.ReplyTokens)))	×
487	}	×
488	}
489	}
490	} else if r.stats.TokensInput > 0 \|\| r.stats.TokensOutput > 0 {	×
491	// Fall back to step-level token tracking (legacy)	×
492	total := r.stats.TokensInput + r.stats.TokensOutput	×
493	r.println(fmt.Sprintf(" 📈 Tokens: %s (input: %s \| output: %s)",	×
494	formatNumber(total),	×
495	formatNumber(r.stats.TokensInput),	×
496	formatNumber(r.stats.TokensOutput)))	×
497	}	×
498
499	r.println(fmt.Sprintf(" ⏱️ Duration: %s", formatDuration(duration)))	×
500		×
501	// Add visual separator before output	×
502	r.println("")	×
503	r.println("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")	×
504	r.println("")	×
505	}
506
507	// SummaryQuick prints a compact one-line summary
508	func (r *Reporter) SummaryQuick(docCount int, duration time.Duration) {	×
509	tokens := r.stats.TokensInput + r.stats.TokensOutput	×
510	if tokens > 0 {	×
511	r.println(fmt.Sprintf("\n📊 Summary: %d document(s), %s tokens, %s",	×
512	docCount, formatNumber(tokens), formatDuration(duration)))	×
513	} else {	×
514	r.println(fmt.Sprintf("\n📊 Summary: %d document(s), %s",	×
515	docCount, formatDuration(duration)))	×
516	}	×
517	}
518
519	// Helper methods
520
521	// UpdateTokens updates token statistics
522	func (r *Reporter) UpdateTokens(inputTokens, outputTokens int) {	1✔
523	r.mu.Lock()	1✔
524	defer r.mu.Unlock()	1✔
525	r.stats.TokensInput += inputTokens	1✔
526	r.stats.TokensOutput += outputTokens	1✔
527	}	1✔
528
529	// GetStats returns a copy of current statistics
530	func (r *Reporter) GetStats() Stats {	1✔
531	r.mu.Lock()	1✔
532	defer r.mu.Unlock()	1✔
533	return r.stats	1✔
534	}	1✔
535
536	// updateLine updates the current line (mutable output)
537	func (r *Reporter) updateLine(text string) {	1✔
538	if r.quiet {	2✔
539	return	1✔
540	}	1✔
541	r.mu.Lock()	1✔
542	defer r.mu.Unlock()	1✔
543		1✔
544	// Clear previous line	1✔
545	if r.lastLine != "" {	1✔
546	fmt.Fprintf(r.w, "\r\033[K")	×
547	}	×
548
549	fmt.Fprint(r.w, text)	1✔
550	r.lastLine = text	1✔
551	}
552
553	// println writes a new line (finalize any mutable line first)
554	func (r *Reporter) println(text string) {	1✔
555	if r.quiet {	2✔
556	return	1✔
557	}	1✔
558	r.mu.Lock()	1✔
559	defer r.mu.Unlock()	1✔
560		1✔
561	if r.lastLine != "" {	2✔
562	if r.hasThinking {	1✔
563	// In thinking mode: keep the previous line, just move to next line	×
564	fmt.Fprintln(r.w)	×
565	} else {	1✔
566	// Without thinking: clear the mutable line and replace it	1✔
567	fmt.Fprint(r.w, "\r\033[K")	1✔
568	}	1✔
569	r.lastLine = ""	1✔
570	}
571	fmt.Fprintln(r.w, text)	1✔
572	}
573
574	// Formatting helpers
575
576	func formatNumber(n int) string {	1✔
577	if n >= 1000000 {	2✔
578	return fmt.Sprintf("%.1fM", float64(n)/1000000)	1✔
579	}	1✔
580	if n >= 1000 {	2✔
581	return fmt.Sprintf("%.1fK", float64(n)/1000)	1✔
582	}	1✔
583	return fmt.Sprintf("%d", n)	1✔
584	}
585
586	func formatDuration(d time.Duration) string {	1✔
587	if d < time.Second {	2✔
588	return fmt.Sprintf("%.0fms", d.Seconds()*1000)	1✔
589	}	1✔
590	if d < time.Minute {	2✔
591	return fmt.Sprintf("%.1fs", d.Seconds())	1✔
592	}	1✔
593	minutes := int(d.Minutes())	1✔
594	seconds := int(d.Seconds()) % 60	1✔
595	return fmt.Sprintf("%dm %ds", minutes, seconds)	1✔
596	}

fogfish / iq / 20696400416

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