feat: reporter ui and plain

Commit `1a7066e`

Eric Bower · 2026-08-16 09:45:42 -0400 EDT

parent 50d7d59

feat: reporter ui and plain

7 files changed, +1704, -211

M main.go

A reporter_plain.go

A reporter_plain_test.go

A reporter_ui.go

A reporter_ui_test.go

A stats.go

A stats_test.go

M main.go

+10, -211

......@@ -16,13 +16,11 @@ import (
	"log/slog"
	"os"
	"os/exec"
	"os/signal"
	"path/filepath"
	"runtime"
	"sort"
	"strconv"
	"strings"
	"syscall"
	"time"
)

......@@ -1027,7 +1025,7 @@ See: https://github.com/picosh/pici
	fmt.Fprintln(os.Stdout, "✅ job launched") //nolint:errcheck

	if cfg.Wait {
		if err := waitAndReport(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
		if err := waitAndReportPlain(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
			runErr = err
			return fmt.Errorf("wait: %w", err)
		}

......@@ -1043,210 +1041,12 @@ See: https://github.com/picosh/pici
}

// waitAndReport polls the job's sessions until all complete, prints live
// progress to stdout, then dumps session history and a final summary.
// interactive progress to stdout, updates stats.json, and outputs a final summary.
func waitAndReport(cfg *Cfg, log *slog.Logger, name, jobID, eventType string) error {
	domain := getDomain(eventType)
	prefix := domain + "." + name + "." + jobID + "."
	interval := cfg.MonitorInterval
	if interval <= 0 {
		interval = 5 * time.Second
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	// Handle ^C and termination signals gracefully
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	defer signal.Stop(sigCh)

	fmt.Fprintln(os.Stdout)                                //nolint:errcheck
	fmt.Fprint(os.Stdout, "⏳ waiting for completion...\n") //nolint:errcheck

	// Track state for each session
	known := make(map[string]*sessionState)
	var jobSessions []SessionInfo
	var sessionOrder []string // track insertion order for deterministic output
	var liveLines []string    // last set of status lines printed (for overwrite)

	var done <-chan struct{}
	if cfg.Ctx != nil {
		done = cfg.Ctx.Done()
	}

	for {
		select {
		case <-done:
			cancelJobSessions(prefix)
			fmt.Fprintln(os.Stdout, "\n⏹ job cancelled") //nolint:errcheck
			return cfg.Ctx.Err()
		case <-sigCh:
			cancelJobSessions(prefix)
			fmt.Fprintln(os.Stdout, "\n⏹ job cancelled") //nolint:errcheck
			return fmt.Errorf("job cancelled by signal")
		case <-ticker.C:
		}

		// Fetch current session list
		listOutput, err := exec.Command("zmx", "list").CombinedOutput()
		if err != nil {
			log.Error("zmx list", "err", err)
			continue
		}

		sessions := parseZMXList(string(listOutput))
		jobSessions = nil
		for _, s := range sessions {
			if strings.HasPrefix(s.Name, prefix) {
				s.Short = cleanSessionShort(s.Name, prefix, name, jobID)
				jobSessions = append(jobSessions, s)
			}
		}

		if len(jobSessions) == 0 {
			continue // runner session may not have appeared yet
		}

		// Update state for each session
		for _, s := range jobSessions {
			state, ok := known[s.Short]
			if !ok {
				state = &sessionState{}
				known[s.Short] = state
				sessionOrder = append(sessionOrder, s.Short)
			}

			if s.Ended == "" {
				state.status = "running"
				created, _ := strconv.ParseInt(s.Created, 10, 64)
				state.duration = fmtDurationTs(created, time.Now().Unix())
			} else {
				if s.ExitCode == "0" {
					state.status = "success"
				} else {
					state.status = "failed"
					state.exitCode = s.ExitCode
				}
				state.duration = fmtDuration(s.Created, s.Ended)
				state.printed = true // lock final state
			}
		}

		// Build current status lines
		lines := make([]string, 0, len(sessionOrder))
		for _, short := range sessionOrder {
			state := known[short]
			icon := map[string]string{"running": "🚀", "success": "✅", "failed": "❌"}[state.status]
			detail := ""
			if state.status == "failed" && state.exitCode != "" {
				detail = fmt.Sprintf(", exit %s", state.exitCode)
			}
			if state.duration != "" && state.duration != "—" {
				detail += fmt.Sprintf(" (%s)", state.duration)
			}
			lines = append(lines, fmt.Sprintf("   %-12s %s %s%s", short, icon, state.status, detail))
		}

		// Overwrite previous lines with cursor-up, or print fresh
		if len(liveLines) > 0 {
			// Move cursor up to overwrite previous lines //nolint:errcheck
			for range len(liveLines) {
				fmt.Fprint(os.Stdout, "\033[A") //nolint:errcheck
			}
			// Clear each line
			for i, line := range lines { //nolint:errcheck
				if i > 0 {
					fmt.Fprint(os.Stdout, "\n") //nolint:errcheck
				}
				fmt.Fprint(os.Stdout, line+"\033[K") //nolint:errcheck
			}
			fmt.Fprint(os.Stdout, "\n") //nolint:errcheck
		} else { //nolint:errcheck
			for _, line := range lines {
				fmt.Fprintln(os.Stdout, line) //nolint:errcheck
			}
		}
		liveLines = lines

		// Stage local artifacts on tick so progress and HTML reports update live
		_ = stageLocalArtifacts(cfg, log, name, jobID, eventType)

		// Check if all sessions are done and the runner session has finished
		if isJobComplete(jobSessions) {
			break
		}
	}

	// Print last 25 lines of history for failed sessions only
	fmt.Fprintln(os.Stdout) //nolint:errcheck
	for _, s := range jobSessions {
		state := known[s.Short]
		if state == nil || state.status != "failed" {
			continue
		}

		separator := strings.Repeat("\u2500", 50)
		fmt.Fprintln(os.Stdout, separator)                                         //nolint:errcheck
		fmt.Fprintf(os.Stdout, "Session: %s (exit %s)\n", s.Short, state.exitCode) //nolint:errcheck
		fmt.Fprintln(os.Stdout, separator)                                         //nolint:errcheck
		fmt.Fprintln(os.Stdout)                                                    //nolint:errcheck

		history, err := fetchHistoryPlain(s.Name)
		if err != nil {
			fmt.Fprintf(os.Stdout, "   (history unavailable: %v)\n", err) //nolint:errcheck
		} else {
			lines := strings.Split(history, "\n")
			// Show last 25 lines
			if len(lines) > 25 {
				fmt.Fprintf(os.Stdout, "   ... (%d lines omitted)\n", len(lines)-25) //nolint:errcheck
				lines = lines[len(lines)-25:]
			}
			for _, line := range lines {
				if line != "" {
					fmt.Fprintf(os.Stdout, "   %s\n", line) //nolint:errcheck
				}
			}
		}
		fmt.Fprintln(os.Stdout) //nolint:errcheck
	}

	// Final summary
	exitCode, status := resolveJobExitCode(jobSessions)
	_, _, duration := computeJobTiming(jobSessions)
	icon := map[string]string{"success": "✅", "failed": "❌"}[status]
	if exitCode == 0 {
		fmt.Fprintf(os.Stdout, "%s job finished: %s (%s)\n", icon, status, duration) //nolint:errcheck
	} else {
		fmt.Fprintf(os.Stdout, "%s job failed: exit %d (%s)\n", icon, exitCode, duration) //nolint:errcheck
	}

	topic := fmt.Sprintf("%s.%s.%s", domain, name, jobID)
	if exitCode != 0 && cfg.DebugOnFail {
		cwd, _ := os.Getwd()
		debugCfg := DebugBridgeConfig{
			Topic:        topic,
			KeyLocation:  cfg.KeyLocation,
			CertLocation: cfg.CertificateLocation,
			WorkDir:      cwd,
			Timeout:      cfg.DebugTimeout,
			Logger:       log,
			Stdout:       os.Stdout,
		}
		if err := startDebugBridge(cfg.Ctx, debugCfg); err != nil {
			log.Error("debug bridge failed", "err", err)
		}
	}

	if cfg.SummaryFile != "" {
		if err := writeSummaryMarkdown(cfg.SummaryFile, name, jobID, jobSessions, known, cfg.DebugOnFail && exitCode != 0, topic); err != nil {
			log.Error("write summary markdown", "err", err, "file", cfg.SummaryFile)
		}
	}

	if exitCode != 0 {
		return &JobFailedError{ExitCode: exitCode}
	}

	return nil
	cwd, _ := os.Getwd()
	branch := detectGitBranch(cwd)
	commit := detectGitCommit(cwd)
	return waitAndReportUI(cfg, log, name, jobID, eventType, branch, commit)
}

func writeSummaryMarkdown(filePath string, repoName, jobID string, jobSessions []SessionInfo, known map[string]*sessionState, debugActive bool, debugTopic string) error {

......@@ -1351,7 +1151,6 @@ type sessionState struct {
	status   string // "running", "success", "failed"
	exitCode string
	duration string
	printed  bool // true once final state (success/failed) is printed
}

// monitorJobState tracks display state for a single job across ticks.

......@@ -2630,14 +2429,14 @@ func runLocal(cfg *Cfg, dest string) error {
		}
	}()

	fmt.Fprintf(os.Stdout, "🚀 starting local job local.%s.%s\n", eventData.Name, jobID) //nolint:errcheck
	log.Info("starting local job", "job_id", jobID)
	if err := eng.Setup(); err != nil {
		return fmt.Errorf("workspace setup: %w", err)
	}
	if cfg.InPlace {
		fmt.Fprintf(os.Stdout, "📁 using in-place workspace directory %s\n", eng.Wk.GetDir()) //nolint:errcheck
		log.Debug("using in-place workspace directory", "dir", eng.Wk.GetDir())
	} else {
		fmt.Fprintf(os.Stdout, "📦 syncing workspace to temp directory %s\n", eng.Wk.GetDir()) //nolint:errcheck
		log.Debug("syncing workspace to temp directory", "dir", eng.Wk.GetDir())
	}
	log.Debug("workspace directory", "dir", eng.Wk.GetDir())

......@@ -2674,7 +2473,7 @@ func runLocal(cfg *Cfg, dest string) error {
		return err
	}

	fmt.Fprintln(os.Stdout, "🏃 launching sessions...") //nolint:errcheck
	log.Debug("launching sessions")
	if err := eng.Run(manifest); err != nil {
		return fmt.Errorf("run: %w", err)
	}

A reporter_plain.go

+145, -0

......@@ -0,0 +1,145 @@
package main

import (
	"fmt"
	"log/slog"
	"os"
	"os/exec"
	"os/signal"
	"path/filepath"
	"strconv"
	"strings"
	"syscall"
	"time"
)

// waitAndReportPlain provides clean, linear append-only output for background services
// like pici-runner.service, bypassing all TTY cursor overwrites and spinner loops.
func waitAndReportPlain(cfg *Cfg, log *slog.Logger, name, jobID, eventType string) error {
	domain := getDomain(eventType)
	prefix := fmt.Sprintf("%s.%s.%s.", domain, name, jobID)
	interval := cfg.MonitorInterval
	if interval <= 0 {
		interval = 1 * time.Second
	}
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	defer signal.Stop(sigCh)

	startedSessions := make(map[string]bool)
	completedSessions := make(map[string]bool)
	var finalSessions []SessionInfo

	var done <-chan struct{}
	if cfg.Ctx != nil {
		done = cfg.Ctx.Done()
	}

	for {
		select {
		case <-done:
			cancelJobSessions(prefix)
			fmt.Fprintf(os.Stdout, "[%s] ⏹ runner cancelled\n", time.Now().Format("15:04:05")) //nolint:errcheck
			return cfg.Ctx.Err()
		case <-sigCh:
			cancelJobSessions(prefix)
			fmt.Fprintf(os.Stdout, "\n[%s] ⏹ runner terminated by signal\n", time.Now().Format("15:04:05")) //nolint:errcheck
			return fmt.Errorf("job cancelled by signal")
		case <-ticker.C:
		}

		listOutput, err := exec.Command("zmx", "list").CombinedOutput()
		if err != nil {
			if log != nil {
				log.Error("zmx list", "err", err)
			}
			continue
		}

		sessions := parseZMXList(string(listOutput))
		var jobSessions []SessionInfo
		for _, s := range sessions {
			if strings.HasPrefix(s.Name, prefix) {
				s.Short = cleanSessionShort(s.Name, prefix, name, jobID)
				jobSessions = append(jobSessions, s)
			}
		}

		finalSessions = jobSessions

		for _, s := range jobSessions {
			if s.Short == "runner" || s.Short == "" {
				continue
			}

			if !startedSessions[s.Short] {
				startedSessions[s.Short] = true
				fmt.Fprintf(os.Stdout, "[%s] ▶ [%s] started\n", time.Now().Format("15:04:05"), s.Short) //nolint:errcheck
			}

			if s.Ended != "" && !completedSessions[s.Short] {
				completedSessions[s.Short] = true
				created, _ := strconv.ParseInt(s.Created, 10, 64)
				ended, _ := strconv.ParseInt(s.Ended, 10, 64)
				dur := time.Duration(ended-created) * time.Second
				if s.ExitCode == "0" {
					fmt.Fprintf(os.Stdout, "[%s] ✔ [%s] succeeded (%s)\n", time.Now().Format("15:04:05"), s.Short, formatShortDuration(dur)) //nolint:errcheck
				} else {
					fmt.Fprintf(os.Stdout, "[%s] ✖ [%s] failed with exit %s (%s)\n", time.Now().Format("15:04:05"), s.Short, s.ExitCode, formatShortDuration(dur)) //nolint:errcheck
					hist, err := fetchHistoryPlain(s.Name)
					if err == nil && hist != "" {
						errPreview := extractErrorPreview(hist, 5)
						for _, line := range errPreview {
							fmt.Fprintf(os.Stdout, "    │ %s\n", line) //nolint:errcheck
						}
					}
				}
			}
		}

		// Stage local artifacts
		_ = stageLocalArtifacts(cfg, log, name, jobID, eventType)

		if isJobComplete(jobSessions) {
			break
		}
	}

	exitCode, status := resolveJobExitCode(finalSessions)
	_, _, duration := computeJobTiming(finalSessions)

	reportPath := filepath.Join(cfg.ArtifactDir, name, jobID, "index.html")
	if exitCode == 0 {
		fmt.Fprintf(os.Stdout, "[%s] ✔ job finished: %s (%s)\n", time.Now().Format("15:04:05"), status, duration) //nolint:errcheck
	} else {
		fmt.Fprintf(os.Stdout, "[%s] ❌ job failed: exit %d (%s)\n", time.Now().Format("15:04:05"), exitCode, duration) //nolint:errcheck
	}
	fmt.Fprintf(os.Stdout, "Artifacts: %s\n", reportPath) //nolint:errcheck

	if cfg.SummaryFile != "" {
		knownStates := make(map[string]*sessionState)
		for _, s := range finalSessions {
			st := &sessionState{
				status:   "failed",
				exitCode: s.ExitCode,
			}
			if s.ExitCode == "0" {
				st.status = "success"
			}
			knownStates[s.Short] = st
		}
		topic := fmt.Sprintf("%s.%s.%s", domain, name, jobID)
		if err := writeSummaryMarkdown(cfg.SummaryFile, name, jobID, finalSessions, knownStates, cfg.DebugOnFail && exitCode != 0, topic); err != nil && log != nil {
			log.Error("write summary markdown", "err", err, "file", cfg.SummaryFile)
		}
	}

	if exitCode != 0 {
		return &JobFailedError{ExitCode: exitCode}
	}

	return nil
}

A reporter_plain_test.go

+49, -0

......@@ -0,0 +1,49 @@
package main

import (
	"context"
	"fmt"
	"io"
	"log/slog"
	"os/exec"
	"testing"
	"time"
)

func TestWaitAndReportPlain_Cancellation(t *testing.T) {
	if _, err := exec.LookPath("zmx"); err != nil {
		t.Skip("zmx not found, skipping integration test")
	}

	ctx, cancel := context.WithCancel(context.Background())
	cfg := &Cfg{
		Ctx:             ctx,
		Cancel:          cancel,
		MonitorInterval: 50 * time.Millisecond,
		Logger:          slog.New(slog.NewTextHandler(io.Discard, nil)),
		ArtifactDir:     t.TempDir(),
	}

	jobID := fmt.Sprintf("plain-cancel-%d", time.Now().UnixNano())
	prefix := "local.testrepo." + jobID + "."
	runnerSession := prefix + "runner"

	t.Cleanup(func() {
		_ = exec.Command("zmx", "kill", "-f", runnerSession).Run()
	})

	cmd := exec.Command("zmx", "run", runnerSession, "-d", "sleep", "30")
	if err := cmd.Run(); err != nil {
		t.Fatalf("failed to start zmx session: %v", err)
	}

	go func() {
		time.Sleep(100 * time.Millisecond)
		cancel()
	}()

	err := waitAndReportPlain(cfg, nil, "testrepo", jobID, "local")
	if err == nil {
		t.Error("expected error when waitAndReportPlain is cancelled, got nil")
	}
}

A reporter_ui.go

+863, -0

......@@ -0,0 +1,863 @@
package main

import (
	"context"
	"fmt"
	"io"
	"log/slog"
	"os"
	"os/exec"
	"os/signal"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"golang.org/x/sys/unix"
)

// ---------------------------------------------------------
// UI State Data Model
// ---------------------------------------------------------

type TaskUIState struct {
	Name          string
	Status        string // "expected", "running", "success", "failed", "cancelled"
	Created       int64
	Ended         int64
	Duration      time.Duration
	ExitCode      string
	LastOutput    string   // Last non-empty line for running peek
	ErrorPreview  []string // Captured error lines on failure
	PredictedDur  time.Duration
	PrevStatus    string
	PrevExitCode  int
	FailureStreak int
	Transition    string
}

type PipelineUIState struct {
	Repo           string
	Branch         string
	Commit         string
	JobID          string
	StartedAt      time.Time
	Elapsed        time.Duration
	Tasks          map[string]*TaskUIState
	TaskOrder      []string
	PredictedTotal time.Duration
	IsComplete     bool
	Cancelled      bool
	Stats          *RepoStats
	mu             sync.Mutex
}

var spinnerFrames = []rune{'⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'}

func newPipelineUIState(repo, branch, commit, jobID string, stats *RepoStats) *PipelineUIState {
	ui := &PipelineUIState{
		Repo:      repo,
		Branch:    branch,
		Commit:    commit,
		JobID:     jobID,
		StartedAt: time.Now(),
		Tasks:     make(map[string]*TaskUIState),
		Stats:     stats,
	}

	if stats != nil {
		ui.PredictedTotal = time.Duration(stats.AvgWallDurationMs) * time.Millisecond

		// Pre-populate expected tasks in deterministic order
		var names []string
		for name := range stats.Tasks {
			names = append(names, name)
		}
		sort.Strings(names)

		for _, name := range names {
			ts := stats.Tasks[name]
			ui.Tasks[name] = &TaskUIState{
				Name:          name,
				Status:        "expected",
				PredictedDur:  time.Duration(ts.AvgDurationMs) * time.Millisecond,
				PrevStatus:    ts.LastStatus,
				PrevExitCode:  ts.LastExitCode,
				FailureStreak: ts.FailureStreak,
			}
			ui.TaskOrder = append(ui.TaskOrder, name)
		}
	}

	return ui
}

// ---------------------------------------------------------
// UI Helpers & Formatters
// ---------------------------------------------------------

var ansiRegex = regexp.MustCompile(`\x1b\[[0-9;]*[a-zA-Z]|\x1b\([a-zA-Z]`)

func stripANSI(s string) string {
	return ansiRegex.ReplaceAllString(s, "")
}

func getTerminalWidth() int {
	ws, err := unix.IoctlGetWinsize(int(os.Stdout.Fd()), unix.TIOCGWINSZ)
	if err == nil && ws.Col > 0 {
		return int(ws.Col)
	}
	if cols := os.Getenv("COLUMNS"); cols != "" {
		if n, err := strconv.Atoi(cols); err == nil && n > 0 {
			return n
		}
	}
	return 80
}

func truncateLine(s string, maxWidth int) string {
	if maxWidth <= 0 {
		maxWidth = 80
	}
	clean := stripANSI(s)
	runes := []rune(clean)
	if len(runes) <= maxWidth {
		return s
	}
	if maxWidth > 3 {
		return string(runes[:maxWidth-3]) + "..."
	}
	return string(runes[:maxWidth])
}

func extractLastOutputLine(raw string) string {
	clean := stripANSI(raw)
	lines := strings.Split(clean, "\n")
	for i := len(lines) - 1; i >= 0; i-- {
		line := strings.TrimSpace(lines[i])
		if line != "" && !strings.HasPrefix(line, "step starting") && !strings.HasPrefix(line, "step completed") {
			return line
		}
	}
	return ""
}

func extractErrorPreview(raw string, maxLines int) []string {
	clean := stripANSI(raw)
	lines := strings.Split(clean, "\n")
	var nonEmpties []string
	for _, l := range lines {
		trimmed := strings.TrimRight(l, "\r\n ")
		if trimmed != "" {
			nonEmpties = append(nonEmpties, trimmed)
		}
	}
	if len(nonEmpties) == 0 {
		return nil
	}
	if len(nonEmpties) > maxLines {
		nonEmpties = nonEmpties[len(nonEmpties)-maxLines:]
	}
	return nonEmpties
}

func renderProgressBar(elapsed, predicted time.Duration, width int) (string, int) {
	if width <= 0 {
		width = 10
	}
	if predicted <= 0 {
		return "[" + strings.Repeat("█", width) + "]", 100
	}

	pct := int((float64(elapsed) / float64(predicted)) * 100)
	displayPct := pct
	if displayPct > 100 {
		displayPct = 100
	}
	if displayPct < 0 {
		displayPct = 0
	}

	filled := (displayPct * width) / 100
	unfilled := width - filled
	if unfilled < 0 {
		unfilled = 0
	}
	bar := "[" + strings.Repeat("█", filled) + strings.Repeat("░", unfilled) + "]"
	return bar, displayPct
}

func renderErrorPreviewBox(lines []string, width int) string {
	if len(lines) == 0 {
		return ""
	}
	if width < 30 {
		width = 60
	}

	var sb strings.Builder
	headerTitle := "┌─ [error preview] "
	headerLine := headerTitle + strings.Repeat("─", max(0, width-len(headerTitle)))
	sb.WriteString("    " + headerLine + "\n")

	for _, line := range lines {
		if len(line) > width-4 {
			line = line[:width-7] + "..."
		}
		fmt.Fprintf(&sb, "    │ %s\n", line)
	}

	footer := "└" + strings.Repeat("─", max(0, width-1))
	sb.WriteString("    " + footer)
	return sb.String()
}

func formatShortDuration(d time.Duration) string {
	secs := d.Seconds()
	if secs < 0 {
		secs = 0
	}
	if secs >= 60 {
		m := int(secs) / 60
		s := int(secs) % 60
		return fmt.Sprintf("%dm%ds", m, s)
	}
	if secs >= 10 {
		return fmt.Sprintf("%.1fs", secs)
	}
	return fmt.Sprintf("%.1fs", secs)
}

// ---------------------------------------------------------
// Full Interactive UI State Rendering
// ---------------------------------------------------------

func (ui *PipelineUIState) Render(frame int) string {
	ui.mu.Lock()
	defer ui.mu.Unlock()

	var sb strings.Builder
	commitShort := ui.Commit
	if len(commitShort) > 7 {
		commitShort = commitShort[:7]
	}
	if commitShort == "" {
		commitShort = "HEAD"
	}
	branch := ui.Branch
	if branch == "" {
		branch = "local"
	}

	// Header line 1: repo info
	fmt.Fprintf(&sb, "🚀 pici: %s (branch: %s • %s)\n", ui.Repo, branch, commitShort)

	// Calculate counts
	totalTasks := len(ui.TaskOrder)
	completedCount := 0
	failedCount := 0
	for _, name := range ui.TaskOrder {
		task := ui.Tasks[name]
		if task.Status == "success" || task.Status == "failed" {
			completedCount++
		}
		if task.Status == "failed" {
			failedCount++
		}
	}

	elapsed := ui.Elapsed
	if elapsed == 0 {
		elapsed = time.Since(ui.StartedAt)
	}

	// Header line 2: timing & status
	fmt.Fprintf(&sb, "⏱  Running for %s [ ", formatShortDuration(elapsed))
	if failedCount > 0 {
		fmt.Fprintf(&sb, "%d/%d completed • %d failed ]\n", completedCount, totalTasks, failedCount)
	} else if completedCount == 0 {
		estTotal := "~" + formatShortDuration(ui.PredictedTotal)
		if ui.PredictedTotal == 0 {
			estTotal = "estimating"
		}
		fmt.Fprintf(&sb, "0/%d tasks • est. %s ]\n", totalTasks, estTotal)
	} else {
		rem := ui.PredictedTotal - elapsed
		if rem < 0 {
			rem = 0
		}
		fmt.Fprintf(&sb, "%d/%d completed • est. ~%s remaining ]\n", completedCount, totalTasks, formatShortDuration(rem))
	}

	sb.WriteString("\n")

	spinnerRune := spinnerFrames[frame%len(spinnerFrames)]

	// Render task list
	for _, name := range ui.TaskOrder {
		task := ui.Tasks[name]
		switch task.Status {
		case "expected":
			pred := ""
			if task.PredictedDur > 0 {
				pred = fmt.Sprintf("~%s", formatShortDuration(task.PredictedDur))
			}
			notes := ""
			if task.PrevStatus == "failed" {
				notes = " • ⚠️ failed last run"
			}
			detail := ""
			if pred != "" || notes != "" {
				detail = fmt.Sprintf("(expected %s%s)", pred, notes)
			}
			fmt.Fprintf(&sb, "  ◌ %-8s %s\n", task.Name, detail)

		case "running":
			taskDur := task.Duration
			if taskDur == 0 && task.Created > 0 {
				taskDur = time.Duration(time.Now().Unix()-task.Created) * time.Second
			}
			predStr := ""
			var bar string
			var pct int
			if task.PredictedDur > 0 {
				predStr = fmt.Sprintf("/ ~%s", formatShortDuration(task.PredictedDur))
				bar, pct = renderProgressBar(taskDur, task.PredictedDur, 10)
			}
			extraNote := ""
			if task.PrevStatus == "failed" {
				extraNote = " (retesting previous failure)"
			}

			barPart := ""
			if bar != "" {
				barPart = fmt.Sprintf("  %s %d%%%s", bar, pct, extraNote)
			}

			fmt.Fprintf(&sb, "  %c %-8s %s %s%s\n", spinnerRune, task.Name, formatShortDuration(taskDur), predStr, barPart)
			if task.LastOutput != "" {
				fmt.Fprintf(&sb, "    └─ %s\n", task.LastOutput)
			}

		case "success":
			transitionNote := ""
			if task.Transition == "fixed" {
				transitionNote = " • 🎉 fixed!"
			}
			durStr := formatShortDuration(task.Duration)
			if transitionNote != "" {
				fmt.Fprintf(&sb, "  ✔ %-8s %s  (%s)\n", task.Name, durStr, strings.TrimPrefix(transitionNote, " • "))
			} else {
				fmt.Fprintf(&sb, "  ✔ %-8s %s\n", task.Name, durStr)
			}

		case "failed":
			notes := ""
			switch task.Transition {
			case "still_failing":
				notes = " • ⚠️ still failing"
			case "new_failure":
				notes = " • ❌ new failure"
			}
			exitStr := "exit 1"
			if task.ExitCode != "" {
				exitStr = "exit " + task.ExitCode
			}
			fmt.Fprintf(&sb, "  ✖ %-8s %s (%s%s)\n", task.Name, formatShortDuration(task.Duration), exitStr, notes)
			if len(task.ErrorPreview) > 0 {
				sb.WriteString(renderErrorPreviewBox(task.ErrorPreview, 65))
				sb.WriteString("\n")
			}

		case "cancelled":
			fmt.Fprintf(&sb, "  ✖ %-8s cancelled\n", task.Name)
		}
	}

	return sb.String()
}

// RenderSummary renders the post-run completion summary.
func (ui *PipelineUIState) RenderSummary(reportPath string, exitCode int) string {
	ui.mu.Lock()
	defer ui.mu.Unlock()

	var sb strings.Builder
	totalTasks := len(ui.TaskOrder)
	failedCount := 0
	var failedTasks []*TaskUIState

	for _, name := range ui.TaskOrder {
		task := ui.Tasks[name]
		if task.Status == "failed" {
			failedCount++
			failedTasks = append(failedTasks, task)
		}
	}

	if exitCode != 0 && failedCount == 0 {
		failedCount = 1
	}

	elapsed := ui.Elapsed
	if elapsed == 0 {
		elapsed = time.Since(ui.StartedAt)
	}
	wallStr := formatShortDuration(elapsed)

	if exitCode == 0 && failedCount == 0 {
		fmt.Fprintf(&sb, "✔ All %d tasks succeeded (total time: %s)\n\n", totalTasks, wallStr)
		for _, name := range ui.TaskOrder {
			task := ui.Tasks[name]
			avgStr := ""
			if task.PredictedDur > 0 {
				avgStr = fmt.Sprintf("(avg %s", formatShortDuration(task.PredictedDur))
				if task.Transition == "fixed" {
					avgStr += " • 🎉 fixed!)"
				} else {
					avgStr += ")"
				}
			}
			fmt.Fprintf(&sb, "  ✔ %-9s %6s   %s\n", task.Name, formatShortDuration(task.Duration), avgStr)
		}
	} else {
		taskWord := "task"
		if failedCount > 1 {
			taskWord = "tasks"
		}
		fmt.Fprintf(&sb, "❌ %d %s failed (total time: %s)\n\n", failedCount, taskWord, wallStr)
		for _, name := range ui.TaskOrder {
			task := ui.Tasks[name]
			switch task.Status {
			case "success":
				fmt.Fprintf(&sb, "  ✔ %-9s %6s\n", task.Name, formatShortDuration(task.Duration))
			case "failed":
				exitStr := "exit " + task.ExitCode
				if task.ExitCode == "" {
					exitStr = "exit 1"
				}
				fmt.Fprintf(&sb, "  ✖ %-9s %6s   %s (see error preview above)\n", task.Name, formatShortDuration(task.Duration), exitStr)
			default:
				fmt.Fprintf(&sb, "  ✖ %-9s cancelled\n", task.Name)
			}
		}
	}

	sb.WriteString("\nArtifacts:\n")
	if reportPath != "" {
		fmt.Fprintf(&sb, "  📄 Report: %s\n", reportPath)
	}
	if failedCount > 0 {
		if len(failedTasks) > 0 {
			for _, ft := range failedTasks {
				fmt.Fprintf(&sb, "  🔍 Debug:  zmx attach local.%s.%s.step.%s\n", ui.Repo, ui.JobID, ft.Name)
			}
		} else {
			fmt.Fprintf(&sb, "  🔍 Debug:  zmx attach local.%s.%s.runner\n", ui.Repo, ui.JobID)
		}
	}

	return sb.String()
}

// ---------------------------------------------------------
// Non-TTY Plain Stream Reporter
// ---------------------------------------------------------

type plainStreamReporter struct {
	out io.Writer
	ui  *PipelineUIState
	log *slog.Logger
}

func (p *plainStreamReporter) OnTaskStart(name string) {
	fmt.Fprintf(p.out, "▶ starting %s\n", name) //nolint:errcheck
}

func (p *plainStreamReporter) OnTaskOutput(name, line string) {
	if line != "" {
		fmt.Fprintf(p.out, "  [%s] %s\n", name, line) //nolint:errcheck
	}
}

func (p *plainStreamReporter) OnTaskComplete(name string, exitCode int, dur time.Duration, errorLines []string) {
	durStr := formatShortDuration(dur)
	if exitCode == 0 {
		fmt.Fprintf(p.out, "✔ %s succeeded (%s)\n", name, durStr) //nolint:errcheck
	} else {
		fmt.Fprintf(p.out, "✖ %s failed with exit %d (%s)\n", name, exitCode, durStr) //nolint:errcheck
		if len(errorLines) > 0 {
			for _, line := range errorLines {
				fmt.Fprintf(p.out, "    │ %s\n", line) //nolint:errcheck
			}
		}
	}
}

// ---------------------------------------------------------
// Interactive Controller Loop
// ---------------------------------------------------------

type interactiveController struct {
	ctx      context.Context
	out      io.Writer
	ui       *PipelineUIState
	isTTY    bool
	rendered int // number of lines rendered in previous frame for cursor overwrite
	streamer *plainStreamReporter
}

func newInteractiveController(ctx context.Context, out io.Writer, repo, branch, commit, jobID string, stats *RepoStats, isTTY bool) *interactiveController {
	if ctx == nil {
		ctx = context.Background()
	}
	ui := newPipelineUIState(repo, branch, commit, jobID, stats)
	return &interactiveController{
		ctx:      ctx,
		out:      out,
		ui:       ui,
		isTTY:    isTTY,
		streamer: &plainStreamReporter{out: out, ui: ui},
	}
}

func (c *interactiveController) WaitUntilDone(fetchSessions func() ([]SessionInfo, bool)) error {
	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	defer signal.Stop(sigCh)

	ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()

	frame := 0

	// Initial render
	c.redraw(c.ui.Render(frame))

	prevStatus := make(map[string]string)
	prevOutput := make(map[string]string)

	for {
		select {
		case <-c.ctx.Done():
			c.ui.Cancelled = true
			c.clear()
			fmt.Fprintln(c.out, "\n⏹ cancelling job and terminating active sessions...") //nolint:errcheck
			return c.ctx.Err()

		case <-sigCh:
			c.ui.Cancelled = true
			c.clear()
			fmt.Fprintln(c.out, "\n^C")                                                //nolint:errcheck
			fmt.Fprintln(c.out, "⏹ cancelling job and terminating active sessions...") //nolint:errcheck
			return fmt.Errorf("job cancelled by signal")

		case <-ticker.C:
			frame++
			sessions, isComplete := fetchSessions()

			// Update UI state with current sessions
			c.ui.mu.Lock()
			for _, s := range sessions {
				name := s.Short
				if name == "runner" || name == "" {
					continue
				}

				task, exists := c.ui.Tasks[name]
				if !exists {
					task = &TaskUIState{
						Name:   name,
						Status: "running",
					}
					c.ui.Tasks[name] = task
					c.ui.TaskOrder = append(c.ui.TaskOrder, name)
				}

				created, _ := strconv.ParseInt(s.Created, 10, 64)
				ended, _ := strconv.ParseInt(s.Ended, 10, 64)
				task.Created = created
				task.Ended = ended

				oldState := prevStatus[name]
				if s.Ended == "" {
					task.Status = "running"
					if created > 0 {
						task.Duration = time.Duration(time.Now().Unix()-created) * time.Second
					}
					if !c.isTTY && (oldState == "" || oldState == "expected") {
						c.streamer.OnTaskStart(name)
					}
					if !c.isTTY && task.LastOutput != "" && task.LastOutput != prevOutput[name] {
						c.streamer.OnTaskOutput(name, task.LastOutput)
						prevOutput[name] = task.LastOutput
					}
				} else {
					task.ExitCode = s.ExitCode
					exitCodeInt, _ := strconv.Atoi(s.ExitCode)
					if s.ExitCode == "0" {
						task.Status = "success"
					} else {
						task.Status = "failed"
					}
					if created > 0 && ended > 0 {
						task.Duration = time.Duration(ended-created) * time.Second
					}
					if task.Transition == "" && c.ui.Stats != nil {
						task.Transition = computeOutcomeTransition(c.ui.Stats.Tasks[name], exitCodeInt)
					}
					if !c.isTTY && oldState != task.Status {
						c.streamer.OnTaskComplete(name, exitCodeInt, task.Duration, task.ErrorPreview)
					}
				}
				prevStatus[name] = task.Status
			}
			c.ui.mu.Unlock()

			c.redraw(c.ui.Render(frame))

			if isComplete {
				c.ui.Elapsed = time.Since(c.ui.StartedAt)
				c.ui.IsComplete = true
				c.clear()
				return nil
			}
		}
	}
}

func (c *interactiveController) redraw(outStr string) {
	if !c.isTTY {
		return
	}
	trimmed := strings.TrimRight(outStr, "\n")
	rawLines := strings.Split(trimmed, "\n")
	if len(rawLines) == 0 || (len(rawLines) == 1 && rawLines[0] == "") {
		return
	}

	termWidth := getTerminalWidth()
	lines := make([]string, len(rawLines))
	for i, l := range rawLines {
		lines[i] = truncateLine(l, termWidth-1)
	}

	var b strings.Builder

	// Move cursor up from the bottom of previous render block to the top line
	if c.rendered > 0 {
		fmt.Fprintf(&b, "\r\033[%dA", c.rendered)
	}

	// Print each line, clearing to the end of the line and resetting carriage
	for i, line := range lines {
		b.WriteString("\r\033[2K") // clear entire line
		b.WriteString(line)
		if i < len(lines)-1 {
			b.WriteString("\n")
		}
	}

	// If the previous frame had more lines, clear the leftover trailing lines
	if c.rendered > len(lines)-1 {
		extra := c.rendered - (len(lines) - 1)
		for i := 0; i < extra; i++ {
			b.WriteString("\n\r\033[2K")
		}
		fmt.Fprintf(&b, "\r\033[%dA", extra)
	}

	c.rendered = len(lines) - 1
	fmt.Fprint(c.out, b.String()) //nolint:errcheck
}

func (c *interactiveController) clear() {
	if !c.isTTY || c.rendered < 0 {
		return
	}
	var b strings.Builder
	if c.rendered > 0 {
		fmt.Fprintf(&b, "\r\033[%dA", c.rendered)
	}
	for i := 0; i <= c.rendered; i++ {
		b.WriteString("\r\033[2K")
		if i < c.rendered {
			b.WriteString("\n")
		}
	}
	if c.rendered > 0 {
		fmt.Fprintf(&b, "\r\033[%dA", c.rendered)
	}
	c.rendered = 0
	fmt.Fprint(c.out, b.String()) //nolint:errcheck
}

// ---------------------------------------------------------
// Interactive waitAndReport Entrypoint
// ---------------------------------------------------------

func isTerminal(w io.Writer) bool {
	if os.Getenv("CI") != "" || os.Getenv("TERM") == "dumb" {
		return false
	}
	if f, ok := w.(*os.File); ok {
		stat, err := f.Stat()
		if err == nil && (stat.Mode()&os.ModeCharDevice) != 0 {
			return true
		}
	}
	return false
}

// waitAndReportUI executes the rich interactive reporting loop, updating stats.json.
func waitAndReportUI(cfg *Cfg, log *slog.Logger, repoName, jobID, eventType, branch, commit string) error {
	domain := getDomain(eventType)
	prefix := fmt.Sprintf("%s.%s.%s.", domain, repoName, jobID)

	stats, _ := loadRepoStats(repoName)
	isTTY := isTerminal(os.Stdout)

	ctrl := newInteractiveController(cfg.Ctx, os.Stdout, repoName, branch, commit, jobID, stats, isTTY)

	lastStage := time.Now()

	// Session poller & peek fetcher
	fetchSessions := func() ([]SessionInfo, bool) {
		listOutput, err := exec.Command("zmx", "list").CombinedOutput()
		if err != nil {
			if log != nil {
				log.Error("zmx list", "err", err)
			}
			return nil, false
		}

		sessions := parseZMXList(string(listOutput))
		var jobSessions []SessionInfo
		for _, s := range sessions {
			if strings.HasPrefix(s.Name, prefix) {
				s.Short = cleanSessionShort(s.Name, prefix, repoName, jobID)
				jobSessions = append(jobSessions, s)
			}
		}

		// Update tail history for live activity and error peeks
		for _, s := range jobSessions {
			if s.Short == "runner" || s.Short == "" {
				continue
			}
			hist, err := fetchHistoryPlain(s.Name)
			if err == nil && hist != "" {
				ctrl.ui.mu.Lock()
				if task, ok := ctrl.ui.Tasks[s.Short]; ok {
					if s.Ended == "" {
						task.LastOutput = extractLastOutputLine(hist)
					} else if s.ExitCode != "0" && len(task.ErrorPreview) == 0 {
						task.ErrorPreview = extractErrorPreview(hist, 5)
					}
				}
				ctrl.ui.mu.Unlock()
			}
		}

		// Stage local artifacts periodically (every 2 seconds) so reports stay updated live without disk trashing
		if time.Since(lastStage) >= 2*time.Second {
			_ = stageLocalArtifacts(cfg, log, repoName, jobID, eventType)
			lastStage = time.Now()
		}

		return jobSessions, isJobComplete(jobSessions)
	}

	waitErr := ctrl.WaitUntilDone(fetchSessions)
	if waitErr != nil {
		cancelJobSessions(prefix)
		if ctrl.ui.Cancelled {
			for _, task := range ctrl.ui.Tasks {
				if task.Status == "running" {
					task.Status = "cancelled"
				}
			}
			fmt.Fprintln(os.Stdout, "⏹ job cancelled") //nolint:errcheck
		}
		return waitErr
	}

	// Fetch final session list
	var finalSessions []SessionInfo
	if listOutput, err := exec.Command("zmx", "list").CombinedOutput(); err == nil {
		for _, s := range parseZMXList(string(listOutput)) {
			if strings.HasPrefix(s.Name, prefix) {
				s.Short = cleanSessionShort(s.Name, prefix, repoName, jobID)
				finalSessions = append(finalSessions, s)
			}
		}
	}

	exitCode, overallStatus := resolveJobExitCode(finalSessions)

	// Update historical stats.json
	if stats != nil {
		for _, s := range finalSessions {
			if s.Short == "runner" || s.Short == "" {
				continue
			}
			created, _ := strconv.ParseInt(s.Created, 10, 64)
			ended, _ := strconv.ParseInt(s.Ended, 10, 64)
			dur := time.Duration(ended-created) * time.Second
			code, _ := strconv.Atoi(s.ExitCode)
			updateTaskStats(stats, s.Short, dur, code)
		}
		updateRepoStats(stats, ctrl.ui.Elapsed, overallStatus)
		_ = saveRepoStats(stats)
	}

	// Output summary
	reportPath := filepath.Join(cfg.ArtifactDir, repoName, jobID, "index.html")
	fmt.Fprintln(os.Stdout)                                            //nolint:errcheck
	fmt.Fprint(os.Stdout, ctrl.ui.RenderSummary(reportPath, exitCode)) //nolint:errcheck

	// Debug bridge if failed
	topic := fmt.Sprintf("%s.%s.%s", domain, repoName, jobID)
	if exitCode != 0 && cfg.DebugOnFail {
		cwd, _ := os.Getwd()
		debugCfg := DebugBridgeConfig{
			Topic:        topic,
			KeyLocation:  cfg.KeyLocation,
			CertLocation: cfg.CertificateLocation,
			WorkDir:      cwd,
			Timeout:      cfg.DebugTimeout,
			Logger:       log,
			Stdout:       os.Stdout,
		}
		if err := startDebugBridge(cfg.Ctx, debugCfg); err != nil && log != nil {
			log.Error("debug bridge failed", "err", err)
		}
	}

	if cfg.SummaryFile != "" {
		knownStates := make(map[string]*sessionState)
		for _, s := range finalSessions {
			st := &sessionState{
				status:   "failed",
				exitCode: s.ExitCode,
			}
			if s.ExitCode == "0" {
				st.status = "success"
			}
			knownStates[s.Short] = st
		}
		if err := writeSummaryMarkdown(cfg.SummaryFile, repoName, jobID, finalSessions, knownStates, cfg.DebugOnFail && exitCode != 0, topic); err != nil && log != nil {
			log.Error("write summary markdown", "err", err, "file", cfg.SummaryFile)
		}
	}

	if exitCode != 0 {
		return &JobFailedError{ExitCode: exitCode}
	}

	return nil
}

A reporter_ui_test.go

+334, -0

......@@ -0,0 +1,334 @@
package main

import (
	"bytes"
	"context"
	"strings"
	"testing"
	"time"
)

func TestUI_ForecastStateInitialization(t *testing.T) {
	stats := &RepoStats{
		Repo:              "my-repo",
		AvgWallDurationMs: 14200,
		Tasks: map[string]*TaskStats{
			"fmt": {
				AvgDurationMs: 850,
				LastStatus:    "success",
				LastExitCode:  0,
			},
			"lint": {
				AvgDurationMs: 4200,
				LastStatus:    "failed",
				LastExitCode:  1,
				FailureStreak: 1,
			},
			"test": {
				AvgDurationMs: 13500,
				LastStatus:    "success",
				LastExitCode:  0,
			},
		},
	}

	ui := newPipelineUIState("my-repo", "main", "8f2a1b9", "job123", stats)

	if len(ui.Tasks) != 3 {
		t.Fatalf("expected 3 tasks, got %d", len(ui.Tasks))
	}
	if ui.Tasks["fmt"].Status != "expected" {
		t.Errorf("expected fmt status to be expected, got %s", ui.Tasks["fmt"].Status)
	}
	if ui.Tasks["lint"].PrevStatus != "failed" {
		t.Errorf("expected lint prev status failed, got %s", ui.Tasks["lint"].PrevStatus)
	}
	if ui.Tasks["lint"].PredictedDur != 4200*time.Millisecond {
		t.Errorf("expected predicted duration 4.2s, got %v", ui.Tasks["lint"].PredictedDur)
	}
}

func TestUI_RenderProgressBar(t *testing.T) {
	tests := []struct {
		elapsed   time.Duration
		predicted time.Duration
		width     int
		wantBar   string
		wantPct   int
	}{
		{0, 10 * time.Second, 10, "[░░░░░░░░░░]", 0},
		{5 * time.Second, 10 * time.Second, 10, "[█████░░░░░]", 50},
		{10 * time.Second, 10 * time.Second, 10, "[██████████]", 100},
		{15 * time.Second, 10 * time.Second, 10, "[██████████]", 100}, // clamped to 100%
		{1 * time.Second, 0, 10, "[██████████]", 100},                 // zero predicted fallback
	}

	for _, tt := range tests {
		bar, pct := renderProgressBar(tt.elapsed, tt.predicted, tt.width)
		if bar != tt.wantBar || pct != tt.wantPct {
			t.Errorf("renderProgressBar(%v, %v, %d) = (%s, %d), want (%s, %d)",
				tt.elapsed, tt.predicted, tt.width, bar, pct, tt.wantBar, tt.wantPct)
		}
	}
}

func TestUI_EarlyFailurePeekBox(t *testing.T) {
	lines := []string{
		"cmd/server/main.go:58:2: errcheck: unchecked error",
		"api/v1/routes.go:19:1: revive: exported function missing doc",
	}

	box := renderErrorPreviewBox(lines, 60)
	if !strings.Contains(box, "┌─ [error preview]") {
		t.Errorf("expected box header, got:\n%s", box)
	}
	if !strings.Contains(box, "│ cmd/server/main.go:58:2: errcheck: unchecked error") {
		t.Errorf("expected line 1 in box, got:\n%s", box)
	}
	if !strings.Contains(box, "└") {
		t.Errorf("expected box footer, got:\n%s", box)
	}
}

func TestUI_LiveActivityPeekExtraction(t *testing.T) {
	rawHistory := "step starting\n\n   \x1b[32mrunning golangci-lint on 28 packages...\x1b[0m\n\n"
	peek := extractLastOutputLine(rawHistory)
	if peek != "running golangci-lint on 28 packages..." {
		t.Errorf("expected clean peek line, got '%s'", peek)
	}

	emptyHistory := "   \n\n\x1b[0m\n"
	peekEmpty := extractLastOutputLine(emptyHistory)
	if peekEmpty != "" {
		t.Errorf("expected empty peek, got '%s'", peekEmpty)
	}
}

func TestUI_RenderStateTransitions(t *testing.T) {
	stats := &RepoStats{
		Repo:              "my-repo",
		AvgWallDurationMs: 14000,
		Tasks: map[string]*TaskStats{
			"fmt":  {AvgDurationMs: 850, LastStatus: "success", LastExitCode: 0},
			"lint": {AvgDurationMs: 4200, LastStatus: "failed", LastExitCode: 1, FailureStreak: 1},
			"test": {AvgDurationMs: 13500, LastStatus: "success", LastExitCode: 0},
		},
	}

	ui := newPipelineUIState("my-repo", "main", "8f2a1b9", "job123", stats)

	// State 1: Forecast
	render1 := ui.Render(0)
	if !strings.Contains(render1, "◌ lint") || !strings.Contains(render1, "failed last run") {
		t.Errorf("expected State 1 forecast output, got:\n%s", render1)
	}

	// State 2: Running fmt and lint
	ui.Tasks["fmt"].Status = "running"
	ui.Tasks["fmt"].Duration = 400 * time.Millisecond
	ui.Tasks["lint"].Status = "running"
	ui.Tasks["lint"].Duration = 5400 * time.Millisecond
	ui.Tasks["lint"].LastOutput = "running golangci-lint on 28 packages..."

	render2 := ui.Render(0)
	if !strings.Contains(render2, "fmt") || !strings.Contains(render2, "lint") {
		t.Errorf("expected State 2 running tasks, got:\n%s", render2)
	}
	if !strings.Contains(render2, "└─ running golangci-lint on 28 packages...") {
		t.Errorf("expected live activity peek under lint, got:\n%s", render2)
	}

	// State 3: Failure with error preview
	ui.Tasks["fmt"].Status = "success"
	ui.Tasks["fmt"].Duration = 800 * time.Millisecond
	ui.Tasks["lint"].Status = "failed"
	ui.Tasks["lint"].ExitCode = "1"
	ui.Tasks["lint"].Duration = 4200 * time.Millisecond
	ui.Tasks["lint"].Transition = "still_failing"
	ui.Tasks["lint"].ErrorPreview = []string{
		"cmd/server/main.go:58:2: errcheck: unchecked error",
		"api/v1/routes.go:19:1: revive: exported function missing doc",
	}

	render3 := ui.Render(0)
	if !strings.Contains(render3, "lint") || !strings.Contains(render3, "still failing") {
		t.Errorf("expected State 3 failure indicator, got:\n%s", render3)
	}
	if !strings.Contains(render3, "cmd/server/main.go:58:2") {
		t.Errorf("expected inline error preview box, got:\n%s", render3)
	}

	// State 5: Completion summary (success vs failure)
	summary := ui.RenderSummary("/tmp/pici-artifacts/my-repo/job123/index.html", 1)
	if !strings.Contains(summary, "1 task failed") {
		t.Errorf("expected failure summary, got:\n%s", summary)
	}
	if !strings.Contains(summary, "Report: /tmp/pici-artifacts/my-repo/job123/index.html") {
		t.Errorf("expected report path in summary, got:\n%s", summary)
	}
}

func TestUI_RenderSummary_SuccessWithFixed(t *testing.T) {
	stats := &RepoStats{
		Repo: "my-repo",
		Tasks: map[string]*TaskStats{
			"fmt":  {AvgDurationMs: 900, LastStatus: "success", LastExitCode: 0},
			"lint": {AvgDurationMs: 4200, LastStatus: "failed", LastExitCode: 1, FailureStreak: 1},
			"test": {AvgDurationMs: 13500, LastStatus: "success", LastExitCode: 0},
		},
	}

	ui := newPipelineUIState("my-repo", "main", "8f2a1b9", "job123", stats)
	ui.Tasks["fmt"].Status = "success"
	ui.Tasks["fmt"].Duration = 800 * time.Millisecond
	ui.Tasks["lint"].Status = "success"
	ui.Tasks["lint"].Duration = 4100 * time.Millisecond
	ui.Tasks["lint"].Transition = "fixed"
	ui.Tasks["test"].Status = "success"
	ui.Tasks["test"].Duration = 14100 * time.Millisecond

	summary := ui.RenderSummary("/tmp/report.html", 0)
	if !strings.Contains(summary, "All 3 tasks succeeded") {
		t.Errorf("expected success summary, got:\n%s", summary)
	}
	if !strings.Contains(summary, "🎉 fixed!") {
		t.Errorf("expected fixed badge in summary, got:\n%s", summary)
	}
	if !strings.Contains(summary, "Report: /tmp/report.html") {
		t.Errorf("expected report path, got:\n%s", summary)
	}
}

func TestUI_NonTTYStreaming(t *testing.T) {
	var buf bytes.Buffer
	logger := newLogger("ci", "error")

	stats := &RepoStats{Repo: "test"}
	ui := newPipelineUIState("test", "main", "abc", "job1", stats)

	streamer := &plainStreamReporter{
		out: &buf,
		ui:  ui,
		log: logger,
	}

	streamer.OnTaskStart("fmt")
	streamer.OnTaskOutput("fmt", "formatting...")
	streamer.OnTaskComplete("fmt", 0, 800*time.Millisecond, nil)

	out := buf.String()
	if !strings.Contains(out, "starting fmt") {
		t.Errorf("expected non-TTY start event, got: %s", out)
	}
	if !strings.Contains(out, "fmt succeeded (0.8s)") {
		t.Errorf("expected non-TTY complete event, got: %s", out)
	}
}

func TestInteractiveController_Cancellation(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	var out bytes.Buffer

	ctrl := newInteractiveController(ctx, &out, "testrepo", "main", "head", "job1", nil, false)
	go func() {
		time.Sleep(50 * time.Millisecond)
		cancel()
	}()

	err := ctrl.WaitUntilDone(func() ([]SessionInfo, bool) {
		return []SessionInfo{
			{Name: "local.testrepo.job1.step.test", Short: "test"},
		}, false
	})

	if err == nil {
		t.Fatal("expected error on cancellation, got nil")
	}
	if !strings.Contains(out.String(), "cancelling job and terminating active sessions") && !ctrl.ui.Cancelled {
		t.Errorf("expected cancellation message, got: %s", out.String())
	}
}

func TestInteractiveController_FullRunSimulation(t *testing.T) {
	var out bytes.Buffer
	ctx := context.Background()

	stats := &RepoStats{
		Repo: "my-repo",
		Tasks: map[string]*TaskStats{
			"test": {AvgDurationMs: 1000, LastStatus: "success"},
		},
	}

	ctrl := newInteractiveController(ctx, &out, "my-repo", "main", "abcd123", "job99", stats, true)

	step := 0
	err := ctrl.WaitUntilDone(func() ([]SessionInfo, bool) {
		step++
		if step == 1 {
			// Step running
			return []SessionInfo{
				{Name: "local.my-repo.job99.step.test", Short: "test", Created: "1000", Ended: ""},
			}, false
		}
		// Step completed
		return []SessionInfo{
			{Name: "local.my-repo.job99.step.test", Short: "test", Created: "1000", Ended: "1002", ExitCode: "0"},
			{Name: "local.my-repo.job99.runner", Short: "runner", Created: "1000", Ended: "1002", ExitCode: "0"},
		}, true
	})

	if err != nil {
		t.Fatalf("expected nil error, got %v", err)
	}
	if !ctrl.ui.IsComplete {
		t.Errorf("expected UI to be marked complete")
	}
	if ctrl.ui.Tasks["test"].Status != "success" {
		t.Errorf("expected test status success, got %s", ctrl.ui.Tasks["test"].Status)
	}
}

func TestController_RedrawNoDrift(t *testing.T) {
	var buf bytes.Buffer
	ctrl := &interactiveController{
		out:   &buf,
		isTTY: true,
	}

	// Frame 1: 3 lines
	ctrl.redraw("line1\nline2\nline3\n")
	if ctrl.rendered != 2 {
		t.Errorf("expected rendered=2, got %d", ctrl.rendered)
	}

	// Frame 2: same 3 lines (should move up 2 lines, not drift)
	buf.Reset()
	ctrl.redraw("line1\nline2\nline3\n")
	out := buf.String()
	if !strings.HasPrefix(out, "\r\033[2A") {
		t.Errorf("expected move up 2 lines '\\r\\033[2A', got %q", out)
	}

	// Frame 3: expanded to 5 lines
	buf.Reset()
	ctrl.redraw("line1\nline2\nline3\nline4\nline5\n")
	out = buf.String()
	if !strings.HasPrefix(out, "\r\033[2A") {
		t.Errorf("expected move up 2 lines before expanding, got %q", out)
	}
	if ctrl.rendered != 4 {
		t.Errorf("expected rendered=4, got %d", ctrl.rendered)
	}

	// Frame 4: shrink back to 3 lines
	buf.Reset()
	ctrl.redraw("line1\nline2\nline3\n")
	out = buf.String()
	if !strings.HasPrefix(out, "\r\033[4A") {
		t.Errorf("expected move up 4 lines before shrinking, got %q", out)
	}
	if ctrl.rendered != 2 {
		t.Errorf("expected rendered=2, got %d", ctrl.rendered)
	}
}

A stats.go

+146, -0

......@@ -0,0 +1,146 @@
package main

import (
	"encoding/json"
	"math"
	"os"
	"path/filepath"
	"time"
)

// RepoStats tracks historical run durations and outcomes across runs.
type RepoStats struct {
	Repo              string                `json:"repo"`
	LastRunAt         string                `json:"last_run_at"`
	LastOverallStatus string                `json:"last_overall_status"`
	RunsCount         int                   `json:"runs_count"`
	AvgWallDurationMs int64                 `json:"avg_wall_duration_ms"`
	Tasks             map[string]*TaskStats `json:"tasks"`
}

// TaskStats tracks duration averages, exit codes, and failure streaks per task.
type TaskStats struct {
	AvgDurationMs  int64  `json:"avg_duration_ms"`
	LastDurationMs int64  `json:"last_duration_ms"`
	LastStatus     string `json:"last_status"`
	LastExitCode   int    `json:"last_exit_code"`
	FailureStreak  int    `json:"failure_streak,omitempty"`
	SeenCount      int    `json:"seen_count"`
}

// getStatsPath returns the path to stats.json for the specified repository.
func getStatsPath(repo string) string {
	cacheDir := os.Getenv("XDG_CACHE_HOME")
	if cacheDir == "" {
		home, err := os.UserHomeDir()
		if err != nil {
			cacheDir = "/tmp"
		} else {
			cacheDir = filepath.Join(home, ".cache")
		}
	}
	return filepath.Join(cacheDir, "pici", repo, "stats.json")
}

// loadRepoStats loads historical stats for a repo, or returns an initialized empty struct.
func loadRepoStats(repo string) (*RepoStats, error) {
	path := getStatsPath(repo)
	data, err := os.ReadFile(path)
	if err != nil {
		if os.IsNotExist(err) {
			return &RepoStats{
				Repo:  repo,
				Tasks: make(map[string]*TaskStats),
			}, nil
		}
		return nil, err
	}

	var stats RepoStats
	if err := json.Unmarshal(data, &stats); err != nil {
		return &RepoStats{
			Repo:  repo,
			Tasks: make(map[string]*TaskStats),
		}, nil
	}
	if stats.Tasks == nil {
		stats.Tasks = make(map[string]*TaskStats)
	}
	return &stats, nil
}

// saveRepoStats writes repo stats to stats.json.
func saveRepoStats(stats *RepoStats) error {
	if stats == nil || stats.Repo == "" {
		return nil
	}
	path := getStatsPath(stats.Repo)
	if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
		return err
	}
	data, err := json.MarshalIndent(stats, "", "  ")
	if err != nil {
		return err
	}
	return os.WriteFile(path, data, 0644)
}

// updateTaskStats updates single task statistics using EMA duration (alpha = 0.3).
func updateTaskStats(stats *RepoStats, name string, actualDur time.Duration, exitCode int) {
	if stats.Tasks == nil {
		stats.Tasks = make(map[string]*TaskStats)
	}
	task, ok := stats.Tasks[name]
	if !ok {
		task = &TaskStats{}
		stats.Tasks[name] = task
	}

	actualMs := actualDur.Milliseconds()
	if task.SeenCount == 0 || task.AvgDurationMs == 0 {
		task.AvgDurationMs = actualMs
	} else {
		// EMA: alpha = 0.3
		task.AvgDurationMs = int64(math.Round(0.3*float64(actualMs) + 0.7*float64(task.AvgDurationMs)))
	}

	task.LastDurationMs = actualMs
	task.LastExitCode = exitCode
	task.SeenCount++

	if exitCode == 0 {
		task.LastStatus = "success"
		task.FailureStreak = 0
	} else {
		task.LastStatus = "failed"
		task.FailureStreak++
	}
}

// updateRepoStats updates overall repository stats after a run.
func updateRepoStats(stats *RepoStats, wallDur time.Duration, overallStatus string) {
	stats.RunsCount++
	stats.LastRunAt = time.Now().UTC().Format(time.RFC3339)
	stats.LastOverallStatus = overallStatus

	wallMs := wallDur.Milliseconds()
	if stats.RunsCount == 1 || stats.AvgWallDurationMs == 0 {
		stats.AvgWallDurationMs = wallMs
	} else {
		stats.AvgWallDurationMs = int64(math.Round(0.3*float64(wallMs) + 0.7*float64(stats.AvgWallDurationMs)))
	}
}

// computeOutcomeTransition returns "fixed", "still_failing", "new_failure", or "".
func computeOutcomeTransition(prev *TaskStats, currentExitCode int) string {
	if prev != nil && (prev.LastStatus == "failed" || prev.LastExitCode != 0) {
		if currentExitCode == 0 {
			return "fixed"
		}
		return "still_failing"
	}
	if currentExitCode != 0 {
		return "new_failure"
	}
	return ""
}

A stats_test.go

+157, -0

......@@ -0,0 +1,157 @@
package main

import (
	"path/filepath"
	"testing"
	"time"
)

func TestStats_EMACalculation(t *testing.T) {
	// Formula: EMA_new = 0.3 * actual + 0.7 * EMA_prev
	// If EMA_prev == 0, EMA_new = actual
	stats := &RepoStats{
		Repo: "testrepo",
		Tasks: map[string]*TaskStats{
			"fmt": {
				AvgDurationMs:  1000,
				LastDurationMs: 1000,
				LastStatus:     "success",
				LastExitCode:   0,
				SeenCount:      1,
			},
		},
	}

	// Update with a new run of 2000ms
	updateTaskStats(stats, "fmt", 2000*time.Millisecond, 0)
	task := stats.Tasks["fmt"]
	if task.SeenCount != 2 {
		t.Errorf("expected seen count 2, got %d", task.SeenCount)
	}
	// 0.3 * 2000 + 0.7 * 1000 = 600 + 700 = 1300
	if task.AvgDurationMs != 1300 {
		t.Errorf("expected AvgDurationMs 1300, got %d", task.AvgDurationMs)
	}
	if task.LastDurationMs != 2000 {
		t.Errorf("expected LastDurationMs 2000, got %d", task.LastDurationMs)
	}
	if task.LastStatus != "success" {
		t.Errorf("expected LastStatus success, got %s", task.LastStatus)
	}
	if task.FailureStreak != 0 {
		t.Errorf("expected FailureStreak 0, got %d", task.FailureStreak)
	}

	// Update with new task (not seen before)
	updateTaskStats(stats, "lint", 4000*time.Millisecond, 1)
	lintTask := stats.Tasks["lint"]
	if lintTask.AvgDurationMs != 4000 {
		t.Errorf("expected initial AvgDurationMs 4000, got %d", lintTask.AvgDurationMs)
	}
	if lintTask.LastStatus != "failed" {
		t.Errorf("expected LastStatus failed, got %s", lintTask.LastStatus)
	}
	if lintTask.FailureStreak != 1 {
		t.Errorf("expected FailureStreak 1, got %d", lintTask.FailureStreak)
	}

	// Fail again -> streak increments
	updateTaskStats(stats, "lint", 4100*time.Millisecond, 1)
	lintTask = stats.Tasks["lint"]
	if lintTask.FailureStreak != 2 {
		t.Errorf("expected FailureStreak 2, got %d", lintTask.FailureStreak)
	}

	// Succeed -> streak resets to 0
	updateTaskStats(stats, "lint", 3900*time.Millisecond, 0)
	lintTask = stats.Tasks["lint"]
	if lintTask.FailureStreak != 0 {
		t.Errorf("expected FailureStreak reset to 0, got %d", lintTask.FailureStreak)
	}
	if lintTask.LastStatus != "success" {
		t.Errorf("expected LastStatus success, got %s", lintTask.LastStatus)
	}
}

func TestStats_StorageAndPath(t *testing.T) {
	tempCache := t.TempDir()
	t.Setenv("XDG_CACHE_HOME", tempCache)

	repo := "my-project"
	expectedPath := filepath.Join(tempCache, "pici", repo, "stats.json")
	path := getStatsPath(repo)
	if path != expectedPath {
		t.Errorf("expected stats path %s, got %s", expectedPath, path)
	}

	// Test Save & Load
	stats := &RepoStats{
		Repo:              repo,
		LastRunAt:         time.Now().UTC().Format(time.RFC3339),
		LastOverallStatus: "success",
		RunsCount:         5,
		AvgWallDurationMs: 12000,
		Tasks: map[string]*TaskStats{
			"build": {
				AvgDurationMs:  6000,
				LastDurationMs: 6100,
				LastStatus:     "success",
				LastExitCode:   0,
				SeenCount:      5,
			},
		},
	}

	if err := saveRepoStats(stats); err != nil {
		t.Fatalf("failed to save stats: %v", err)
	}

	loaded, err := loadRepoStats(repo)
	if err != nil {
		t.Fatalf("failed to load stats: %v", err)
	}
	if loaded.Repo != repo || loaded.RunsCount != 5 || loaded.AvgWallDurationMs != 12000 {
		t.Errorf("loaded stats mismatch: %+v", loaded)
	}
	if task, ok := loaded.Tasks["build"]; !ok || task.AvgDurationMs != 6000 {
		t.Errorf("loaded task stats mismatch: %+v", task)
	}
}

func TestStats_OutcomeTransitions(t *testing.T) {
	// 1. Fixed transition (failed in prev run, succeeded in current run)
	t1 := &TaskStats{
		LastStatus:   "failed",
		LastExitCode: 1,
	}
	if trans := computeOutcomeTransition(t1, 0); trans != "fixed" {
		t.Errorf("expected 'fixed', got '%s'", trans)
	}

	// 2. Still failing transition (failed in prev run, failed in current run)
	if trans := computeOutcomeTransition(t1, 1); trans != "still_failing" {
		t.Errorf("expected 'still_failing', got '%s'", trans)
	}

	// 3. New failure transition (succeeded in prev run, failed in current run)
	t2 := &TaskStats{
		LastStatus:   "success",
		LastExitCode: 0,
	}
	if trans := computeOutcomeTransition(t2, 2); trans != "new_failure" {
		t.Errorf("expected 'new_failure', got '%s'", trans)
	}

	// 4. Clean success (succeeded in prev run, succeeded in current run)
	if trans := computeOutcomeTransition(t2, 0); trans != "" {
		t.Errorf("expected '', got '%s'", trans)
	}

	// 5. First time seen (nil prev)
	if trans := computeOutcomeTransition(nil, 0); trans != "" {
		t.Errorf("expected '', got '%s'", trans)
	}
	if trans := computeOutcomeTransition(nil, 1); trans != "new_failure" {
		t.Errorf("expected 'new_failure', got '%s'", trans)
	}
}

...	...		@@ -1351,7 +1151,6 @@ type sessionState struct {
1351	1151		status string // "running", "success", "failed"
1352	1152		exitCode string
1353	1153		duration string
1354		-	printed bool // true once final state (success/failed) is printed
1355	1154		}
1356	1155
1357	1156		// monitorJobState tracks display state for a single job across ticks.

...	...		@@ -2630,14 +2429,14 @@ func runLocal(cfg *Cfg, dest string) error {
2630	2429		}
2631	2430		}()
2632	2431
2633		-	fmt.Fprintf(os.Stdout, "🚀 starting local job local.%s.%s\n", eventData.Name, jobID) //nolint:errcheck
	2432	+	log.Info("starting local job", "job_id", jobID)
