feat(tui,app): surface & correlate background-job elicitations (#3584)

pkg/app/app.go

@@ -194,6 +194,22 @@ func (a *App) Start(ctx context.Context) {
 			case <-ctx.Done():
 			}
 		})
+
+		// Forward elicitation requests raised anywhere in the runtime —
+		// including background-job (run_background_agent) sub-sessions whose
+		// RunStream has no live UI reading its own events channel — so they
+		// always reach the TUI as a dialog. For runtimes that mirror sink
+		// deliveries onto their RunStream channel too (LocalRuntime: the
+		// swap-based bridge best-effort-sends the same event on the very
+		// RunStream channel Run/Retry/RunWithMessage read from below), this
+		// sink is the single, exactly-once delivery point and those loops skip
+		// the mirrored copy (see mustSkipMirroredElicitation). Runtimes that
+		// don't mirror it (RemoteRuntime, whose OnElicitationRequest below is
+		// a no-op) deliver elicitations only through that RunStream copy,
+		// which those loops forward unfiltered (#3584 review).
+		a.runtime.OnElicitationRequest(func(event runtime.Event) {
+			a.sendEvent(ctx, event)
+		})
 	})
 }
 
@@ -499,26 +515,7 @@ func (a *App) Run(ctx context.Context, cancel context.CancelFunc, message string
 		} else {
 			a.session.AddMessage(session.UserMessage(message))
 		}
-		for event := range a.runtime.RunStream(ctx, a.session) {
-			// 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.
-			if ctx.Err() != nil {
-				if _, ok := event.(*runtime.StreamStoppedEvent); ok {
-					// ctx is cancelled; detach cancellation but keep its trace
-					// context so the stop event still reaches subscribers.
-					a.sendEvent(context.WithoutCancel(ctx), event)
-				}
-				continue
-			}
-
-			// Clear titleGenerating flag when title is generated (from server for remote runtime)
-			if _, ok := event.(*runtime.SessionTitleEvent); ok {
-				a.titleGenerating.Store(false)
-			}
-
-			a.sendEvent(ctx, event)
-		}
+		a.forwardRunStreamEvents(ctx, a.runtime.RunStream(ctx, a.session), nil)
 	}()
 }
 
@@ -685,6 +682,91 @@ func (a *App) sendEvent(ctx context.Context, event tea.Msg) {
 	}
 }
 
+// isElicitationRequestEvent reports whether event is an
+// *runtime.ElicitationRequestEvent.
+func isElicitationRequestEvent(event runtime.Event) bool {
+	_, ok := event.(*runtime.ElicitationRequestEvent)
+	return ok
+}
+
+// elicitationSinkMirror is an optional runtime capability satisfied by
+// runtimes whose OnElicitationRequest sink is the single, exactly-once
+// delivery point for elicitation requests even though the runtime ALSO
+// best-effort-mirrors the same event onto its RunStream channel, for the
+// benefit of out-of-process consumers reading RunStream directly (see
+// runtime.LocalRuntime.MirrorsElicitationOnRunStream and elicitationHandler,
+// #3584).
+type elicitationSinkMirror interface {
+	MirrorsElicitationOnRunStream()
+}
+
+// mustSkipMirroredElicitation reports whether rt implements
+// elicitationSinkMirror, i.e. whether Start's OnElicitationRequest sink
+// (below) already delivers every elicitation from rt exactly once, making
+// any *runtime.ElicitationRequestEvent read directly off RunStream a
+// duplicate that Run/Retry/RunWithMessage must skip to avoid a second
+// dialog for the same request.
+//
+// Runtimes that do NOT implement it (RemoteRuntime, whose
+// OnElicitationRequest is a documented no-op) deliver elicitations ONLY via
+// RunStream, so that copy is the sole delivery and must reach a.events
+// unfiltered — an earlier fix skipped this event unconditionally and
+// silently dropped every remote elicitation as a result (#3584 review).
+func mustSkipMirroredElicitation(rt runtime.Runtime) bool {
+	_, ok := rt.(elicitationSinkMirror)
+	return ok
+}
+
+// forwardRunStreamEvents drains ch and forwards each event to a.events,
+// applying the bookkeeping shared by Run, Retry, and RunWithMessage. This is
+// the SINGLE place that decides whether a RunStream event reaches a.events:
+// centralizing it here (instead of three copies of the same loop) is what
+// lets one set of tests cover all three entry points, and makes it
+// impossible for one of them to silently regress independently of the
+// others (#3584 review — Retry and RunWithMessage had duplicated this exact
+// logic untested).
+//
+// filter, when non-nil, runs after the elicitation-mirroring skip below and
+// may itself veto forwarding an event by returning false. Retry uses it to
+// suppress the pre-StreamStarted re-emitted user message; Run and
+// 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)
+	for event := range ch {
+		// 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.
+		if ctx.Err() != nil {
+			if _, ok := event.(*runtime.StreamStoppedEvent); ok {
+				// ctx is cancelled; detach cancellation but keep its trace
+				// context so the stop event still reaches subscribers.
+				a.sendEvent(context.WithoutCancel(ctx), event)
+			}
+			continue
+		}
+
+		// Already delivered via the OnElicitationRequest sink (Start) for
+		// runtimes that mirror it onto RunStream too; skip that duplicate
+		// copy to avoid a second dialog for the same request. Runtimes that
+		// deliver ONLY via RunStream (e.g. RemoteRuntime) are unaffected —
+		// mustSkipMirroredElicitation is false for them (#3584 review).
+		if skipMirroredElicitation && isElicitationRequestEvent(event) {
+			continue
+		}
+
+		if filter != nil && !filter(event) {
+			continue
+		}
+
+		// Clear titleGenerating flag when title is generated (from server for remote runtime)
+		if _, ok := event.(*runtime.SessionTitleEvent); ok {
+			a.titleGenerating.Store(false)
+		}
+
+		a.sendEvent(ctx, event)
+	}
+}
+
 // acquireStreamGuard locks a.streamGuard (set via WithStreamGuard) and
 // returns the matching release func, or a no-op release when no guard is
 // attached (a bare App with no SessionManager to race against). Callers must
