dreampiler/kandev
1
0
Fork 0
Code Issues 43 Pull Requests Actions 6 Packages Projects Releases Wiki Activity

feat: improve task launch (#297)

Browse Source
This commit is contained in:
Carlos Florêncio
2026-02-26 23:06:07 +00:00
committed by GitHub
parent 028f5e413f
commit ea1b731fd6
38 changed files with 1119 additions and 432 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/cmd/kandev/adapters.go
+4
View File
@@ -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 {
apps/backend/internal/integration/simulated_agent_manager_test.go
+3
View File
@@ -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
}
apps/backend/internal/orchestrator/event_handlers_test.go
+3
View File
@@ -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 ---
apps/backend/internal/orchestrator/executor/executor.go
+5
View File
@@ -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.
apps/backend/internal/orchestrator/executor/executor_resume.go
+24
View File
@@ -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),
apps/backend/internal/orchestrator/executor/executor_test.go
+3
View File
@@ -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 {
apps/backend/internal/orchestrator/handlers/handlers.go
+68 -72
View File
@@ -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 {
apps/backend/internal/orchestrator/scheduler/scheduler_test.go
+3
View File
@@ -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
apps/backend/internal/orchestrator/session_launch.go
+180
View File
@@ -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
}
apps/backend/internal/orchestrator/session_launch_test.go
+113
View File
@@ -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)
}
})
}
}
apps/backend/internal/orchestrator/task_operations.go
+37
View File
@@ -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 (
apps/backend/internal/orchestrator/task_operations_test.go
+158
View File
@@ -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
}
apps/backend/internal/orchestrator/workflow_callbacks.go
+7 -18
View File
@@ -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
}
apps/backend/internal/task/handlers/task_handlers.go
+3 -10
View File
@@ -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 {
apps/backend/internal/task/handlers/task_http_handlers.go
+32 -17
View File
@@ -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))
}()
}
apps/backend/internal/task/handlers/task_ws_handlers.go
+17 -6
View File
@@ -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)
}
apps/backend/pkg/websocket/actions.go
+3
View File
@@ -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"
apps/web/components/editors/tiptap/tiptap-mermaid-extension.ts
+7 -1
View File
@@ -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
apps/web/components/kanban-with-preview.tsx
+16 -1
View File
@@ -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<string | null>(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));
apps/web/components/kanban/graph2-step-node.tsx
+7 -56
View File
@@ -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 <FutureNode step={step} />;
// 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))",
}
}
>
<div className="flex items-center gap-1.5 w-full">
<div className="shrink-0">{getTaskStateIcon(task.state, "h-3 w-3")}</div>
<span className={cn("text-[10px] font-medium", status.color)}>{status.label}</span>
<span className="text-[11px] font-medium text-foreground truncate">
{step.title}
</span>
</div>
<span className="text-[11px] text-foreground/80 truncate w-full text-left">
{step.title}
</span>
</button>
{isHovered && hasNext && nextStepId && (
apps/web/components/kanban/graph2-task-pipeline.tsx
+3 -3
View File
@@ -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({
<div className="flex items-center gap-3">
<button
type="button"
onClick={() => onEditTask(task)}
onClick={() => onOpenTask(task)}
className={cn(
"w-[160px] shrink-0 rounded-md px-2.5 py-1.5 text-left transition-colors cursor-pointer",
"hover:bg-accent/60 active:bg-accent/80",
apps/web/components/kanban/swimlane-graph2-content.tsx
+2 -2
View File
@@ -10,7 +10,7 @@ export function SwimlaneGraph2Content({
steps,
tasks,
onPreviewTask,
onEditTask,
onOpenTask,
onDeleteTask,
onMoveError,
deletingTaskId,
@@ -58,7 +58,7 @@ export function SwimlaneGraph2Content({
steps={steps}
onMoveTask={handleMoveTask}
onPreviewTask={onPreviewTask}
onEditTask={onEditTask}
onOpenTask={onOpenTask}
onDeleteTask={onDeleteTask}
isMoving={movingTaskId === task.id}
isDeleting={deletingTaskId === task.id}
apps/web/components/task-create-dialog-submit.tsx
+25 -55
View File
@@ -5,7 +5,8 @@ import { useRouter } from "next/navigation";
import type { Task, LocalRepository } from "@/lib/types/http";
import { createTask, updateTask } from "@/lib/api";
import { useAppStore } from "@/components/state-provider";
import { getWebSocketClient } from "@/lib/ws/connection";
import { launchSession } from "@/lib/services/session-launch-service";
import { buildStartRequest } from "@/lib/services/session-launch-helpers";
import { useToast } from "@/components/toast-provider";
import { linkToSession } from "@/lib/links";
import {
@@ -17,14 +18,6 @@ import {
const GENERIC_ERROR_MESSAGE = "An error occurred";
interface OrchestratorStartResponse {
success: boolean;
task_id: string;
agent_instance_id: string;
session_id?: string;
state: string;
}
export type SubmitHandlersDeps = {
isSessionMode: boolean;
isEditMode: boolean;
@@ -166,20 +159,12 @@ export function useTaskSubmitHandlers({
setIsCreatingSession(true);
try {
const client = getWebSocketClient();
if (!client) throw new Error("WebSocket client not available");
const response = await client.request<OrchestratorStartResponse>(
"orchestrator.start",
{
task_id: taskId,
agent_profile_id: agentProfileId,
executor_id: executorId,
executor_profile_id: executorProfileId || undefined,
prompt: trimmedDescription,
},
15000,
);
const { request } = buildStartRequest(taskId, agentProfileId, {
executorId,
executorProfileId: executorProfileId || undefined,
prompt: trimmedDescription,
});
const response = await launchSession(request);
const newSessionId = response?.session_id;
if (newSessionId) {
@@ -237,24 +222,16 @@ export function useTaskSubmitHandlers({
let taskSessionId: string | null = null;
if (agentProfileId) {
const client = getWebSocketClient();
if (client) {
try {
const response = await client.request<OrchestratorStartResponse>(
"orchestrator.start",
{
task_id: updatedTask.id,
agent_profile_id: agentProfileId,
executor_id: executorId,
executor_profile_id: executorProfileId || undefined,
prompt: trimmedDescription || "",
},
15000,
);
taskSessionId = response?.session_id ?? null;
} catch (error) {
console.error("[TaskCreateDialog] failed to start agent:", error);
}
try {
const { request } = buildStartRequest(updatedTask.id, agentProfileId, {
executorId,
executorProfileId: executorProfileId || undefined,
prompt: trimmedDescription || "",
});
const response = await launchSession(request);
taskSessionId = response?.session_id ?? null;
} catch (error) {
console.error("[TaskCreateDialog] failed to start agent:", error);
}
}
@@ -400,20 +377,13 @@ export function useTaskSubmitHandlers({
const result = await performTaskUpdate();
if (!result) return;
const { updatedTask, trimmedDescription } = result;
const client = getWebSocketClient();
if (!client) throw new Error("WebSocket client not available");
const response = await client.request<OrchestratorStartResponse>(
"orchestrator.start",
{
task_id: updatedTask.id,
agent_profile_id: agentProfileId,
executor_id: executorId,
executor_profile_id: executorProfileId || undefined,
prompt: trimmedDescription || "",
plan_mode: true,
},
15000,
);
const { request } = buildStartRequest(updatedTask.id, agentProfileId, {
executorId,
executorProfileId: executorProfileId || undefined,
prompt: trimmedDescription || "",
planMode: true,
});
const response = await launchSession(request);
const newSessionId = response?.session_id ?? null;
onSuccess?.(updatedTask, "edit", { taskSessionId: newSessionId });
onOpenChange(false);
apps/web/components/task/chat/message-renderer.tsx
+4 -4
View File
@@ -5,7 +5,8 @@ import { IconPlayerPlay } from "@tabler/icons-react";
import { Button } from "@kandev/ui/button";
import type { Message, TaskSessionState, TaskState } from "@/lib/types/http";
import type { ToolCallMetadata } from "@/components/task/chat/types";
import { getWebSocketClient } from "@/lib/ws/connection";
import { launchSession } from "@/lib/services/session-launch-service";
import { buildStartCreatedRequest } from "@/lib/services/session-launch-helpers";
import { ChatMessage } from "@/components/task/chat/messages/chat-message";
import { PermissionRequestMessage } from "@/components/task/chat/messages/permission-request-message";
import { StatusMessage } from "@/components/task/chat/messages/status-message";
@@ -37,11 +38,10 @@ type AdapterContext = {
function TaskDescriptionStartButton({ taskId, sessionId }: { taskId: string; sessionId: string }) {
const [isStarting, setIsStarting] = useState(false);
const handleStart = useCallback(async () => {
const client = getWebSocketClient();
if (!client) return;
setIsStarting(true);
try {
await client.request("orchestrator.start", { task_id: taskId, session_id: sessionId }, 15000);
const { request } = buildStartCreatedRequest(taskId, sessionId);
await launchSession(request);
} catch (error) {
console.error("Failed to start agent:", error);
} finally {
apps/web/components/task/mobile/session-task-switcher-sheet.tsx
+23 -35
View File
@@ -10,15 +10,17 @@ import { TaskCreateDialog } from "@/components/task-create-dialog";
import { useAppStore, useAppStoreApi } from "@/components/state-provider";
import { replaceSessionUrl } from "@/lib/links";
import { fetchWorkflowSnapshot, listWorkflows } from "@/lib/api";
import { launchSession } from "@/lib/services/session-launch-service";
import { buildPrepareRequest } from "@/lib/services/session-launch-helpers";
import { useTasks } from "@/hooks/use-tasks";
import { useTaskActions } from "@/hooks/use-task-actions";
import { useTaskRemoval } from "@/hooks/use-task-removal";
import { getSessionInfoForTask } from "@/lib/utils/session-info";
import type {
TaskState,
TaskSessionState,
Workspace,
Repository,
TaskSession,
Task,
WorkflowSnapshot,
} from "@/lib/types/http";
@@ -90,40 +92,13 @@ function useSheetData(workspaceId: string | null, workflowId: string | null) {
return activeTaskId;
}, [activeSessionId, activeTaskId, sessionsById]);
const getSessionInfoForTask = useCallback(
(taskId: string) => {
const sessions = sessionsByTaskId[taskId] ?? [];
if (sessions.length === 0) {
return { diffStats: undefined, updatedAt: undefined, sessionState: undefined };
}
const primarySession = sessions.find((s: TaskSession) => s.is_primary);
const latestSession = primarySession ?? sessions[0];
if (!latestSession) {
return { diffStats: undefined, updatedAt: undefined, sessionState: undefined };
}
const updatedAt = latestSession.updated_at;
const sessionState = latestSession.state as TaskSessionState | undefined;
const gitStatus = gitStatusBySessionId[latestSession.id];
if (!gitStatus?.files) return { diffStats: undefined, updatedAt, sessionState };
let additions = 0;
let deletions = 0;
for (const file of Object.values(gitStatus.files)) {
additions += file.additions ?? 0;
deletions += file.deletions ?? 0;
}
const diffStats = additions === 0 && deletions === 0 ? undefined : { additions, deletions };
return { diffStats, updatedAt, sessionState };
},
[sessionsByTaskId, gitStatusBySessionId],
);
const tasksWithRepositories = useMemo(() => {
const repositories = workspaceId ? (repositoriesByWorkspace[workspaceId] ?? []) : [];
const repositoryPathsById = new Map(
repositories.map((repo: Repository) => [repo.id, repo.local_path]),
);
return tasks.map((task: KanbanState["tasks"][number]) => {
const sessionInfo = getSessionInfoForTask(task.id);
const sessionInfo = getSessionInfoForTask(task.id, sessionsByTaskId, gitStatusBySessionId);
return {
id: task.id,
title: task.title,
@@ -141,7 +116,7 @@ function useSheetData(workspaceId: string | null, workflowId: string | null) {
primarySessionId: task.primarySessionId ?? null,
};
});
}, [repositoriesByWorkspace, tasks, workspaceId, getSessionInfoForTask]);
}, [repositoriesByWorkspace, tasks, workspaceId, sessionsByTaskId, gitStatusBySessionId]);
const dialogSteps = useMemo(
() =>
@@ -299,15 +274,28 @@ function useSheetActions(workspaceId: string | null, onOpenChange: (open: boolea
onOpenChange(false);
return;
}
loadTaskSessionsForTask(taskId).then((sessions) => {
loadTaskSessionsForTask(taskId).then(async (sessions) => {
const sessionId = sessions[0]?.id ?? null;
if (!sessionId) {
setActiveTask(taskId);
if (sessionId) {
setActiveSession(taskId, sessionId);
replaceSessionUrl(sessionId);
onOpenChange(false);
return;
}
setActiveSession(taskId, sessionId);
replaceSessionUrl(sessionId);
// No session — prepare workspace
const { request } = buildPrepareRequest(taskId);
try {
const resp = await launchSession(request);
if (resp.session_id) {
setActiveSession(taskId, resp.session_id);
replaceSessionUrl(resp.session_id);
onOpenChange(false);
return;
}
} catch {
// Fall through to default behavior
}
setActiveTask(taskId);
onOpenChange(false);
});
},
apps/web/components/task/task-center-panel.tsx
+2 -20
View File
@@ -21,8 +21,8 @@ import { PassthroughTerminal } from "./passthrough-terminal";
import type { OpenFileTab, FileContentResponse } from "@/lib/types/backend";
import { useAppStore } from "@/components/state-provider";
import { SessionTabs, type SessionTab } from "@/components/session-tabs";
import { approveSessionAction } from "@/app/actions/workspaces";
import { getWebSocketClient } from "@/lib/ws/connection";
import { executeApprove } from "@/lib/services/session-approve";
import { requestFileContent } from "@/lib/ws/workspace-files";
import {
setOpenFileTabs as saveOpenFileTabs,
@@ -64,25 +64,7 @@ function useSessionApprove(activeSessionId: string | null, activeTaskId: string
const handleApprove = useCallback(async () => {
if (!activeSessionId || !activeTaskId) return;
try {
const response = await approveSessionAction(activeSessionId);
if (response?.session) setTaskSession(response.session);
if (
response?.workflow_step?.events?.on_enter?.some(
(a: { type: string }) => a.type === "auto_start_agent",
)
) {
const client = getWebSocketClient();
if (client)
client.send({
type: "request",
action: "orchestrator.start",
payload: {
task_id: activeTaskId,
session_id: activeSessionId,
workflow_step_id: response.workflow_step.id,
},
});
}
await executeApprove(activeSessionId, activeTaskId, setTaskSession);
} catch (error) {
console.error("Failed to approve session:", error);
}
apps/web/components/task/task-session-sidebar.tsx
+67 -47
View File
@@ -1,7 +1,7 @@
"use client";
import { useCallback, useMemo, useState, memo } from "react";
import type { TaskState, TaskSessionState, Repository, TaskSession, Task } from "@/lib/types/http";
import type { TaskState, TaskSessionState, Repository, Task } from "@/lib/types/http";
import type { KanbanState } from "@/lib/state/slices";
import { TaskSwitcher } from "./task-switcher";
import { TaskRenameDialog } from "./task-rename-dialog";
@@ -14,7 +14,11 @@ import { replaceSessionUrl } from "@/lib/links";
import { useAllWorkflowSnapshots } from "@/hooks/domains/kanban/use-all-workflow-snapshots";
import { useTaskActions } from "@/hooks/use-task-actions";
import { useTaskRemoval } from "@/hooks/use-task-removal";
import { performLayoutSwitch } from "@/lib/state/dockview-store";
import { performLayoutSwitch, useDockviewStore } from "@/lib/state/dockview-store";
import { INTENT_PR_REVIEW } from "@/lib/state/layout-manager";
import { launchSession } from "@/lib/services/session-launch-service";
import { buildPrepareRequest } from "@/lib/services/session-launch-helpers";
import { getSessionInfoForTask } from "@/lib/utils/session-info";
import { useArchivedTaskState } from "./task-archived-context";
/** Find a task across all workflow snapshots */
@@ -86,45 +90,6 @@ type TaskSessionSidebarProps = {
workflowId: string | null;
};
type SessionInfo = {
diffStats: { additions: number; deletions: number } | undefined;
updatedAt: string | undefined;
sessionState: TaskSessionState | undefined;
};
type GitStatusMap = Record<
string,
{ files?: Record<string, { additions?: number; deletions?: number }> }
>;
function getSessionInfoForTask(
taskId: string,
sessionsByTaskId: Record<string, TaskSession[]>,
gitStatusBySessionId: GitStatusMap,
): SessionInfo {
const sessions = sessionsByTaskId[taskId] ?? [];
if (sessions.length === 0) {
return { diffStats: undefined, updatedAt: undefined, sessionState: undefined };
}
const primarySession = sessions.find((s: TaskSession) => s.is_primary);
const latestSession = primarySession ?? sessions[0];
if (!latestSession) {
return { diffStats: undefined, updatedAt: undefined, sessionState: undefined };
}
const updatedAt = latestSession.updated_at;
const sessionState = latestSession.state as TaskSessionState | undefined;
const gitStatus = gitStatusBySessionId[latestSession.id];
if (!gitStatus?.files) return { diffStats: undefined, updatedAt, sessionState };
let additions = 0;
let deletions = 0;
for (const file of Object.values(gitStatus.files)) {
additions += file.additions ?? 0;
deletions += file.deletions ?? 0;
}
const diffStats = additions === 0 && deletions === 0 ? undefined : { additions, deletions };
return { diffStats, updatedAt, sessionState };
}
function useSidebarData(workspaceId: string | null) {
const activeTaskId = useAppStore((state) => state.tasks.activeTaskId);
const activeSessionId = useAppStore((state) => state.tasks.activeSessionId);
@@ -219,10 +184,46 @@ function useSidebarData(workspaceId: string | null) {
return { activeTaskId, selectedTaskId, allSteps, isLoadingWorkflow, tasksWithRepositories };
}
function useSidebarActions(store: ReturnType<typeof useAppStoreApi>) {
type SwitchFn = (
taskId: string,
sessionId: string,
oldSessionId: string | null | undefined,
) => void;
type StoreApi = ReturnType<typeof useAppStoreApi>;
async function prepareAndSwitchTask(
taskId: string,
store: StoreApi,
switchToSession: SwitchFn,
setPreparingTaskId: (id: string | null) => void,
): Promise<boolean> {
setPreparingTaskId(taskId);
try {
const { request } = buildPrepareRequest(taskId);
const resp = await launchSession(request);
if (resp.session_id) {
const oldSessionId = store.getState().tasks.activeSessionId;
switchToSession(taskId, resp.session_id, oldSessionId);
// Apply PR review layout if the task has PR metadata
if (store.getState().taskPRs.byTaskId[taskId]) {
const { api, buildDefaultLayout } = useDockviewStore.getState();
if (api) buildDefaultLayout(api, INTENT_PR_REVIEW);
}
return true;
}
return false;
} catch {
return false;
} finally {
setPreparingTaskId(null);
}
}
function useSidebarActions(store: StoreApi) {
const setActiveTask = useAppStore((state) => state.setActiveTask);
const setActiveSession = useAppStore((state) => state.setActiveSession);
const [deletingTaskId, setDeletingTaskId] = useState<string | null>(null);
const [preparingTaskId, setPreparingTaskId] = useState<string | null>(null);
const { deleteTaskById, archiveTaskById, renameTaskById } = useTaskActions();
const { removeTaskFromBoard, loadTaskSessionsForTask } = useTaskRemoval({
store,
@@ -247,14 +248,21 @@ function useSidebarActions(store: ReturnType<typeof useAppStoreApi>) {
loadTaskSessionsForTask(taskId);
return;
}
loadTaskSessionsForTask(taskId).then((sessions) => {
loadTaskSessionsForTask(taskId).then(async (sessions) => {
const currentOldSessionId = store.getState().tasks.activeSessionId;
const sessionId = sessions[0]?.id ?? null;
if (!sessionId) {
setActiveTask(taskId);
if (sessionId) {
switchToSession(taskId, sessionId, currentOldSessionId);
return;
}
switchToSession(taskId, sessionId, currentOldSessionId);
// No session — prepare workspace and switch to it
const switched = await prepareAndSwitchTask(
taskId,
store,
switchToSession,
setPreparingTaskId,
);
if (!switched) setActiveTask(taskId);
});
},
[loadTaskSessionsForTask, switchToSession, setActiveTask, store],
@@ -306,6 +314,7 @@ function useSidebarActions(store: ReturnType<typeof useAppStoreApi>) {
return {
deletingTaskId,
preparingTaskId,
handleSelectTask,
handleArchiveTask,
handleDeleteTask,
@@ -326,6 +335,7 @@ export const TaskSessionSidebar = memo(function TaskSessionSidebar({
useSidebarData(workspaceId);
const {
deletingTaskId,
preparingTaskId,
handleSelectTask,
handleArchiveTask,
handleDeleteTask,
@@ -335,11 +345,21 @@ export const TaskSessionSidebar = memo(function TaskSessionSidebar({
handleRenameSubmit,
} = useSidebarActions(store);
const displayTasks = useMemo(
() =>
preparingTaskId
? tasksWithRepositories.map((t) =>
t.id === preparingTaskId ? { ...t, sessionState: "STARTING" as TaskSessionState } : t,
)
: tasksWithRepositories,
[tasksWithRepositories, preparingTaskId],
);
return (
<PanelRoot>
<PanelBody className="space-y-4 p-0">
<TaskSwitcher
tasks={tasksWithRepositories}
tasks={displayTasks}
steps={allSteps.map((step) => ({ id: step.id, title: step.title, color: step.color }))}
activeTaskId={activeTaskId}
selectedTaskId={selectedTaskId}
apps/web/hooks/domains/kanban/use-kanban-navigation.ts
+13 -21
View File
@@ -5,6 +5,8 @@ import { useRouter } from "next/navigation";
import { getWebSocketClient } from "@/lib/ws/connection";
import { linkToSession } from "@/lib/links";
import { INTENT_PR_REVIEW } from "@/lib/state/layout-manager";
import { launchSession } from "@/lib/services/session-launch-service";
import { buildPrepareRequest } from "@/lib/services/session-launch-helpers";
import { useAppStore } from "@/components/state-provider";
import type { Task } from "@/components/kanban-card";
@@ -38,22 +40,6 @@ async function fetchLatestSession(
}
}
async function preparePRSession(taskId: string): Promise<string | null> {
const client = getWebSocketClient();
if (!client) return null;
try {
const response = await client.request<{ session_id: string }>(
"task.session.prepare",
{ task_id: taskId },
15000,
);
return response.session_id ?? null;
} catch (error) {
console.error("Failed to prepare session for PR task:", error);
return null;
}
}
export function useKanbanNavigation({
enablePreviewOnClick,
isMobile,
@@ -81,17 +67,23 @@ export function useKanbanNavigation({
const handleNoSession = useCallback(
async (task: Task) => {
if (taskPRs[task.id]) {
const sessionId = await preparePRSession(task.id);
if (sessionId) {
router.push(linkToSession(sessionId, INTENT_PR_REVIEW));
try {
const { request } = buildPrepareRequest(task.id);
const resp = await launchSession(request);
if (resp.session_id) {
// Apply PR review layout if the task has PR metadata
const layoutIntent = taskPRs[task.id] ? INTENT_PR_REVIEW : undefined;
if (onOpenTask) onOpenTask(task, resp.session_id);
else router.push(linkToSession(resp.session_id, layoutIntent));
return;
}
} catch {
// Fall through to dialog as last resort
}
setEditingTask(task);
setIsDialogOpen(true);
},
[taskPRs, router, setEditingTask, setIsDialogOpen],
[taskPRs, router, onOpenTask, setEditingTask, setIsDialogOpen],
);
const handleOpenTask = useCallback(
apps/web/hooks/domains/session/use-session-agent.ts
+6 -13
View File
@@ -1,5 +1,7 @@
import { useCallback, useMemo, useState } from "react";
import { getWebSocketClient } from "@/lib/ws/connection";
import { launchSession } from "@/lib/services/session-launch-service";
import { buildStartRequest } from "@/lib/services/session-launch-helpers";
import { useAppStore } from "@/components/state-provider";
import type { TaskSessionState, Task, TaskSession } from "@/lib/types/http";
@@ -49,21 +51,12 @@ export function useSessionAgent(task: Task | null): UseSessionAgentReturn {
if (!task?.id) return;
if (!agentProfileId) return;
const client = getWebSocketClient();
if (!client) return;
setIsAgentLoading(true);
try {
// Store will be updated via WebSocket notifications when session starts
await client.request(
"orchestrator.start",
{
task_id: task.id,
agent_profile_id: agentProfileId,
prompt: prompt ?? task.description ?? "",
},
15000,
);
const { request } = buildStartRequest(task.id, agentProfileId, {
prompt: prompt ?? task.description ?? "",
});
await launchSession(request);
} catch {
// Failed to start agent
} finally {
apps/web/hooks/domains/session/use-session-launch.ts
+40
View File
@@ -0,0 +1,40 @@
import { useCallback, useState } from "react";
import {
launchSession,
type LaunchSessionRequest,
type LaunchSessionResponse,
} from "@/lib/services/session-launch-service";
export function useSessionLaunch(options?: {
onSuccess?: (resp: LaunchSessionResponse) => void;
onError?: (err: Error) => void;
}): {
launch: (request: LaunchSessionRequest) => Promise<LaunchSessionResponse | null>;
isLoading: boolean;
error: string | null;
} {
const [isLoading, setIsLoading] = useState(false);
const [error, setError] = useState<string | null>(null);
const launch = useCallback(
async (request: LaunchSessionRequest): Promise<LaunchSessionResponse | null> => {
setIsLoading(true);
setError(null);
try {
const resp = await launchSession(request);
options?.onSuccess?.(resp);
return resp;
} catch (err) {
const message = err instanceof Error ? err.message : "Unknown error";
setError(message);
options?.onError?.(err instanceof Error ? err : new Error(message));
return null;
} finally {
setIsLoading(false);
}
},
[options?.onSuccess, options?.onError], // eslint-disable-line react-hooks/exhaustive-deps
);
return { launch, isLoading, error };
}
apps/web/hooks/domains/session/use-session-resumption.ts
+13 -13
View File
@@ -1,5 +1,7 @@
import { useCallback, useEffect, useState, useRef } from "react";
import { getWebSocketClient } from "@/lib/ws/connection";
import { launchSession } from "@/lib/services/session-launch-service";
import { buildResumeRequest } from "@/lib/services/session-launch-helpers";
import { useAppStore } from "@/components/state-provider";
import type { TaskSessionState } from "@/lib/types/http";
@@ -129,11 +131,14 @@ async function checkAndResume({
}
if (status.needs_resume && status.is_resumable) {
setters.setResumptionState("resuming");
const resumeResp = await client.request<ResumeResponse>(
"task.session.resume",
{ task_id: taskId, session_id: sessionId },
30000,
);
const { request } = buildResumeRequest(taskId, sessionId);
const launchResp = await launchSession(request);
const resumeResp: ResumeResponse = {
success: launchResp.success,
state: launchResp.state,
worktree_path: launchResp.worktree_path,
worktree_branch: launchResp.worktree_branch,
};
applyResumeResponse(resumeResp, taskId, sessionId, session, setters);
} else {
setters.setResumptionState("idle");
@@ -226,16 +231,11 @@ export function useSessionResumption(
// Manual resume function
const resumeSession = useCallback(async (): Promise<boolean> => {
if (!taskId || !sessionId) return false;
const client = getWebSocketClient();
if (!client) return false;
setResumptionState("resuming");
setError(null);
try {
const response = await client.request<ResumeResponse>(
"task.session.resume",
{ task_id: taskId, session_id: sessionId },
30000,
);
const { request } = buildResumeRequest(taskId, sessionId);
const response = await launchSession(request);
if (response.success) {
setResumptionState("resumed");
if (response.state) {
@@ -252,7 +252,7 @@ export function useSessionResumption(
return true;
}
setResumptionState("error");
setError(response.error ?? "Failed to resume session");
setError("Failed to resume session");
return false;
} catch (err) {
setResumptionState("error");
apps/web/hooks/use-session-layout-state.ts
+1 -33
View File
@@ -5,8 +5,7 @@ import { useAppStore } from "@/components/state-provider";
import { useSessionGitStatus } from "@/hooks/domains/session/use-session-git-status";
import { useSessionCommits } from "@/hooks/domains/session/use-session-commits";
import { getPlanLastSeen } from "@/lib/local-storage";
import { approveSessionAction } from "@/app/actions/workspaces";
import { getWebSocketClient } from "@/lib/ws/connection";
import { executeApprove } from "@/lib/services/session-approve";
import type { OpenFileTab } from "@/lib/types/backend";
import type { MobileSessionPanel } from "@/lib/state/slices/ui/types";
@@ -43,37 +42,6 @@ function useOpenFileRequestState() {
return { openFileRequest, handleOpenFile, handleFileOpenHandled };
}
/** Handle approve action: call server action and trigger auto-start if configured. */
async function executeApprove(
sessionId: string,
taskId: string,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
setTaskSession: (session: any) => void,
) {
const response = await approveSessionAction(sessionId);
if (response?.session) {
setTaskSession(response.session);
}
if (
response?.workflow_step?.events?.on_enter?.some(
(a: { type: string }) => a.type === "auto_start_agent",
)
) {
const client = getWebSocketClient();
if (client) {
client.send({
type: "request",
action: "orchestrator.start",
payload: {
task_id: taskId,
session_id: sessionId,
workflow_step_id: response.workflow_step.id,
},
});
}
}
}
/**
* Shared hook for session layout state used across mobile, tablet, and desktop layouts.
* Consolidates common state and logic to avoid duplication.
apps/web/lib/api/domains/session-api.ts
+10
View File
@@ -57,3 +57,13 @@ export async function openSessionFolder(sessionId: string, options?: ApiRequestO
init: { method: "POST", ...(options?.init ?? {}) },
});
}
export { launchSession, type LaunchSessionResponse } from "@/lib/services/session-launch-service";
export {
buildPRPrepareRequest,
buildPrepareRequest,
buildStartRequest,
buildStartCreatedRequest,
buildResumeRequest,
buildWorkflowStepRequest,
} from "@/lib/services/session-launch-helpers";
apps/web/lib/services/session-approve.ts
+26
View File
@@ -0,0 +1,26 @@
import { approveSessionAction } from "@/app/actions/workspaces";
import { launchSession } from "@/lib/services/session-launch-service";
import { buildWorkflowStepRequest } from "@/lib/services/session-launch-helpers";
/** Handle approve action: call server action and trigger auto-start if configured. */
export async function executeApprove(
sessionId: string,
taskId: string,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
setTaskSession: (session: any) => void,
) {
const response = await approveSessionAction(sessionId);
if (response?.session) {
setTaskSession(response.session);
}
if (
response?.workflow_step?.events?.on_enter?.some(
(a: { type: string }) => a.type === "auto_start_agent",
)
) {
const { request } = buildWorkflowStepRequest(taskId, sessionId, response.workflow_step.id);
launchSession(request).catch((err) =>
console.error("Failed to auto-start workflow step:", err),
);
}
}
apps/web/lib/services/session-launch-helpers.ts
+107
View File
@@ -0,0 +1,107 @@
import type { LaunchSessionRequest } from "./session-launch-service";
export type LayoutIntentHint = "default" | "plan" | "pr-review" | "keep";
type BuildResult = {
request: LaunchSessionRequest;
layout: LayoutIntentHint;
};
export function buildStartRequest(
taskId: string,
agentProfileId: string,
opts?: {
executorId?: string;
executorProfileId?: string;
prompt?: string;
planMode?: boolean;
priority?: number;
},
): BuildResult {
return {
request: {
task_id: taskId,
intent: "start",
agent_profile_id: agentProfileId,
executor_id: opts?.executorId,
executor_profile_id: opts?.executorProfileId,
prompt: opts?.prompt,
plan_mode: opts?.planMode,
priority: opts?.priority,
},
layout: opts?.planMode ? "plan" : "default",
};
}
export function buildPrepareRequest(
taskId: string,
opts?: { agentProfileId?: string; executorId?: string; executorProfileId?: string },
): BuildResult {
return {
request: {
task_id: taskId,
intent: "prepare",
agent_profile_id: opts?.agentProfileId,
executor_id: opts?.executorId,
executor_profile_id: opts?.executorProfileId,
launch_workspace: true,
},
layout: "default",
};
}
export function buildPRPrepareRequest(taskId: string): BuildResult {
return {
request: {
task_id: taskId,
intent: "prepare",
launch_workspace: true,
},
layout: "pr-review",
};
}
export function buildStartCreatedRequest(
taskId: string,
sessionId: string,
opts?: { agentProfileId?: string; prompt?: string; skipMessageRecord?: boolean },
): BuildResult {
return {
request: {
task_id: taskId,
intent: "start_created",
session_id: sessionId,
agent_profile_id: opts?.agentProfileId,
prompt: opts?.prompt,
skip_message_record: opts?.skipMessageRecord,
},
layout: "keep",
};
}
export function buildResumeRequest(taskId: string, sessionId: string): BuildResult {
return {
request: {
task_id: taskId,
intent: "resume",
session_id: sessionId,
},
layout: "keep",
};
}
export function buildWorkflowStepRequest(
taskId: string,
sessionId: string,
stepId: string,
): BuildResult {
return {
request: {
task_id: taskId,
intent: "workflow_step",
session_id: sessionId,
workflow_step_id: stepId,
},
layout: "keep",
};
}
apps/web/lib/services/session-launch-service.ts
+38
View File
@@ -0,0 +1,38 @@
import { getWebSocketClient } from "@/lib/ws/connection";
export type SessionIntent = "prepare" | "start" | "start_created" | "resume" | "workflow_step";
export type LaunchSessionRequest = {
task_id: string;
intent?: SessionIntent;
session_id?: string;
agent_profile_id?: string;
executor_id?: string;
executor_profile_id?: string;
prompt?: string;
plan_mode?: boolean;
workflow_step_id?: string;
priority?: number;
launch_workspace?: boolean;
skip_message_record?: boolean;
};
export type LaunchSessionResponse = {
success: boolean;
task_id: string;
session_id?: string;
agent_execution_id?: string;
state: string;
worktree_path?: string;
worktree_branch?: string;
};
export async function launchSession(
request: LaunchSessionRequest,
timeout?: number,
): Promise<LaunchSessionResponse> {
const client = getWebSocketClient();
if (!client) throw new Error("WebSocket client not available");
const effectiveTimeout = timeout ?? (request.intent === "resume" ? 30_000 : 15_000);
return client.request<LaunchSessionResponse>("session.launch", request, effectiveTimeout);
}
apps/web/lib/ui/state-icons.tsx
+6 -5
View File
@@ -8,12 +8,13 @@ type IconConfig = {
className: string;
};
const STYLE_MUTED = "text-muted-foreground";
const STYLE_LOADING = "text-blue-500 animate-spin";
const STYLE_WARNING = "text-yellow-500";
const STYLE_ERROR = "text-red-500";
const TASK_STATE_ICONS: Record<TaskState, IconConfig> = {
CREATED: { Icon: IconLoader2, className: STYLE_LOADING },
CREATED: { Icon: IconAlertCircle, className: STYLE_MUTED },
SCHEDULING: { Icon: IconLoader2, className: STYLE_LOADING },
IN_PROGRESS: { Icon: IconLoader2, className: STYLE_LOADING },
REVIEW: { Icon: IconCheck, className: STYLE_WARNING },
@@ -22,11 +23,11 @@ const TASK_STATE_ICONS: Record<TaskState, IconConfig> = {
COMPLETED: { Icon: IconCheck, className: "text-green-500" },
FAILED: { Icon: IconX, className: STYLE_ERROR },
CANCELLED: { Icon: IconX, className: STYLE_ERROR },
TODO: { Icon: IconAlertCircle, className: "text-muted-foreground" },
TODO: { Icon: IconAlertCircle, className: STYLE_MUTED },
};
const SESSION_STATE_ICONS: Record<TaskSessionState, IconConfig> = {
CREATED: { Icon: IconLoader2, className: STYLE_LOADING },
CREATED: { Icon: IconAlertCircle, className: STYLE_MUTED },
STARTING: { Icon: IconLoader2, className: STYLE_LOADING },
RUNNING: { Icon: IconLoader2, className: STYLE_LOADING },
WAITING_FOR_INPUT: { Icon: IconCheck, className: STYLE_WARNING },
@@ -37,12 +38,12 @@ const SESSION_STATE_ICONS: Record<TaskSessionState, IconConfig> = {
const DEFAULT_TASK_ICON: IconConfig = {
Icon: IconAlertCircle,
className: "text-muted-foreground",
className: STYLE_MUTED,
};
const DEFAULT_SESSION_ICON: IconConfig = {
Icon: IconAlertCircle,
className: "text-muted-foreground",
className: STYLE_MUTED,
};
export function getTaskStateIcon(state?: TaskState, className?: string) {
apps/web/lib/utils/session-info.ts
+40
View File
@@ -0,0 +1,40 @@
import type { TaskSession, TaskSessionState } from "@/lib/types/http";
export type SessionInfo = {
diffStats: { additions: number; deletions: number } | undefined;
updatedAt: string | undefined;
sessionState: TaskSessionState | undefined;
};
type GitStatusMap = Record<
string,
{ files?: Record<string, { additions?: number; deletions?: number }> }
>;
export function getSessionInfoForTask(
taskId: string,
sessionsByTaskId: Record<string, TaskSession[]>,
gitStatusBySessionId: GitStatusMap,
): SessionInfo {
const sessions = sessionsByTaskId[taskId] ?? [];
if (sessions.length === 0) {
return { diffStats: undefined, updatedAt: undefined, sessionState: undefined };
}
const primarySession = sessions.find((s: TaskSession) => s.is_primary);
const latestSession = primarySession ?? sessions[0];
if (!latestSession) {
return { diffStats: undefined, updatedAt: undefined, sessionState: undefined };
}
const updatedAt = latestSession.updated_at;
const sessionState = latestSession.state as TaskSessionState | undefined;
const gitStatus = gitStatusBySessionId[latestSession.id];
if (!gitStatus?.files) return { diffStats: undefined, updatedAt, sessionState };
let additions = 0;
let deletions = 0;
for (const file of Object.values(gitStatus.files)) {
additions += file.additions ?? 0;
deletions += file.deletions ?? 0;
}
const diffStats = additions === 0 && deletions === 0 ? undefined : { additions, deletions };
return { diffStats, updatedAt, sessionState };
}