feat: improve task creation (#220)

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