fix: sidebar task switcher shows outdated session state (#349)

This commit is contained in:
Carlos Florêncio
2026-03-02 20:15:48 +00:00
committed by GitHub
parent fb50b726df
commit 2aab682cd8
12 changed files with 317 additions and 30 deletions
+5 -2
View File
@@ -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)
}
}
+6
View File
@@ -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,
@@ -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)
@@ -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
@@ -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)
@@ -16,6 +16,7 @@ import (
const (
agentEventComplete = "complete"
agentEventError = "error"
agentEventToolCall = "tool_call"
)
// buildTaskEventPayload builds the standard map payload for TaskUpdated events.
@@ -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)
@@ -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]
@@ -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 := "<nil>"
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 := "<nil>"
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) {
+8
View File
@@ -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)/ });
@@ -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 });
});
});
+20 -3
View File
@@ -29,10 +29,28 @@ function upsertTaskSessionList(
sessionUpdate: Record<string, unknown>,
): 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<AppState>): 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) => {