diff --git a/apps/backend/internal/agent/runtime/lifecycle/event_types.go b/apps/backend/internal/agent/runtime/lifecycle/event_types.go index 42d0b7f61b..043442be43 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/event_types.go +++ b/apps/backend/internal/agent/runtime/lifecycle/event_types.go @@ -24,6 +24,22 @@ type AgentEventPayload struct { PromptGeneration uint64 `json:"prompt_generation,omitempty"` } +// AgentStalledPayload describes a prompt that has stopped receiving agent events. +// It is advisory only; the lifecycle remains running until the provider completes +// or the user cancels the turn. +type AgentStalledPayload struct { + AgentExecutionID string `json:"agent_execution_id"` + TaskID string `json:"task_id"` + SessionID string `json:"session_id"` + PromptGeneration uint64 `json:"prompt_generation"` + LastActivityAt time.Time `json:"last_activity_at"` + StalledFor time.Duration `json:"stalled_for"` + ToolCallID string `json:"tool_call_id,omitempty"` + ToolName string `json:"tool_name,omitempty"` + ToolTitle string `json:"tool_title,omitempty"` + ToolStatus string `json:"tool_status,omitempty"` +} + // AgentctlEventPayload is the payload for agentctl lifecycle events (starting, ready, error). type AgentctlEventPayload struct { TaskID string `json:"task_id"` diff --git a/apps/backend/internal/agent/runtime/lifecycle/events.go b/apps/backend/internal/agent/runtime/lifecycle/events.go index ee21020ff7..49919ce6e7 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/events.go +++ b/apps/backend/internal/agent/runtime/lifecycle/events.go @@ -34,6 +34,39 @@ func (p *EventPublisher) PublishAgentEvent(ctx context.Context, eventType string p.publishAgentEventPayload(ctx, eventType, newAgentEventPayload(execution)) } +// PublishAgentStalled publishes one advisory inactivity signal for a prompt. +func (p *EventPublisher) PublishAgentStalled( + ctx context.Context, + execution *AgentExecution, + promptGeneration uint64, + lastActivityAt time.Time, + stalledFor time.Duration, +) { + if p.eventBus == nil { + return + } + payload := AgentStalledPayload{ + AgentExecutionID: execution.ID, + TaskID: execution.TaskID, + SessionID: execution.SessionID, + PromptGeneration: promptGeneration, + LastActivityAt: lastActivityAt, + StalledFor: stalledFor, + } + if tool := execution.activeToolSnapshot(); tool != nil { + payload.ToolCallID = tool.ToolCallID + payload.ToolName = tool.Name + payload.ToolTitle = tool.Title + payload.ToolStatus = tool.Status + } + event := bus.NewEvent(events.AgentStalled, "agent-manager", payload) + if err := p.eventBus.Publish(ctx, events.AgentStalled, event); err != nil { + p.logger.Error("failed to publish stalled agent event", + zap.String("execution_id", execution.ID), + zap.Error(err)) + } +} + // publishAgentEventPayload publishes an immutable agent lifecycle snapshot. func (p *EventPublisher) publishAgentEventPayload(ctx context.Context, eventType string, payload AgentEventPayload) { if p.eventBus == nil { diff --git a/apps/backend/internal/agent/runtime/lifecycle/execution_store.go b/apps/backend/internal/agent/runtime/lifecycle/execution_store.go index 2d25690107..4776e65a2f 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/execution_store.go +++ b/apps/backend/internal/agent/runtime/lifecycle/execution_store.go @@ -220,6 +220,7 @@ func (s *ExecutionStore) BeginPrompt(executionID string) (uint64, error) { func beginExecutionPrompt(execution *AgentExecution) uint64 { execution.promptGeneration++ execution.Status = v1.AgentStatusRunning + execution.resetActiveTool() return execution.promptGeneration } diff --git a/apps/backend/internal/agent/runtime/lifecycle/execution_store_test.go b/apps/backend/internal/agent/runtime/lifecycle/execution_store_test.go index c631825577..b592caf291 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/execution_store_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/execution_store_test.go @@ -119,3 +119,19 @@ func TestExecutionStore_BeginPromptAlwaysAdvancesGeneration(t *testing.T) { t.Fatal("replacement prompt must create generation 2 while already running") } } + +func TestExecutionStore_BeginPromptClearsActiveTopLevelTool(t *testing.T) { + store := NewExecutionStore() + exec := &AgentExecution{ID: "exec-1", SessionID: "session-1"} + if err := store.Add(exec); err != nil { + t.Fatalf("Add: %v", err) + } + exec.setActiveTool(activeTopLevelTool{ToolCallID: "tool-1", Name: "shell"}) + + if _, err := store.BeginPrompt(exec.ID); err != nil { + t.Fatalf("BeginPrompt: %v", err) + } + if got := exec.activeToolSnapshot(); got != nil { + t.Fatalf("active tool after BeginPrompt = %#v, want nil", got) + } +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_events.go b/apps/backend/internal/agent/runtime/lifecycle/manager_events.go index efb5421456..bb878dcdd6 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_events.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_events.go @@ -346,6 +346,12 @@ func (m *Manager) handleCompleteEvent(execution *AgentExecution, event *agentctl // separate DB rows mid-sentence, breaking markdown that spans the boundary. func (m *Manager) handleToolCallEvent(execution *AgentExecution, event agentctl.AgentEvent) agentctl.AgentEvent { if event.ParentToolCallID == "" { + execution.setActiveTool(activeTopLevelTool{ + ToolCallID: event.ToolCallID, + Name: event.ToolName, + Title: event.ToolTitle, + Status: event.ToolStatus, + }) // flushMessageBuffer publishes any remaining buffered content through // the streaming path itself and always returns "". m.flushMessageBuffer(execution) @@ -365,6 +371,9 @@ func (m *Manager) handleToolCallEvent(execution *AgentExecution, event agentctl. // handleToolUpdateEvent stores completed tool results in session history. func (m *Manager) handleToolUpdateEvent(execution *AgentExecution, event agentctl.AgentEvent) { + if event.ParentToolCallID == "" && isTerminalToolUpdate(event) { + execution.clearActiveTool(event.ToolCallID) + } if m.historyManager != nil && execution.historyEnabled && execution.SessionID != "" && event.ToolStatus == toolStatusComplete { m.flushAssistantHistory(execution) if err := m.historyManager.AppendToolResult(execution.SessionID, event); err != nil { diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go index d34e73c18d..788e1d0894 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_events_test.go @@ -1392,6 +1392,47 @@ func TestHandleAgentEvent_UpdatesLastActivityAt(t *testing.T) { } } +func TestHandleAgentEvent_TracksActiveTopLevelTool(t *testing.T) { + mgr, _ := createTestManagerWithTracking() + execution := createTestExecution("exec-1", "task-1", "session-1") + + mgr.handleAgentEvent(execution, agentctl.AgentEvent{ + Type: "tool_call", + ToolCallID: "top-level-tool", + ToolName: "shell", + ToolTitle: "Start dev server", + ToolStatus: "in_progress", + }) + + if got := execution.activeToolSnapshot(); got == nil || + got.ToolCallID != "top-level-tool" || got.Title != "Start dev server" { + t.Fatalf("active top-level tool = %#v, want Start dev server", got) + } + + mgr.handleAgentEvent(execution, agentctl.AgentEvent{ + Type: "tool_call", + ToolCallID: "subagent-tool", + ToolName: "shell", + ToolTitle: "Run subagent command", + ParentToolCallID: "top-level-tool", + ToolStatus: "in_progress", + }) + + if got := execution.activeToolSnapshot(); got == nil || got.ToolCallID != "top-level-tool" { + t.Fatalf("subagent tool replaced top-level tool: %#v", got) + } + + mgr.handleAgentEvent(execution, agentctl.AgentEvent{ + Type: "tool_update", + ToolCallID: "top-level-tool", + ToolStatus: toolStatusComplete, + }) + + if got := execution.activeToolSnapshot(); got != nil { + t.Fatalf("terminal update left active top-level tool: %#v", got) + } +} + // TestHandleAgentEvent_PromptDoneChDoesNotBlockWhenFull verifies that signaling // promptDoneCh with a full channel (no receiver) doesn't block the event handler. func TestHandleAgentEvent_PromptDoneChDoesNotBlockWhenFull(t *testing.T) { diff --git a/apps/backend/internal/agent/runtime/lifecycle/session.go b/apps/backend/internal/agent/runtime/lifecycle/session.go index f971d909da..a18e900609 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/session.go +++ b/apps/backend/internal/agent/runtime/lifecycle/session.go @@ -565,8 +565,9 @@ func (sm *SessionManager) waitForPromptDone( execution *AgentExecution, promptGeneration uint64, ) (*PromptResult, error) { - stallTicker := time.NewTicker(30 * time.Second) + stallTicker := time.NewTicker(time.Minute) defer stallTicker.Stop() + stallReported := false for { select { @@ -618,11 +619,21 @@ func (sm *SessionManager) waitForPromptDone( lastActivity := execution.lastActivityAt execution.lastActivityAtMu.Unlock() - if elapsed > 5*time.Minute { + if elapsed >= 5*time.Minute && !stallReported { sm.logger.Warn("agent stall detected: no events received", zap.String("execution_id", execution.ID), zap.Duration("elapsed_since_last_event", elapsed), zap.Time("last_activity", lastActivity)) + if sm.eventPublisher != nil { + sm.eventPublisher.PublishAgentStalled( + ctx, + execution, + promptGeneration, + lastActivity, + elapsed, + ) + } + stallReported = true } } } diff --git a/apps/backend/internal/agent/runtime/lifecycle/session_test.go b/apps/backend/internal/agent/runtime/lifecycle/session_test.go index 61f7c41aec..bc97c9feee 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/session_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/session_test.go @@ -11,6 +11,7 @@ import ( "strings" "sync" "testing" + "testing/synctest" "time" "github.com/coder/acp-go-sdk" @@ -1734,6 +1735,58 @@ func TestWaitForPromptDone_IgnoresSupersededGenerationSignal(t *testing.T) { } } +func TestWaitForPromptDone_PublishesSingleStall(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eventBus := &MockEventBusWithTracking{} + sm := NewSessionManager(newSessionTestLogger(), make(chan struct{})) + sm.eventPublisher = NewEventPublisher(eventBus, newSessionTestLogger()) + execution := &AgentExecution{ + ID: "test-exec", + TaskID: "test-task", + SessionID: "test-session", + promptDoneCh: make(chan PromptCompletionSignal, 1), + } + execution.lastActivityAt = time.Now() + + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + waitResult := make(chan error, 1) + go func() { + _, err := sm.waitForPromptDone(ctx, execution, 7) + waitResult <- err + }() + + time.Sleep(5 * time.Minute) + synctest.Wait() + if got := countPublishedEvents(eventBus, "agent.stalled"); got != 1 { + t.Fatalf("published stalled events = %d, want 1", got) + } + + time.Sleep(2 * time.Minute) + synctest.Wait() + if got := countPublishedEvents(eventBus, "agent.stalled"); got != 1 { + t.Fatalf("published stalled events after later checks = %d, want 1", got) + } + + cancel() + if err := <-waitResult; !errors.Is(err, context.Canceled) { + t.Fatalf("waitForPromptDone error = %v, want context canceled", err) + } + }) +} + +func countPublishedEvents(eventBus *MockEventBusWithTracking, subject string) int { + eventBus.mu.Lock() + defer eventBus.mu.Unlock() + var count int + for _, event := range eventBus.PublishedEvents { + if event.Subject == subject { + count++ + } + } + return count +} + func TestSendPrompt_TriggerTimeCancelReleaseReturnsErrCancelEscalated(t *testing.T) { mock := newMockAgentServer(t) defer mock.Close() diff --git a/apps/backend/internal/agent/runtime/lifecycle/types.go b/apps/backend/internal/agent/runtime/lifecycle/types.go index 6f2541162b..43870a51a1 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/types.go +++ b/apps/backend/internal/agent/runtime/lifecycle/types.go @@ -178,6 +178,8 @@ type AgentExecution struct { // Last time an agent event was received (for stall detection) lastActivityAt time.Time lastActivityAtMu sync.Mutex + activeTool *activeTopLevelTool + activeToolMu sync.RWMutex // Fires once on the first agent event to publish AgentRunning. firstActivityOnce sync.Once @@ -187,6 +189,43 @@ type AgentExecution struct { sessionSpanMu sync.RWMutex } +type activeTopLevelTool struct { + ToolCallID string + Name string + Title string + Status string +} + +func (e *AgentExecution) setActiveTool(tool activeTopLevelTool) { + e.activeToolMu.Lock() + e.activeTool = &tool + e.activeToolMu.Unlock() +} + +func (e *AgentExecution) clearActiveTool(toolCallID string) { + e.activeToolMu.Lock() + if e.activeTool != nil && e.activeTool.ToolCallID == toolCallID { + e.activeTool = nil + } + e.activeToolMu.Unlock() +} + +func (e *AgentExecution) resetActiveTool() { + e.activeToolMu.Lock() + e.activeTool = nil + e.activeToolMu.Unlock() +} + +func (e *AgentExecution) activeToolSnapshot() *activeTopLevelTool { + e.activeToolMu.RLock() + defer e.activeToolMu.RUnlock() + if e.activeTool == nil { + return nil + } + tool := *e.activeTool + return &tool +} + // setRuntimeEnvironment stores a defensive copy of the effective task runtime // environment. Callers must use RuntimeEnvironment when reading it. func (e *AgentExecution) setRuntimeEnvironment(env map[string]string) { diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index fd71cb7f16..c39c7cd0e9 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -168,6 +168,7 @@ const ( AgentReady = "agent.ready" // Agent finished a prompt turn, ready for follow-up AgentCompleted = "agent.completed" AgentFailed = "agent.failed" + AgentStalled = "agent.stalled" AgentStopped = "agent.stopped" AgentContextReset = "agent.context_reset" // Agent subprocess restarted with fresh ACP session AgentACPSessionCreated = "agent.acp_session_created" diff --git a/apps/backend/internal/orchestrator/event_handlers_stall.go b/apps/backend/internal/orchestrator/event_handlers_stall.go new file mode 100644 index 0000000000..5b36ee6793 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_stall.go @@ -0,0 +1,87 @@ +package orchestrator + +import ( + "context" + "fmt" + "strings" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +const ( + stallCancelTurnButtonTestID = "stall-cancel-turn-button" + metaKeyActionVisibility = "action_visibility" +) + +// handleAgentStalled persists an advisory recovery affordance without changing +// the prompt, session, or task lifecycle. +func (s *Service) handleAgentStalled(ctx context.Context, payload lifecycle.AgentStalledPayload) { + if s.messageCreator == nil || s.repo == nil || payload.TaskID == "" || payload.SessionID == "" { + return + } + session, err := s.repo.GetTaskSession(ctx, payload.SessionID) + if err != nil || session == nil || session.State != models.TaskSessionStateRunning { + return + } + generationOwner, ok := s.agentManager.(interface { + OwnsPromptGeneration(sessionID, executionID string, generation uint64) bool + }) + if !ok || payload.PromptGeneration == 0 || !generationOwner.OwnsPromptGeneration( + payload.SessionID, + payload.AgentExecutionID, + payload.PromptGeneration, + ) { + return + } + turnID, err := s.peekActiveTurnID(ctx, payload.SessionID) + if err != nil { + return + } + if s.turnService != nil && turnID == "" { + return + } + metadata := map[string]interface{}{ + metaKeyActionVisibility: "running", + metaKeySessionID: payload.SessionID, + metaKeyTaskID: payload.TaskID, + "actions": []map[string]interface{}{{ + actionMetaKeyType: "ws_request", + actionMetaKeyLabel: "Cancel turn", + actionMetaKeyTestID: stallCancelTurnButtonTestID, + "params": map[string]interface{}{ + "method": "agent.cancel", + "payload": map[string]interface{}{"session_id": payload.SessionID}, + }, + }}, + } + if err := s.messageCreator.CreateSessionMessage( + ctx, + payload.TaskID, + stallNoticeContent(payload), + payload.SessionID, + string(v1.MessageTypeStatus), + turnID, + metadata, + false, + ); err != nil { + s.logger.Warn("failed to create agent stall notice", + zap.String("task_id", payload.TaskID), + zap.String("session_id", payload.SessionID), + zap.Error(err)) + } +} + +func stallNoticeContent(payload lifecycle.AgentStalledPayload) string { + tool := strings.TrimSpace(payload.ToolTitle) + if tool == "" { + tool = strings.TrimSpace(payload.ToolName) + } + if tool == "" { + return "Still waiting for the agent." + } + return fmt.Sprintf("Still waiting on %s.", tool) +} diff --git a/apps/backend/internal/orchestrator/event_handlers_stall_test.go b/apps/backend/internal/orchestrator/event_handlers_stall_test.go new file mode 100644 index 0000000000..a036e49202 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_stall_test.go @@ -0,0 +1,124 @@ +package orchestrator + +import ( + "context" + "strings" + "testing" + + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" + "github.com/kandev/kandev/internal/task/models" +) + +func TestHandleAgentStalled_PersistsNeutralRunningNotice(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task-1", "session-1", "step-1") + before, err := repo.GetTaskSession(ctx, "session-1") + if err != nil { + t.Fatalf("get session before handling stall: %v", err) + } + agentMgr := &mockAgentManager{ + repoForExecutionLookup: repo, + currentPromptExecutionID: "execution-1", + } + agentMgr.currentPromptGeneration.Store(7) + svc := createTestServiceWithScheduler( + repo, + newMockStepGetter(), + newMockTaskRepo(), + agentMgr, + ) + svc.turnService = &repoTurnService{repo: repo} + activeTurn, err := svc.turnService.StartTurn(ctx, "session-1") + if err != nil { + t.Fatalf("start active turn: %v", err) + } + messages := &mockMessageCreator{} + svc.messageCreator = messages + + svc.handleAgentStalled(ctx, lifecycle.AgentStalledPayload{ + AgentExecutionID: "execution-1", + TaskID: "task-1", + SessionID: "session-1", + PromptGeneration: 7, + ToolName: "shell", + ToolTitle: "Start dev server", + ToolStatus: "in_progress", + }) + + if len(messages.sessionMessages) != 1 { + t.Fatalf("session messages = %d, want 1", len(messages.sessionMessages)) + } + message := messages.sessionMessages[0] + if !strings.Contains(message.content, "Still waiting on Start dev server") { + t.Fatalf("notice content = %q, want sanitized tool title", message.content) + } + if message.metadata["action_visibility"] != "running" { + t.Fatalf("action visibility = %v, want running", message.metadata["action_visibility"]) + } + if message.turnID != activeTurn.ID { + t.Fatalf("notice turn ID = %q, want active turn %q", message.turnID, activeTurn.ID) + } + if _, hasVariant := message.metadata["variant"]; hasVariant { + t.Fatalf("notice metadata unexpectedly set a warning/error variant: %#v", message.metadata) + } + actions, ok := message.metadata["actions"].([]map[string]interface{}) + if !ok || len(actions) != 1 { + t.Fatalf("actions = %#v, want one cancel action", message.metadata["actions"]) + } + action := actions[0] + if action["label"] != "Cancel turn" || action["test_id"] != "stall-cancel-turn-button" { + t.Fatalf("cancel action = %#v", action) + } + params, ok := action["params"].(map[string]interface{}) + if !ok || params["method"] != "agent.cancel" { + t.Fatalf("cancel params = %#v, want agent.cancel", action["params"]) + } + + after, err := repo.GetTaskSession(ctx, "session-1") + if err != nil { + t.Fatalf("get session after handling stall: %v", err) + } + if after.State != before.State { + t.Fatalf("session state changed from %q to %q", before.State, after.State) + } +} + +func TestHandleAgentStalled_RejectsSettledOrStalePrompt(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task-1", "session-1", "step-1") + agentMgr := &mockAgentManager{currentPromptExecutionID: "execution-1"} + agentMgr.currentPromptGeneration.Store(7) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + messages := &mockMessageCreator{} + svc.messageCreator = messages + payload := lifecycle.AgentStalledPayload{ + AgentExecutionID: "execution-1", + TaskID: "task-1", + SessionID: "session-1", + PromptGeneration: 7, + } + + if err := repo.UpdateTaskSessionState(ctx, "session-1", models.TaskSessionStateWaitingForInput, ""); err != nil { + t.Fatalf("settle session: %v", err) + } + svc.handleAgentStalled(ctx, payload) + if len(messages.sessionMessages) != 0 { + t.Fatalf("settled session messages = %d, want 0", len(messages.sessionMessages)) + } + if err := repo.UpdateTaskSessionState(ctx, "session-1", models.TaskSessionStateRunning, ""); err != nil { + t.Fatalf("resume session: %v", err) + } + agentMgr.currentPromptGeneration.Store(8) + svc.handleAgentStalled(ctx, payload) + if len(messages.sessionMessages) != 0 { + t.Fatalf("stale generation messages = %d, want 0", len(messages.sessionMessages)) + } +} + +func TestStallNoticeContentFallsBackWithoutTool(t *testing.T) { + if got := stallNoticeContent(lifecycle.AgentStalledPayload{}); got != "Still waiting for the agent." { + t.Fatalf("stallNoticeContent() = %q, want generic fallback", got) + } +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index cfb5c44cee..4eea878549 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -675,6 +675,7 @@ func NewService( OnAgentReady: s.handleAgentReady, OnAgentCompleted: s.handleAgentCompleted, OnAgentFailed: s.handleAgentFailed, + OnAgentStalled: s.handleAgentStalled, OnAgentStopped: s.handleAgentStopped, OnAgentStreamEvent: s.handleAgentStreamEvent, OnACPSessionCreated: s.handleACPSessionCreated, diff --git a/apps/backend/internal/orchestrator/watcher/watcher.go b/apps/backend/internal/orchestrator/watcher/watcher.go index df3483b888..b6af1fcbfb 100644 --- a/apps/backend/internal/orchestrator/watcher/watcher.go +++ b/apps/backend/internal/orchestrator/watcher/watcher.go @@ -99,6 +99,7 @@ type EventHandlers struct { OnAgentReady func(ctx context.Context, data AgentEventData) // Turn ended; agent idle waiting for follow-up OnAgentCompleted func(ctx context.Context, data AgentEventData) OnAgentFailed func(ctx context.Context, data AgentEventData) + OnAgentStalled func(ctx context.Context, payload lifecycle.AgentStalledPayload) OnAgentStopped func(ctx context.Context, data AgentEventData) OnACPSessionCreated func(ctx context.Context, data ACPSessionEventData) @@ -162,6 +163,10 @@ func (w *Watcher) Start(ctx context.Context) error { w.unsubscribeAll() return err } + if err := w.subscribeToAgentStallEvents(); err != nil { + w.unsubscribeAll() + return err + } // Subscribe to ACP session events if err := w.subscribeToACPSessionEvents(); err != nil { @@ -320,6 +325,23 @@ func (w *Watcher) subscribeToAgentEvents() error { return nil } +func (w *Watcher) subscribeToAgentStallEvents() error { + if w.handlers.OnAgentStalled == nil { + return nil + } + sub, err := w.eventBus.QueueSubscribe( + events.AgentStalled, + w.queue, + w.createAgentStalledEventHandler(w.handlers.OnAgentStalled), + ) + if err != nil { + w.logger.Error("failed to subscribe to agent stalled event", zap.Error(err)) + return err + } + w.subscriptions = append(w.subscriptions, sub) + return nil +} + // subscribeToACPSessionEvents subscribes to ACP session lifecycle events func (w *Watcher) subscribeToACPSessionEvents() error { if w.handlers.OnACPSessionCreated == nil { @@ -425,6 +447,20 @@ func (w *Watcher) createAgentEventHandler(handler func(ctx context.Context, data } } +func (w *Watcher) createAgentStalledEventHandler( + handler func(ctx context.Context, payload lifecycle.AgentStalledPayload), +) bus.EventHandler { + return func(ctx context.Context, event *bus.Event) error { + var payload lifecycle.AgentStalledPayload + if err := w.parseEventData(event.Data, &payload); err != nil { + w.logger.Error("failed to parse agent stalled event", zap.Error(err)) + return nil + } + handler(ctx, payload) + return nil + } +} + // createACPSessionEventHandler creates a bus.EventHandler for ACP session events func (w *Watcher) createACPSessionEventHandler(handler func(ctx context.Context, data ACPSessionEventData)) bus.EventHandler { return func(ctx context.Context, event *bus.Event) error { diff --git a/apps/backend/internal/orchestrator/watcher/watcher_test.go b/apps/backend/internal/orchestrator/watcher/watcher_test.go index 707771459b..ecfe1de0bc 100644 --- a/apps/backend/internal/orchestrator/watcher/watcher_test.go +++ b/apps/backend/internal/orchestrator/watcher/watcher_test.go @@ -303,6 +303,48 @@ func TestAgentEventHandling(t *testing.T) { }) } +func TestWatcherDispatchesAgentStalled(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + eventBus := newMockEventBus() + var received lifecycle.AgentStalledPayload + var handled bool + var mu sync.Mutex + w := NewWatcher(eventBus, EventHandlers{ + OnAgentStalled: func(_ context.Context, payload lifecycle.AgentStalledPayload) { + mu.Lock() + received = payload + handled = true + mu.Unlock() + }, + }, "orchestrator-test", createTestLogger()) + if err := w.Start(context.Background()); err != nil { + t.Fatalf("start watcher: %v", err) + } + t.Cleanup(func() { _ = w.Stop() }) + + event := bus.NewEvent(events.AgentStalled, "test", map[string]interface{}{ + "agent_execution_id": "agent-456", + "task_id": "task-123", + "session_id": "session-789", + "prompt_generation": 7, + "tool_title": "Start dev server", + }) + if err := eventBus.Publish(context.Background(), events.AgentStalled, event); err != nil { + t.Fatalf("publish stalled event: %v", err) + } + synctest.Wait() + + mu.Lock() + defer mu.Unlock() + if !handled { + t.Fatal("OnAgentStalled handler was not called") + } + if received.SessionID != "session-789" || received.ToolTitle != "Start dev server" { + t.Fatalf("received stalled payload = %#v", received) + } + }) +} + func TestAgentStreamEventHandling(t *testing.T) { synctest.Test(t, func(t *testing.T) { eventBus := newMockEventBus() diff --git a/apps/web/components/task/chat/messages/action-message.test.tsx b/apps/web/components/task/chat/messages/action-message.test.tsx index 5ca4616476..8131e60efa 100644 --- a/apps/web/components/task/chat/messages/action-message.test.tsx +++ b/apps/web/components/task/chat/messages/action-message.test.tsx @@ -30,6 +30,7 @@ const CANCEL_TEST_ID = "recovery-cancel-retry-button"; const TECHNICAL_DETAILS = "Technical details"; const RECOVERY_MESSAGE = "Agent encountered an error"; const RESUME_TEST_ID = "recovery-resume-button"; +const STALL_CANCEL_TEST_ID = "stall-cancel-turn-button"; function retryMessage(overrides: Partial = {}): Message { return { @@ -90,9 +91,32 @@ function recoveryMessage(withParams = false): Message { } as Partial); } +function stalledMessage(turnId = "turn-1"): Message { + return retryMessage({ + turn_id: turnId, + content: "Still waiting on Start dev server.", + metadata: { + action_visibility: "running", + actions: [ + { + type: "ws_request", + label: "Cancel turn", + test_id: STALL_CANCEL_TEST_ID, + params: { method: "agent.cancel", payload: { session_id: "sess-1" } }, + }, + ], + }, + } as Partial); +} + /** ActionMessage reads session state from the store (keyed by comment.session_id), * so seed it via the provider instead of passing a prop. */ -function renderAction(comment: Message, sessionState?: TaskSessionState, sessionError?: string) { +function renderAction( + comment: Message, + sessionState?: TaskSessionState, + sessionError?: string, + activeTurnId?: string, +) { const initialState: Partial = sessionState ? { taskSessions: { @@ -100,6 +124,10 @@ function renderAction(comment: Message, sessionState?: TaskSessionState, session "sess-1": { state: sessionState, error_message: sessionError } as TaskSession, }, }, + turns: { + bySession: {}, + activeBySession: activeTurnId ? { "sess-1": activeTurnId } : {}, + }, } : {}; return render(, { @@ -172,6 +200,50 @@ describe("ActionMessage — transient retry (warning variant)", () => { }); }); +describe("ActionMessage — running stall notice", () => { + it("renders a neutral compact notice only while the session is running", () => { + renderAction(stalledMessage(), "RUNNING", undefined, "turn-1"); + + const notice = screen.getByTestId("running-action-notice"); + expect(notice.className).toContain("text-muted-foreground"); + expect(notice.className).not.toContain("text-amber"); + expect(notice.className).not.toContain("text-red"); + expect(notice.querySelector("svg")).toBeNull(); + const button = screen.getByTestId(STALL_CANCEL_TEST_ID); + expect(button.className).toContain("min-h-11"); + expect(button.className).not.toContain("w-full"); + }); + + it("hides the running-only notice after the session settles", () => { + const { container } = renderAction(stalledMessage(), "WAITING_FOR_INPUT"); + expect(container.firstChild).toBeNull(); + }); + + it("sends agent.cancel when Cancel turn is activated", async () => { + renderAction(stalledMessage(), "RUNNING", undefined, "turn-1"); + + fireEvent.click(screen.getByTestId(STALL_CANCEL_TEST_ID)); + + await waitFor(() => + expect(requestMock).toHaveBeenCalledWith("agent.cancel", { session_id: "sess-1" }), + ); + }); + + it("hides an old notice when a later turn is active", () => { + const { container } = renderAction(stalledMessage("turn-1"), "RUNNING", undefined, "turn-2"); + expect(container.firstChild).toBeNull(); + }); + + it("hides a running-only notice without a turn ID", () => { + const message = stalledMessage(); + message.turn_id = undefined; + + const { container } = renderAction(message, "RUNNING", undefined, "turn-1"); + + expect(container.firstChild).toBeNull(); + }); +}); + describe("ActionMessage — missing PR branch", () => { it("renders a plain-language recovery panel with collapsed technical details", () => { renderAction( diff --git a/apps/web/components/task/chat/messages/action-message.tsx b/apps/web/components/task/chat/messages/action-message.tsx index ee2d128129..7e7ccd2d6a 100644 --- a/apps/web/components/task/chat/messages/action-message.tsx +++ b/apps/web/components/task/chat/messages/action-message.tsx @@ -38,6 +38,7 @@ const ICON_MAP: Record = { type ActionMeta = { actions?: MessageAction[]; + action_visibility?: "running"; variant?: string; recovery_actions?: boolean; is_auth_error?: boolean; @@ -55,21 +56,45 @@ export const ActionMessage = memo(function ActionMessage({ comment }: { comment: // Read session state from the store instead of receiving it as a prop, so a // state transition doesn't re-render every message in the list (only the // rare action messages that actually depend on it). - const sessionState = useAppStore((state) => - comment.session_id - ? (state.taskSessions.items[comment.session_id]?.state ?? undefined) - : undefined, - ); - const sessionError = useAppStore((state) => - comment.session_id - ? (state.taskSessions.items[comment.session_id]?.error_message as string | undefined) - : undefined, - ); - const [recoveryRequested, setRecoveryRequested] = useState(false); + const { sessionState, sessionError, activeTurnId } = useActionMessageSession(comment.session_id); const metadata = comment.metadata as ActionMeta | undefined; - const isWarning = metadata?.variant === "warning"; const message = comment.content || "An error occurred"; + if (metadata?.action_visibility === "running") { + if (sessionState !== "RUNNING" || !comment.turn_id || activeTurnId !== comment.turn_id) { + return null; + } + return ( + + ); + } + + return ( + + ); +}); + +function SettledActionMessage({ + metadata, + message, + sessionError, + sessionState, + taskId, +}: { + metadata: ActionMeta | undefined; + message: string; + sessionError?: string; + sessionState?: TaskSessionState; + taskId?: string; +}) { + const [recoveryRequested, setRecoveryRequested] = useState(false); + // A waiting session can still need recovery, so only hide this persisted // card after its own Resume request is acknowledged (or the session starts). if (isSessionActive(sessionState) || (metadata?.recovery_actions && recoveryRequested)) @@ -79,17 +104,18 @@ export const ActionMessage = memo(function ActionMessage({ comment }: { comment: return ( ); } - const iconClass = isWarning ? "text-amber-500" : "text-red-500"; - const textClass = isWarning - ? "text-amber-600 dark:text-amber-400" - : "text-red-600 dark:text-red-400"; + const iconClass = metadata?.variant === "warning" ? "text-amber-500" : "text-red-500"; + const textClass = + metadata?.variant === "warning" + ? "text-amber-600 dark:text-amber-400" + : "text-red-600 dark:text-red-400"; return (
@@ -103,7 +129,7 @@ export const ActionMessage = memo(function ActionMessage({ comment }: { comment: {metadata?.actions && metadata.actions.length > 0 && ( setRecoveryRequested(true) : undefined } @@ -113,7 +139,42 @@ export const ActionMessage = memo(function ActionMessage({ comment }: { comment:
); -}); +} + +function useActionMessageSession(sessionId: Message["session_id"]) { + const sessionState = useAppStore((state) => + sessionId ? (state.taskSessions.items[sessionId]?.state ?? undefined) : undefined, + ); + const sessionError = useAppStore((state) => + sessionId + ? (state.taskSessions.items[sessionId]?.error_message as string | undefined) + : undefined, + ); + const activeTurnId = useAppStore((state) => + sessionId ? (state.turns.activeBySession[sessionId] ?? undefined) : undefined, + ); + return { sessionState, sessionError, activeTurnId }; +} + +function RunningActionNotice({ + actions, + message, + taskId, +}: { + actions?: MessageAction[]; + message: string; + taskId?: string; +}) { + return ( +
+ {message} + {actions && actions.length > 0 && } +
+ ); +} function MissingBranchRecovery({ metadata, @@ -227,19 +288,28 @@ function ActionButtons({ actions, taskId, onRecoveryRequested, + compact = false, }: { actions: MessageAction[]; taskId?: string; onRecoveryRequested?: () => void; + compact?: boolean; }) { return ( -
+
{actions.map((action, i) => ( ))}
@@ -250,10 +320,12 @@ function ActionButton({ action, messageTaskId, onCompleted, + compact = false, }: { action: MessageAction; messageTaskId?: string; onCompleted?: () => void; + compact?: boolean; }): ReactElement | null { const [state, setState] = useState<"idle" | "busy" | "done" | "error">("idle"); const activeTaskId = useAppStore((s) => s.tasks.activeTaskId); @@ -312,10 +384,12 @@ function ActionButton({ const button = (