fix(review): TUI sink must never backpressure the orchestrator

a91d234→main·

payment-alt·1w ago·2 files·+263 added/-28 removed

During a live full-crew dogfood run (2026-07-07, run 6), the parent entire process wedged mid-run: the TUI stopped rendering (elapsed froze at 11m58s, spinner ticks dead), the armed --timeout 20m never surfaced, no judge ran, and the process had to be killed externally 15 minutes later. The mechanism is structural: Program.Send is an unbuffered BLOCKING send, and TUISink called it directly from the orchestrator's serial dispatch goroutine — so any stall in the Bubble Tea Update/render pipeline freezes sink dispatch, the bounded fanIn drain loop, the forwarding goroutines, the stdout parsers, and reviewer-timeout event handling with it. (The stall's own trigger was not reproduced in three targeted shim experiments; this closes the amplification path that turned a display stall into a full orchestrator freeze.)

TUISink now enqueues onto a bounded internal queue drained by a pump goroutine — the only goroutine allowed to block on Send. Display events (AgentEvent) never block: overflow beyond the 4096-message cap is dropped and counted. Rare control messages (run summary, phase transitions, quit) use a bounded wait, and PostRunComplete keeps its Kill fallback, so a wedged TUI degrades to lost frames and a stale footer instead of a hung run. Order is preserved through the single queue for a healthy program.

Tests inject a deterministically wedged teaRunner: AgentEvent must complete 3x the queue cap without blocking, control messages must return within their bounded wait, and a recording runner pins FIFO delivery.

Co-Authored-By: Claude Fable 5 noreply@anthropic.com

Sessions

01KX0HNDFDWYN2F02BKXA5AJ6AView transcript

Changes

2

• cmd/entire/cli/review

• Mtui_sink.go+95/-28

• Mtui_sink_test.go+168

17 unmodified lines

// teaRunner is the slice of *tea.Program the sink depends on, extracted so  
// tests can substitute a program with a deterministically stalled event loop.  
type teaRunner interface {  
    Run() (tea.Model, error)  
    Send(msg tea.Msg)  
    Kill()  
}

// tuiSinkQueueCap bounds the sink's internal dispatch queue. Program.Send is  
// an unbuffered BLOCKING send: if the Bubble Tea Update/render pipeline ever  
// stalls, a direct Send from the orchestrator's dispatch goroutine parks  
// forever — freezing sink dispatch, the fanIn drain loop, the parsers, and  
// reviewer-timeout handling with it (observed live: the 2026-07-07 run-6  
// wedge, where the TUI froze mid-run and a 20m --timeout never surfaced).  
// The queue absorbs bursts; overflow beyond the cap is dropped and counted —  
// a display that can lag must never backpressure the data plane.  
const tuiSinkQueueCap = 4096

// TUISink is a Sink that renders a Bubble Tea dashboard. The orchestrator  
// calls AgentEvent/RunFinished from a single goroutine (CU4 serial-dispatch  
// contract); the sink translates each event into a tea.Msg and sends it via  
// Program.Send. Bubble Tea's Send is thread-safe, but we never rely on that  
// property — the serial-dispatch promise means Send is only called from the  
// orchestrator's dispatch goroutine.  
// contract); the sink translates each event into a tea.Msg, enqueues it on a  
// bounded internal queue, and a pump goroutine forwards it via Program.Send —  
// so only the pump can ever block on a stalled Bubble Tea loop, never the  
// orchestrator.

// Cancellation: cancel is the same context.CancelFunc that controls the  
// orchestrator's run context. The first KeyCtrlC in the dashboard fires this
// root's context, which cancels the same function — no parallel signal.Notify  
// goroutine is needed here.

// Compile-time interface check.

// wedgedProgram is a teaRunner whose event loop never consumes messages:
// Send blocks until Kill, modeling a Bubble Tea program whose Update/render
// pipeline has stalled (the 2026-07-07 run-6 incident shape). Run blocks
// until Kill so the sink's done channel behaves like a live program's.

// TestTUISink_AgentEventNeverBlocksWhenProgramLoopIsWedged pins the wedge
// hardening: a stalled Bubble Tea loop must never backpressure the
// orchestrator. Before the fix, the first AgentEvent after the stall parked
// forever inside Program.Send, freezing sink dispatch, the fanIn drain loop,
// the parsers, and reviewer-timeout handling with them.
func TestTUISink_AgentEventNeverBlocksWhenProgramLoopIsWedged(t *testing.T) {
t.Parallel()
prog := newWedgedProgram()
sink := newTUISinkWithProgram(prog)
sink.Start()
defer func() {
prog.Kill()
sink.Wait()
}()

finished := make(chan struct{})  
go func() {  
    for range 3 * tuiSinkQueueCap {  
        sink.AgentEvent("agent-a", reviewtypes.AssistantText{Text: "x"})  
    }  
    close(finished)  
}()

select {
case <-finished:
case <-time.After(5 * time.Second):
t.Fatal("AgentEvent blocked on a wedged TUI loop — orchestrator freeze")
}

if got := sink.droppedCount(); got == 0 {
t.Error("expected overflow drops to be counted when the queue jams")
}
}

// TestTUISink_EventsReachProgramInOrder pins that the async pump preserves
// dispatch order for a healthy program. func TestTUISink_EventsReachProgramInOrder(t *testing.T) {
t.Parallel()
prog := newRecordingProgram()
sink := newTUISinkWithProgram(prog)
sink.Start()
defer func() {
prog.Kill()
sink.Wait()
}()

for i := range 50 {
sink.AgentEvent("agent-a", reviewtypes.AssistantText{Text: string(rune('a' + i%26))})
}
sink.RunFinished(reviewtypes.RunSummary{})

deadline := time.After(5 * time.Second)
for {
msgs := prog.recorded()
if len(msgs) >= 51 {
for i := range 50 {
if _, ok := msgs[i].(agentEventMsg); !ok {
t.Fatalf("msgs[%d] = %T, want agentEventMsg", i, msgs[i])
}
}
if _, ok := msgs[50].(runFinishedMsg); !ok {
t.Fatalf("msgs[50] = %T, want runFinishedMsg (order violated)", msgs[50])
}
return
}
select {
case <-deadline:
t.Fatalf("only %d/51 messages reached the program", len(msgs))
case <-time.After(10 * time.Millisecond):
}
}
}

// TestTUISink_RunFinishedBoundedWhenWedged pins that control messages use a
// bounded wait rather than blocking forever when the queue is jammed.
func TestTUISink_RunFinishedBoundedWhenWedged(t *testing.T) {
t.Parallel()
prog := newWedgedProgram()
sink := newTUISinkWithProgram(prog)
sink.Start()
defer func() {
prog.Kill()
sink.Wait()
}()

// Jam the queue.
for range 2 * tuiSinkQueueCap {
sink.AgentEvent("agent-a", reviewtypes.AssistantText{Text: "x"})
}

finished := make(chan struct{})
go func() {
sink.RunFinished(reviewtypes.RunSummary{})
close(finished)
}()
select {
case <-finished:
case <-time.After(tuiPostRunCompleteGrace + 3*time.Second):
t.Fatal("RunFinished blocked past its bounded wait on a wedged TUI")
}
}

Mcmd/entire/cli/review/tui_sink_test.go+168