fix(app): synthesize StreamStopped when the runtime stream closes without one

pkg/app/app.go

@@ -421,6 +421,7 @@ func (a *App) RunSkillFork(ctx context.Context, cancel context.CancelFunc, skill
 	// so the supervisor marks the session idle.
 	go func() {
 		events := make(chan runtime.Event, defaultRuntimeEventBuffer)
+		var failed atomic.Bool
 		go func() {
 			defer close(events)
 			result, err := a.runtime.RunSkillFork(ctx, a.session, skillstool.RunSkillArgs{
@@ -430,16 +431,41 @@ func (a *App) RunSkillFork(ctx context.Context, cancel context.CancelFunc, skill
 			switch {
 			case errors.Is(err, runtime.ErrUnsupported):
 				slog.WarnContext(ctx, "Runtime does not support fork-mode skills; skill not executed", "skill", skillName)
+				failed.Store(true)
 				a.sendEvent(ctx, runtime.Error(fmt.Sprintf("Skill %q cannot run: this runtime does not support fork-mode skills.", skillName)))
 			case err != nil:
 				slog.ErrorContext(ctx, "Failed to run fork-mode skill", "skill", skillName, "error", err)
+				failed.Store(true)
 				a.sendEvent(ctx, runtime.Error(fmt.Sprintf("Skill %q failed: %v", skillName, err)))
 			case result != nil && result.IsError:
+				failed.Store(true)
 				a.sendEvent(ctx, runtime.Error(result.Output))
 			}
 		}()
 
+		// sawStop tracks ANY StreamStoppedEvent forwarded here — unlike
+		// forwardRunStreamEvents' root-only rule (#4136), every event on this
+		// channel belongs to the fork's own sub-session, so a root-session
+		// check would always be false here and synthesize a spurious
+		// duplicate stop on every fork run, successful or not.
+		var (
+			sawStop       bool
+			lastSessionID string
+			agentName     string
+		)
 		for event := range events {
+			if scoped, ok := event.(runtime.SessionScoped); ok {
+				if id := scoped.GetSessionID(); id != "" {
+					lastSessionID = id
+				}
+			}
+			if name := event.GetAgentName(); name != "" {
+				agentName = name
+			}
+			if _, ok := event.(*runtime.StreamStoppedEvent); ok {
+				sawStop = true
+			}
+
 			if ctx.Err() != nil {
 				if _, ok := event.(*runtime.StreamStoppedEvent); ok {
 					// ctx is cancelled; detach cancellation but keep its trace
@@ -450,6 +476,10 @@ func (a *App) RunSkillFork(ctx context.Context, cancel context.CancelFunc, skill
 			}
 			a.sendEvent(ctx, event)
 		}
+
+		if !sawStop {
+			a.synthesizeStreamStopped(ctx, cmp.Or(lastSessionID, a.session.ID), agentName, failed.Load())
+		}
 	}()
 }
 
@@ -758,7 +788,36 @@ func mustSkipMirroredElicitation(rt runtime.Runtime) bool {
 // RunWithMessage pass nil.
 func (a *App) forwardRunStreamEvents(ctx context.Context, ch <-chan runtime.Event, filter func(event runtime.Event) (forward bool)) {
 	skipMirroredElicitation := mustSkipMirroredElicitation(a.runtime)
+
+	// sawRootStop/sawRootError/agentName drive the #4136 fallback below: the
+	// runtime documents channel close (not receipt of StreamStoppedEvent) as
+	// the terminal signal, and may drop the event under back-pressure
+	// (LocalRuntime.finalizeEventChannel). Only a root-session stop counts —
+	// a sub-session (delegation/fork) stop leaves the root stream running.
+	var (
+		sawRootStop  bool
+		sawRootError bool
+		agentName    string
+	)
+
 	for event := range ch {
+		isRoot := isRootSessionEvent(event, a.session.ID)
+		if isRoot {
+			if name := event.GetAgentName(); name != "" {
+				agentName = name
+			}
+			switch event.(type) {
+			case *runtime.StreamStoppedEvent:
+				// Tracked here — before the cancellation/filter/dedupe checks
+				// below — so a filter that happens to veto a genuine root stop
+				// can never leave this bookkeeping thinking none arrived and
+				// synthesize a spurious duplicate.
+				sawRootStop = true
+			case *runtime.ErrorEvent:
+				sawRootError = true
+			}
+		}
+
 		// If context is cancelled, continue draining but don't forward events
 		// — except StreamStoppedEvent, which must always propagate so the
 		// supervisor can mark the session as no longer running.
@@ -791,6 +850,51 @@ func (a *App) forwardRunStreamEvents(ctx context.Context, ch <-chan runtime.Even
 
 		a.sendEvent(ctx, event)
 	}
+
+	if !sawRootStop {
+		a.synthesizeStreamStopped(ctx, a.session.ID, agentName, sawRootError)
+	}
+}
+
+// isRootSessionEvent reports whether event belongs to the given root
+// session rather than a sub-session (delegation or fork-skill child).
+// Events that don't implement [runtime.SessionScoped], or that carry an
+// empty SessionID, are treated as root-scoped by convention — matching how
+// [runtime.SessionScoped] consumers elsewhere (e.g. the supervisor's
+// isTopLevelStream) already interpret an absent session id.
+func isRootSessionEvent(event runtime.Event, sessionID string) bool {
+	scoped, ok := event.(runtime.SessionScoped)
+	if !ok {
+		return true
+	}
+	id := scoped.GetSessionID()
+	return id == "" || id == sessionID
+}
+
+// synthesizeStreamStopped sends a StreamStoppedEvent for sessionID when the
+// runtime's event channel closed without forwarding one (#4136). Consumers
+// (the chat page, the supervisor) clear their "Working…" state only on
+// receipt of this event and treat channel close as the terminal signal only
+// in the runtime's documentation, not in their own code — so a dropped
+// event (see LocalRuntime.finalizeEventChannel's non-blocking emit) leaves
+// them stuck indefinitely without this fallback.
+func (a *App) synthesizeStreamStopped(ctx context.Context, sessionID, agentName string, sawError bool) {
+	reason := runtime.TurnEndReasonNormal
+	switch {
+	case ctx.Err() != nil:
+		reason = runtime.TurnEndReasonCanceled
+	case sawError:
+		reason = runtime.TurnEndReasonError
+	}
+	if agentName == "" {
+		agentName = a.runtime.CurrentAgentName(ctx)
+	}
+	slog.WarnContext(ctx, "runtime stream closed without a StreamStoppedEvent; synthesizing one",
+		"session_id", sessionID, "reason", reason)
+	// ctx may already be cancelled; detach cancellation but keep its trace
+	// context so the synthesized stop still reaches subscribers, mirroring
+	// the ctx-cancelled StreamStoppedEvent forwarding above.
+	a.sendEvent(context.WithoutCancel(ctx), runtime.StreamStopped(sessionID, agentName, reason))
 }
 
 // acquireStreamGuard locks a.streamGuard (set via WithStreamGuard) and

pkg/app/app_test.go

@@ -1240,3 +1240,215 @@ func TestElicitationDeliveryAcrossEntryPoints(t *testing.T) {
 		})
 	}
 }
+
+// scriptedStreamMockRuntime replays a fixed sequence of events on RunStream
+// and then closes the channel, without appending a StreamStoppedEvent
+// unless the script itself contains one. This drives the #4136 regression
+// tests below through every shape the drain loop must fall back for: no
+// stop at all, a real root stop, a sub-session-only stop, a mid-stream
+// cancellation, and a root error — uniformly across every entry point in
+// elicitationEntryPoints.
+type scriptedStreamMockRuntime struct {
+	mockRuntime
+
+	script []runtime.Event
+	// cancelAfterScript, when set, is called after every scripted event has
+	// been pushed onto the channel and before it closes, simulating ctx
+	// being cancelled mid-stream (e.g. Esc) with the channel closing right
+	// after — without it ever carrying a StreamStoppedEvent of its own.
+	cancelAfterScript context.CancelFunc
+}
+
+func (m *scriptedStreamMockRuntime) RunStream(_ context.Context, _ *session.Session) <-chan runtime.Event {
+	ch := make(chan runtime.Event, len(m.script)+1)
+	go func() {
+		defer close(ch)
+		for _, ev := range m.script {
+			ch <- ev
+		}
+		if m.cancelAfterScript != nil {
+			m.cancelAfterScript()
+		}
+	}()
+	return ch
+}
+
+// collectUntilQuietWindow is how long collectUntilQuiet waits for a lull
+// before deciding the producer is done.
+const collectUntilQuietWindow = 500 * time.Millisecond
+
+// collectUntilQuiet drains events until no further message arrives within
+// collectUntilQuietWindow, returning everything collected in order. The
+// #4136 regression tests below have no single sentinel event to stop on (a
+// synthesized StreamStoppedEvent looks just like a real one, and a
+// sub-session stop can arrive before it), so a fixed quiet window is the
+// simplest way to know forwardRunStreamEvents' goroutine has finished.
+func collectUntilQuiet(t *testing.T, events <-chan tea.Msg) []tea.Msg {
+	t.Helper()
+
+	var collected []tea.Msg
+	for {
+		select {
+		case msg := <-events:
+			collected = append(collected, msg)
+		case <-time.After(collectUntilQuietWindow):
+			return collected
+		}
+	}
+}
+
+// streamStoppedEvents returns every *runtime.StreamStoppedEvent in msgs, in
+// order.
+func streamStoppedEvents(msgs []tea.Msg) []*runtime.StreamStoppedEvent {
+	var stops []*runtime.StreamStoppedEvent
+	for _, msg := range msgs {
+		if stop, ok := msg.(*runtime.StreamStoppedEvent); ok {
+			stops = append(stops, stop)
+		}
+	}
+	return stops
+}
+
+// TestForwardRunStreamEvents_SynthesizesRootStreamStopped is the regression
+// coverage for #4136 (TUI stuck on "Working…" after the stream ends): the
+// runtime documents the channel close, not receipt of StreamStoppedEvent,
+// as the terminal signal, and drops the event under back-pressure
+// (LocalRuntime.finalizeEventChannel's non-blocking emit) — a fast local
+// model streaming faster than the persistence observer can keep up
+// reliably triggers this. Run/Retry/RunWithMessage all funnel through
+// forwardRunStreamEvents, so every case is driven across all three via
+// elicitationEntryPoints, mirroring TestElicitationDeliveryAcrossEntryPoints.
+func TestForwardRunStreamEvents_SynthesizesRootStreamStopped(t *testing.T) {
+	t.Parallel()
+
+	for _, ep := range elicitationEntryPoints {
+		t.Run(ep.name, func(t *testing.T) {
+			t.Parallel()
+
+			t.Run("no stop event: synthesizes exactly one, reason normal, as the last event", func(t *testing.T) {
+				t.Parallel()
+
+				sess := session.New()
+				rt := &scriptedStreamMockRuntime{script: []runtime.Event{
+					runtime.StreamStarted(sess.ID, "mock"),
+					runtime.AgentChoice("mock", sess.ID, "partial content"),
+				}}
+				events := make(chan tea.Msg, 16)
+				app := &App{runtime: rt, session: sess, events: events}
+
+				ctx, cancel := context.WithCancel(t.Context())
+				defer cancel()
+				ep.invoke(app, ctx, cancel)
+
+				collected := collectUntilQuiet(t, events)
+				require.NotEmpty(t, collected, "the script's own events must still be forwarded")
+				stops := streamStoppedEvents(collected)
+				require.Len(t, stops, 1, "exactly one StreamStoppedEvent must be synthesized")
+				assert.Equal(t, sess.ID, stops[0].SessionID)
+				assert.Equal(t, "normal", stops[0].Reason)
+				assert.Equal(t, "mock", stops[0].AgentName, "must fall back to the last observed root-session agent name")
+				assert.Same(t, tea.Msg(stops[0]), collected[len(collected)-1],
+					"the synthesized stop must be the last event forwarded")
+			})
+
+			t.Run("real root stop: no duplicate is synthesized", func(t *testing.T) {
+				t.Parallel()
+
+				sess := session.New()
+				rt := &scriptedStreamMockRuntime{script: []runtime.Event{
+					runtime.StreamStarted(sess.ID, "mock"),
+					runtime.AgentChoice("mock", sess.ID, "partial content"),
+					runtime.StreamStopped(sess.ID, "mock", "normal"),
+				}}
+				events := make(chan tea.Msg, 16)
+				app := &App{runtime: rt, session: sess, events: events}
+
+				ctx, cancel := context.WithCancel(t.Context())
+				defer cancel()
+				ep.invoke(app, ctx, cancel)
+
+				collected := collectUntilQuiet(t, events)
+				stops := streamStoppedEvents(collected)
+				require.Len(t, stops, 1, "the real StreamStoppedEvent must not be duplicated")
+				assert.Equal(t, sess.ID, stops[0].SessionID)
+				assert.Equal(t, "normal", stops[0].Reason)
+			})
+
+			t.Run("sub-session stop only: root stop is still synthesized", func(t *testing.T) {
+				t.Parallel()
+
+				sess := session.New()
+				const subSessionID = "sub-session"
+				rt := &scriptedStreamMockRuntime{script: []runtime.Event{
+					runtime.StreamStarted(sess.ID, "mock"),
+					runtime.StreamStopped(subSessionID, "worker", "normal"),
+				}}
+				events := make(chan tea.Msg, 16)
+				app := &App{runtime: rt, session: sess, events: events}
+
+				ctx, cancel := context.WithCancel(t.Context())
+				defer cancel()
+				ep.invoke(app, ctx, cancel)
+
+				collected := collectUntilQuiet(t, events)
+				stops := streamStoppedEvents(collected)
+				require.Len(t, stops, 2, "the sub-session stop must not satisfy the root fallback")
+				assert.Equal(t, subSessionID, stops[0].SessionID, "the sub-session's own stop is forwarded first")
+				assert.Equal(t, sess.ID, stops[1].SessionID, "a root stop must still be synthesized")
+				assert.Equal(t, "normal", stops[1].Reason)
+				assert.Equal(t, "mock", stops[1].AgentName,
+					"must use the root StreamStarted's agent name, not the sub-session's")
+				assert.Same(t, tea.Msg(stops[1]), collected[len(collected)-1],
+					"the synthesized root stop must be the last event forwarded")
+			})
+
+			t.Run("ctx cancelled mid-stream: synthesized with reason canceled", func(t *testing.T) {
+				t.Parallel()
+
+				sess := session.New()
+				ctx, cancel := context.WithCancel(t.Context())
+				defer cancel()
+				rt := &scriptedStreamMockRuntime{
+					script: []runtime.Event{
+						runtime.StreamStarted(sess.ID, "mock"),
+						runtime.AgentChoice("mock", sess.ID, "partial content"),
+					},
+					cancelAfterScript: cancel,
+				}
+				events := make(chan tea.Msg, 16)
+				app := &App{runtime: rt, session: sess, events: events}
+
+				ep.invoke(app, ctx, cancel)
+
+				collected := collectUntilQuiet(t, events)
+				stops := streamStoppedEvents(collected)
+				require.Len(t, stops, 1, "exactly one StreamStoppedEvent must be synthesized")
+				assert.Equal(t, sess.ID, stops[0].SessionID)
+				assert.Equal(t, "canceled", stops[0].Reason)
+			})
+
+			t.Run("root error event: synthesized with reason error", func(t *testing.T) {
+				t.Parallel()
+
+				sess := session.New()
+				rt := &scriptedStreamMockRuntime{script: []runtime.Event{
+					runtime.StreamStarted(sess.ID, "mock"),
+					runtime.ErrorForSession(sess.ID, "boom"),
+				}}
+				events := make(chan tea.Msg, 16)
+				app := &App{runtime: rt, session: sess, events: events}
+
+				ctx, cancel := context.WithCancel(t.Context())
+				defer cancel()
+				ep.invoke(app, ctx, cancel)
+
+				collected := collectUntilQuiet(t, events)
+				stops := streamStoppedEvents(collected)
+				require.Len(t, stops, 1, "exactly one StreamStoppedEvent must be synthesized")
+				assert.Equal(t, sess.ID, stops[0].SessionID)
+				assert.Equal(t, "error", stops[0].Reason)
+				assert.Equal(t, "mock", stops[0].AgentName)
+			})
+		})
+	}
+}