From ea1b731fd6137fde90ab8e3c26fd56b63e6b562f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Thu, 26 Feb 2026 23:06:07 +0000 Subject: [PATCH] feat: improve task launch (#297) --- apps/backend/cmd/kandev/adapters.go | 4 + .../simulated_agent_manager_test.go | 3 + .../orchestrator/event_handlers_test.go | 3 + .../orchestrator/executor/executor.go | 5 + .../orchestrator/executor/executor_resume.go | 24 +++ .../orchestrator/executor/executor_test.go | 3 + .../orchestrator/handlers/handlers.go | 140 +++++++------- .../orchestrator/scheduler/scheduler_test.go | 3 + .../internal/orchestrator/session_launch.go | 180 ++++++++++++++++++ .../orchestrator/session_launch_test.go | 113 +++++++++++ .../internal/orchestrator/task_operations.go | 37 ++++ .../orchestrator/task_operations_test.go | 158 +++++++++++++++ .../orchestrator/workflow_callbacks.go | 25 +-- .../internal/task/handlers/task_handlers.go | 13 +- .../task/handlers/task_http_handlers.go | 49 +++-- .../task/handlers/task_ws_handlers.go | 23 ++- apps/backend/pkg/websocket/actions.go | 3 + .../tiptap/tiptap-mermaid-extension.ts | 8 +- apps/web/components/kanban-with-preview.tsx | 17 +- .../components/kanban/graph2-step-node.tsx | 63 +----- .../kanban/graph2-task-pipeline.tsx | 6 +- .../kanban/swimlane-graph2-content.tsx | 4 +- .../components/task-create-dialog-submit.tsx | 80 +++----- .../components/task/chat/message-renderer.tsx | 8 +- .../mobile/session-task-switcher-sheet.tsx | 58 +++--- .../web/components/task/task-center-panel.tsx | 22 +-- .../components/task/task-session-sidebar.tsx | 114 ++++++----- .../domains/kanban/use-kanban-navigation.ts | 34 ++-- .../domains/session/use-session-agent.ts | 19 +- .../domains/session/use-session-launch.ts | 40 ++++ .../domains/session/use-session-resumption.ts | 26 +-- apps/web/hooks/use-session-layout-state.ts | 34 +--- apps/web/lib/api/domains/session-api.ts | 10 + apps/web/lib/services/session-approve.ts | 26 +++ .../lib/services/session-launch-helpers.ts | 107 +++++++++++ .../lib/services/session-launch-service.ts | 38 ++++ apps/web/lib/ui/state-icons.tsx | 11 +- apps/web/lib/utils/session-info.ts | 40 ++++ 38 files changed, 1119 insertions(+), 432 deletions(-) create mode 100644 apps/backend/internal/orchestrator/session_launch.go create mode 100644 apps/backend/internal/orchestrator/session_launch_test.go create mode 100644 apps/web/hooks/domains/session/use-session-launch.ts create mode 100644 apps/web/lib/services/session-approve.ts create mode 100644 apps/web/lib/services/session-launch-helpers.ts create mode 100644 apps/web/lib/services/session-launch-service.ts create mode 100644 apps/web/lib/utils/session-info.ts diff --git a/apps/backend/cmd/kandev/adapters.go b/apps/backend/cmd/kandev/adapters.go index ca62ebbc27..2675324ac4 100644 --- a/apps/backend/cmd/kandev/adapters.go +++ b/apps/backend/cmd/kandev/adapters.go @@ -234,6 +234,10 @@ func (a *lifecycleAdapter) PollRemoteStatusForRecords(ctx context.Context, recor a.mgr.PollRemoteStatusForRecords(ctx, lcRecords) } +func (a *lifecycleAdapter) CleanupStaleExecutionBySessionID(ctx context.Context, sessionID string) error { + return a.mgr.CleanupStaleExecutionBySessionID(ctx, sessionID) +} + func (a *lifecycleAdapter) GetRemoteRuntimeStatusBySession(ctx context.Context, sessionID string) (*executor.RemoteRuntimeStatus, error) { status, ok := a.mgr.GetRemoteStatusBySessionID(ctx, sessionID) if !ok || status == nil { diff --git a/apps/backend/internal/integration/simulated_agent_manager_test.go b/apps/backend/internal/integration/simulated_agent_manager_test.go index 88b348ef25..627c016a9e 100644 --- a/apps/backend/internal/integration/simulated_agent_manager_test.go +++ b/apps/backend/internal/integration/simulated_agent_manager_test.go @@ -415,3 +415,6 @@ func (s *SimulatedAgentManagerClient) GetRemoteRuntimeStatusBySession(ctx contex func (s *SimulatedAgentManagerClient) PollRemoteStatusForRecords(ctx context.Context, records []executor.RemoteStatusPollRequest) { } +func (s *SimulatedAgentManagerClient) CleanupStaleExecutionBySessionID(ctx context.Context, sessionID string) error { + return nil +} diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index 9e331bfb14..476e17d6cc 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_test.go @@ -144,6 +144,9 @@ func (m *mockAgentManager) GetRemoteRuntimeStatusBySession(_ context.Context, _ } func (m *mockAgentManager) PollRemoteStatusForRecords(_ context.Context, _ []executor.RemoteStatusPollRequest) { } +func (m *mockAgentManager) CleanupStaleExecutionBySessionID(_ context.Context, _ string) error { + return nil +} // --- Helpers --- diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index 8836a236a9..7ce871024b 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -109,6 +109,11 @@ type AgentManagerClient interface { // executor records. Used during startup to populate remote status cache before any // sessions are lazily resumed. PollRemoteStatusForRecords(ctx context.Context, records []RemoteStatusPollRequest) + + // CleanupStaleExecutionBySessionID removes a stale execution from the in-memory + // tracking store. Used when the agent process has exited but the execution entry + // was not cleaned up (e.g. prepared workspace where agent was never started). + CleanupStaleExecutionBySessionID(ctx context.Context, sessionID string) error } // RemoteRuntimeStatus mirrors runtime status details needed by orchestrator/UI. diff --git a/apps/backend/internal/orchestrator/executor/executor_resume.go b/apps/backend/internal/orchestrator/executor/executor_resume.go index 821b926436..e385b4c4de 100644 --- a/apps/backend/internal/orchestrator/executor/executor_resume.go +++ b/apps/backend/internal/orchestrator/executor/executor_resume.go @@ -2,6 +2,7 @@ package executor import ( "context" + "strings" "time" "github.com/kandev/kandev/internal/agent/lifecycle" @@ -10,6 +11,14 @@ import ( "go.uber.org/zap" ) +// isStaleExecutionError checks if an error from LaunchAgent indicates a stale +// execution in the lifecycle manager's in-memory store. This happens when +// PrepareTaskSession registered an execution but the agent was never started, +// or the agent exited without proper cleanup. +func isStaleExecutionError(err error) bool { + return err != nil && strings.Contains(err.Error(), "already has an agent running") +} + // repoInfo holds resolved repository details for agent launch. type repoInfo struct { RepositoryID string @@ -146,6 +155,21 @@ func (e *Executor) ResumeSession(ctx context.Context, session *models.TaskSessio req.Env = e.applyPreferredShellEnv(ctx, req.ExecutorType, req.Env) resp, err := e.agentManager.LaunchAgent(ctx, req) + if err != nil && isStaleExecutionError(err) { + // The lifecycle manager's in-memory store has a stale execution (agent not actually + // running). This happens when PrepareTaskSession registered an execution but the + // agent was never started, or the agent exited without cleanup. Clean up the stale + // entry and retry. + e.logger.Info("cleaning up stale execution and retrying launch", + zap.String("task_id", task.ID), + zap.String("session_id", session.ID)) + if cleanupErr := e.agentManager.CleanupStaleExecutionBySessionID(ctx, session.ID); cleanupErr != nil { + e.logger.Warn("failed to clean up stale execution", + zap.String("session_id", session.ID), + zap.Error(cleanupErr)) + } + resp, err = e.agentManager.LaunchAgent(ctx, req) + } if err != nil { e.logger.Error("failed to relaunch agent for session", zap.String("task_id", task.ID), diff --git a/apps/backend/internal/orchestrator/executor/executor_test.go b/apps/backend/internal/orchestrator/executor/executor_test.go index e45fc17a37..4af20585fb 100644 --- a/apps/backend/internal/orchestrator/executor/executor_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_test.go @@ -86,6 +86,9 @@ func (m *mockAgentManager) GetRemoteRuntimeStatusBySession(ctx context.Context, } func (m *mockAgentManager) PollRemoteStatusForRecords(ctx context.Context, records []RemoteStatusPollRequest) { } +func (m *mockAgentManager) CleanupStaleExecutionBySessionID(ctx context.Context, sessionID string) error { + return nil +} func (m *mockAgentManager) ResolveAgentProfile(ctx context.Context, profileID string) (*AgentProfileInfo, error) { if m.resolveAgentProfileFunc != nil { diff --git a/apps/backend/internal/orchestrator/handlers/handlers.go b/apps/backend/internal/orchestrator/handlers/handlers.go index 8be75c12e2..4472522578 100644 --- a/apps/backend/internal/orchestrator/handlers/handlers.go +++ b/apps/backend/internal/orchestrator/handlers/handlers.go @@ -3,7 +3,6 @@ package handlers import ( "context" - "strings" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/orchestrator" @@ -40,6 +39,7 @@ func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { d.RegisterFunc(ws.ActionTaskSessionStatus, h.wsGetTaskSessionStatus) d.RegisterFunc(ws.ActionTaskSessionPrepare, h.wsPrepareTaskSession) d.RegisterFunc(ws.ActionAgentCancel, h.wsCancelAgent) + d.RegisterFunc(ws.ActionSessionLaunch, h.wsLaunchSession) } // WS handlers @@ -93,6 +93,26 @@ func (h *Handlers) wsTriggerTask(ctx context.Context, msg *ws.Message) (*ws.Mess return ws.NewResponse(msg.ID, msg.Action, resp) } +func (h *Handlers) wsLaunchSession(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req orchestrator.LaunchSessionRequest + 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.LaunchSession(ctx, &req) + if err != nil { + h.logger.Error("failed to launch session", + zap.String("task_id", req.TaskID), + zap.String("intent", string(orchestrator.ResolveIntent(&req))), + zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to launch session: "+err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + type wsStartTaskRequest struct { TaskID string `json:"task_id"` SessionID string `json:"session_id,omitempty"` @@ -105,80 +125,47 @@ type wsStartTaskRequest struct { PlanMode bool `json:"plan_mode,omitempty"` } +// wsStartTask delegates to LaunchSession for backward compatibility. func (h *Handlers) wsStartTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req wsStartTaskRequest if err := msg.ParsePayload(&req); err != nil { return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) } - req.Prompt = strings.TrimSpace(req.Prompt) if req.TaskID == "" { return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) } - // agent_profile_id is only required when creating a new session (no session_id provided) if req.SessionID == "" && req.AgentProfileID == "" { 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 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)) - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to start task: "+err.Error(), nil) - } - resp := dto.StartTaskResponse{ - Success: true, - TaskID: req.TaskID, - TaskSessionID: req.SessionID, - State: "RUNNING", - } - return ws.NewResponse(msg.ID, msg.Action, resp) + launchReq := &orchestrator.LaunchSessionRequest{ + TaskID: req.TaskID, + SessionID: req.SessionID, + AgentProfileID: req.AgentProfileID, + ExecutorID: req.ExecutorID, + ExecutorProfileID: req.ExecutorProfileID, + Priority: req.Priority, + Prompt: req.Prompt, + WorkflowStepID: req.WorkflowStepID, + PlanMode: req.PlanMode, } - - // 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, false) - 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.ExecutorProfileID, req.Priority, req.Prompt, req.WorkflowStepID, req.PlanMode) + resp, err := h.service.LaunchSession(ctx, launchReq) if err != nil { h.logger.Error("failed to start task", zap.String("task_id", req.TaskID), zap.Error(err)) return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to start task: "+err.Error(), nil) } - resp := dto.StartTaskResponse{ - Success: true, - TaskID: execution.TaskID, - AgentExecutionID: execution.AgentExecutionID, - TaskSessionID: execution.SessionID, - State: string(execution.SessionState), + // Map to legacy StartTaskResponse format for backward compatibility + legacyResp := dto.StartTaskResponse{ + Success: resp.Success, + TaskID: resp.TaskID, + AgentExecutionID: resp.AgentExecutionID, + TaskSessionID: resp.SessionID, + State: resp.State, + WorktreePath: resp.WorktreePath, + WorktreeBranch: resp.WorktreeBranch, } - if execution.WorktreePath != "" { - resp.WorktreePath = &execution.WorktreePath - } - if execution.WorktreeBranch != "" { - resp.WorktreeBranch = &execution.WorktreeBranch - } - return ws.NewResponse(msg.ID, msg.Action, resp) + return ws.NewResponse(msg.ID, msg.Action, legacyResp) } type wsResumeTaskSessionRequest struct { @@ -186,6 +173,7 @@ type wsResumeTaskSessionRequest struct { TaskSessionID string `json:"session_id"` } +// wsResumeTaskSession delegates to LaunchSession with resume intent for backward compatibility. func (h *Handlers) wsResumeTaskSession(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req wsResumeTaskSessionRequest if err := msg.ParsePayload(&req); err != nil { @@ -198,7 +186,12 @@ func (h *Handlers) wsResumeTaskSession(ctx context.Context, msg *ws.Message) (*w return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil) } - execution, err := h.service.ResumeTaskSession(ctx, req.TaskID, req.TaskSessionID) + launchReq := &orchestrator.LaunchSessionRequest{ + TaskID: req.TaskID, + SessionID: req.TaskSessionID, + Intent: orchestrator.IntentResume, + } + resp, err := h.service.LaunchSession(ctx, launchReq) if err != nil { h.logger.Error("failed to resume task session", zap.String("task_id", req.TaskID), @@ -207,20 +200,17 @@ func (h *Handlers) wsResumeTaskSession(ctx context.Context, msg *ws.Message) (*w return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to resume task session: "+err.Error(), nil) } - resp := dto.ResumeTaskSessionResponse{ - Success: true, - TaskID: execution.TaskID, - AgentExecutionID: execution.AgentExecutionID, - TaskSessionID: execution.SessionID, - State: string(execution.SessionState), + // Map to legacy ResumeTaskSessionResponse format + legacyResp := dto.ResumeTaskSessionResponse{ + Success: resp.Success, + TaskID: resp.TaskID, + AgentExecutionID: resp.AgentExecutionID, + TaskSessionID: resp.SessionID, + State: resp.State, + WorktreePath: resp.WorktreePath, + WorktreeBranch: resp.WorktreeBranch, } - if execution.WorktreePath != "" { - resp.WorktreePath = &execution.WorktreePath - } - if execution.WorktreeBranch != "" { - resp.WorktreeBranch = &execution.WorktreeBranch - } - return ws.NewResponse(msg.ID, msg.Action, resp) + return ws.NewResponse(msg.ID, msg.Action, legacyResp) } type wsPrepareTaskSessionRequest struct { @@ -231,6 +221,7 @@ type wsPrepareTaskSessionResponse struct { SessionID string `json:"session_id"` } +// wsPrepareTaskSession delegates to LaunchSession with prepare intent for backward compatibility. func (h *Handlers) wsPrepareTaskSession(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req wsPrepareTaskSessionRequest if err := msg.ParsePayload(&req); err != nil { @@ -240,7 +231,12 @@ func (h *Handlers) wsPrepareTaskSession(ctx context.Context, msg *ws.Message) (* return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) } - sessionID, err := h.service.PrepareTaskSession(ctx, req.TaskID, "", "", "", "", true) + launchReq := &orchestrator.LaunchSessionRequest{ + TaskID: req.TaskID, + Intent: orchestrator.IntentPrepare, + LaunchWorkspace: true, + } + resp, err := h.service.LaunchSession(ctx, launchReq) if err != nil { h.logger.Error("failed to prepare task session", zap.String("task_id", req.TaskID), @@ -248,7 +244,7 @@ func (h *Handlers) wsPrepareTaskSession(ctx context.Context, msg *ws.Message) (* return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to prepare session: "+err.Error(), nil) } - return ws.NewResponse(msg.ID, msg.Action, wsPrepareTaskSessionResponse{SessionID: sessionID}) + return ws.NewResponse(msg.ID, msg.Action, wsPrepareTaskSessionResponse{SessionID: resp.SessionID}) } type wsStopTaskRequest struct { diff --git a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go index e5c25062bd..e8f0d11cf4 100644 --- a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go +++ b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go @@ -85,6 +85,9 @@ func (m *mockAgentManager) GetRemoteRuntimeStatusBySession(ctx context.Context, } func (m *mockAgentManager) PollRemoteStatusForRecords(ctx context.Context, records []executor.RemoteStatusPollRequest) { } +func (m *mockAgentManager) CleanupStaleExecutionBySessionID(ctx context.Context, sessionID string) error { + return nil +} func (m *mockAgentManager) CancelAgent(ctx context.Context, sessionID string) error { return nil diff --git a/apps/backend/internal/orchestrator/session_launch.go b/apps/backend/internal/orchestrator/session_launch.go new file mode 100644 index 0000000000..bd1593b365 --- /dev/null +++ b/apps/backend/internal/orchestrator/session_launch.go @@ -0,0 +1,180 @@ +package orchestrator + +import ( + "context" + "fmt" + "strings" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// SessionIntent represents the type of session operation requested. +type SessionIntent string + +const ( + IntentPrepare SessionIntent = "prepare" // Create session, optionally launch workspace, NO agent + IntentStart SessionIntent = "start" // Create session + launch agent (new session) + IntentStartCreated SessionIntent = "start_created" // Start agent on existing CREATED session + IntentResume SessionIntent = "resume" // Restart stopped session with resume token + IntentWorkflowStep SessionIntent = "workflow_step" // Start session with workflow step prompt config +) + +// LaunchSessionRequest is the unified request for session.launch. +type LaunchSessionRequest struct { + TaskID string `json:"task_id"` + Intent SessionIntent `json:"intent,omitempty"` + SessionID string `json:"session_id,omitempty"` + AgentProfileID string `json:"agent_profile_id,omitempty"` + ExecutorID string `json:"executor_id,omitempty"` + ExecutorProfileID string `json:"executor_profile_id,omitempty"` + Prompt string `json:"prompt,omitempty"` + PlanMode bool `json:"plan_mode,omitempty"` + WorkflowStepID string `json:"workflow_step_id,omitempty"` + Priority int `json:"priority,omitempty"` + LaunchWorkspace bool `json:"launch_workspace,omitempty"` + SkipMessageRecord bool `json:"skip_message_record,omitempty"` +} + +// LaunchSessionResponse is the unified response for session.launch. +type LaunchSessionResponse struct { + Success bool `json:"success"` + TaskID string `json:"task_id"` + SessionID string `json:"session_id,omitempty"` + AgentExecutionID string `json:"agent_execution_id,omitempty"` + State string `json:"state"` + WorktreePath *string `json:"worktree_path,omitempty"` + WorktreeBranch *string `json:"worktree_branch,omitempty"` +} + +// ResolveIntent infers the session intent from request fields when Intent is empty. +func ResolveIntent(req *LaunchSessionRequest) SessionIntent { + if req.Intent != "" { + return req.Intent + } + if req.SessionID != "" && req.WorkflowStepID != "" { + return IntentWorkflowStep + } + if req.SessionID != "" && req.Prompt == "" && req.AgentProfileID == "" { + return IntentResume + } + if req.SessionID != "" { + return IntentStartCreated + } + if req.LaunchWorkspace && req.Prompt == "" { + return IntentPrepare + } + return IntentStart +} + +// LaunchSession is the unified entry point for all session operations. +func (s *Service) LaunchSession(ctx context.Context, req *LaunchSessionRequest) (*LaunchSessionResponse, error) { + 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) + case IntentStart: + return s.launchStart(ctx, req) + case IntentStartCreated: + return s.launchStartCreated(ctx, req) + case IntentResume: + return s.launchResume(ctx, req) + case IntentWorkflowStep: + return s.launchWorkflowStep(ctx, req) + default: + return nil, fmt.Errorf("unknown intent: %s", intent) + } +} + +// launchPrepare creates a session entry without launching the agent. +func (s *Service) launchPrepare(ctx context.Context, req *LaunchSessionRequest) (*LaunchSessionResponse, error) { + sessionID, err := s.PrepareTaskSession( + ctx, req.TaskID, req.AgentProfileID, req.ExecutorID, + req.ExecutorProfileID, req.WorkflowStepID, req.LaunchWorkspace, + ) + if err != nil { + return nil, err + } + return &LaunchSessionResponse{ + Success: true, + TaskID: req.TaskID, + SessionID: sessionID, + State: string(models.TaskSessionStateCreated), + }, nil +} + +// launchStart creates a new session and launches the agent. +func (s *Service) launchStart(ctx context.Context, req *LaunchSessionRequest) (*LaunchSessionResponse, error) { + execution, err := s.StartTask( + ctx, req.TaskID, req.AgentProfileID, req.ExecutorID, + req.ExecutorProfileID, req.Priority, req.Prompt, + req.WorkflowStepID, req.PlanMode, + ) + if err != nil { + return nil, err + } + return executionToLaunchResponse(req.TaskID, execution), nil +} + +// launchStartCreated starts agent execution on an existing CREATED session. +func (s *Service) launchStartCreated(ctx context.Context, req *LaunchSessionRequest) (*LaunchSessionResponse, error) { + execution, err := s.StartCreatedSession( + ctx, req.TaskID, req.SessionID, req.AgentProfileID, + req.Prompt, req.SkipMessageRecord, + ) + if err != nil { + return nil, err + } + return executionToLaunchResponse(req.TaskID, execution), nil +} + +// launchResume resumes a stopped session. +func (s *Service) launchResume(ctx context.Context, req *LaunchSessionRequest) (*LaunchSessionResponse, error) { + execution, err := s.ResumeTaskSession(ctx, req.TaskID, req.SessionID) + if err != nil { + return nil, err + } + return executionToLaunchResponse(req.TaskID, execution), nil +} + +// launchWorkflowStep starts a session with workflow step prompt configuration. +func (s *Service) launchWorkflowStep(ctx context.Context, req *LaunchSessionRequest) (*LaunchSessionResponse, error) { + err := s.StartSessionForWorkflowStep(ctx, req.TaskID, req.SessionID, req.WorkflowStepID) + if err != nil { + return nil, err + } + return &LaunchSessionResponse{ + Success: true, + TaskID: req.TaskID, + SessionID: req.SessionID, + State: string(v1.TaskSessionStateRunning), + }, nil +} + +// executionToLaunchResponse converts a TaskExecution to a LaunchSessionResponse. +func executionToLaunchResponse(taskID string, exec *executor.TaskExecution) *LaunchSessionResponse { + resp := &LaunchSessionResponse{ + Success: true, + TaskID: taskID, + SessionID: exec.SessionID, + AgentExecutionID: exec.AgentExecutionID, + State: string(exec.SessionState), + } + if exec.WorktreePath != "" { + resp.WorktreePath = &exec.WorktreePath + } + if exec.WorktreeBranch != "" { + resp.WorktreeBranch = &exec.WorktreeBranch + } + return resp +} diff --git a/apps/backend/internal/orchestrator/session_launch_test.go b/apps/backend/internal/orchestrator/session_launch_test.go new file mode 100644 index 0000000000..eca59bb01e --- /dev/null +++ b/apps/backend/internal/orchestrator/session_launch_test.go @@ -0,0 +1,113 @@ +package orchestrator + +import ( + "testing" +) + +func TestResolveIntent(t *testing.T) { + tests := []struct { + name string + req LaunchSessionRequest + want SessionIntent + }{ + // Explicit intents take priority + { + name: "explicit start intent", + req: LaunchSessionRequest{TaskID: "t1", Intent: IntentStart}, + want: IntentStart, + }, + { + name: "explicit resume intent", + req: LaunchSessionRequest{TaskID: "t1", Intent: IntentResume, SessionID: "s1"}, + want: IntentResume, + }, + { + name: "explicit prepare intent", + req: LaunchSessionRequest{TaskID: "t1", Intent: IntentPrepare}, + want: IntentPrepare, + }, + { + name: "explicit start_created intent", + req: LaunchSessionRequest{TaskID: "t1", Intent: IntentStartCreated, SessionID: "s1"}, + want: IntentStartCreated, + }, + { + name: "explicit workflow_step intent", + req: LaunchSessionRequest{TaskID: "t1", Intent: IntentWorkflowStep, SessionID: "s1", WorkflowStepID: "ws1"}, + want: IntentWorkflowStep, + }, + + // Inferred intents (no explicit intent set) + { + name: "workflow_step inferred from session_id + workflow_step_id", + req: LaunchSessionRequest{TaskID: "t1", SessionID: "s1", WorkflowStepID: "ws1"}, + want: IntentWorkflowStep, + }, + { + name: "resume inferred from session_id only", + req: LaunchSessionRequest{TaskID: "t1", SessionID: "s1"}, + want: IntentResume, + }, + { + name: "resume inferred from session_id with no prompt and no agent_profile", + req: LaunchSessionRequest{TaskID: "t1", SessionID: "s1"}, + want: IntentResume, + }, + { + name: "start_created inferred from session_id + prompt", + req: LaunchSessionRequest{TaskID: "t1", SessionID: "s1", Prompt: "hello"}, + want: IntentStartCreated, + }, + { + name: "start_created inferred from session_id + agent_profile_id", + req: LaunchSessionRequest{TaskID: "t1", SessionID: "s1", AgentProfileID: "ap1"}, + want: IntentStartCreated, + }, + { + name: "start_created inferred from session_id + prompt + agent_profile_id", + req: LaunchSessionRequest{TaskID: "t1", SessionID: "s1", Prompt: "hello", AgentProfileID: "ap1"}, + want: IntentStartCreated, + }, + { + name: "prepare inferred from launch_workspace without prompt", + req: LaunchSessionRequest{TaskID: "t1", LaunchWorkspace: true}, + want: IntentPrepare, + }, + { + name: "start inferred from minimal request", + req: LaunchSessionRequest{TaskID: "t1"}, + want: IntentStart, + }, + { + name: "start inferred when prompt provided without session_id", + req: LaunchSessionRequest{TaskID: "t1", Prompt: "do something"}, + want: IntentStart, + }, + { + name: "start inferred when launch_workspace + prompt (not prepare)", + req: LaunchSessionRequest{TaskID: "t1", LaunchWorkspace: true, Prompt: "do something"}, + want: IntentStart, + }, + + // Edge cases + { + name: "resume wins over start_created when session_id set, no prompt, no agent_profile", + req: LaunchSessionRequest{TaskID: "t1", SessionID: "s1", ExecutorID: "e1"}, + want: IntentResume, + }, + { + name: "workflow_step wins over resume when both session_id and workflow_step_id set", + req: LaunchSessionRequest{TaskID: "t1", SessionID: "s1", WorkflowStepID: "ws1"}, + want: IntentWorkflowStep, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := ResolveIntent(&tt.req) + if got != tt.want { + t.Errorf("ResolveIntent() = %q, want %q", got, tt.want) + } + }) + } +} diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index d58743347d..057c45ef25 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -576,6 +576,14 @@ func (s *Service) ensureSessionRunning(ctx context.Context, sessionID string, se zap.String("session_id", sessionID), zap.String("session_state", string(session.State))) + // If the session is in CREATED state with an existing workspace (AgentExecutionID set), + // the workspace was prepared but the agent was never started. Use LaunchPreparedSession + // which routes to startAgentOnExistingWorkspace to reuse the workspace rather than + // ResumeSession which tries a full LaunchAgent and conflicts with the existing execution. + if session.State == models.TaskSessionStateCreated && session.AgentExecutionID != "" { + return s.startAgentOnPreparedWorkspace(ctx, sessionID, session) + } + running, err := s.repo.GetExecutorRunningBySessionID(ctx, sessionID) if err != nil || running == nil { return fmt.Errorf("session is not resumable: no executor record (state: %s)", session.State) @@ -611,6 +619,35 @@ func (s *Service) ensureSessionRunning(ctx context.Context, sessionID string, se return nil } +// startAgentOnPreparedWorkspace starts the agent subprocess on a session whose workspace +// was prepared (agentctl running) but whose agent process was never started. This avoids +// the "session already has an agent running" error from ResumeSession which tries a full +// LaunchAgent and conflicts with the existing execution in the lifecycle manager's store. +func (s *Service) startAgentOnPreparedWorkspace(ctx context.Context, sessionID string, session *models.TaskSession) error { + s.logger.Debug("session has prepared workspace but no agent, starting agent on existing workspace", + zap.String("session_id", sessionID), + zap.String("agent_execution_id", session.AgentExecutionID)) + + launchCtx := context.WithoutCancel(ctx) + task, err := s.scheduler.GetTask(launchCtx, session.TaskID) + if err != nil { + return fmt.Errorf("failed to get task for prepared session: %w", err) + } + if _, err = s.executor.LaunchPreparedSession(launchCtx, task, sessionID, executor.LaunchOptions{ + AgentProfileID: session.AgentProfileID, + ExecutorID: session.ExecutorID, + StartAgent: true, + }); err != nil { + return fmt.Errorf("failed to start agent on prepared workspace: %w", err) + } + + if err := s.waitForSessionReady(ctx, sessionID); err != nil { + return fmt.Errorf("session not ready after starting agent: %w", err) + } + s.logger.Debug("agent started on prepared workspace and ready for prompt") + return nil +} + // waitForSessionReady polls the session state until the agent is ready for prompts. func (s *Service) waitForSessionReady(ctx context.Context, sessionID string) error { const ( diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index 7c6225e9cc..f14fb4bf18 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -4,12 +4,15 @@ import ( "context" "errors" "fmt" + "strings" "testing" "time" sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator/queue" + "github.com/kandev/kandev/internal/orchestrator/scheduler" "github.com/kandev/kandev/internal/task/models" v1 "github.com/kandev/kandev/pkg/api/v1" ) @@ -541,3 +544,158 @@ func TestReconcileSessionsOnStartup(t *testing.T) { } }) } + +// --- ensureSessionRunning: prepared workspace --- + +func TestEnsureSessionRunning_PreparedWorkspace(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + + // Seed task and session in CREATED state (workspace prepared, agent not started) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateCreated) + + // Set AgentExecutionID to simulate a prepared workspace + session, err := repo.GetTaskSession(ctx, "session1") + if err != nil { + t.Fatalf("failed to load session: %v", err) + } + session.AgentExecutionID = "exec-prepare-1" + session.AgentProfileID = "profile1" + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("failed to update session: %v", err) + } + + // Create a wrapped mock agent manager that transitions the session to WAITING_FOR_INPUT + // when StartAgentProcess is called (simulating the agent starting successfully). + startAgentProcessCalled := false + wrappedMgr := &sessionUpdatingAgentManager{ + mockAgentManager: &mockAgentManager{isAgentRunning: false}, + repo: repo, + sessionID: "session1", + taskID: "task1", + onStartCalled: &startAgentProcessCalled, + } + + taskRepo := newMockTaskRepo() + taskRepo.tasks["task1"] = &v1.Task{ + ID: "task1", + Title: "Test Task", + Description: "desc", + State: v1.TaskStateInProgress, + } + + log := testLogger() + exec := executor.NewExecutor(wrappedMgr, repo, log, executor.ExecutorConfig{}) + sched := scheduler.NewScheduler(queue.NewTaskQueue(100), exec, taskRepo, log, scheduler.DefaultSchedulerConfig()) + + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, wrappedMgr) + svc.executor = exec + svc.scheduler = sched + + // Re-load session for the call + session, err = repo.GetTaskSession(ctx, "session1") + if err != nil { + t.Fatalf("failed to reload session: %v", err) + } + + err = svc.ensureSessionRunning(ctx, "session1", session) + if err != nil { + t.Fatalf("ensureSessionRunning failed: %v", err) + } + + if !startAgentProcessCalled { + t.Fatal("expected StartAgentProcess to be called (prepared workspace path)") + } + + // Verify the session transitioned through STARTING + updated, err := repo.GetTaskSession(ctx, "session1") + if err != nil { + t.Fatalf("failed to reload session: %v", err) + } + if updated.State != models.TaskSessionStateWaitingForInput { + t.Fatalf("expected session state %q, got %q", models.TaskSessionStateWaitingForInput, updated.State) + } +} + +func TestEnsureSessionRunning_WaitingForInputUsesResumePath(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + + // Session in WAITING_FOR_INPUT without executor running record → resume path fails gracefully + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateWaitingForInput) + + agentMgr := &mockAgentManager{isAgentRunning: false} + log := testLogger() + exec := executor.NewExecutor(agentMgr, repo, log, executor.ExecutorConfig{}) + + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + svc.executor = exec + + session, err := repo.GetTaskSession(ctx, "session1") + if err != nil { + t.Fatalf("failed to load session: %v", err) + } + + // Should fail because there is no executor running record (resume path) + err = svc.ensureSessionRunning(ctx, "session1", session) + if err == nil { + t.Fatal("expected error for WAITING_FOR_INPUT session without executor record") + } + // Verify it took the resume path (error mentions "not resumable") + if !strings.Contains(err.Error(), "not resumable") { + t.Fatalf("expected 'not resumable' error from resume path, got: %v", err) + } +} + +func TestEnsureSessionRunning_CreatedWithoutExecutionUsesResumePath(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + + // Session in CREATED state WITHOUT AgentExecutionID → resume path (not prepared workspace) + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateCreated) + + agentMgr := &mockAgentManager{isAgentRunning: false} + log := testLogger() + exec := executor.NewExecutor(agentMgr, repo, log, executor.ExecutorConfig{}) + + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + svc.executor = exec + + session, err := repo.GetTaskSession(ctx, "session1") + if err != nil { + t.Fatalf("failed to load session: %v", err) + } + + // AgentExecutionID is empty → should NOT take prepared workspace path + // Should fail with "not resumable" because no executor running record + err = svc.ensureSessionRunning(ctx, "session1", session) + if err == nil { + t.Fatal("expected error for CREATED session without executor record") + } + if !strings.Contains(err.Error(), "not resumable") { + t.Fatalf("expected 'not resumable' error from resume path, got: %v", err) + } +} + +// sessionUpdatingAgentManager wraps mockAgentManager to update session state +// when StartAgentProcess is called, simulating the agent initialization flow. +type sessionUpdatingAgentManager struct { + *mockAgentManager + repo *sqliterepo.Repository + sessionID string + taskID string + onStartCalled *bool +} + +func (m *sessionUpdatingAgentManager) StartAgentProcess(_ context.Context, _ string) error { + *m.onStartCalled = true + // Simulate the agent starting by transitioning session to WAITING_FOR_INPUT + ctx := context.Background() + sess, err := m.repo.GetTaskSession(ctx, m.sessionID) + if err == nil && sess != nil { + sess.State = models.TaskSessionStateWaitingForInput + sess.UpdatedAt = time.Now().UTC() + _ = m.repo.UpdateTaskSession(ctx, sess) + } + return nil +} diff --git a/apps/backend/internal/orchestrator/workflow_callbacks.go b/apps/backend/internal/orchestrator/workflow_callbacks.go index 05e6ac53eb..1f1d839fa1 100644 --- a/apps/backend/internal/orchestrator/workflow_callbacks.go +++ b/apps/backend/internal/orchestrator/workflow_callbacks.go @@ -5,7 +5,6 @@ import ( "fmt" "github.com/kandev/kandev/internal/workflow/engine" - wfmodels "github.com/kandev/kandev/internal/workflow/models" ) // buildWorkflowCallbacks creates the callback registry for the workflow engine. @@ -82,24 +81,14 @@ func (c *autoStartAgentCallback) Execute(ctx context.Context, in engine.ActionIn return engine.ActionResult{}, nil } - session, err := c.svc.repo.GetTaskSession(ctx, in.State.SessionID) + _, err := c.svc.LaunchSession(ctx, &LaunchSessionRequest{ + TaskID: in.State.TaskID, + Intent: IntentWorkflowStep, + SessionID: in.State.SessionID, + WorkflowStepID: in.Step.ID, + }) if err != nil { - return engine.ActionResult{}, fmt.Errorf("load session for auto-start: %w", err) - } - - // Reconstruct the workflow step to build the prompt. - // The step's prompt template and plan mode flag drive prompt construction. - step, err := c.svc.workflowStepGetter.GetStep(ctx, in.Step.ID) - if err != nil { - return engine.ActionResult{}, fmt.Errorf("load step for auto-start: %w", err) - } - - effectivePrompt := c.svc.buildWorkflowPrompt(in.State.TaskDescription, step, in.State.TaskID) - planMode := step.HasOnEnterAction(wfmodels.OnEnterEnablePlanMode) - - err = c.svc.autoStartStepPrompt(ctx, in.State.TaskID, session, in.Step.Name, effectivePrompt, planMode, true) - if err != nil { - return engine.ActionResult{}, fmt.Errorf("auto-start prompt failed: %w", err) + return engine.ActionResult{}, fmt.Errorf("auto-start via LaunchSession failed: %w", err) } return engine.ActionResult{}, nil } diff --git a/apps/backend/internal/task/handlers/task_handlers.go b/apps/backend/internal/task/handlers/task_handlers.go index 57031e8f24..ce5dcf5e57 100644 --- a/apps/backend/internal/task/handlers/task_handlers.go +++ b/apps/backend/internal/task/handlers/task_handlers.go @@ -6,7 +6,7 @@ import ( "github.com/gin-gonic/gin" "github.com/kandev/kandev/internal/common/logger" - "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator" "github.com/kandev/kandev/internal/task/dto" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/task/service" @@ -31,15 +31,8 @@ type TaskHandlers struct { } type OrchestratorStarter interface { - // StartTask starts agent execution for a task. - StartTask(ctx context.Context, taskID string, agentProfileID string, executorID string, executorProfileID string, priority int, prompt string, workflowStepID string, planMode bool) (*executor.TaskExecution, error) - // PrepareTaskSession creates a session entry without launching the agent. - // When launchWorkspace is true, the workspace infrastructure (agentctl) is launched - // synchronously so file browsing works immediately. When false, the workspace launch - // is deferred to StartTaskWithSession (useful for remote executors where provisioning is slow). - PrepareTaskSession(ctx context.Context, taskID string, agentProfileID string, executorID string, executorProfileID string, workflowStepID string, launchWorkspace bool) (string, error) - // StartTaskWithSession starts agent execution for a task using a pre-created session. - StartTaskWithSession(ctx context.Context, taskID string, sessionID string, agentProfileID string, executorID string, executorProfileID string, priority int, prompt string, workflowStepID string, planMode bool) (*executor.TaskExecution, error) + // LaunchSession is the unified entry point for all session operations. + LaunchSession(ctx context.Context, req *orchestrator.LaunchSessionRequest) (*orchestrator.LaunchSessionResponse, error) } func NewTaskHandlers(svc *service.Service, orchestrator OrchestratorStarter, repo handlerRepo, planService *service.PlanService, log *logger.Logger) *TaskHandlers { diff --git a/apps/backend/internal/task/handlers/task_http_handlers.go b/apps/backend/internal/task/handlers/task_http_handlers.go index 78b2042855..f1bec9b38f 100644 --- a/apps/backend/internal/task/handlers/task_http_handlers.go +++ b/apps/backend/internal/task/handlers/task_http_handlers.go @@ -9,6 +9,7 @@ import ( "github.com/gin-gonic/gin" "github.com/kandev/kandev/internal/common/constants" + "github.com/kandev/kandev/internal/orchestrator" "github.com/kandev/kandev/internal/task/dto" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/task/service" @@ -397,14 +398,19 @@ func (h *TaskHandlers) handlePostCreateTaskSession( return } if body.PrepareSession && !body.StartAgent { - // Create session entry without launching the agent. - // The session stays in CREATED state until the user triggers it. - // Launch workspace so file browsing works immediately. - sessionID, err := h.orchestrator.PrepareTaskSession(c.Request.Context(), taskID, body.AgentProfileID, body.ExecutorID, body.ExecutorProfileID, resolvedStepID, true) + resp, err := h.orchestrator.LaunchSession(c.Request.Context(), &orchestrator.LaunchSessionRequest{ + TaskID: taskID, + Intent: orchestrator.IntentPrepare, + AgentProfileID: body.AgentProfileID, + ExecutorID: body.ExecutorID, + ExecutorProfileID: body.ExecutorProfileID, + WorkflowStepID: resolvedStepID, + LaunchWorkspace: true, + }) if err != nil { h.logger.Error("failed to prepare session for task", zap.Error(err), zap.String("task_id", taskID)) } else { - response.TaskSessionID = sessionID + response.TaskSessionID = resp.SessionID } } else if body.StartAgent { h.startAgentForNewTask(c.Request.Context(), response, taskID, description, body, resolvedStepID) @@ -422,36 +428,45 @@ func (h *TaskHandlers) startAgentForNewTask( resolvedStepID string, ) { // Create session entry synchronously so we can return the session ID immediately. - // Skip workspace launch — StartTaskWithSession will handle it in the background goroutine. + // Skip workspace launch — the start intent will handle it in the background goroutine. // This prevents blocking for 30-60s on remote executors (sprites, remote_docker). - sessionID, err := h.orchestrator.PrepareTaskSession(ctx, taskID, body.AgentProfileID, body.ExecutorID, body.ExecutorProfileID, resolvedStepID, false) + prepResp, err := h.orchestrator.LaunchSession(ctx, &orchestrator.LaunchSessionRequest{ + TaskID: taskID, + Intent: orchestrator.IntentPrepare, + AgentProfileID: body.AgentProfileID, + ExecutorID: body.ExecutorID, + ExecutorProfileID: body.ExecutorProfileID, + WorkflowStepID: resolvedStepID, + }) if err != nil { h.logger.Error("failed to prepare session for task", zap.Error(err), zap.String("task_id", taskID)) - // Continue without session - task was created successfully return } + sessionID := prepResp.SessionID response.TaskSessionID = sessionID // Launch agent asynchronously so the HTTP request can return immediately. // The frontend will receive WebSocket updates when the agent actually starts. - executorID := body.ExecutorID // Capture for goroutine - stepID := resolvedStepID // Capture for goroutine go func() { - // Use a longer timeout to accommodate setup scripts (which can take minutes for npm install, etc.) startCtx, cancel := context.WithTimeout(context.Background(), constants.AgentLaunchTimeout) defer cancel() - // Use task description as the initial prompt with workflow step config (prompt prefix/suffix, plan mode) - execution, err := h.orchestrator.StartTaskWithSession(startCtx, taskID, sessionID, body.AgentProfileID, executorID, body.ExecutorProfileID, body.Priority, description, stepID, body.PlanMode) + launchResp, err := h.orchestrator.LaunchSession(startCtx, &orchestrator.LaunchSessionRequest{ + TaskID: taskID, + Intent: orchestrator.IntentStartCreated, + SessionID: sessionID, + AgentProfileID: body.AgentProfileID, + Prompt: description, + SkipMessageRecord: true, + PlanMode: body.PlanMode, + }) if err != nil { h.logger.Error("failed to start agent for task (async)", zap.Error(err), zap.String("task_id", taskID), zap.String("session_id", sessionID)) return } h.logger.Info("agent started for task (async)", zap.String("task_id", taskID), - zap.String("session_id", execution.SessionID), - zap.String("executor_id", executorID), - zap.String("workflow_step_id", stepID), - zap.String("execution_id", execution.AgentExecutionID)) + zap.String("session_id", launchResp.SessionID), + zap.String("execution_id", launchResp.AgentExecutionID)) }() } diff --git a/apps/backend/internal/task/handlers/task_ws_handlers.go b/apps/backend/internal/task/handlers/task_ws_handlers.go index b61230d80f..a30dd96e7f 100644 --- a/apps/backend/internal/task/handlers/task_ws_handlers.go +++ b/apps/backend/internal/task/handlers/task_ws_handlers.go @@ -4,6 +4,7 @@ import ( "context" "strings" + "github.com/kandev/kandev/internal/orchestrator" "github.com/kandev/kandev/internal/task/dto" "github.com/kandev/kandev/internal/task/service" v1 "github.com/kandev/kandev/pkg/api/v1" @@ -146,9 +147,19 @@ func (h *TaskHandlers) wsCreateTask(ctx context.Context, msg *ws.Message) (*ws.M taskDTO := dto.FromTask(task) response := createTaskResponse{TaskDTO: taskDTO} if req.StartAgent && req.AgentProfileID != "" && h.orchestrator != nil { - // Use task description as the initial prompt with workflow step config (prompt prefix/suffix, plan mode) + // Use task description as the initial prompt with workflow step config // Use taskDTO.WorkflowStepID (backend-resolved) instead of req.WorkflowStepID - execution, err := h.orchestrator.StartTask(ctx, taskDTO.ID, req.AgentProfileID, req.ExecutorID, req.ExecutorProfileID, req.Priority, taskDTO.Description, taskDTO.WorkflowStepID, req.PlanMode) + launchResp, err := h.orchestrator.LaunchSession(ctx, &orchestrator.LaunchSessionRequest{ + TaskID: taskDTO.ID, + Intent: orchestrator.IntentStart, + AgentProfileID: req.AgentProfileID, + ExecutorID: req.ExecutorID, + ExecutorProfileID: req.ExecutorProfileID, + Priority: req.Priority, + Prompt: taskDTO.Description, + WorkflowStepID: taskDTO.WorkflowStepID, + PlanMode: req.PlanMode, + }) if err != nil { h.logger.Error("failed to start agent for task", zap.Error(err)) return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to start agent for task", nil) @@ -156,10 +167,10 @@ func (h *TaskHandlers) wsCreateTask(ctx context.Context, msg *ws.Message) (*ws.M h.logger.Info("wsCreateTask started agent", zap.String("task_id", taskDTO.ID), zap.String("executor_id", req.ExecutorID), - zap.String("workflow_step_id", req.WorkflowStepID), - zap.String("session_id", execution.SessionID)) - response.TaskSessionID = execution.SessionID - response.AgentExecutionID = execution.AgentExecutionID + zap.String("workflow_step_id", taskDTO.WorkflowStepID), + zap.String("session_id", launchResp.SessionID)) + response.TaskSessionID = launchResp.SessionID + response.AgentExecutionID = launchResp.AgentExecutionID } return ws.NewResponse(msg.ID, msg.Action, response) } diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 26e475da4f..4b213ed3cd 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -74,6 +74,9 @@ const ( ActionTaskSessionStatus = "task.session.status" ActionTaskSessionPrepare = "task.session.prepare" + // Unified session launch + ActionSessionLaunch = "session.launch" + // Agent actions ActionAgentList = "agent.list" ActionAgentLaunch = "agent.launch" diff --git a/apps/web/components/editors/tiptap/tiptap-mermaid-extension.ts b/apps/web/components/editors/tiptap/tiptap-mermaid-extension.ts index 3508dbe5db..53279adcfc 100644 --- a/apps/web/components/editors/tiptap/tiptap-mermaid-extension.ts +++ b/apps/web/components/editors/tiptap/tiptap-mermaid-extension.ts @@ -8,7 +8,13 @@ import type { Node as PmNode } from "@tiptap/pm/model"; import type { NodeView } from "@tiptap/pm/view"; import CodeBlockLowlight from "@tiptap/extension-code-block-lowlight"; import mermaid from "mermaid"; -import { DEFAULT_SCALE, SCALE_STEP, MIN_SCALE, MAX_SCALE, getSvgDimensions } from "@/components/shared/mermaid-utils"; +import { + DEFAULT_SCALE, + SCALE_STEP, + MIN_SCALE, + MAX_SCALE, + getSvgDimensions, +} from "@/components/shared/mermaid-utils"; /** * Known mermaid diagram type keywords. If a code block's first line starts diff --git a/apps/web/components/kanban-with-preview.tsx b/apps/web/components/kanban-with-preview.tsx index 9c10fd70b6..404be7971b 100644 --- a/apps/web/components/kanban-with-preview.tsx +++ b/apps/web/components/kanban-with-preview.tsx @@ -12,6 +12,8 @@ import { Task } from "./kanban-card"; import type { KanbanState } from "@/lib/state/slices"; import { PREVIEW_PANEL } from "@/lib/settings/constants"; import { linkToSession } from "@/lib/links"; +import { launchSession } from "@/lib/services/session-launch-service"; +import { buildPrepareRequest } from "@/lib/services/session-launch-helpers"; type KanbanWithPreviewProps = { initialTaskId?: string; @@ -99,7 +101,7 @@ export function KanbanWithPreview({ initialTaskId }: KanbanWithPreviewProps) { // Use custom hooks for layout and session management const { containerRef, shouldFloat, kanbanWidth } = useKanbanLayout(isOpen, previewWidthPx); - const { sessionId: selectedTaskSessionId } = useTaskSession(selectedTaskId ?? null); + const { sessionId: selectedTaskSessionId, isLoading } = useTaskSession(selectedTaskId ?? null); // Track resize state const isResizingRef = useRef(false); @@ -129,6 +131,19 @@ export function KanbanWithPreview({ initialTaskId }: KanbanWithPreviewProps) { } }, [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]); + const handleNavigateToTask = useCallback( (task: Task, sessionId: string) => { router.push(linkToSession(sessionId)); diff --git a/apps/web/components/kanban/graph2-step-node.tsx b/apps/web/components/kanban/graph2-step-node.tsx index e1354300f3..733987485f 100644 --- a/apps/web/components/kanban/graph2-step-node.tsx +++ b/apps/web/components/kanban/graph2-step-node.tsx @@ -16,43 +16,6 @@ import type { WorkflowStep } from "@/components/kanban-column"; type StepPhase = "past" | "current" | "future"; -type StatusInfo = { - label: string; - color: string; -}; - -function getStatusInfo(task: Task): StatusInfo { - if (task.reviewStatus === "pending" && task.state !== "IN_PROGRESS") { - return { label: "Needs Approval", color: "text-amber-500" }; - } - if (task.reviewStatus === "changes_requested") { - return { label: "Changes Requested", color: "text-amber-500" }; - } - - switch (task.state) { - case "IN_PROGRESS": - case "SCHEDULING": - return { label: "Running", color: "text-blue-500" }; - case "WAITING_FOR_INPUT": - return { label: "Waiting", color: "text-amber-500" }; - case "COMPLETED": - return { label: "Completed", color: "text-green-500" }; - case "FAILED": - return { label: "Failed", color: "text-red-500" }; - case "CANCELLED": - return { label: "Cancelled", color: "text-red-500" }; - case "TODO": - case "CREATED": - return { label: "Todo", color: "text-muted-foreground" }; - case "REVIEW": - return { label: "Review", color: "text-yellow-500" }; - case "BLOCKED": - return { label: "Blocked", color: "text-yellow-500" }; - default: - return { label: "Pending", color: "text-muted-foreground" }; - } -} - function isRunningState(state?: string): boolean { return state === "IN_PROGRESS" || state === "SCHEDULING"; } @@ -71,7 +34,7 @@ export type Graph2StepNodeProps = { }; const NODE_CLASS = - "w-[130px] h-[44px] rounded-lg shrink-0 px-2.5 flex flex-col items-start justify-center"; + "w-[130px] h-[36px] rounded-lg shrink-0 px-2.5 flex flex-col items-start justify-center"; function PastNode({ step }: { step: WorkflowStep }) { return ( @@ -143,7 +106,6 @@ export function Graph2StepNode({ if (phase === "future") return ; // Current phase - const status = getStatusInfo(task); const running = isRunningState(task.state); const handleClick = () => { @@ -176,29 +138,18 @@ export function Graph2StepNode({ onClick={handleClick} className={cn( NODE_CLASS, - "bg-background cursor-pointer", - "hover:bg-accent/30 transition-colors", + "cursor-pointer transition-colors bg-background hover:bg-accent/30", running - ? "border border-muted-foreground/20 node-border-running" - : "border-2 ring-1 ring-offset-1 ring-offset-background", + ? "border-1 border-accent/50 node-border-running" + : "border-1 border-accent/50", )} - style={ - running - ? undefined - : { - borderColor: step.color || "hsl(var(--primary))", - // @ts-expect-error -- CSS custom property for ring color - "--tw-ring-color": step.color || "hsl(var(--primary))", - } - } >
{getTaskStateIcon(task.state, "h-3 w-3")}
- {status.label} + + {step.title} +
- - {step.title} - {isHovered && hasNext && nextStepId && ( diff --git a/apps/web/components/kanban/graph2-task-pipeline.tsx b/apps/web/components/kanban/graph2-task-pipeline.tsx index f0a215ee71..4b82476953 100644 --- a/apps/web/components/kanban/graph2-task-pipeline.tsx +++ b/apps/web/components/kanban/graph2-task-pipeline.tsx @@ -38,7 +38,7 @@ export type Graph2TaskPipelineProps = { steps: WorkflowStep[]; onMoveTask: (task: Task, targetStepId: string) => void; onPreviewTask: (task: Task) => void; - onEditTask: (task: Task) => void; + onOpenTask: (task: Task) => void; onDeleteTask: (task: Task) => void; isMoving?: boolean; isDeleting?: boolean; @@ -110,7 +110,7 @@ export function Graph2TaskPipeline({ steps, onMoveTask, onPreviewTask, - onEditTask, + onOpenTask, onDeleteTask, isMoving, isDeleting, @@ -128,7 +128,7 @@ export function Graph2TaskPipeline({