2634	2433		if err := eng.Setup(); err != nil {
2635	2434		return fmt.Errorf("workspace setup: %w", err)
2636	2435		}
2637	2436		if cfg.InPlace {
2638		-	fmt.Fprintf(os.Stdout, "📁 using in-place workspace directory %s\n", eng.Wk.GetDir()) //nolint:errcheck
	2437	+	log.Debug("using in-place workspace directory", "dir", eng.Wk.GetDir())
2639	2438		} else {
2640		-	fmt.Fprintf(os.Stdout, "📦 syncing workspace to temp directory %s\n", eng.Wk.GetDir()) //nolint:errcheck
	2439	+	log.Debug("syncing workspace to temp directory", "dir", eng.Wk.GetDir())
2641	2440		}
2642	2441		log.Debug("workspace directory", "dir", eng.Wk.GetDir())
2643	2442

...	...		@@ -2674,7 +2473,7 @@ func runLocal(cfg *Cfg, dest string) error {
2674	2473		return err
2675	2474		}
2676	2475
2677		-	fmt.Fprintln(os.Stdout, "🏃 launching sessions...") //nolint:errcheck
	2476	+	log.Debug("launching sessions")
2678	2477		if err := eng.Run(manifest); err != nil {
2679	2478		return fmt.Errorf("run: %w", err)
2680	2479		}

