diff --git a/apps/backend/cmd/mock-agent/script_test.go b/apps/backend/cmd/mock-agent/script_test.go index 74ef2b0d56..1b393e0a71 100644 --- a/apps/backend/cmd/mock-agent/script_test.go +++ b/apps/backend/cmd/mock-agent/script_test.go @@ -612,8 +612,11 @@ func assertContentText(t *testing.T, msg map[string]any, wantText string) { if content["type"] != BlockText { t.Errorf("content type = %q, want %q", content["type"], BlockText) } - if content["text"] != wantText { - t.Errorf("text = %q, want %q", content["text"], wantText) + // emitTextBlock appends a trailing newline to match real agent streaming behavior; + // allow the expected text with or without the trailing newline. + got, _ := content["text"].(string) + if got != wantText && got != wantText+"\n" { + t.Errorf("text = %q, want %q (or with trailing newline)", got, wantText) } } diff --git a/apps/backend/cmd/mock-agent/sequences.go b/apps/backend/cmd/mock-agent/sequences.go index 3d87d48d3d..0a2fb291d7 100644 --- a/apps/backend/cmd/mock-agent/sequences.go +++ b/apps/backend/cmd/mock-agent/sequences.go @@ -58,7 +58,13 @@ func emitThinkingBlock(enc *json.Encoder, thought, parentToolUseID string) { } // emitTextBlock emits an assistant message with a text content block. +// Text always ends with a newline to match real agent streaming behavior, +// which allows the lifecycle manager's message buffer to flush immediately +// and publish "message_streaming" events for activity detection. func emitTextBlock(enc *json.Encoder, text, parentToolUseID string) { + if !strings.HasSuffix(text, "\n") { + text += "\n" + } _ = enc.Encode(AssistantMsg{ Type: TypeAssistant, ParentToolUseID: parentToolUseID, diff --git a/apps/backend/internal/agent/lifecycle/manager_events.go b/apps/backend/internal/agent/lifecycle/manager_events.go index 16824361de..9adcbdcade 100644 --- a/apps/backend/internal/agent/lifecycle/manager_events.go +++ b/apps/backend/internal/agent/lifecycle/manager_events.go @@ -1,6 +1,7 @@ package lifecycle import ( + "context" "strings" "time" @@ -9,6 +10,7 @@ import ( agentctl "github.com/kandev/kandev/internal/agentctl/client" "github.com/kandev/kandev/internal/agentctl/tracing" + "github.com/kandev/kandev/internal/events" ) const toolStatusComplete = "complete" @@ -231,6 +233,29 @@ func (m *Manager) handleToolCallEvent(execution *AgentExecution, event agentctl. return event } +// handleToolUpdateEvent stores completed tool results in session history. +func (m *Manager) handleToolUpdateEvent(execution *AgentExecution, event agentctl.AgentEvent) { + if m.historyManager != nil && execution.historyEnabled && execution.SessionID != "" && event.ToolStatus == toolStatusComplete { + if err := m.historyManager.AppendToolResult(execution.SessionID, event); err != nil { + m.logger.Warn("failed to store tool result to history", zap.Error(err)) + } + } +} + +// handleErrorEvent processes the "error" agent event: flushes buffers, marks state as failed, +// and signals prompt completion. The raw error event is not published to the frontend stream; +// the agent failure path (handleAgentFailed) sets session FAILED with the error message. +func (m *Manager) handleErrorEvent(execution *AgentExecution, event agentctl.AgentEvent) { + m.flushMessageBuffer(execution) + m.logger.Error("agent error", + zap.String("execution_id", execution.ID), + zap.String("error", event.Error), + zap.String("text", event.Text), + zap.Any("data", event.Data)) + m.handleCompleteEventMarkState(execution, &event, true) + handleCompleteEventSignal(execution, &event, true) +} + // handleContextWindowEvent processes the "context_window" agent event: logs and publishes it. // Returns true because no further stream publishing is needed. func (m *Manager) handleContextWindowEvent(execution *AgentExecution, event agentctl.AgentEvent) { @@ -261,13 +286,22 @@ func (m *Manager) handleAvailableCommandsEvent(execution *AgentExecution, event m.eventPublisher.PublishAvailableCommands(execution, event.AvailableCommands) } -// handleAgentEvent processes incoming agent events from the agent -func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.AgentEvent) { - // Update last activity timestamp for stall detection +// recordActivity updates the last-activity timestamp and, on the very first +// event from an execution, publishes AgentRunning to transition STARTING → RUNNING. +func (m *Manager) recordActivity(execution *AgentExecution) { execution.lastActivityAtMu.Lock() execution.lastActivityAt = time.Now() execution.lastActivityAtMu.Unlock() + execution.firstActivityOnce.Do(func() { + m.eventPublisher.PublishAgentEvent(context.Background(), events.AgentRunning, execution) + }) +} + +// handleAgentEvent processes incoming agent events from the agent +func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.AgentEvent) { + m.recordActivity(execution) + m.logger.Debug("handleAgentEvent entry", zap.String("execution_id", execution.ID), zap.String("event_type", event.Type), @@ -287,29 +321,15 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age event = m.handleToolCallEvent(execution, event) case "tool_update": - if m.historyManager != nil && execution.historyEnabled && execution.SessionID != "" && event.ToolStatus == toolStatusComplete { - if err := m.historyManager.AppendToolResult(execution.SessionID, event); err != nil { - m.logger.Warn("failed to store tool result to history", zap.Error(err)) - } - } + m.handleToolUpdateEvent(execution, event) case "plan": m.logger.Debug("agent plan update", zap.String("execution_id", execution.ID)) case "error": - m.flushMessageBuffer(execution) - m.logger.Error("agent error", - zap.String("execution_id", execution.ID), - zap.String("error", event.Error), - zap.String("text", event.Text), - zap.Any("data", event.Data)) - // Signal prompt completion as an error so waitForPromptDone unblocks - // and the session state transitions from RUNNING to FAILED. - m.handleCompleteEventMarkState(execution, &event, true) - handleCompleteEventSignal(execution, &event, true) - return // Don't publish raw error event to frontend stream; the agent failure - // path (handleAgentFailed) sets session FAILED with the error message. + m.handleErrorEvent(execution, event) + return case "complete": m.handleCompleteEvent(execution, &event) diff --git a/apps/backend/internal/agent/lifecycle/types.go b/apps/backend/internal/agent/lifecycle/types.go index 5cb2f37205..54ed9deea7 100644 --- a/apps/backend/internal/agent/lifecycle/types.go +++ b/apps/backend/internal/agent/lifecycle/types.go @@ -83,6 +83,9 @@ type AgentExecution struct { lastActivityAt time.Time lastActivityAtMu sync.Mutex + // Fires once on the first agent event to publish AgentRunning. + firstActivityOnce sync.Once + // Session-level trace span for grouping all operations under one trace sessionSpan trace.Span sessionSpanMu sync.RWMutex diff --git a/apps/backend/internal/gateway/websocket/task_notifications.go b/apps/backend/internal/gateway/websocket/task_notifications.go index 906a7d0f24..f2cfb02692 100644 --- a/apps/backend/internal/gateway/websocket/task_notifications.go +++ b/apps/backend/internal/gateway/websocket/task_notifications.go @@ -101,7 +101,7 @@ func (b *TaskEventBroadcaster) subscribe(eventBus bus.EventBus, subject, action if action == ws.ActionSessionStateChanged { if data, ok := event.Data.(map[string]interface{}); ok { if metadata, ok := data["metadata"]; ok { - b.logger.Info("received session.state_changed with metadata", + b.logger.Debug("received session.state_changed with metadata", zap.String("action", action), zap.String("session_id", sessionID), zap.Any("metadata", metadata)) @@ -110,11 +110,16 @@ func (b *TaskEventBroadcaster) subscribe(eventBus bus.EventBus, subject, action } switch action { - case ws.ActionSessionAgentctlStarting, ws.ActionSessionAgentctlReady, ws.ActionSessionAgentctlError, ws.ActionSessionStateChanged: + case ws.ActionSessionAgentctlStarting, ws.ActionSessionAgentctlReady, ws.ActionSessionAgentctlError: if sessionID != "" { b.hub.BroadcastToSession(sessionID, msg) return nil } + case ws.ActionSessionStateChanged: + // Broadcast globally so the sidebar task switcher can track + // session state changes for all tasks, not just the active one. + b.hub.Broadcast(msg) + return nil case ws.ActionSessionMessageAdded, ws.ActionSessionMessageUpdated: if sessionID != "" { b.hub.BroadcastToSession(sessionID, msg) diff --git a/apps/backend/internal/orchestrator/event_handlers.go b/apps/backend/internal/orchestrator/event_handlers.go index a6b96c97ac..2fddf5deb6 100644 --- a/apps/backend/internal/orchestrator/event_handlers.go +++ b/apps/backend/internal/orchestrator/event_handlers.go @@ -16,6 +16,7 @@ import ( const ( agentEventComplete = "complete" agentEventError = "error" + agentEventToolCall = "tool_call" ) // buildTaskEventPayload builds the standard map payload for TaskUpdated events. diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go index 22d1f98376..cfd7725261 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent.go +++ b/apps/backend/internal/orchestrator/event_handlers_agent.go @@ -34,7 +34,12 @@ func (s *Service) handleAgentRunning(ctx context.Context, data watcher.AgentEven zap.Error(err)) return } - s.processOnTurnStartViaEngine(ctx, data.TaskID, session) + // on_turn_start is only needed for passthrough sessions where there's + // no PromptTask call to handle it. For ACP sessions, PromptTask or + // dispatchPromptAsync already processes on_turn_start. + if s.agentManager.IsPassthroughSession(ctx, data.SessionID) { + s.processOnTurnStartViaEngine(ctx, data.TaskID, session) + } // Move session to running and task to in progress s.setSessionRunning(ctx, data.TaskID, data.SessionID, session) diff --git a/apps/backend/internal/orchestrator/event_handlers_streaming.go b/apps/backend/internal/orchestrator/event_handlers_streaming.go index 8aee54ae85..478777c358 100644 --- a/apps/backend/internal/orchestrator/event_handlers_streaming.go +++ b/apps/backend/internal/orchestrator/event_handlers_streaming.go @@ -36,7 +36,7 @@ func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle case "thinking_streaming": s.handleThinkingStreamingEvent(ctx, payload) - case "tool_call": + case agentEventToolCall: s.saveAgentTextIfPresent(ctx, payload) s.handleToolCallEvent(ctx, payload) @@ -329,7 +329,6 @@ func (s *Service) handleToolUpdateEvent(ctx context.Context, payload *lifecycle. // goroutine may have changed the state in the DB. The guards are best-effort to avoid // unnecessary writes; the DB update via UpdateTaskSessionState is the atomic source of truth. func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID string, nextState models.TaskSessionState, errorMessage string, allowWakeFromWaiting bool, preloadedSession ...*models.TaskSession) { - var session *models.TaskSession if len(preloadedSession) > 0 && preloadedSession[0] != nil { session = preloadedSession[0] diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index 5465dfc3f5..544cd55310 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_test.go @@ -756,6 +756,93 @@ func TestCleanupAgentExecution_SkipsEmptyExecutionID(t *testing.T) { } } +func TestHandleAgentRunning_PassthroughGuard(t *testing.T) { + ctx := context.Background() + + t.Run("ACP session skips on_turn_start", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0, + Events: wfmodels.StepEvents{ + OnTurnStart: []wfmodels.OnTurnStartAction{ + {Type: wfmodels.OnTurnStartMoveToNext}, + }, + }, + } + stepGetter.steps["step2"] = &wfmodels.WorkflowStep{ + ID: "step2", WorkflowID: "wf1", Name: "Step 2", Position: 1, + } + + taskRepo := newMockTaskRepo() + agentMgr := &mockAgentManager{isPassthrough: false} + svc := createTestServiceWithAgent(repo, stepGetter, taskRepo, agentMgr) + + svc.handleAgentRunning(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: "s1"}) + + // Workflow step must remain step1 because on_turn_start is skipped for ACP sessions. + updated, err := repo.GetTaskSession(ctx, "s1") + if err != nil { + t.Fatalf("failed to get session: %v", err) + } + if updated.WorkflowStepID == nil || *updated.WorkflowStepID != "step1" { + got := "" + if updated.WorkflowStepID != nil { + got = *updated.WorkflowStepID + } + t.Errorf("expected session workflow step to remain 'step1', got %s", got) + } + }) + + t.Run("passthrough session fires on_turn_start", func(t *testing.T) { + repo := setupTestRepo(t) + seedSession(t, repo, "t1", "s1", "step1") + + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0, + Events: wfmodels.StepEvents{ + OnTurnStart: []wfmodels.OnTurnStartAction{ + {Type: wfmodels.OnTurnStartMoveToNext}, + }, + }, + } + stepGetter.steps["step2"] = &wfmodels.WorkflowStep{ + ID: "step2", WorkflowID: "wf1", Name: "Step 2", Position: 1, + } + + taskRepo := newMockTaskRepo() + agentMgr := &mockAgentManager{isPassthrough: true} + svc := createTestServiceWithAgent(repo, stepGetter, taskRepo, agentMgr) + + svc.handleAgentRunning(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: "s1"}) + + // Workflow step must move to step2 because passthrough sessions fire on_turn_start. + updated, err := repo.GetTaskSession(ctx, "s1") + if err != nil { + t.Fatalf("failed to get session: %v", err) + } + if updated.WorkflowStepID == nil || *updated.WorkflowStepID != "step2" { + got := "" + if updated.WorkflowStepID != nil { + got = *updated.WorkflowStepID + } + t.Errorf("expected session workflow step to be 'step2', got %s", got) + } + }) + + t.Run("missing session_id is ignored", func(t *testing.T) { + repo := setupTestRepo(t) + taskRepo := newMockTaskRepo() + svc := createTestService(repo, newMockStepGetter(), taskRepo) + + // Should not panic or error with empty session ID. + svc.handleAgentRunning(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: ""}) + }) +} + // waitForStopCall polls until the mock agent manager has received at least one // StopAgentWithReason call, or fails the test after a timeout. func waitForStopCall(t *testing.T, agentMgr *mockAgentManager) { diff --git a/apps/web/e2e/pages/session-page.ts b/apps/web/e2e/pages/session-page.ts index 2bae682218..ed9d0e50eb 100644 --- a/apps/web/e2e/pages/session-page.ts +++ b/apps/web/e2e/pages/session-page.ts @@ -31,6 +31,14 @@ export class SessionPage { return this.sidebar.getByTestId(`sidebar-section-${label}`); } + /** Task title scoped to a specific sidebar section (Review, In Progress, Backlog). */ + taskInSection(title: string, sectionLabel: string): Locator { + return this.sidebar + .getByTestId(`sidebar-section-${sectionLabel}`) + .locator("..") + .getByText(title, { exact: false }); + } + /** Agent STARTING or RUNNING status indicator. */ agentStatus(): Locator { return this.page.getByRole("status", { name: /Agent is (starting|running)/ }); diff --git a/apps/web/e2e/tests/task-switcher-status.spec.ts b/apps/web/e2e/tests/task-switcher-status.spec.ts new file mode 100644 index 0000000000..f68d6eb3f6 --- /dev/null +++ b/apps/web/e2e/tests/task-switcher-status.spec.ts @@ -0,0 +1,133 @@ +import { test, expect } from "../fixtures/test-base"; +import { KanbanPage } from "../pages/kanban-page"; +import { SessionPage } from "../pages/session-page"; + +test.describe("Task switcher sidebar status", () => { + /** + * Verifies that the sidebar task switcher reflects real-time status updates + * for tasks the user hasn't clicked, and that status is preserved correctly + * when switching between tasks. + * + * This tests the fix for stale sidebar status: without the fix, + * session.state_changed WS events were only sent to the active session's + * subscribers. Tasks completing in the background would show outdated + * status until the user clicked them to trigger a fresh fetch. + * + * Setup: + * Inbox → Working (auto_start, message+15s delay, on_turn_complete → Done) → Done + * 3 tasks in Inbox. + * + * The mock agent sends a message immediately (triggers RUNNING / "In Progress"), + * then delays 15s (stays in RUNNING), then completes (→ WAITING_FOR_INPUT / "Review"). + * Each task progresses through: Backlog → In Progress → Review. + */ + test("sidebar reflects real-time status updates and preserves state on task switch", async ({ + testPage, + apiClient, + seedData, + }) => { + test.setTimeout(120_000); + + // --- Seed workflow --- + const workflow = await apiClient.createWorkflow( + seedData.workspaceId, + "Switcher Status Workflow", + ); + + const inboxStep = await apiClient.createWorkflowStep(workflow.id, "Inbox", 0); + const workingStep = await apiClient.createWorkflowStep(workflow.id, "Working", 1); + const doneStep = await apiClient.createWorkflowStep(workflow.id, "Done", 2); + + // message first (triggers RUNNING), delay 15s, then complete + await apiClient.updateWorkflowStep(workingStep.id, { + prompt: + 'e2e:message("working...")\ne2e:delay(15000)\ne2e:message("task finished")\n{{task_prompt}}', + events: { + on_enter: [{ type: "auto_start_agent" }], + on_turn_complete: [{ type: "move_to_step", config: { step_id: doneStep.id } }], + }, + }); + + await apiClient.saveUserSettings({ + workspace_id: seedData.workspaceId, + workflow_filter_id: workflow.id, + enable_preview_on_click: false, + }); + + // --- Seed 3 tasks --- + const taskAlpha = await apiClient.createTask(seedData.workspaceId, "Alpha Task", { + workflow_id: workflow.id, + workflow_step_id: inboxStep.id, + agent_profile_id: seedData.agentProfileId, + repository_ids: [seedData.repositoryId], + }); + + const taskBeta = await apiClient.createTask(seedData.workspaceId, "Beta Task", { + workflow_id: workflow.id, + workflow_step_id: inboxStep.id, + agent_profile_id: seedData.agentProfileId, + repository_ids: [seedData.repositoryId], + }); + + const taskGamma = await apiClient.createTask(seedData.workspaceId, "Gamma Task", { + workflow_id: workflow.id, + workflow_step_id: inboxStep.id, + agent_profile_id: seedData.agentProfileId, + repository_ids: [seedData.repositoryId], + }); + + // --- Complete Alpha first, then navigate to its session --- + await apiClient.moveTask(taskAlpha.id, workflow.id, workingStep.id); + + const kanban = new KanbanPage(testPage); + await kanban.goto(); + + const alphaInDone = kanban.taskCardInColumn("Alpha Task", doneStep.id); + await expect(alphaInDone).toBeVisible({ timeout: 45_000 }); + + await alphaInDone.click(); + await expect(testPage).toHaveURL(/\/s\//, { timeout: 15_000 }); + + const session = new SessionPage(testPage); + await session.waitForLoad(); + + // --- Initial state: Alpha in Review, Beta + Gamma in Backlog --- + await expect(session.taskInSection("Alpha Task", "Review")).toBeVisible({ timeout: 15_000 }); + await expect(session.taskInSection("Beta Task", "Backlog")).toBeVisible({ timeout: 10_000 }); + await expect(session.taskInSection("Gamma Task", "Backlog")).toBeVisible({ timeout: 10_000 }); + + // --- Move Beta to Working (background task) --- + await apiClient.moveTask(taskBeta.id, workflow.id, workingStep.id); + + // Beta should appear in "In Progress" (RUNNING state) while the 15s delay runs + await expect(session.taskInSection("Beta Task", "In Progress")).toBeVisible({ + timeout: 30_000, + }); + + // While Beta is in progress, Alpha should remain in Review and Gamma in Backlog + await expect(session.taskInSection("Alpha Task", "Review")).toBeVisible({ timeout: 5_000 }); + await expect(session.taskInSection("Gamma Task", "Backlog")).toBeVisible({ timeout: 5_000 }); + + // Beta transitions from "In Progress" → "Review" after the 15s delay completes + await expect(session.taskInSection("Beta Task", "Review")).toBeVisible({ timeout: 30_000 }); + + // --- Move Gamma to Working (background task) --- + await apiClient.moveTask(taskGamma.id, workflow.id, workingStep.id); + + // Gamma should appear in "In Progress" while running + await expect(session.taskInSection("Gamma Task", "In Progress")).toBeVisible({ + timeout: 30_000, + }); + + // Gamma transitions to "Review" + await expect(session.taskInSection("Gamma Task", "Review")).toBeVisible({ timeout: 30_000 }); + + // --- Switch to Beta's session by clicking in sidebar --- + await session.taskInSidebar("Beta Task").click(); + + // All three tasks should remain in "Review" after switching sessions + await expect(session.taskInSection("Alpha Task", "Review")).toBeVisible({ timeout: 10_000 }); + await expect(session.taskInSection("Beta Task", "Review")).toBeVisible({ timeout: 10_000 }); + await expect(session.taskInSection("Gamma Task", "Review")).toBeVisible({ timeout: 10_000 }); + }); +}); diff --git a/apps/web/lib/ws/handlers/agent-session.ts b/apps/web/lib/ws/handlers/agent-session.ts index 761a53e9df..2ce11ab748 100644 --- a/apps/web/lib/ws/handlers/agent-session.ts +++ b/apps/web/lib/ws/handlers/agent-session.ts @@ -29,10 +29,28 @@ function upsertTaskSessionList( sessionUpdate: Record, ): void { const sessionsByTask = store.getState().taskSessionsByTask.itemsByTaskId[taskId]; - if (!sessionsByTask) return; + const newState = payload.new_state as TaskSessionState | undefined; + + if (!sessionsByTask) { + // Initialize per-task sessions from WS event so the sidebar can + // display up-to-date status even for tasks not yet clicked. + if (newState) { + store.getState().setTaskSessionsForTask(taskId, [ + { + id: sessionId, + task_id: taskId, + state: newState, + started_at: "", + updated_at: "", + ...(payload.agent_profile_id ? { agent_profile_id: payload.agent_profile_id } : {}), + ...sessionUpdate, + }, + ]); + } + return; + } const hasSession = sessionsByTask.some((s) => s.id === sessionId); - const newState = payload.new_state as TaskSessionState | undefined; if (!hasSession && newState) { store.getState().setTaskSessionsForTask(taskId, [ @@ -114,7 +132,6 @@ export function registerTaskSessionHandlers(store: StoreApi): WsHandle const sessionId = payload.session_id; const isQueued = payload.is_queued as boolean; const queuedMessage = payload.message as QueuedMessage | null | undefined; - console.log("[Queue] Status changed:", { sessionId, isQueued, hasMessage: !!queuedMessage }); store.getState().setQueueStatus(sessionId, { is_queued: isQueued, message: queuedMessage }); }, "session.state_changed": (message) => {