diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 9c74743694..6f8e5ea284 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -1482,6 +1482,46 @@ func (s *Service) publishSessionWaitingEvent(ctx context.Context, taskID, sessio )) } +// publishSessionCreatedEvent publishes a session state change event for CREATED. +// PrepareTaskSession only writes the row to the DB without going through +// updateTaskSessionState, so without this the frontend's per-task session list +// stays empty until a manual reload (e.g. the kanban preview "No agents yet" +// staleness bug). Mirrors publishSessionWaitingEvent's payload shape so the +// existing session.state_changed handler can upsert the new session into the +// store. +func (s *Service) publishSessionCreatedEvent(ctx context.Context, taskID, sessionID, stepID string) { + if s.eventBus == nil { + return + } + eventData := map[string]interface{}{ + "task_id": taskID, + "session_id": sessionID, + "new_state": string(models.TaskSessionStateCreated), + } + if stepID != "" { + eventData["workflow_step_id"] = stepID + } + if session, err := s.repo.GetTaskSession(ctx, sessionID); err == nil && session != nil { + if session.AgentProfileID != "" { + eventData["agent_profile_id"] = session.AgentProfileID + } + if len(session.AgentProfileSnapshot) > 0 { + eventData["agent_profile_snapshot"] = session.AgentProfileSnapshot + } + if session.TaskEnvironmentID != "" { + eventData["task_environment_id"] = session.TaskEnvironmentID + } + if len(session.Metadata) > 0 { + eventData["session_metadata"] = session.Metadata + } + } + _ = s.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent( + events.TaskSessionStateChanged, + "orchestrator", + eventData, + )) +} + // resolveTurnStartTargetStep resolves the target step ID for an on_turn_start transition action. // Returns the step ID and true if resolved; empty string and false if not resolvable. func (s *Service) resolveTurnStartTargetStep(ctx context.Context, currentStep *wfmodels.WorkflowStep, action *wfmodels.OnTurnStartAction) (string, bool) { diff --git a/apps/backend/internal/orchestrator/handlers/handlers.go b/apps/backend/internal/orchestrator/handlers/handlers.go index 009f41554f..44295c38e6 100644 --- a/apps/backend/internal/orchestrator/handlers/handlers.go +++ b/apps/backend/internal/orchestrator/handlers/handlers.go @@ -3,6 +3,7 @@ package handlers import ( "context" + "strings" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/orchestrator" @@ -40,6 +41,7 @@ func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { d.RegisterFunc(ws.ActionTaskSessionPrepare, h.wsPrepareTaskSession) d.RegisterFunc(ws.ActionAgentCancel, h.wsCancelAgent) d.RegisterFunc(ws.ActionSessionLaunch, h.wsLaunchSession) + d.RegisterFunc(ws.ActionSessionEnsure, h.wsEnsureSession) d.RegisterFunc(ws.ActionSessionRecover, h.wsRecoverSession) d.RegisterFunc(ws.ActionSessionResetContext, h.wsResetContext) d.RegisterFunc(ws.ActionSessionStop, h.wsStopSession) @@ -119,6 +121,35 @@ func (h *Handlers) wsLaunchSession(ctx context.Context, msg *ws.Message) (*ws.Me return ws.NewResponse(msg.ID, msg.Action, resp) } +type wsEnsureSessionRequest struct { + TaskID string `json:"task_id"` +} + +func (h *Handlers) wsEnsureSession(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsEnsureSessionRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + + resp, err := h.service.EnsureSession(ctx, req.TaskID) + if err != nil { + h.logger.Error("failed to ensure session", + zap.String("task_id", req.TaskID), + zap.Error(err)) + // Mirror httpEnsureTaskSession's NotFound mapping so the frontend can + // distinguish unknown task ids from real server errors. EnsureSession + // wraps the repo error as "task not found: %w". + if strings.Contains(err.Error(), "task not found") { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "Task not found", nil) + } + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to ensure session: "+err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + type wsResetContextRequest struct { SessionID string `json:"session_id"` } diff --git a/apps/backend/internal/orchestrator/session_ensure.go b/apps/backend/internal/orchestrator/session_ensure.go new file mode 100644 index 0000000000..547906a0f0 --- /dev/null +++ b/apps/backend/internal/orchestrator/session_ensure.go @@ -0,0 +1,162 @@ +package orchestrator + +import ( + "context" + "fmt" + "sync" + + "github.com/kandev/kandev/internal/task/models" + wfmodels "github.com/kandev/kandev/internal/workflow/models" +) + +// EnsureSessionResponse describes the outcome of EnsureSession. +type EnsureSessionResponse struct { + Success bool `json:"success"` + TaskID string `json:"task_id"` + SessionID string `json:"session_id,omitempty"` + State string `json:"state"` + AgentProfileID string `json:"agent_profile_id,omitempty"` + Source string `json:"source"` // existing_primary | existing_newest | created_prepare | created_start + NewlyCreated bool `json:"newly_created"` // true when a new session was created by this call +} + +// ensureLocks serializes EnsureSession calls per task id so concurrent callers +// observe the same session rather than racing to create duplicates. Entries are +// not deleted on release: deletion would race with a concurrent waiter +// (it could acquire the about-to-be-discarded mutex while a new caller LoadOrStores +// a fresh one, putting two goroutines in the critical section for the same task). +// Growth is bounded by the number of distinct task IDs (~160 B per entry). +var ensureLocks sync.Map // map[taskID]*sync.Mutex + +func acquireEnsureLock(taskID string) func() { + v, _ := ensureLocks.LoadOrStore(taskID, &sync.Mutex{}) + mu := v.(*sync.Mutex) + mu.Lock() + return mu.Unlock +} + +// EnsureSession is the server-authoritative idempotent entry point for opening +// a task: it returns the existing primary (or newest) session if any, otherwise +// resolves the agent profile from the task's full context and creates a session +// via prepare (workspace-only) or start (with agent), gated by the task's +// workflow step. +func (s *Service) EnsureSession(ctx context.Context, taskID string) (*EnsureSessionResponse, error) { + if taskID == "" { + return nil, fmt.Errorf("task_id is required") + } + release := acquireEnsureLock(taskID) + defer release() + + if existing := s.findExistingSession(ctx, taskID); existing != nil { + return existing, nil + } + + task, err := s.repo.GetTask(ctx, taskID) + if err != nil { + return nil, fmt.Errorf("task not found: %w", err) + } + + agentProfileID, step := s.resolveTaskAgentProfile(ctx, task) + autoStart := stepAllowsAutoStart(step) + + intent := IntentPrepare + source := "created_prepare" + if agentProfileID != "" && autoStart { + intent = IntentStart + source = "created_start" + } + + launchResp, err := s.LaunchSession(ctx, &LaunchSessionRequest{ + TaskID: taskID, + Intent: intent, + AgentProfileID: agentProfileID, + WorkflowStepID: task.WorkflowStepID, + LaunchWorkspace: true, + AutoStart: intent == IntentStart, + }) + if err != nil { + return nil, err + } + + return &EnsureSessionResponse{ + Success: true, + TaskID: taskID, + SessionID: launchResp.SessionID, + State: launchResp.State, + AgentProfileID: agentProfileID, + Source: source, + NewlyCreated: true, + }, nil +} + +// findExistingSession returns the task's existing primary session (or the +// newest if none is marked primary). Returns nil when the task has no sessions. +func (s *Service) findExistingSession(ctx context.Context, taskID string) *EnsureSessionResponse { + sessions, err := s.repo.ListTaskSessions(ctx, taskID) + if err != nil || len(sessions) == 0 { + return nil + } + for _, sess := range sessions { + if sess.IsPrimary { + return existingResponse(taskID, sess, "existing_primary") + } + } + // ListTaskSessions returns rows ordered by started_at DESC. + return existingResponse(taskID, sessions[0], "existing_newest") +} + +func existingResponse(taskID string, sess *models.TaskSession, source string) *EnsureSessionResponse { + return &EnsureSessionResponse{ + Success: true, + TaskID: taskID, + SessionID: sess.ID, + State: string(sess.State), + AgentProfileID: sess.AgentProfileID, + Source: source, + NewlyCreated: false, + } +} + +// resolveTaskAgentProfile applies the 4-step resolution chain on the backend: +// 1) task.metadata.agent_profile_id, 2) workflow step override, +// 3) workflow default, 4) workspace default. Returns the resolved profile id +// (or "" when none resolve) along with the workflow step it loaded (or nil). +// Returning the step lets callers reuse it (e.g. to gate auto-start) without a +// second DB lookup. +func (s *Service) resolveTaskAgentProfile(ctx context.Context, task *models.Task) (string, *wfmodels.WorkflowStep) { + step := s.lookupWorkflowStep(ctx, task.WorkflowStepID) + if v, ok := task.Metadata["agent_profile_id"].(string); ok && v != "" { + return v, step + } + if step != nil { + if id := s.resolveStepAgentProfile(ctx, step); id != "" { + return id, step + } + } + ws, err := s.repo.GetWorkspace(ctx, task.WorkspaceID) + if err == nil && ws != nil && ws.DefaultAgentProfileID != nil && *ws.DefaultAgentProfileID != "" { + return *ws.DefaultAgentProfileID, step + } + return "", step +} + +func (s *Service) lookupWorkflowStep(ctx context.Context, stepID string) *wfmodels.WorkflowStep { + if stepID == "" || s.workflowStepGetter == nil { + return nil + } + step, err := s.workflowStepGetter.GetStep(ctx, stepID) + if err != nil { + return nil + } + return step +} + +// stepAllowsAutoStart reports whether the workflow step (if any) has the +// auto_start_agent on-enter action. Tasks without a workflow step default to +// allowing auto-start (mirrors shouldBlockAutoStart's behavior). +func stepAllowsAutoStart(step *wfmodels.WorkflowStep) bool { + if step == nil { + return true + } + return step.HasOnEnterAction(wfmodels.OnEnterAutoStartAgent) +} diff --git a/apps/backend/internal/orchestrator/session_ensure_test.go b/apps/backend/internal/orchestrator/session_ensure_test.go new file mode 100644 index 0000000000..502c736598 --- /dev/null +++ b/apps/backend/internal/orchestrator/session_ensure_test.go @@ -0,0 +1,374 @@ +package orchestrator + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/task/models" + wfmodels "github.com/kandev/kandev/internal/workflow/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +func TestEnsureSession_RequiresTaskID(t *testing.T) { + repo := setupTestRepo(t) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + + if _, err := svc.EnsureSession(context.Background(), ""); err == nil { + t.Fatal("expected error when task_id is empty") + } +} + +func TestEnsureSession_TaskNotFound(t *testing.T) { + repo := setupTestRepo(t) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + + if _, err := svc.EnsureSession(context.Background(), "missing"); err == nil { + t.Fatal("expected error when task is missing") + } +} + +func TestEnsureSession_ReturnsExistingPrimary(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + + seedTaskAndSession(t, repo, "task1", "session-old", models.TaskSessionStateCompleted) + // Add a newer non-primary session, then mark the older one primary. + now := time.Now().UTC() + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-new", TaskID: "task1", State: models.TaskSessionStateRunning, + StartedAt: now.Add(time.Minute), UpdatedAt: now.Add(time.Minute), + }); err != nil { + t.Fatalf("failed to create newer session: %v", err) + } + if err := repo.SetSessionPrimary(ctx, "session-old"); err != nil { + t.Fatalf("failed to mark primary: %v", err) + } + + resp, err := svc.EnsureSession(ctx, "task1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if resp.SessionID != "session-old" { + t.Errorf("expected primary session-old, got %q", resp.SessionID) + } + if resp.Source != "existing_primary" { + t.Errorf("expected source=existing_primary, got %q", resp.Source) + } + if resp.NewlyCreated { + t.Error("expected NewlyCreated=false") + } +} + +func TestEnsureSession_ReturnsExistingNewest_NoPrimary(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + + seedTaskAndSession(t, repo, "task1", "session-old", models.TaskSessionStateCompleted) + now := time.Now().UTC() + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-new", TaskID: "task1", State: models.TaskSessionStateRunning, + StartedAt: now.Add(time.Minute), UpdatedAt: now.Add(time.Minute), + }); err != nil { + t.Fatalf("failed to create newer session: %v", err) + } + + resp, err := svc.EnsureSession(ctx, "task1") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if resp.SessionID != "session-new" { + t.Errorf("expected newest session-new, got %q", resp.SessionID) + } + if resp.Source != "existing_newest" { + t.Errorf("expected source=existing_newest, got %q", resp.Source) + } +} + +func TestEnsureSession_Concurrent_ReturnsSameExistingSession(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateRunning) + + const N = 8 + var wg sync.WaitGroup + results := make([]string, N) + wg.Add(N) + for i := 0; i < N; i++ { + go func(idx int) { + defer wg.Done() + resp, err := svc.EnsureSession(ctx, "task1") + if err != nil { + t.Errorf("concurrent ensure failed: %v", err) + return + } + results[idx] = resp.SessionID + }(i) + } + wg.Wait() + + for i, sid := range results { + if sid != "session1" { + t.Errorf("call %d: expected session1, got %q", i, sid) + } + } +} + +// acquireEnsureLock must serialize callers per task id. The previous attempt +// to bound map growth (Delete on release) opened a window where a third +// caller could LoadOrStore a fresh mutex while a second caller still held +// the about-to-be-discarded one, putting two goroutines in the critical +// section at the same time. This test pins down that property by counting +// the maximum observed concurrency under the same task id. +func TestAcquireEnsureLock_SerializesPerTaskID(t *testing.T) { + const N = 16 + var ( + active int + mu sync.Mutex + maxCon int + wg sync.WaitGroup + ) + wg.Add(N) + for i := 0; i < N; i++ { + go func() { + defer wg.Done() + release := acquireEnsureLock("task-x") + defer release() + mu.Lock() + active++ + if active > maxCon { + maxCon = active + } + mu.Unlock() + time.Sleep(2 * time.Millisecond) + mu.Lock() + active-- + mu.Unlock() + }() + } + wg.Wait() + if maxCon != 1 { + t.Errorf("expected max concurrency 1 under same task id, got %d", maxCon) + } +} + +// Distinct task ids must NOT serialize against each other. Uses a +// channel-based rendezvous (rather than a sleep) so the assertion is +// deterministic on slow CI runners: every goroutine signals after acquiring +// its lock, the test releases them all at once, and only then do they +// observe peak concurrency. +func TestAcquireEnsureLock_AllowsConcurrencyAcrossTaskIDs(t *testing.T) { + const N = 8 + acquired := make(chan struct{}, N) + start := make(chan struct{}) + var holding int + var mu sync.Mutex + var peak int + var wg sync.WaitGroup + wg.Add(N) + for i := 0; i < N; i++ { + taskID := fmt.Sprintf("task-%d", i) + go func() { + defer wg.Done() + release := acquireEnsureLock(taskID) + defer release() + acquired <- struct{}{} + <-start + mu.Lock() + holding++ + if holding > peak { + peak = holding + } + mu.Unlock() + time.Sleep(2 * time.Millisecond) + mu.Lock() + holding-- + mu.Unlock() + }() + } + for i := 0; i < N; i++ { + <-acquired + } + close(start) + wg.Wait() + if peak < 2 { + t.Errorf("expected concurrency across distinct task ids, peak=%d", peak) + } +} + +// Service-level companion to the lock primitive tests: with no pre-existing +// session, N concurrent EnsureSession calls must converge on exactly one +// created session. This catches regressions in findExistingSession or +// LaunchSession idempotency that the lock tests alone wouldn't surface. +func TestEnsureSession_Concurrent_CreatesSingleSession(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + taskRepo := newMockTaskRepo() + // Stub LaunchAgent so the background workspace launch goroutine spawned + // by PrepareTaskSession completes without nil-deref'ing on the response. + agentMgr := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, _ *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + return &executor.LaunchAgentResponse{AgentExecutionID: "exec-1"}, nil + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + + now := time.Now().UTC() + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("seed workspace: %v", err) + } + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "Test Workflow", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("seed workflow: %v", err) + } + if err := repo.CreateTask(ctx, &models.Task{ID: "task1", WorkflowID: "wf1", WorkspaceID: "ws1", Title: "T", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("seed task: %v", err) + } + // PrepareTaskSession resolves agent_profile_id from task metadata via + // scheduler.GetTask (backed by the mock taskRepo), so seed both stores. + taskRepo.tasks["task1"] = &v1.Task{ + ID: "task1", + WorkspaceID: "ws1", + Metadata: map[string]any{"agent_profile_id": "profile1"}, + } + + const N = 8 + var wg sync.WaitGroup + results := make([]string, N) + errs := make([]error, N) + wg.Add(N) + for i := 0; i < N; i++ { + go func(idx int) { + defer wg.Done() + resp, err := svc.EnsureSession(ctx, "task1") + if err != nil { + errs[idx] = err + return + } + results[idx] = resp.SessionID + }(i) + } + wg.Wait() + + for i, err := range errs { + if err != nil { + t.Fatalf("call %d failed: %v", i, err) + } + } + first := results[0] + if first == "" { + t.Fatal("expected a session id from the first caller") + } + for i, sid := range results { + if sid != first { + t.Errorf("call %d: expected %q, got %q", i, first, sid) + } + } + + sessions, err := repo.ListTaskSessions(ctx, "task1") + if err != nil { + t.Fatalf("list sessions: %v", err) + } + if len(sessions) != 1 { + t.Errorf("expected exactly 1 session created, got %d", len(sessions)) + } +} + +func TestStepAllowsAutoStart(t *testing.T) { + if !stepAllowsAutoStart(nil) { + t.Error("expected nil step to allow auto-start (no workflow step constraint)") + } + stepWith := &wfmodels.WorkflowStep{ + Events: wfmodels.StepEvents{ + OnEnter: []wfmodels.OnEnterAction{{Type: wfmodels.OnEnterAutoStartAgent}}, + }, + } + if !stepAllowsAutoStart(stepWith) { + t.Error("expected step with auto_start_agent to allow auto-start") + } + stepWithout := &wfmodels.WorkflowStep{} + if stepAllowsAutoStart(stepWithout) { + t.Error("expected step without auto_start_agent to disallow auto-start") + } +} + +func TestResolveTaskAgentProfile_TaskMetadataWins(t *testing.T) { + repo := setupTestRepo(t) + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ID: "step1", AgentProfileID: "step-profile"} + stepGetter.workflowAgentProfileID = "wf-profile" + svc := createTestService(repo, stepGetter, newMockTaskRepo()) + + task := &models.Task{ + ID: "t1", + WorkspaceID: "ws1", + WorkflowStepID: "step1", + Metadata: map[string]interface{}{"agent_profile_id": "task-profile"}, + } + if got, _ := svc.resolveTaskAgentProfile(context.Background(), task); got != "task-profile" { + t.Errorf("expected task-profile, got %q", got) + } +} + +func TestResolveTaskAgentProfile_StepThenWorkflowThenWorkspace(t *testing.T) { + ctx := context.Background() + now := time.Now().UTC() + + t.Run("step override", func(t *testing.T) { + repo := setupTestRepo(t) + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ID: "step1", AgentProfileID: "step-profile"} + svc := createTestService(repo, stepGetter, newMockTaskRepo()) + task := &models.Task{ID: "t1", WorkflowStepID: "step1"} + if got, _ := svc.resolveTaskAgentProfile(ctx, task); got != "step-profile" { + t.Errorf("expected step-profile, got %q", got) + } + }) + + t.Run("workflow default when step has none", func(t *testing.T) { + repo := setupTestRepo(t) + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ID: "step1", WorkflowID: "wf1"} + stepGetter.workflowAgentProfileID = "wf-profile" + svc := createTestService(repo, stepGetter, newMockTaskRepo()) + task := &models.Task{ID: "t1", WorkflowStepID: "step1"} + if got, _ := svc.resolveTaskAgentProfile(ctx, task); got != "wf-profile" { + t.Errorf("expected wf-profile, got %q", got) + } + }) + + t.Run("workspace default when step+workflow have none", func(t *testing.T) { + repo := setupTestRepo(t) + ws := &models.Workspace{ + ID: "ws-x", Name: "X", DefaultAgentProfileID: strPtr("ws-profile"), + CreatedAt: now, UpdatedAt: now, + } + if err := repo.CreateWorkspace(ctx, ws); err != nil { + t.Fatalf("create workspace: %v", err) + } + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + task := &models.Task{ID: "t1", WorkspaceID: "ws-x"} + if got, _ := svc.resolveTaskAgentProfile(ctx, task); got != "ws-profile" { + t.Errorf("expected ws-profile, got %q", got) + } + }) + + t.Run("returns empty when nothing resolves", func(t *testing.T) { + repo := setupTestRepo(t) + ws := &models.Workspace{ID: "ws-y", Name: "Y", CreatedAt: now, UpdatedAt: now} + if err := repo.CreateWorkspace(ctx, ws); err != nil { + t.Fatalf("create workspace: %v", err) + } + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + task := &models.Task{ID: "t1", WorkspaceID: "ws-y"} + if got, _ := svc.resolveTaskAgentProfile(ctx, task); got != "" { + t.Errorf("expected empty, got %q", got) + } + }) +} diff --git a/apps/backend/internal/orchestrator/session_launch.go b/apps/backend/internal/orchestrator/session_launch.go index a048ab1041..cbb847f6c2 100644 --- a/apps/backend/internal/orchestrator/session_launch.go +++ b/apps/backend/internal/orchestrator/session_launch.go @@ -79,11 +79,6 @@ func (s *Service) LaunchSession(ctx context.Context, req *LaunchSessionRequest) intent := ResolveIntent(req) req.Prompt = strings.TrimSpace(req.Prompt) - s.logger.Debug("LaunchSession", - zap.String("task_id", req.TaskID), - zap.String("intent", string(intent)), - zap.String("session_id", req.SessionID)) - switch intent { case IntentPrepare: return s.launchPrepare(ctx, req) diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 0ffed110b6..fe3d0ae7fd 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -168,6 +168,12 @@ func (s *Service) PrepareTaskSession(ctx context.Context, taskID string, agentPr return "", err } + // Notify the frontend that a new CREATED session exists. The start path + // transitions through updateTaskSessionState which broadcasts; the prepare + // path writes the row directly, so without this the per-task session list + // stays empty until a manual reload. + s.publishSessionCreatedEvent(ctx, taskID, sessionID, workflowStepID) + if launchWorkspace { // Launch workspace infrastructure (agentctl) in the background so the WS response // returns the session ID immediately. The frontend navigates to the session page diff --git a/apps/backend/internal/task/handlers/task_handlers.go b/apps/backend/internal/task/handlers/task_handlers.go index b7e060a82f..f0ba66a05f 100644 --- a/apps/backend/internal/task/handlers/task_handlers.go +++ b/apps/backend/internal/task/handlers/task_handlers.go @@ -40,6 +40,9 @@ func (h *TaskHandlers) SetOnTaskCreatedWithPR(fn func(ctx context.Context, taskI type OrchestratorStarter interface { // LaunchSession is the unified entry point for all session operations. LaunchSession(ctx context.Context, req *orchestrator.LaunchSessionRequest) (*orchestrator.LaunchSessionResponse, error) + // EnsureSession returns the task's existing primary/newest session if any, + // otherwise resolves the agent profile server-side and creates one. + EnsureSession(ctx context.Context, taskID string) (*orchestrator.EnsureSessionResponse, error) } func NewTaskHandlers(svc *service.Service, orchestrator OrchestratorStarter, repo handlerRepo, planService *service.PlanService, log *logger.Logger) *TaskHandlers { @@ -66,6 +69,7 @@ func (h *TaskHandlers) registerHTTP(router *gin.Engine) { api.GET("/tasks/:id", h.httpGetTask) api.GET("/task-sessions/:id", h.httpGetTaskSession) api.GET("/tasks/:id/sessions", h.httpListTaskSessions) + api.POST("/tasks/:id/sessions/ensure", h.httpEnsureTaskSession) api.GET("/tasks/:id/environment", h.httpGetTaskEnvironment) api.GET("/task-sessions/:id/turns", h.httpListSessionTurns) api.POST("/tasks", h.httpCreateTask) diff --git a/apps/backend/internal/task/handlers/task_http_handlers.go b/apps/backend/internal/task/handlers/task_http_handlers.go index f07a437490..94d01ca72a 100644 --- a/apps/backend/internal/task/handlers/task_http_handlers.go +++ b/apps/backend/internal/task/handlers/task_http_handlers.go @@ -209,6 +209,23 @@ func (h *TaskHandlers) httpListTaskSessions(c *gin.Context) { }) } +// httpEnsureTaskSession returns the task's existing primary/newest session if any, +// otherwise resolves the agent profile server-side and creates one (prepare or +// start, depending on the workflow step). Idempotent under concurrent calls. +func (h *TaskHandlers) httpEnsureTaskSession(c *gin.Context) { + taskID := c.Param("id") + if taskID == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "task id is required"}) + return + } + resp, err := h.orchestrator.EnsureSession(c.Request.Context(), taskID) + if err != nil { + handleNotFound(c, h.logger, err, "task not found") + return + } + c.JSON(http.StatusOK, resp) +} + func (h *TaskHandlers) httpGetTaskSession(c *gin.Context) { session, err := h.service.GetTaskSession(c.Request.Context(), c.Param("id")) if err != nil { diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index c688b0ccc1..727cbc0edb 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -80,6 +80,7 @@ const ( // Unified session launch ActionSessionLaunch = "session.launch" + ActionSessionEnsure = "session.ensure" ActionSessionRecover = "session.recover" ActionSessionResetContext = "session.reset_context" ActionSessionStop = "session.stop" diff --git a/apps/web/components/kanban-card.tsx b/apps/web/components/kanban-card.tsx index 717f1d1284..8beafcbe33 100644 --- a/apps/web/components/kanban-card.tsx +++ b/apps/web/components/kanban-card.tsx @@ -439,6 +439,8 @@ export function KanbanCard({ disabled: isMultiSelectMode, }); + const isPreviewed = useAppStore((state) => state.kanbanPreviewedTaskId === task.id); + const style = { transform: CSS.Translate.toString(transform), transition: "none", @@ -471,6 +473,7 @@ export function KanbanCard({ needsAction(task) && !isSelected && "border-l-2 border-l-amber-500", isDragging && "opacity-50 z-50", isSelected && "ring-1 ring-primary/60 border-primary/60", + isPreviewed && !isSelected && "ring-2 ring-primary border-primary", )} onClick={handleClick} {...(!isMultiSelectMode ? listeners : {})} diff --git a/apps/web/components/kanban-with-preview.tsx b/apps/web/components/kanban-with-preview.tsx index 1681d7c9c7..6d723add32 100644 --- a/apps/web/components/kanban-with-preview.tsx +++ b/apps/web/components/kanban-with-preview.tsx @@ -21,8 +21,11 @@ import { Task } from "./kanban-card"; import type { KanbanState } from "@/lib/state/slices"; import { PREVIEW_PANEL } from "@/lib/settings/constants"; import { linkToTask } from "@/lib/links"; -import { launchSession } from "@/lib/services/session-launch-service"; -import { buildPrepareRequest } from "@/lib/services/session-launch-helpers"; +import { findTaskInSnapshots } from "@/lib/kanban/find-task"; +import { + useEnsureTaskSession, + type UseEnsureTaskSessionResult, +} from "@/hooks/domains/session/use-ensure-task-session"; type KanbanWithPreviewProps = { initialTaskId?: string; @@ -122,10 +125,17 @@ function useSessionSelectionReset( function useSelectedTask( selectedTaskId: string | null | undefined, kanbanTasks: KanbanState["tasks"], + snapshots: Record, ) { return useMemo(() => { - if (!selectedTaskId || kanbanTasks.length === 0) return null; - const task = kanbanTasks.find((t: KanbanState["tasks"][number]) => t.id === selectedTaskId); + if (!selectedTaskId) return null; + // The active workflow's tasks live in `kanban.tasks`, but cards from other + // workflows can also appear in the board (multi-workflow swimlane view via + // `kanbanMulti.snapshots`). Fall back to those so cross-workflow previews + // are not auto-closed by the "task no longer exists" guard below. + const task = + kanbanTasks.find((t: KanbanState["tasks"][number]) => t.id === selectedTaskId) ?? + findTaskInSnapshots(selectedTaskId, snapshots); if (!task) return null; return { id: task.id, @@ -137,7 +147,7 @@ function useSelectedTask( repositoryId: task.repositoryId, primarySessionId: task.primarySessionId, }; - }, [selectedTaskId, kanbanTasks]); + }, [selectedTaskId, kanbanTasks, snapshots]); } export function KanbanWithPreview({ initialTaskId, initialSessionId }: KanbanWithPreviewProps) { @@ -146,6 +156,8 @@ export function KanbanWithPreview({ initialTaskId, initialSessionId }: KanbanWit // Get tasks from the kanban store const kanbanTasks = useAppStore((state) => state.kanban.tasks); + const kanbanMultiSnapshots = useAppStore((state) => state.kanbanMulti.snapshots); + const setKanbanPreviewedTaskId = useAppStore((state) => state.setKanbanPreviewedTaskId); const { selectedTaskId, isOpen, previewWidthPx, open, close, updatePreviewWidth } = useKanbanPreview({ @@ -155,9 +167,18 @@ export function KanbanWithPreview({ initialTaskId, initialSessionId }: KanbanWit }, }); + // Mirror the previewed task id into the store so kanban cards can highlight + // the currently-previewed card without prop-drilling through swimlanes. + useEffect(() => { + setKanbanPreviewedTaskId(isOpen ? (selectedTaskId ?? null) : null); + }, [isOpen, selectedTaskId, setKanbanPreviewedTaskId]); + useEffect(() => { + return () => setKanbanPreviewedTaskId(null); + }, [setKanbanPreviewedTaskId]); + // Use custom hooks for layout and session management const { containerRef, shouldFloat, kanbanWidth } = useKanbanLayout(isOpen, previewWidthPx); - const { sessionId: selectedTaskSessionId, isLoading } = useTaskSession(selectedTaskId ?? null); + const { sessionId: selectedTaskSessionId } = useTaskSession(selectedTaskId ?? null); // User-selected tab overrides the default primary session pick. // Reset when the selected task changes. @@ -169,27 +190,20 @@ export function KanbanWithPreview({ initialTaskId, initialSessionId }: KanbanWit // Track resize state const isResizingRef = useRef(false); - const selectedTask = useSelectedTask(selectedTaskId, kanbanTasks); + const selectedTask = useSelectedTask(selectedTaskId, kanbanTasks, kanbanMultiSnapshots); - // Close panel if selected task no longer exists + // Close panel if selected task no longer exists in either the active + // workflow's tasks or any loaded multi-workflow snapshot. useEffect(() => { if (isOpen && selectedTaskId && !selectedTask) { close(); } }, [isOpen, selectedTaskId, selectedTask, close]); - // Prepare workspace when preview opens for a task with no session - const preparedTaskRef = useRef(null); - useEffect(() => { - if (!isOpen || !selectedTaskId || isLoading || selectedTaskSessionId) return; - if (preparedTaskRef.current === selectedTaskId) return; - preparedTaskRef.current = selectedTaskId; - - const { request } = buildPrepareRequest(selectedTaskId); - launchSession(request).catch(() => { - // Prepare failed silently — user can still start agent manually - }); - }, [isOpen, selectedTaskId, selectedTaskSessionId, isLoading]); + // Auto-start a session when the preview opens on a task with no session, + // mirroring the full task page so the preview doesn't dead-end on + // "No agents yet." The hook no-ops when no agent profile resolves. + const ensureSession = useEnsureTaskSession(selectedTask, { enabled: isOpen }); const handleNavigateToTask = useCallback( (task: Task) => { @@ -236,6 +250,7 @@ export function KanbanWithPreview({ initialTaskId, initialSessionId }: KanbanWit previewWidthPx={previewWidthPx} selectedTask={selectedTask} activeSessionId={activeSessionId} + ensureSession={ensureSession} onPreviewTask={handlePreviewTaskWithData} onNavigateToTask={handleNavigateToTask} onClose={close} @@ -249,6 +264,7 @@ export function KanbanWithPreview({ initialTaskId, initialSessionId }: KanbanWit isOpen={isOpen} selectedTask={selectedTask} activeSessionId={activeSessionId} + ensureSession={ensureSession} onPreviewTask={handlePreviewTaskWithData} onNavigateToTask={handleNavigateToTask} onClose={close} @@ -277,6 +293,7 @@ type PreviewLayoutProps = { previewWidthPx: number; selectedTask: Task | null; activeSessionId: string | null; + ensureSession: UseEnsureTaskSessionResult; onPreviewTask: (task: Task) => void; onNavigateToTask: (task: Task) => void; onClose: () => void; @@ -289,6 +306,7 @@ function FloatingPreviewLayout({ previewWidthPx, selectedTask, activeSessionId, + ensureSession, onPreviewTask, onNavigateToTask, onClose, @@ -317,6 +335,7 @@ function FloatingPreviewLayout({ onNavigateToTask(task)} onSessionChange={onSessionChange} @@ -333,6 +352,7 @@ function InlinePreviewLayout({ isOpen, selectedTask, activeSessionId, + ensureSession, onPreviewTask, onNavigateToTask, onClose, @@ -354,6 +374,7 @@ function InlinePreviewLayout({ onNavigateToTask(task)} onSessionChange={onSessionChange} diff --git a/apps/web/components/kanban/kanban-header.tsx b/apps/web/components/kanban/kanban-header.tsx index 3f10233fef..6861f86bd5 100644 --- a/apps/web/components/kanban/kanban-header.tsx +++ b/apps/web/components/kanban/kanban-header.tsx @@ -1,8 +1,8 @@ "use client"; import Link from "next/link"; +import { useEffect, useRef, useState } from "react"; import { useRouter } from "next/navigation"; -import { useState } from "react"; import { Button } from "@kandev/ui/button"; import { ToggleGroup, ToggleGroupItem } from "@kandev/ui/toggle-group"; import { Tooltip, TooltipContent, TooltipProvider, TooltipTrigger } from "@kandev/ui/tooltip"; @@ -269,6 +269,25 @@ function TabletHeader({ ); } +// Width below which the centered search input no longer fits between the +// left ("KanDev" + GitHub/Jira/Stats) and right (Add task + chat + view +// toggle + indicators + display + settings) action groups. +const DESKTOP_HEADER_NARROW_PX = 1100; + +function useIsHeaderNarrow(ref: React.RefObject): boolean { + const [isNarrow, setIsNarrow] = useState(false); + useEffect(() => { + const el = ref.current; + if (!el) return; + const update = () => setIsNarrow(el.clientWidth < DESKTOP_HEADER_NARROW_PX); + update(); + const observer = new ResizeObserver(update); + observer.observe(el); + return () => observer.disconnect(); + }, [ref]); + return isNarrow; +} + function DesktopHeader({ onCreateTask, workspaceId, @@ -294,8 +313,15 @@ function DesktopHeader({ showHealthIndicator: boolean; onOpenHealthDialog: () => void; }) { + // The search input is absolutely centered, so the left/right action groups + // can grow into it and overlap. Hide the search when the header gets too + // narrow to fit all three regions side-by-side (e.g. when the kanban preview + // panel is open and squeezes the board area). + const headerRef = useRef(null); + const isNarrow = useIsHeaderNarrow(headerRef); + const showSearch = !!onSearchChange && !isNarrow; return ( -
+
KanDev @@ -319,8 +345,8 @@ function DesktopHeader({
- {onSearchChange && ( -
+ {showSearch && ( +
state.kanbanPreviewedTaskId === task.id); const statusIcon = getTaskStateIcon(task.state, "h-3 w-3"); return ( @@ -86,6 +87,7 @@ function DraggableTaskChip({ "border border-border/50", needsAction(task) && "border-l-2 border-l-amber-500", isDragging && "opacity-30", + isPreviewed && "ring-2 ring-primary border-primary", )} style={transform ? { transform: `translate(${transform.x}px, ${transform.y}px)` } : undefined} > diff --git a/apps/web/components/task-preview-panel.tsx b/apps/web/components/task-preview-panel.tsx index 7906a524b2..06ac32e73e 100644 --- a/apps/web/components/task-preview-panel.tsx +++ b/apps/web/components/task-preview-panel.tsx @@ -2,12 +2,14 @@ import { IconArrowsMaximize, IconX } from "@tabler/icons-react"; import { Button } from "@kandev/ui/button"; +import type { UseEnsureTaskSessionResult } from "@/hooks/domains/session/use-ensure-task-session"; import type { Task } from "./kanban-card"; import { PreviewSessionTabs } from "./task/preview-session-tabs"; interface TaskPreviewPanelProps { task: Task | null; sessionId?: string | null; + ensureSession?: UseEnsureTaskSessionResult; onClose: () => void; onMaximize?: (task: Task) => void; onSessionChange?: (sessionId: string | null) => void; @@ -16,6 +18,7 @@ interface TaskPreviewPanelProps { export function TaskPreviewPanel({ task, sessionId = null, + ensureSession, onClose, onMaximize, onSessionChange, @@ -54,6 +57,7 @@ export function TaskPreviewPanel({ ) : ( diff --git a/apps/web/components/task/preview-session-tabs.tsx b/apps/web/components/task/preview-session-tabs.tsx index 0e70cb2ca5..d3c2642547 100644 --- a/apps/web/components/task/preview-session-tabs.tsx +++ b/apps/web/components/task/preview-session-tabs.tsx @@ -2,12 +2,14 @@ import { useCallback, useMemo } from "react"; import { IconLoader2 } from "@tabler/icons-react"; +import { Button } from "@kandev/ui/button"; import { AgentLogo } from "@/components/agent-logo"; import { SessionTabs, type SessionTab } from "@/components/session-tabs"; import { useAppStore } from "@/components/state-provider"; import { useToast } from "@/components/toast-provider"; import { useSessionResumption } from "@/hooks/domains/session/use-session-resumption"; import { useTaskSessions } from "@/hooks/use-task-sessions"; +import type { UseEnsureTaskSessionResult } from "@/hooks/domains/session/use-ensure-task-session"; import type { AgentProfileOption } from "@/lib/state/slices"; import type { TaskSession, TaskSessionState } from "@/lib/types/http"; import { getWebSocketClient } from "@/lib/ws/connection"; @@ -25,6 +27,7 @@ const LABEL_SEPARATOR = " \u2022 "; type PreviewSessionTabsProps = { taskId: string; sessionId: string | null; + ensureSession?: UseEnsureTaskSessionResult; onSessionChange?: (sessionId: string | null) => void; }; @@ -37,6 +40,7 @@ type PreviewSessionTabsProps = { export function PreviewSessionTabs({ taskId, sessionId, + ensureSession, onSessionChange, }: PreviewSessionTabsProps) { const { sessions, isLoaded } = useTaskSessions(taskId); @@ -83,17 +87,17 @@ export function PreviewSessionTabs({ ); if (!isLoaded && sortedSessions.length === 0) { - return ; + return ; } if (sortedSessions.length === 0) { - return ( -
-
- No agents yet. -
-
- ); + if (ensureSession?.status === "preparing") { + return ; + } + if (ensureSession?.status === "error") { + return ; + } + return ; } return ( @@ -178,10 +182,41 @@ function RunningSpinner() { return ; } -function PreviewLoadingState() { +function PreviewLoadingState({ label }: { label: string }) { return ( -
- Loading agents… +
+ + {label} +
+ ); +} + +function PreviewEmptyState() { + return ( +
+
+ No agents yet. +
+
+ ); +} + +function PreviewEnsureError({ onRetry }: { onRetry: () => void }) { + return ( +
+ Failed to prepare workspace. +
); } diff --git a/apps/web/components/task/task-page-content.tsx b/apps/web/components/task/task-page-content.tsx index a2cc01f930..9f382b1124 100644 --- a/apps/web/components/task/task-page-content.tsx +++ b/apps/web/components/task/task-page-content.tsx @@ -1,6 +1,6 @@ "use client"; -import { useEffect, useMemo, useRef, useState } from "react"; +import { useEffect, useMemo, useState } from "react"; import { TaskTopBar } from "@/components/task/task-top-bar"; import { TaskLayout } from "@/components/task/task-layout"; import { DebugOverlay } from "@/components/debug-overlay"; @@ -15,7 +15,7 @@ import { useSessionResumption } from "@/hooks/domains/session/use-session-resump import { useSessionAgentctl } from "@/hooks/domains/session/use-session-agentctl"; import { useTaskFocus } from "@/hooks/domains/session/use-task-focus"; import { useAppStore } from "@/components/state-provider"; -import { useTaskSessions } from "@/hooks/use-task-sessions"; +import { useEnsureTaskSession } from "@/hooks/domains/session/use-ensure-task-session"; import { fetchTask } from "@/lib/api"; import { useTasks } from "@/hooks/use-tasks"; import { useResponsiveBreakpoint } from "@/hooks/use-responsive-breakpoint"; @@ -30,12 +30,6 @@ import { resolveTaskProps, } from "@/components/task/task-page-content-helpers"; -// Stable empty array used as the fallback in useAppStore selectors that return -// arrays — avoids creating a new reference on every call, which would cause -// React's useSyncExternalStore (getServerSnapshot) to throw an infinite-loop -// error during SSR / hydration. -const EMPTY_SESSIONS: never[] = []; - type TaskPageContentProps = { task: Task | null; taskId?: string | null; @@ -492,40 +486,6 @@ function useTaskDetails(activeTaskId: string | null, initialTask: Task | null) { return { task, kanbanTask }; } -function useAutoStartSession( - task: Task | null, - handleStartAgent: ( - agentProfileId: string, - opts?: { prompt?: string; autoStart?: boolean }, - ) => Promise, -) { - const { isLoaded } = useTaskSessions(task?.id ?? null); - const sessions = useAppStore((state) => - task?.id ? (state.taskSessionsByTask.itemsByTaskId[task.id] ?? EMPTY_SESSIONS) : EMPTY_SESSIONS, - ); - const workspaceDefaultProfileId = useAppStore((state) => { - const activeId = state.workspaces.activeId; - if (!activeId) return null; - return state.workspaces.items.find((w) => w.id === activeId)?.default_agent_profile_id ?? null; - }); - const taskMetaProfileId = (task?.metadata as Record | null)?.agent_profile_id; - const agentProfileId = - typeof taskMetaProfileId === "string" ? taskMetaProfileId : (workspaceDefaultProfileId ?? null); - const startedRef = useRef(false); - - useEffect(() => { - if (!task?.id) return; - if (!isLoaded) return; - if (sessions.length > 0) return; - if (!agentProfileId) return; - if (startedRef.current) return; - startedRef.current = true; - handleStartAgent(agentProfileId, { autoStart: true }).catch(() => { - startedRef.current = false; - }); - }, [task?.id, isLoaded, sessions.length, agentProfileId, handleStartAgent]); -} - function useTaskPageData( initialTask: Task | null, fallbackTaskId: string | null | undefined, @@ -548,7 +508,7 @@ function useTaskPageData( const { task } = useTaskDetails(activeTaskId, initialTask); const agent = useSessionAgent(task); - useAutoStartSession(task, agent.handleStartAgent); + useEnsureTaskSession(task); const initialSessionId = sessionId ?? agent.taskSessionId ?? null; const effectiveSessionId = validatedActiveSessionId ?? initialSessionId; diff --git a/apps/web/e2e/tests/kanban/kanban-board.spec.ts b/apps/web/e2e/tests/kanban/kanban-board.spec.ts index 04a4743b44..8242ceabec 100644 --- a/apps/web/e2e/tests/kanban/kanban-board.spec.ts +++ b/apps/web/e2e/tests/kanban/kanban-board.spec.ts @@ -42,4 +42,43 @@ test.describe("Kanban board", () => { // Use toHaveURL (polling assertion) since replaceState doesn't fire navigation events. await expect(testPage).toHaveURL(/taskId=/, { timeout: 10000 }); }); + + // The desktop kanban header centers the search input absolutely. When the + // preview panel opens, the kanban area shrinks (`kanbanWidth = container - + // previewWidth`); the header narrows along with it. Below ~1100px there is + // no longer room between the left/right action groups for the centered + // search, so the header hides it (see useIsHeaderNarrow in kanban-header). + test("hides header search when preview panel narrows the kanban area", async ({ + testPage, + apiClient, + seedData, + }) => { + await apiClient.saveUserSettings({ enable_preview_on_click: true }); + + const task = await apiClient.createTask(seedData.workspaceId, "Header Squeeze Task", { + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + }); + const kanban = new KanbanPage(testPage); + await kanban.goto(); + + // With no preview open, the kanban area uses the full viewport width + // (Desktop Chrome = 1280px) so the centered search is visible. + const search = testPage.getByTestId("kanban-header-search"); + await expect(search).toBeVisible(); + + // Open the preview — kanban width drops below 1100px (1280 - 500 default + // preview width = 780px) and the search must hide. + const card = kanban.taskCard(task.id); + await expect(card).toBeVisible(); + await card.click(); + await expect(testPage.getByTestId("task-preview-panel")).toBeVisible({ timeout: 10_000 }); + await expect(search).toBeHidden({ timeout: 5_000 }); + + // Closing the preview restores the full kanban width and brings the + // search back. + await testPage.keyboard.press("Escape"); + await expect(testPage.getByTestId("task-preview-panel")).toBeHidden({ timeout: 5_000 }); + await expect(search).toBeVisible({ timeout: 5_000 }); + }); }); diff --git a/apps/web/e2e/tests/kanban/preview-session-tabs.spec.ts b/apps/web/e2e/tests/kanban/preview-session-tabs.spec.ts index d0b90084ad..958493f96e 100644 --- a/apps/web/e2e/tests/kanban/preview-session-tabs.spec.ts +++ b/apps/web/e2e/tests/kanban/preview-session-tabs.spec.ts @@ -145,3 +145,138 @@ test.describe("Preview session tabs", () => { await expect(previewPanel.getByRole("button", { name: "+" })).toHaveCount(0); }); }); + +/** + * Verifies the lazy-workspace-setup behavior: opening the kanban preview for + * a task with no sessions auto-launches one (using the workspace default agent + * profile) so the user lands on a usable agent tab instead of the + * "No agents yet." dead-end. + */ +test.describe("Preview auto-prepare", () => { + test("auto-starts a session when previewing a task with no sessions", async ({ + testPage, + apiClient, + seedData, + }) => { + test.setTimeout(120_000); + + // 1. Make sure the workspace has a default agent profile so the preview + // can resolve one to start. The seed creates an agent profile but + // doesn't necessarily wire it as the workspace default. + await apiClient.updateWorkspace(seedData.workspaceId, { + default_agent_profile_id: seedData.agentProfileId, + }); + + // 2. Create a task with NO agent — it lands on the kanban with 0 sessions. + const task = await apiClient.createTask(seedData.workspaceId, "Auto Prepare Task", { + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }); + + // Sanity-check the precondition: the freshly created task must have no + // sessions. Otherwise the "auto-prepare" path is never exercised. + const before = await apiClient.listTaskSessions(task.id); + expect(before.sessions ?? []).toHaveLength(0); + + // 3. Enable preview-on-click and open the kanban. + await apiClient.saveUserSettings({ enable_preview_on_click: true }); + const kanban = new KanbanPage(testPage); + await kanban.goto(); + + const card = kanban.taskCard(task.id); + await expect(card).toBeVisible({ timeout: 10_000 }); + await card.click(); + + // 4. Preview panel renders. The empty "No agents yet." state must NOT + // appear at any point — the user should see "Preparing workspace…" + // bridging the gap and then the session tab. + const previewPanel = testPage.getByTestId("task-preview-panel"); + await expect(previewPanel).toBeVisible({ timeout: 10_000 }); + await expect(previewPanel.getByTestId("preview-empty-state")).toHaveCount(0); + + // 5. Eventually a session tab appears for the auto-started session. + const sessionTab = previewPanel.locator('[data-testid^="preview-session-tab-"]'); + await expect(sessionTab.first()).toBeVisible({ timeout: 30_000 }); + + // 6. The auto-launched session is reflected in the backend. + await expect + .poll( + async () => { + const { sessions } = await apiClient.listTaskSessions(task.id); + return sessions.length; + }, + { timeout: 30_000, message: "Waiting for auto-prepared session to be created" }, + ) + .toBeGreaterThan(0); + }); + + // Regression test for the snapshot/PR-review case: tasks that don't carry + // their own metadata.agent_profile_id used to dead-end on "No agents yet." + // The resolver now also walks the workflow step → workflow chain, so a step + // with its own agent_profile_id is enough to auto-start even when the task + // and workspace have nothing set. + test("auto-starts using the workflow step's agent_profile_id when task has none", async ({ + testPage, + apiClient, + seedData, + }) => { + test.setTimeout(120_000); + + // 1. Create a second agent profile distinct from the seeded one so we can + // prove the resolver picked the step's profile (not the workspace + // default that the previous test in this file may have left set). + const { agents } = await apiClient.listAgents(); + const stepProfile = await apiClient.createAgentProfile(agents[0].id, "Step Profile", { + model: "mock-fast", + }); + + // 2. Pin that profile on the start step. The workspace default is left + // alone — whether or not it is set, the step value must win. + await apiClient.updateWorkflowStep(seedData.startStepId, { + agent_profile_id: stepProfile.id, + }); + + // 3. Task with NO agent and NO metadata override — the only place a + // profile can come from is the step. + const task = await apiClient.createTask(seedData.workspaceId, "Step Profile Task", { + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }); + + const before = await apiClient.listTaskSessions(task.id); + expect(before.sessions ?? []).toHaveLength(0); + + await apiClient.saveUserSettings({ enable_preview_on_click: true }); + const kanban = new KanbanPage(testPage); + await kanban.goto(); + + const card = kanban.taskCard(task.id); + await expect(card).toBeVisible({ timeout: 10_000 }); + await card.click(); + + // 4. Preview panel opens and skips the empty state. + const previewPanel = testPage.getByTestId("task-preview-panel"); + await expect(previewPanel).toBeVisible({ timeout: 10_000 }); + await expect(previewPanel.getByTestId("preview-empty-state")).toHaveCount(0); + + // 5. A session tab appears for the auto-started session. + const sessionTab = previewPanel.locator('[data-testid^="preview-session-tab-"]'); + await expect(sessionTab.first()).toBeVisible({ timeout: 30_000 }); + + // 6. The auto-launched session uses the STEP's profile, not the workspace + // default — this is the regression-bait assertion that proves the + // backend session.ensure resolution chain (task metadata → step → workflow + // → workspace default) honors the step override. + await expect + .poll( + async () => { + const { sessions } = await apiClient.listTaskSessions(task.id); + return sessions[0]?.agent_profile_id ?? null; + }, + { timeout: 30_000, message: "Waiting for session created with step's profile" }, + ) + .toBe(stepProfile.id); + }); +}); diff --git a/apps/web/hooks/domains/session/use-ensure-task-session.test.ts b/apps/web/hooks/domains/session/use-ensure-task-session.test.ts new file mode 100644 index 0000000000..87f223e4db --- /dev/null +++ b/apps/web/hooks/domains/session/use-ensure-task-session.test.ts @@ -0,0 +1,131 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { act, renderHook } from "@testing-library/react"; + +const mockEnsureTaskSession = vi.fn(); +const mockLoadSessions = vi.fn().mockResolvedValue(undefined); +let mockSessionsResult: { + sessions: Array<{ id: string }>; + isLoaded: boolean; + loadSessions: (force?: boolean) => Promise; +} = { + sessions: [], + isLoaded: true, + loadSessions: mockLoadSessions, +}; + +vi.mock("@/lib/services/session-launch-service", () => ({ + ensureTaskSession: (taskId: string) => mockEnsureTaskSession(taskId), +})); + +vi.mock("@/hooks/use-task-sessions", () => ({ + useTaskSessions: () => mockSessionsResult, +})); + +import { useEnsureTaskSession } from "./use-ensure-task-session"; + +const TASK = { id: "task-1" }; + +function flushMicrotasks() { + return act(() => Promise.resolve()); +} + +describe("useEnsureTaskSession", () => { + beforeEach(() => { + vi.clearAllMocks(); + mockLoadSessions.mockResolvedValue(undefined); + mockSessionsResult = { sessions: [], isLoaded: true, loadSessions: mockLoadSessions }; + mockEnsureTaskSession.mockResolvedValue({ + success: true, + task_id: "task-1", + session_id: "sess-new", + state: "CREATED", + source: "created_prepare", + newly_created: true, + }); + }); + + it("calls the backend ensure endpoint once when the task has no sessions", async () => { + const { result } = renderHook(() => useEnsureTaskSession(TASK)); + + expect(mockEnsureTaskSession).toHaveBeenCalledTimes(1); + expect(mockEnsureTaskSession).toHaveBeenCalledWith("task-1"); + expect(result.current.status).toBe("preparing"); + await flushMicrotasks(); + expect(result.current.status).toBe("idle"); + }); + + it("force-reloads the session list after a successful ensure", async () => { + renderHook(() => useEnsureTaskSession(TASK)); + await flushMicrotasks(); + // Two awaits: ensure().then(loadSessions(true)).then(setStatus). + await flushMicrotasks(); + expect(mockLoadSessions).toHaveBeenCalledWith(true); + }); + + it("no-ops when the task already has a session", () => { + mockSessionsResult = { + sessions: [{ id: "sess-1" }], + isLoaded: true, + loadSessions: mockLoadSessions, + }; + renderHook(() => useEnsureTaskSession(TASK)); + expect(mockEnsureTaskSession).not.toHaveBeenCalled(); + }); + + it("no-ops while sessions are still loading", () => { + mockSessionsResult = { sessions: [], isLoaded: false, loadSessions: mockLoadSessions }; + renderHook(() => useEnsureTaskSession(TASK)); + expect(mockEnsureTaskSession).not.toHaveBeenCalled(); + }); + + it("no-ops when disabled", () => { + renderHook(() => useEnsureTaskSession(TASK, { enabled: false })); + expect(mockEnsureTaskSession).not.toHaveBeenCalled(); + }); + + it("no-ops when task id is missing", () => { + renderHook(() => useEnsureTaskSession(null)); + expect(mockEnsureTaskSession).not.toHaveBeenCalled(); + }); + + it("is idempotent across re-renders for the same task", () => { + const { rerender } = renderHook(() => useEnsureTaskSession(TASK)); + rerender(); + rerender(); + expect(mockEnsureTaskSession).toHaveBeenCalledTimes(1); + }); + + it("reports an error and exposes a working retry()", async () => { + mockEnsureTaskSession.mockRejectedValueOnce(new Error("boom")); + const { result } = renderHook(() => useEnsureTaskSession(TASK)); + + expect(mockEnsureTaskSession).toHaveBeenCalledTimes(1); + await flushMicrotasks(); + expect(result.current.status).toBe("error"); + expect(result.current.error?.message).toBe("boom"); + + mockEnsureTaskSession.mockResolvedValueOnce({ + success: true, + task_id: "task-1", + session_id: "sess-new", + state: "CREATED", + source: "created_prepare", + newly_created: true, + }); + act(() => result.current.retry()); + expect(mockEnsureTaskSession).toHaveBeenCalledTimes(2); + await flushMicrotasks(); + expect(result.current.status).toBe("idle"); + }); + + it("calls ensure again when the task id changes", () => { + const { rerender } = renderHook( + ({ task }: { task: { id: string } }) => useEnsureTaskSession(task), + { initialProps: { task: TASK } }, + ); + expect(mockEnsureTaskSession).toHaveBeenCalledTimes(1); + rerender({ task: { id: "task-2" } }); + expect(mockEnsureTaskSession).toHaveBeenCalledTimes(2); + expect(mockEnsureTaskSession).toHaveBeenLastCalledWith("task-2"); + }); +}); diff --git a/apps/web/hooks/domains/session/use-ensure-task-session.ts b/apps/web/hooks/domains/session/use-ensure-task-session.ts new file mode 100644 index 0000000000..ce18366b5f --- /dev/null +++ b/apps/web/hooks/domains/session/use-ensure-task-session.ts @@ -0,0 +1,87 @@ +import { useCallback, useEffect, useRef, useState } from "react"; +import { ensureTaskSession } from "@/lib/services/session-launch-service"; +import { useTaskSessions } from "@/hooks/use-task-sessions"; + +/** Minimal task shape consumed by useEnsureTaskSession. */ +export type EnsureTaskInput = { + id?: string | null; +} | null; + +export type EnsureTaskSessionStatus = "idle" | "preparing" | "error"; + +export type UseEnsureTaskSessionResult = { + status: EnsureTaskSessionStatus; + error: Error | null; + retry: () => void; +}; + +/** + * Ensures the task has at least one session by delegating to the backend's + * idempotent `session.ensure` endpoint. The backend resolves the agent profile + * (task metadata → workflow step → workflow → workspace default) and chooses + * prepare vs start based on the workflow step's auto_start_agent action — the + * frontend stays thin and contract-free. + * + * Behavior: + * - No-op while sessions are still loading. + * - No-op when the task already has at least one session. + * - No-op when `enabled === false` or `task.id` is missing. + * - Idempotent per task id within a mount; switching tasks resets the latch. + */ +export function useEnsureTaskSession( + task: EnsureTaskInput, + opts?: { enabled?: boolean }, +): UseEnsureTaskSessionResult { + const enabled = opts?.enabled ?? true; + const taskId = task?.id ?? null; + const { sessions, isLoaded, loadSessions } = useTaskSessions(taskId); + + const [status, setStatus] = useState("idle"); + const [error, setError] = useState(null); + const [retryToken, setRetryToken] = useState(0); + + // Latch keyed by `${taskId}:${retryToken}` so a re-mount on the same task + // doesn't refire, but switching tasks or calling retry() does. + const launchedKeyRef = useRef(null); + + /* eslint-disable react-hooks/set-state-in-effect -- ensuring a session is a side effect; status mirrors that external work */ + useEffect(() => { + if (!enabled || !taskId || !isLoaded) return; + if (sessions.length > 0) return; + const key = `${taskId}:${retryToken}`; + if (launchedKeyRef.current === key) return; + launchedKeyRef.current = key; + + // Cancel guard so a stale resolution can't overwrite a switched-away task. + let cancelled = false; + setStatus("preparing"); + setError(null); + ensureTaskSession(taskId) + .then(async () => { + if (cancelled || launchedKeyRef.current !== key) return; + // Force-reload: backend may have returned an existing_* source our initial list missed. + await loadSessions(true); + if (cancelled || launchedKeyRef.current !== key) return; + setStatus("idle"); + }) + .catch((err: unknown) => { + if (cancelled || launchedKeyRef.current !== key) return; + setStatus("error"); + setError(err instanceof Error ? err : new Error(String(err))); + launchedKeyRef.current = null; + }); + + return () => { + cancelled = true; + }; + }, [enabled, taskId, isLoaded, loadSessions, sessions.length, retryToken]); + /* eslint-enable react-hooks/set-state-in-effect */ + + const retry = useCallback(() => { + setStatus("idle"); + setError(null); + setRetryToken((n) => n + 1); + }, []); + + return { status, error, retry }; +} diff --git a/apps/web/hooks/use-task-sessions.ts b/apps/web/hooks/use-task-sessions.ts index c7883b0b58..0969c98b2d 100644 --- a/apps/web/hooks/use-task-sessions.ts +++ b/apps/web/hooks/use-task-sessions.ts @@ -25,15 +25,6 @@ export function useTaskSessions(taskId: string | null) { setTaskSessionsLoading(taskId, true); try { const response = await listTaskSessions(taskId, { cache: "no-store" }); - // Debug: log the fetched sessions to see if they have agent_profile_snapshot - console.log( - "[useTaskSessions] fetched sessions:", - response.sessions?.map((s) => ({ - id: s.id, - hasSnapshot: !!s.agent_profile_snapshot, - cli_passthrough: s.agent_profile_snapshot?.cli_passthrough, - })), - ); setTaskSessionsForTask(taskId, response.sessions ?? []); } catch { setTaskSessionsForTask(taskId, []); diff --git a/apps/web/hooks/use-task.ts b/apps/web/hooks/use-task.ts index 71cfc939aa..90df4fdcc8 100644 --- a/apps/web/hooks/use-task.ts +++ b/apps/web/hooks/use-task.ts @@ -1,15 +1,23 @@ import { useEffect } from "react"; import { useAppStore } from "@/components/state-provider"; import { getWebSocketClient } from "@/lib/ws/connection"; +import { findTaskInSnapshots } from "@/lib/kanban/find-task"; import type { KanbanState } from "@/lib/state/slices"; +type Task = KanbanState["tasks"][number]; + export function useTask(taskId: string | null) { - const task = useAppStore((state) => - taskId - ? (state.kanban.tasks.find((item: KanbanState["tasks"][number]) => item.id === taskId) ?? - null) - : null, - ); + // The active workflow's tasks live in `kanban.tasks`, but cross-workflow + // tasks (PR-review boards, multi-workflow swimlanes) live in + // `kanbanMulti.snapshots[*].tasks`. Mirror the lookup used by + // `KanbanWithPreview.useSelectedTask` so consumers like the chat panel + // can still resolve the task description for cross-workflow previews. + const task = useAppStore((state) => { + if (!taskId) return null; + const fromActive = state.kanban.tasks.find((item: Task) => item.id === taskId); + if (fromActive) return fromActive; + return findTaskInSnapshots(taskId, state.kanbanMulti.snapshots); + }); useEffect(() => { if (!taskId) return; diff --git a/apps/web/lib/kanban/find-task.ts b/apps/web/lib/kanban/find-task.ts new file mode 100644 index 0000000000..3931b39105 --- /dev/null +++ b/apps/web/lib/kanban/find-task.ts @@ -0,0 +1,17 @@ +import type { KanbanState } from "@/lib/state/slices"; + +type Task = KanbanState["tasks"][number]; + +// findTaskInSnapshots searches the multi-workflow snapshot map for a task by id. +// Used by callers that need to resolve cross-workflow tasks (e.g. PR-review +// boards, multi-workflow swimlanes) which do not live in `kanban.tasks`. +export function findTaskInSnapshots( + taskId: string, + snapshots: Record, +): Task | null { + for (const snapshot of Object.values(snapshots)) { + const found = snapshot.tasks.find((t) => t.id === taskId); + if (found) return found; + } + return null; +} diff --git a/apps/web/lib/services/session-launch-service.ts b/apps/web/lib/services/session-launch-service.ts index c7c410afbf..d12e19777d 100644 --- a/apps/web/lib/services/session-launch-service.ts +++ b/apps/web/lib/services/session-launch-service.ts @@ -51,3 +51,33 @@ export async function launchSession( const effectiveTimeout = timeout ?? (request.intent === "resume" ? 30_000 : 15_000); return client.request("session.launch", request, effectiveTimeout); } + +export type EnsureSessionResponse = { + success: boolean; + task_id: string; + session_id?: string; + state: string; + agent_profile_id?: string; + source: "existing_primary" | "existing_newest" | "created_prepare" | "created_start"; + newly_created: boolean; +}; + +/** + * Server-authoritative idempotent ensure: returns the task's existing primary + * (or newest) session, otherwise creates one with the agent profile resolved + * server-side from task metadata, workflow step, workflow, or workspace + * default. Safe for preview mode — the backend chooses prepare vs start based + * on the workflow step's auto_start_agent action. + */ +export async function ensureTaskSession( + taskId: string, + timeout?: number, +): Promise { + const client = getWebSocketClient(); + if (!client) throw new Error("WebSocket client not available"); + return client.request( + "session.ensure", + { task_id: taskId }, + timeout ?? 15_000, + ); +} diff --git a/apps/web/lib/state/default-state.ts b/apps/web/lib/state/default-state.ts index 41ad95d509..327a33681d 100644 --- a/apps/web/lib/state/default-state.ts +++ b/apps/web/lib/state/default-state.ts @@ -78,6 +78,7 @@ export const defaultState = { bottomTerminal: defaultUIState.bottomTerminal, sidebarViews: defaultUIState.sidebarViews, collapsedSubtaskParents: defaultUIState.collapsedSubtaskParents, + kanbanPreviewedTaskId: defaultUIState.kanbanPreviewedTaskId, }; export type DefaultState = typeof defaultState; diff --git a/apps/web/lib/state/slices/ui/types.ts b/apps/web/lib/state/slices/ui/types.ts index 865cc98079..bceb5cf9a2 100644 --- a/apps/web/lib/state/slices/ui/types.ts +++ b/apps/web/lib/state/slices/ui/types.ts @@ -117,6 +117,8 @@ export type UISliceState = { sidebarViews: SidebarSliceState; /** Parent task IDs whose subtasks are collapsed in the sidebar. Tab-scoped (sessionStorage). */ collapsedSubtaskParents: string[]; + /** Task ID currently shown in the kanban preview side-panel, or null if closed. */ + kanbanPreviewedTaskId: string | null; }; export type UISliceActions = { @@ -167,6 +169,7 @@ export type UISliceActions = { toggleSubtaskCollapsed: (parentTaskId: string) => void; clearSidebarSyncError: () => void; migrateLocalViewsToBackend: () => void; + setKanbanPreviewedTaskId: (taskId: string | null) => void; }; export type { SidebarView, SidebarViewDraft }; diff --git a/apps/web/lib/state/slices/ui/ui-slice.ts b/apps/web/lib/state/slices/ui/ui-slice.ts index 4ec14b96ce..daa5a58419 100644 --- a/apps/web/lib/state/slices/ui/ui-slice.ts +++ b/apps/web/lib/state/slices/ui/ui-slice.ts @@ -94,6 +94,7 @@ export const defaultUIState: UISliceState = { bottomTerminal: { isOpen: false, pendingCommand: null }, sidebarViews: loadSidebarState(), collapsedSubtaskParents: [], + kanbanPreviewedTaskId: null, }; type ImmerSet = Parameters[0]; @@ -514,6 +515,11 @@ export const createUISlice: StateCreator + set((draft) => { + if (draft.kanbanPreviewedTaskId === taskId) return; + draft.kanbanPreviewedTaskId = taskId; + }), openQuickChat: (sessionId, workspaceId, agentProfileId) => set((draft) => { draft.quickChat.isOpen = true; diff --git a/apps/web/lib/state/store.ts b/apps/web/lib/state/store.ts index 393874c5bd..ac6b33db45 100644 --- a/apps/web/lib/state/store.ts +++ b/apps/web/lib/state/store.ts @@ -229,6 +229,7 @@ export type AppState = { bottomTerminal: (typeof defaultUIState)["bottomTerminal"]; sidebarViews: (typeof defaultUIState)["sidebarViews"]; collapsedSubtaskParents: (typeof defaultUIState)["collapsedSubtaskParents"]; + kanbanPreviewedTaskId: (typeof defaultUIState)["kanbanPreviewedTaskId"]; // GitHub actions setGitHubStatus: (status: GitHubStatus | null) => void; @@ -478,6 +479,7 @@ export type AppState = { toggleSubtaskCollapsed: UIA["toggleSubtaskCollapsed"]; clearSidebarSyncError: UIA["clearSidebarSyncError"]; migrateLocalViewsToBackend: UIA["migrateLocalViewsToBackend"]; + setKanbanPreviewedTaskId: UIA["setKanbanPreviewedTaskId"]; }; export type AppStore = ReturnType;