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