diff --git a/internal/interfaces/api/sse.go b/internal/interfaces/api/sse.go index 8fbf9cc1..f1a308c2 100644 --- a/internal/interfaces/api/sse.go +++ b/internal/interfaces/api/sse.go @@ -23,6 +23,7 @@ type sseEventData struct { Status string `json:"status,omitempty"` Step string `json:"step,omitempty"` Error string `json:"error,omitempty"` + Prompt string `json:"prompt,omitempty"` } func eventDataFromEvent(ev ports.Event) sseEventData { //nolint:gocritic // Event is the public facade value type @@ -55,14 +56,17 @@ func eventDataFromEvent(ev ports.Event) sseEventData { //nolint:gocritic // Even data.Status = "cancelled" } } + case ports.EventInputRequired: + if payload, ok := ev.Payload.(*ports.EnrichedInputRequest); ok && payload != nil { + data.Prompt = payload.Prompt + } case ports.EventKindUnknown, ports.EventRunStarted, ports.EventRunCompleted, ports.EventMessageUser, ports.EventMessageAssistant, ports.EventToolCall, - ports.EventToolResult, - ports.EventInputRequired: + ports.EventToolResult: } return data diff --git a/internal/interfaces/api/sse_test.go b/internal/interfaces/api/sse_test.go index 8550347a..ba7c3aa6 100644 --- a/internal/interfaces/api/sse_test.go +++ b/internal/interfaces/api/sse_test.go @@ -131,6 +131,22 @@ func TestSSEStream_SentDataTypeIsRegistered(t *testing.T) { "an unregistered type makes huma log 'unknown event type' and dump a stack on every frame", sentType) } +func TestSSEEventDataFromEvent_InputRequiredIncludesPrompt(t *testing.T) { + data := eventDataFromEvent(ports.Event{ + Seq: 7, + Kind: ports.EventInputRequired, + RunID: "run-x", + Payload: &ports.EnrichedInputRequest{ + Prompt: "> ", + }, + }) + + assert.Equal(t, "input.required", data.Kind) + assert.Equal(t, "run-x", data.RunID) + assert.Equal(t, uint64(7), data.Seq) + assert.Equal(t, "> ", data.Prompt) +} + func TestHTTPSSE_StreamConsumesRunSessionEvents(t *testing.T) { // Acceptance: A seeded execution must be tracked in the bridge so that the SSE // stream endpoint can resolve it. The route must be registered (no 405). diff --git a/internal/interfaces/cli/run_wiring_conversation_test.go b/internal/interfaces/cli/run_wiring_conversation_test.go index 6e85ec1e..3798daa4 100644 --- a/internal/interfaces/cli/run_wiring_conversation_test.go +++ b/internal/interfaces/cli/run_wiring_conversation_test.go @@ -33,7 +33,7 @@ states: mode: conversation prompt: "Hello" options: - model: sonnet + model: haiku on_success: done done: type: terminal @@ -53,7 +53,7 @@ states: mode: conversation prompt: "Hello" options: - model: sonnet + model: haiku on_success: done done: type: terminal @@ -78,7 +78,7 @@ states: system_prompt: "You are a helpful code reviewer" prompt: "Task: {{inputs.task}}" options: - model: sonnet + model: haiku on_success: done done: type: terminal @@ -98,7 +98,7 @@ states: mode: conversation prompt: "Start conversation" options: - model: sonnet + model: haiku on_success: second_chat second_chat: type: agent @@ -108,7 +108,7 @@ states: conversation: continue_from: first_chat options: - model: sonnet + model: haiku on_success: done done: type: terminal @@ -135,7 +135,7 @@ states: mode: conversation prompt: "First conversation" options: - model: sonnet + model: haiku on_success: done chat_2: type: agent @@ -143,7 +143,7 @@ states: mode: conversation prompt: "Second conversation" options: - model: sonnet + model: haiku on_success: done done: type: terminal @@ -163,7 +163,7 @@ states: mode: conversation prompt: "Hello from single step" options: - model: sonnet + model: haiku on_success: done done: type: terminal diff --git a/internal/interfaces/tui/tab_monitoring.go b/internal/interfaces/tui/tab_monitoring.go index 4cfab440..2a252e30 100644 --- a/internal/interfaces/tui/tab_monitoring.go +++ b/internal/interfaces/tui/tab_monitoring.go @@ -120,6 +120,7 @@ type MonitoringTab struct { // Conversation input and streaming display. inputReader *TUIInputReader + activeSession ports.RunSession inputField textinput.Model inputActive bool convBuf *strings.Builder @@ -189,6 +190,7 @@ func (t MonitoringTab) Update(msg tea.Msg) (MonitoringTab, tea.Cmd) { //nolint:c t.selectedIdx = 0 t.states = make(map[string]workflow.StepState) t.finalStatus = "" // clear any previous run's outcome so terminal steps start pending + t.activeSession = msg.Session t.agentStep = "" t.autoScroll = true t.showSpinner = false @@ -203,10 +205,14 @@ func (t MonitoringTab) Update(msg tea.Msg) (MonitoringTab, tea.Cmd) { //nolint:c // EventWorkflowCompleted/Failed event; this only clears the in-flight indicators. t.ticking = false t.showSpinner = false + t.activeSession = nil return t, nil case facadeEventMsg: t.applyFacadeEvent(msg.Event) + if msg.Event.Kind == ports.EventInputRequired { + t.beginInputRequest() + } return t, nil case spinner.TickMsg: @@ -218,31 +224,13 @@ func (t MonitoringTab) Update(msg tea.Msg) (MonitoringTab, tea.Cmd) { //nolint:c return t, nil case InputRequestedMsg: - t.inputActive = true - t.resizeInputField() - t.resizeViewport() - _ = t.inputField.Focus() - t.autoSelectRunning() - // Only append new turns during an active conversation; never restart - // tracking (convStart resets convTurnCount, duplicating user messages - // already rendered by convAppendUser). - if t.convStep != "" { - if state, ok := t.states[t.convStep]; ok && state.Conversation != nil { - t.convAppendNewTurns(state.Conversation) - } - } else if len(t.flatNodes) > 0 && t.selectedIdx >= 0 && t.selectedIdx < len(t.flatNodes) { - name := t.flatNodes[t.selectedIdx].Name - if state, ok := t.states[name]; ok && state.Conversation != nil { - t.convStart(name, state.Conversation) - } - } - t.convApplyToViewport() + t.beginInputRequest() return t, nil case tea.KeyPressMsg: if t.inputActive { switch { - case key.Matches(msg, keySelect) && t.inputReader != nil: + case key.Matches(msg, keySelect): text := t.inputField.Value() t.inputField.Reset() t.inputActive = false @@ -252,7 +240,13 @@ func (t MonitoringTab) Update(msg tea.Msg) (MonitoringTab, tea.Cmd) { //nolint:c t.stream.Reset() } t.convApplyToViewport() - t.inputReader.Respond(text) + if t.activeSession != nil { + if err := t.activeSession.Respond(ports.InputResponse{Value: text}); err != nil { + return t, nil + } + } else if t.inputReader != nil { + t.inputReader.Respond(text) + } return t, nil case msg.Code == tea.KeyUp: if t.selectedIdx > 0 { @@ -376,11 +370,33 @@ func (t MonitoringTab) InputActive() bool { //nolint:gocritic // read-only return t.inputActive } +func (t *MonitoringTab) beginInputRequest() { + t.inputActive = true + t.resizeInputField() + t.resizeViewport() + _ = t.inputField.Focus() + t.autoSelectRunning() + // Only append new turns during an active conversation; never restart + // tracking (convStart resets convTurnCount, duplicating user messages + // already rendered by convAppendUser). + if t.convStep != "" { + if state, ok := t.states[t.convStep]; ok && state.Conversation != nil { + t.convAppendNewTurns(state.Conversation) + } + } else if len(t.flatNodes) > 0 && t.selectedIdx >= 0 && t.selectedIdx < len(t.flatNodes) { + name := t.flatNodes[t.selectedIdx].Name + if state, ok := t.states[name]; ok && state.Conversation != nil { + t.convStart(name, state.Conversation) + } + } + t.convApplyToViewport() +} + // View renders the monitoring tab. // //nolint:gocritic // Bubbletea convention: value receivers return a new model on each update func (t MonitoringTab) View() string { - if t.active == nil && t.wf == nil { + if t.active == nil && t.wf == nil && !t.inputActive { return EmptyStateView("📡", "No active execution", "Launch a workflow from the Workflows tab.") } diff --git a/internal/interfaces/tui/tab_monitoring_test.go b/internal/interfaces/tui/tab_monitoring_test.go index c3f7845f..376ee627 100644 --- a/internal/interfaces/tui/tab_monitoring_test.go +++ b/internal/interfaces/tui/tab_monitoring_test.go @@ -53,6 +53,7 @@ import ( // mockRunSession implements ports.RunSession for testing event loop behavior. type mockRunSession struct { eventsChan <-chan ports.Event + responses []ports.InputResponse } func (m *mockRunSession) ID() string { @@ -64,6 +65,7 @@ func (m *mockRunSession) Events() <-chan ports.Event { } func (m *mockRunSession) Respond(r ports.InputResponse) error { + m.responses = append(m.responses, r) return nil } @@ -1200,6 +1202,33 @@ func TestTabMonitoring_ConversationTurnsFromFacadeEvents(t *testing.T) { assert.Contains(t, view, "production", "the user's answer must appear in the conversation") } +func TestTabMonitoring_InputRequiredFromFacadeShowsInputAndRespondsToSession(t *testing.T) { + tab := newMonitoringTab() + session := &mockRunSession{} + tab, _ = tab.Update(tea.WindowSizeMsg{Width: 100, Height: 40}) + tab, _ = tab.Update(ExecutionStartedMsg{ExecutionID: "clarifier", Session: session}) + + tab, cmd := tab.Update(facadeEventMsg{Event: ports.Event{ + Kind: ports.EventInputRequired, + RunID: "clarifier", + Payload: &ports.EnrichedInputRequest{ + Prompt: "> ", + }, + }}) + + require.Nil(t, cmd) + require.True(t, tab.InputActive(), "facade input request must display the conversation input box") + assert.Contains(t, tab.View(), "empty to end conversation", "input box must be rendered while awaiting user input") + + tab.inputField.SetValue("production") + tab, cmd = tab.Update(tea.KeyPressMsg{Code: tea.KeyEnter}) + + require.Nil(t, cmd) + require.False(t, tab.InputActive(), "submitting input should hide the input box") + require.Len(t, session.responses, 1) + assert.Equal(t, "production", session.responses[0].Value) +} + // --- End-to-end: AC gate TestMonitoringTab --- // TestMonitoringTab is the acceptance-criteria gate named in the task spec.