main
reporter_ui.go
Eric Bower
·
2026-09-15
1package main
2
3import (
4 "context"
5 "fmt"
6 "io"
7 "log/slog"
8 "os"
9 "os/exec"
10 "os/signal"
11 "path/filepath"
12 "regexp"
13 "sort"
14 "strconv"
15 "strings"
16 "sync"
17 "syscall"
18 "time"
19
20 "golang.org/x/sys/unix"
21)
22
23// ---------------------------------------------------------
24// UI State Data Model
25// ---------------------------------------------------------
26
27type TaskUIState struct {
28 Name string
29 Status string // "expected", "running", "success", "failed", "cancelled"
30 Created int64
31 Ended int64
32 Duration time.Duration
33 ExitCode string
34 LastOutput string // Last non-empty line for running peek
35 RunningOutput []string // Captured recent output lines while running (up to 5)
36 ErrorPreview []string // Captured error lines on failure
37 PredictedDur time.Duration
38 PrevStatus string
39 PrevExitCode int
40 FailureStreak int
41 Transition string
42}
43
44type PipelineUIState struct {
45 Repo string
46 Domain string
47 Branch string
48 Commit string
49 JobID string
50 StartedAt time.Time
51 Elapsed time.Duration
52 Tasks map[string]*TaskUIState
53 TaskOrder []string
54 PredictedTotal time.Duration
55 IsComplete bool
56 Cancelled bool
57 Stats *RepoStats
58 mu sync.Mutex
59}
60
61var spinnerFrames = []rune{'⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'}
62
63func newPipelineUIState(repo, branch, commit, jobID string, stats *RepoStats) *PipelineUIState {
64 ui := &PipelineUIState{
65 Repo: repo,
66 Branch: branch,
67 Commit: commit,
68 JobID: jobID,
69 StartedAt: time.Now(),
70 Tasks: make(map[string]*TaskUIState),
71 Stats: stats,
72 }
73
74 if stats != nil {
75 ui.PredictedTotal = time.Duration(stats.AvgWallDurationMs) * time.Millisecond
76
77 // Pre-populate expected tasks sorted by start time
78 var names []string
79 for name := range stats.Tasks {
80 names = append(names, name)
81 }
82 sort.Slice(names, func(i, j int) bool {
83 ti := stats.Tasks[names[i]]
84 tj := stats.Tasks[names[j]]
85 if ti.LastStartedAt != tj.LastStartedAt {
86 return ti.LastStartedAt < tj.LastStartedAt
87 }
88 return names[i] < names[j]
89 })
90
91 for _, name := range names {
92 ts := stats.Tasks[name]
93 ui.Tasks[name] = &TaskUIState{
94 Name: name,
95 Status: "expected",
96 Created: ts.LastStartedAt,
97 PredictedDur: time.Duration(ts.AvgDurationMs) * time.Millisecond,
98 PrevStatus: ts.LastStatus,
99 PrevExitCode: ts.LastExitCode,
100 FailureStreak: ts.FailureStreak,
101 }
102 ui.TaskOrder = append(ui.TaskOrder, name)
103 }
104 }
105
106 return ui
107}
108
109// PruneUnexecuted removes tasks that were expected from previous stats but never actually ran.
110func (ui *PipelineUIState) PruneUnexecuted() {
111 ui.mu.Lock()
112 defer ui.mu.Unlock()
113
114 var newOrder []string
115 for _, name := range ui.TaskOrder {
116 task := ui.Tasks[name]
117 if task != nil && task.Status == "expected" {
118 delete(ui.Tasks, name)
119 } else {
120 newOrder = append(newOrder, name)
121 }
122 }
123 ui.TaskOrder = newOrder
124}
125
126// ---------------------------------------------------------
127
128// UI Helpers & Formatters
129// ---------------------------------------------------------
130
131var ansiRegex = regexp.MustCompile(`\x1b\[[0-9;]*[a-zA-Z]|\x1b\([a-zA-Z]`)
132
133func stripANSI(s string) string {
134 return ansiRegex.ReplaceAllString(s, "")
135}
136
137func getTerminalWidth() int {
138 ws, err := unix.IoctlGetWinsize(int(os.Stdout.Fd()), unix.TIOCGWINSZ)
139 if err == nil && ws.Col > 0 {
140 return int(ws.Col)
141 }
142 if cols := os.Getenv("COLUMNS"); cols != "" {
143 if n, err := strconv.Atoi(cols); err == nil && n > 0 {
144 return n
145 }
146 }
147 return 80
148}
149
150func truncateLine(s string, maxWidth int) string {
151 if maxWidth <= 0 {
152 maxWidth = 80
153 }
154 clean := stripANSI(s)
155 runes := []rune(clean)
156 if len(runes) <= maxWidth {
157 return s
158 }
159 if maxWidth > 3 {
160 return string(runes[:maxWidth-3]) + "..."
161 }
162 return string(runes[:maxWidth])
163}
164
165func extractLastOutputLines(raw string, maxLines int) []string {
166 clean := stripANSI(raw)
167 lines := strings.Split(clean, "\n")
168 var nonEmpties []string
169 for _, l := range lines {
170 line := strings.TrimSpace(l)
171 if line != "" && !strings.HasPrefix(line, "step starting") && !strings.HasPrefix(line, "step completed") {
172 nonEmpties = append(nonEmpties, line)
173 }
174 }
175 if len(nonEmpties) == 0 {
176 return nil
177 }
178 if maxLines > 0 && len(nonEmpties) > maxLines {
179 nonEmpties = nonEmpties[len(nonEmpties)-maxLines:]
180 }
181 return nonEmpties
182}
183
184func extractLastOutputLine(raw string) string {
185 lines := extractLastOutputLines(raw, 1)
186 if len(lines) > 0 {
187 return lines[0]
188 }
189 return ""
190}
191
192func extractErrorPreview(raw string, maxLines int) []string {
193 clean := stripANSI(raw)
194 lines := strings.Split(clean, "\n")
195 var nonEmpties []string
196 for _, l := range lines {
197 trimmed := strings.TrimRight(l, "\r\n ")
198 if trimmed != "" {
199 nonEmpties = append(nonEmpties, trimmed)
200 }
201 }
202 if len(nonEmpties) == 0 {
203 return nil
204 }
205 if len(nonEmpties) > maxLines {
206 nonEmpties = nonEmpties[len(nonEmpties)-maxLines:]
207 }
208 return nonEmpties
209}
210
211func renderProgressBar(elapsed, predicted time.Duration, width int) (string, int) {
212 if width <= 0 {
213 width = 10
214 }
215 if predicted <= 0 {
216 return "[" + strings.Repeat("█", width) + "]", 100
217 }
218
219 pct := int((float64(elapsed) / float64(predicted)) * 100)
220 displayPct := pct
221 if displayPct > 100 {
222 displayPct = 100
223 }
224 if displayPct < 0 {
225 displayPct = 0
226 }
227
228 filled := (displayPct * width) / 100
229 unfilled := width - filled
230 if unfilled < 0 {
231 unfilled = 0
232 }
233 bar := "[" + strings.Repeat("█", filled) + strings.Repeat("░", unfilled) + "]"
234 return bar, displayPct
235}
236
237func renderErrorPreviewBox(lines []string, width int) string {
238 if len(lines) == 0 {
239 return ""
240 }
241 if width < 30 {
242 width = 60
243 }
244
245 var sb strings.Builder
246 headerTitle := "┌─ [error preview] "
247 headerLine := headerTitle + strings.Repeat("─", max(0, width-len(headerTitle)))
248 sb.WriteString(" " + headerLine + "\n")
249
250 for _, line := range lines {
251 if len(line) > width-4 {
252 line = line[:width-7] + "..."
253 }
254 fmt.Fprintf(&sb, " │ %s\n", line)
255 }
256
257 footer := "└" + strings.Repeat("─", max(0, width-1))
258 sb.WriteString(" " + footer)
259 return sb.String()
260}
261
262func formatShortDuration(d time.Duration) string {
263 secs := d.Seconds()
264 if secs < 0 {
265 secs = 0
266 }
267 if secs >= 60 {
268 m := int(secs) / 60
269 s := int(secs) % 60
270 return fmt.Sprintf("%dm%ds", m, s)
271 }
272 if secs >= 10 {
273 return fmt.Sprintf("%.1fs", secs)
274 }
275 return fmt.Sprintf("%.1fs", secs)
276}
277
278// ---------------------------------------------------------
279// Full Interactive UI State Rendering
280// ---------------------------------------------------------
281
282func (ui *PipelineUIState) Render(frame int) string {
283 ui.mu.Lock()
284 defer ui.mu.Unlock()
285
286 var sb strings.Builder
287 commitShort := ui.Commit
288 if len(commitShort) > 7 {
289 commitShort = commitShort[:7]
290 }
291 if commitShort == "" {
292 commitShort = "HEAD"
293 }
294 branch := ui.Branch
295 if branch == "" {
296 branch = "local"
297 }
298
299 // Header line 1: repo info
300 fmt.Fprintf(&sb, "🚀 pici: %s (branch: %s • %s)\n", ui.Repo, branch, commitShort)
301
302 // Calculate counts
303 totalTasks := len(ui.TaskOrder)
304 completedCount := 0
305 failedCount := 0
306 for _, name := range ui.TaskOrder {
307 task := ui.Tasks[name]
308 if task.Status == "success" || task.Status == "failed" {
309 completedCount++
310 }
311 if task.Status == "failed" {
312 failedCount++
313 }
314 }
315
316 elapsed := ui.Elapsed
317 if elapsed == 0 {
318 elapsed = time.Since(ui.StartedAt)
319 }
320
321 // Header line 2: timing & status
322 fmt.Fprintf(&sb, "⏱ Running for %s [ ", formatShortDuration(elapsed))
323 if failedCount > 0 {
324 fmt.Fprintf(&sb, "%d/%d completed • %d failed ]\n", completedCount, totalTasks, failedCount)
325 } else if completedCount == 0 {
326 estTotal := "~" + formatShortDuration(ui.PredictedTotal)
327 if ui.PredictedTotal == 0 {
328 estTotal = "estimating"
329 }
330 fmt.Fprintf(&sb, "0/%d tasks • est. %s ]\n", totalTasks, estTotal)
331 } else {
332 rem := ui.PredictedTotal - elapsed
333 if rem < 0 {
334 rem = 0
335 }
336 fmt.Fprintf(&sb, "%d/%d completed • est. ~%s remaining ]\n", completedCount, totalTasks, formatShortDuration(rem))
337 }
338
339 sb.WriteString("\n")
340
341 spinnerRune := spinnerFrames[frame%len(spinnerFrames)]
342
343 // Render task list
344 for _, name := range ui.TaskOrder {
345 task := ui.Tasks[name]
346 switch task.Status {
347 case "expected":
348 pred := ""
349 if task.PredictedDur > 0 {
350 pred = fmt.Sprintf("~%s", formatShortDuration(task.PredictedDur))
351 }
352 notes := ""
353 if task.PrevStatus == "failed" {
354 notes = " • ⚠️ failed last run"
355 }
356 detail := ""
357 if pred != "" || notes != "" {
358 detail = fmt.Sprintf("(expected %s%s)", pred, notes)
359 }
360 fmt.Fprintf(&sb, " ◌ %-8s %s\n", task.Name, detail)
361
362 case "running":
363 taskDur := task.Duration
364 if taskDur == 0 && task.Created > 0 {
365 taskDur = time.Duration(time.Now().Unix()-task.Created) * time.Second
366 }
367 predStr := ""
368 var bar string
369 var pct int
370 if task.PredictedDur > 0 {
371 predStr = fmt.Sprintf("/ ~%s", formatShortDuration(task.PredictedDur))
372 bar, pct = renderProgressBar(taskDur, task.PredictedDur, 10)
373 }
374 extraNote := ""
375 if task.PrevStatus == "failed" {
376 extraNote = " (retesting previous failure)"
377 }
378
379 barPart := ""
380 if bar != "" {
381 barPart = fmt.Sprintf(" %s %d%%%s", bar, pct, extraNote)
382 }
383
384 fmt.Fprintf(&sb, " %c %-8s %s %s%s\n", spinnerRune, task.Name, formatShortDuration(taskDur), predStr, barPart)
385 if len(task.RunningOutput) > 0 {
386 for i, line := range task.RunningOutput {
387 if i == len(task.RunningOutput)-1 {
388 fmt.Fprintf(&sb, " └─ %s\n", line)
389 } else {
390 fmt.Fprintf(&sb, " │ %s\n", line)
391 }
392 }
393 } else if task.LastOutput != "" {
394 fmt.Fprintf(&sb, " └─ %s\n", task.LastOutput)
395 }
396
397 case "success":
398 transitionNote := ""
399 if task.Transition == "fixed" {
400 transitionNote = " • 🎉 fixed!"
401 }
402 durStr := formatShortDuration(task.Duration)
403 if transitionNote != "" {
404 fmt.Fprintf(&sb, " ✔ %-8s %s (%s)\n", task.Name, durStr, strings.TrimPrefix(transitionNote, " • "))
405 } else {
406 fmt.Fprintf(&sb, " ✔ %-8s %s\n", task.Name, durStr)
407 }
408
409 case "failed":
410 notes := ""
411 switch task.Transition {
412 case "still_failing":
413 notes = " • ⚠️ still failing"
414 case "new_failure":
415 notes = " • ❌ new failure"
416 }
417 exitStr := "exit 1"
418 if task.ExitCode != "" {
419 exitStr = "exit " + task.ExitCode
420 }
421 fmt.Fprintf(&sb, " ✖ %-8s %s (%s%s)\n", task.Name, formatShortDuration(task.Duration), exitStr, notes)
422 if len(task.ErrorPreview) > 0 {
423 sb.WriteString(renderErrorPreviewBox(task.ErrorPreview, 65))
424 sb.WriteString("\n")
425 }
426
427 case "cancelled":
428 fmt.Fprintf(&sb, " ✖ %-8s cancelled\n", task.Name)
429 }
430 }
431
432 return sb.String()
433}
434
435// RenderSummary renders the post-run completion summary.
436func (ui *PipelineUIState) RenderSummary(reportPath string, exitCode int) string {
437 ui.mu.Lock()
438 defer ui.mu.Unlock()
439
440 var sb strings.Builder
441 failedCount := 0
442 var failedTasks []*TaskUIState
443
444 for _, name := range ui.TaskOrder {
445 task := ui.Tasks[name]
446 if task.Status == "failed" {
447 failedCount++
448 failedTasks = append(failedTasks, task)
449 }
450 }
451
452 if exitCode != 0 && failedCount == 0 {
453 failedCount = 1
454 }
455
456 elapsed := ui.Elapsed
457 if elapsed == 0 {
458 elapsed = time.Since(ui.StartedAt)
459 }
460 wallStr := formatShortDuration(elapsed)
461
462 if exitCode == 0 && failedCount == 0 {
463 var succeededTasks []*TaskUIState
464 for _, name := range ui.TaskOrder {
465 task := ui.Tasks[name]
466 if task != nil && task.Status == "success" {
467 succeededTasks = append(succeededTasks, task)
468 }
469 }
470 fmt.Fprintf(&sb, "✔ All %d tasks succeeded (total time: %s)\n\n", len(succeededTasks), wallStr)
471 for _, task := range succeededTasks {
472 avgStr := ""
473 if task.PredictedDur > 0 {
474 avgStr = fmt.Sprintf("(avg %s", formatShortDuration(task.PredictedDur))
475 if task.Transition == "fixed" {
476 avgStr += " • 🎉 fixed!)"
477 } else {
478 avgStr += ")"
479 }
480 }
481 fmt.Fprintf(&sb, " ✔ %-9s %6s %s\n", task.Name, formatShortDuration(task.Duration), avgStr)
482 }
483 } else {
484 taskWord := "task"
485 if failedCount > 1 {
486 taskWord = "tasks"
487 }
488 fmt.Fprintf(&sb, "❌ %d %s failed (total time: %s)\n\n", failedCount, taskWord, wallStr)
489 for _, name := range ui.TaskOrder {
490 task := ui.Tasks[name]
491 switch task.Status {
492 case "success":
493 fmt.Fprintf(&sb, " ✔ %-9s %6s\n", task.Name, formatShortDuration(task.Duration))
494 case "failed":
495 exitStr := "exit " + task.ExitCode
496 if task.ExitCode == "" {
497 exitStr = "exit 1"
498 }
499 fmt.Fprintf(&sb, " ✖ %-9s %6s %s\n", task.Name, formatShortDuration(task.Duration), exitStr)
500 if len(task.ErrorPreview) > 0 {
501 sb.WriteString(renderErrorPreviewBox(task.ErrorPreview, 65))
502 sb.WriteString("\n")
503 }
504 default:
505 fmt.Fprintf(&sb, " ✖ %-9s cancelled\n", task.Name)
506 }
507 }
508 }
509
510 sb.WriteString("\nArtifacts:\n")
511 if reportPath != "" {
512 fmt.Fprintf(&sb, " 📄 Report: %s\n", reportPath)
513 }
514 domain := ui.Domain
515 if domain == "" {
516 domain = "local"
517 }
518 if failedCount > 0 {
519 if len(failedTasks) > 0 {
520 for _, ft := range failedTasks {
521 fmt.Fprintf(&sb, " 🔍 Debug: zmx attach %s.%s.%s.step.%s\n", domain, ui.Repo, ui.JobID, ft.Name)
522 }
523 } else {
524 fmt.Fprintf(&sb, " 🔍 Debug: zmx attach %s.%s.%s.runner\n", domain, ui.Repo, ui.JobID)
525 }
526 }
527
528 return sb.String()
529}
530
531// ---------------------------------------------------------
532// Non-TTY Plain Stream Reporter
533// ---------------------------------------------------------
534
535type plainStreamReporter struct {
536 out io.Writer
537 ui *PipelineUIState
538 log *slog.Logger
539}
540
541func (p *plainStreamReporter) OnTaskStart(name string) {
542 fmt.Fprintf(p.out, "▶ starting %s\n", name) //nolint:errcheck
543}
544
545func (p *plainStreamReporter) OnTaskOutput(name, line string) {
546 if line != "" {
547 fmt.Fprintf(p.out, " [%s] %s\n", name, line) //nolint:errcheck
548 }
549}
550
551func (p *plainStreamReporter) OnTaskComplete(name string, exitCode int, dur time.Duration, errorLines []string) {
552 durStr := formatShortDuration(dur)
553 if exitCode == 0 {
554 fmt.Fprintf(p.out, "✔ %s succeeded (%s)\n", name, durStr) //nolint:errcheck
555 } else {
556 fmt.Fprintf(p.out, "✖ %s failed with exit %d (%s)\n", name, exitCode, durStr) //nolint:errcheck
557 if len(errorLines) > 0 {
558 for _, line := range errorLines {
559 fmt.Fprintf(p.out, " │ %s\n", line) //nolint:errcheck
560 }
561 }
562 }
563}
564
565// ---------------------------------------------------------
566// Interactive Controller Loop
567// ---------------------------------------------------------
568
569type interactiveController struct {
570 ctx context.Context
571 out io.Writer
572 ui *PipelineUIState
573 isTTY bool
574 rendered int // number of lines rendered in previous frame for cursor overwrite
575 streamer *plainStreamReporter
576}
577
578func newInteractiveController(ctx context.Context, out io.Writer, repo, branch, commit, jobID string, stats *RepoStats, isTTY bool) *interactiveController {
579 if ctx == nil {
580 ctx = context.Background()
581 }
582 ui := newPipelineUIState(repo, branch, commit, jobID, stats)
583 return &interactiveController{
584 ctx: ctx,
585 out: out,
586 ui: ui,
587 isTTY: isTTY,
588 streamer: &plainStreamReporter{out: out, ui: ui},
589 }
590}
591
592func (c *interactiveController) WaitUntilDone(fetchSessions func() ([]SessionInfo, bool)) error {
593 sigCh := make(chan os.Signal, 1)
594 signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
595 defer signal.Stop(sigCh)
596
597 ticker := time.NewTicker(150 * time.Millisecond)
598 defer ticker.Stop()
599
600 frame := 0
601
602 // Initial render
603 c.redraw(c.ui.Render(frame))
604
605 prevStatus := make(map[string]string)
606 prevOutput := make(map[string]string)
607
608 for {
609 select {
610 case <-c.ctx.Done():
611 c.ui.Cancelled = true
612 c.clear()
613 fmt.Fprintln(c.out, "\n⏹ cancelling job and terminating active sessions...") //nolint:errcheck
614 return c.ctx.Err()
615
616 case <-sigCh:
617 c.ui.Cancelled = true
618 c.clear()
619 fmt.Fprintln(c.out, "\n^C") //nolint:errcheck
620 fmt.Fprintln(c.out, "⏹ cancelling job and terminating active sessions...") //nolint:errcheck
621 return fmt.Errorf("job cancelled by signal")
622
623 case <-ticker.C:
624 frame++
625 sessions, isComplete := fetchSessions()
626
627 // Update UI state with current sessions
628 c.ui.mu.Lock()
629 for _, s := range sessions {
630 name := s.Short
631 if name == "runner" || name == "" {
632 continue
633 }
634
635 task, exists := c.ui.Tasks[name]
636 if !exists {
637 task = &TaskUIState{
638 Name: name,
639 Status: "running",
640 }
641 c.ui.Tasks[name] = task
642 c.ui.TaskOrder = append(c.ui.TaskOrder, name)
643 }
644
645 created, _ := strconv.ParseInt(s.Created, 10, 64)
646 ended, _ := strconv.ParseInt(s.Ended, 10, 64)
647 task.Created = created
648 task.Ended = ended
649
650 oldState := prevStatus[name]
651 if s.Ended == "" {
652 task.Status = "running"
653 if created > 0 {
654 task.Duration = time.Duration(time.Now().Unix()-created) * time.Second
655 }
656 if !c.isTTY && (oldState == "" || oldState == "expected") {
657 c.streamer.OnTaskStart(name)
658 }
659 if !c.isTTY && task.LastOutput != "" && task.LastOutput != prevOutput[name] {
660 c.streamer.OnTaskOutput(name, task.LastOutput)
661 prevOutput[name] = task.LastOutput
662 }
663 } else {
664 task.ExitCode = s.ExitCode
665 exitCodeInt, _ := strconv.Atoi(s.ExitCode)
666 if s.ExitCode == "0" {
667 task.Status = "success"
668 } else {
669 task.Status = "failed"
670 }
671 if created > 0 && ended > 0 {
672 task.Duration = time.Duration(ended-created) * time.Second
673 }
674 if task.Transition == "" && c.ui.Stats != nil {
675 task.Transition = computeOutcomeTransition(c.ui.Stats.Tasks[name], exitCodeInt)
676 }
677 if !c.isTTY && oldState != task.Status {
678 c.streamer.OnTaskComplete(name, exitCodeInt, task.Duration, task.ErrorPreview)
679 }
680 }
681 prevStatus[name] = task.Status
682 }
683 c.ui.mu.Unlock()
684
685 c.redraw(c.ui.Render(frame))
686
687 if isComplete {
688 c.ui.Elapsed = time.Since(c.ui.StartedAt)
689 c.ui.IsComplete = true
690 c.clear()
691 return nil
692 }
693 }
694 }
695}
696
697func (c *interactiveController) redraw(outStr string) {
698 if !c.isTTY {
699 return
700 }
701 trimmed := strings.TrimRight(outStr, "\n")
702 rawLines := strings.Split(trimmed, "\n")
703 if len(rawLines) == 0 || (len(rawLines) == 1 && rawLines[0] == "") {
704 return
705 }
706
707 termWidth := getTerminalWidth()
708 lines := make([]string, len(rawLines))
709 for i, l := range rawLines {
710 lines[i] = truncateLine(l, termWidth-1)
711 }
712
713 var b strings.Builder
714
715 // Move cursor up from the bottom of previous render block to the top line
716 if c.rendered > 0 {
717 fmt.Fprintf(&b, "\r\033[%dA", c.rendered)
718 }
719
720 // Print each line, clearing to the end of the line and resetting carriage
721 for i, line := range lines {
722 b.WriteString("\r\033[2K") // clear entire line
723 b.WriteString(line)
724 if i < len(lines)-1 {
725 b.WriteString("\n")
726 }
727 }
728
729 // If the previous frame had more lines, clear the leftover trailing lines
730 if c.rendered > len(lines)-1 {
731 extra := c.rendered - (len(lines) - 1)
732 for i := 0; i < extra; i++ {
733 b.WriteString("\n\r\033[2K")
734 }
735 fmt.Fprintf(&b, "\r\033[%dA", extra)
736 }
737
738 c.rendered = len(lines) - 1
739 fmt.Fprint(c.out, b.String()) //nolint:errcheck
740}
741
742func (c *interactiveController) clear() {
743 if !c.isTTY || c.rendered < 0 {
744 return
745 }
746 var b strings.Builder
747 if c.rendered > 0 {
748 fmt.Fprintf(&b, "\r\033[%dA", c.rendered) // move cursor up
749 }
750 for i := 0; i <= c.rendered; i++ {
751 b.WriteString("\r\033[2K") // erase entire line
752 if i < c.rendered {
753 b.WriteString("\n")
754 }
755 }
756 if c.rendered > 0 {
757 fmt.Fprintf(&b, "\r\033[%dA", c.rendered)
758 }
759 c.rendered = 0
760 fmt.Fprint(c.out, b.String()) //nolint:errcheck
761}
762
763// ---------------------------------------------------------
764// Interactive waitAndReport Entrypoint
765// ---------------------------------------------------------
766
767func isTerminal(w io.Writer) bool {
768 if os.Getenv("CI") != "" || os.Getenv("TERM") == "dumb" {
769 return false
770 }
771 if f, ok := w.(*os.File); ok {
772 stat, err := f.Stat()
773 if err == nil && (stat.Mode()&os.ModeCharDevice) != 0 {
774 return true
775 }
776 }
777 return false
778}
779
780// waitAndReportUI executes the rich interactive reporting loop, updating stats.json.
781func waitAndReportUI(cfg *Cfg, log *slog.Logger, repoName, jobID, domain, eventType, branch, commit string) error {
782 if domain == "" {
783 domain = "local"
784 }
785 if eventType == "" {
786 eventType = "local"
787 }
788 prefix := fmt.Sprintf("%s.%s.%s.", domain, repoName, jobID)
789
790 stats, _ := loadRepoStats(repoName, eventType)
791 isTTY := isTerminal(os.Stdout)
792
793 ctrl := newInteractiveController(cfg.Ctx, os.Stdout, repoName, branch, commit, jobID, stats, isTTY)
794 ctrl.ui.Domain = domain
795
796 lastStage := time.Now()
797
798 // Session poller & peek fetcher
799 fetchSessions := func() ([]SessionInfo, bool) {
800 listOutput, err := exec.Command("zmx", "list").CombinedOutput()
801 if err != nil {
802 if log != nil {
803 log.Error("zmx list", "err", err)
804 }
805 return nil, false
806 }
807
808 sessions := parseZMXList(string(listOutput))
809 var jobSessions []SessionInfo
810 for _, s := range sessions {
811 if strings.HasPrefix(s.Name, prefix) {
812 s.Short = cleanSessionShort(s.Name, prefix, repoName, jobID)
813 jobSessions = append(jobSessions, s)
814 }
815 }
816
817 // Update tail history for live activity and error peeks
818 for _, s := range jobSessions {
819 if s.Short == "runner" || s.Short == "" {
820 continue
821 }
822 hist, err := fetchHistoryPlain(s.Name)
823 if err == nil && hist != "" {
824 ctrl.ui.mu.Lock()
825 if task, ok := ctrl.ui.Tasks[s.Short]; ok {
826 if s.Ended == "" {
827 task.RunningOutput = extractLastOutputLines(hist, 5)
828 if len(task.RunningOutput) > 0 {
829 task.LastOutput = task.RunningOutput[len(task.RunningOutput)-1]
830 }
831 } else if s.ExitCode != "0" && len(task.ErrorPreview) == 0 {
832 task.ErrorPreview = extractErrorPreview(hist, 5)
833 }
834 }
835 ctrl.ui.mu.Unlock()
836 }
837 }
838
839 // Stage local artifacts periodically (every 2 seconds) so reports stay updated live without disk trashing
840 if time.Since(lastStage) >= 2*time.Second {
841 _ = stageLocalArtifacts(cfg, log, repoName, jobID, domain)
842 lastStage = time.Now()
843 }
844
845 return jobSessions, isJobComplete(jobSessions)
846 }
847
848 waitErr := ctrl.WaitUntilDone(fetchSessions)
849 if waitErr != nil {
850 cancelJobSessions(prefix)
851 if ctrl.ui.Cancelled {
852 for _, task := range ctrl.ui.Tasks {
853 if task.Status == "running" {
854 task.Status = "cancelled"
855 }
856 }
857 fmt.Fprintln(os.Stdout, "⏹ job cancelled") //nolint:errcheck
858 }
859 return waitErr
860 }
861
862 // Fetch final session list
863 var finalSessions []SessionInfo
864 if listOutput, err := exec.Command("zmx", "list").CombinedOutput(); err == nil {
865 for _, s := range parseZMXList(string(listOutput)) {
866 if strings.HasPrefix(s.Name, prefix) {
867 s.Short = cleanSessionShort(s.Name, prefix, repoName, jobID)
868 finalSessions = append(finalSessions, s)
869 }
870 }
871 }
872
873 for _, s := range finalSessions {
874 if s.Short == "runner" || s.Short == "" {
875 continue
876 }
877 if s.ExitCode != "0" && s.ExitCode != "" {
878 ctrl.ui.mu.Lock()
879 if task, ok := ctrl.ui.Tasks[s.Short]; ok && len(task.ErrorPreview) == 0 {
880 if hist, err := fetchHistoryPlain(s.Name); err == nil && hist != "" {
881 task.ErrorPreview = extractErrorPreview(hist, 5)
882 }
883 }
884 ctrl.ui.mu.Unlock()
885 }
886 }
887
888 exitCode, overallStatus := resolveJobExitCode(finalSessions)
889
890 if overallStatus == "success" {
891 ctrl.ui.PruneUnexecuted()
892 }
893
894 // Update historical stats.json
895 if stats != nil {
896 if overallStatus == "success" {
897 currentTasks := make(map[string]bool)
898 for _, s := range finalSessions {
899 if s.Short != "runner" && s.Short != "" {
900 currentTasks[s.Short] = true
901 }
902 }
903 ctrl.ui.mu.Lock()
904 for name, t := range ctrl.ui.Tasks {
905 if t.Status != "expected" {
906 currentTasks[name] = true
907 }
908 }
909 ctrl.ui.mu.Unlock()
910
911 pruneStaleTasks(stats, currentTasks)
912 }
913
914 for _, s := range finalSessions {
915 if s.Short == "runner" || s.Short == "" {
916 continue
917 }
918 created, _ := strconv.ParseInt(s.Created, 10, 64)
919 ended, _ := strconv.ParseInt(s.Ended, 10, 64)
920 dur := time.Duration(ended-created) * time.Second
921 code, _ := strconv.Atoi(s.ExitCode)
922 updateTaskStats(stats, s.Short, dur, code, created)
923 }
924 updateRepoStats(stats, ctrl.ui.Elapsed, overallStatus)
925 _ = saveRepoStats(stats)
926 }
927
928 // Output summary
929 reportPath := filepath.Join(cfg.ArtifactDir, repoName, jobID, "index.html")
930 fmt.Fprintln(os.Stdout) //nolint:errcheck
931 fmt.Fprint(os.Stdout, ctrl.ui.RenderSummary(reportPath, exitCode)) //nolint:errcheck
932
933 // Debug bridge if failed
934 topic := fmt.Sprintf("%s.%s.%s", domain, repoName, jobID)
935 if exitCode != 0 && cfg.DebugOnFail {
936 cwd, _ := os.Getwd()
937 debugCfg := DebugBridgeConfig{
938 Topic: topic,
939 KeyLocation: cfg.KeyLocation,
940 CertLocation: cfg.CertificateLocation,
941 WorkDir: cwd,
942 Timeout: cfg.DebugTimeout,
943 Logger: log,
944 Stdout: os.Stdout,
945 }
946 if err := startDebugBridge(cfg.Ctx, debugCfg); err != nil && log != nil {
947 log.Error("debug bridge failed", "err", err)
948 }
949 }
950
951 if cfg.SummaryFile != "" {
952 knownStates := make(map[string]*sessionState)
953 for _, s := range finalSessions {
954 st := &sessionState{
955 status: "failed",
956 exitCode: s.ExitCode,
957 }
958 if s.ExitCode == "0" {
959 st.status = "success"
960 }
961 knownStates[s.Short] = st
962 }
963 if err := writeSummaryMarkdown(cfg.SummaryFile, repoName, jobID, finalSessions, knownStates, cfg.DebugOnFail && exitCode != 0, topic); err != nil && log != nil {
964 log.Error("write summary markdown", "err", err, "file", cfg.SummaryFile)
965 }
966 }
967
968 if exitCode != 0 {
969 return &JobFailedError{ExitCode: exitCode}
970 }
971
972 return nil
973}