diff --git a/docs/features/api-server/index.md b/docs/features/api-server/index.md index 2a1ba6432d..3943f55c4c 100644 --- a/docs/features/api-server/index.md +++ b/docs/features/api-server/index.md @@ -46,7 +46,7 @@ All endpoints are under the `/api` prefix. | `GET` | `/api/sessions/:id` | Get a session by ID (messages, tokens, permissions) | | `GET` | `/api/sessions/:id/status` | Lightweight runtime state (streaming, title, agent, tokens). Requires an attached runtime. | | `GET` | `/api/sessions/:id/snapshot` | Full state in one call (stored fields + runtime state + `last_event_seq`) for gapless resync — see [Reconnecting without gaps](#reconnecting-without-gaps). | -| `GET` | `/api/sessions/:id/events` | Live session event stream (SSE) with sequence numbers and replay. Available for a run attached via [`--listen`](#listen). | +| `GET` | `/api/sessions/:id/events` | Live session event stream (SSE) with sequence numbers and replay. Available for a run attached via [`--listen`](#listen), or once a session has raised at least one out-of-band event (e.g. a background job's elicitation, answered via `POST .../elicitation`) that created a session-scoped event log on demand. | | `DELETE` | `/api/sessions/:id` | Delete a session | | `PATCH` | `/api/sessions/:id/title` | Update session title | | `PATCH` | `/api/sessions/:id/permissions` | Update session permissions | @@ -255,7 +255,10 @@ session's runtime events — `stream_started`, `agent_choice`, `tool_call`, per-request stream returned by the agent-execution endpoint, it is session-scoped and survives across turns, so a client can watch a session for its whole lifetime. It is available for a run attached via -[`--listen`](#listen). +[`--listen`](#listen), and — since a session-scoped event log is created on +demand the first time a session raises an out-of-band event — for any +API-created session that has produced at least one (see the +[Sessions endpoint table](#sessions) above). Each event carries a monotonic **sequence number** in the SSE `id:` field, and the server buffers recent events. This makes the stream resumable: diff --git a/docs/guides/go-sdk/index.md b/docs/guides/go-sdk/index.md index 6fa025a031..11ae699894 100644 --- a/docs/guides/go-sdk/index.md +++ b/docs/guides/go-sdk/index.md @@ -129,6 +129,23 @@ if err := chat.Restart(); err != nil { For advanced use (custom elicitation, raw event inspection), call `chat.Runtime()` to access the underlying `runtime.Runtime` directly. +> [!WARNING] +> **Breaking change: `Runtime.ResumeElicitation` (#3584)** +> +> `Runtime.ResumeElicitation` gained an `elicitationID` parameter so responses can +> be correlated with a specific concurrent elicitation request (needed once +> multiple background jobs can be eliciting input at the same time). It is +> declared **variadic** (`elicitationID ...string`) specifically so existing +> *callers* of the 3-argument form keep compiling unchanged — `rt.ResumeElicitation(ctx, action, content)` +> still works and falls back to resolving the sole pending request. +> +> If you implement your own `runtime.Runtime` (rather than embedding +> `runtime.LocalRuntime`/`runtime.RemoteRuntime`), you do need to update your +> method's signature to match, and also add an `OnElicitationRequest(handler +> func(runtime.Event))` method (a no-op is fine if your runtime never raises +> elicitations) — both are required interface methods, matching the existing +> no-op-able pattern already used by `OnToolsChanged`/`OnBackgroundEvent`. + ## Optional Provider Build Tags By default docker-agent includes all four cloud providers (OpenAI, Anthropic, Google, Amazon Bedrock). When embedding docker-agent in your own binary you can compile out unneeded providers — together with their transitive SDK dependencies — to reduce binary size. diff --git a/pkg/api/types.go b/pkg/api/types.go index a4b2995474..a0e772031d 100644 --- a/pkg/api/types.go +++ b/pkg/api/types.go @@ -174,6 +174,13 @@ type DesktopTokenResponse struct { type ResumeElicitationRequest struct { Action string `json:"action"` // "accept", "decline", or "cancel" Content map[string]any `json:"content"` // The submitted form data (only present when action is "accept") + // ElicitationID correlates this response with a specific concurrent + // elicitation request (see the elicitation_id field on the + // ElicitationRequestEvent stream event). Optional and additive: when + // empty, the server falls back to resolving the sole pending request, + // for backward compatibility with clients that predate per-request + // correlation (#3584). + ElicitationID string `json:"elicitation_id,omitempty"` } // SteerSessionRequest represents a request to inject user messages into a diff --git a/pkg/app/app.go b/pkg/app/app.go index 4ff34bb41d..2061cdf2c5 100644 --- a/pkg/app/app.go +++ b/pkg/app/app.go @@ -178,6 +178,18 @@ 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. This is the runtime's single, + // exactly-once delivery point for elicitation requests (#3584); the + // swap-based bridge is a separate best-effort path for remote/SSE + // consumers only and never also reaches this sink, so no App-side + // dedupe is needed here. + a.runtime.OnElicitationRequest(func(event runtime.Event) { + a.sendEvent(ctx, event) + }) }) } @@ -933,9 +945,11 @@ 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 additive: pass "" to fall back to resolving the +// sole pending request (see runtime.Runtime.ResumeElicitation). +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() { diff --git a/pkg/app/app_test.go b/pkg/app/app_test.go index 1ce4a5331f..ebb32d3dfc 100644 --- a/pkg/app/app_test.go +++ b/pkg/app/app_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "path/filepath" + "sync" "testing" "time" @@ -58,7 +59,7 @@ func (m *mockRuntime) Run(ctx context.Context, sess *session.Session) ([]session return nil, nil } func (m *mockRuntime) Resume(ctx context.Context, req runtime.ResumeRequest) {} -func (m *mockRuntime) ResumeElicitation(ctx context.Context, action tools.ElicitationAction, content map[string]any) error { +func (m *mockRuntime) ResumeElicitation(ctx context.Context, action tools.ElicitationAction, content map[string]any, elicitationID ...string) error { return nil } func (m *mockRuntime) SessionStore() session.Store { return m.store } @@ -107,6 +108,7 @@ func (m *mockRuntime) AvailableModels(context.Context) []runtime.ModelChoice { r func (m *mockRuntime) SupportsModelSwitching() bool { return false } func (m *mockRuntime) OnToolsChanged(func(runtime.Event)) {} func (m *mockRuntime) OnBackgroundEvent(func(runtime.Event)) {} +func (m *mockRuntime) OnElicitationRequest(func(runtime.Event)) {} // Verify mockRuntime implements runtime.Runtime var _ runtime.Runtime = (*mockRuntime)(nil) @@ -217,6 +219,113 @@ func TestApp_Start_ForwardsBackgroundEvents(t *testing.T) { } } +// elicitationRequestMockRuntime captures the handler App.Start registers via +// OnElicitationRequest and records every ResumeElicitation call, so tests can +// drive the sink and assert on what App forwards back to the runtime. +type elicitationRequestMockRuntime struct { + mockRuntime + + handler func(runtime.Event) + + mu sync.Mutex + resumedIDs []string + resumedActions []tools.ElicitationAction +} + +// firstOrEmpty returns the first element of ids, or "" when empty. Mirrors +// runtime.firstElicitationID for tests that record a variadic call's ID. +func firstOrEmpty(ids []string) string { + if len(ids) == 0 { + return "" + } + return ids[0] +} + +func (m *elicitationRequestMockRuntime) OnElicitationRequest(handler func(runtime.Event)) { + m.handler = handler +} + +func (m *elicitationRequestMockRuntime) ResumeElicitation(_ context.Context, action tools.ElicitationAction, _ map[string]any, elicitationID ...string) error { + m.mu.Lock() + defer m.mu.Unlock() + m.resumedIDs = append(m.resumedIDs, firstOrEmpty(elicitationID)) + m.resumedActions = append(m.resumedActions, action) + return nil +} + +// TestApp_Start_ForwardsElicitationRequests verifies Start wires the +// runtime's OnElicitationRequest sink into the app's event stream, so +// background-job elicitations (which have no live channel of their own +// reaching the TUI) are surfaced (#3584). +func TestApp_Start_ForwardsElicitationRequests(t *testing.T) { + t.Parallel() + + rt := &elicitationRequestMockRuntime{} + events := make(chan tea.Msg, 16) + app := &App{ + runtime: rt, + session: session.New(), + events: events, + } + + app.Start(t.Context()) + require.NotNil(t, rt.handler, "Start must register the OnElicitationRequest handler") + + ev := runtime.ElicitationRequest("need input", "form", nil, "", "eid-1", "", "sess-1", nil, "worker") + rt.handler(ev) + + select { + case msg := <-events: + assert.Equal(t, ev, msg, "the elicitation request must reach the app's event stream unchanged") + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the forwarded elicitation request") + } +} + +// TestApp_SendEvent_DeliversElicitationWithoutDedupe pins the #3584 fix that +// removed the App-side ElicitationID dedupe: the runtime's +// OnElicitationRequest sink is now the single, exactly-once delivery point +// (elicitationHandler calls it directly, synchronously, and unconditionally, +// and runCollecting no longer re-forwards a bridge-observed copy), so +// sendEvent must forward every event unconditionally — including two +// distinct events that happen to share an ElicitationID, which a stateful +// dedupe would have incorrectly collapsed. +func TestApp_SendEvent_DeliversElicitationWithoutDedupe(t *testing.T) { + t.Parallel() + + events := make(chan tea.Msg, 16) + app := &App{events: events} + ctx := t.Context() + + ev := runtime.ElicitationRequest("need input", "form", nil, "", "eid-dup", "", "sess-1", nil, "worker") + app.sendEvent(ctx, ev) + require.Len(t, events, 1, "the sink's single delivery must reach the app's event stream") + <-events + + // A second event that happens to carry the same ElicitationID (e.g. a + // canceled request's ID reused later) must still go through: nothing in + // the App layer keys off ElicitationID any more. + app.sendEvent(ctx, ev) + require.Len(t, events, 1, "sendEvent must not drop a delivery based on ElicitationID") +} + +// TestApp_ResumeElicitation_ForwardsID verifies ResumeElicitation passes the +// elicitation ID through to the runtime unchanged. There is no dedupe state +// to clear any more (#3584): the runtime's per-request waiter registry is +// the sole source of truth for whether an ID is still answerable. +func TestApp_ResumeElicitation_ForwardsID(t *testing.T) { + t.Parallel() + + rt := &elicitationRequestMockRuntime{} + app := &App{runtime: rt, events: make(chan tea.Msg, 16)} + + require.NoError(t, app.ResumeElicitation(t.Context(), tools.ElicitationActionAccept, nil, "eid-clear")) + + require.Len(t, rt.resumedIDs, 1) + assert.Equal(t, "eid-clear", rt.resumedIDs[0]) + assert.Equal(t, tools.ElicitationActionAccept, rt.resumedActions[0]) +} + // stubSnapshotController is a tiny SnapshotController used by the app // tests to drive /undo without spinning up a real shadow-git // repository. enabled gates SnapshotsEnabled(), and the (files, ok, diff --git a/pkg/chatserver/agent.go b/pkg/chatserver/agent.go index f28eda8aa1..4673bccf3a 100644 --- a/pkg/chatserver/agent.go +++ b/pkg/chatserver/agent.go @@ -238,7 +238,7 @@ func runAgentLoop(ctx context.Context, rt runtime.Runtime, sess *session.Session case *runtime.ElicitationRequestEvent: // Required: the runtime blocks until we respond, regardless // of NonInteractive. Decline so the tool call fails fast. - _ = rt.ResumeElicitation(ctx, tools.ElicitationActionDecline, nil) + _ = rt.ResumeElicitation(ctx, tools.ElicitationActionDecline, nil, e.ElicitationID) case *runtime.MaxIterationsReachedEvent: // Defensive: in non-interactive mode the runtime already // stops on its own and this Resume is dropped. diff --git a/pkg/cli/runner.go b/pkg/cli/runner.go index eb842d807c..dcc0223a6c 100644 --- a/pkg/cli/runner.go +++ b/pkg/cli/runner.go @@ -124,7 +124,7 @@ func Run(ctx context.Context, out *Printer, cfg Config, rt runtime.Runtime, sess // still Reject (the hook said Ask, not Approve). rt.Resume(ctx, runtime.ResumeReject("")) case *runtime.ElicitationRequestEvent: - _ = rt.ResumeElicitation(ctx, "decline", nil) + _ = rt.ResumeElicitation(ctx, "decline", nil, e.ElicitationID) case *runtime.MaxIterationsReachedEvent: switch handleMaxIterationsAutoApprove(cfg.AutoApprove, &autoExtensions, e.MaxIterations) { case maxIterContinue: @@ -242,7 +242,7 @@ func Run(ctx context.Context, out *Printer, cfg Config, rt runtime.Runtime, sess serverURL, ok := e.Meta["docker-agent/server_url"].(string) if !ok || serverURL == "" { slog.WarnContext(ctx, "Skipping elicitation: missing or invalid server_url (non-interactive session?)") - _ = rt.ResumeElicitation(ctx, "decline", nil) + _ = rt.ResumeElicitation(ctx, "decline", nil, e.ElicitationID) return nil } @@ -254,9 +254,9 @@ func Run(ctx context.Context, out *Printer, cfg Config, rt runtime.Runtime, sess switch result { case ConfirmationApprove: - _ = rt.ResumeElicitation(ctx, "accept", nil) + _ = rt.ResumeElicitation(ctx, "accept", nil, e.ElicitationID) case ConfirmationReject: - _ = rt.ResumeElicitation(ctx, "decline", nil) + _ = rt.ResumeElicitation(ctx, "decline", nil, e.ElicitationID) return errors.New("OAuth authorization rejected by user") } } diff --git a/pkg/cli/runner_test.go b/pkg/cli/runner_test.go index 91b6bebe0e..a45f1d1371 100644 --- a/pkg/cli/runner_test.go +++ b/pkg/cli/runner_test.go @@ -69,7 +69,7 @@ func (m *mockRuntime) Run(context.Context, *session.Session) ([]session.Message, return nil, nil } -func (m *mockRuntime) ResumeElicitation(_ context.Context, action tools.ElicitationAction, _ map[string]any) error { +func (m *mockRuntime) ResumeElicitation(_ context.Context, action tools.ElicitationAction, _ map[string]any, _ ...string) error { m.mu.Lock() defer m.mu.Unlock() m.elicitationDeclines++ @@ -110,6 +110,7 @@ func (m *mockRuntime) AvailableModels(context.Context) []runtime.ModelChoice func (m *mockRuntime) SupportsModelSwitching() bool { return false } func (m *mockRuntime) OnToolsChanged(func(runtime.Event)) {} func (m *mockRuntime) OnBackgroundEvent(func(runtime.Event)) {} +func (m *mockRuntime) OnElicitationRequest(func(runtime.Event)) {} func (m *mockRuntime) RegenerateTitle(context.Context, *session.Session, chan runtime.Event) {} func (m *mockRuntime) Resume(_ context.Context, req runtime.ResumeRequest) { diff --git a/pkg/embeddedchat/embeddedchat.go b/pkg/embeddedchat/embeddedchat.go index 3f1d07b1f8..42476f1e9d 100644 --- a/pkg/embeddedchat/embeddedchat.go +++ b/pkg/embeddedchat/embeddedchat.go @@ -81,7 +81,7 @@ type ToolActivity struct { type runtimeRunner interface { RunStream(ctx context.Context, sess *session.Session) <-chan dagentruntime.Event Resume(ctx context.Context, req dagentruntime.ResumeRequest) - ResumeElicitation(ctx context.Context, action tools.ElicitationAction, content map[string]any) error + ResumeElicitation(ctx context.Context, action tools.ElicitationAction, content map[string]any, elicitationID ...string) error Close() error } @@ -303,7 +303,7 @@ func (s *Session) forwardEvents(ctx context.Context, events <-chan dagentruntime // This headless wrapper has no built-in elicitation UI. Decline so the // run cannot hang forever; embedders that need elicitation can consume // RuntimeEvent directly by driving the runtime themselves. - _ = s.rt.ResumeElicitation(ctx, tools.ElicitationActionDecline, nil) + _ = s.rt.ResumeElicitation(ctx, tools.ElicitationActionDecline, nil, e.ElicitationID) case *dagentruntime.MaxIterationsReachedEvent: s.rt.Resume(ctx, dagentruntime.ResumeReject("")) case *dagentruntime.ErrorEvent: diff --git a/pkg/embeddedchat/embeddedchat_test.go b/pkg/embeddedchat/embeddedchat_test.go index 95a145e892..b4d2d8fc9b 100644 --- a/pkg/embeddedchat/embeddedchat_test.go +++ b/pkg/embeddedchat/embeddedchat_test.go @@ -62,7 +62,7 @@ func (f *fakeRuntime) Resume(_ context.Context, req dagentruntime.ResumeRequest) f.resumes = append(f.resumes, req) } -func (f *fakeRuntime) ResumeElicitation(_ context.Context, action tools.ElicitationAction, _ map[string]any) error { +func (f *fakeRuntime) ResumeElicitation(_ context.Context, action tools.ElicitationAction, _ map[string]any, _ ...string) error { f.elicitations = append(f.elicitations, action) return nil } @@ -166,7 +166,7 @@ func TestSessionSendDeclinesElicitationAndRejectsMaxIterations(t *testing.T) { out, err := s.Send(t.Context(), "hi") require.NoError(t, err) - rt.events <- dagentruntime.ElicitationRequest("authorize", "url", nil, "https://example.com", "id", nil, "agent") + rt.events <- dagentruntime.ElicitationRequest("authorize", "url", nil, "https://example.com", "id", "", "sess", nil, "agent") rt.events <- dagentruntime.MaxIterationsReached(3) close(rt.events) diff --git a/pkg/leantui/update_test.go b/pkg/leantui/update_test.go index a2c06d52a0..9a589c21c8 100644 --- a/pkg/leantui/update_test.go +++ b/pkg/leantui/update_test.go @@ -56,7 +56,7 @@ func (r *cycleThinkingRuntime) Run(context.Context, *session.Session) ([]session return nil, nil } func (r *cycleThinkingRuntime) Resume(context.Context, runtime.ResumeRequest) {} -func (r *cycleThinkingRuntime) ResumeElicitation(context.Context, tools.ElicitationAction, map[string]any) error { +func (r *cycleThinkingRuntime) ResumeElicitation(context.Context, tools.ElicitationAction, map[string]any, ...string) error { return nil } func (r *cycleThinkingRuntime) SessionStore() session.Store { return nil } @@ -120,6 +120,7 @@ func (r *cycleThinkingRuntime) AvailableModels(context.Context) []runtime.ModelC func (r *cycleThinkingRuntime) SupportsModelSwitching() bool { return r.supports } func (r *cycleThinkingRuntime) OnToolsChanged(func(runtime.Event)) {} func (r *cycleThinkingRuntime) OnBackgroundEvent(func(runtime.Event)) {} +func (r *cycleThinkingRuntime) OnElicitationRequest(func(runtime.Event)) {} var _ runtime.Runtime = (*cycleThinkingRuntime)(nil) diff --git a/pkg/runtime/agent_delegation.go b/pkg/runtime/agent_delegation.go index 01df570672..f1af4c6f38 100644 --- a/pkg/runtime/agent_delegation.go +++ b/pkg/runtime/agent_delegation.go @@ -378,6 +378,15 @@ func (r *LocalRuntime) runCollecting(ctx context.Context, parent *session.Sessio if usage, ok := event.(*TokenUsageEvent); ok { r.emitBackgroundEvent(usage) } + // Elicitation requests are NOT re-forwarded here: elicitationHandler + // already delivered this event to the OnElicitationRequest sink + // directly, synchronously, and exactly once (#3584). Forwarding it + // again here — as the bridge's best-effort copy happens to flow + // through this same channel when this background sub-session + // currently owns the bridge slot — used to cause a second sink + // delivery for the same request, which only a stateful App-side + // dedupe (since removed) papered over. This branch is intentionally + // absent; ElicitationRequestEvents seen here are simply ignored. if errEvt, ok := event.(*ErrorEvent); ok { errMsg = errEvt.Error break @@ -416,15 +425,31 @@ func (r *LocalRuntime) runCollecting(ctx context.Context, parent *session.Sessio // no explanation. Prepend an actionable note so the model (and, through it, // the user) learns the server must be authorized interactively first. if note := backgroundAuthRequiredNote(child); note != "" { - if result != "" { - result = note + "\n\n" + result - } else { - result = note - } + result = prependNote(result, note) + } + // Mid-call elicitations that were auto-declined because this background + // session had no UI to answer them (see elicitationHandler) are recorded + // against this sub-session's ID; surface them the same way (#3584). + for _, note := range r.elicitationDeclines.drain(s.ID) { + result = prependNote(result, note) } return &agenttool.RunResult{Result: result} } +// prependNote prepends note to result, separated by a blank line, handling +// the case where either side is empty. Used to surface model-readable +// context (OAuth-required, elicitation auto-declined) ahead of a background +// sub-session's actual response. +func prependNote(result, note string) string { + if note == "" { + return result + } + if result == "" { + return note + } + return note + "\n\n" + result +} + // backgroundAuthRequiredNote returns a model-readable note naming the child // agent's MCP toolsets that could not start because they require first-time // interactive OAuth authorization, which a background agent cannot complete diff --git a/pkg/runtime/client.go b/pkg/runtime/client.go index 1aa8747f0f..fff3a7835e 100644 --- a/pkg/runtime/client.go +++ b/pkg/runtime/client.go @@ -458,8 +458,8 @@ func (c *Client) DeleteRemoteSession(ctx context.Context, sessionID string) erro return c.doRequest(ctx, http.MethodDelete, "/api/sessions/"+sessionID, nil, nil) } -func (c *Client) ResumeElicitation(ctx context.Context, sessionID string, action tools.ElicitationAction, content map[string]any) error { - req := api.ResumeElicitationRequest{Action: string(action), Content: content} +func (c *Client) ResumeElicitation(ctx context.Context, sessionID string, action tools.ElicitationAction, content map[string]any, elicitationID ...string) error { + req := api.ResumeElicitationRequest{Action: string(action), Content: content, ElicitationID: firstElicitationID(elicitationID)} return c.doRequest(ctx, http.MethodPost, "/api/sessions/"+sessionID+"/elicitation", req, nil) } diff --git a/pkg/runtime/commands_test.go b/pkg/runtime/commands_test.go index 8e44007914..a32aa96c55 100644 --- a/pkg/runtime/commands_test.go +++ b/pkg/runtime/commands_test.go @@ -49,7 +49,7 @@ func (m *mockRuntime) Run(context.Context, *session.Session) ([]session.Message, return nil, nil } func (m *mockRuntime) Resume(context.Context, ResumeRequest) {} -func (m *mockRuntime) ResumeElicitation(context.Context, tools.ElicitationAction, map[string]any) error { +func (m *mockRuntime) ResumeElicitation(context.Context, tools.ElicitationAction, map[string]any, ...string) error { return nil } func (m *mockRuntime) SessionStore() session.Store { return nil } @@ -93,6 +93,7 @@ func (m *mockRuntime) AvailableModels(context.Context) []ModelChoice { return ni func (m *mockRuntime) SupportsModelSwitching() bool { return false } func (m *mockRuntime) OnToolsChanged(func(Event)) {} func (m *mockRuntime) OnBackgroundEvent(func(Event)) {} +func (m *mockRuntime) OnElicitationRequest(func(Event)) {} func (m *mockRuntime) RegenerateTitle(context.Context, *session.Session, chan Event) { } diff --git a/pkg/runtime/contract_test.go b/pkg/runtime/contract_test.go index d3891b6464..47519fff53 100644 --- a/pkg/runtime/contract_test.go +++ b/pkg/runtime/contract_test.go @@ -124,7 +124,7 @@ func runRuntimeContract(t *testing.T, newRT func(t *testing.T) Runtime) { t.Cleanup(func() { _ = rt.Close() }) ctx, cancel := context.WithCancel(t.Context()) cancel() - _ = rt.ResumeElicitation(ctx, tools.ElicitationActionDecline, nil) + _ = rt.ResumeElicitation(ctx, tools.ElicitationActionDecline, nil, "") }) } diff --git a/pkg/runtime/elicitation.go b/pkg/runtime/elicitation.go index e868b4b80a..fdc3468635 100644 --- a/pkg/runtime/elicitation.go +++ b/pkg/runtime/elicitation.go @@ -6,10 +6,14 @@ import ( "fmt" "log/slog" "sync" + "sync/atomic" + "github.com/google/uuid" "github.com/modelcontextprotocol/go-sdk/mcp" + "github.com/docker/docker-agent/pkg/telemetry/genai" "github.com/docker/docker-agent/pkg/tools" + mcptools "github.com/docker/docker-agent/pkg/tools/mcp" ) // ElicitationResult represents the result of an elicitation request. @@ -40,6 +44,12 @@ type ElicitationRequestHandler func(ctx context.Context, message string, schema // configured (no RunStream is active). var errNoElicitationChannel = errors.New("no events channel available for elicitation") +// errNoSuchElicitation is returned by ResumeElicitation when the given +// elicitation ID (or, in the empty-ID fallback, "the single pending +// request") no longer has a registered waiter — already answered, timed +// out, or never existed. +var errNoSuchElicitation = errors.New("no elicitation request in progress") + // elicitationBridge owns the events channel that the runtime's MCP // elicitation handler sends requests to. Each RunStream call swaps in its // own channel on entry and the previous one back on exit, so nested @@ -54,6 +64,17 @@ var errNoElicitationChannel = errors.New("no events channel available for elicit // contract testable in isolation (with the race detector) without // spinning up a runtime, and keeps LocalRuntime free of the two raw // fields it used to expose. +// +// Concurrent (non-nested) RunStreams — most notably background jobs +// started via run_background_agent — can swap this single slot out from +// under each other; see elicitationWaiters and OnElicitationRequest for +// the routing/delivery fix (#3584). The bridge itself is kept only as a +// best-effort secondary delivery path for remote/SSE consumers that read +// events directly off a RunStream channel (see remote_runtime.go). It is +// never allowed to hold up the reliable sink or response processing: send +// is bounded by the caller's ctx (see elicitationHandler), and callers +// invoke it from a detached goroutine so a wedged or abandoned channel +// cannot block the request/response path at all (#3584 review item 1). type elicitationBridge struct { mu sync.RWMutex ch chan Event @@ -70,13 +91,18 @@ func (b *elicitationBridge) swap(ch chan Event) chan Event { } // send delivers ev to the current channel, holding the read lock across -// the send. This blocks concurrent teardown until the send completes, -// preserving the invariant that the channel reference held by an -// in-flight sender stays open until that sender finishes. +// the send so a concurrent restoreAndClose cannot close the channel out +// from under an in-flight send without going through recover() below. The +// send itself is bounded by ctx: if ctx is done before the channel accepts +// the event, send returns ctx.Err() instead of blocking forever. Combined +// with callers invoking send from a detached goroutine (see +// elicitationHandler), a full or abandoned channel can no longer delay — +// let alone indefinitely block — the reliable sink delivery or response +// handling that used to be sequenced before this call (#3584 item 1). // -// Returns errNoElicitationChannel when no channel is configured or when -// a defensive recover catches an externally closed channel. -func (b *elicitationBridge) send(ev Event) (err error) { +// Returns errNoElicitationChannel when no channel is configured or when a +// defensive recover catches an externally closed channel. +func (b *elicitationBridge) send(ctx context.Context, ev Event) (err error) { b.mu.RLock() defer b.mu.RUnlock() defer func() { @@ -87,8 +113,12 @@ func (b *elicitationBridge) send(ev Event) (err error) { if b.ch == nil { return errNoElicitationChannel } - b.ch <- ev - return nil + select { + case b.ch <- ev: + return nil + case <-ctx.Done(): + return ctx.Err() + } } // restoreAndClose restores the previous stream channel and closes the current @@ -100,9 +130,11 @@ func (b *elicitationBridge) send(ev Event) (err error) { // lock makes restoreAndClose wait for any in-flight send to finish, because // send holds the read lock across "b.ch <- ev". If the stream consumer has // gone away and current is full (or unbuffered), that parked send never -// drains, so this call blocks on Lock forever and the teardown goroutine -// leaks. A leaked goroutine is the deliberate, accepted alternative to -// crashing the whole process with a send-on-closed-channel panic. +// drains until its own ctx is done, so this call blocks on Lock until then. A +// bounded wait is the deliberate, accepted alternative to crashing the whole +// process with a send-on-closed-channel panic; #3584 bounded the wait (send +// used to have no ctx at all and could block indefinitely) and moved the +// caller onto a detached goroutine so this can never stall the request path. func (b *elicitationBridge) restoreAndClose(current, previous chan Event) { b.mu.Lock() defer b.mu.Unlock() @@ -110,34 +142,306 @@ func (b *elicitationBridge) restoreAndClose(current, previous chan Event) { close(current) } +// waiterState is the terminal-state machine for a single elicitationWaiter. +// Exactly one of resolve/cancel ever wins the transition out of pending, +// closing the #3584 cancellation-vs-response race: a resolve that already +// flipped the state keeps its value in the channel for the ctx.Done() branch +// to drain, instead of the handler discarding a response ResumeElicitation +// already reported as delivered. +type waiterState int32 + +const ( + waiterPending waiterState = iota + waiterResolved + waiterCanceled +) + +// elicitationWaiter is one pending elicitation request's response slot. +type elicitationWaiter struct { + ch chan ElicitationResult + state atomic.Int32 +} + +func newElicitationWaiter() *elicitationWaiter { + return &elicitationWaiter{ch: make(chan ElicitationResult, 1)} +} + +// tryResolve attempts to deliver result, winning the terminal-state race +// only if the waiter is still pending. Returns false without sending when +// the waiter was already resolved or cancelled. +func (w *elicitationWaiter) tryResolve(result ElicitationResult) bool { + if !w.state.CompareAndSwap(int32(waiterPending), int32(waiterResolved)) { + return false + } + w.ch <- result + return true +} + +// tryCancel attempts to mark the waiter cancelled, winning the terminal-state +// race only if it is still pending. Returns false when resolve already won — +// the caller must then receive from ch instead of treating this as a +// cancellation, since a value is already there (or is about to land). +func (w *elicitationWaiter) tryCancel() bool { + return w.state.CompareAndSwap(int32(waiterPending), int32(waiterCanceled)) +} + +// elicitationWaiters routes an elicitation response to the specific request +// that is waiting for it, keyed by a correlation ID that is unique per +// request (see elicitationHandler). This replaces the single shared +// elicitationRequestCh, which could only ever have one request in flight: +// with concurrent (background-job) elicitations, a response arriving on +// that shared channel could be delivered to an arbitrary waiter, and +// ResumeElicitation had no way to tell "no request in flight" from "the +// request hasn't parked on the channel yet" (a TOCTOU race). +// +// Each waiter is registered BEFORE the corresponding request event is +// emitted, so a response that arrives immediately after — even before the +// handler reaches its receive — is never lost. The registry key is always an +// internally-generated ID (never the MCP wire ElicitationID, which two +// different MCP servers can coincidentally reuse): see elicitationHandler. +type elicitationWaiters struct { + mu sync.Mutex + pending map[string]*elicitationWaiter +} + +// register creates a waiter for id and stores it. The channel is buffered +// (capacity 1), so resolve never blocks even if the registrant hasn't +// reached its receive yet. +func (w *elicitationWaiters) register(id string) *elicitationWaiter { + wt := newElicitationWaiter() + w.mu.Lock() + defer w.mu.Unlock() + if w.pending == nil { + w.pending = make(map[string]*elicitationWaiter) + } + w.pending[id] = wt + return wt +} + +// abandon removes id's waiter from the registry, if it is still the one +// registered (defends against a hypothetical ID reuse racing a fresh +// register call), without touching its terminal state. Called once a waiter +// is done being awaited via any path, so a later resolve() for a reused ID +// cannot be confused with this one. +func (w *elicitationWaiters) abandon(id string, wt *elicitationWaiter) { + w.mu.Lock() + defer w.mu.Unlock() + if w.pending[id] == wt { + delete(w.pending, id) + } +} + +// cancel marks wt cancelled if it is still pending and removes it from the +// registry. Returns true when this call won the cancel-vs-resolve race — the +// caller (elicitationHandler's ctx.Done() branch) should then return ctx.Err(). +// Returns false when resolve() already won: the caller must receive from +// wt.ch instead, since a result is already there (or is about to land — the +// buffered send in tryResolve never blocks). +func (w *elicitationWaiters) cancel(id string, wt *elicitationWaiter) bool { + won := wt.tryCancel() + if won { + w.abandon(id, wt) + } + return won +} + +// resolve delivers result to the waiter registered for id and returns true. +// Returns false without side effects when no waiter is currently registered +// for that ID, or when it was already resolved/cancelled — already +// answered, timed out, or unknown. +func (w *elicitationWaiters) resolve(id string, result ElicitationResult) bool { + w.mu.Lock() + wt, ok := w.pending[id] + if ok { + delete(w.pending, id) + } + w.mu.Unlock() + if !ok { + return false + } + return wt.tryResolve(result) +} + +// resolveSingle delivers result to the sole pending waiter. It exists for +// backward compatibility with clients that don't send an elicitation_id +// (the pre-#3584 API contract only ever supported one request in flight). +// It is a deliberate no-op — returning false — when zero or more than one +// request is pending, since there is then no way to disambiguate which one +// the caller meant. +func (w *elicitationWaiters) resolveSingle(result ElicitationResult) bool { + w.mu.Lock() + if len(w.pending) != 1 { + w.mu.Unlock() + return false + } + var id string + var wt *elicitationWaiter + for k, v := range w.pending { + id, wt = k, v + } + delete(w.pending, id) + w.mu.Unlock() + return wt.tryResolve(result) +} + +// count reports the number of elicitations currently awaiting a response. +func (w *elicitationWaiters) count() int { + w.mu.Lock() + defer w.mu.Unlock() + return len(w.pending) +} + +// firstElicitationID extracts the (at most one meaningful) elicitation ID +// from a variadic parameter. The parameter is variadic — rather than a +// required positional string — purely so pre-#3584 callers of +// Runtime.ResumeElicitation (3 args) keep compiling unchanged; see +// docs/guides/go-sdk/index.md for the Go API compatibility note (this +// project's CHANGELOG.md is generated per release from merged PRs, not +// edited alongside the change, so it carries no such note pre-release). +func firstElicitationID(elicitationID []string) string { + if len(elicitationID) == 0 { + return "" + } + return elicitationID[0] +} + // ResumeElicitation sends an elicitation response back to a waiting -// elicitation request. Returns an error if no elicitation is in progress -// or if the context is cancelled before the response can be delivered. -func (r *LocalRuntime) ResumeElicitation(ctx context.Context, action tools.ElicitationAction, content map[string]any) error { - slog.DebugContext(ctx, "Resuming runtime with elicitation response", "agent", r.currentAgentName(), "action", action) +// elicitation request. When elicitationID is non-empty it is routed to that +// specific request; when empty, it falls back to resolving the sole pending +// request for backward compatibility with callers that predate per-request +// correlation. Returns an error if no matching elicitation is in progress. +// +// Unlike the old shared-channel implementation, this never blocks on ctx: +// each waiter channel is buffered (capacity 1) and registered before its +// request event is emitted, so resolving it is always a non-blocking map +// lookup plus a buffered send — there is no TOCTOU window to race. A +// resolve that raced a ctx-cancellation on the handler side is decided by +// an atomic terminal state (see elicitationWaiter), so a true here always +// means the response will reach the caller, never a discarded one. +func (r *LocalRuntime) ResumeElicitation(ctx context.Context, action tools.ElicitationAction, content map[string]any, elicitationID ...string) error { + id := firstElicitationID(elicitationID) + slog.DebugContext(ctx, "Resuming runtime with elicitation response", "agent", r.currentAgentName(), "action", action, "elicitation_id", id) result := ElicitationResult{ Action: action, Content: content, } - select { - case <-ctx.Done(): - slog.DebugContext(ctx, "Context cancelled while sending elicitation response") - return ctx.Err() - case r.elicitationRequestCh <- result: - slog.DebugContext(ctx, "Elicitation response sent successfully", "action", action) - return nil - default: - slog.DebugContext(ctx, "Elicitation channel not ready") - return errors.New("no elicitation request in progress") + var ok bool + if id != "" { + ok = r.elicitationWaiters.resolve(id, result) + } else { + ok = r.elicitationWaiters.resolveSingle(result) + } + if !ok { + slog.DebugContext(ctx, "No matching elicitation request in progress", "elicitation_id", id) + return errNoSuchElicitation + } + slog.DebugContext(ctx, "Elicitation response sent successfully", "action", action) + return nil +} + +// OnElicitationRequest registers a handler invoked whenever an MCP toolset +// raises an elicitation request. This is the reliable route for +// background-job elicitations (run_background_agent): their RunStream runs +// on a detached goroutine and can race concurrent streams for the bridge's +// single channel slot (#3584), so elicitationHandler calls this sink +// directly, synchronously, and unconditionally — before it ever touches the +// best-effort bridge — as the single, exactly-once delivery point. Embedders +// (e.g. the TUI's App, or the API server for session-scoped SSE delivery) +// register a handler here that forwards the event to their UI/transport. +func (r *LocalRuntime) OnElicitationRequest(handler func(Event)) { + r.elicitationSinkMu.Lock() + defer r.elicitationSinkMu.Unlock() + r.onElicitationRequest = handler +} + +// emitElicitationRequest forwards an elicitation request event to the +// registered sink, if any. Besides [LocalRuntime.EmitElicitationRequestForTesting], +// this is the ONLY call site that invokes the sink (see elicitationHandler): +// production callers must not add a second delivery path (e.g. re-forwarding +// an event observed on a RunStream channel), or the exactly-once guarantee +// this type documents no longer holds (#3584 item 5 — dual delivery +// previously required a stateful App-side dedupe to paper over). +func (r *LocalRuntime) emitElicitationRequest(event Event) { + r.elicitationSinkMu.RLock() + handler := r.onElicitationRequest + r.elicitationSinkMu.RUnlock() + if handler != nil { + handler(event) + } +} + +// EmitElicitationRequestForTesting invokes whatever OnElicitationRequest sink +// is currently registered, exactly as elicitationHandler would, but without +// the real MCP elicitation handshake. elicitationHandler is unexported, so +// callers outside this package (e.g. pkg/server) cannot drive it directly to +// prove a *specific* runtime instance has the expected sink wired; this +// gives them a seam to do that instead of reconstructing the sink separately +// and invoking it in isolation, which would pass even if the runtime under +// test was never actually wired up (#3584 re-review should-fix 1). +func (r *LocalRuntime) EmitElicitationRequestForTesting(event Event) { + r.emitElicitationRequest(event) +} + +// hasElicitationSink reports whether an embedder has registered an +// OnElicitationRequest handler. Used to decide whether a background +// session's elicitation has any chance of reaching a user (see +// elicitationHandler's headless fast-decline path). +func (r *LocalRuntime) hasElicitationSink() bool { + r.elicitationSinkMu.RLock() + defer r.elicitationSinkMu.RUnlock() + return r.onElicitationRequest != nil +} + +// elicitationDeclineNotes accumulates model-readable notes for elicitations +// that were auto-declined because a background session had no UI available +// to answer them (see elicitationHandler). runCollecting drains these after +// the sub-session completes and prepends them to the tool result, mirroring +// backgroundAuthRequiredNote's #3200 pattern for OAuth-at-Start failures. +type elicitationDeclineNotes struct { + mu sync.Mutex + bySession map[string][]string +} + +// record appends note under sessionID. No-op when either is empty. +func (n *elicitationDeclineNotes) record(sessionID, note string) { + if sessionID == "" || note == "" { + return } + n.mu.Lock() + defer n.mu.Unlock() + if n.bySession == nil { + n.bySession = make(map[string][]string) + } + n.bySession[sessionID] = append(n.bySession[sessionID], note) +} + +// drain returns and clears the notes recorded for sessionID. +func (n *elicitationDeclineNotes) drain(sessionID string) []string { + n.mu.Lock() + defer n.mu.Unlock() + notes := n.bySession[sessionID] + delete(n.bySession, sessionID) + return notes +} + +// backgroundElicitationDeclinedNote returns a model-readable explanation for +// an elicitation that was auto-declined because it originated from a +// background (non-interactive) session with no UI available to answer it. +func backgroundElicitationDeclinedNote(message string) string { + return fmt.Sprintf( + "Note: a tool requested user input (%q) while running as a background task. "+ + "Background tasks have no interactive UI to answer such requests, so it was "+ + "automatically declined. If the tool truly needs user input, ask the user to "+ + "run this task in the foreground instead.", + message, + ) } // elicitationHandler is the MCP-toolset-side hook that turns an inbound -// elicitation request from a server into an ElicitationRequest event on the -// active stream's events channel and waits for the embedder's response on -// elicitationRequestCh. +// elicitation request from a server into an ElicitationRequest event and +// waits for the embedder's response, correlated by elicitation ID. func (r *LocalRuntime) elicitationHandler(ctx context.Context, req *mcp.ElicitParams) (tools.ElicitationResult, error) { slog.DebugContext(ctx, "Elicitation request received from MCP server", "message", req.Message) @@ -150,30 +454,98 @@ func (r *LocalRuntime) elicitationHandler(ctx context.Context, req *mcp.ElicitPa }, nil } + // A background session (run_background_agent) marks its context so + // toolset Start() OAuth fails fast instead of eliciting (#3200). Mid-call + // elicitations reach here regardless of that marker, so extend the same + // fast-fail idea: if this call is running in such a context AND no + // embedder has registered a sink to surface it (headless use — e.g. the + // --exec CLI path, which never registers OnElicitationRequest), nobody at + // all can answer this request. Decline immediately with a model-readable + // note instead of parking a goroutine forever (#3584). + if !mcptools.InteractivePromptsAllowed(ctx) && !r.hasElicitationSink() { + slog.WarnContext(ctx, "Declining elicitation: background session has no UI to answer it", "message", req.Message) + r.elicitationDeclines.record(genai.ConversationIDFromContext(ctx), backgroundElicitationDeclinedNote(req.Message)) + return tools.ElicitationResult{ + Action: tools.ElicitationActionDecline, + }, nil + } + r.executeOnUserInputHooks(ctx, "", "elicitation") + // The registry key (and the ElicitationID surfaced to clients for + // ResumeElicitation routing) is always a freshly generated, internal + // ID — never the MCP wire req.ElicitationID. The wire value is only + // ever set for URL-mode elicitations and is chosen by the originating + // MCP server; two independent servers (e.g. two background jobs each + // talking to their own MCP process) can legitimately reuse the same + // value. Trusting it as the registry key would let the second + // request's register() silently evict the first request's waiter, + // orphaning it (#3584 review item 2a). The wire ID is preserved + // separately on the event (ServerElicitationID) for callers that want + // to correlate with server-side logs; it is never used for routing. + correlationID := uuid.NewString() + + // Register the waiter BEFORE emitting the request event. This is the + // #3584 TOCTOU fix: previously a response that arrived before the + // handler reached its receive on the shared channel was lost because + // there was nothing to receive it into yet. + wt := r.elicitationWaiters.register(correlationID) + defer r.elicitationWaiters.abandon(correlationID, wt) + slog.DebugContext(ctx, "Sending elicitation request event to client", "message", req.Message, "mode", req.Mode, "requested_schema", req.RequestedSchema, - "url", req.URL) + "url", req.URL, + "elicitation_id", correlationID, + "server_elicitation_id", req.ElicitationID) slog.DebugContext(ctx, "Elicitation request meta", "meta", req.Meta) - if err := r.elicitation.send( - ElicitationRequest(req.Message, req.Mode, req.RequestedSchema, req.URL, req.ElicitationID, req.Meta, r.currentAgentName()), - ); err != nil { - return tools.ElicitationResult{}, err - } + sessionID := genai.ConversationIDFromContext(ctx) + ev := ElicitationRequest(req.Message, req.Mode, req.RequestedSchema, req.URL, correlationID, req.ElicitationID, sessionID, req.Meta, r.currentAgentName()) + + // Reliable delivery: invoked synchronously, unconditionally, and exactly + // once, BEFORE anything that could block (#3584 review item 1). This + // must never be gated behind the best-effort bridge below. + r.emitElicitationRequest(ev) + + // Best-effort secondary delivery on the owning stream's events channel, + // kept for remote/SSE consumers that read directly off RunStream + // (remote_runtime.go depends on it). Dispatched on a detached goroutine, + // bounded by ctx, so a wedged or abandoned bridge channel (concurrent + // RunStreams racing the swap-based single slot, or a dead consumer) can + // never delay — let alone block — sink delivery or the response wait + // below (#3584 review item 1). runCollecting no longer treats a bridge + // delivery as a second source of truth (#3584 review item 5): this send + // exists solely for out-of-process consumers. + go func() { + if err := r.elicitation.send(ctx, ev); err != nil { + slog.DebugContext(ctx, "Elicitation bridge send failed or abandoned; relying on the registered sink", "error", err) + } + }() - // Wait for response from the client. + // Wait for the response addressed to this specific request. The + // ctx.Done() branch cannot simply return ctx.Err(): resolve() may have + // already won the terminal-state race an instant earlier and be about + // to (or have already) delivered into wt.ch, in which case + // ResumeElicitation already reported success to its caller and this + // handler must not silently discard that response (#3584 review item + // 2b). cancel() decides the winner atomically. select { - case result := <-r.elicitationRequestCh: + case result := <-wt.ch: return tools.ElicitationResult{ Action: result.Action, Content: result.Content, }, nil case <-ctx.Done(): slog.DebugContext(ctx, "Context cancelled while waiting for elicitation response") - return tools.ElicitationResult{}, ctx.Err() + if r.elicitationWaiters.cancel(correlationID, wt) { + return tools.ElicitationResult{}, ctx.Err() + } + result := <-wt.ch + return tools.ElicitationResult{ + Action: result.Action, + Content: result.Content, + }, nil } } diff --git a/pkg/runtime/elicitation_concurrency_test.go b/pkg/runtime/elicitation_concurrency_test.go new file mode 100644 index 0000000000..85c8292021 --- /dev/null +++ b/pkg/runtime/elicitation_concurrency_test.go @@ -0,0 +1,641 @@ +package runtime + +import ( + "context" + "fmt" + "maps" + "sync" + "testing" + "time" + + "github.com/modelcontextprotocol/go-sdk/mcp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/docker/docker-agent/pkg/agent" + "github.com/docker/docker-agent/pkg/chat" + "github.com/docker/docker-agent/pkg/session" + "github.com/docker/docker-agent/pkg/team" + "github.com/docker/docker-agent/pkg/telemetry/genai" + "github.com/docker/docker-agent/pkg/tools" + agenttool "github.com/docker/docker-agent/pkg/tools/builtin/agent" + mcptools "github.com/docker/docker-agent/pkg/tools/mcp" +) + +// --- elicitationWaiters: unit + concurrency regression tests (#3584) --- + +func TestElicitationWaiters_ResolveRoutesToCorrectID(t *testing.T) { + t.Parallel() + + var w elicitationWaiters + wtA := w.register("a") + wtB := w.register("b") + + require.True(t, w.resolve("b", ElicitationResult{Action: tools.ElicitationActionDecline})) + require.True(t, w.resolve("a", ElicitationResult{Action: tools.ElicitationActionAccept})) + + select { + case result := <-wtA.ch: + assert.Equal(t, tools.ElicitationActionAccept, result.Action, "waiter a must receive a's response, not b's") + default: + t.Fatal("waiter a never received its response") + } + select { + case result := <-wtB.ch: + assert.Equal(t, tools.ElicitationActionDecline, result.Action, "waiter b must receive b's response, not a's") + default: + t.Fatal("waiter b never received its response") + } +} + +// TestElicitationWaiters_ResolveBeforeReceiveIsNotLost pins the TOCTOU fix: +// registering the waiter before the request event is emitted means a +// response that arrives before anyone has read from the channel is still +// captured (the channel is buffered), instead of the old shared unbuffered +// elicitationRequestCh's `default:` branch reporting "no elicitation request +// in progress". See TestElicitationHandler_TOCTOU_ResolveImmediatelyAfterRegister +// below for the same pin exercised through the real handler path. +func TestElicitationWaiters_ResolveBeforeReceiveIsNotLost(t *testing.T) { + t.Parallel() + + var w elicitationWaiters + wt := w.register("id-1") + + ok := w.resolve("id-1", ElicitationResult{Action: tools.ElicitationActionAccept, Content: map[string]any{"k": "v"}}) + require.True(t, ok, "resolve must succeed even though nothing has received from the channel yet") + + select { + case result := <-wt.ch: + assert.Equal(t, tools.ElicitationActionAccept, result.Action) + assert.Equal(t, map[string]any{"k": "v"}, result.Content) + default: + t.Fatal("the buffered waiter channel should already hold the resolved result") + } +} + +func TestElicitationWaiters_ResolveUnknownIDReturnsFalse(t *testing.T) { + t.Parallel() + + var w elicitationWaiters + assert.False(t, w.resolve("missing", ElicitationResult{})) +} + +func TestElicitationWaiters_ResolveSingle_FallsBackForEmptyID(t *testing.T) { + t.Parallel() + + var w elicitationWaiters + wt := w.register("only-one") + + require.True(t, w.resolveSingle(ElicitationResult{Action: tools.ElicitationActionAccept})) + select { + case result := <-wt.ch: + assert.Equal(t, tools.ElicitationActionAccept, result.Action) + default: + t.Fatal("resolveSingle should have delivered to the sole pending waiter") + } +} + +// TestElicitationWaiters_ResolveSingle_AmbiguousWithMultiplePending verifies +// resolveSingle refuses to guess when more than one request is in flight — +// the whole point of per-ID correlation is that an empty-ID caller cannot +// safely disambiguate concurrent requests. +func TestElicitationWaiters_ResolveSingle_AmbiguousWithMultiplePending(t *testing.T) { + t.Parallel() + + var w elicitationWaiters + w.register("a") + w.register("b") + + assert.False(t, w.resolveSingle(ElicitationResult{Action: tools.ElicitationActionAccept})) + assert.Equal(t, 2, w.count(), "an ambiguous resolveSingle must not consume either waiter") +} + +func TestElicitationWaiters_Abandon(t *testing.T) { + t.Parallel() + + var w elicitationWaiters + wt := w.register("a") + require.Equal(t, 1, w.count()) + + w.abandon("a", wt) + assert.Equal(t, 0, w.count()) + assert.False(t, w.resolve("a", ElicitationResult{}), "abandoned waiter must not be resolvable") +} + +// TestElicitationWaiters_DuplicateWireIDsDoNotCollide pins #3584 review item +// 2a: registry keys are always internally-generated IDs, never the MCP wire +// ElicitationID, precisely because two independent MCP servers (e.g. two +// concurrent background jobs, each talking to their own server) can +// legitimately reuse the same wire ID. Registering two waiters under +// different (internal) IDs — even when both requests logically share one +// wire ID — must never let the second registration evict the first's +// channel. +func TestElicitationWaiters_DuplicateWireIDsDoNotCollide(t *testing.T) { + t.Parallel() + + var w elicitationWaiters + // Simulates two servers both using wire ID "1": elicitationHandler + // always mints a fresh internal correlation ID regardless, so the two + // registrations land under distinct keys. + wtServerA := w.register("internal-a") + wtServerB := w.register("internal-b") + + require.Equal(t, 2, w.count(), "distinct internal IDs must not collide even if the wire IDs would have") + + require.True(t, w.resolve("internal-a", ElicitationResult{Action: tools.ElicitationActionAccept, Content: map[string]any{"who": "a"}})) + require.True(t, w.resolve("internal-b", ElicitationResult{Action: tools.ElicitationActionAccept, Content: map[string]any{"who": "b"}})) + + select { + case result := <-wtServerA.ch: + assert.Equal(t, map[string]any{"who": "a"}, result.Content, "server A's waiter must not have been evicted or overwritten") + default: + t.Fatal("server A's waiter never received its response") + } + select { + case result := <-wtServerB.ch: + assert.Equal(t, map[string]any{"who": "b"}, result.Content, "server B's waiter must not have been evicted or overwritten") + default: + t.Fatal("server B's waiter never received its response") + } +} + +// TestElicitationWaiter_CancelWinsWhenFirst pins the #3584 review item 2b +// cancellation-vs-response race: when the handler's ctx.Done() branch wins +// the terminal-state CAS first, a subsequent resolve() must be told it lost +// (return false) instead of silently succeeding into a channel nobody will +// ever read again. +func TestElicitationWaiter_CancelWinsWhenFirst(t *testing.T) { + t.Parallel() + + var w elicitationWaiters + wt := w.register("a") + + require.True(t, w.cancel("a", wt), "cancel must win when nothing resolved yet") + assert.False(t, w.resolve("a", ElicitationResult{Action: tools.ElicitationActionAccept}), + "a resolve racing after cancel already won must not report success") + assert.Equal(t, 0, w.count(), "cancel must remove the waiter from the registry") +} + +// TestElicitationWaiter_ResolveWinsWhenFirst is the mirror image: resolve() +// wins the race first, so the handler's cancel() call must lose and report +// false, telling the handler to drain the value from the channel instead of +// discarding a response ResumeElicitation already reported as delivered. +func TestElicitationWaiter_ResolveWinsWhenFirst(t *testing.T) { + t.Parallel() + + var w elicitationWaiters + wt := w.register("a") + + require.True(t, w.resolve("a", ElicitationResult{Action: tools.ElicitationActionAccept, Content: map[string]any{"k": "v"}})) + assert.False(t, w.cancel("a", wt), "cancel must lose once resolve already won the terminal-state race") + + select { + case result := <-wt.ch: + assert.Equal(t, map[string]any{"k": "v"}, result.Content, "the resolved value must still be retrievable by the loser of the race") + default: + t.Fatal("resolve's value must be in the channel even though cancel lost the race") + } +} + +// TestElicitationWaiters_ConcurrentRegisterResolveDeregister runs many +// concurrent request/response pairs through the registry to catch data races +// (run with -race) and confirm no response is ever misdelivered under +// concurrent load, mirroring the concurrent background-job scenario from +// the audit. +func TestElicitationWaiters_ConcurrentRegisterResolveDeregister(t *testing.T) { + t.Parallel() + + var w elicitationWaiters + const n = 200 + + var wg sync.WaitGroup + for i := range n { + wg.Go(func() { + id := fmt.Sprintf("req-%d", i) + wt := w.register(id) + defer w.abandon(id, wt) + + done := make(chan struct{}) + go func() { + defer close(done) + ok := w.resolve(id, ElicitationResult{Action: tools.ElicitationActionAccept, Content: map[string]any{"i": i}}) + assert.True(t, ok) + }() + + select { + case result := <-wt.ch: + assert.Equal(t, map[string]any{"i": i}, result.Content, "response must route back to its own request") + case <-time.After(2 * time.Second): + t.Errorf("waiter %s never received its response", id) + } + <-done + }) + } + wg.Wait() +} + +// TestElicitationWaiters_ConcurrentResolveCancelRace hammers a single waiter +// with concurrent resolve/cancel attempts (run with -race) to confirm the +// terminal-state CAS lets exactly one of them win, never both and never +// neither. +func TestElicitationWaiters_ConcurrentResolveCancelRace(t *testing.T) { + t.Parallel() + + const n = 300 + for i := range n { + var w elicitationWaiters + id := fmt.Sprintf("race-%d", i) + wt := w.register(id) + + var wg sync.WaitGroup + var resolveWon, cancelWon atomicBool + wg.Add(2) + go func() { + defer wg.Done() + if w.resolve(id, ElicitationResult{Action: tools.ElicitationActionAccept}) { + resolveWon.set(true) + } + }() + go func() { + defer wg.Done() + if w.cancel(id, wt) { + cancelWon.set(true) + } + }() + wg.Wait() + + require.NotEqual(t, resolveWon.get(), cancelWon.get(), "exactly one of resolve/cancel must win, never both or neither") + } +} + +// atomicBool is a tiny test-local helper; sync/atomic.Bool is available but +// spelling out set/get keeps the race-loop above terse. +type atomicBool struct { + mu sync.Mutex + v bool +} + +func (a *atomicBool) set(v bool) { + a.mu.Lock() + defer a.mu.Unlock() + a.v = v +} + +func (a *atomicBool) get() bool { + a.mu.Lock() + defer a.mu.Unlock() + return a.v +} + +// --- elicitationBridge: bounded/non-blocking send (#3584 review item 1) --- + +// TestElicitationBridge_SendBlocksUntilCtxDone pins the review-item-1 fix: an +// unbuffered, unconsumed bridge channel used to block send() forever with no +// way out. send() must now be bounded by ctx and release with ctx.Err() +// instead of hanging, and must never panic. +func TestElicitationBridge_SendBlocksUntilCtxDone(t *testing.T) { + t.Parallel() + + var b elicitationBridge + ch := make(chan Event) // unbuffered, nobody ever reads it + b.swap(ch) + + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + + start := time.Now() + err := b.send(ctx, Warning("hello", "agent")) + elapsed := time.Since(start) + + require.ErrorIs(t, err, context.DeadlineExceeded, "send on a full/abandoned channel must release via ctx, not block forever") + assert.Less(t, elapsed, 2*time.Second, "send must not block substantially past the ctx deadline") +} + +// TestElicitationBridge_SendNeverBlocksReliableSink is the end-to-end version +// of the review-item-1 fix: a wedged bridge channel must not delay — let +// alone block — elicitationHandler's reliable OnElicitationRequest sink +// delivery or its subsequent wait for a response. Before the fix, the bridge +// send was awaited synchronously and BEFORE the sink call, so an abandoned +// channel meant the sink (and thus the whole request) never even started. +func TestElicitationBridge_SendNeverBlocksReliableSink(t *testing.T) { + t.Parallel() + + rt := newElicitationTestRuntime(t) + + // Wedge the bridge: swap in an unbuffered channel with no reader, as if + // a concurrent RunStream's swap left a dead consumer behind. + wedged := make(chan Event) + rt.elicitation.swap(wedged) + + sinkCalled := make(chan Event, 1) + rt.OnElicitationRequest(func(ev Event) { sinkCalled <- ev }) + + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + type handlerResult struct { + result tools.ElicitationResult + err error + } + done := make(chan handlerResult, 1) + go func() { + result, err := rt.elicitationHandler(ctx, &mcp.ElicitParams{Message: "confirm?"}) + done <- handlerResult{result, err} + }() + + // The sink must fire almost immediately, regardless of the wedged bridge. + var ev *ElicitationRequestEvent + select { + case e := <-sinkCalled: + ev = e.(*ElicitationRequestEvent) + case <-time.After(1 * time.Second): + t.Fatal("the reliable sink must not be blocked by a wedged bridge channel") + } + + require.NoError(t, rt.ResumeElicitation(t.Context(), tools.ElicitationActionAccept, nil, ev.ElicitationID)) + + select { + case got := <-done: + require.NoError(t, got.err) + assert.Equal(t, tools.ElicitationActionAccept, got.result.Action) + case <-time.After(1 * time.Second): + t.Fatal("elicitationHandler must not be blocked by a wedged bridge channel") + } +} + +// --- elicitationHandler: headless fast-decline (#3584 item 5) --- + +func TestElicitationHandler_HeadlessBackgroundFastDeclines(t *testing.T) { + t.Parallel() + + rt := newElicitationTestRuntime(t) + + // No OnElicitationRequest sink registered, and the context is marked + // non-interactive the way runStreamLoop marks a background + // (run_background_agent) session's context (#3200). + ctx := mcptools.WithoutInteractivePrompts(t.Context()) + ctx = genai.WithConversationID(ctx, "bg-sess-1") + + result, err := rt.elicitationHandler(ctx, &mcp.ElicitParams{Message: "need sudo password"}) + require.NoError(t, err) + assert.Equal(t, tools.ElicitationActionDecline, result.Action, + "a background session with no UI sink must fast-decline instead of blocking forever") + + notes := rt.elicitationDeclines.drain("bg-sess-1") + require.Len(t, notes, 1) + assert.Contains(t, notes[0], "need sudo password") +} + +// TestElicitationHandler_BackgroundWithSinkStillWaitsForResponse verifies +// that registering an OnElicitationRequest sink is enough to opt a +// background-session elicitation back into the normal wait-for-response +// path instead of being fast-declined. +func TestElicitationHandler_BackgroundWithSinkStillWaitsForResponse(t *testing.T) { + t.Parallel() + + rt := newElicitationTestRuntime(t) + + received := make(chan Event, 1) + rt.OnElicitationRequest(func(ev Event) { received <- ev }) + + ctx := mcptools.WithoutInteractivePrompts(t.Context()) + ctx = genai.WithConversationID(ctx, "bg-sess-2") + + type handlerResult struct { + result tools.ElicitationResult + err error + } + done := make(chan handlerResult, 1) + go func() { + result, err := rt.elicitationHandler(ctx, &mcp.ElicitParams{Message: "confirm?"}) + done <- handlerResult{result, err} + }() + + var ev *ElicitationRequestEvent + select { + case e := <-received: + ev = e.(*ElicitationRequestEvent) + case <-time.After(2 * time.Second): + t.Fatal("sink never received the elicitation request") + } + assert.Equal(t, "bg-sess-2", ev.SessionID, "the event must carry the originating (sub-)session ID") + + require.NoError(t, rt.ResumeElicitation(t.Context(), tools.ElicitationActionAccept, nil, ev.ElicitationID)) + + select { + case got := <-done: + require.NoError(t, got.err) + assert.Equal(t, tools.ElicitationActionAccept, got.result.Action) + case <-time.After(2 * time.Second): + t.Fatal("elicitationHandler never returned") + } + assert.Empty(t, rt.elicitationDeclines.drain("bg-sess-2"), "must not fast-decline once a sink is registered") +} + +// TestElicitationHandler_TOCTOU_ResolveImmediatelyAfterRegister promotes the +// former helper-level TOCTOU pin (TestElicitationWaiters_ResolveBeforeReceiveIsNotLost) +// to exercise the real elicitationHandler code path end-to-end: the sink +// fires synchronously before the handler ever reaches its response select, +// so a caller that resolves the instant it observes the sink-delivered event +// — beating the handler to its `select` — must still have its response +// delivered rather than racing the old shared-channel `default:` branch. +func TestElicitationHandler_TOCTOU_ResolveImmediatelyAfterRegister(t *testing.T) { + t.Parallel() + + rt := newElicitationTestRuntime(t) + + sinkDone := make(chan struct{}) + rt.OnElicitationRequest(func(ev Event) { + // Resolve from inside the sink callback itself, i.e. before + // elicitationHandler's goroutine has any chance to reach its + // `select` on the waiter channel. This is the tightest possible + // version of the TOCTOU window. + e := ev.(*ElicitationRequestEvent) + ok := rt.elicitationWaiters.resolve(e.ElicitationID, ElicitationResult{Action: tools.ElicitationActionAccept, Content: map[string]any{"answered": true}}) + assert.True(t, ok, "resolve issued synchronously from the sink callback must still find the just-registered waiter") + close(sinkDone) + }) + + result, err := rt.elicitationHandler(t.Context(), &mcp.ElicitParams{Message: "confirm?"}) + require.NoError(t, err) + assert.Equal(t, tools.ElicitationActionAccept, result.Action) + assert.Equal(t, map[string]any{"answered": true}, result.Content) + <-sinkDone +} + +// --- Full-stack regression: concurrent background-job elicitations (#3584) --- + +// elicitingToolSet is a minimal toolset whose one tool blocks on an MCP +// elicitation via the handler ConfigureHandlers wires onto it — mirroring +// how a real MCP toolset elicits mid-tool-call — without needing a real MCP +// server. Used to reproduce the reported bug: elicitations raised from +// multiple concurrent background jobs (run_background_agent) must all +// surface and each response must reach its own waiter. +type elicitingToolSet struct { + mu sync.Mutex + handler tools.ElicitationHandler + message string + // received captures the ElicitationResult the tool call got back, so + // tests can verify the response that reached this specific worker + // without having to scrape it back out of the model transcript. + received tools.ElicitationResult + gotResult bool +} + +var _ tools.Elicitable = (*elicitingToolSet)(nil) + +func (e *elicitingToolSet) SetElicitationHandler(handler tools.ElicitationHandler) { + e.mu.Lock() + defer e.mu.Unlock() + e.handler = handler +} + +func (e *elicitingToolSet) snapshot() (tools.ElicitationResult, bool) { + e.mu.Lock() + defer e.mu.Unlock() + return e.received, e.gotResult +} + +func (e *elicitingToolSet) Tools(context.Context) ([]tools.Tool, error) { + return []tools.Tool{{ + Name: "ask_user", + Handler: func(ctx context.Context, _ tools.ToolCall, _ tools.Runtime) (*tools.ToolCallResult, error) { + e.mu.Lock() + handler := e.handler + e.mu.Unlock() + if handler == nil { + return tools.ResultError("no elicitation handler configured"), nil + } + result, err := handler(ctx, &mcp.ElicitParams{Message: e.message}) + if err != nil { + return nil, err + } + e.mu.Lock() + e.received = result + e.gotResult = true + e.mu.Unlock() + return tools.ResultSuccess(fmt.Sprintf("%s:%v", result.Action, result.Content)), nil + }, + }}, nil +} + +// TestConcurrentBackgroundElicitations_AllSurfaceAndRouteToCorrectWaiter is +// the end-to-end regression test for issue #3584 / audit finding A6: two +// concurrent background jobs (as run_background_agent spawns them) each +// raise an elicitation mid-tool-call. Before the fix, the single-slot +// elicitationBridge and shared elicitationRequestCh meant: (a) runCollecting +// silently dropped the ElicitationRequestEvent, so nothing was ever +// displayed, and (b) even if something had displayed it, a response could be +// delivered to the wrong waiter. This asserts both jobs surface via the new +// OnElicitationRequest sink — exactly once each, with no dedupe map required +// (runCollecting no longer re-forwards a bridge-observed copy; see #3584 +// review item 5) — and each receives its own, non-swapped response. +func TestConcurrentBackgroundElicitations_AllSurfaceAndRouteToCorrectWaiter(t *testing.T) { + t.Parallel() + + newWorker := func(name, question string) (*agent.Agent, *elicitingToolSet) { + ts := &elicitingToolSet{message: question} + toolCallStream := newStreamBuilder().AddToolCallWithStop("call_1", "ask_user", "{}").Build() + followUpStream := newStreamBuilder().AddContent("done").AddStopWithUsage(5, 5).Build() + prov := &queueProvider{id: "test/mock-model", streams: []chat.MessageStream{toolCallStream, followUpStream}} + a := agent.New(name, "worker", agent.WithModel(prov), agent.WithToolSets(ts)) + return a, ts + } + + worker1, ts1 := newWorker("worker1", "worker1 needs input") + worker2, ts2 := newWorker("worker2", "worker2 needs input") + root := agent.New("root", "root", agent.WithModel(&mockProvider{id: "test/mock-model", stream: &mockStream{}})) + agent.WithSubAgents(worker1, worker2)(root) + + tm := team.New(team.WithAgents(root, worker1, worker2)) + rt, err := NewLocalRuntime(t.Context(), tm, WithSessionCompaction(false), WithModelStore(mockModelStore{})) + require.NoError(t, err) + + // deliveries counts sink invocations per ElicitationID. Each ID must be + // delivered exactly once: any count > 1 means the exactly-once guarantee + // (elicitationHandler is the sole caller of emitElicitationRequest) has + // regressed, since nothing else in the App/runtime layer masks + // duplicates any more. + var mu sync.Mutex + deliveries := make(map[string]int) + requests := make(map[string]*ElicitationRequestEvent) + rt.OnElicitationRequest(func(ev Event) { + req := ev.(*ElicitationRequestEvent) + mu.Lock() + defer mu.Unlock() + deliveries[req.ElicitationID]++ + requests[req.ElicitationID] = req + }) + + results := make(chan *agenttool.RunResult, 2) + launch := func(agentName string) { + go func() { + parent := session.New(session.WithUserMessage("go"), session.WithToolsApproved(true)) + res := rt.RunAgent(t.Context(), agenttool.RunParams{ + AgentName: agentName, + Task: "do it", + ParentSession: parent, + }) + results <- res + }() + } + launch("worker1") + launch("worker2") + + // Wait until both elicitations have surfaced, then respond to each by + // its own ID with a distinguishable payload so a swapped response would + // be caught by the assertions below. + var reqs map[string]*ElicitationRequestEvent + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + reqs = make(map[string]*ElicitationRequestEvent, len(requests)) + maps.Copy(reqs, requests) + return len(reqs) == 2 + }, 5*time.Second, 10*time.Millisecond, "both concurrent background elicitations must surface via the sink") + + mu.Lock() + for id, count := range deliveries { + assert.Equal(t, 1, count, "elicitation %s must be delivered to the sink exactly once", id) + } + mu.Unlock() + + var worker1ID, worker2ID string + for id, ev := range reqs { + switch ev.Message { + case "worker1 needs input": + worker1ID = id + case "worker2 needs input": + worker2ID = id + } + } + require.NotEmpty(t, worker1ID, "worker1's elicitation must have surfaced") + require.NotEmpty(t, worker2ID, "worker2's elicitation must have surfaced") + require.NotEqual(t, worker1ID, worker2ID, "concurrent elicitations must get distinct correlation IDs") + + require.NoError(t, rt.ResumeElicitation(t.Context(), tools.ElicitationActionAccept, map[string]any{"answer": "1"}, worker1ID)) + require.NoError(t, rt.ResumeElicitation(t.Context(), tools.ElicitationActionAccept, map[string]any{"answer": "2"}, worker2ID)) + + var got []*agenttool.RunResult + for range 2 { + select { + case r := <-results: + got = append(got, r) + case <-time.After(5 * time.Second): + t.Fatal("a background job never completed after its elicitation was resumed") + } + } + for _, r := range got { + require.Empty(t, r.ErrMsg, "background job must not fail") + } + + worker1Result, ok1 := ts1.snapshot() + require.True(t, ok1, "worker1's tool call must have received an elicitation result") + worker2Result, ok2 := ts2.snapshot() + require.True(t, ok2, "worker2's tool call must have received an elicitation result") + + assert.Equal(t, map[string]any{"answer": "1"}, worker1Result.Content, + "worker1 must receive its own response, not worker2's (no swap)") + assert.Equal(t, map[string]any{"answer": "2"}, worker2Result.Content, + "worker2 must receive its own response, not worker1's (no swap)") +} diff --git a/pkg/runtime/elicitation_test.go b/pkg/runtime/elicitation_test.go index 11ab050d9a..17d7d4a195 100644 --- a/pkg/runtime/elicitation_test.go +++ b/pkg/runtime/elicitation_test.go @@ -27,7 +27,7 @@ func TestElicitationBridge_SendBeforeSwapReturnsError(t *testing.T) { t.Parallel() var b elicitationBridge - err := b.send(Error("nothing")) + err := b.send(t.Context(), Error("nothing")) assert.ErrorIs(t, err, errNoElicitationChannel) } @@ -55,7 +55,7 @@ func TestElicitationBridge_SendDeliversToCurrentChannel(t *testing.T) { ch := make(chan Event, 1) b.swap(ch) - require.NoError(t, b.send(Error("hello"))) + require.NoError(t, b.send(t.Context(), Error("hello"))) select { case ev := <-ch: @@ -75,7 +75,7 @@ func TestElicitationBridge_SendRecoversClosedChannel(t *testing.T) { b.swap(ch) close(ch) - err := b.send(Error("closed")) + err := b.send(t.Context(), Error("closed")) assert.ErrorIs(t, err, errNoElicitationChannel) } @@ -97,7 +97,7 @@ func TestElicitationBridge_RestoreAndCloseWaitsForInflightSenders(t *testing.T) sendDone := make(chan error, 1) go func() { - sendDone <- b.send(Error("inflight")) + sendDone <- b.send(t.Context(), Error("inflight")) }() // Wait until the sender holds the read lock: TryLock fails only while @@ -167,7 +167,7 @@ func TestElicitationBridge_ConcurrentSendsAndCloseAreSerializedSafely(t *testing for range 10 { wg.Go(func() { for range 5 { - _ = b.send(Error("x")) + _ = b.send(t.Context(), Error("x")) } }) } diff --git a/pkg/runtime/event.go b/pkg/runtime/event.go index 080ded540f..e5e4b0b1c0 100644 --- a/pkg/runtime/event.go +++ b/pkg/runtime/event.go @@ -583,28 +583,52 @@ func (e *PausedEvent) GetSessionID() string { return e.SessionID } type ElicitationRequestEvent struct { AgentContext - Type string `json:"type"` - Message string `json:"message"` - Mode string `json:"mode,omitempty"` // "form" or "url" - Schema any `json:"schema,omitempty"` - URL string `json:"url,omitempty"` - ElicitationID string `json:"elicitation_id,omitempty"` - Meta map[string]any `json:"meta,omitempty"` -} - -func ElicitationRequest(message, mode string, schema any, url, elicitationID string, meta map[string]any, agentName string) Event { + Type string `json:"type"` + Message string `json:"message"` + Mode string `json:"mode,omitempty"` // "form" or "url" + Schema any `json:"schema,omitempty"` + URL string `json:"url,omitempty"` + // ElicitationID is the internally-generated correlation ID used to route + // a ResumeElicitation response back to this specific request (see + // elicitationHandler). It is always set and always unique, regardless of + // whether the originating MCP server supplied its own wire ID: two + // independent servers can coincidentally reuse the same wire ID, so that + // value is never used for routing (#3584). + ElicitationID string `json:"elicitation_id,omitempty"` + // ServerElicitationID is the MCP wire-protocol elicitationId as supplied + // by the originating server, if any (URL-mode elicitations only; form + // elicitations leave it empty). It is informational only — useful for + // correlating with server-side logs — and must never be used as a + // routing key. + ServerElicitationID string `json:"server_elicitation_id,omitempty"` + // SessionID is the session (or sub-session) on whose behalf this + // elicitation was raised. A detached background job's sub-session ID + // differs from its parent/foreground session, which lets consumers (see + // the TUI supervisor) tell apart a foreground stream's own elicitation + // from a still-live background job's when the foreground stream stops + // (#3584 review item 4). + SessionID string `json:"session_id,omitempty"` + Meta map[string]any `json:"meta,omitempty"` +} + +func ElicitationRequest(message, mode string, schema any, url, elicitationID, serverElicitationID, sessionID string, meta map[string]any, agentName string) Event { return &ElicitationRequestEvent{ - Type: "elicitation_request", - Message: message, - Mode: mode, - Schema: schema, - URL: url, - ElicitationID: elicitationID, - Meta: meta, - AgentContext: newAgentContext(agentName), + Type: "elicitation_request", + Message: message, + Mode: mode, + Schema: schema, + URL: url, + ElicitationID: elicitationID, + ServerElicitationID: serverElicitationID, + SessionID: sessionID, + Meta: meta, + AgentContext: newAgentContext(agentName), } } +// GetSessionID makes ElicitationRequestEvent satisfy [SessionScoped]. +func (e *ElicitationRequestEvent) GetSessionID() string { return e.SessionID } + type AuthorizationEvent struct { AgentContext diff --git a/pkg/runtime/remote_client.go b/pkg/runtime/remote_client.go index 2029869d2e..3cff900765 100644 --- a/pkg/runtime/remote_client.go +++ b/pkg/runtime/remote_client.go @@ -22,8 +22,11 @@ type RemoteClient interface { // ResumeSession resumes a paused session with optional rejection reason or tool name ResumeSession(ctx context.Context, id, confirmation, reason, toolName string) error - // ResumeElicitation sends an elicitation response - ResumeElicitation(ctx context.Context, sessionID string, action tools.ElicitationAction, content map[string]any) error + // ResumeElicitation sends an elicitation response. elicitationID is + // additive: pass "" to fall back to the sole-pending-request behavior. + // Variadic for the same Go-API-compatibility reason as + // [Runtime.ResumeElicitation]: at most one value is meaningful. + ResumeElicitation(ctx context.Context, sessionID string, action tools.ElicitationAction, content map[string]any, elicitationID ...string) error // RunAgent executes an agent and returns a channel of streaming events. // model, when non-empty, is applied as a persistent override on the diff --git a/pkg/runtime/remote_contract_test.go b/pkg/runtime/remote_contract_test.go index 2c2fbb1b14..ec055db91b 100644 --- a/pkg/runtime/remote_contract_test.go +++ b/pkg/runtime/remote_contract_test.go @@ -41,7 +41,7 @@ func (s *stubRemoteClient) ResumeSession(context.Context, string, string, string return nil } -func (s *stubRemoteClient) ResumeElicitation(context.Context, string, tools.ElicitationAction, map[string]any) error { +func (s *stubRemoteClient) ResumeElicitation(context.Context, string, tools.ElicitationAction, map[string]any, ...string) error { return nil } diff --git a/pkg/runtime/remote_runtime.go b/pkg/runtime/remote_runtime.go index b849005e23..acc337d31f 100644 --- a/pkg/runtime/remote_runtime.go +++ b/pkg/runtime/remote_runtime.go @@ -394,15 +394,16 @@ func (r *RemoteRuntime) convertSessionMessages(sess *session.Session) []api.Mess } // ResumeElicitation sends an elicitation response back to a waiting elicitation request -func (r *RemoteRuntime) ResumeElicitation(ctx context.Context, action tools.ElicitationAction, content map[string]any) error { - slog.DebugContext(ctx, "Resuming remote runtime with elicitation response", "agent", r.currentAgent, "action", action, "session_id", r.sessionID) +func (r *RemoteRuntime) ResumeElicitation(ctx context.Context, action tools.ElicitationAction, content map[string]any, elicitationID ...string) error { + id := firstElicitationID(elicitationID) + slog.DebugContext(ctx, "Resuming remote runtime with elicitation response", "agent", r.currentAgent, "action", action, "session_id", r.sessionID, "elicitation_id", id) err := r.handleOAuthElicitation(ctx, r.pendingOAuthElicitation) if err != nil { return err } - if err := r.client.ResumeElicitation(ctx, r.sessionID, action, content); err != nil { + if err := r.client.ResumeElicitation(ctx, r.sessionID, action, content, id); err != nil { return err } @@ -420,7 +421,7 @@ func (r *RemoteRuntime) handleOAuthElicitation(ctx context.Context, req *Elicita if !ok { err := errors.New("server_url missing from elicitation metadata") slog.ErrorContext(ctx, "Failed to extract server_url", "error", err) - _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil) + _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil, req.ElicitationID) return err } @@ -428,7 +429,7 @@ func (r *RemoteRuntime) handleOAuthElicitation(ctx context.Context, req *Elicita if !ok { err := errors.New("auth_server_metadata missing from elicitation metadata") slog.ErrorContext(ctx, "Failed to extract auth_server_metadata", "error", err) - _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil) + _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil, req.ElicitationID) return err } @@ -436,12 +437,12 @@ func (r *RemoteRuntime) handleOAuthElicitation(ctx context.Context, req *Elicita metadataBytes, err := json.Marshal(authServerMetadata) if err != nil { slog.ErrorContext(ctx, "Failed to marshal auth_server_metadata", "error", err) - _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil) + _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil, req.ElicitationID) return fmt.Errorf("failed to marshal auth_server_metadata: %w", err) } if err := json.Unmarshal(metadataBytes, &authMetadata); err != nil { slog.ErrorContext(ctx, "Failed to unmarshal auth_server_metadata", "error", err) - _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil) + _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil, req.ElicitationID) return fmt.Errorf("failed to unmarshal auth_server_metadata: %w", err) } @@ -461,7 +462,7 @@ func (r *RemoteRuntime) handleOAuthElicitation(ctx context.Context, req *Elicita callbackServer, err := mcp.NewCallbackServer(ctx) if err != nil { slog.ErrorContext(ctx, "Failed to create callback server", "error", err) - _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil) + _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil, req.ElicitationID) return fmt.Errorf("failed to create callback server: %w", err) } defer func() { @@ -476,7 +477,7 @@ func (r *RemoteRuntime) handleOAuthElicitation(ctx context.Context, req *Elicita if err := callbackServer.Start(); err != nil { slog.ErrorContext(ctx, "Failed to start callback server", "error", err) - _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil) + _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil, req.ElicitationID) return fmt.Errorf("failed to start callback server: %w", err) } @@ -489,21 +490,21 @@ func (r *RemoteRuntime) handleOAuthElicitation(ctx context.Context, req *Elicita clientID, clientSecret, err = mcp.RegisterClient(oauthCtx, &authMetadata, redirectURI, nil) if err != nil { slog.ErrorContext(ctx, "Dynamic client registration failed", "error", err) - _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil) + _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil, req.ElicitationID) return fmt.Errorf("failed to register client: %w", err) } slog.DebugContext(ctx, "Client registered successfully", "client_id", clientID) } else { err := errors.New("authorization server does not support dynamic client registration") slog.ErrorContext(ctx, "Client registration not supported", "error", err) - _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil) + _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil, req.ElicitationID) return err } state, err := mcp.GenerateState() if err != nil { slog.ErrorContext(ctx, "Failed to generate state", "error", err) - _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil) + _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil, req.ElicitationID) return fmt.Errorf("failed to generate state: %w", err) } @@ -526,14 +527,14 @@ func (r *RemoteRuntime) handleOAuthElicitation(ctx context.Context, req *Elicita code, receivedState, err := mcp.RequestAuthorizationCode(oauthCtx, authURL, callbackServer, state) if err != nil { slog.ErrorContext(ctx, "Failed to get authorization code", "error", err) - _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil) + _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil, req.ElicitationID) return fmt.Errorf("failed to get authorization code: %w", err) } if receivedState != state { err := fmt.Errorf("state mismatch: expected %s, got %s", state, receivedState) slog.ErrorContext(ctx, "State mismatch in authorization response", "error", err) - _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil) + _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil, req.ElicitationID) return err } @@ -551,7 +552,7 @@ func (r *RemoteRuntime) handleOAuthElicitation(ctx context.Context, req *Elicita ) if err != nil { slog.ErrorContext(ctx, "Failed to exchange code for token", "error", err) - _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil) + _ = r.client.ResumeElicitation(ctx, r.sessionID, "decline", nil, req.ElicitationID) return fmt.Errorf("failed to exchange code for token: %w", err) } @@ -569,7 +570,7 @@ func (r *RemoteRuntime) handleOAuthElicitation(ctx context.Context, req *Elicita } slog.DebugContext(ctx, "Sending token to server") - if err := r.client.ResumeElicitation(ctx, r.sessionID, tools.ElicitationActionAccept, tokenData); err != nil { + if err := r.client.ResumeElicitation(ctx, r.sessionID, tools.ElicitationActionAccept, tokenData, req.ElicitationID); err != nil { slog.ErrorContext(ctx, "Failed to send token to server", "error", err) return fmt.Errorf("failed to send token to server: %w", err) } @@ -706,6 +707,12 @@ func (r *RemoteRuntime) OnToolsChanged(func(Event)) {} // run server-side and their events are not forwarded out-of-band. func (r *RemoteRuntime) OnBackgroundEvent(func(Event)) {} +// OnElicitationRequest is a no-op for remote runtimes; elicitation requests +// (including from server-side background jobs) arrive as ElicitationRequestEvent +// values on the RunStream channel itself (see the RunStream forwarding loop +// above), so there is no separate out-of-band sink to register. +func (r *RemoteRuntime) OnElicitationRequest(func(Event)) {} + // Close is a no-op for remote runtimes. func (r *RemoteRuntime) Close() error { return nil diff --git a/pkg/runtime/runtime.go b/pkg/runtime/runtime.go index fa6a5365f1..8f5d21e637 100644 --- a/pkg/runtime/runtime.go +++ b/pkg/runtime/runtime.go @@ -82,8 +82,16 @@ type Runtime interface { // Resume allows resuming execution after user confirmation. // The ResumeRequest carries the decision type and an optional reason (for rejections). Resume(ctx context.Context, req ResumeRequest) - // ResumeElicitation sends an elicitation response back to a waiting elicitation request - ResumeElicitation(_ context.Context, action tools.ElicitationAction, content map[string]any) error + // ResumeElicitation sends an elicitation response back to a waiting + // elicitation request. elicitationID correlates the response with a + // specific concurrent request (see ElicitationRequestEvent.ElicitationID); + // omit it (or pass "") to fall back to resolving the sole pending + // request, for backward compatibility with older clients. The parameter + // is variadic — rather than a required 4th positional argument — purely + // so pre-#3584 callers of this interface method keep compiling unchanged; + // implementations should treat more than one value as a caller error and + // use only the first. + ResumeElicitation(_ context.Context, action tools.ElicitationAction, content map[string]any, elicitationID ...string) error // SessionStore returns the session store for browsing/loading past sessions. // Returns nil if no persistent session store is configured. SessionStore() session.Store @@ -176,6 +184,14 @@ type Runtime interface { // background work can implement this as a no-op. OnBackgroundEvent(handler func(Event)) + // OnElicitationRequest registers a handler invoked whenever an MCP + // toolset raises an elicitation request, regardless of which stream + // (foreground or a detached background job) is currently "active" for + // the runtime's internal elicitation bridge. This is the reliable + // delivery route for background-job elicitations (#3584); runtimes + // that don't run local background work can implement this as a no-op. + OnElicitationRequest(handler func(Event)) + // QueueStatus returns the current depth and capacity of message queues QueueStatus() QueueStatus @@ -224,16 +240,23 @@ type LocalRuntime struct { managedOAuth bool unmanagedOAuthRedirectURI string nonInteractive bool - startupInfoEmitted bool // Track if startup info has been emitted to avoid unnecessary duplication - elicitationRequestCh chan ElicitationResult // Channel for receiving elicitation responses - elicitation elicitationBridge // Owns the per-stream events channel for outbound elicitation requests - sessionStore session.Store - workingDir string // Working directory for hooks execution - env []string // Environment variables for hooks execution - modelSwitcherCfg *ModelSwitcherConfig - providerRegistry *provider.Registry - gatewayModels gatewayModelsCache - dmrModels dmrModelsCache + startupInfoEmitted bool // Track if startup info has been emitted to avoid unnecessary duplication + elicitation elicitationBridge // Owns the per-stream events channel for outbound elicitation requests + elicitationWaiters elicitationWaiters // Routes elicitation responses to the request awaiting them, keyed by ID (#3584) + elicitationDeclines elicitationDeclineNotes + + // elicitationSinkMu guards onElicitationRequest; elicitation requests can + // arrive from background-job goroutines (runCollecting) concurrently with + // the sink being (re)registered. + elicitationSinkMu sync.RWMutex + onElicitationRequest func(Event) + sessionStore session.Store + workingDir string // Working directory for hooks execution + env []string // Environment variables for hooks execution + modelSwitcherCfg *ModelSwitcherConfig + providerRegistry *provider.Registry + gatewayModels gatewayModelsCache + dmrModels dmrModelsCache // hooksRegistry is the runtime-private hooks.Registry used to build // every Executor. It carries the runtime-owned builtin hooks @@ -610,7 +633,6 @@ func NewLocalRuntime(ctx context.Context, agents *team.Team, opts ...Opt) (*Loca team: agents, agents: newAgentRouter(agents, defaultAgent.Name()), resumeChan: make(chan ResumeRequest), - elicitationRequestCh: make(chan ElicitationResult), steerQueue: NewInMemoryMessageQueue(defaultSteerQueueCapacity), followUpQueue: NewInMemoryMessageQueue(defaultFollowUpQueueCapacity), sessionCompaction: true, diff --git a/pkg/runtime/runtime_test.go b/pkg/runtime/runtime_test.go index 858b556b33..833a925e46 100644 --- a/pkg/runtime/runtime_test.go +++ b/pkg/runtime/runtime_test.go @@ -4145,6 +4145,12 @@ func TestElicitationHandler_NonInteractive(t *testing.T) { assert.Equal(t, tools.ElicitationActionDecline, result.Action, "non-interactive runtime should decline elicitation") } +// TestElicitationHandler_Interactive_NoChannel verifies that an interactive +// elicitationHandler call succeeds even when the elicitation bridge has no +// channel configured (no RunStream is active). This used to be a hard +// failure because delivery depended entirely on the bridge; with per-request +// waiters (#3584) the bridge send is best-effort only, and ResumeElicitation +// still routes the response correctly via the waiter registry. func TestElicitationHandler_Interactive_NoChannel(t *testing.T) { t.Parallel() @@ -4152,7 +4158,7 @@ func TestElicitationHandler_Interactive_NoChannel(t *testing.T) { root := agent.New("root", "test", agent.WithModel(prov)) tm := team.New(team.WithAgents(root)) - // Default runtime (interactive mode) with no events channel set + // Default runtime (interactive mode) with no events channel set on the bridge. rt, err := NewLocalRuntime(t.Context(), tm) require.NoError(t, err) @@ -4160,10 +4166,29 @@ func TestElicitationHandler_Interactive_NoChannel(t *testing.T) { Message: "Authorize OAuth?", } - _, err = rt.elicitationHandler(t.Context(), params) + type handlerResult struct { + result tools.ElicitationResult + err error + } + done := make(chan handlerResult, 1) + go func() { + result, err := rt.elicitationHandler(t.Context(), params) + done <- handlerResult{result: result, err: err} + }() + + require.Eventually(t, func() bool { return rt.elicitationWaiters.count() == 1 }, time.Second, time.Millisecond, + "elicitationHandler must register a waiter even though the bridge has no channel") - require.Error(t, err, "interactive runtime with no events channel should error") - assert.ErrorIs(t, err, errNoElicitationChannel) + require.NoError(t, rt.ResumeElicitation(t.Context(), tools.ElicitationActionAccept, map[string]any{"ok": true}, "")) + + select { + case got := <-done: + require.NoError(t, got.err) + assert.Equal(t, tools.ElicitationActionAccept, got.result.Action) + assert.Equal(t, map[string]any{"ok": true}, got.result.Content) + case <-time.After(2 * time.Second): + t.Fatal("elicitationHandler did not return after ResumeElicitation") + } } // TestRunAgentPersistsSubSessionToStore is the regression test for the diff --git a/pkg/server/server.go b/pkg/server/server.go index 6f1c7e3b74..292d66ea16 100644 --- a/pkg/server/server.go +++ b/pkg/server/server.go @@ -485,7 +485,7 @@ func (s *Server) elicitation(c echo.Context) error { return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("invalid request body: %v", err)) } - if err := s.sm.ResumeElicitation(c.Request().Context(), sessionID, req.Action, req.Content); err != nil { + if err := s.sm.ResumeElicitation(c.Request().Context(), sessionID, req.Action, req.Content, req.ElicitationID); err != nil { return echo.NewHTTPError(http.StatusInternalServerError, fmt.Sprintf("failed to resume elicitation: %v", err)) } diff --git a/pkg/server/session_manager.go b/pkg/server/session_manager.go index d6d77c05e0..198b4eb43c 100644 --- a/pkg/server/session_manager.go +++ b/pkg/server/session_manager.go @@ -159,6 +159,54 @@ func (sm *SessionManager) HasEventSource(sessionID string) bool { return ok } +// ensureEventLog returns sessionID's [pumpedEventLog], creating a bare one — +// ring buffer only, no pump goroutine — if none is registered yet. Used by +// appendSessionEvent to give a session a replayable event route even when it +// was never attached via RegisterEventSource (the common case for a session +// created directly by the API server rather than an external embedder like +// the TUI). +// +// cancel closes the log (delivering session_exited and disconnecting any +// live listener) instead of being a no-op: there is no external source pump +// to stop, but DeleteSession/BatchDeleteSessions call pe.cancel() +// unconditionally expecting it to end the log's lifetime. A no-op here left +// a connected GET /api/sessions/:id/events request on a lazily-created log +// blocked forever after deletion — it never saw session_exited and never +// closed, contradicting the end-of-session contract documented on +// Server.sessionEvents (#3584 re-review). +func (sm *SessionManager) ensureEventLog(sessionID string) *pumpedEventLog { + if pe, ok := sm.eventLogs.Load(sessionID); ok { + return pe + } + log := newEventLog(defaultEventLogCapacity) + pe := &pumpedEventLog{log: log, cancel: func() { log.close("session ended") }} + actual, _ := sm.eventLogs.LoadOrStore(sessionID, pe) + return actual +} + +// appendSessionEvent appends event to sessionID's event log, creating the +// log on demand (see ensureEventLog) if this is the first out-of-band event +// the session has ever produced. Registered as the runtime's +// OnElicitationRequest sink in runtimeForSession (#3584 review item 3). +func (sm *SessionManager) appendSessionEvent(sessionID string, event any) { + sm.ensureEventLog(sessionID).log.append(event) +} + +// sessionElicitationSink returns the OnElicitationRequest handler that +// runtimeForSession registers on every API/server-created runtime: it +// appends the event to sessionID's (lazily created) event log, giving +// out-of-band elicitations — chiefly from detached background jobs — a +// session-scoped route to any client polling/streaming +// GET /api/sessions/:id/events, instead of the runtime treating an absent +// local sink as "no UI" and auto-declining (#3584 review item 3). Split out +// as its own method so the exact wiring is unit-testable without spinning up +// a real runtime/team. +func (sm *SessionManager) sessionElicitationSink(sessionID string) func(runtime.Event) { + return func(ev runtime.Event) { + sm.appendSessionEvent(sessionID, ev) + } +} + // LastEventSeq returns the most recent event sequence number for sessionID, // so a snapshot can advertise the exact point from which a client should tail. // Returns 0 and false when no event log exists. @@ -911,8 +959,9 @@ func (sm *SessionManager) recallSession(ctx context.Context, sessionID string, m return nil } -// ResumeElicitation resumes an elicitation request. -func (sm *SessionManager) ResumeElicitation(ctx context.Context, sessionID, action string, content map[string]any) error { +// ResumeElicitation resumes an elicitation request. elicitationID is +// additive: pass "" to fall back to resolving the sole pending request. +func (sm *SessionManager) ResumeElicitation(ctx context.Context, sessionID, action string, content map[string]any, elicitationID string) error { sm.mux.Lock() defer sm.mux.Unlock() rt, exists := sm.runtimeSessions.Load(sessionID) @@ -920,7 +969,7 @@ func (sm *SessionManager) ResumeElicitation(ctx context.Context, sessionID, acti return errors.New("session not found") } - return rt.runtime.ResumeElicitation(ctx, tools.ElicitationAction(action), content) + return rt.runtime.ResumeElicitation(ctx, tools.ElicitationAction(action), content, elicitationID) } // ToggleToolApproval toggles the tool approval mode for a session. @@ -1085,6 +1134,22 @@ func (sm *SessionManager) runtimeForSession(ctx context.Context, sess *session.S return nil, nil, err } + // Give this session an out-of-band, session-scoped route for + // elicitations raised while nobody is synchronously reading this + // specific RunSession call's stream — most notably background jobs + // (run_background_agent) that outlive the request that started them. + // Before this, only pkg/app (the TUI) ever registered an + // OnElicitationRequest sink; every API/server-created runtime had none, + // so elicitationHandler's headless fast-decline path ("no sink means no + // UI") fired for every background elicitation raised through the API, + // even though RemoteRuntime.OnElicitationRequest's contract promises + // remote/SSE clients CAN answer them via ResumeElicitation. Appending to + // this session's event log makes the request replayable via GET + // /api/sessions/:id/events (lazily creating the log if this session was + // never attached with RegisterEventSource) and answerable via the + // existing POST .../elicitation route (#3584 review item 3). + run.OnElicitationRequest(sm.sessionElicitationSink(sess.ID)) + // Apply any stored per-agent model overrides so that a session // resumed (or freshly created with overrides via CreateSession) uses // the requested models instead of the agent's defaults. diff --git a/pkg/server/session_manager_test.go b/pkg/server/session_manager_test.go index f0e2247913..108bbcab55 100644 --- a/pkg/server/session_manager_test.go +++ b/pkg/server/session_manager_test.go @@ -63,7 +63,7 @@ func (f *fakeRuntime) Steer(_ context.Context, _ runtime.QueuedMessage) error { func (f *fakeRuntime) FollowUp(_ context.Context, _ runtime.QueuedMessage) error { return nil } -func (f *fakeRuntime) ResumeElicitation(_ context.Context, _ tools.ElicitationAction, _ map[string]any) error { +func (f *fakeRuntime) ResumeElicitation(_ context.Context, _ tools.ElicitationAction, _ map[string]any, _ ...string) error { return nil } @@ -651,3 +651,276 @@ func storedToolResultContent(t *testing.T, sess *session.Session, toolCallID str require.Failf(t, "tool result not found", "tool_call_id=%s", toolCallID) return "" } + +// --- #3584 review item 3: session-scoped elicitation sink for API/server runtimes --- + +// TestSessionElicitationSink_MakesSessionEventSourceReplayable pins the fix +// for review item 3: before this, only pkg/app (the TUI) ever registered an +// OnElicitationRequest sink, so every API/server-created runtime had none — +// elicitationHandler's headless fast-decline path ("no sink means no UI") +// therefore fired for every background elicitation raised through the API, +// even though a remote/SSE client could otherwise answer it. +// runtimeForSession registers sessionElicitationSink on every runtime it +// builds; this exercises that exact closure (without needing a full +// runtime/team) and confirms the session gains a replayable event source it +// didn't have before. +func TestSessionElicitationSink_MakesSessionEventSourceReplayable(t *testing.T) { + t.Parallel() + + ctx := t.Context() + sm := NewSessionManager(ctx, config.Sources{}, session.NewInMemorySessionStore(), 0, &config.RuntimeConfig{}) + + require.False(t, sm.HasEventSource("sess-1"), + "a session that was never attached and produced no out-of-band event must have no event source") + + sink := sm.sessionElicitationSink("sess-1") + ev := runtime.ElicitationRequest("need input", "form", nil, "", "eid-1", "", "bg-child", nil, "agent") + sink(ev) + + require.True(t, sm.HasEventSource("sess-1"), + "the sink must lazily create a session-scoped event source on first use") + + streamCtx, cancel := context.WithCancel(ctx) + var mu sync.Mutex + var replayed []any + done := make(chan struct{}) + go func() { + defer close(done) + ok := sm.StreamEvents(streamCtx, "sess-1", nil, func(_ uint64, event any) { + mu.Lock() + defer mu.Unlock() + replayed = append(replayed, event) + }) + assert.True(t, ok) + }() + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(replayed) == 1 + }, 2*time.Second, time.Millisecond) + cancel() + <-done + + mu.Lock() + defer mu.Unlock() + require.Len(t, replayed, 1) + assert.Same(t, ev, replayed[0], "the elicitation event must be replayable via GET .../events") +} + +// TestSessionElicitationSink_ReusesEventLogAcrossCalls verifies the sink +// doesn't clobber an existing event log (e.g. one already registered via +// RegisterEventSource for an attached session) and that repeated sink +// invocations accumulate rather than overwrite. +func TestSessionElicitationSink_ReusesEventLogAcrossCalls(t *testing.T) { + t.Parallel() + + ctx := t.Context() + sm := NewSessionManager(ctx, config.Sources{}, session.NewInMemorySessionStore(), 0, &config.RuntimeConfig{}) + + sink := sm.sessionElicitationSink("sess-2") + first := runtime.ElicitationRequest("first", "form", nil, "", "eid-1", "", "bg-child-1", nil, "agent") + second := runtime.ElicitationRequest("second", "form", nil, "", "eid-2", "", "bg-child-2", nil, "agent") + sink(first) + sink(second) + + seq, ok := sm.LastEventSeq("sess-2") + require.True(t, ok) + assert.Equal(t, uint64(2), seq, "both sink deliveries must land in the same event log") + + streamCtx, cancel := context.WithCancel(ctx) + var mu sync.Mutex + var replayed []any + done := make(chan struct{}) + go func() { + defer close(done) + ok := sm.StreamEvents(streamCtx, "sess-2", nil, func(_ uint64, event any) { + mu.Lock() + defer mu.Unlock() + replayed = append(replayed, event) + }) + assert.True(t, ok) + }() + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(replayed) == 2 + }, 2*time.Second, time.Millisecond) + cancel() + <-done + + mu.Lock() + defer mu.Unlock() + require.Len(t, replayed, 2) + assert.Same(t, first, replayed[0]) + assert.Same(t, second, replayed[1]) +} + +// TestRuntimeForSession_RegistersSessionScopedElicitationSink is an +// end-to-end check that runtimeForSession actually wires +// sessionElicitationSink onto every runtime it builds (not just that the +// sink mechanics work in isolation, per the tests above). It uses a +// harness-backed agent (harness: type: claude-code) so no model provider or +// API key is needed — runtime construction only requires *a* valid agent, +// per LocalRuntime's "has no valid model" guard. +// +// Crucially, this drives the sink through the RETURNED runtime itself (via +// EmitElicitationRequestForTesting) rather than reconstructing a fresh +// sm.sessionElicitationSink(...) closure and calling that in isolation: the +// latter would keep passing even if runtimeForSession's +// `run.OnElicitationRequest(...)` registration were deleted, since it never +// actually exercises what got wired onto `run` (#3584 re-review should-fix +// 1). +func TestRuntimeForSession_RegistersSessionScopedElicitationSink(t *testing.T) { + t.Parallel() + + ctx := t.Context() + cfg := []byte(`agents: + root: + description: Test agent + instruction: Be helpful. + harness: + type: claude-code +`) + store := session.NewInMemorySessionStore() + sess := session.New() + require.NoError(t, store.AddSession(ctx, sess)) + + sources := config.Sources{"agent.yaml": config.NewBytesSource("agent.yaml", cfg)} + sm := NewSessionManager(ctx, sources, store, 0, &config.RuntimeConfig{}) + + require.False(t, sm.HasEventSource(sess.ID)) + + run, _, err := sm.runtimeForSession(ctx, sess, "agent.yaml", "", &config.RuntimeConfig{}) + require.NoError(t, err) + t.Cleanup(func() { _ = run.Close() }) + + lr, ok := run.(*runtime.LocalRuntime) + require.True(t, ok, "runtimeForSession is expected to build a *runtime.LocalRuntime, got %T", run) + + // Simulate what elicitationHandler does when it raises a background + // elicitation, but through the sink runtimeForSession actually registered + // on this specific runtime instance, not a freshly built one. + lr.EmitElicitationRequestForTesting(runtime.ElicitationRequest("need input", "form", nil, "", "eid-1", "", sess.ID, nil, "root")) + require.True(t, sm.HasEventSource(sess.ID), + "runtimeForSession must leave the session able to surface out-of-band elicitations") +} + +// --- #3584 re-review blocker: lazily-created event logs must actually close on delete --- + +// TestReview_DeleteSessionClosesLazyElicitationEventLog is the regression +// test for the #3584 re-review blocker: ensureEventLog handed lazily-created +// (API-only) event logs a no-op cancel function, so DeleteSession's +// unconditional pe.cancel() call did nothing to the underlying eventLog — a +// client already streaming GET /api/sessions/:id/events for such a session +// would never receive a terminal session_exited event and would never see +// its stream close, contradicting the end-of-session contract documented on +// Server.sessionEvents (and docs/features/api-server/index.md). This proves +// the lazily-created log is actually closed on deletion: session_exited is +// delivered and the blocked StreamEvents call returns. +func TestReview_DeleteSessionClosesLazyElicitationEventLog(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := session.NewInMemorySessionStore() + sm := NewSessionManager(ctx, config.Sources{}, store, 0, &config.RuntimeConfig{}) + + sess := session.New() + require.NoError(t, store.AddSession(ctx, sess)) + + // Lazily create the session's event log the way runtimeForSession's + // sessionElicitationSink does for a background job's elicitation — no + // runtime/RegisterEventSource pump is ever registered for this session. + sm.sessionElicitationSink(sess.ID)(runtime.ElicitationRequest("need input", "form", nil, "", "eid-1", "", sess.ID, nil, "root")) + require.True(t, sm.HasEventSource(sess.ID)) + + streamCtx, cancelStream := context.WithCancel(ctx) + defer cancelStream() + var mu sync.Mutex + var received []any + streamDone := make(chan struct{}) + go func() { + defer close(streamDone) + sm.StreamEvents(streamCtx, sess.ID, nil, func(_ uint64, event any) { + mu.Lock() + defer mu.Unlock() + received = append(received, event) + }) + }() + + // Wait for the replay of the elicitation event so the stream is known to + // be actively connected (not merely about to start) before we delete. + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(received) == 1 + }, 2*time.Second, time.Millisecond) + + require.NoError(t, sm.DeleteSession(ctx, sess.ID)) + + select { + case <-streamDone: + case <-time.After(2 * time.Second): + t.Fatal("StreamEvents must return once the session is deleted; a no-op cancel on the lazily-created event log leaves connected /events streams blocked forever") + } + + mu.Lock() + defer mu.Unlock() + require.Len(t, received, 2, "expected the elicitation event followed by a terminal session_exited event") + exited, ok := received[1].(sessionExitedEvent) + require.True(t, ok, "expected sessionExitedEvent, got %T", received[1]) + assert.Equal(t, "session_exited", exited.Type) +} + +// TestReview_BatchDeleteSessionsClosesLazyElicitationEventLog is the batch +// variant of the above: BatchDeleteSessions goes through the same +// pe.cancel() call per session, so it shares the exact same bug and fix. +func TestReview_BatchDeleteSessionsClosesLazyElicitationEventLog(t *testing.T) { + t.Parallel() + + ctx := t.Context() + store := session.NewInMemorySessionStore() + sm := NewSessionManager(ctx, config.Sources{}, store, 0, &config.RuntimeConfig{}) + + sess := session.New() + require.NoError(t, store.AddSession(ctx, sess)) + + sm.sessionElicitationSink(sess.ID)(runtime.ElicitationRequest("need input", "form", nil, "", "eid-1", "", sess.ID, nil, "root")) + require.True(t, sm.HasEventSource(sess.ID)) + + streamCtx, cancelStream := context.WithCancel(ctx) + defer cancelStream() + var mu sync.Mutex + var received []any + streamDone := make(chan struct{}) + go func() { + defer close(streamDone) + sm.StreamEvents(streamCtx, sess.ID, nil, func(_ uint64, event any) { + mu.Lock() + defer mu.Unlock() + received = append(received, event) + }) + }() + + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return len(received) == 1 + }, 2*time.Second, time.Millisecond) + + deleted, failed := sm.BatchDeleteSessions(ctx, []string{sess.ID}) + assert.Equal(t, 1, deleted) + assert.Empty(t, failed) + + select { + case <-streamDone: + case <-time.After(2 * time.Second): + t.Fatal("StreamEvents must return once the session is batch-deleted; a no-op cancel on the lazily-created event log leaves connected /events streams blocked forever") + } + + mu.Lock() + defer mu.Unlock() + require.Len(t, received, 2, "expected the elicitation event followed by a terminal session_exited event") + exited, ok := received[1].(sessionExitedEvent) + require.True(t, ok, "expected sessionExitedEvent, got %T", received[1]) + assert.Equal(t, "session_exited", exited.Type) +} diff --git a/pkg/tui/compact_routing_test.go b/pkg/tui/compact_routing_test.go index 8872280019..a128b1ad51 100644 --- a/pkg/tui/compact_routing_test.go +++ b/pkg/tui/compact_routing_test.go @@ -47,7 +47,7 @@ func (stubRuntime) Run(context.Context, *session.Session) ([]session.Message, er return nil, nil } func (stubRuntime) Resume(context.Context, runtime.ResumeRequest) {} -func (stubRuntime) ResumeElicitation(context.Context, tools.ElicitationAction, map[string]any) error { +func (stubRuntime) ResumeElicitation(context.Context, tools.ElicitationAction, map[string]any, ...string) error { return nil } func (stubRuntime) SessionStore() session.Store { return nil } @@ -84,6 +84,7 @@ func (stubRuntime) AvailableModels(context.Context) []runtime.ModelChoice { retu func (stubRuntime) SupportsModelSwitching() bool { return false } func (stubRuntime) OnToolsChanged(func(runtime.Event)) {} func (stubRuntime) OnBackgroundEvent(func(runtime.Event)) {} +func (stubRuntime) OnElicitationRequest(func(runtime.Event)) {} func (stubRuntime) QueueStatus() runtime.QueueStatus { return runtime.QueueStatus{} } func (stubRuntime) TogglePause(context.Context) (bool, error) { return false, nil } func (stubRuntime) Close() error { return nil } diff --git a/pkg/tui/dialog/base.go b/pkg/tui/dialog/base.go index 40065d74a9..6665b1bc43 100644 --- a/pkg/tui/dialog/base.go +++ b/pkg/tui/dialog/base.go @@ -101,10 +101,10 @@ func ContentEndRow(dialogRow, dialogHeight int) int { } // CloseWithElicitationResponse returns a command that closes the dialog and sends an elicitation response. -func CloseWithElicitationResponse(action tools.ElicitationAction, content map[string]any) tea.Cmd { +func CloseWithElicitationResponse(action tools.ElicitationAction, content map[string]any, elicitationID string) tea.Cmd { return tea.Sequence( core.CmdHandler(CloseDialogMsg{}), - core.CmdHandler(messages.ElicitationResponseMsg{Action: action, Content: content}), + core.CmdHandler(messages.ElicitationResponseMsg{Action: action, Content: content, ElicitationID: elicitationID}), ) } diff --git a/pkg/tui/dialog/elicitation.go b/pkg/tui/dialog/elicitation.go index 3cc9f368eb..8ce0956e26 100644 --- a/pkg/tui/dialog/elicitation.go +++ b/pkg/tui/dialog/elicitation.go @@ -59,6 +59,7 @@ type ElicitationDialog struct { title string message string + elicitationID string fields []ElicitationField inputs []textinput.Model boolValues map[int]bool @@ -87,7 +88,7 @@ func (d *ElicitationDialog) hasFreeFormInput() bool { } // NewElicitationDialog creates a new elicitation dialog. -func NewElicitationDialog(message string, schema any, meta map[string]any) Dialog { +func NewElicitationDialog(message string, schema any, meta map[string]any, elicitationID string) Dialog { fields := parseElicitationSchema(schema) // Determine dialog title from meta, defaulting to "Question" @@ -99,13 +100,14 @@ func NewElicitationDialog(message string, schema any, meta map[string]any) Dialo } d := &ElicitationDialog{ - title: title, - message: message, - fields: fields, - inputs: make([]textinput.Model, len(fields)), - boolValues: make(map[int]bool), - enumIndexes: make(map[int]int), - fieldErrors: make(map[int]string), + title: title, + message: message, + elicitationID: elicitationID, + fields: fields, + inputs: make([]textinput.Model, len(fields)), + boolValues: make(map[int]bool), + enumIndexes: make(map[int]int), + fieldErrors: make(map[int]string), keyMap: elicitationKeyMap{ Up: key.NewBinding(key.WithKeys("up")), Down: key.NewBinding(key.WithKeys("down")), @@ -338,7 +340,7 @@ func (d *ElicitationDialog) isTextInputField() bool { } func (d *ElicitationDialog) close(action tools.ElicitationAction, content map[string]any) tea.Cmd { - return CloseWithElicitationResponse(action, content) + return CloseWithElicitationResponse(action, content, d.elicitationID) } // collectAndValidate validates all fields and returns the collected values. diff --git a/pkg/tui/dialog/elicitation_test.go b/pkg/tui/dialog/elicitation_test.go index 179e67d1fe..42ec09cd90 100644 --- a/pkg/tui/dialog/elicitation_test.go +++ b/pkg/tui/dialog/elicitation_test.go @@ -339,7 +339,7 @@ func TestNewElicitationDialog(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() - dialog := NewElicitationDialog(tt.message, tt.schema, tt.meta) + dialog := NewElicitationDialog(tt.message, tt.schema, tt.meta, "") require.NotNil(t, dialog) ed, ok := dialog.(*ElicitationDialog) @@ -559,7 +559,7 @@ func TestElicitationDialog_collectAndValidate(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() - dialog := NewElicitationDialog("test", tt.schema, nil).(*ElicitationDialog) + dialog := NewElicitationDialog("test", tt.schema, nil, "").(*ElicitationDialog) if tt.setupInputs != nil { tt.setupInputs(dialog) } @@ -587,7 +587,7 @@ func TestElicitationDialog_PasswordFieldMaskedAndUntrimmed(t *testing.T) { }, "required": []any{"password"}, } - dialog := NewElicitationDialog("sudo", schema, nil).(*ElicitationDialog) + dialog := NewElicitationDialog("sudo", schema, nil, "").(*ElicitationDialog) // The password input is masked, not echoed in clear text. assert.Equal(t, textinput.EchoPassword, dialog.inputs[0].EchoMode) @@ -615,7 +615,7 @@ func TestElicitationDialog_LongEnumScrolls(t *testing.T) { "enum": enumValues, } - dialog := NewElicitationDialog("Choose an option:", schema, nil).(*ElicitationDialog) + dialog := NewElicitationDialog("Choose an option:", schema, nil, "").(*ElicitationDialog) require.Len(t, dialog.fields, 1) require.Len(t, dialog.fields[0].EnumValues, 30) @@ -648,7 +648,7 @@ func TestElicitationDialog_FieldsBelowFold_AreReachable(t *testing.T) { } schema := map[string]any{"type": "object", "properties": props, "required": required} - dialog := NewElicitationDialog("Fill in the form", schema, nil).(*ElicitationDialog) + dialog := NewElicitationDialog("Fill in the form", schema, nil, "").(*ElicitationDialog) _, _ = dialog.Update(tea.WindowSizeMsg{Width: 100, Height: 18}) _ = dialog.View() @@ -680,7 +680,7 @@ func TestElicitationDialog_SmallContent_NoScrollbar(t *testing.T) { "required": []any{"name"}, } - dialog := NewElicitationDialog("Enter your name", schema, nil).(*ElicitationDialog) + dialog := NewElicitationDialog("Enter your name", schema, nil, "").(*ElicitationDialog) _, _ = dialog.Update(tea.WindowSizeMsg{Width: 100, Height: 30}) _ = dialog.View() @@ -707,7 +707,7 @@ func TestElicitationDialog_TypingRevealsBelowFoldField(t *testing.T) { "required": []any{"name"}, } - dialog := NewElicitationDialog(longMessage, schema, nil).(*ElicitationDialog) + dialog := NewElicitationDialog(longMessage, schema, nil, "").(*ElicitationDialog) _, _ = dialog.Update(tea.WindowSizeMsg{Width: 80, Height: 16}) _ = dialog.View() @@ -747,7 +747,7 @@ func TestElicitationDialog_OpensScrolledToTop(t *testing.T) { "enum": enumValues, } - dialog := NewElicitationDialog(longMessage, schema, nil).(*ElicitationDialog) + dialog := NewElicitationDialog(longMessage, schema, nil, "").(*ElicitationDialog) _, _ = dialog.Update(tea.WindowSizeMsg{Width: 80, Height: 18}) _ = dialog.View() @@ -769,7 +769,7 @@ func TestElicitationDialog_UserScrollUp_NotSnappedBack(t *testing.T) { } schema := map[string]any{"type": "string", "title": "Pick one", "enum": enumValues} - dialog := NewElicitationDialog(longMessage, schema, nil).(*ElicitationDialog) + dialog := NewElicitationDialog(longMessage, schema, nil, "").(*ElicitationDialog) _, _ = dialog.Update(tea.WindowSizeMsg{Width: 80, Height: 16}) _ = dialog.View() diff --git a/pkg/tui/dialog/manager_test.go b/pkg/tui/dialog/manager_test.go index c15fb5f92f..68e89fe474 100644 --- a/pkg/tui/dialog/manager_test.go +++ b/pkg/tui/dialog/manager_test.go @@ -32,7 +32,7 @@ func TestManagerBackgroundDialog(t *testing.T) { // Stack a background dialog (i.e. one carrying an originating event) on top. type fakeEvent struct{ id int } event := &fakeEvent{id: 42} - bg := NewElicitationDialog("Pick a value", nil, nil) + bg := NewElicitationDialog("Pick a value", nil, nil, "") mgr.handleOpen(OpenDialogMsg{ Model: bg, OriginatingEvent: event, diff --git a/pkg/tui/dialog/oauth_authorization.go b/pkg/tui/dialog/oauth_authorization.go index df0b1b3ecf..f3707d2847 100644 --- a/pkg/tui/dialog/oauth_authorization.go +++ b/pkg/tui/dialog/oauth_authorization.go @@ -18,18 +18,20 @@ type oauthAuthorizationDialog struct { ctx func() context.Context - serverURL string - app *app.App - keyMap ConfirmKeyMap + serverURL string + elicitationID string + app *app.App + keyMap ConfirmKeyMap } // NewOAuthAuthorizationDialog creates a new OAuth authorization confirmation dialog -func NewOAuthAuthorizationDialog(ctx context.Context, serverURL string, appInstance *app.App) Dialog { +func NewOAuthAuthorizationDialog(ctx context.Context, serverURL, elicitationID string, appInstance *app.App) Dialog { return &oauthAuthorizationDialog{ - ctx: func() context.Context { return context.WithoutCancel(ctx) }, - serverURL: serverURL, - app: appInstance, - keyMap: DefaultConfirmKeyMap(), + ctx: func() context.Context { return context.WithoutCancel(ctx) }, + serverURL: serverURL, + elicitationID: elicitationID, + app: appInstance, + keyMap: DefaultConfirmKeyMap(), } } @@ -52,11 +54,11 @@ func (d *oauthAuthorizationDialog) Update(msg tea.Msg) (layout.Model, tea.Cmd) { model, cmd, handled := HandleConfirmKeys(msg, d.keyMap, func() (layout.Model, tea.Cmd) { - _ = d.app.ResumeElicitation(d.ctx(), tools.ElicitationActionAccept, nil) + _ = d.app.ResumeElicitation(d.ctx(), tools.ElicitationActionAccept, nil, d.elicitationID) return d, core.CmdHandler(CloseDialogMsg{}) }, func() (layout.Model, tea.Cmd) { - _ = d.app.ResumeElicitation(d.ctx(), tools.ElicitationActionDecline, nil) + _ = d.app.ResumeElicitation(d.ctx(), tools.ElicitationActionDecline, nil, d.elicitationID) return d, core.CmdHandler(CloseDialogMsg{}) }, ) diff --git a/pkg/tui/dialog/url_elicitation.go b/pkg/tui/dialog/url_elicitation.go index 0c00c56396..dec734bb0a 100644 --- a/pkg/tui/dialog/url_elicitation.go +++ b/pkg/tui/dialog/url_elicitation.go @@ -20,21 +20,23 @@ type URLElicitationDialog struct { ctx func() context.Context - message string - url string - keyMap ConfirmKeyMap - escape key.Binding - openBrowser key.Binding + message string + url string + elicitationID string + keyMap ConfirmKeyMap + escape key.Binding + openBrowser key.Binding } // NewURLElicitationDialog creates a new URL elicitation dialog. -func NewURLElicitationDialog(ctx context.Context, message, url string) Dialog { +func NewURLElicitationDialog(ctx context.Context, message, url, elicitationID string) Dialog { return &URLElicitationDialog{ - ctx: func() context.Context { return context.WithoutCancel(ctx) }, - message: message, - url: url, - keyMap: DefaultConfirmKeyMap(), - escape: key.NewBinding(key.WithKeys("esc")), + ctx: func() context.Context { return context.WithoutCancel(ctx) }, + message: message, + url: url, + elicitationID: elicitationID, + keyMap: DefaultConfirmKeyMap(), + escape: key.NewBinding(key.WithKeys("esc")), openBrowser: key.NewBinding( key.WithKeys("o"), key.WithHelp("o", "open"), @@ -79,7 +81,7 @@ func (d *URLElicitationDialog) Update(msg tea.Msg) (layout.Model, tea.Cmd) { } func (d *URLElicitationDialog) respond(action tools.ElicitationAction) tea.Cmd { - return CloseWithElicitationResponse(action, nil) + return CloseWithElicitationResponse(action, nil, d.elicitationID) } func (d *URLElicitationDialog) openURLInBrowser() tea.Cmd { diff --git a/pkg/tui/dialog/url_elicitation_test.go b/pkg/tui/dialog/url_elicitation_test.go index 3991cc1012..4be4d33af1 100644 --- a/pkg/tui/dialog/url_elicitation_test.go +++ b/pkg/tui/dialog/url_elicitation_test.go @@ -30,7 +30,7 @@ func TestNewURLElicitationDialog(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() - dialog := NewURLElicitationDialog(t.Context(), tt.message, tt.url) + dialog := NewURLElicitationDialog(t.Context(), tt.message, tt.url, "") require.NotNil(t, dialog) ud, ok := dialog.(*URLElicitationDialog) @@ -44,7 +44,7 @@ func TestNewURLElicitationDialog(t *testing.T) { func TestURLElicitationDialog_View(t *testing.T) { t.Parallel() - dialog := NewURLElicitationDialog(t.Context(), "Please visit the URL", "https://example.com/callback").(*URLElicitationDialog) + dialog := NewURLElicitationDialog(t.Context(), "Please visit the URL", "https://example.com/callback", "").(*URLElicitationDialog) dialog.SetSize(100, 50) view := dialog.View() @@ -61,7 +61,7 @@ func TestURLElicitationDialog_View(t *testing.T) { func TestURLElicitationDialog_HasOpenKeyBinding(t *testing.T) { t.Parallel() - dialog := NewURLElicitationDialog(t.Context(), "Test", "https://example.com").(*URLElicitationDialog) + dialog := NewURLElicitationDialog(t.Context(), "Test", "https://example.com", "").(*URLElicitationDialog) // Verify the openBrowser key binding exists and is configured correctly require.NotNil(t, dialog.openBrowser) diff --git a/pkg/tui/handlers.go b/pkg/tui/handlers.go index d57d201087..fa4e5a2d9a 100644 --- a/pkg/tui/handlers.go +++ b/pkg/tui/handlers.go @@ -1151,8 +1151,8 @@ func (m *appModel) closeTranscriptCh() { } } -func (m *appModel) handleElicitationResponse(action tools.ElicitationAction, content map[string]any) (tea.Model, tea.Cmd) { - if err := m.application.ResumeElicitation(m.ctx(), action, content); err != nil { +func (m *appModel) handleElicitationResponse(action tools.ElicitationAction, content map[string]any, elicitationID string) (tea.Model, tea.Cmd) { + if err := m.application.ResumeElicitation(m.ctx(), action, content, elicitationID); err != nil { slog.Error("Failed to resume elicitation", "action", action, "error", err) return m, notification.ErrorCmd("Failed to complete server request: " + err.Error()) } diff --git a/pkg/tui/messages/mcp.go b/pkg/tui/messages/mcp.go index 3d18624a06..231c5bd20d 100644 --- a/pkg/tui/messages/mcp.go +++ b/pkg/tui/messages/mcp.go @@ -20,5 +20,9 @@ type ( ElicitationResponseMsg struct { Action tools.ElicitationAction Content map[string]any + // ElicitationID correlates this response with the specific request it + // answers; empty for dialogs built before the ID was plumbed through + // (falls back to the runtime's sole-pending-request behavior). + ElicitationID string } ) diff --git a/pkg/tui/page/chat/queue_test.go b/pkg/tui/page/chat/queue_test.go index 716b0cdbb3..483c78c5fd 100644 --- a/pkg/tui/page/chat/queue_test.go +++ b/pkg/tui/page/chat/queue_test.go @@ -69,7 +69,7 @@ func (queueTestRuntime) Run(context.Context, *session.Session) ([]session.Messag return nil, nil } func (queueTestRuntime) Resume(context.Context, runtime.ResumeRequest) {} -func (queueTestRuntime) ResumeElicitation(context.Context, tools.ElicitationAction, map[string]any) error { +func (queueTestRuntime) ResumeElicitation(context.Context, tools.ElicitationAction, map[string]any, ...string) error { return nil } func (queueTestRuntime) SessionStore() session.Store { return nil } @@ -110,6 +110,7 @@ func (queueTestRuntime) AvailableModels(context.Context) []runtime.ModelChoice { func (queueTestRuntime) SupportsModelSwitching() bool { return false } func (queueTestRuntime) OnToolsChanged(func(runtime.Event)) {} func (queueTestRuntime) OnBackgroundEvent(func(runtime.Event)) {} +func (queueTestRuntime) OnElicitationRequest(func(runtime.Event)) {} func (queueTestRuntime) QueueStatus() runtime.QueueStatus { return runtime.QueueStatus{} } func (queueTestRuntime) TogglePause(context.Context) (bool, error) { return false, nil } func (queueTestRuntime) Close() error { return nil } diff --git a/pkg/tui/page/chat/runtime_events.go b/pkg/tui/page/chat/runtime_events.go index cb2d4f1857..d3f1d41ee4 100644 --- a/pkg/tui/page/chat/runtime_events.go +++ b/pkg/tui/page/chat/runtime_events.go @@ -450,7 +450,7 @@ func (p *chatPage) handleElicitationRequest(msg *runtime.ElicitationRequestEvent serverURL = url } dialogCmd := core.CmdHandler(dialog.OpenDialogMsg{ - Model: dialog.NewOAuthAuthorizationDialog(p.ctx(), serverURL, p.app), + Model: dialog.NewOAuthAuthorizationDialog(p.ctx(), serverURL, msg.ElicitationID, p.app), OriginatingEvent: msg, }) return tea.Batch(spinnerCmd, dialogCmd) @@ -462,7 +462,7 @@ func (p *chatPage) handleElicitationRequest(msg *runtime.ElicitationRequestEvent case "url": // URL-based elicitation - show URL dialog dialogCmd := core.CmdHandler(dialog.OpenDialogMsg{ - Model: dialog.NewURLElicitationDialog(p.ctx(), msg.Message, msg.URL), + Model: dialog.NewURLElicitationDialog(p.ctx(), msg.Message, msg.URL, msg.ElicitationID), OriginatingEvent: msg, }) return tea.Batch(spinnerCmd, dialogCmd) @@ -470,7 +470,7 @@ func (p *chatPage) handleElicitationRequest(msg *runtime.ElicitationRequestEvent default: // Form-based elicitation (default) - show form dialog dialogCmd := core.CmdHandler(dialog.OpenDialogMsg{ - Model: dialog.NewElicitationDialog(msg.Message, msg.Schema, msg.Meta), + Model: dialog.NewElicitationDialog(msg.Message, msg.Schema, msg.Meta, msg.ElicitationID), OriginatingEvent: msg, }) return tea.Batch(spinnerCmd, dialogCmd) diff --git a/pkg/tui/service/supervisor/supervisor.go b/pkg/tui/service/supervisor/supervisor.go index 379ca7fdcf..16b0e5ce79 100644 --- a/pkg/tui/service/supervisor/supervisor.go +++ b/pkg/tui/service/supervisor/supervisor.go @@ -19,15 +19,21 @@ import ( // SessionRunner represents a running session. type SessionRunner struct { - ID string - App *app.App - WorkingDir string - Title string - IsRunning bool // True when stream is active - NeedsAttn bool // True when user attention is needed - PendingEvent tea.Msg // Event that triggered attention (for replay on tab switch) - cancel context.CancelFunc - cleanup func() + ID string + App *app.App + WorkingDir string + Title string + IsRunning bool // True when stream is active + NeedsAttn bool // True when user attention is needed + // PendingEvents queues attention events (tool confirmation, max + // iterations, elicitation) that arrived while this tab was inactive, in + // arrival order, for replay when the user switches to it. A single slot + // used to overwrite an earlier event with a later one, silently dropping + // it (#3584) — e.g. two concurrent background-job elicitations on the + // same unfocused tab would leave only the second visible. + PendingEvents []tea.Msg + cancel context.CancelFunc + cleanup func() } // SessionSpawner is a function that creates new sessions. @@ -179,15 +185,28 @@ func (s *Supervisor) handleRuntimeEvent(sessionID string, msg tea.Msg) { case *runtime.StreamStartedEvent: if isTopLevelStream(runner.ID, ev.SessionID) { runner.IsRunning = true - runner.PendingEvent = nil // New top-level stream supersedes any stale pending event + // A new top-level turn supersedes any stale attention events raised + // by ITS OWN previous turn, but must not discard a still-live, + // unanswered elicitation from a detached background job + // (run_background_agent outlives the turn boundary via + // context.WithoutCancel) — that job's waiter goroutine is still + // blocked and would be orphaned if its prompt vanished from the + // queue (#3584 review item 4). + runner.PendingEvents = retainDetachedElicitations(runner.ID, runner.PendingEvents) + runner.NeedsAttn = len(runner.PendingEvents) > 0 s.notifyTabsUpdated() } case *runtime.StreamStoppedEvent: if isTopLevelStream(runner.ID, ev.SessionID) { runner.IsRunning = false - runner.PendingEvent = nil // Clear any pending attention event since the top-level stream ended - runner.NeedsAttn = false + // Same rule as StreamStarted above: only this runner's own + // top-level attention events are moot now that its stream ended. + // A detached background job's live elicitation must survive the + // foreground stream's stop so its waiter isn't orphaned (#3584 + // review item 4). + runner.PendingEvents = retainDetachedElicitations(runner.ID, runner.PendingEvents) + runner.NeedsAttn = len(runner.PendingEvents) > 0 s.notifyTabsUpdated() } @@ -199,7 +218,7 @@ func (s *Supervisor) handleRuntimeEvent(sessionID string, msg tea.Msg) { // These require user attention if sessionID != s.activeID { runner.NeedsAttn = true - runner.PendingEvent = msg + runner.PendingEvents = append(runner.PendingEvents, msg) s.notifyTabsUpdated() // Ring the terminal bell to alert the user if p := s.program; p != nil { @@ -209,6 +228,21 @@ func (s *Supervisor) handleRuntimeEvent(sessionID string, msg tea.Msg) { } } +// retainDetachedElicitations filters pending to keep only +// ElicitationRequestEvents raised by a session other than runnerID — i.e. a +// detached background job's sub-session, whose elicitation waiter is still +// blocked awaiting a response regardless of what the runner's own top-level +// stream is doing. Everything else (ToolCallConfirmation, MaxIterationsReached, +// and elicitations belonging to runnerID's own top-level stream) is dropped: +// those are inherently scoped to the stream that just started or stopped, so +// they are genuinely moot once it does. +func retainDetachedElicitations(runnerID string, pending []tea.Msg) []tea.Msg { + return slices.DeleteFunc(pending, func(msg tea.Msg) bool { + elic, ok := msg.(*runtime.ElicitationRequestEvent) + return !ok || isTopLevelStream(runnerID, elic.SessionID) + }) +} + // notifyTabsUpdated sends a tabs updated message (must be called with lock held). func (s *Supervisor) notifyTabsUpdated() { p := s.program @@ -274,24 +308,25 @@ func (s *Supervisor) SwitchTo(sessionID string) *SessionRunner { return runner } -// ConsumePendingEvent returns and clears the pending event for the given session. -// Returns nil if no event is pending. +// ConsumePendingEvent pops and returns the oldest pending event for the given +// session (FIFO). Returns nil if none is pending. func (s *Supervisor) ConsumePendingEvent(sessionID string) tea.Msg { s.mu.Lock() defer s.mu.Unlock() runner, ok := s.runners[sessionID] - if !ok || runner.PendingEvent == nil { + if !ok || len(runner.PendingEvents) == 0 { return nil } - event := runner.PendingEvent - runner.PendingEvent = nil + event := runner.PendingEvents[0] + runner.PendingEvents = runner.PendingEvents[1:] return event } -// SetPendingEvent stores an attention event for the given session so it can -// be replayed when the user later switches to that tab. Used to re-stash a +// SetPendingEvent re-queues an attention event at the FRONT of the given +// session's pending queue, ahead of anything queued behind it, so it can be +// replayed when the user later switches to that tab. Used to re-stash a // background dialog's originating event when the user navigates away from // the tab that opened it. // @@ -303,7 +338,7 @@ func (s *Supervisor) SetPendingEvent(sessionID string, event tea.Msg) { defer s.mu.Unlock() if runner, ok := s.runners[sessionID]; ok { - runner.PendingEvent = event + runner.PendingEvents = append([]tea.Msg{event}, runner.PendingEvents...) } } diff --git a/pkg/tui/service/supervisor/supervisor_test.go b/pkg/tui/service/supervisor/supervisor_test.go index e32af5e7f6..80bcc9f207 100644 --- a/pkg/tui/service/supervisor/supervisor_test.go +++ b/pkg/tui/service/supervisor/supervisor_test.go @@ -3,6 +3,7 @@ package supervisor import ( "testing" + tea "charm.land/bubbletea/v2" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -118,12 +119,36 @@ func TestSetPendingEvent_RoundTrip(t *testing.T) { s.SetPendingEvent("A", event) - assert.Equal(t, event, s.runners["A"].PendingEvent, "event is stored on the runner") + assert.Equal(t, []tea.Msg{event}, s.runners["A"].PendingEvents, "event is stored on the runner") assert.False(t, s.runners["A"].NeedsAttn, "SetPendingEvent must NOT raise NeedsAttn (the user is already aware)") got := s.ConsumePendingEvent("A") assert.Equal(t, event, got) - assert.Nil(t, s.runners["A"].PendingEvent, "event is cleared after consumption") + assert.Empty(t, s.runners["A"].PendingEvents, "event is cleared after consumption") +} + +// TestSetPendingEvent_Queue verifies that multiple events queued for the same +// inactive session are replayed in FIFO order and that SetPendingEvent +// re-queues at the front, ahead of anything queued behind it (#3584). +func TestSetPendingEvent_Queue(t *testing.T) { + t.Parallel() + s := newTestSupervisor([]string{"A"}, "B") + + type fakeEvent struct{ id int } + first := &fakeEvent{id: 1} + second := &fakeEvent{id: 2} + + s.runners["A"].PendingEvents = []tea.Msg{first, second} + + // Re-stash a third event (e.g. the live dialog instance for the event the + // user was looking at) ahead of the two already queued. + stashed := &fakeEvent{id: 0} + s.SetPendingEvent("A", stashed) + + assert.Equal(t, stashed, s.ConsumePendingEvent("A")) + assert.Equal(t, first, s.ConsumePendingEvent("A")) + assert.Equal(t, second, s.ConsumePendingEvent("A")) + assert.Nil(t, s.ConsumePendingEvent("A"), "queue is drained") } // TestSetPendingEvent_UnknownSession is a no-op (and must not panic). @@ -133,7 +158,7 @@ func TestSetPendingEvent_UnknownSession(t *testing.T) { s.SetPendingEvent("does-not-exist", "payload") - assert.Nil(t, s.runners["A"].PendingEvent, "unrelated runner is untouched") + assert.Empty(t, s.runners["A"].PendingEvents, "unrelated runner is untouched") } // --- #3217: session-aware stream lifecycle tests --- @@ -166,8 +191,8 @@ func TestStreamStarted_SubSessionDoesNotDropPendingEvent(t *testing.T) { t.Parallel() s := newTestSupervisor([]string{"sess-A", "sess-B"}, "sess-B") // sess-A is background - elicitation := runtime.ElicitationRequest("confirm?", "form", nil, "", "eid-1", nil, "agent") - s.runners["sess-A"].PendingEvent = elicitation + elicitation := runtime.ElicitationRequest("confirm?", "form", nil, "", "eid-1", "", "", nil, "agent") + s.runners["sess-A"].PendingEvents = []tea.Msg{elicitation} s.runners["sess-A"].NeedsAttn = true s.runners["sess-A"].IsRunning = true // already running a top-level turn @@ -177,7 +202,7 @@ func TestStreamStarted_SubSessionDoesNotDropPendingEvent(t *testing.T) { SessionID: "child-xyz", }) - require.NotNil(t, s.runners["sess-A"].PendingEvent, + require.NotEmpty(t, s.runners["sess-A"].PendingEvents, "nested StreamStarted must NOT clear the parent's pending elicitation") assert.True(t, s.runners["sess-A"].NeedsAttn, "nested StreamStarted must NOT clear NeedsAttn") @@ -192,8 +217,8 @@ func TestStreamStopped_SubSessionDoesNotDropPendingEvent(t *testing.T) { t.Parallel() s := newTestSupervisor([]string{"sess-A", "sess-B"}, "sess-B") - elicitation := runtime.ElicitationRequest("confirm?", "form", nil, "", "eid-2", nil, "agent") - s.runners["sess-A"].PendingEvent = elicitation + elicitation := runtime.ElicitationRequest("confirm?", "form", nil, "", "eid-2", "", "", nil, "agent") + s.runners["sess-A"].PendingEvents = []tea.Msg{elicitation} s.runners["sess-A"].NeedsAttn = true s.runners["sess-A"].IsRunning = true @@ -203,7 +228,7 @@ func TestStreamStopped_SubSessionDoesNotDropPendingEvent(t *testing.T) { SessionID: "child-xyz", }) - require.NotNil(t, s.runners["sess-A"].PendingEvent, + require.NotEmpty(t, s.runners["sess-A"].PendingEvents, "nested StreamStopped must NOT clear the parent's pending elicitation") assert.True(t, s.runners["sess-A"].NeedsAttn, "nested StreamStopped must NOT clear NeedsAttn") @@ -218,9 +243,9 @@ func TestStreamStarted_TopLevelSupersedesStalePending(t *testing.T) { t.Parallel() s := newTestSupervisor([]string{"sess-A"}, "sess-A") - s.runners["sess-A"].PendingEvent = runtime.ElicitationRequest( - "old?", "form", nil, "", "eid-stale", nil, "agent", - ) + s.runners["sess-A"].PendingEvents = []tea.Msg{runtime.ElicitationRequest( + "old?", "form", nil, "", "eid-stale", "", "", nil, "agent", + )} s.runners["sess-A"].IsRunning = false // New top-level turn starts. @@ -229,7 +254,7 @@ func TestStreamStarted_TopLevelSupersedesStalePending(t *testing.T) { SessionID: "sess-A", }) - assert.Nil(t, s.runners["sess-A"].PendingEvent, + assert.Empty(t, s.runners["sess-A"].PendingEvents, "top-level StreamStarted must supersede any stale pending event") assert.True(t, s.runners["sess-A"].IsRunning, "top-level StreamStarted must set IsRunning") @@ -245,7 +270,7 @@ func TestStreamStopped_TopLevelClearsPendingAndNeedsAttn(t *testing.T) { }{ { name: "elicitation pending", - pending: runtime.ElicitationRequest("q?", "form", nil, "", "eid-3", nil, "agent"), + pending: runtime.ElicitationRequest("q?", "form", nil, "", "eid-3", "", "", nil, "agent"), }, { name: "tool confirmation pending", @@ -259,7 +284,7 @@ func TestStreamStopped_TopLevelClearsPendingAndNeedsAttn(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { s := newTestSupervisor([]string{"sess-A"}, "sess-B") - s.runners["sess-A"].PendingEvent = tc.pending + s.runners["sess-A"].PendingEvents = []tea.Msg{tc.pending} s.runners["sess-A"].NeedsAttn = true s.runners["sess-A"].IsRunning = true @@ -268,8 +293,8 @@ func TestStreamStopped_TopLevelClearsPendingAndNeedsAttn(t *testing.T) { SessionID: "sess-A", }) - assert.Nil(t, s.runners["sess-A"].PendingEvent, - "top-level StreamStopped must clear PendingEvent") + assert.Empty(t, s.runners["sess-A"].PendingEvents, + "top-level StreamStopped must clear PendingEvents") assert.False(t, s.runners["sess-A"].NeedsAttn, "top-level StreamStopped must clear NeedsAttn") assert.False(t, s.runners["sess-A"].IsRunning, @@ -285,9 +310,9 @@ func TestStreamStarted_EmptySessionID_TreatedAsTopLevel(t *testing.T) { t.Parallel() s := newTestSupervisor([]string{"sess-A"}, "sess-A") - s.runners["sess-A"].PendingEvent = runtime.ElicitationRequest( - "old?", "form", nil, "", "eid-old", nil, "agent", - ) + s.runners["sess-A"].PendingEvents = []tea.Msg{runtime.ElicitationRequest( + "old?", "form", nil, "", "eid-old", "", "", nil, "agent", + )} // Emitter omits SessionID (empty string). s.handleRuntimeEvent("sess-A", &runtime.StreamStartedEvent{ @@ -295,7 +320,90 @@ func TestStreamStarted_EmptySessionID_TreatedAsTopLevel(t *testing.T) { SessionID: "", }) - assert.Nil(t, s.runners["sess-A"].PendingEvent, + assert.Empty(t, s.runners["sess-A"].PendingEvents, "empty SessionID must be treated as top-level and supersede stale pending event") assert.True(t, s.runners["sess-A"].IsRunning) } + +// --- #3584 review item 4: foreground stream stop must not discard a +// still-live detached background job's elicitation --- + +// TestStreamStopped_TopLevel_PreservesDetachedBackgroundElicitation is the +// regression test for review item 4: a background job started via +// run_background_agent outlives its parent's top-level stream +// (context.WithoutCancel), so its own live, unanswered elicitation can still +// be queued on this runner when the FOREGROUND stream stops. The old +// unconditional `PendingEvents = nil` wiped it out from under the background +// job's still-blocked waiter goroutine. Only the runner's OWN top-level +// attention events (here, none) are moot; the background elicitation +// (SessionID "bg-child", distinct from the runner's own "sess-A") must +// survive. +func TestStreamStopped_TopLevel_PreservesDetachedBackgroundElicitation(t *testing.T) { + t.Parallel() + s := newTestSupervisor([]string{"sess-A"}, "sess-B") // sess-A inactive + + bgElicitation := runtime.ElicitationRequest("bg needs input", "form", nil, "", "eid-bg", "", "bg-child", nil, "agent") + s.runners["sess-A"].PendingEvents = []tea.Msg{bgElicitation} + s.runners["sess-A"].NeedsAttn = true + s.runners["sess-A"].IsRunning = true + + // The foreground (top-level) stream for sess-A stops. + s.handleRuntimeEvent("sess-A", &runtime.StreamStoppedEvent{ + Type: "stream_stopped", + SessionID: "sess-A", + }) + + assert.False(t, s.runners["sess-A"].IsRunning, "the foreground stream itself must still be marked stopped") + require.Equal(t, []tea.Msg{bgElicitation}, s.runners["sess-A"].PendingEvents, + "a detached background job's live elicitation must survive the foreground stream's stop") + assert.True(t, s.runners["sess-A"].NeedsAttn, + "NeedsAttn must stay true while a background elicitation is still queued") +} + +// TestStreamStopped_TopLevel_MixedPending_OnlyForegroundEventsCleared covers +// the more realistic mixed case: a foreground-owned tool confirmation and a +// background job's elicitation are both queued. Stopping the foreground +// stream must clear only the foreground-scoped entry. +func TestStreamStopped_TopLevel_MixedPending_OnlyForegroundEventsCleared(t *testing.T) { + t.Parallel() + s := newTestSupervisor([]string{"sess-A"}, "sess-B") + + foregroundConfirmation := runtime.ToolCallConfirmation(tools.ToolCall{}, tools.Tool{}, "agent", nil) + bgElicitation := runtime.ElicitationRequest("bg needs input", "form", nil, "", "eid-bg", "", "bg-child", nil, "agent") + s.runners["sess-A"].PendingEvents = []tea.Msg{foregroundConfirmation, bgElicitation} + s.runners["sess-A"].NeedsAttn = true + s.runners["sess-A"].IsRunning = true + + s.handleRuntimeEvent("sess-A", &runtime.StreamStoppedEvent{ + Type: "stream_stopped", + SessionID: "sess-A", + }) + + require.Equal(t, []tea.Msg{bgElicitation}, s.runners["sess-A"].PendingEvents, + "only the foreground-scoped tool confirmation must be dropped; the background elicitation stays queued") + assert.True(t, s.runners["sess-A"].NeedsAttn) +} + +// TestStreamStarted_TopLevel_PreservesDetachedBackgroundElicitation mirrors +// the StreamStopped case for a NEW top-level turn starting on the same tab +// while a background job from a previous turn is still live: starting a new +// foreground turn must not orphan the background job's queued elicitation +// either. +func TestStreamStarted_TopLevel_PreservesDetachedBackgroundElicitation(t *testing.T) { + t.Parallel() + s := newTestSupervisor([]string{"sess-A"}, "sess-B") + + bgElicitation := runtime.ElicitationRequest("bg needs input", "form", nil, "", "eid-bg2", "", "bg-child-2", nil, "agent") + s.runners["sess-A"].PendingEvents = []tea.Msg{bgElicitation} + s.runners["sess-A"].NeedsAttn = true + + s.handleRuntimeEvent("sess-A", &runtime.StreamStartedEvent{ + Type: "stream_started", + SessionID: "sess-A", + }) + + assert.True(t, s.runners["sess-A"].IsRunning) + require.Equal(t, []tea.Msg{bgElicitation}, s.runners["sess-A"].PendingEvents, + "a new top-level turn must not discard a still-live detached background elicitation") + assert.True(t, s.runners["sess-A"].NeedsAttn) +} diff --git a/pkg/tui/tui.go b/pkg/tui/tui.go index 4b608ef5f6..c249306cfd 100644 --- a/pkg/tui/tui.go +++ b/pkg/tui/tui.go @@ -1309,7 +1309,7 @@ func (m *appModel) update(msg tea.Msg) (tea.Model, tea.Cmd) { // --- Elicitation --- case messages.ElicitationResponseMsg: - return m.handleElicitationResponse(msg.Action, msg.Content) + return m.handleElicitationResponse(msg.Action, msg.Content, msg.ElicitationID) // --- Errors --- @@ -1802,42 +1802,76 @@ func (m *appModel) applySidebarCollapsed(sessionID string) tea.Cmd { return m.resizeAll() } -// replayPendingEvent checks if a session has a pending attention event (e.g. tool confirmation, -// max iterations, elicitation) that was received while the tab was inactive. -// If found, it opens the appropriate dialog. The event was already processed by the chat page -// (updating the message list), but the dialog command was discarded for inactive sessions. +// replayPendingEvent checks if a session has pending attention events (e.g. +// tool confirmation, max iterations, elicitation) that were received while +// the tab was inactive. Every queued event is replayed, in arrival order, so +// concurrent attention events (e.g. two background-job elicitations) all +// reopen as stacked dialogs instead of only the most recent one (#3584). Each +// event was already processed by the chat page (updating the message list), +// but the dialog command was discarded for inactive sessions. // // If a stashed dialog instance is available for this session and its -// associated event still matches the pending one, the same instance is +// associated event still matches the first pending one, the same instance is // re-opened so any in-progress input survives the round trip (issue #2770). // Otherwise the stash is discarded and a fresh dialog is built. func (m *appModel) replayPendingEvent(sessionID string) tea.Cmd { - pendingEvent := m.supervisor.ConsumePendingEvent(sessionID) - if pendingEvent == nil { - // No pending event: any stash is stale (e.g. the agent finished). - delete(m.stashedDialogs, sessionID) - return nil - } - sessionState, ok := m.sessionStates[sessionID] if !ok { delete(m.stashedDialogs, sessionID) return nil } - // If we stashed the live dialog instance when leaving this tab and the - // pending event hasn't changed, re-open the same instance so the user's - // in-progress input is preserved. - if stash, ok := m.stashedDialogs[sessionID]; ok { - delete(m.stashedDialogs, sessionID) - if stash.event == pendingEvent && stash.dialog != nil { - return core.CmdHandler(dialog.OpenDialogMsg{ - Model: stash.dialog, - OriginatingEvent: pendingEvent, - }) + var cmds []tea.Cmd + for first := true; ; first = false { + pendingEvent := m.supervisor.ConsumePendingEvent(sessionID) + if pendingEvent == nil { + if first { + // No pending event at all: any stash is stale (e.g. the agent finished). + delete(m.stashedDialogs, sessionID) + } + break + } + + // Only the first (oldest) event can match a stashed live dialog + // instance: the stash holds exactly the one dialog that was on + // screen when the user left the tab. + if first { + if stash, ok := m.stashedDialogs[sessionID]; ok { + delete(m.stashedDialogs, sessionID) + if stash.event == pendingEvent && stash.dialog != nil { + cmds = append(cmds, core.CmdHandler(dialog.OpenDialogMsg{ + Model: stash.dialog, + OriginatingEvent: pendingEvent, + })) + continue + } + } + } + + if cmd := m.dialogCmdForPendingEvent(pendingEvent, sessionState); cmd != nil { + cmds = append(cmds, cmd) } } + if len(cmds) == 0 { + return nil + } + // tea.Sequence (not tea.Batch) is required for deterministic ordering: + // tea.Batch runs commands concurrently with no ordering guarantee on + // which resulting Msg reaches Update first, so two concurrent attention + // events (e.g. two background-job elicitations queued while this tab was + // inactive) could stack in a random order on every replay even though + // they were popped off the FIFO queue above in arrival order (#3584 + // should-fix: FIFO replay). tea.Sequence guarantees each OpenDialogMsg is + // delivered to Update in the order the commands were built, so the + // dialog stack's bottom-to-top order matches arrival order every time. + return tea.Sequence(cmds...) +} + +// dialogCmdForPendingEvent builds the OpenDialogMsg command for a single +// replayed attention event. Shared by every event replayPendingEvent pops off +// the queue after the first (stash-eligible) one. +func (m *appModel) dialogCmdForPendingEvent(pendingEvent tea.Msg, sessionState *service.SessionState) tea.Cmd { switch ev := pendingEvent.(type) { case *runtime.ToolCallConfirmationEvent: return core.CmdHandler(dialog.OpenDialogMsg{ @@ -1868,7 +1902,7 @@ func (m *appModel) replayElicitationEvent(ev *runtime.ElicitationRequestEvent) t serverURL = url } return core.CmdHandler(dialog.OpenDialogMsg{ - Model: dialog.NewOAuthAuthorizationDialog(m.ctx(), serverURL, m.application), + Model: dialog.NewOAuthAuthorizationDialog(m.ctx(), serverURL, ev.ElicitationID, m.application), OriginatingEvent: ev, }) } @@ -1877,12 +1911,12 @@ func (m *appModel) replayElicitationEvent(ev *runtime.ElicitationRequestEvent) t switch ev.Mode { case "url": return core.CmdHandler(dialog.OpenDialogMsg{ - Model: dialog.NewURLElicitationDialog(m.ctx(), ev.Message, ev.URL), + Model: dialog.NewURLElicitationDialog(m.ctx(), ev.Message, ev.URL, ev.ElicitationID), OriginatingEvent: ev, }) default: return core.CmdHandler(dialog.OpenDialogMsg{ - Model: dialog.NewElicitationDialog(ev.Message, ev.Schema, ev.Meta), + Model: dialog.NewElicitationDialog(ev.Message, ev.Schema, ev.Meta, ev.ElicitationID), OriginatingEvent: ev, }) } diff --git a/pkg/tui/tui_stash_dialog_test.go b/pkg/tui/tui_stash_dialog_test.go index 46217e1290..88a847bf96 100644 --- a/pkg/tui/tui_stash_dialog_test.go +++ b/pkg/tui/tui_stash_dialog_test.go @@ -1,6 +1,7 @@ package tui import ( + "reflect" "testing" tea "charm.land/bubbletea/v2" @@ -170,3 +171,85 @@ func TestReplayPendingEvent_NoPendingEvent_ClearsStash(t *testing.T) { _, stillStashed := m.stashedDialogs[sessionID] assert.False(t, stillStashed, "orphaned stash must be cleared") } + +// drainInOrder unwraps a tea.Cmd exactly the way bubbletea's real +// Program.execSequenceMsg does (see charm.land/bubbletea/v2 tea.go): a +// sequenceMsg (produced by tea.Sequence) is drained by executing each cmd +// ONE AT A TIME, IN ORDER, recursing into any nested sequence; anything else +// is a leaf Msg. sequenceMsg is an unexported `[]tea.Cmd`, so reflection is +// used to recognize its shape without depending on bubbletea internals by +// name — but a slice-of-Cmd shape alone is not enough to tell it apart from +// the exported tea.BatchMsg (also `[]tea.Cmd`, produced by tea.Batch and run +// CONCURRENTLY with no ordering guarantee by the real Program). Treating +// both shapes the same here would make this helper — and the FIFO-order test +// below — pass identically whether production used tea.Sequence or +// tea.Batch, silently defeating the regression test it exists for. So a +// tea.BatchMsg is rejected outright, and only a value whose concrete type is +// bubbletea's private sequenceMsg is recursed into. +func drainInOrder(t *testing.T, cmd tea.Cmd) []tea.Msg { + t.Helper() + if cmd == nil { + return nil + } + msg := cmd() + if msg == nil { + return []tea.Msg{msg} + } + rt := reflect.TypeOf(msg) + if rt == reflect.TypeFor[tea.BatchMsg]() { + t.Fatalf("drainInOrder got a tea.BatchMsg: production must batch replayed dialogs with tea.Sequence (ordered), not tea.Batch (concurrent, unordered)") + } + if rt.Kind() == reflect.Slice && rt.Elem() == reflect.TypeFor[tea.Cmd]() && rt.Name() == "sequenceMsg" { + v := reflect.ValueOf(msg) + var out []tea.Msg + for i := range v.Len() { + sub, _ := v.Index(i).Interface().(tea.Cmd) + out = append(out, drainInOrder(t, sub)...) + } + return out + } + return []tea.Msg{msg} +} + +// TestReplayPendingEvent_ReplaysConcurrentElicitationsInFIFOOrder is the +// should-fix regression test for deterministic replay ordering: two +// concurrent background-job elicitations queued on the same inactive tab +// must reopen as dialogs in the order they arrived. Before the fix, +// replayPendingEvent batched the OpenDialogMsg commands with tea.Batch, +// which bubbletea's real Program executes CONCURRENTLY with no ordering +// guarantee on which resulting Msg reaches Update first (see +// Program.execBatchMsg) — so which prompt ended up on top of the dialog +// stack was nondeterministic. This drains the actual tea.Cmd graph the way +// the real Program does (see drainInOrder) instead of only inspecting the +// internal queue slice order, so it would catch a regression back to +// tea.Batch. +func TestReplayPendingEvent_ReplaysConcurrentElicitationsInFIFOOrder(t *testing.T) { + t.Parallel() + + const sessionID = "session-A" + + m, _ := newTestModel(t) + m.supervisor = supervisor.New(nil) + m.supervisor.AddSession(t.Context(), nil, &session.Session{ID: sessionID}, "/tmp", nil) + m.sessionStates[sessionID] = service.NewSessionState(&session.Session{ID: sessionID}) + + first := &runtime.ElicitationRequestEvent{Message: "worker1 needs input", ElicitationID: "e1"} + second := &runtime.ElicitationRequestEvent{Message: "worker2 needs input", ElicitationID: "e2"} + runner := m.supervisor.GetRunner(sessionID) + require.NotNil(t, runner) + runner.PendingEvents = []tea.Msg{first, second} + + cmd := m.replayPendingEvent(sessionID) + require.NotNil(t, cmd) + + msgs := drainInOrder(t, cmd) + require.Len(t, msgs, 2, "both queued elicitations must be replayed") + + open1, ok := msgs[0].(dialog.OpenDialogMsg) + require.True(t, ok, "expected dialog.OpenDialogMsg, got %T", msgs[0]) + open2, ok := msgs[1].(dialog.OpenDialogMsg) + require.True(t, ok, "expected dialog.OpenDialogMsg, got %T", msgs[1]) + + assert.Same(t, first, open1.OriginatingEvent, "the first-queued elicitation must be replayed first") + assert.Same(t, second, open2.OriginatingEvent, "the second-queued elicitation must be replayed second, not first") +}