...	...		@@ -16,13 +16,11 @@ import (
16	16		"log/slog"
17	17		"os"
18	18		"os/exec"
19		-	"os/signal"
20	19		"path/filepath"
21	20		"runtime"
22	21		"sort"
23	22		"strconv"
24	23		"strings"
25		-	"syscall"
26	24		"time"
27	25		)
28	26

...	...		@@ -1027,7 +1025,7 @@ See: https://github.com/picosh/pici
1027	1025		fmt.Fprintln(os.Stdout, "✅ job launched") //nolint:errcheck
1028	1026
1029	1027		if cfg.Wait {
1030		-	if err := waitAndReport(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
	1028	+	if err := waitAndReportPlain(cfg, log, eventData.Name, jobID, eventData.Type); err != nil {
1031	1029		runErr = err
1032	1030		return fmt.Errorf("wait: %w", err)
1033	1031		}

...	...		@@ -1043,210 +1041,12 @@ See: https://github.com/picosh/pici
1043	1041		}
1044	1042
1045	1043		// waitAndReport polls the job's sessions until all complete, prints live
1046		-	// progress to stdout, then dumps session history and a final summary.
	1044	+	// interactive progress to stdout, updates stats.json, and outputs a final summary.
1047	1045		func waitAndReport(cfg Cfg, log slog.Logger, name, jobID, eventType string) error {
1048		-	domain := getDomain(eventType)
1049		-	prefix := domain + "." + name + "." + jobID + "."
1050		-	interval := cfg.MonitorInterval
1051		-	if interval <= 0 {
1052		-	interval = 5 * time.Second
1053		-	}
1054		-	ticker := time.NewTicker(interval)
1055		-	defer ticker.Stop()
1056		-
1057		-	// Handle ^C and termination signals gracefully
1058		-	sigCh := make(chan os.Signal, 1)
1059		-	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
1060		-	defer signal.Stop(sigCh)
1061		-
1062		-	fmt.Fprintln(os.Stdout) //nolint:errcheck
1063		-	fmt.Fprint(os.Stdout, "⏳ waiting for completion...\n") //nolint:errcheck
1064		-
1065		-	// Track state for each session
1066		-	known := make(map[string]*sessionState)
1067		-	var jobSessions []SessionInfo
1068		-	var sessionOrder []string // track insertion order for deterministic output
1069		-	var liveLines []string // last set of status lines printed (for overwrite)
1070		-
1071		-	var done <-chan struct{}
1072		-	if cfg.Ctx != nil {
1073		-	done = cfg.Ctx.Done()
1074		-	}
1075		-
1076		-	for {
1077		-	select {
1078		-	case <-done:
1079		-	cancelJobSessions(prefix)
1080		-	fmt.Fprintln(os.Stdout, "\n⏹ job cancelled") //nolint:errcheck
1081		-	return cfg.Ctx.Err()
1082		-	case <-sigCh:
1083		-	cancelJobSessions(prefix)
1084		-	fmt.Fprintln(os.Stdout, "\n⏹ job cancelled") //nolint:errcheck
1085		-	return fmt.Errorf("job cancelled by signal")
1086		-	case <-ticker.C:
1087		-	}
1088		-
1089		-	// Fetch current session list
1090		-	listOutput, err := exec.Command("zmx", "list").CombinedOutput()
1091		-	if err != nil {
1092		-	log.Error("zmx list", "err", err)
1093		-	continue
1094		-	}
1095		-
1096		-	sessions := parseZMXList(string(listOutput))
1097		-	jobSessions = nil
1098		-	for _, s := range sessions {
1099		-	if strings.HasPrefix(s.Name, prefix) {
1100		-	s.Short = cleanSessionShort(s.Name, prefix, name, jobID)
1101		-	jobSessions = append(jobSessions, s)
1102		-	}
1103		-	}
1104		-
1105		-	if len(jobSessions) == 0 {
1106		-	continue // runner session may not have appeared yet
1107		-	}
1108		-
1109		-	// Update state for each session
1110		-	for _, s := range jobSessions {
1111		-	state, ok := known[s.Short]
1112		-	if !ok {
1113		-	state = &sessionState{}
1114		-	known[s.Short] = state
1115		-	sessionOrder = append(sessionOrder, s.Short)
1116		-	}
1117		-
1118		-	if s.Ended == "" {
1119		-	state.status = "running"
1120		-	created, _ := strconv.ParseInt(s.Created, 10, 64)
1121		-	state.duration = fmtDurationTs(created, time.Now().Unix())
1122		-	} else {
1123		-	if s.ExitCode == "0" {
1124		-	state.status = "success"
1125		-	} else {
1126		-	state.status = "failed"
1127		-	state.exitCode = s.ExitCode
1128		-	}
1129		-	state.duration = fmtDuration(s.Created, s.Ended)
1130		-	state.printed = true // lock final state
1131		-	}
1132		-	}
1133		-
1134		-	// Build current status lines
1135		-	lines := make([]string, 0, len(sessionOrder))
1136		-	for _, short := range sessionOrder {
1137		-	state := known[short]
1138		-	icon := map[string]string{"running": "🚀", "success": "✅", "failed": "❌"}[state.status]
1139		-	detail := ""
1140		-	if state.status == "failed" && state.exitCode != "" {
1141		-	detail = fmt.Sprintf(", exit %s", state.exitCode)
1142		-	}
1143		-	if state.duration != "" && state.duration != "—" {
1144		-	detail += fmt.Sprintf(" (%s)", state.duration)
1145		-	}
1146		-	lines = append(lines, fmt.Sprintf(" %-12s %s %s%s", short, icon, state.status, detail))
1147		-	}
1148		-
1149		-	// Overwrite previous lines with cursor-up, or print fresh
1150		-	if len(liveLines) > 0 {
1151		-	// Move cursor up to overwrite previous lines //nolint:errcheck
1152		-	for range len(liveLines) {
1153		-	fmt.Fprint(os.Stdout, "\033[A") //nolint:errcheck
1154		-	}
1155		-	// Clear each line
1156		-	for i, line := range lines { //nolint:errcheck
1157		-	if i > 0 {
1158		-	fmt.Fprint(os.Stdout, "\n") //nolint:errcheck
1159		-	}
1160		-	fmt.Fprint(os.Stdout, line+"\033[K") //nolint:errcheck
1161		-	}
1162		-	fmt.Fprint(os.Stdout, "\n") //nolint:errcheck
1163		-	} else { //nolint:errcheck
1164		-	for _, line := range lines {
1165		-	fmt.Fprintln(os.Stdout, line) //nolint:errcheck
1166		-	}
1167		-	}
1168		-	liveLines = lines
1169		-
1170		-	// Stage local artifacts on tick so progress and HTML reports update live
1171		-	_ = stageLocalArtifacts(cfg, log, name, jobID, eventType)
1172		-
1173		-	// Check if all sessions are done and the runner session has finished
1174		-	if isJobComplete(jobSessions) {
1175		-	break
1176		-	}
1177		-	}
1178		-
1179		-	// Print last 25 lines of history for failed sessions only
1180		-	fmt.Fprintln(os.Stdout) //nolint:errcheck
1181		-	for _, s := range jobSessions {
1182		-	state := known[s.Short]
1183		-	if state == nil \|\| state.status != "failed" {
1184		-	continue
1185		-	}
1186		-
1187		-	separator := strings.Repeat("\u2500", 50)
1188		-	fmt.Fprintln(os.Stdout, separator) //nolint:errcheck
1189		-	fmt.Fprintf(os.Stdout, "Session: %s (exit %s)\n", s.Short, state.exitCode) //nolint:errcheck
1190		-	fmt.Fprintln(os.Stdout, separator) //nolint:errcheck
1191		-	fmt.Fprintln(os.Stdout) //nolint:errcheck
1192		-
1193		-	history, err := fetchHistoryPlain(s.Name)
1194		-	if err != nil {
1195		-	fmt.Fprintf(os.Stdout, " (history unavailable: %v)\n", err) //nolint:errcheck
1196		-	} else {
1197		-	lines := strings.Split(history, "\n")
1198		-	// Show last 25 lines
1199		-	if len(lines) > 25 {
1200		-	fmt.Fprintf(os.Stdout, " ... (%d lines omitted)\n", len(lines)-25) //nolint:errcheck
1201		-	lines = lines[len(lines)-25:]
1202		-	}
1203		-	for _, line := range lines {
1204		-	if line != "" {
1205		-	fmt.Fprintf(os.Stdout, " %s\n", line) //nolint:errcheck
1206		-	}
1207		-	}
1208		-	}
1209		-	fmt.Fprintln(os.Stdout) //nolint:errcheck
1210		-	}
1211		-
1212		-	// Final summary
1213		-	exitCode, status := resolveJobExitCode(jobSessions)
1214		-	_, _, duration := computeJobTiming(jobSessions)
1215		-	icon := map[string]string{"success": "✅", "failed": "❌"}[status]
1216		-	if exitCode == 0 {
1217		-	fmt.Fprintf(os.Stdout, "%s job finished: %s (%s)\n", icon, status, duration) //nolint:errcheck
1218		-	} else {
1219		-	fmt.Fprintf(os.Stdout, "%s job failed: exit %d (%s)\n", icon, exitCode, duration) //nolint:errcheck
1220		-	}
1221		-
1222		-	topic := fmt.Sprintf("%s.%s.%s", domain, name, jobID)
1223		-	if exitCode != 0 && cfg.DebugOnFail {
1224		-	cwd, _ := os.Getwd()
1225		-	debugCfg := DebugBridgeConfig{
1226		-	Topic: topic,
1227		-	KeyLocation: cfg.KeyLocation,
1228		-	CertLocation: cfg.CertificateLocation,
1229		-	WorkDir: cwd,
1230		-	Timeout: cfg.DebugTimeout,
1231		-	Logger: log,
1232		-	Stdout: os.Stdout,
1233		-	}
1234		-	if err := startDebugBridge(cfg.Ctx, debugCfg); err != nil {
1235		-	log.Error("debug bridge failed", "err", err)
1236		-	}
1237		-	}
1238		-
1239		-	if cfg.SummaryFile != "" {
1240		-	if err := writeSummaryMarkdown(cfg.SummaryFile, name, jobID, jobSessions, known, cfg.DebugOnFail && exitCode != 0, topic); err != nil {
1241		-	log.Error("write summary markdown", "err", err, "file", cfg.SummaryFile)
1242		-	}
1243		-	}
1244		-
1245		-	if exitCode != 0 {
1246		-	return &JobFailedError{ExitCode: exitCode}
1247		-	}
1248		-
1249		-	return nil
	1046	+	cwd, _ := os.Getwd()
	1047	+	branch := detectGitBranch(cwd)
	1048	+	commit := detectGitCommit(cwd)
	1049	+	return waitAndReportUI(cfg, log, name, jobID, eventType, branch, commit)
1250	1050		}
1251	1051
1252	1052		func writeSummaryMarkdown(filePath string, repoName, jobID string, jobSessions []SessionInfo, known map[string]*sessionState, debugActive bool, debugTopic string) error {

...	...		@@ -0,0 +1,145 @@
	1	+	package main
	2	+
	3	+	import (
	4	+	"fmt"
	5	+	"log/slog"
	6	+	"os"
	7	+	"os/exec"
	8	+	"os/signal"
	9	+	"path/filepath"
	10	+	"strconv"
	11	+	"strings"
	12	+	"syscall"
	13	+	"time"
	14	+	)
	15	+
	16	+	// waitAndReportPlain provides clean, linear append-only output for background services
	17	+	// like pici-runner.service, bypassing all TTY cursor overwrites and spinner loops.
	18	+	func waitAndReportPlain(cfg Cfg, log slog.Logger, name, jobID, eventType string) error {
	19	+	domain := getDomain(eventType)
	20	+	prefix := fmt.Sprintf("%s.%s.%s.", domain, name, jobID)
	21	+	interval := cfg.MonitorInterval
	22	+	if interval <= 0 {
	23	+	interval = 1 * time.Second
	24	+	}
	25	+	ticker := time.NewTicker(interval)
	26	+	defer ticker.Stop()
	27	+
	28	+	sigCh := make(chan os.Signal, 1)
	29	+	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	30	+	defer signal.Stop(sigCh)
	31	+
	32	+	startedSessions := make(map[string]bool)
	33	+	completedSessions := make(map[string]bool)
	34	+	var finalSessions []SessionInfo
	35	+
	36	+	var done <-chan struct{}
	37	+	if cfg.Ctx != nil {
	38	+	done = cfg.Ctx.Done()
	39	+	}
	40	+
	41	+	for {
	42	+	select {
	43	+	case <-done:
	44	+	cancelJobSessions(prefix)
	45	+	fmt.Fprintf(os.Stdout, "[%s] ⏹ runner cancelled\n", time.Now().Format("15:04:05")) //nolint:errcheck
	46	+	return cfg.Ctx.Err()
	47	+	case <-sigCh:
	48	+	cancelJobSessions(prefix)
	49	+	fmt.Fprintf(os.Stdout, "\n[%s] ⏹ runner terminated by signal\n", time.Now().Format("15:04:05")) //nolint:errcheck
	50	+	return fmt.Errorf("job cancelled by signal")
	51	+	case <-ticker.C:
	52	+	}
	53	+
	54	+	listOutput, err := exec.Command("zmx", "list").CombinedOutput()
	55	+	if err != nil {
	56	+	if log != nil {
	57	+	log.Error("zmx list", "err", err)
	58	+	}
	59	+	continue
	60	+	}
	61	+
	62	+	sessions := parseZMXList(string(listOutput))
	63	+	var jobSessions []SessionInfo
	64	+	for _, s := range sessions {
	65	+	if strings.HasPrefix(s.Name, prefix) {
	66	+	s.Short = cleanSessionShort(s.Name, prefix, name, jobID)
	67	+	jobSessions = append(jobSessions, s)
	68	+	}
	69	+	}
	70	+
	71	+	finalSessions = jobSessions
	72	+
	73	+	for _, s := range jobSessions {
	74	+	if s.Short == "runner" \|\| s.Short == "" {
	75	+	continue
	76	+	}
	77	+
	78	+	if !startedSessions[s.Short] {
	79	+	startedSessions[s.Short] = true
	80	+	fmt.Fprintf(os.Stdout, "[%s] ▶ [%s] started\n", time.Now().Format("15:04:05"), s.Short) //nolint:errcheck
	81	+	}
	82	+
	83	+	if s.Ended != "" && !completedSessions[s.Short] {
	84	+	completedSessions[s.Short] = true
	85	+	created, _ := strconv.ParseInt(s.Created, 10, 64)
	86	+	ended, _ := strconv.ParseInt(s.Ended, 10, 64)
	87	+	dur := time.Duration(ended-created) * time.Second
	88	+	if s.ExitCode == "0" {
	89	+	fmt.Fprintf(os.Stdout, "[%s] ✔ [%s] succeeded (%s)\n", time.Now().Format("15:04:05"), s.Short, formatShortDuration(dur)) //nolint:errcheck
	90	+	} else {
	91	+	fmt.Fprintf(os.Stdout, "[%s] ✖ [%s] failed with exit %s (%s)\n", time.Now().Format("15:04:05"), s.Short, s.ExitCode, formatShortDuration(dur)) //nolint:errcheck
	92	+	hist, err := fetchHistoryPlain(s.Name)
	93	+	if err == nil && hist != "" {
	94	+	errPreview := extractErrorPreview(hist, 5)
	95	+	for _, line := range errPreview {
	96	+	fmt.Fprintf(os.Stdout, " │ %s\n", line) //nolint:errcheck
	97	+	}
	98	+	}
	99	+	}
	100	+	}
	101	+	}
	102	+
	103	+	// Stage local artifacts
	104	+	_ = stageLocalArtifacts(cfg, log, name, jobID, eventType)
	105	+
	106	+	if isJobComplete(jobSessions) {
	107	+	break
	108	+	}
	109	+	}
	110	+
	111	+	exitCode, status := resolveJobExitCode(finalSessions)
	112	+	_, _, duration := computeJobTiming(finalSessions)
	113	+
	114	+	reportPath := filepath.Join(cfg.ArtifactDir, name, jobID, "index.html")
	115	+	if exitCode == 0 {
	116	+	fmt.Fprintf(os.Stdout, "[%s] ✔ job finished: %s (%s)\n", time.Now().Format("15:04:05"), status, duration) //nolint:errcheck
	117	+	} else {
	118	+	fmt.Fprintf(os.Stdout, "[%s] ❌ job failed: exit %d (%s)\n", time.Now().Format("15:04:05"), exitCode, duration) //nolint:errcheck
	119	+	}
	120	+	fmt.Fprintf(os.Stdout, "Artifacts: %s\n", reportPath) //nolint:errcheck
	121	+
	122	+	if cfg.SummaryFile != "" {
	123	+	knownStates := make(map[string]*sessionState)
	124	+	for _, s := range finalSessions {
	125	+	st := &sessionState{
	126	+	status: "failed",
	127	+	exitCode: s.ExitCode,
	128	+	}
	129	+	if s.ExitCode == "0" {
	130	+	st.status = "success"
	131	+	}
	132	+	knownStates[s.Short] = st
	133	+	}
	134	+	topic := fmt.Sprintf("%s.%s.%s", domain, name, jobID)
	135	+	if err := writeSummaryMarkdown(cfg.SummaryFile, name, jobID, finalSessions, knownStates, cfg.DebugOnFail && exitCode != 0, topic); err != nil && log != nil {
	136	+	log.Error("write summary markdown", "err", err, "file", cfg.SummaryFile)
	137	+	}
	138	+	}
	139	+
	140	+	if exitCode != 0 {
	141	+	return &JobFailedError{ExitCode: exitCode}
	142	+	}
	143	+
	144	+	return nil
	145	+	}

...	...		@@ -0,0 +1,49 @@
	1	+	package main
	2	+
	3	+	import (
	4	+	"context"
	5	+	"fmt"
	6	+	"io"
	7	+	"log/slog"
	8	+	"os/exec"
	9	+	"testing"
	10	+	"time"
	11	+	)
	12	+
	13	+	func TestWaitAndReportPlain_Cancellation(t *testing.T) {
	14	+	if _, err := exec.LookPath("zmx"); err != nil {
	15	+	t.Skip("zmx not found, skipping integration test")
	16	+	}
	17	+
	18	+	ctx, cancel := context.WithCancel(context.Background())
	19	+	cfg := &Cfg{
	20	+	Ctx: ctx,
	21	+	Cancel: cancel,
	22	+	MonitorInterval: 50 * time.Millisecond,
	23	+	Logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
	24	+	ArtifactDir: t.TempDir(),
	25	+	}
	26	+
	27	+	jobID := fmt.Sprintf("plain-cancel-%d", time.Now().UnixNano())
	28	+	prefix := "local.testrepo." + jobID + "."
	29	+	runnerSession := prefix + "runner"
	30	+
	31	+	t.Cleanup(func() {
	32	+	_ = exec.Command("zmx", "kill", "-f", runnerSession).Run()
	33	+	})
	34	+
	35	+	cmd := exec.Command("zmx", "run", runnerSession, "-d", "sleep", "30")
	36	+	if err := cmd.Run(); err != nil {
	37	+	t.Fatalf("failed to start zmx session: %v", err)
	38	+	}
	39	+
	40	+	go func() {
	41	+	time.Sleep(100 * time.Millisecond)
	42	+	cancel()
	43	+	}()
	44	+
	45	+	err := waitAndReportPlain(cfg, nil, "testrepo", jobID, "local")
	46	+	if err == nil {
	47	+	t.Error("expected error when waitAndReportPlain is cancelled, got nil")
	48	+	}
	49	+	}

...	...		@@ -0,0 +1,334 @@
	1	+	package main
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"context"
	6	+	"strings"
	7	+	"testing"
	8	+	"time"
	9	+	)
	10	+
	11	+	func TestUI_ForecastStateInitialization(t *testing.T) {
	12	+	stats := &RepoStats{
	13	+	Repo: "my-repo",
	14	+	AvgWallDurationMs: 14200,
	15	+	Tasks: map[string]*TaskStats{
	16	+	"fmt": {
	17	+	AvgDurationMs: 850,
	18	+	LastStatus: "success",
	19	+	LastExitCode: 0,
	20	+	},
	21	+	"lint": {
	22	+	AvgDurationMs: 4200,
	23	+	LastStatus: "failed",
	24	+	LastExitCode: 1,
	25	+	FailureStreak: 1,
	26	+	},
	27	+	"test": {
	28	+	AvgDurationMs: 13500,
	29	+	LastStatus: "success",
	30	+	LastExitCode: 0,
	31	+	},
	32	+	},
	33	+	}
	34	+
	35	+	ui := newPipelineUIState("my-repo", "main", "8f2a1b9", "job123", stats)
	36	+
	37	+	if len(ui.Tasks) != 3 {
	38	+	t.Fatalf("expected 3 tasks, got %d", len(ui.Tasks))
	39	+	}
	40	+	if ui.Tasks["fmt"].Status != "expected" {
	41	+	t.Errorf("expected fmt status to be expected, got %s", ui.Tasks["fmt"].Status)
	42	+	}
	43	+	if ui.Tasks["lint"].PrevStatus != "failed" {
	44	+	t.Errorf("expected lint prev status failed, got %s", ui.Tasks["lint"].PrevStatus)
	45	+	}
	46	+	if ui.Tasks["lint"].PredictedDur != 4200*time.Millisecond {
	47	+	t.Errorf("expected predicted duration 4.2s, got %v", ui.Tasks["lint"].PredictedDur)
	48	+	}
	49	+	}
	50	+
	51	+	func TestUI_RenderProgressBar(t *testing.T) {
	52	+	tests := []struct {
	53	+	elapsed time.Duration
	54	+	predicted time.Duration
	55	+	width int
	56	+	wantBar string
	57	+	wantPct int
	58	+	}{
	59	+	{0, 10 * time.Second, 10, "[░░░░░░░░░░]", 0},
	60	+	{5 * time.Second, 10 * time.Second, 10, "[█████░░░░░]", 50},
	61	+	{10 * time.Second, 10 * time.Second, 10, "[██████████]", 100},
	62	+	{15 * time.Second, 10 * time.Second, 10, "[██████████]", 100}, // clamped to 100%
	63	+	{1 * time.Second, 0, 10, "[██████████]", 100}, // zero predicted fallback
	64	+	}
	65	+
	66	+	for _, tt := range tests {
	67	+	bar, pct := renderProgressBar(tt.elapsed, tt.predicted, tt.width)
	68	+	if bar != tt.wantBar \|\| pct != tt.wantPct {
	69	+	t.Errorf("renderProgressBar(%v, %v, %d) = (%s, %d), want (%s, %d)",
	70	+	tt.elapsed, tt.predicted, tt.width, bar, pct, tt.wantBar, tt.wantPct)
	71	+	}
	72	+	}
	73	+	}
	74	+
	75	+	func TestUI_EarlyFailurePeekBox(t *testing.T) {
	76	+	lines := []string{
	77	+	"cmd/server/main.go:58:2: errcheck: unchecked error",
	78	+	"api/v1/routes.go:19:1: revive: exported function missing doc",
	79	+	}
	80	+
	81	+	box := renderErrorPreviewBox(lines, 60)
	82	+	if !strings.Contains(box, "┌─ [error preview]") {
	83	+	t.Errorf("expected box header, got:\n%s", box)
	84	+	}
	85	+	if !strings.Contains(box, "│ cmd/server/main.go:58:2: errcheck: unchecked error") {
	86	+	t.Errorf("expected line 1 in box, got:\n%s", box)
	87	+	}
	88	+	if !strings.Contains(box, "└") {
	89	+	t.Errorf("expected box footer, got:\n%s", box)
	90	+	}
	91	+	}
	92	+
	93	+	func TestUI_LiveActivityPeekExtraction(t *testing.T) {
	94	+	rawHistory := "step starting\n\n \x1b[32mrunning golangci-lint on 28 packages...\x1b[0m\n\n"
	95	+	peek := extractLastOutputLine(rawHistory)
	96	+	if peek != "running golangci-lint on 28 packages..." {
	97	+	t.Errorf("expected clean peek line, got '%s'", peek)
	98	+	}
	99	+
	100	+	emptyHistory := " \n\n\x1b[0m\n"
	101	+	peekEmpty := extractLastOutputLine(emptyHistory)
	102	+	if peekEmpty != "" {
	103	+	t.Errorf("expected empty peek, got '%s'", peekEmpty)
	104	+	}
	105	+	}
	106	+
	107	+	func TestUI_RenderStateTransitions(t *testing.T) {
	108	+	stats := &RepoStats{
	109	+	Repo: "my-repo",
	110	+	AvgWallDurationMs: 14000,
	111	+	Tasks: map[string]*TaskStats{
	112	+	"fmt": {AvgDurationMs: 850, LastStatus: "success", LastExitCode: 0},
	113	+	"lint": {AvgDurationMs: 4200, LastStatus: "failed", LastExitCode: 1, FailureStreak: 1},
	114	+	"test": {AvgDurationMs: 13500, LastStatus: "success", LastExitCode: 0},
	115	+	},
	116	+	}
	117	+
	118	+	ui := newPipelineUIState("my-repo", "main", "8f2a1b9", "job123", stats)
	119	+
	120	+	// State 1: Forecast
	121	+	render1 := ui.Render(0)
	122	+	if !strings.Contains(render1, "◌ lint") \|\| !strings.Contains(render1, "failed last run") {
	123	+	t.Errorf("expected State 1 forecast output, got:\n%s", render1)
	124	+	}
	125	+
	126	+	// State 2: Running fmt and lint
	127	+	ui.Tasks["fmt"].Status = "running"
	128	+	ui.Tasks["fmt"].Duration = 400 * time.Millisecond
	129	+	ui.Tasks["lint"].Status = "running"
	130	+	ui.Tasks["lint"].Duration = 5400 * time.Millisecond
	131	+	ui.Tasks["lint"].LastOutput = "running golangci-lint on 28 packages..."
	132	+
	133	+	render2 := ui.Render(0)
	134	+	if !strings.Contains(render2, "fmt") \|\| !strings.Contains(render2, "lint") {
	135	+	t.Errorf("expected State 2 running tasks, got:\n%s", render2)
	136	+	}
	137	+	if !strings.Contains(render2, "└─ running golangci-lint on 28 packages...") {
	138	+	t.Errorf("expected live activity peek under lint, got:\n%s", render2)
	139	+	}
	140	+
	141	+	// State 3: Failure with error preview
	142	+	ui.Tasks["fmt"].Status = "success"
	143	+	ui.Tasks["fmt"].Duration = 800 * time.Millisecond
	144	+	ui.Tasks["lint"].Status = "failed"
	145	+	ui.Tasks["lint"].ExitCode = "1"
	146	+	ui.Tasks["lint"].Duration = 4200 * time.Millisecond
	147	+	ui.Tasks["lint"].Transition = "still_failing"
	148	+	ui.Tasks["lint"].ErrorPreview = []string{
	149	+	"cmd/server/main.go:58:2: errcheck: unchecked error",
	150	+	"api/v1/routes.go:19:1: revive: exported function missing doc",
	151	+	}
	152	+
	153	+	render3 := ui.Render(0)
	154	+	if !strings.Contains(render3, "lint") \|\| !strings.Contains(render3, "still failing") {
	155	+	t.Errorf("expected State 3 failure indicator, got:\n%s", render3)
	156	+	}
	157	+	if !strings.Contains(render3, "cmd/server/main.go:58:2") {
	158	+	t.Errorf("expected inline error preview box, got:\n%s", render3)
	159	+	}
	160	+
	161	+	// State 5: Completion summary (success vs failure)
	162	+	summary := ui.RenderSummary("/tmp/pici-artifacts/my-repo/job123/index.html", 1)
	163	+	if !strings.Contains(summary, "1 task failed") {
	164	+	t.Errorf("expected failure summary, got:\n%s", summary)
	165	+	}
	166	+	if !strings.Contains(summary, "Report: /tmp/pici-artifacts/my-repo/job123/index.html") {
	167	+	t.Errorf("expected report path in summary, got:\n%s", summary)
	168	+	}
	169	+	}
	170	+
	171	+	func TestUI_RenderSummary_SuccessWithFixed(t *testing.T) {
	172	+	stats := &RepoStats{
	173	+	Repo: "my-repo",
	174	+	Tasks: map[string]*TaskStats{
	175	+	"fmt": {AvgDurationMs: 900, LastStatus: "success", LastExitCode: 0},
	176	+	"lint": {AvgDurationMs: 4200, LastStatus: "failed", LastExitCode: 1, FailureStreak: 1},
	177	+	"test": {AvgDurationMs: 13500, LastStatus: "success", LastExitCode: 0},
	178	+	},
	179	+	}
	180	+
	181	+	ui := newPipelineUIState("my-repo", "main", "8f2a1b9", "job123", stats)
	182	+	ui.Tasks["fmt"].Status = "success"
	183	+	ui.Tasks["fmt"].Duration = 800 * time.Millisecond
	184	+	ui.Tasks["lint"].Status = "success"
	185	+	ui.Tasks["lint"].Duration = 4100 * time.Millisecond
	186	+	ui.Tasks["lint"].Transition = "fixed"
	187	+	ui.Tasks["test"].Status = "success"
	188	+	ui.Tasks["test"].Duration = 14100 * time.Millisecond
	189	+
	190	+	summary := ui.RenderSummary("/tmp/report.html", 0)
	191	+	if !strings.Contains(summary, "All 3 tasks succeeded") {
	192	+	t.Errorf("expected success summary, got:\n%s", summary)
	193	+	}
	194	+	if !strings.Contains(summary, "🎉 fixed!") {
	195	+	t.Errorf("expected fixed badge in summary, got:\n%s", summary)
	196	+	}
	197	+	if !strings.Contains(summary, "Report: /tmp/report.html") {
	198	+	t.Errorf("expected report path, got:\n%s", summary)
	199	+	}
	200	+	}
	201	+
	202	+	func TestUI_NonTTYStreaming(t *testing.T) {
	203	+	var buf bytes.Buffer
	204	+	logger := newLogger("ci", "error")
	205	+
	206	+	stats := &RepoStats{Repo: "test"}
	207	+	ui := newPipelineUIState("test", "main", "abc", "job1", stats)
	208	+
	209	+	streamer := &plainStreamReporter{
	210	+	out: &buf,
	211	+	ui: ui,
	212	+	log: logger,
	213	+	}
	214	+
	215	+	streamer.OnTaskStart("fmt")
	216	+	streamer.OnTaskOutput("fmt", "formatting...")
	217	+	streamer.OnTaskComplete("fmt", 0, 800*time.Millisecond, nil)
	218	+
	219	+	out := buf.String()
	220	+	if !strings.Contains(out, "starting fmt") {
	221	+	t.Errorf("expected non-TTY start event, got: %s", out)
	222	+	}
	223	+	if !strings.Contains(out, "fmt succeeded (0.8s)") {
	224	+	t.Errorf("expected non-TTY complete event, got: %s", out)
	225	+	}
	226	+	}
	227	+
	228	+	func TestInteractiveController_Cancellation(t *testing.T) {
	229	+	ctx, cancel := context.WithCancel(context.Background())
	230	+	var out bytes.Buffer
	231	+
	232	+	ctrl := newInteractiveController(ctx, &out, "testrepo", "main", "head", "job1", nil, false)
	233	+	go func() {
	234	+	time.Sleep(50 * time.Millisecond)
	235	+	cancel()
	236	+	}()
	237	+
	238	+	err := ctrl.WaitUntilDone(func() ([]SessionInfo, bool) {
	239	+	return []SessionInfo{
	240	+	{Name: "local.testrepo.job1.step.test", Short: "test"},
	241	+	}, false
	242	+	})
	243	+
	244	+	if err == nil {
	245	+	t.Fatal("expected error on cancellation, got nil")
	246	+	}
	247	+	if !strings.Contains(out.String(), "cancelling job and terminating active sessions") && !ctrl.ui.Cancelled {
	248	+	t.Errorf("expected cancellation message, got: %s", out.String())
	249	+	}
	250	+	}
	251	+
	252	+	func TestInteractiveController_FullRunSimulation(t *testing.T) {
	253	+	var out bytes.Buffer
	254	+	ctx := context.Background()
	255	+
	256	+	stats := &RepoStats{
	257	+	Repo: "my-repo",
	258	+	Tasks: map[string]*TaskStats{
	259	+	"test": {AvgDurationMs: 1000, LastStatus: "success"},
	260	+	},
	261	+	}
	262	+
	263	+	ctrl := newInteractiveController(ctx, &out, "my-repo", "main", "abcd123", "job99", stats, true)
	264	+
	265	+	step := 0
	266	+	err := ctrl.WaitUntilDone(func() ([]SessionInfo, bool) {
	267	+	step++
	268	+	if step == 1 {
	269	+	// Step running
	270	+	return []SessionInfo{
	271	+	{Name: "local.my-repo.job99.step.test", Short: "test", Created: "1000", Ended: ""},
	272	+	}, false
	273	+	}
	274	+	// Step completed
	275	+	return []SessionInfo{
	276	+	{Name: "local.my-repo.job99.step.test", Short: "test", Created: "1000", Ended: "1002", ExitCode: "0"},
	277	+	{Name: "local.my-repo.job99.runner", Short: "runner", Created: "1000", Ended: "1002", ExitCode: "0"},
	278	+	}, true
	279	+	})
	280	+
	281	+	if err != nil {
	282	+	t.Fatalf("expected nil error, got %v", err)
	283	+	}
	284	+	if !ctrl.ui.IsComplete {
	285	+	t.Errorf("expected UI to be marked complete")
	286	+	}
	287	+	if ctrl.ui.Tasks["test"].Status != "success" {
	288	+	t.Errorf("expected test status success, got %s", ctrl.ui.Tasks["test"].Status)
	289	+	}
	290	+	}
	291	+
	292	+	func TestController_RedrawNoDrift(t *testing.T) {
	293	+	var buf bytes.Buffer
	294	+	ctrl := &interactiveController{
	295	+	out: &buf,
	296	+	isTTY: true,
	297	+	}
	298	+
	299	+	// Frame 1: 3 lines
	300	+	ctrl.redraw("line1\nline2\nline3\n")
	301	+	if ctrl.rendered != 2 {
	302	+	t.Errorf("expected rendered=2, got %d", ctrl.rendered)
	303	+	}
	304	+
	305	+	// Frame 2: same 3 lines (should move up 2 lines, not drift)
	306	+	buf.Reset()
	307	+	ctrl.redraw("line1\nline2\nline3\n")
	308	+	out := buf.String()
	309	+	if !strings.HasPrefix(out, "\r\033[2A") {
	310	+	t.Errorf("expected move up 2 lines '\\r\\033[2A', got %q", out)
	311	+	}
	312	+
	313	+	// Frame 3: expanded to 5 lines
	314	+	buf.Reset()
	315	+	ctrl.redraw("line1\nline2\nline3\nline4\nline5\n")
	316	+	out = buf.String()
	317	+	if !strings.HasPrefix(out, "\r\033[2A") {
	318	+	t.Errorf("expected move up 2 lines before expanding, got %q", out)
	319	+	}
	320	+	if ctrl.rendered != 4 {
	321	+	t.Errorf("expected rendered=4, got %d", ctrl.rendered)
	322	+	}
	323	+
	324	+	// Frame 4: shrink back to 3 lines
	325	+	buf.Reset()
	326	+	ctrl.redraw("line1\nline2\nline3\n")
	327	+	out = buf.String()
	328	+	if !strings.HasPrefix(out, "\r\033[4A") {
	329	+	t.Errorf("expected move up 4 lines before shrinking, got %q", out)
	330	+	}
	331	+	if ctrl.rendered != 2 {
	332	+	t.Errorf("expected rendered=2, got %d", ctrl.rendered)
	333	+	}
	334	+	}

...	...		@@ -0,0 +1,146 @@
	1	+	package main
	2	+
	3	+	import (
	4	+	"encoding/json"
	5	+	"math"
	6	+	"os"
	7	+	"path/filepath"
	8	+	"time"
	9	+	)
	10	+
	11	+	// RepoStats tracks historical run durations and outcomes across runs.
	12	+	type RepoStats struct {
	13	+	Repo string `json:"repo"`
	14	+	LastRunAt string `json:"last_run_at"`
	15	+	LastOverallStatus string `json:"last_overall_status"`
	16	+	RunsCount int `json:"runs_count"`
	17	+	AvgWallDurationMs int64 `json:"avg_wall_duration_ms"`
	18	+	Tasks map[string]*TaskStats `json:"tasks"`
	19	+	}
	20	+
	21	+	// TaskStats tracks duration averages, exit codes, and failure streaks per task.
	22	+	type TaskStats struct {
	23	+	AvgDurationMs int64 `json:"avg_duration_ms"`
	24	+	LastDurationMs int64 `json:"last_duration_ms"`
	25	+	LastStatus string `json:"last_status"`
	26	+	LastExitCode int `json:"last_exit_code"`
	27	+	FailureStreak int `json:"failure_streak,omitempty"`
	28	+	SeenCount int `json:"seen_count"`
	29	+	}
	30	+
	31	+	// getStatsPath returns the path to stats.json for the specified repository.
	32	+	func getStatsPath(repo string) string {
	33	+	cacheDir := os.Getenv("XDG_CACHE_HOME")
	34	+	if cacheDir == "" {
	35	+	home, err := os.UserHomeDir()
	36	+	if err != nil {
	37	+	cacheDir = "/tmp"
	38	+	} else {
	39	+	cacheDir = filepath.Join(home, ".cache")
	40	+	}
	41	+	}
	42	+	return filepath.Join(cacheDir, "pici", repo, "stats.json")
	43	+	}
	44	+
	45	+	// loadRepoStats loads historical stats for a repo, or returns an initialized empty struct.
	46	+	func loadRepoStats(repo string) (*RepoStats, error) {
	47	+	path := getStatsPath(repo)
	48	+	data, err := os.ReadFile(path)
	49	+	if err != nil {
	50	+	if os.IsNotExist(err) {
	51	+	return &RepoStats{
	52	+	Repo: repo,
	53	+	Tasks: make(map[string]*TaskStats),
	54	+	}, nil
	55	+	}
	56	+	return nil, err
	57	+	}
	58	+
	59	+	var stats RepoStats
	60	+	if err := json.Unmarshal(data, &stats); err != nil {
	61	+	return &RepoStats{
	62	+	Repo: repo,
	63	+	Tasks: make(map[string]*TaskStats),
	64	+	}, nil
	65	+	}
	66	+	if stats.Tasks == nil {
	67	+	stats.Tasks = make(map[string]*TaskStats)
	68	+	}
	69	+	return &stats, nil
	70	+	}
	71	+
	72	+	// saveRepoStats writes repo stats to stats.json.
	73	+	func saveRepoStats(stats *RepoStats) error {
	74	+	if stats == nil \|\| stats.Repo == "" {
	75	+	return nil
	76	+	}
	77	+	path := getStatsPath(stats.Repo)
	78	+	if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
	79	+	return err
	80	+	}
	81	+	data, err := json.MarshalIndent(stats, "", " ")
	82	+	if err != nil {
	83	+	return err
	84	+	}
	85	+	return os.WriteFile(path, data, 0644)
	86	+	}
	87	+
	88	+	// updateTaskStats updates single task statistics using EMA duration (alpha = 0.3).
	89	+	func updateTaskStats(stats *RepoStats, name string, actualDur time.Duration, exitCode int) {
	90	+	if stats.Tasks == nil {
	91	+	stats.Tasks = make(map[string]*TaskStats)
	92	+	}
	93	+	task, ok := stats.Tasks[name]
	94	+	if !ok {
	95	+	task = &TaskStats{}
	96	+	stats.Tasks[name] = task
	97	+	}
	98	+
	99	+	actualMs := actualDur.Milliseconds()
	100	+	if task.SeenCount == 0 \|\| task.AvgDurationMs == 0 {
	101	+	task.AvgDurationMs = actualMs
	102	+	} else {
	103	+	// EMA: alpha = 0.3
	104	+	task.AvgDurationMs = int64(math.Round(0.3float64(actualMs) + 0.7float64(task.AvgDurationMs)))
	105	+	}
	106	+
	107	+	task.LastDurationMs = actualMs
	108	+	task.LastExitCode = exitCode
	109	+	task.SeenCount++
	110	+
	111	+	if exitCode == 0 {
	112	+	task.LastStatus = "success"
	113	+	task.FailureStreak = 0
	114	+	} else {
	115	+	task.LastStatus = "failed"
	116	+	task.FailureStreak++
	117	+	}
	118	+	}
	119	+
	120	+	// updateRepoStats updates overall repository stats after a run.
	121	+	func updateRepoStats(stats *RepoStats, wallDur time.Duration, overallStatus string) {
	122	+	stats.RunsCount++
	123	+	stats.LastRunAt = time.Now().UTC().Format(time.RFC3339)
	124	+	stats.LastOverallStatus = overallStatus
	125	+
	126	+	wallMs := wallDur.Milliseconds()
	127	+	if stats.RunsCount == 1 \|\| stats.AvgWallDurationMs == 0 {
	128	+	stats.AvgWallDurationMs = wallMs
	129	+	} else {
	130	+	stats.AvgWallDurationMs = int64(math.Round(0.3float64(wallMs) + 0.7float64(stats.AvgWallDurationMs)))
	131	+	}
	132	+	}
	133	+
	134	+	// computeOutcomeTransition returns "fixed", "still_failing", "new_failure", or "".
	135	+	func computeOutcomeTransition(prev *TaskStats, currentExitCode int) string {
	136	+	if prev != nil && (prev.LastStatus == "failed" \|\| prev.LastExitCode != 0) {
	137	+	if currentExitCode == 0 {
	138	+	return "fixed"
	139	+	}
	140	+	return "still_failing"
	141	+	}
	142	+	if currentExitCode != 0 {
	143	+	return "new_failure"
	144	+	}
	145	+	return ""
	146	+	}

