fix(terminal): handle refresh shutdown safely

Author: omarluq

internal/terminal/refresh.go

@@ -40,22 +40,22 @@ type terminalRefreshResult struct {
 
 type terminalRefreshLoader func(context.Context, *terminalRefreshRequest) terminalRefreshSnapshot
 
-func (app *App) requestTerminalRefresh(ctx context.Context) {
+func (app *App) requestTerminalRefresh(ctx context.Context) <-chan struct{} {
 	// Keep the parent task/workflow summary stable while inspecting a child.
 	if len(app.agentTaskSessionStack) > 0 {
-		return
+		return nil
 	}
 
 	if app.refreshInFlight {
 		app.refreshPending = true
 
 		app.logTerminalRefresh("coalesce", 0, "pending")
 
-		return
+		return nil
 	}
 
 	if app.runtime == nil || app.sessionID == "" {
-		return
+		return nil
 	}
 
 	app.refreshGeneration++
@@ -73,8 +73,11 @@ func (app *App) requestTerminalRefresh(ctx context.Context) {
 
 	screen := app.screen
 	diagnostics := app.refreshDiagnostics
+	done := make(chan struct{})
 
 	go func() {
+		defer close(done)
+
 		refreshCtx, cancel := context.WithTimeout(loadCtx, terminalRefreshTimeout)
 		defer cancel()
 
@@ -93,6 +96,8 @@ func (app *App) requestTerminalRefresh(ctx context.Context) {
 		logTerminalRefreshTimings(diagnostics, result, outcome)
 		postTerminalRefreshResult(ctx, screen, result)
 	}()
+
+	return done
 }
 
 func refreshOutcome(result *terminalRefreshResult) string {
@@ -133,10 +138,37 @@ func postTerminalRefreshResult(ctx context.Context, screen terminalScreen, resul
 		return
 	}
 
+	stop := screenStop(screen)
+	select {
+	case <-stop:
+		return
+	default:
+	}
+
+	postTerminalRefreshEvent(ctx, screen.EventQ(), stop, tcell.NewEventInterrupt(result))
+}
+
+func postTerminalRefreshEvent(
+	ctx context.Context,
+	events chan<- tcell.Event,
+	stop <-chan struct{},
+	event tcell.Event,
+) (posted bool) {
+	// Test screens may close their event queue during shutdown. Treat that race
+	// like a stopped screen instead of crashing the refresh worker.
+	defer func() {
+		if recover() != nil {
+			posted = false
+		}
+	}()
+
 	select {
-	case screen.EventQ() <- tcell.NewEventInterrupt(result):
+	case events <- event:
+		return true
 	case <-ctx.Done():
-	case <-screenStop(screen):
+		return false
+	case <-stop:
+		return false
 	}
 }
 
@@ -290,17 +322,10 @@ func (diagnostics *terminalRefreshDiagnostics) log(
 	outcome string,
 	count int,
 ) {
-	attributes := []any{
-		slog.String("operation", operation),
-		slog.Duration("duration", duration),
-		slog.String("outcome", outcome),
-	}
-	if count > 0 {
-		attributes = append(attributes, slog.Int("count", count))
-	}
-
 	logger := slog.Default()
-	logger.Debug("terminal refresh", attributes...)
+	if logger.Enabled(context.Background(), slog.LevelDebug) {
+		logger.Debug("terminal refresh", terminalRefreshAttributes(operation, duration, outcome, count)...)
+	}
 
 	if duration < terminalSlowOperationThreshold {
 		return
@@ -324,6 +349,20 @@ func (diagnostics *terminalRefreshDiagnostics) log(
 	diagnostics.suppressed[key] = 0
 	diagnostics.Unlock()
 
+	attributes := terminalRefreshAttributes(operation, duration, outcome, count)
 	attributes = append(attributes, slog.Int("suppressed_count", suppressed))
 	logger.Warn("terminal refresh slow", attributes...)
 }
+
+func terminalRefreshAttributes(operation string, duration time.Duration, outcome string, count int) []any {
+	attributes := []any{
+		slog.String("operation", operation),
+		slog.Duration("duration", duration),
+		slog.String("outcome", outcome),
+	}
+	if count > 0 {
+		attributes = append(attributes, slog.Int("count", count))
+	}
+
+	return attributes
+}

internal/terminal/refresh_acceptance_internal_test.go

@@ -207,10 +207,12 @@ func TestTerminalRefreshCancellationWhileRuntimeLoadIsBlocked(t *testing.T) {
 	})
 
 	ctx, cancel := context.WithCancel(context.Background())
-	app.requestTerminalRefresh(ctx)
+	done := app.requestTerminalRefresh(ctx)
+	require.NotNil(t, done)
 	awaitSignal(t, started, "refresh did not enter Runtime.AgentTasks")
 	cancel()
 	awaitSignal(t, stub.exited, "Runtime.AgentTasks did not return after application cancellation")
+	awaitSignal(t, done, "refresh worker did not finish after application cancellation")
 
 	select {
 	case event := <-screen.EventQ():
@@ -249,25 +251,42 @@ func TestTerminalRefreshPublicationStopsWithScreen(t *testing.T) {
 	t.Parallel()
 
 	app, screen := newRefreshTestApp(t)
-	for index := 0; index < cap(screen.EventQ()); index++ {
-		screen.EventQ() <- tcell.NewEventInterrupt(index)
+	close(screen.stop)
+
+	postTerminalRefreshResult(t.Context(), app.screen, &terminalRefreshResult{
+		Snapshot: newTerminalRefreshSnapshot(""), SessionID: "", Timing: terminalRefreshTiming{
+			Total: 0, AgentTasks: 0, AgentPanel: 0, ToolTasks: 0,
+			Workflows: 0, Details: 0, WorkflowPanel: 0,
+		},
+		Generation: 0, TimedOut: false, Canceled: false,
+	})
+
+	select {
+	case event := <-screen.EventQ():
+		t.Fatalf("stopped screen received refresh event %T", event)
+	default:
 	}
+}
 
-	done := make(chan struct{})
+func TestTerminalRefreshPublicationHandlesClosedEventQueue(t *testing.T) {
+	t.Parallel()
 
-	go func() {
-		postTerminalRefreshResult(t.Context(), app.screen, &terminalRefreshResult{
+	events := make(chan tcell.Event)
+	close(events)
+
+	posted := postTerminalRefreshEvent(
+		t.Context(),
+		events,
+		nil,
+		tcell.NewEventInterrupt(&terminalRefreshResult{
 			Snapshot: newTerminalRefreshSnapshot(""), SessionID: "", Timing: terminalRefreshTiming{
 				Total: 0, AgentTasks: 0, AgentPanel: 0, ToolTasks: 0,
 				Workflows: 0, Details: 0, WorkflowPanel: 0,
 			},
 			Generation: 0, TimedOut: false, Canceled: false,
-		})
-		close(done)
-	}()
-
-	close(screen.stop)
-	awaitSignal(t, done, "blocked refresh publication did not stop with the screen")
+		}),
+	)
+	assert.False(t, posted)
 }
 
 func TestTerminalRefreshLatestStateCoalescing(t *testing.T) {