19209726209

Committed 09 Nov 2025 02:14PM UTC coverage: 38.525%. First build

Build # 19209726209

Build Type

Pull #26

github

Committed by

fogfish

Commit Message

support yaml, json as structured input

Pull Request Pull Request #26: Compose AI workflow within the shell

Coverage Stats

1141 of 2873 new or added lines in 35 files covered. (39.71%)

1165 of 3024 relevant lines covered (38.53%)

0.41 hits per line

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

29.39

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

// 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) {
        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()))
                }
        }
        r.stats.TokensInput += tokens
}

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

fogfish / iq / 19209726209

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