diff --git a/apps/backend/cmd/kandev/adapters.go b/apps/backend/cmd/kandev/adapters.go index 6882bcb572..8c8aa91517 100644 --- a/apps/backend/cmd/kandev/adapters.go +++ b/apps/backend/cmd/kandev/adapters.go @@ -115,6 +115,11 @@ func (a *lifecycleAdapter) LaunchAgent(ctx context.Context, req *executor.Launch }, nil } +// SetExecutionDescription updates the task description in an existing execution's metadata. +func (a *lifecycleAdapter) SetExecutionDescription(ctx context.Context, agentExecutionID string, description string) error { + return a.mgr.SetExecutionDescription(ctx, agentExecutionID, description) +} + // StartAgentProcess starts the agent subprocess for an instance. // The command is built internally based on the instance's agent profile. func (a *lifecycleAdapter) StartAgentProcess(ctx context.Context, agentInstanceID string) error { @@ -232,6 +237,12 @@ func (w *orchestratorWrapper) ResumeTaskSession(ctx context.Context, taskID, tas return err } +// StartCreatedSession forwards to the orchestrator service, discarding the TaskExecution result. +func (w *orchestratorWrapper) StartCreatedSession(ctx context.Context, taskID, sessionID, agentProfileID, prompt string) error { + _, err := w.svc.StartCreatedSession(ctx, taskID, sessionID, agentProfileID, prompt) + return err +} + // messageCreatorAdapter adapts the task service to the orchestrator.MessageCreator interface type messageCreatorAdapter struct { svc *taskservice.Service diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 5cf058763f..5b7e9b15f4 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -867,6 +867,20 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti return execution, nil } +// SetExecutionDescription updates the task description stored in an execution's metadata. +// This is used when starting an agent on a workspace that was launched without a prompt. +func (m *Manager) SetExecutionDescription(_ context.Context, executionID string, description string) error { + execution, exists := m.executionStore.Get(executionID) + if !exists { + return fmt.Errorf("execution %q not found", executionID) + } + if execution.Metadata == nil { + execution.Metadata = make(map[string]interface{}) + } + execution.Metadata["task_description"] = description + return nil +} + // StartAgentProcess configures and starts the agent subprocess for an execution. // This must be called after Launch() to actually start the agent (e.g., auggie, codex). // The command is built internally based on the execution's agent profile. diff --git a/apps/backend/internal/integration/orchestrator_test.go b/apps/backend/internal/integration/orchestrator_test.go index 675d438355..efb953d3df 100644 --- a/apps/backend/internal/integration/orchestrator_test.go +++ b/apps/backend/internal/integration/orchestrator_test.go @@ -151,6 +151,11 @@ func (s *SimulatedAgentManagerClient) LaunchAgent(ctx context.Context, req *exec }, nil } +// SetExecutionDescription updates the task description for an execution +func (s *SimulatedAgentManagerClient) SetExecutionDescription(ctx context.Context, agentExecutionID string, description string) error { + return nil +} + // StartAgentProcess simulates starting the agent subprocess func (s *SimulatedAgentManagerClient) StartAgentProcess(ctx context.Context, agentExecutionID string) error { s.logger.Info("simulated: starting agent process", diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index 67eec03d38..3b4b69c2b0 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -62,6 +62,10 @@ type AgentManagerClient interface { // ResolveAgentProfile resolves an agent profile ID to profile information ResolveAgentProfile(ctx context.Context, profileID string) (*AgentProfileInfo, error) + + // SetExecutionDescription updates the task description in an existing execution's metadata. + // Used when starting an agent on a session whose workspace was already launched. + SetExecutionDescription(ctx context.Context, agentExecutionID string, description string) error } // AgentProfileInfo contains resolved profile information @@ -150,6 +154,11 @@ func agentSessionStateToV1(state models.TaskSessionState) v1.TaskSessionState { return v1.TaskSessionState(state) } +// TaskStateChangeFunc is called when the executor needs to update a task's state. +// When set, it replaces direct repo.UpdateTaskState calls so the caller can +// publish events (e.g. WebSocket notifications) alongside the DB update. +type TaskStateChangeFunc func(ctx context.Context, taskID string, state v1.TaskState) error + // Executor manages agent execution for tasks type Executor struct { agentManager AgentManagerClient @@ -161,6 +170,10 @@ type Executor struct { retryLimit int retryDelay time.Duration + // Callback for task state changes that need event publishing. + // Set by the orchestrator to route through the task service layer. + onTaskStateChange TaskStateChangeFunc + // Per-session locks to prevent concurrent resume/launch operations on the same session. // This prevents race conditions when the backend restarts and multiple resume requests // arrive simultaneously (e.g., from frontend auto-resume). @@ -188,6 +201,57 @@ func NewExecutor(agentManager AgentManagerClient, repo repository.Repository, lo } } +// SetOnTaskStateChange sets a callback for task state changes. +// This allows the orchestrator to route state changes through the task service layer +// which publishes WebSocket events. Without this, async goroutines would only update +// the database, leaving the frontend out of sync. +func (e *Executor) SetOnTaskStateChange(fn TaskStateChangeFunc) { + e.onTaskStateChange = fn +} + +// startAgentProcessAsync starts the agent subprocess in a background goroutine. +// On success it transitions the task to IN_PROGRESS; on failure it marks both the +// session and task as FAILED. +func (e *Executor) startAgentProcessAsync(taskID, sessionID, agentExecutionID string) { + go func() { + startCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + + if err := e.agentManager.StartAgentProcess(startCtx, agentExecutionID); err != nil { + e.logger.Error("failed to start agent process", + zap.String("task_id", taskID), + zap.String("agent_execution_id", agentExecutionID), + zap.Error(err)) + if updateErr := e.repo.UpdateTaskSessionState(context.Background(), sessionID, models.TaskSessionStateFailed, err.Error()); updateErr != nil { + e.logger.Warn("failed to mark session as failed after start error", + zap.String("session_id", sessionID), + zap.Error(updateErr)) + } + if updateErr := e.updateTaskState(context.Background(), taskID, v1.TaskStateFailed); updateErr != nil { + e.logger.Warn("failed to mark task as failed after start error", + zap.String("task_id", taskID), + zap.Error(updateErr)) + } + return + } + + if updateErr := e.updateTaskState(context.Background(), taskID, v1.TaskStateInProgress); updateErr != nil { + e.logger.Warn("failed to update task state to IN_PROGRESS after agent start", + zap.String("task_id", taskID), + zap.Error(updateErr)) + } + }() +} + +// updateTaskState updates a task's state, using the callback if set for event publishing, +// or falling back to the raw repository. +func (e *Executor) updateTaskState(ctx context.Context, taskID string, state v1.TaskState) error { + if e.onTaskStateChange != nil { + return e.onTaskStateChange(ctx, taskID, state) + } + return e.repo.UpdateTaskState(ctx, taskID, state) +} + // shouldUseWorktree returns true if the given executor type should use Git worktrees. func shouldUseWorktree(executorType string) bool { return models.ExecutorType(executorType) == models.ExecutorTypeWorktree @@ -260,7 +324,7 @@ func (e *Executor) ExecuteWithProfile(ctx context.Context, task *v1.Task, agentP } // Launch the agent for the prepared session - return e.LaunchPreparedSession(ctx, task, sessionID, agentProfileID, executorID, prompt, workflowStepID) + return e.LaunchPreparedSession(ctx, task, sessionID, agentProfileID, executorID, prompt, workflowStepID, true) } // PrepareSession creates a session entry in the database without launching the agent. @@ -357,9 +421,13 @@ func (e *Executor) PrepareSession(ctx context.Context, task *v1.Task, agentProfi return sessionID, nil } -// LaunchPreparedSession launches the agent for a pre-created session. +// LaunchPreparedSession launches the workspace (and optionally the agent) for a pre-created session. // The session must have been created using PrepareSession. -func (e *Executor) LaunchPreparedSession(ctx context.Context, task *v1.Task, sessionID string, agentProfileID string, executorID string, prompt string, workflowStepID string) (*TaskExecution, error) { +// When startAgent is false, only the workspace infrastructure (agentctl) is launched; the agent +// subprocess is not started and the session state remains CREATED. +// When startAgent is true and the workspace was already launched (AgentExecutionID set), only the +// agent subprocess is started. +func (e *Executor) LaunchPreparedSession(ctx context.Context, task *v1.Task, sessionID string, agentProfileID string, executorID string, prompt string, workflowStepID string, startAgent bool) (*TaskExecution, error) { // Fetch the session to get its configuration session, err := e.repo.GetTaskSession(ctx, sessionID) if err != nil { @@ -373,6 +441,12 @@ func (e *Executor) LaunchPreparedSession(ctx context.Context, task *v1.Task, ses return nil, fmt.Errorf("session does not belong to task") } + // Fast path: workspace already launched (e.g., from PrepareSession with workspace). + // Only start the agent subprocess if requested; otherwise return early. + if session.AgentExecutionID != "" { + return e.startAgentOnExistingWorkspace(ctx, task, session, prompt, startAgent) + } + metadata := cloneMetadata(task.Metadata) var repositoryPath string var repositoryID string @@ -469,7 +543,7 @@ func (e *Executor) LaunchPreparedSession(ctx context.Context, task *v1.Task, ses zap.String("session_id", sessionID), zap.Error(updateErr)) } - if updateErr := e.repo.UpdateTaskState(ctx, task.ID, v1.TaskStateFailed); updateErr != nil { + if updateErr := e.updateTaskState(ctx, task.ID, v1.TaskStateFailed); updateErr != nil { e.logger.Warn("failed to mark task as failed after launch error", zap.String("task_id", task.ID), zap.Error(updateErr)) @@ -480,7 +554,9 @@ func (e *Executor) LaunchPreparedSession(ctx context.Context, task *v1.Task, ses now := time.Now().UTC() session.AgentExecutionID = resp.AgentExecutionID session.ContainerID = resp.ContainerID - session.State = models.TaskSessionStateStarting + if startAgent { + session.State = models.TaskSessionStateStarting + } session.ErrorMessage = "" session.UpdatedAt = now @@ -535,47 +611,25 @@ func (e *Executor) LaunchPreparedSession(ctx context.Context, task *v1.Task, ses } } + sessionState := v1.TaskSessionStateCreated + if startAgent { + sessionState = v1.TaskSessionStateStarting + } execution := &TaskExecution{ TaskID: task.ID, AgentExecutionID: resp.AgentExecutionID, AgentProfileID: agentProfileID, StartedAt: session.StartedAt, - SessionState: v1.TaskSessionStateStarting, + SessionState: sessionState, LastUpdate: now, SessionID: sessionID, WorktreePath: resp.WorktreePath, WorktreeBranch: resp.WorktreeBranch, } - // Start the agent process asynchronously - go func() { - startCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) - defer cancel() - - if err := e.agentManager.StartAgentProcess(startCtx, resp.AgentExecutionID); err != nil { - e.logger.Error("failed to start agent process", - zap.String("task_id", task.ID), - zap.String("agent_execution_id", resp.AgentExecutionID), - zap.Error(err)) - if updateErr := e.repo.UpdateTaskSessionState(context.Background(), sessionID, models.TaskSessionStateFailed, err.Error()); updateErr != nil { - e.logger.Warn("failed to mark session as failed after start error", - zap.String("session_id", sessionID), - zap.Error(updateErr)) - } - if updateErr := e.repo.UpdateTaskState(context.Background(), task.ID, v1.TaskStateFailed); updateErr != nil { - e.logger.Warn("failed to mark task as failed after start error", - zap.String("task_id", task.ID), - zap.Error(updateErr)) - } - return - } - - if updateErr := e.repo.UpdateTaskState(context.Background(), task.ID, v1.TaskStateInProgress); updateErr != nil { - e.logger.Warn("failed to update task state to IN_PROGRESS after agent start", - zap.String("task_id", task.ID), - zap.Error(updateErr)) - } - }() + if startAgent { + e.startAgentProcessAsync(task.ID, sessionID, resp.AgentExecutionID) + } e.logger.Info("agent launched for prepared session", zap.String("task_id", task.ID), @@ -585,6 +639,66 @@ func (e *Executor) LaunchPreparedSession(ctx context.Context, task *v1.Task, ses return execution, nil } +// startAgentOnExistingWorkspace handles the case where LaunchPreparedSession is called on a session +// whose workspace (agentctl) was already launched. It optionally starts just the agent subprocess. +func (e *Executor) startAgentOnExistingWorkspace(ctx context.Context, task *v1.Task, session *models.TaskSession, prompt string, startAgent bool) (*TaskExecution, error) { + if !startAgent { + // Workspace already launched, nothing else to do + now := time.Now().UTC() + return &TaskExecution{ + TaskID: task.ID, + AgentExecutionID: session.AgentExecutionID, + AgentProfileID: session.AgentProfileID, + StartedAt: session.StartedAt, + SessionState: v1.TaskSessionState(session.State), + LastUpdate: now, + SessionID: session.ID, + }, nil + } + + // Update the task description in the existing execution so StartAgentProcess picks it up + if prompt != "" { + if err := e.agentManager.SetExecutionDescription(ctx, session.AgentExecutionID, prompt); err != nil { + e.logger.Warn("failed to set execution description for existing workspace", + zap.String("session_id", session.ID), + zap.String("agent_execution_id", session.AgentExecutionID), + zap.Error(err)) + // Non-fatal: agent may start without description + } + } + + // Transition session to STARTING + now := time.Now().UTC() + session.State = models.TaskSessionStateStarting + session.ErrorMessage = "" + session.UpdatedAt = now + if err := e.repo.UpdateTaskSession(ctx, session); err != nil { + e.logger.Error("failed to update session state for agent start", + zap.String("session_id", session.ID), + zap.Error(err)) + } + + execution := &TaskExecution{ + TaskID: task.ID, + AgentExecutionID: session.AgentExecutionID, + AgentProfileID: session.AgentProfileID, + StartedAt: now, + SessionState: v1.TaskSessionStateStarting, + LastUpdate: now, + SessionID: session.ID, + } + + // Start the agent process asynchronously + e.startAgentProcessAsync(task.ID, session.ID, session.AgentExecutionID) + + e.logger.Info("agent starting on existing workspace", + zap.String("task_id", task.ID), + zap.String("session_id", session.ID), + zap.String("agent_execution_id", session.AgentExecutionID)) + + return execution, nil +} + // ResumeSession restarts an existing task session using its stored worktree. // When startAgent is false, only the executor runtime is started (agent process is not launched). func (e *Executor) ResumeSession(ctx context.Context, session *models.TaskSession, startAgent bool) (*TaskExecution, error) { @@ -858,7 +972,7 @@ func (e *Executor) ResumeSession(ctx context.Context, session *models.TaskSessio zap.String("session_id", session.ID), zap.Error(updateErr)) } - if updateErr := e.repo.UpdateTaskState(context.Background(), task.ID, v1.TaskStateFailed); updateErr != nil { + if updateErr := e.updateTaskState(context.Background(), task.ID, v1.TaskStateFailed); updateErr != nil { e.logger.Warn("failed to mark task as failed after start error on resume", zap.String("task_id", task.ID), zap.Error(updateErr)) @@ -870,7 +984,7 @@ func (e *Executor) ResumeSession(ctx context.Context, session *models.TaskSessio // If the session is waiting for input, the task should be in REVIEW state. // This ensures the task state reflects the actual agent status. if session.State == models.TaskSessionStateWaitingForInput { - if updateErr := e.repo.UpdateTaskState(context.Background(), task.ID, v1.TaskStateReview); updateErr != nil { + if updateErr := e.updateTaskState(context.Background(), task.ID, v1.TaskStateReview); updateErr != nil { e.logger.Warn("failed to update task state to REVIEW after resume", zap.String("task_id", task.ID), zap.Error(updateErr)) diff --git a/apps/backend/internal/orchestrator/executor/executor_test.go b/apps/backend/internal/orchestrator/executor/executor_test.go index db1652c1b0..8aa336529a 100644 --- a/apps/backend/internal/orchestrator/executor/executor_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_test.go @@ -12,9 +12,10 @@ import ( // mockAgentManager implements AgentManagerClient for testing type mockAgentManager struct { - launchAgentFunc func(ctx context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) - startAgentProcessFunc func(ctx context.Context, agentExecutionID string) error - resolveAgentProfileFunc func(ctx context.Context, profileID string) (*AgentProfileInfo, error) + launchAgentFunc func(ctx context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) + startAgentProcessFunc func(ctx context.Context, agentExecutionID string) error + resolveAgentProfileFunc func(ctx context.Context, profileID string) (*AgentProfileInfo, error) + setExecutionDescriptionFunc func(ctx context.Context, agentExecutionID string, description string) error } func (m *mockAgentManager) LaunchAgent(ctx context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) { @@ -28,6 +29,13 @@ func (m *mockAgentManager) LaunchAgent(ctx context.Context, req *LaunchAgentRequ }, nil } +func (m *mockAgentManager) SetExecutionDescription(ctx context.Context, agentExecutionID string, description string) error { + if m.setExecutionDescriptionFunc != nil { + return m.setExecutionDescriptionFunc(ctx, agentExecutionID, description) + } + return nil +} + func (m *mockAgentManager) StartAgentProcess(ctx context.Context, agentExecutionID string) error { if m.startAgentProcessFunc != nil { return m.startAgentProcessFunc(ctx, agentExecutionID) @@ -611,7 +619,7 @@ func TestLaunchPreparedSession_Success(t *testing.T) { Description: "Test description", } - execution, err := executor.LaunchPreparedSession(context.Background(), task, "session-123", "profile-123", "", "test prompt", "") + execution, err := executor.LaunchPreparedSession(context.Background(), task, "session-123", "profile-123", "", "test prompt", "", true) if err != nil { t.Fatalf("LaunchPreparedSession failed: %v", err) } @@ -651,12 +659,146 @@ func TestLaunchPreparedSession_SessionNotBelongsToTask(t *testing.T) { WorkspaceID: "workspace-123", } - _, err := executor.LaunchPreparedSession(context.Background(), task, "session-123", "profile-123", "", "test prompt", "") + _, err := executor.LaunchPreparedSession(context.Background(), task, "session-123", "profile-123", "", "test prompt", "", true) if err == nil { t.Error("Expected error when session doesn't belong to task") } } +func TestLaunchPreparedSession_WorkspaceOnly(t *testing.T) { + repo := newMockRepository() + + // Pre-create session (as PrepareSession would) + session := &models.TaskSession{ + ID: "session-123", + TaskID: "task-123", + AgentProfileID: "profile-123", + State: models.TaskSessionStateCreated, + StartedAt: time.Now(), + UpdatedAt: time.Now(), + } + repo.sessions[session.ID] = session + + launchCalled := false + startAgentCalled := false + agentManager := &mockAgentManager{ + launchAgentFunc: func(ctx context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) { + launchCalled = true + return &LaunchAgentResponse{ + AgentExecutionID: "exec-123", + ContainerID: "container-123", + Status: v1.AgentStatusStarting, + }, nil + }, + startAgentProcessFunc: func(ctx context.Context, id string) error { + startAgentCalled = true + return nil + }, + } + + executor := newTestExecutor(t, agentManager, repo) + + task := &v1.Task{ + ID: "task-123", + WorkspaceID: "workspace-123", + Title: "Test Task", + Description: "Test description", + } + + // startAgent=false: should launch workspace but NOT start agent + execution, err := executor.LaunchPreparedSession(context.Background(), task, "session-123", "profile-123", "", "", "", false) + if err != nil { + t.Fatalf("LaunchPreparedSession(startAgent=false) failed: %v", err) + } + + if !launchCalled { + t.Error("Expected LaunchAgent to be called (workspace setup)") + } + + // Give goroutines a moment to run (there shouldn't be any) + time.Sleep(50 * time.Millisecond) + + if startAgentCalled { + t.Error("Expected StartAgentProcess NOT to be called when startAgent=false") + } + + if execution.SessionState != v1.TaskSessionStateCreated { + t.Errorf("Expected session state CREATED, got %s", execution.SessionState) + } + + // Session in DB should remain CREATED + updatedSession := repo.sessions["session-123"] + if updatedSession.State != models.TaskSessionStateCreated { + t.Errorf("Expected DB session state CREATED, got %s", updatedSession.State) + } +} + +func TestLaunchPreparedSession_ExistingWorkspace_StartAgent(t *testing.T) { + repo := newMockRepository() + + // Session already has an AgentExecutionID (workspace previously launched) + session := &models.TaskSession{ + ID: "session-123", + TaskID: "task-123", + AgentProfileID: "profile-123", + AgentExecutionID: "exec-existing", + State: models.TaskSessionStateCreated, + StartedAt: time.Now(), + UpdatedAt: time.Now(), + } + repo.sessions[session.ID] = session + + startAgentCalled := false + descriptionSet := "" + agentManager := &mockAgentManager{ + startAgentProcessFunc: func(ctx context.Context, id string) error { + startAgentCalled = true + if id != "exec-existing" { + t.Errorf("Expected execution ID exec-existing, got %s", id) + } + return nil + }, + } + agentManager.setExecutionDescriptionFunc = func(ctx context.Context, id, desc string) error { + descriptionSet = desc + return nil + } + + executor := newTestExecutor(t, agentManager, repo) + + task := &v1.Task{ + ID: "task-123", + WorkspaceID: "workspace-123", + Title: "Test Task", + } + + execution, err := executor.LaunchPreparedSession(context.Background(), task, "session-123", "profile-123", "", "build the feature", "", true) + if err != nil { + t.Fatalf("LaunchPreparedSession(existing workspace) failed: %v", err) + } + + // Should use the existing execution ID + if execution.AgentExecutionID != "exec-existing" { + t.Errorf("Expected agent execution ID exec-existing, got %s", execution.AgentExecutionID) + } + + if execution.SessionState != v1.TaskSessionStateStarting { + t.Errorf("Expected session state STARTING, got %s", execution.SessionState) + } + + // Description should have been set + if descriptionSet != "build the feature" { + t.Errorf("Expected description 'build the feature', got %q", descriptionSet) + } + + // Wait for async goroutine + time.Sleep(100 * time.Millisecond) + + if !startAgentCalled { + t.Error("Expected StartAgentProcess to be called") + } +} + func TestExecuteWithProfile_UsesPrepareThenLaunch(t *testing.T) { repo := newMockRepository() diff --git a/apps/backend/internal/orchestrator/handlers/handlers.go b/apps/backend/internal/orchestrator/handlers/handlers.go index bdbeffce57..bb4ea44d65 100644 --- a/apps/backend/internal/orchestrator/handlers/handlers.go +++ b/apps/backend/internal/orchestrator/handlers/handlers.go @@ -116,8 +116,8 @@ func (h *Handlers) wsStartTask(ctx context.Context, msg *ws.Message) (*ws.Messag return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_profile_id is required when creating a new session", nil) } - // If session_id is provided, resume existing session with workflow step config - if req.SessionID != "" { + // If session_id is provided with a workflow step, resume with step config + if req.SessionID != "" && req.WorkflowStepID != "" { err := h.service.StartSessionForWorkflowStep(ctx, req.TaskID, req.SessionID, req.WorkflowStepID) if err != nil { h.logger.Error("failed to start task", zap.String("task_id", req.TaskID), zap.Error(err)) @@ -132,6 +132,29 @@ func (h *Handlers) wsStartTask(ctx context.Context, msg *ws.Message) (*ws.Messag return ws.NewResponse(msg.ID, msg.Action, resp) } + // If session_id is provided without a workflow step, start a CREATED session with prompt + if req.SessionID != "" { + execution, err := h.service.StartCreatedSession(ctx, req.TaskID, req.SessionID, req.AgentProfileID, req.Prompt) + if err != nil { + h.logger.Error("failed to start created session", zap.String("task_id", req.TaskID), zap.String("session_id", req.SessionID), zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to start session: "+err.Error(), nil) + } + resp := dto.StartTaskResponse{ + Success: true, + TaskID: execution.TaskID, + AgentExecutionID: execution.AgentExecutionID, + TaskSessionID: execution.SessionID, + State: string(execution.SessionState), + } + if execution.WorktreePath != "" { + resp.WorktreePath = &execution.WorktreePath + } + if execution.WorktreeBranch != "" { + resp.WorktreeBranch = &execution.WorktreeBranch + } + return ws.NewResponse(msg.ID, msg.Action, resp) + } + // Otherwise, create a new session execution, err := h.service.StartTask(ctx, req.TaskID, req.AgentProfileID, req.ExecutorID, req.Priority, req.Prompt, req.WorkflowStepID, false) if err != nil { diff --git a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go index 314267a209..2b1c8ebde4 100644 --- a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go +++ b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go @@ -43,6 +43,10 @@ func (m *mockAgentManager) LaunchAgent(ctx context.Context, req *executor.Launch }, nil } +func (m *mockAgentManager) SetExecutionDescription(ctx context.Context, agentExecutionID string, description string) error { + return nil +} + func (m *mockAgentManager) StartAgentProcess(ctx context.Context, agentExecutionID string) error { return nil } diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index a324fed322..75e344ec56 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -199,6 +199,12 @@ func NewService( } exec := executor.NewExecutor(agentManager, repo, log, execCfg) + // Wire executor task state changes through the task service layer so + // events are published (e.g. WebSocket notifications to the frontend). + exec.SetOnTaskStateChange(func(ctx context.Context, taskID string, state v1.TaskState) error { + return taskRepo.UpdateTaskState(ctx, taskID, state) + }) + // Create the scheduler with queue, executor, and task repository sched := scheduler.NewScheduler(taskQueue, exec, taskRepo, log, cfg.Scheduler) diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index b087941bdc..b5a9bfbabe 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -6,9 +6,12 @@ import ( "fmt" "os" "strings" + "time" "go.uber.org/zap" + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/orchestrator/dto" "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/orchestrator/queue" @@ -71,6 +74,16 @@ func (s *Service) PrepareTaskSession(ctx context.Context, taskID string, agentPr return "", err } + // Launch workspace infrastructure (agentctl) without starting the agent subprocess. + // This enables file browsing, editing, etc. while the session is in CREATED state. + if _, err := s.executor.LaunchPreparedSession(ctx, task, sessionID, agentProfileID, executorID, "", workflowStepID, false); err != nil { + s.logger.Warn("failed to launch workspace for prepared session (file browsing may be unavailable)", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(err)) + // Non-fatal: session is still usable, workspace will be launched when agent starts + } + s.logger.Info("task session prepared", zap.String("task_id", taskID), zap.String("session_id", sessionID)) @@ -111,7 +124,7 @@ func (s *Service) StartTaskWithSession(ctx context.Context, taskID string, sessi effectivePrompt, planModeActive := s.applyWorkflowAndPlanMode(ctx, effectivePrompt, task.ID, workflowStepID, planMode) - execution, err := s.executor.LaunchPreparedSession(ctx, task, sessionID, agentProfileID, executorID, effectivePrompt, workflowStepID) + execution, err := s.executor.LaunchPreparedSession(ctx, task, sessionID, agentProfileID, executorID, effectivePrompt, workflowStepID, true) if err != nil { return nil, err } @@ -123,6 +136,67 @@ func (s *Service) StartTaskWithSession(ctx context.Context, taskID string, sessi return execution, nil } +// StartCreatedSession starts agent execution for a task using a session that is in CREATED state. +// This is used when a session was prepared (via PrepareSession) but the agent was not launched, +// and the user now wants to start the agent with a prompt (e.g., from the plan panel or chat). +func (s *Service) StartCreatedSession(ctx context.Context, taskID, sessionID, agentProfileID, prompt string) (*executor.TaskExecution, error) { + s.logger.Debug("starting created session", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.String("agent_profile_id", agentProfileID)) + + // Load and verify session + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + return nil, fmt.Errorf("failed to get session: %w", err) + } + if session.TaskID != taskID { + return nil, fmt.Errorf("session does not belong to task") + } + if session.State != models.TaskSessionStateCreated { + return nil, fmt.Errorf("session is not in CREATED state (current: %s)", session.State) + } + + // Use agent profile from request, fall back to session's stored value + effectiveProfileID := agentProfileID + if effectiveProfileID == "" { + effectiveProfileID = session.AgentProfileID + } + if effectiveProfileID == "" { + return nil, fmt.Errorf("agent_profile_id is required") + } + + // Transition task state: CREATED → SCHEDULING → (IN_PROGRESS via executor) + if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateScheduling); err != nil { + s.logger.Warn("failed to update task state to SCHEDULING", + zap.String("task_id", taskID), + zap.Error(err)) + } + + task, err := s.scheduler.GetTask(ctx, taskID) + if err != nil { + return nil, err + } + + effectivePrompt := prompt + if effectivePrompt == "" { + effectivePrompt = task.Description + } + + executorID := session.ExecutorID + + execution, err := s.executor.LaunchPreparedSession(ctx, task, sessionID, effectiveProfileID, executorID, effectivePrompt, "", true) + if err != nil { + return nil, err + } + + if execution.SessionID != "" { + s.recordInitialMessage(ctx, taskID, execution.SessionID, effectivePrompt, false) + } + + return execution, nil +} + // StartTask manually starts agent execution for a task. // If workflowStepID is provided and workflowStepGetter is set, the prompt will be built // using the step's prompt_prefix + base prompt + prompt_suffix, and plan mode will be @@ -145,6 +219,36 @@ func (s *Service) StartTask(ctx context.Context, taskID string, agentProfileID s zap.Error(err)) } + // Move task to the target workflow step if provided and different from current + if workflowStepID != "" { + dbTask, err := s.repo.GetTask(ctx, taskID) + if err == nil && dbTask.WorkflowStepID != workflowStepID { + dbTask.WorkflowStepID = workflowStepID + dbTask.UpdatedAt = time.Now().UTC() + if err := s.repo.UpdateTask(ctx, dbTask); err != nil { + s.logger.Warn("failed to move task to workflow step", + zap.String("task_id", taskID), + zap.String("workflow_step_id", workflowStepID), + zap.Error(err)) + } else if s.eventBus != nil { + _ = s.eventBus.Publish(ctx, events.TaskUpdated, bus.NewEvent( + events.TaskUpdated, + "orchestrator", + map[string]interface{}{ + "task_id": dbTask.ID, + "board_id": dbTask.BoardID, + "workflow_step_id": dbTask.WorkflowStepID, + "title": dbTask.Title, + "description": dbTask.Description, + "state": string(dbTask.State), + "priority": dbTask.Priority, + "position": dbTask.Position, + }, + )) + } + } + } + // Fetch the task from the repository to get complete task info task, err := s.scheduler.GetTask(ctx, taskID) if err != nil { diff --git a/apps/backend/internal/task/handlers/message_handlers.go b/apps/backend/internal/task/handlers/message_handlers.go index 6d4fe318c1..2b7a161184 100644 --- a/apps/backend/internal/task/handlers/message_handlers.go +++ b/apps/backend/internal/task/handlers/message_handlers.go @@ -26,6 +26,7 @@ import ( type OrchestratorService interface { PromptTask(ctx context.Context, taskID, sessionID, prompt, model string, planMode bool, attachments []v1.MessageAttachment) (*orchestrator.PromptResult, error) ResumeTaskSession(ctx context.Context, taskID, taskSessionID string) error + StartCreatedSession(ctx context.Context, taskID, sessionID, agentProfileID, prompt string) error } // MessageHandlers handles WebSocket requests for messages @@ -162,6 +163,9 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w "Session has ended. Please create a new session to continue.", nil) } + // Handle CREATED sessions: save the message, then start the agent with it as the prompt + isCreatedSession := sessionResp.Session.State == models.TaskSessionStateCreated + // Get the current task state to determine if we need to transition task, err := h.taskController.GetTask(ctx, dto.GetTaskRequest{ID: req.TaskID}) if err != nil { @@ -217,8 +221,39 @@ func (h *MessageHandlers) wsAddMessage(ctx context.Context, msg *ws.Message) (*w model := req.Model planMode := req.PlanMode attachments := req.Attachments + startCreated := isCreatedSession + agentProfileID := sessionResp.Session.AgentProfileID go func() { promptCtx := context.WithoutCancel(ctx) + + // For CREATED sessions, start the agent with this message as the initial prompt + if startCreated { + if err := h.orchestrator.StartCreatedSession(promptCtx, taskID, sessionID, agentProfileID, content); err != nil { + h.logger.Warn("failed to start created session from message", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(err)) + + errorMsg := "Failed to start agent" + if _, createErr := h.messageController.CreateMessage(promptCtx, dto.CreateMessageRequest{ + TaskSessionID: sessionID, + TaskID: taskID, + Content: errorMsg, + AuthorType: "agent", + Type: string(v1.MessageTypeError), + Metadata: map[string]interface{}{ + "error": err.Error(), + }, + }); createErr != nil { + h.logger.Error("failed to create error message", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(createErr)) + } + } + return + } + _, err := h.orchestrator.PromptTask(promptCtx, taskID, sessionID, content, model, planMode, attachments) if err != nil { if errors.Is(err, executor.ErrExecutionNotFound) { diff --git a/apps/backend/internal/task/handlers/task_handlers.go b/apps/backend/internal/task/handlers/task_handlers.go index b3c6dc32a2..17d2c66ef0 100644 --- a/apps/backend/internal/task/handlers/task_handlers.go +++ b/apps/backend/internal/task/handlers/task_handlers.go @@ -228,6 +228,7 @@ type httpCreateTaskRequest struct { Position int `json:"position,omitempty"` Metadata map[string]interface{} `json:"metadata,omitempty"` StartAgent bool `json:"start_agent,omitempty"` + PrepareSession bool `json:"prepare_session,omitempty"` AgentProfileID string `json:"agent_profile_id,omitempty"` ExecutorID string `json:"executor_id,omitempty"` PlanMode bool `json:"plan_mode,omitempty"` @@ -249,7 +250,7 @@ func (h *TaskHandlers) httpCreateTask(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"error": "workspace_id is required"}) return } - if body.StartAgent && body.AgentProfileID == "" { + if (body.StartAgent || body.PrepareSession) && body.AgentProfileID == "" { c.JSON(http.StatusBadRequest, gin.H{"error": "agent_profile_id is required to start agent"}) return } @@ -289,7 +290,16 @@ func (h *TaskHandlers) httpCreateTask(c *gin.Context) { } response := createTaskResponse{TaskDTO: resp} - if body.StartAgent && body.AgentProfileID != "" && h.orchestrator != nil { + if body.PrepareSession && !body.StartAgent && body.AgentProfileID != "" && h.orchestrator != nil { + // Create session entry without launching the agent. + // The session stays in CREATED state until the user triggers it. + sessionID, err := h.orchestrator.PrepareTaskSession(reqCtx, resp.ID, body.AgentProfileID, body.ExecutorID, body.WorkflowStepID) + if err != nil { + h.logger.Error("failed to prepare session for task", zap.Error(err), zap.String("task_id", resp.ID)) + } else { + response.TaskSessionID = sessionID + } + } else if body.StartAgent && body.AgentProfileID != "" && h.orchestrator != nil { // Create session entry synchronously so we can return the session ID immediately sessionID, err := h.orchestrator.PrepareTaskSession(reqCtx, resp.ID, body.AgentProfileID, body.ExecutorID, body.WorkflowStepID) if err != nil { diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index 6ff722fc90..4e3d0196fe 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -487,12 +487,66 @@ func (s *Service) UpdateTask(ctx context.Context, id string, req *UpdateTaskRequ return nil, err } + // Resolve local paths to repository IDs (same logic as CreateTask) + var repoByPath map[string]*models.Repository + for _, repoInput := range req.Repositories { + if repoInput.RepositoryID == "" && repoInput.LocalPath != "" { + repos, err := s.repo.ListRepositories(ctx, task.WorkspaceID) + if err != nil { + s.logger.Error("failed to list repositories", zap.Error(err)) + return nil, err + } + repoByPath = make(map[string]*models.Repository, len(repos)) + for _, repo := range repos { + if repo.LocalPath == "" { + continue + } + repoByPath[repo.LocalPath] = repo + } + break + } + } + // Create new repositories for i, repoInput := range req.Repositories { + repositoryID := repoInput.RepositoryID + baseBranch := repoInput.BaseBranch + if repositoryID == "" && repoInput.LocalPath != "" { + repo := repoByPath[repoInput.LocalPath] + if repo == nil { + name := strings.TrimSpace(repoInput.Name) + if name == "" { + name = filepath.Base(repoInput.LocalPath) + } + defaultBranch := repoInput.DefaultBranch + if defaultBranch == "" { + defaultBranch = repoInput.BaseBranch + } + created, err := s.CreateRepository(ctx, &CreateRepositoryRequest{ + WorkspaceID: task.WorkspaceID, + Name: name, + SourceType: "local", + LocalPath: repoInput.LocalPath, + DefaultBranch: defaultBranch, + }) + if err != nil { + return nil, err + } + repo = created + if repoByPath == nil { + repoByPath = make(map[string]*models.Repository) + } + repoByPath[repoInput.LocalPath] = repo + } + repositoryID = repo.ID + if baseBranch == "" { + baseBranch = repo.DefaultBranch + } + } taskRepo := &models.TaskRepository{ TaskID: task.ID, - RepositoryID: repoInput.RepositoryID, - BaseBranch: repoInput.BaseBranch, + RepositoryID: repositoryID, + BaseBranch: baseBranch, Position: i, Metadata: make(map[string]interface{}), } @@ -1651,6 +1705,10 @@ func (s *Service) publishTaskEvent(ctx context.Context, eventType string, task * } } + if len(task.Repositories) > 0 { + data["repository_id"] = task.Repositories[0].RepositoryID + } + if task.Metadata != nil { data["metadata"] = task.Metadata } diff --git a/apps/web/app/tasks/tasks-page-client.tsx b/apps/web/app/tasks/tasks-page-client.tsx index 493159e53b..8757d8ef17 100644 --- a/apps/web/app/tasks/tasks-page-client.tsx +++ b/apps/web/app/tasks/tasks-page-client.tsx @@ -199,7 +199,6 @@ export function TasksPageClient({ setCreateDialogOpen(false) fetchTasks() }} - navigateOnSessionCreate={false} /> )} diff --git a/apps/web/components/combobox.tsx b/apps/web/components/combobox.tsx index 8dfbb64df4..d70fa510d4 100644 --- a/apps/web/components/combobox.tsx +++ b/apps/web/components/combobox.tsx @@ -54,11 +54,17 @@ export const Combobox = memo(function Combobox({ showSearch = true, }: ComboboxProps) { const [open, setOpen] = useState(false) + // Track the highlighted item. Defaults to the selected value so the current + // selection is highlighted when the popover opens (not the first item). + const [highlighted, setHighlighted] = useState("") const selectedOption = options.find((option) => option.value === value) return ( - + { + setOpen(next) + if (next) setHighlighted(value) + }}>