...	...		@@ -0,0 +1,157 @@
	1	+	package main
	2	+
	3	+	import (
	4	+	"path/filepath"
	5	+	"testing"
	6	+	"time"
	7	+	)
	8	+
	9	+	func TestStats_EMACalculation(t *testing.T) {
	10	+	// Formula: EMA_new = 0.3 * actual + 0.7 * EMA_prev
	11	+	// If EMA_prev == 0, EMA_new = actual
	12	+	stats := &RepoStats{
	13	+	Repo: "testrepo",
	14	+	Tasks: map[string]*TaskStats{
	15	+	"fmt": {
	16	+	AvgDurationMs: 1000,
	17	+	LastDurationMs: 1000,
	18	+	LastStatus: "success",
	19	+	LastExitCode: 0,
	20	+	SeenCount: 1,
	21	+	},
	22	+	},
	23	+	}
	24	+
	25	+	// Update with a new run of 2000ms
	26	+	updateTaskStats(stats, "fmt", 2000*time.Millisecond, 0)
	27	+	task := stats.Tasks["fmt"]
	28	+	if task.SeenCount != 2 {
	29	+	t.Errorf("expected seen count 2, got %d", task.SeenCount)
	30	+	}
	31	+	// 0.3 * 2000 + 0.7 * 1000 = 600 + 700 = 1300
	32	+	if task.AvgDurationMs != 1300 {
	33	+	t.Errorf("expected AvgDurationMs 1300, got %d", task.AvgDurationMs)
	34	+	}
	35	+	if task.LastDurationMs != 2000 {
	36	+	t.Errorf("expected LastDurationMs 2000, got %d", task.LastDurationMs)
	37	+	}
	38	+	if task.LastStatus != "success" {
	39	+	t.Errorf("expected LastStatus success, got %s", task.LastStatus)
	40	+	}
	41	+	if task.FailureStreak != 0 {
	42	+	t.Errorf("expected FailureStreak 0, got %d", task.FailureStreak)
	43	+	}
	44	+
	45	+	// Update with new task (not seen before)
	46	+	updateTaskStats(stats, "lint", 4000*time.Millisecond, 1)
	47	+	lintTask := stats.Tasks["lint"]
	48	+	if lintTask.AvgDurationMs != 4000 {
	49	+	t.Errorf("expected initial AvgDurationMs 4000, got %d", lintTask.AvgDurationMs)
	50	+	}
	51	+	if lintTask.LastStatus != "failed" {
	52	+	t.Errorf("expected LastStatus failed, got %s", lintTask.LastStatus)
	53	+	}
	54	+	if lintTask.FailureStreak != 1 {
	55	+	t.Errorf("expected FailureStreak 1, got %d", lintTask.FailureStreak)
	56	+	}
	57	+
	58	+	// Fail again -> streak increments
	59	+	updateTaskStats(stats, "lint", 4100*time.Millisecond, 1)
	60	+	lintTask = stats.Tasks["lint"]
	61	+	if lintTask.FailureStreak != 2 {
	62	+	t.Errorf("expected FailureStreak 2, got %d", lintTask.FailureStreak)
	63	+	}
	64	+
	65	+	// Succeed -> streak resets to 0
	66	+	updateTaskStats(stats, "lint", 3900*time.Millisecond, 0)
	67	+	lintTask = stats.Tasks["lint"]
	68	+	if lintTask.FailureStreak != 0 {
	69	+	t.Errorf("expected FailureStreak reset to 0, got %d", lintTask.FailureStreak)
	70	+	}
	71	+	if lintTask.LastStatus != "success" {
	72	+	t.Errorf("expected LastStatus success, got %s", lintTask.LastStatus)
	73	+	}
	74	+	}
	75	+
	76	+	func TestStats_StorageAndPath(t *testing.T) {
	77	+	tempCache := t.TempDir()
	78	+	t.Setenv("XDG_CACHE_HOME", tempCache)
	79	+
	80	+	repo := "my-project"
	81	+	expectedPath := filepath.Join(tempCache, "pici", repo, "stats.json")
	82	+	path := getStatsPath(repo)
	83	+	if path != expectedPath {
	84	+	t.Errorf("expected stats path %s, got %s", expectedPath, path)
	85	+	}
	86	+
	87	+	// Test Save & Load
	88	+	stats := &RepoStats{
	89	+	Repo: repo,
	90	+	LastRunAt: time.Now().UTC().Format(time.RFC3339),
	91	+	LastOverallStatus: "success",
	92	+	RunsCount: 5,
	93	+	AvgWallDurationMs: 12000,
	94	+	Tasks: map[string]*TaskStats{
	95	+	"build": {
	96	+	AvgDurationMs: 6000,
	97	+	LastDurationMs: 6100,
	98	+	LastStatus: "success",
	99	+	LastExitCode: 0,
	100	+	SeenCount: 5,
	101	+	},
	102	+	},
	103	+	}
	104	+
	105	+	if err := saveRepoStats(stats); err != nil {
	106	+	t.Fatalf("failed to save stats: %v", err)
	107	+	}
	108	+
	109	+	loaded, err := loadRepoStats(repo)
	110	+	if err != nil {
	111	+	t.Fatalf("failed to load stats: %v", err)
	112	+	}
	113	+	if loaded.Repo != repo \|\| loaded.RunsCount != 5 \|\| loaded.AvgWallDurationMs != 12000 {
	114	+	t.Errorf("loaded stats mismatch: %+v", loaded)
	115	+	}
	116	+	if task, ok := loaded.Tasks["build"]; !ok \|\| task.AvgDurationMs != 6000 {
	117	+	t.Errorf("loaded task stats mismatch: %+v", task)
	118	+	}
	119	+	}
	120	+
	121	+	func TestStats_OutcomeTransitions(t *testing.T) {
	122	+	// 1. Fixed transition (failed in prev run, succeeded in current run)
	123	+	t1 := &TaskStats{
	124	+	LastStatus: "failed",
	125	+	LastExitCode: 1,
	126	+	}
	127	+	if trans := computeOutcomeTransition(t1, 0); trans != "fixed" {
	128	+	t.Errorf("expected 'fixed', got '%s'", trans)
	129	+	}
	130	+
	131	+	// 2. Still failing transition (failed in prev run, failed in current run)
	132	+	if trans := computeOutcomeTransition(t1, 1); trans != "still_failing" {
	133	+	t.Errorf("expected 'still_failing', got '%s'", trans)
	134	+	}
	135	+
	136	+	// 3. New failure transition (succeeded in prev run, failed in current run)
	137	+	t2 := &TaskStats{
	138	+	LastStatus: "success",
	139	+	LastExitCode: 0,
	140	+	}
	141	+	if trans := computeOutcomeTransition(t2, 2); trans != "new_failure" {
	142	+	t.Errorf("expected 'new_failure', got '%s'", trans)
	143	+	}
	144	+
	145	+	// 4. Clean success (succeeded in prev run, succeeded in current run)
	146	+	if trans := computeOutcomeTransition(t2, 0); trans != "" {
	147	+	t.Errorf("expected '', got '%s'", trans)
	148	+	}
	149	+
	150	+	// 5. First time seen (nil prev)
	151	+	if trans := computeOutcomeTransition(nil, 0); trans != "" {
	152	+	t.Errorf("expected '', got '%s'", trans)
	153	+	}
	154	+	if trans := computeOutcomeTransition(nil, 1); trans != "new_failure" {
	155	+	t.Errorf("expected 'new_failure', got '%s'", trans)
	156	+	}
	157	+	}

Commit 1a7066e

Commit `1a7066e`