fix(web): stop double-publishing live dashboard events

cmd/ralphex/main.go

@@ -517,10 +517,17 @@ func buildNotifyResult(req executePlanRequest, branch, elapsed string, stats git
 // what the planMoveErr note points at: it is non-nil only when the archive was attempted
 // and failed, and it goes last because the warning at the failure site is well above the
 // completion line.
-func displayStats(req executePlanRequest, baseLog *progress.Logger, stats git.DiffStats, elapsed, branch string,
+func displayStats(req executePlanRequest, baseLog *progress.Logger, runnerLog processor.Logger, stats git.DiffStats, elapsed, branch string,
 	planMoved bool, planMoveErr error) {
 	if stats.Files > 0 {
-		baseLog.LogDiffStats(stats.Files, stats.Additions, stats.Deletions)
+		// prefer the runner logger so a --serve dashboard gets the line as well:
+		// the progress logger writes it to the file only, and a live session has no
+		// tailer to read it back.
+		if w, ok := runnerLog.(interface{ LogDiffStats(int, int, int) }); ok {
+			w.LogDiffStats(stats.Files, stats.Additions, stats.Deletions)
+		} else {
+			baseLog.LogDiffStats(stats.Files, stats.Additions, stats.Deletions)
+		}
 		req.Colors.Info().Printf("\ncompleted in %s (%d files, +%d/-%d lines)\n",
 			elapsed, stats.Files, stats.Additions, stats.Deletions)
 	} else {
@@ -690,7 +697,7 @@ func executePlan(ctx context.Context, o opts, req executePlanRequest) error {
 	// notify after the archive so a retained worktree is already recorded when the message goes out
 	sendNotification(req, branch, elapsed, stats, nil)
 
-	displayStats(req, plr.baseLog, stats, elapsed, branch, planMoved, planMoveErr)
+	displayStats(req, plr.baseLog, runnerLog, stats, elapsed, branch, planMoved, planMoveErr)
 	keepDashboardAlive(ctx, o, req, plr.closeLog)
 
 	return nil

cmd/ralphex/main_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/umputun/ralphex/pkg/processor"
 	"github.com/umputun/ralphex/pkg/progress"
 	"github.com/umputun/ralphex/pkg/status"
+	"github.com/umputun/ralphex/pkg/web"
 )
 
 // captureStdout runs fn while redirecting os.Stdout (and the fatih/color Output
@@ -2975,7 +2976,7 @@ func TestDisplayStats(t *testing.T) {
 
 		req := executePlanRequest{PlanFile: "docs/plans/feature.md", Colors: colors}
 		stats := git.DiffStats{Files: 5, Additions: 200, Deletions: 50}
-		displayStats(req, baseLog, stats, "2m15s", "feature-branch", false, nil)
+		displayStats(req, baseLog, baseLog, stats, "2m15s", "feature-branch", false, nil)
 	})
 
 	t.Run("without_diff_stats", func(t *testing.T) {
@@ -2990,7 +2991,30 @@ func TestDisplayStats(t *testing.T) {
 		defer func() { _ = baseLog.Close() }()
 
 		req := executePlanRequest{Colors: colors}
-		displayStats(req, baseLog, git.DiffStats{}, "30s", "main", false, nil)
+		displayStats(req, baseLog, baseLog, git.DiffStats{}, "30s", "main", false, nil)
+	})
+
+	t.Run("diff_stats_reach_broadcast_session", func(t *testing.T) {
+		chdirTemp(t)
+
+		colors := testColors()
+		holder := &status.PhaseHolder{}
+		baseLog, err := progress.NewLogger(progress.Config{
+			PlanFile: "broadcast-stats.md", Mode: "full", Branch: "main", NoColor: true,
+		}, colors, holder)
+		require.NoError(t, err)
+		defer func() { _ = baseLog.Close() }()
+
+		session := web.NewSession("broadcast-stats", baseLog.Path())
+		defer session.Close()
+		bl := web.NewBroadcastLogger(baseLog, session, holder)
+
+		req := executePlanRequest{Colors: colors}
+		displayStats(req, baseLog, bl, git.DiffStats{Files: 2, Additions: 3, Deletions: 1}, "1s", "main", false, nil)
+
+		// a live dashboard session has no tailer to read the DIFFSTATS line back,
+		// so the stats must be recorded through the broadcast logger.
+		require.Equal(t, &web.DiffStats{Files: 2, Additions: 3, Deletions: 1}, session.GetDiffStats())
 	})
 
 	// #450: a worktree archive exists only in the branch, so the path alone names a file that
@@ -3014,7 +3038,7 @@ func TestDisplayStats(t *testing.T) {
 			WtPreserve: &atomic.Bool{},
 		}
 		output := captureStdout(t, func() {
-			displayStats(req, baseLog, git.DiffStats{Files: 1, Additions: 10, Deletions: 5}, "10s", "feature-wt", true, nil)
+			displayStats(req, baseLog, baseLog, git.DiffStats{Files: 1, Additions: 10, Deletions: 5}, "10s", "feature-wt", true, nil)
 		})
 		assert.Contains(t, output,
 			"  plan: "+filepath.Join("docs", "plans", "completed", "feature.md")+" (committed on branch feature-wt)\n")
@@ -3033,7 +3057,7 @@ func TestDisplayStats(t *testing.T) {
 
 		req := executePlanRequest{PlanFile: filepath.Join("docs", "plans", "feature.md"), Colors: colors}
 		output := captureStdout(t, func() {
-			displayStats(req, baseLog, git.DiffStats{}, "10s", "feature", true, nil)
+			displayStats(req, baseLog, baseLog, git.DiffStats{}, "10s", "feature", true, nil)
 		})
 		assert.Contains(t, output, "  plan: "+filepath.Join("docs", "plans", "completed", "feature.md")+"\n")
 		assert.NotContains(t, output, "committed on branch")
@@ -3112,7 +3136,7 @@ func TestDisplayStats(t *testing.T) {
 				req.Colors = colors
 
 				output := captureStdout(t, func() {
-					displayStats(req, baseLog, git.DiffStats{}, "1s", "main", tc.planMoved, tc.planMoveErr)
+					displayStats(req, baseLog, baseLog, git.DiffStats{}, "1s", "main", tc.planMoved, tc.planMoveErr)
 				})
 				assert.Contains(t, output, "  plan: "+tc.wantPath+"\n")
 				if tc.wantNote {

pkg/web/broadcast_logger.go

@@ -22,6 +22,11 @@ type Logger interface {
 	Path() string
 }
 
+// DiffStatsWriter is implemented by loggers that persist diff statistics.
+type DiffStatsWriter interface {
+	LogDiffStats(files, additions, deletions int)
+}
+
 // BroadcastLogger wraps a Logger and broadcasts events to SSE clients.
 // implements the decorator pattern - all calls are forwarded to the inner logger
 // while also being converted to events for web streaming.
@@ -147,6 +152,18 @@ func (b *BroadcastLogger) LogDraftReview(action, feedback string) {
 	}
 }
 
+// LogDiffStats records completion diff stats in the wrapped logger and publishes the
+// same line for the dashboard. the progress logger writes this line to the file only,
+// so a live session (which has no tailer by design) would otherwise never see it.
+func (b *BroadcastLogger) LogDiffStats(files, additions, deletions int) {
+	if w, ok := b.inner.(DiffStatsWriter); ok {
+		w.LogDiffStats(files, additions, deletions)
+	}
+	b.session.SetDiffStats(DiffStats{Files: files, Additions: additions, Deletions: deletions})
+	b.broadcast(NewOutputEvent(b.holder.Get(),
+		fmt.Sprintf("DIFFSTATS: files=%d additions=%d deletions=%d", files, additions, deletions)))
+}
+
 // Path returns the progress file path.
 func (b *BroadcastLogger) Path() string {
 	return b.inner.Path()

pkg/web/broadcast_logger_test.go

@@ -402,6 +402,17 @@ func TestBroadcastLogger_LogDraftReview_ReviseWithFeedback(t *testing.T) {
 	assert.Equal(t, "Please add more details to Task 3", mockLogger.LogDraftReviewCalls()[0].Feedback)
 }
 
+func TestBroadcastLogger_LogDiffStats(t *testing.T) {
+	session := NewSession("test", "/tmp/test.txt")
+	defer session.Close()
+
+	bl := NewBroadcastLogger(&mocks.LoggerMock{}, session, &status.PhaseHolder{})
+	bl.LogDiffStats(4, 12, 3)
+
+	// the live session has no tailer, so the wrapper itself must record the stats
+	require.Equal(t, &DiffStats{Files: 4, Additions: 12, Deletions: 3}, session.GetDiffStats())
+}
+
 func TestExtractTerminalSignal(t *testing.T) {
 	cases := []struct {
 		name   string

pkg/web/dashboard.go

@@ -73,6 +73,13 @@ func NewDashboard(cfg DashboardConfig, holder *status.PhaseHolder) *Dashboard {
 func (d *Dashboard) Start(ctx context.Context) (*BroadcastLogger, error) {
 	// create session for SSE streaming (handles both live streaming and history replay)
 	session := NewSession("main", d.baseLog.Path())
+	// the run starts now, so the card is active even before watcher discovery
+	// (which only happens when a watch dir covers the project).
+	session.SetState(SessionStateActive)
+	// events for this session come from the BroadcastLogger, not from a file tailer.
+	// without the mark, watcher discovery would start a tailer on the same progress
+	// file and publish every live line a second time.
+	session.markLiveOutput()
 	broadcastLog := NewBroadcastLogger(d.baseLog, session, d.holder)
 
 	// extract plan name for display

pkg/web/dashboard_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -301,6 +302,66 @@ func TestDashboard_PrintWatchInfo(t *testing.T) {
 	d.printWatchInfo([]string{"/tmp", "/var"})
 }
 
+func TestDashboard_Start_MultiSessionDoesNotDuplicateLiveEvents(t *testing.T) {
+	tmpDir := t.TempDir()
+	oldWd, wdErr := os.Getwd()
+	require.NoError(t, wdErr)
+	require.NoError(t, os.Chdir(tmpDir))
+	t.Cleanup(func() { _ = os.Chdir(oldWd) })
+
+	colors := testColors()
+	holder := &status.PhaseHolder{}
+	baseLog, err := progress.NewLogger(progress.Config{
+		PlanFile: "test.md",
+		Mode:     "full",
+		Branch:   "main",
+		NoColor:  true,
+	}, colors, holder)
+	require.NoError(t, err)
+	defer baseLog.Close()
+
+	cfg := DashboardConfig{
+		BaseLog:   baseLog,
+		Port:      0, // use random port
+		PlanFile:  "test.md",
+		Branch:    "main",
+		WatchDirs: []string{tmpDir}, // multi-session mode
+		Colors:    colors,
+	}
+
+	d := NewDashboard(cfg, holder)
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	broadcastLog, err := d.Start(ctx)
+	require.NoError(t, err)
+	require.NotNil(t, broadcastLog)
+
+	// the watcher discovers the progress file of the live session. a tailer started
+	// for it would republish every line the BroadcastLogger already streamed.
+	sess := broadcastLog.session
+	require.Eventually(t, func() bool {
+		return sess.GetState() == SessionStateActive
+	}, 2*time.Second, 10*time.Millisecond, "watcher should discover the live progress file")
+	time.Sleep(200 * time.Millisecond) // allow an erroneous tailer to start
+	assert.False(t, sess.IsTailing(), "live session must not be tailed; file events would duplicate live events")
+
+	events, cleanup := subscribeSSEEvents(t, sess)
+	defer cleanup()
+	_ = drainChannel(events, 200*time.Millisecond)
+
+	broadcastLog.Print("unique live line %d", 7)
+	delivered := drainChannel(events, 500*time.Millisecond)
+
+	var matches int
+	for _, ev := range delivered {
+		if strings.Contains(ev, "unique live line 7") {
+			matches++
+		}
+	}
+	assert.Equal(t, 1, matches, "live line must reach SSE exactly once")
+}
+
 func TestConnectHost(t *testing.T) {
 	tests := []struct {
 		host, want string

pkg/web/session.go

@@ -106,6 +106,12 @@ type Session struct {
 	state    SessionState    // current state (active/completed)
 	tailer   *Tailer         // file tailer for reading new content (nil if not tailing)
 
+	// liveOutput is true for the in-process session registered by Dashboard.Start.
+	// its events are published directly by the BroadcastLogger, so a tailer must
+	// not run for it: reading the same progress file would publish every line twice.
+	// set once before registration; guarded by mu for readers on watcher goroutines.
+	liveOutput bool
+
 	// lastModified tracks the file's last modification time for change detection
 	lastModified time.Time
 
@@ -229,6 +235,32 @@ func (s *Session) GetState() SessionState {
 	return s.state
 }
 
+// markLiveOutput flags the session as fed directly by a BroadcastLogger.
+// called once by Dashboard.Start before the session is registered so that
+// discovery, reactivation, and start-tailing never attach a second event source.
+func (s *Session) markLiveOutput() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.liveOutput = true
+}
+
+// isLiveOutput reports whether the session's events come from a BroadcastLogger.
+func (s *Session) isLiveOutput() bool {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.liveOutput
+}
+
+// clearLiveOutput removes the BroadcastLogger mark once the run it belonged to has
+// finished. the dashboard process can outlive the run (keepDashboardAlive) and later
+// observe another run of the same plan path: that run has no broadcast logger, so it
+// must be served by the ordinary discovery/tailer path again.
+func (s *Session) clearLiveOutput() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.liveOutput = false
+}
+
 // GetTailer returns the session's tailer thread-safely.
 func (s *Session) GetTailer() *Tailer {
 	s.mu.RLock()
@@ -308,6 +340,9 @@ func (s *Session) resetForNewRun() {
 	s.lastTask = 0
 	s.loaded = false
 	s.diffStats = nil
+	// a new run on the same progress file is not fed by the previous run's
+	// BroadcastLogger, so hand it back to the discovery/tailer path.
+	s.liveOutput = false
 }
 
 // tailerStartMode selects how startTailerLocked begins tailing.
@@ -384,6 +419,10 @@ func (s *Session) StartTailing(fromStart bool) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
+	if s.liveOutput {
+		return nil // events already stream live through the BroadcastLogger
+	}
+
 	if s.stopping {
 		return nil // StopTailing mid-flight; caller retries on next event
 	}
@@ -422,6 +461,10 @@ func (s *Session) Reactivate() error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
+	if s.liveOutput {
+		return nil // events already stream live through the BroadcastLogger
+	}
+
 	if s.stopping {
 		return nil // StopTailing mid-flight; watcher retries on next Write event
 	}

pkg/web/session_manager.go

@@ -209,8 +209,10 @@ func (m *SessionManager) updateSession(session *Session) error {
 	// pointing inside an unfinished header. without this gate, a later Reactivate
 	// would resume from mid-header and emit the remaining header lines as output
 	// events. the loader will run on a later updateSession call once the header
-	// separator is visible.
-	if newState == SessionStateCompleted && headerComplete && session.MarkLoadedIfNot() {
+	// separator is visible. live-output sessions are skipped before MarkLoadedIfNot
+	// so they are not marked loaded without loading anything; their content is
+	// already published by the BroadcastLogger.
+	if newState == SessionStateCompleted && headerComplete && !session.isLiveOutput() && session.MarkLoadedIfNot() {
 		if session.getLastOffset() == 0 {
 			m.loadProgressFileIntoSession(session.Path, session)
 		}
@@ -387,8 +389,9 @@ func (m *SessionManager) RefreshStates() {
 	m.mu.RUnlock()
 
 	for _, session := range sessions {
-		// only check sessions that are currently tailing
-		if !session.IsTailing() {
+		// live-output sessions never tail; refresh their state too so a finished
+		// run does not stay marked active in the dashboard.
+		if !session.IsTailing() && !session.isLiveOutput() {
 			continue
 		}
 
@@ -402,6 +405,9 @@ func (m *SessionManager) RefreshStates() {
 			// session completed, update state and stop tailing
 			session.SetState(SessionStateCompleted)
 			session.StopTailing()
+			// the run that the BroadcastLogger fed has finished; a later run of the
+			// same plan path must fall back to the discovery/tailer path.
+			session.clearLiveOutput()
 		}
 	}
 }