@@ -729,32 +811,17 @@ func (a *App) Retry(ctx context.Context, cancel context.CancelFunc) {
 		defer release()
 
 		streamStarted := false
-		for event := range a.runtime.RunStream(ctx, a.session) {
-			// 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.
-			if ctx.Err() != nil {
-				if _, ok := event.(*runtime.StreamStoppedEvent); ok {
-					a.sendEvent(context.WithoutCancel(ctx), event)
-				}
-				continue
-			}
-
+		a.forwardRunStreamEvents(ctx, a.runtime.RunStream(ctx, a.session), func(event runtime.Event) bool {
 			switch event.(type) {
 			case *runtime.StreamStartedEvent:
 				streamStarted = true
 			case *runtime.UserMessageEvent:
 				if !streamStarted {
-					continue
+					return false
 				}
 			}
-
-			if _, ok := event.(*runtime.SessionTitleEvent); ok {
-				a.titleGenerating.Store(false)
-			}
-
-			a.sendEvent(ctx, event)
-		}
+			return true
+		})
 	}()
 }
 
@@ -784,26 +851,7 @@ func (a *App) RunWithMessage(ctx context.Context, cancel context.CancelFunc, msg
 		defer release()
 
 		a.session.AddMessage(msg)
-		for event := range a.runtime.RunStream(ctx, a.session) {
-			// 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.
-			if ctx.Err() != nil {
-				if _, ok := event.(*runtime.StreamStoppedEvent); ok {
-					// ctx is cancelled; detach cancellation but keep its trace
-					// context so the stop event still reaches subscribers.
-					a.sendEvent(context.WithoutCancel(ctx), event)
-				}
-				continue
-			}
-
-			// Clear titleGenerating flag when title is generated (from server for remote runtime)
-			if _, ok := event.(*runtime.SessionTitleEvent); ok {
-				a.titleGenerating.Store(false)
-			}
-
-			a.sendEvent(ctx, event)
-		}
+		a.forwardRunStreamEvents(ctx, a.runtime.RunStream(ctx, a.session), nil)
 	}()
 }
 
@@ -974,9 +1022,13 @@ func (a *App) TogglePause() (paused, supported bool) {
 	return p, true
 }
 
-// ResumeElicitation resumes an elicitation request with the given action and content
-func (a *App) ResumeElicitation(ctx context.Context, action tools.ElicitationAction, content map[string]any) error {
-	return a.runtime.ResumeElicitation(ctx, action, content)
+// ResumeElicitation resumes an elicitation request with the given action and
+// content. elicitationID is variadic, mirroring runtime.Runtime.ResumeElicitation,
+// purely so pre-#3584 3-arg callers keep compiling unchanged; at most the
+// first value is meaningful, and "" (or omitting it) falls back to resolving
+// the sole pending request.
+func (a *App) ResumeElicitation(ctx context.Context, action tools.ElicitationAction, content map[string]any, elicitationID ...string) error {
+	return a.runtime.ResumeElicitation(ctx, action, content, elicitationID...)
 }
 
 func (a *App) NewSession() {