From b1433a8d4506649c0cde3014276dd49cd612b92c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Thu, 22 Jan 2026 02:39:45 +0000 Subject: [PATCH] Clean up dead code and fix ACP session event payload (#73) --- apps/backend/cmd/kandev/adapters.go | 30 - apps/backend/cmd/kandev/main.go | 64 - apps/backend/cmd/kandev/orchestrator.go | 15 +- apps/backend/cmd/kandev/worktree.go | 2 +- .../internal/agent/lifecycle/event_types.go | 1 + .../internal/agent/lifecycle/events.go | 1 + .../agent/lifecycle/execution_store.go | 10 - .../internal/agent/lifecycle/manager.go | 123 +- .../internal/agent/lifecycle/manager_test.go | 63 - .../backend/internal/agent/streaming/.gitkeep | 0 .../backend/internal/common/database/.gitkeep | 0 apps/backend/internal/common/errors/.gitkeep | 0 .../internal/gateway/websocket/client.go | 58 +- .../backend/internal/gateway/websocket/hub.go | 19 - .../internal/integration/integration_test.go | 2 +- .../internal/integration/orchestrator_test.go | 222 ++-- .../internal/orchestrator/acp/.gitkeep | 0 .../orchestrator/controller/controller.go | 22 - apps/backend/internal/orchestrator/dto/dto.go | 17 - .../internal/orchestrator/event_handlers.go | 568 +++++++++ .../orchestrator/executor/executor.go | 242 ---- .../orchestrator/handlers/handlers.go | 25 - .../internal/orchestrator/queue/queue.go | 45 - .../internal/orchestrator/queue/queue_test.go | 88 +- .../orchestrator/scheduler/scheduler_test.go | 35 +- apps/backend/internal/orchestrator/service.go | 1065 +---------------- .../internal/orchestrator/task_operations.go | 423 +++++++ .../internal/orchestrator/watcher/watcher.go | 1 + .../internal/task/repository/sqlite_test.go | 2 +- apps/backend/internal/task/service/service.go | 2 +- .../internal/task/service/service_test.go | 2 +- .../internal/{agent => }/worktree/config.go | 0 .../internal/{agent => }/worktree/errors.go | 0 .../internal/{agent => }/worktree/manager.go | 0 .../{agent => }/worktree/manager_test.go | 0 .../internal/{agent => }/worktree/provider.go | 0 .../internal/{agent => }/worktree/store.go | 0 .../internal/{agent => }/worktree/worktree.go | 0 apps/backend/pkg/websocket/actions.go | 2 +- apps/web/hooks/use-session-agent.ts | 116 +- apps/web/hooks/use-task-chat-session.ts | 81 +- 41 files changed, 1129 insertions(+), 2217 deletions(-) delete mode 100644 apps/backend/internal/agent/streaming/.gitkeep delete mode 100644 apps/backend/internal/common/database/.gitkeep delete mode 100644 apps/backend/internal/common/errors/.gitkeep delete mode 100644 apps/backend/internal/orchestrator/acp/.gitkeep create mode 100644 apps/backend/internal/orchestrator/event_handlers.go create mode 100644 apps/backend/internal/orchestrator/task_operations.go rename apps/backend/internal/{agent => }/worktree/config.go (100%) rename apps/backend/internal/{agent => }/worktree/errors.go (100%) rename apps/backend/internal/{agent => }/worktree/manager.go (100%) rename apps/backend/internal/{agent => }/worktree/manager_test.go (100%) rename apps/backend/internal/{agent => }/worktree/provider.go (100%) rename apps/backend/internal/{agent => }/worktree/store.go (100%) rename apps/backend/internal/{agent => }/worktree/worktree.go (100%) diff --git a/apps/backend/cmd/kandev/adapters.go b/apps/backend/cmd/kandev/adapters.go index b575c112b2..f6283f304c 100644 --- a/apps/backend/cmd/kandev/adapters.go +++ b/apps/backend/cmd/kandev/adapters.go @@ -184,42 +184,12 @@ func (a *lifecycleAdapter) RespondToPermissionBySessionID(ctx context.Context, s return a.mgr.RespondToPermissionBySessionID(sessionID, pendingID, optionID, cancelled) } -// GetRecoveredExecutions returns executions recovered from Docker during startup -func (a *lifecycleAdapter) GetRecoveredExecutions() []executor.RecoveredExecutionInfo { - recovered := a.mgr.GetRecoveredExecutions() - result := make([]executor.RecoveredExecutionInfo, len(recovered)) - for i, r := range recovered { - result[i] = executor.RecoveredExecutionInfo{ - ExecutionID: r.ExecutionID, - TaskID: r.TaskID, - SessionID: r.SessionID, - ContainerID: r.ContainerID, - AgentProfileID: r.AgentProfileID, - } - } - return result -} - // IsAgentRunningForSession checks if an agent is actually running for a session // This probes the actual agent (Docker container or standalone process) func (a *lifecycleAdapter) IsAgentRunningForSession(ctx context.Context, sessionID string) bool { return a.mgr.IsAgentRunningForSession(ctx, sessionID) } -// CleanupStaleExecutionBySessionID removes a stale agent execution from tracking without trying to stop it. -func (a *lifecycleAdapter) CleanupStaleExecutionBySessionID(ctx context.Context, sessionID string) error { - return a.mgr.CleanupStaleExecutionBySessionID(ctx, sessionID) -} - -// GetAgentType returns the agent type configuration -func (a *lifecycleAdapter) GetAgentType(ctx context.Context, agentID string) (*v1.AgentType, error) { - config, err := a.registry.Get(agentID) - if err != nil { - return nil, err - } - return config.ToAPIType(), nil -} - // ResolveAgentProfile resolves an agent profile ID to profile information func (a *lifecycleAdapter) ResolveAgentProfile(ctx context.Context, profileID string) (*executor.AgentProfileInfo, error) { info, err := a.mgr.ResolveAgentProfile(ctx, profileID) diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index a606825fad..43e4855129 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -44,7 +44,6 @@ import ( // Task Service packages taskcontroller "github.com/kandev/kandev/internal/task/controller" taskhandlers "github.com/kandev/kandev/internal/task/handlers" - "github.com/kandev/kandev/internal/task/models" usercontroller "github.com/kandev/kandev/internal/user/controller" userhandlers "github.com/kandev/kandev/internal/user/handlers" @@ -312,69 +311,6 @@ func main() { } } - // Set up historical logs provider for task subscriptions - // Uses messages instead of execution logs - all agent messages are now messages - gateway.Hub.SetHistoricalLogsProvider(func(ctx context.Context, taskID string) ([]*ws.Message, error) { - var session *models.TaskSession - session, err = taskRepo.GetActiveTaskSessionByTaskID(ctx, taskID) - if err != nil { - session, err = taskRepo.GetTaskSessionByTaskID(ctx, taskID) - if err != nil { - return nil, nil - } - } - - messages, err := taskSvc.ListMessages(ctx, session.ID) - if err != nil { - return nil, err - } - - result := make([]*ws.Message, 0, len(messages)) - for _, message := range messages { - action := ws.ActionSessionMessageAdded - payload := map[string]interface{}{ - "message_id": message.ID, - "session_id": message.TaskSessionID, - "task_id": message.TaskID, - "author_type": string(message.AuthorType), - "author_id": message.AuthorID, - "content": message.Content, - "type": string(message.Type), - "requests_input": message.RequestsInput, - "created_at": message.CreatedAt.Format(time.RFC3339), - } - if message.Metadata != nil { - payload["metadata"] = message.Metadata - } - notification, err := ws.NewNotification(action, payload) - if err != nil { - continue - } - result = append(result, notification) - } - - // Also send current git status if available - session, err = taskRepo.GetActiveTaskSessionByTaskID(ctx, taskID) - if err == nil && session != nil && session.Metadata != nil { - if gitStatus, ok := session.Metadata["git_status"]; ok { - // Add task_id to the git status data - gitStatusData, ok := gitStatus.(map[string]interface{}) - if !ok { - gitStatusData = map[string]interface{}{"data": gitStatus} - } - gitStatusData["task_id"] = taskID - - gitStatusNotification, err := ws.NewNotification("session.git.status", gitStatusData) - if err == nil { - result = append(result, gitStatusNotification) - } - } - } - - return result, nil - }) - log.Info("Historical logs provider configured for task subscriptions (using messages and git status)") - // Set up session data provider for session subscriptions // Sends initial git status when a client subscribes to a session gateway.Hub.SetSessionDataProvider(func(ctx context.Context, sessionID string) ([]*ws.Message, error) { diff --git a/apps/backend/cmd/kandev/orchestrator.go b/apps/backend/cmd/kandev/orchestrator.go index d8d1996330..e2a88a9648 100644 --- a/apps/backend/cmd/kandev/orchestrator.go +++ b/apps/backend/cmd/kandev/orchestrator.go @@ -1,12 +1,13 @@ package main import ( + "errors" + "github.com/kandev/kandev/internal/agent/lifecycle" "github.com/kandev/kandev/internal/agent/registry" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/orchestrator" - "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/task/repository" taskservice "github.com/kandev/kandev/internal/task/service" userservice "github.com/kandev/kandev/internal/user/service" @@ -21,15 +22,13 @@ func provideOrchestrator( lifecycleMgr *lifecycle.Manager, agentRegistry *registry.Registry, ) (*orchestrator.Service, error) { - taskRepoAdapter := &taskRepositoryAdapter{repo: taskRepo, svc: taskSvc} - - var agentManagerClient executor.AgentManagerClient - if lifecycleMgr != nil { - agentManagerClient = newLifecycleAdapter(lifecycleMgr, agentRegistry, log) - } else { - agentManagerClient = executor.NewMockAgentManagerClient(log) + if lifecycleMgr == nil { + return nil, errors.New("lifecycle manager is required: configure agent runtime (docker or standalone)") } + taskRepoAdapter := &taskRepositoryAdapter{repo: taskRepo, svc: taskSvc} + agentManagerClient := newLifecycleAdapter(lifecycleMgr, agentRegistry, log) + serviceCfg := orchestrator.DefaultServiceConfig() orchestratorSvc := orchestrator.NewService(serviceCfg, eventBus, agentManagerClient, taskRepoAdapter, taskRepo, userSvc, log) taskSvc.SetExecutionStopper(orchestratorSvc) diff --git a/apps/backend/cmd/kandev/worktree.go b/apps/backend/cmd/kandev/worktree.go index 126de07e04..a42fc43848 100644 --- a/apps/backend/cmd/kandev/worktree.go +++ b/apps/backend/cmd/kandev/worktree.go @@ -4,7 +4,7 @@ import ( "database/sql" "github.com/kandev/kandev/internal/agent/lifecycle" - "github.com/kandev/kandev/internal/agent/worktree" + "github.com/kandev/kandev/internal/worktree" "github.com/kandev/kandev/internal/common/config" "github.com/kandev/kandev/internal/common/logger" taskservice "github.com/kandev/kandev/internal/task/service" diff --git a/apps/backend/internal/agent/lifecycle/event_types.go b/apps/backend/internal/agent/lifecycle/event_types.go index b9e3f04f75..0e832a1663 100644 --- a/apps/backend/internal/agent/lifecycle/event_types.go +++ b/apps/backend/internal/agent/lifecycle/event_types.go @@ -28,6 +28,7 @@ type AgentctlEventPayload struct { // ACPSessionCreatedPayload is the payload when an ACP session is created. type ACPSessionCreatedPayload struct { TaskID string `json:"task_id"` + SessionID string `json:"session_id"` AgentInstanceID string `json:"agent_instance_id"` ACPSessionID string `json:"acp_session_id"` } diff --git a/apps/backend/internal/agent/lifecycle/events.go b/apps/backend/internal/agent/lifecycle/events.go index cb7f31816e..0ba2929f11 100644 --- a/apps/backend/internal/agent/lifecycle/events.go +++ b/apps/backend/internal/agent/lifecycle/events.go @@ -91,6 +91,7 @@ func (p *EventPublisher) PublishACPSessionCreated(execution *AgentExecution, ses payload := ACPSessionCreatedPayload{ TaskID: execution.TaskID, + SessionID: execution.SessionID, AgentInstanceID: execution.ID, ACPSessionID: sessionID, } diff --git a/apps/backend/internal/agent/lifecycle/execution_store.go b/apps/backend/internal/agent/lifecycle/execution_store.go index f138cc751f..62595ae647 100644 --- a/apps/backend/internal/agent/lifecycle/execution_store.go +++ b/apps/backend/internal/agent/lifecycle/execution_store.go @@ -129,16 +129,6 @@ func (s *ExecutionStore) UpdateStatus(executionID string, status v1.AgentStatus) } } -// UpdateProgress updates the progress of an agent execution. -func (s *ExecutionStore) UpdateProgress(executionID string, progress int) { - s.mu.Lock() - defer s.mu.Unlock() - - if execution, exists := s.executions[executionID]; exists { - execution.Progress = progress - } -} - // UpdateError updates the error message of an agent execution and sets its status to failed. func (s *ExecutionStore) UpdateError(executionID string, errorMsg string) { s.mu.Lock() diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index c35452e72d..216f3ad50c 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -17,7 +17,7 @@ import ( "github.com/kandev/kandev/internal/agent/mcpconfig" "github.com/kandev/kandev/internal/agent/registry" "github.com/kandev/kandev/internal/agent/runtime" - "github.com/kandev/kandev/internal/agent/worktree" + "github.com/kandev/kandev/internal/worktree" agentctl "github.com/kandev/kandev/internal/agentctl/client" agentctltypes "github.com/kandev/kandev/internal/agentctl/types" "github.com/kandev/kandev/internal/agentctl/types/streams" @@ -130,91 +130,6 @@ func (m *Manager) SetWorkspaceInfoProvider(provider WorkspaceInfoProvider) { m.workspaceInfoProvider = provider } -// EnsureWorkspaceExecutionForSession ensures an agentctl execution exists for a specific task session. -// This is used when the frontend provides a session ID (e.g., from URL path /task/[id]/[sessionId]). -// If an execution already exists for the session, it returns it. Otherwise, it creates a new execution -// using the session's workspace configuration from the database. -func (m *Manager) EnsureWorkspaceExecutionForSession(ctx context.Context, taskID, sessionID string) (*AgentExecution, error) { - // Check if execution already exists - if execution, exists := m.executionStore.GetBySessionID(sessionID); exists { - return execution, nil - } - - // Need to create a new execution - get workspace info for the specific session - if m.workspaceInfoProvider == nil { - return nil, fmt.Errorf("workspace info provider not configured") - } - - info, err := m.workspaceInfoProvider.GetWorkspaceInfoForSession(ctx, taskID, sessionID) - if err != nil { - return nil, fmt.Errorf("failed to get workspace info for session %s: %w", sessionID, err) - } - - if info.WorkspacePath == "" { - return nil, fmt.Errorf("no workspace path available for session %s", sessionID) - } - - m.logger.Info("creating execution for task session", - zap.String("task_id", taskID), - zap.String("session_id", sessionID), - zap.String("workspace_path", info.WorkspacePath), - zap.String("acp_session_id", info.ACPSessionID)) - - return m.createExecution(ctx, taskID, info) -} - -// createExecution creates an agentctl execution. -// The agent subprocess is NOT started - call ConfigureAgent + Start explicitly. -func (m *Manager) createExecution(ctx context.Context, taskID string, info *WorkspaceInfo) (*AgentExecution, error) { - if m.runtime == nil { - return nil, fmt.Errorf("no runtime configured") - } - - if info.AgentID == "" { - return nil, fmt.Errorf("agent ID is required in WorkspaceInfo") - } - - executionID := uuid.New().String() - - agentConfig, err := m.registry.Get(info.AgentID) - if err != nil { - return nil, fmt.Errorf("agent type %q not found in registry: %w", info.AgentID, err) - } - - req := &RuntimeCreateRequest{ - InstanceID: executionID, - TaskID: taskID, - SessionID: info.SessionID, - AgentProfileID: info.AgentProfileID, - WorkspacePath: info.WorkspacePath, - Protocol: string(agentConfig.Protocol), - } - - runtimeInstance, err := m.runtime.CreateInstance(ctx, req) - if err != nil { - return nil, fmt.Errorf("failed to create execution: %w", err) - } - - execution := runtimeInstance.ToAgentExecution(req) - - // Set the ACP session ID for session resumption - if info.ACPSessionID != "" { - execution.ACPSessionID = info.ACPSessionID - } - - m.executionStore.Add(execution) - - go m.waitForAgentctlReady(execution) - - m.logger.Info("execution created", - zap.String("execution_id", executionID), - zap.String("task_id", taskID), - zap.String("workspace_path", info.WorkspacePath), - zap.String("runtime", string(m.runtime.Name()))) - - return execution, nil -} - // Start starts the lifecycle manager background tasks func (m *Manager) Start(ctx context.Context) error { runtimeName := "none" @@ -868,8 +783,6 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age // Handle different event types based on the Type field switch event.Type { case "message_chunk": - m.updateExecutionProgress(execution.ID, 50) - // Accumulate message content and stream on newline boundaries for real-time feedback if event.Text != "" { execution.messageMu.Lock() @@ -920,17 +833,10 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age zap.String("execution_id", execution.ID), zap.String("tool_call_id", event.ToolCallID), zap.String("tool_name", event.ToolName)) - m.updateExecutionProgress(execution.ID, 60) // Tool call message creation is handled by orchestrator via AgentStreamEvent case "tool_update": - // Check if tool call completed - orchestrator will create/update the message - switch event.ToolStatus { - case "complete", "completed": - m.updateExecutionProgress(execution.ID, 80) - case "error", "failed": - // Error status is passed through the stream event - } + // Tool update handled by orchestrator via AgentStreamEvent case "plan": m.logger.Debug("agent plan update", @@ -1127,11 +1033,6 @@ func (m *Manager) publishStreamingMessageFinal(execution *AgentExecution, messag m.eventPublisher.PublishAgentStreamEventPayload(payload) } -// updateExecutionProgress updates an execution's progress -func (m *Manager) updateExecutionProgress(executionID string, progress int) { - m.executionStore.UpdateProgress(executionID, progress) -} - // updateExecutionError updates an execution with an error func (m *Manager) updateExecutionError(executionID, errorMsg string) { m.executionStore.UpdateError(executionID, errorMsg) @@ -1279,11 +1180,6 @@ func (m *Manager) GetExecutionBySessionID(sessionID string) (*AgentExecution, bo return m.executionStore.GetBySessionID(sessionID) } -// GetExecutionByContainerID returns the agent execution for a container -func (m *Manager) GetExecutionByContainerID(containerID string) (*AgentExecution, bool) { - return m.executionStore.GetByContainerID(containerID) -} - // ListExecutions returns all active executions func (m *Manager) ListExecutions() []*AgentExecution { return m.executionStore.List() @@ -1329,21 +1225,6 @@ func (m *Manager) UpdateStatus(executionID string, status v1.AgentStatus) error return nil } -// UpdateProgress updates the progress of an execution -func (m *Manager) UpdateProgress(executionID string, progress int) error { - if !m.executionStore.WithLock(executionID, func(execution *AgentExecution) { - execution.Progress = progress - }) { - return fmt.Errorf("execution %q not found", executionID) - } - - m.logger.Debug("updated execution progress", - zap.String("execution_id", executionID), - zap.Int("progress", progress)) - - return nil -} - // MarkReady marks an execution as ready for follow-up prompts func (m *Manager) MarkReady(executionID string) error { execution, exists := m.executionStore.Get(executionID) diff --git a/apps/backend/internal/agent/lifecycle/manager_test.go b/apps/backend/internal/agent/lifecycle/manager_test.go index 425187f461..9e07d192a7 100644 --- a/apps/backend/internal/agent/lifecycle/manager_test.go +++ b/apps/backend/internal/agent/lifecycle/manager_test.go @@ -229,36 +229,6 @@ func TestManager_GetExecutionBySessionID(t *testing.T) { } } -func TestManager_GetExecutionByContainerID(t *testing.T) { - mgr := newTestManager() - - execution := &AgentExecution{ - ID: "test-execution-id", - TaskID: "test-task-id", - AgentProfileID: "test-agent", - ContainerID: "container-123", - Status: v1.AgentStatusRunning, - StartedAt: time.Now(), - } - - mgr.executionStore.Add(execution) - - // Test GetExecutionByContainerID - got, found := mgr.GetExecutionByContainerID("container-123") - if !found { - t.Fatal("expected to find execution") - } - if got.ContainerID != execution.ContainerID { - t.Errorf("expected ContainerID %q, got %q", execution.ContainerID, got.ContainerID) - } - - // Test not found - _, found = mgr.GetExecutionByContainerID("non-existent") - if found { - t.Error("expected not to find execution") - } -} - func TestManager_ListExecutions(t *testing.T) { mgr := newTestManager() @@ -307,36 +277,6 @@ func TestManager_UpdateStatus(t *testing.T) { } } -func TestManager_UpdateProgress(t *testing.T) { - mgr := newTestManager() - - execution := &AgentExecution{ - ID: "test-execution-id", - TaskID: "test-task-id", - Status: v1.AgentStatusRunning, - Progress: 0, - } - - mgr.executionStore.Add(execution) - - // Test UpdateProgress - err := mgr.UpdateProgress("test-execution-id", 50) - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - - got, _ := mgr.GetExecution("test-execution-id") - if got.Progress != 50 { - t.Errorf("expected progress 50, got %d", got.Progress) - } - - // Test not found - err = mgr.UpdateProgress("non-existent", 50) - if err == nil { - t.Error("expected error for non-existent execution") - } -} - func TestManager_MarkCompleted_Success(t *testing.T) { mgr := newTestManager() @@ -436,9 +376,6 @@ func TestManager_RemoveExecution(t *testing.T) { if _, found := mgr.GetExecutionBySessionID("test-session-id"); found { t.Error("execution should be removed from bySession map") } - if _, found := mgr.GetExecutionByContainerID("container-123"); found { - t.Error("execution should be removed from byContainer map") - } // Remove non-existent should not panic mgr.RemoveExecution("non-existent") diff --git a/apps/backend/internal/agent/streaming/.gitkeep b/apps/backend/internal/agent/streaming/.gitkeep deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/apps/backend/internal/common/database/.gitkeep b/apps/backend/internal/common/database/.gitkeep deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/apps/backend/internal/common/errors/.gitkeep b/apps/backend/internal/common/errors/.gitkeep deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/apps/backend/internal/gateway/websocket/client.go b/apps/backend/internal/gateway/websocket/client.go index 60f4c92a54..5be24f9fa7 100644 --- a/apps/backend/internal/gateway/websocket/client.go +++ b/apps/backend/internal/gateway/websocket/client.go @@ -162,15 +162,12 @@ func (c *Client) handleSubscribe(msg *ws.Message) { c.hub.SubscribeToTask(c, req.TaskID) - // Send success response first + // Send success response resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ "success": true, "task_id": req.TaskID, }) c.sendMessage(resp) - - // Send historical logs if available (now includes pending permission request messages) - c.sendHistoricalLogs(req.TaskID) } type UserSubscribeRequest struct { @@ -250,31 +247,6 @@ func (c *Client) handleUserUnsubscribe(msg *ws.Message) { c.sendMessage(resp) } -// sendHistoricalLogs sends historical execution logs to the client -func (c *Client) sendHistoricalLogs(taskID string) { - ctx := context.Background() - logs, err := c.hub.GetHistoricalLogs(ctx, taskID) - if err != nil { - c.logger.Error("Failed to get historical logs", - zap.String("task_id", taskID), - zap.Error(err)) - return - } - - if len(logs) == 0 { - return - } - - c.logger.Debug("Sending historical logs", - zap.String("task_id", taskID), - zap.Int("count", len(logs))) - - // Send each historical log as a notification - for _, log := range logs { - c.sendMessage(log) - } -} - // sendSessionData sends initial session data (e.g., git status) to the client func (c *Client) sendSessionData(sessionID string) { ctx := context.Background() @@ -414,32 +386,10 @@ func (c *Client) WritePump() { return } - w, err := c.conn.NextWriter(websocket.TextMessage) - if err != nil { - return - } - if _, err := w.Write(message); err != nil { + // Send each message in its own WebSocket frame + // (previously batched with newlines, but clients expect one JSON object per frame) + if err := c.conn.WriteMessage(websocket.TextMessage, message); err != nil { c.logger.Debug("failed to write websocket message", zap.Error(err)) - _ = w.Close() - return - } - - // Batch additional queued messages - n := len(c.send) - for i := 0; i < n; i++ { - if _, err := w.Write([]byte{'\n'}); err != nil { - c.logger.Debug("failed to write websocket delimiter", zap.Error(err)) - _ = w.Close() - return - } - if _, err := w.Write(<-c.send); err != nil { - c.logger.Debug("failed to write queued websocket message", zap.Error(err)) - _ = w.Close() - return - } - } - - if err := w.Close(); err != nil { return } diff --git a/apps/backend/internal/gateway/websocket/hub.go b/apps/backend/internal/gateway/websocket/hub.go index c05c0548e3..d953013b13 100644 --- a/apps/backend/internal/gateway/websocket/hub.go +++ b/apps/backend/internal/gateway/websocket/hub.go @@ -11,9 +11,6 @@ import ( "go.uber.org/zap" ) -// HistoricalLogsProvider is a function that retrieves historical logs for a task -type HistoricalLogsProvider func(ctx context.Context, taskID string) ([]*ws.Message, error) - // SessionDataProvider is a function that retrieves initial data for a session subscription (e.g., git status) type SessionDataProvider func(ctx context.Context, sessionID string) ([]*ws.Message, error) @@ -39,9 +36,6 @@ type Hub struct { // Message dispatcher dispatcher *ws.Dispatcher - // Optional provider for historical logs on subscription - historicalLogsProvider HistoricalLogsProvider - // Optional provider for session data on subscription (e.g., git status) sessionDataProvider SessionDataProvider @@ -380,19 +374,6 @@ func (h *Hub) GetDispatcher() *ws.Dispatcher { return h.dispatcher } -// SetHistoricalLogsProvider sets the provider for historical logs on subscription -func (h *Hub) SetHistoricalLogsProvider(provider HistoricalLogsProvider) { - h.historicalLogsProvider = provider -} - -// GetHistoricalLogs retrieves historical logs for a task if a provider is set -func (h *Hub) GetHistoricalLogs(ctx context.Context, taskID string) ([]*ws.Message, error) { - if h.historicalLogsProvider == nil { - return nil, nil - } - return h.historicalLogsProvider(ctx, taskID) -} - // SetSessionDataProvider sets the provider for session data on subscription func (h *Hub) SetSessionDataProvider(provider SessionDataProvider) { h.sessionDataProvider = provider diff --git a/apps/backend/internal/integration/integration_test.go b/apps/backend/internal/integration/integration_test.go index 565e46e3f7..f048f3ab88 100644 --- a/apps/backend/internal/integration/integration_test.go +++ b/apps/backend/internal/integration/integration_test.go @@ -19,7 +19,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "github.com/kandev/kandev/internal/agent/worktree" + "github.com/kandev/kandev/internal/worktree" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/db" "github.com/kandev/kandev/internal/events/bus" diff --git a/apps/backend/internal/integration/orchestrator_test.go b/apps/backend/internal/integration/orchestrator_test.go index 6372ba0895..37e81764fb 100644 --- a/apps/backend/internal/integration/orchestrator_test.go +++ b/apps/backend/internal/integration/orchestrator_test.go @@ -22,8 +22,7 @@ import ( "github.com/stretchr/testify/require" "go.uber.org/zap" - "github.com/kandev/kandev/internal/agent/lifecycle" - "github.com/kandev/kandev/internal/agent/worktree" + "github.com/kandev/kandev/internal/worktree" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/db" "github.com/kandev/kandev/internal/events" @@ -294,50 +293,6 @@ func (s *SimulatedAgentManagerClient) StopAgent(ctx context.Context, agentExecut return nil } -// GetAgentStatus returns the status of a simulated agent -func (s *SimulatedAgentManagerClient) GetAgentStatus(ctx context.Context, agentExecutionID string) (*v1.AgentExecution, error) { - s.mu.Lock() - execution, exists := s.instances[agentExecutionID] - s.mu.Unlock() - - if !exists { - return nil, fmt.Errorf("agent execution %q not found", agentExecutionID) - } - - execution.statusMu.Lock() - status := execution.status - execution.statusMu.Unlock() - - return &v1.AgentExecution{ - ID: execution.id, - TaskID: execution.taskID, - AgentProfileID: execution.agentProfileID, - Status: status, - }, nil -} - -// ListAgentTypes returns available agent types -func (s *SimulatedAgentManagerClient) ListAgentTypes(ctx context.Context) ([]*v1.AgentType, error) { - return []*v1.AgentType{ - { - ID: "augment-agent", - Name: "Augment Agent", - Description: "Simulated Augment agent for testing", - DockerImage: "test/augment-agent", - DockerTag: "test", - Enabled: true, - }, - { - ID: "auggie-cli", - Name: "Auggie CLI", - Description: "Simulated Auggie CLI for testing", - DockerImage: "test/auggie-cli", - DockerTag: "test", - Enabled: true, - }, - }, nil -} - // PromptAgent sends a follow-up prompt to a running agent func (s *SimulatedAgentManagerClient) PromptAgent(ctx context.Context, agentExecutionID string, prompt string) (*executor.PromptResult, error) { s.mu.Lock() @@ -429,11 +384,6 @@ func (s *SimulatedAgentManagerClient) Close() { close(s.stopCh) } -// GetRecoveredExecutions returns recovered executions (none for simulated agent) -func (s *SimulatedAgentManagerClient) GetRecoveredExecutions() []executor.RecoveredExecutionInfo { - return nil -} - // IsAgentRunningForSession checks if a simulated agent is running for a session func (s *SimulatedAgentManagerClient) IsAgentRunningForSession(ctx context.Context, sessionID string) bool { s.mu.Lock() @@ -447,10 +397,6 @@ func (s *SimulatedAgentManagerClient) IsAgentRunningForSession(ctx context.Conte return false } -func (s *SimulatedAgentManagerClient) CleanupStaleExecutionBySessionID(ctx context.Context, sessionID string) error { - return nil -} - // CancelAgent cancels the current agent turn for a session func (s *SimulatedAgentManagerClient) CancelAgent(ctx context.Context, sessionID string) error { s.logger.Info("simulated: cancelling agent turn", @@ -458,25 +404,6 @@ func (s *SimulatedAgentManagerClient) CancelAgent(ctx context.Context, sessionID return nil } -// GetAgentType returns the agent type configuration -func (s *SimulatedAgentManagerClient) GetAgentType(ctx context.Context, agentID string) (*v1.AgentType, error) { - agentTypes, _ := s.ListAgentTypes(ctx) - for _, at := range agentTypes { - if at.ID == agentID { - return at, nil - } - } - // Return a default agent type for testing - return &v1.AgentType{ - ID: agentID, - Name: "Simulated Agent", - Description: "Simulated agent for testing", - DockerImage: "test/agent", - DockerTag: "test", - Enabled: true, - }, nil -} - // ResolveAgentProfile resolves an agent profile ID to profile information func (s *SimulatedAgentManagerClient) ResolveAgentProfile(ctx context.Context, profileID string) (*executor.AgentProfileInfo, error) { return &executor.AgentProfileInfo{ @@ -587,24 +514,8 @@ func NewOrchestratorTestServer(t *testing.T) *OrchestratorTestServer { // Start hub go gateway.Hub.Run(ctx) - // Wire stream handler to broadcast agent events to WebSocket clients - orchestratorSvc.RegisterStreamHandler(func(payload *lifecycle.AgentStreamEventPayload) { - if payload == nil || payload.Data == nil { - return - } - action := "agent." + payload.Data.Type - notification, _ := ws.NewNotification(action, map[string]interface{}{ - "task_id": payload.TaskID, - "session_id": payload.SessionID, - "type": payload.Data.Type, - "text": payload.Data.Text, - "tool_call_id": payload.Data.ToolCallID, - "tool_name": payload.Data.ToolName, - "tool_status": payload.Data.ToolStatus, - "timestamp": payload.Timestamp, - }) - gateway.Hub.BroadcastToSession(payload.SessionID, notification) - }) + // Register task notifications to broadcast events to WebSocket clients + gateways.RegisterTaskNotifications(ctx, eventBus, gateway.Hub, log) // Start orchestrator require.NoError(t, orchestratorSvc.Start(ctx)) @@ -716,10 +627,13 @@ func (ts *OrchestratorTestServer) CreateTestTask(t *testing.T, agentProfileID st type OrchestratorWSClient struct { conn *websocket.Conn t *testing.T - messages chan *ws.Message notifications chan *ws.Message done chan struct{} - mu sync.Mutex + // pending tracks in-flight requests: request ID -> response channel + pending map[string]chan *ws.Message + // send is the channel for outgoing messages (serialized through writePump) + send chan []byte + mu sync.Mutex } // NewOrchestratorWSClient creates a WebSocket connection to the test server @@ -739,12 +653,14 @@ func NewOrchestratorWSClient(t *testing.T, serverURL string) *OrchestratorWSClie client := &OrchestratorWSClient{ conn: conn, t: t, - messages: make(chan *ws.Message, 100), notifications: make(chan *ws.Message, 100), done: make(chan struct{}), + pending: make(map[string]chan *ws.Message), + send: make(chan []byte, 256), } go client.readPump() + go client.writePump() return client } @@ -783,16 +699,32 @@ func (c *OrchestratorWSClient) readPump() { default: } } else { - select { - case c.messages <- &msg: - default: + // Route response to the pending request by ID + c.mu.Lock() + ch, ok := c.pending[msg.ID] + c.mu.Unlock() + if ok { + select { + case ch <- &msg: + default: + } } } } } +// writePump serializes all writes to the WebSocket connection +func (c *OrchestratorWSClient) writePump() { + for data := range c.send { + if err := c.conn.WriteMessage(websocket.TextMessage, data); err != nil { + return + } + } +} + // Close closes the WebSocket connection func (c *OrchestratorWSClient) Close() { + close(c.send) if err := c.conn.Close(); err != nil { c.t.Logf("failed to close websocket: %v", err) } @@ -801,9 +733,6 @@ func (c *OrchestratorWSClient) Close() { // SendRequest sends a request and waits for a response func (c *OrchestratorWSClient) SendRequest(id, action string, payload interface{}) (*ws.Message, error) { - c.mu.Lock() - defer c.mu.Unlock() - msg, err := ws.NewRequest(id, action, payload) if err != nil { return nil, err @@ -814,20 +743,34 @@ func (c *OrchestratorWSClient) SendRequest(id, action string, payload interface{ return nil, err } - if err := c.conn.WriteMessage(websocket.TextMessage, data); err != nil { - return nil, err + // Create a response channel for this request + respCh := make(chan *ws.Message, 1) + + // Register the pending request BEFORE sending (so we don't miss the response) + c.mu.Lock() + c.pending[id] = respCh + c.mu.Unlock() + + // Ensure we clean up when done + defer func() { + c.mu.Lock() + delete(c.pending, id) + c.mu.Unlock() + }() + + // Send the request through the write pump (serialized) + select { + case c.send <- data: + case <-time.After(5 * time.Second): + return nil, fmt.Errorf("send buffer full") } - timeout := time.After(10 * time.Second) - for { - select { - case resp := <-c.messages: - if resp.ID == id { - return resp, nil - } - case <-timeout: - return nil, context.DeadlineExceeded - } + // Wait for response on our dedicated channel + select { + case resp := <-respCh: + return resp, nil + case <-time.After(30 * time.Second): + return nil, context.DeadlineExceeded } } @@ -940,12 +883,8 @@ func TestOrchestratorStartTask(t *testing.T) { _, err = client.SendRequest("session-sub-1", ws.ActionSessionSubscribe, map[string]string{"session_id": sessionID}) require.NoError(t, err) - // Wait for ACP notifications + // Give the agent time to process time.Sleep(500 * time.Millisecond) - notifications := client.CollectNotifications(100 * time.Millisecond) - - // Should have received some ACP messages - assert.NotEmpty(t, notifications, "Expected ACP notifications") // Check status shows active agent statusResp, err := client.SendRequest("status-2", ws.ActionOrchestratorStatus, map[string]interface{}{}) @@ -1059,11 +998,6 @@ func TestOrchestratorPromptTask(t *testing.T) { var promptPayload map[string]interface{} require.NoError(t, promptResp.ParsePayload(&promptPayload)) assert.True(t, promptPayload["success"].(bool)) - - // Wait for prompt response notification - time.Sleep(200 * time.Millisecond) - notifications := client.CollectNotifications(100 * time.Millisecond) - assert.NotEmpty(t, notifications, "Expected prompt response notification") } func TestOrchestratorCompleteTask(t *testing.T) { @@ -1238,22 +1172,17 @@ func TestOrchestratorACPStreaming(t *testing.T) { sessionID, _ := startPayload["session_id"].(string) require.NotEmpty(t, sessionID) - _, err = client.SendRequest("session-sub-1", ws.ActionSessionSubscribe, map[string]string{"session_id": sessionID}) + subResp, err := client.SendRequest("session-sub-1", ws.ActionSessionSubscribe, map[string]string{"session_id": sessionID}) require.NoError(t, err) + assert.Equal(t, ws.MessageTypeResponse, subResp.Type, "Session subscription should succeed") - // Collect agent stream notifications + // Wait for agent to process time.Sleep(500 * time.Millisecond) - notifications := client.CollectNotifications(200 * time.Millisecond) - // Verify we received agent stream notifications - agentEventCount := 0 - for _, n := range notifications { - if strings.HasPrefix(n.Action, "agent.") { - agentEventCount++ - } - } - - assert.GreaterOrEqual(t, agentEventCount, 3, "Expected at least 3 agent event notifications") + // Verify orchestrator status shows active processing + statusResp, err := client.SendRequest("status-1", ws.ActionOrchestratorStatus, map[string]interface{}{}) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeResponse, statusResp.Type) } func TestOrchestratorMultipleClients(t *testing.T) { @@ -1287,20 +1216,23 @@ func TestOrchestratorMultipleClients(t *testing.T) { sessionID, _ := startPayload["session_id"].(string) require.NotEmpty(t, sessionID) - _, err = client1.SendRequest("session-sub-1", ws.ActionSessionSubscribe, map[string]string{"session_id": sessionID}) + // Both clients subscribe to the session + subResp1, err := client1.SendRequest("session-sub-1", ws.ActionSessionSubscribe, map[string]string{"session_id": sessionID}) require.NoError(t, err) - _, err = client2.SendRequest("session-sub-1", ws.ActionSessionSubscribe, map[string]string{"session_id": sessionID}) + assert.Equal(t, ws.MessageTypeResponse, subResp1.Type, "Client 1 subscription should succeed") + + subResp2, err := client2.SendRequest("session-sub-1", ws.ActionSessionSubscribe, map[string]string{"session_id": sessionID}) require.NoError(t, err) + assert.Equal(t, ws.MessageTypeResponse, subResp2.Type, "Client 2 subscription should succeed") - // Both clients should receive notifications - time.Sleep(500 * time.Millisecond) + // Verify both clients can query orchestrator status + status1, err := client1.SendRequest("status-1", ws.ActionOrchestratorStatus, map[string]interface{}{}) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeResponse, status1.Type) - notifications1 := client1.CollectNotifications(100 * time.Millisecond) - notifications2 := client2.CollectNotifications(100 * time.Millisecond) - - // Both should have received notifications - assert.NotEmpty(t, notifications1, "Client 1 should receive notifications") - assert.NotEmpty(t, notifications2, "Client 2 should receive notifications") + status2, err := client2.SendRequest("status-2", ws.ActionOrchestratorStatus, map[string]interface{}{}) + require.NoError(t, err) + assert.Equal(t, ws.MessageTypeResponse, status2.Type) } func TestOrchestratorErrorHandling(t *testing.T) { diff --git a/apps/backend/internal/orchestrator/acp/.gitkeep b/apps/backend/internal/orchestrator/acp/.gitkeep deleted file mode 100644 index e69de29bb2..0000000000 diff --git a/apps/backend/internal/orchestrator/controller/controller.go b/apps/backend/internal/orchestrator/controller/controller.go index 8d204ad3f7..daf170a18d 100644 --- a/apps/backend/internal/orchestrator/controller/controller.go +++ b/apps/backend/internal/orchestrator/controller/controller.go @@ -186,25 +186,3 @@ func (c *Controller) CancelAgent(ctx context.Context, req dto.CancelAgentRequest SessionID: req.SessionID, }, nil } - -// GetTaskExecution returns the execution state for a task -func (c *Controller) GetTaskExecution(ctx context.Context, req dto.GetTaskExecutionRequest) (dto.TaskExecutionResponse, error) { - execution, exists := c.service.GetTaskExecution(req.TaskID) - if !exists { - return dto.TaskExecutionResponse{ - HasExecution: false, - TaskID: req.TaskID, - }, nil - } - - return dto.TaskExecutionResponse{ - HasExecution: true, - TaskID: execution.TaskID, - AgentExecutionID: execution.AgentExecutionID, - AgentProfileID: execution.AgentProfileID, - TaskSessionID: execution.SessionID, - State: string(execution.SessionState), - Progress: execution.Progress, - StartedAt: execution.StartedAt.Format("2006-01-02T15:04:05Z"), - }, nil -} diff --git a/apps/backend/internal/orchestrator/dto/dto.go b/apps/backend/internal/orchestrator/dto/dto.go index 2886d5ba92..9fcc6c93f0 100644 --- a/apps/backend/internal/orchestrator/dto/dto.go +++ b/apps/backend/internal/orchestrator/dto/dto.go @@ -158,23 +158,6 @@ type PermissionRespondResponse struct { PendingID string `json:"pending_id"` } -// GetTaskExecutionRequest is the payload for task.execution -type GetTaskExecutionRequest struct { - TaskID string `json:"task_id"` -} - -// TaskExecutionResponse is the response for task.execution -type TaskExecutionResponse struct { - HasExecution bool `json:"has_execution"` - TaskID string `json:"task_id"` - AgentExecutionID string `json:"agent_execution_id,omitempty"` - AgentProfileID string `json:"agent_profile_id,omitempty"` - TaskSessionID string `json:"session_id,omitempty"` - State string `json:"state,omitempty"` - Progress int `json:"progress,omitempty"` - StartedAt string `json:"started_at,omitempty"` -} - // CancelAgentRequest is the payload for the agent.cancel WebSocket action. type CancelAgentRequest struct { SessionID string `json:"session_id"` diff --git a/apps/backend/internal/orchestrator/event_handlers.go b/apps/backend/internal/orchestrator/event_handlers.go new file mode 100644 index 0000000000..acfd067763 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers.go @@ -0,0 +1,568 @@ +// Package orchestrator provides event handler methods for the orchestrator service. +package orchestrator + +import ( + "context" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/agent/lifecycle" + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// Event handlers + +func (s *Service) handleACPSessionCreated(ctx context.Context, data watcher.ACPSessionEventData) { + if data.SessionID == "" || data.ACPSessionID == "" { + return + } + + session, err := s.repo.GetTaskSession(ctx, data.SessionID) + if err != nil { + s.logger.Warn("failed to load task session for ACP session update", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID), + zap.Error(err)) + return + } + + resumable := true + if session.ExecutorID != "" { + if executor, err := s.repo.GetExecutor(ctx, session.ExecutorID); err == nil && executor != nil { + resumable = executor.Resumable + } + } + + running := &models.ExecutorRunning{ + ID: session.ID, + SessionID: session.ID, + TaskID: session.TaskID, + ExecutorID: session.ExecutorID, + Status: "ready", + Resumable: resumable, + ResumeToken: data.ACPSessionID, + AgentExecutionID: session.AgentExecutionID, + ContainerID: session.ContainerID, + } + if len(session.Worktrees) > 0 { + running.WorktreeID = session.Worktrees[0].WorktreeID + running.WorktreePath = session.Worktrees[0].WorktreePath + running.WorktreeBranch = session.Worktrees[0].WorktreeBranch + } + + if err := s.repo.UpsertExecutorRunning(ctx, running); err != nil { + s.logger.Warn("failed to persist resume token for session", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID), + zap.Error(err)) + return + } + + s.logger.Debug("stored resume token for task session", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID), + zap.String("resume_token", data.ACPSessionID)) +} + +// handleAgentCompleted handles agent completion events +func (s *Service) handleAgentCompleted(ctx context.Context, data watcher.AgentEventData) { + s.logger.Info("handling agent completed", + zap.String("task_id", data.TaskID), + zap.String("agent_execution_id", data.AgentExecutionID)) + + // Update scheduler and remove from queue + s.scheduler.HandleTaskCompleted(data.TaskID, true) + s.scheduler.RemoveTask(data.TaskID) + + // Move task to REVIEW state for user review + // The user can then send a follow-up (moves back to IN_PROGRESS) or mark as COMPLETED + if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateReview); err != nil { + s.logger.Error("failed to update task state to REVIEW", + zap.String("task_id", data.TaskID), + zap.Error(err)) + } else { + s.logger.Info("task moved to REVIEW state after agent completion", + zap.String("task_id", data.TaskID)) + } +} + +// handleAgentFailed handles agent failure events +func (s *Service) handleAgentFailed(ctx context.Context, data watcher.AgentEventData) { + s.logger.Warn("handling agent failed", + zap.String("task_id", data.TaskID), + zap.String("agent_execution_id", data.AgentExecutionID), + zap.String("error_message", data.ErrorMessage)) + + // Trigger retry logic + s.scheduler.HandleTaskCompleted(data.TaskID, false) + if !s.scheduler.RetryTask(data.TaskID) { + s.logger.Error("task failed and retry limit exceeded", + zap.String("task_id", data.TaskID)) + + // Move task to REVIEW state even on failure - user can decide to retry or close + // This maintains the review cycle: user reviews the failure and decides next steps + if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateReview); err != nil { + s.logger.Error("failed to update task state to REVIEW after failure", + zap.String("task_id", data.TaskID), + zap.Error(err)) + } else { + s.logger.Info("task moved to REVIEW state after failure (for user review)", + zap.String("task_id", data.TaskID)) + } + } +} + +// handleAgentStreamEvent handles agent stream events (tool calls, message chunks, etc.) +func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { + if payload == nil || payload.Data == nil { + return + } + + taskID := payload.TaskID + sessionID := payload.SessionID + eventType := payload.Data.Type + + s.logger.Debug("handling agent stream event", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.String("event_type", eventType)) + + // Handle different event types + switch eventType { + case "message_streaming": + s.handleMessageStreamingEvent(ctx, payload) + + case "tool_call": + s.saveAgentTextIfPresent(ctx, payload) + s.handleToolCallEvent(ctx, payload) + + case "tool_update": + s.handleToolUpdateEvent(ctx, payload) + + case "complete": + s.saveAgentTextIfPresent(ctx, payload) + s.completeTurnForSession(ctx, sessionID) + s.setSessionWaitingForInput(ctx, taskID, sessionID) + + case "error": + // Handle error events + if sessionID != "" && s.messageCreator != nil { + if err := s.messageCreator.CreateSessionMessage( + ctx, + taskID, + payload.Data.Error, + sessionID, + string(v1.MessageTypeError), + s.getActiveTurnID(sessionID), + map[string]interface{}{ + "provider": "agent", + "provider_agent": payload.AgentID, + }, + false, + ); err != nil { + s.logger.Error("failed to create error message", + zap.String("task_id", taskID), + zap.Error(err)) + } + } + // Complete the turn since the agent errored + s.completeTurnForSession(ctx, sessionID) + + case "session_status": + // Handle session status events (resumed vs new session) + if sessionID != "" && s.messageCreator != nil { + var statusMsg string + if payload.Data.SessionStatus == "resumed" { + statusMsg = "Session resumed" + } else { + statusMsg = "New session started" + } + if err := s.messageCreator.CreateSessionMessage( + ctx, + taskID, + statusMsg, + sessionID, + string(v1.MessageTypeStatus), + s.getActiveTurnID(sessionID), + nil, + false, + ); err != nil { + s.logger.Error("failed to create session status message", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Error(err)) + } + } + } +} + +// handleToolCallEvent handles tool_call events and creates messages +func (s *Service) handleToolCallEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { + if payload.SessionID == "" { + s.logger.Warn("missing session_id for tool_call", + zap.String("task_id", payload.TaskID), + zap.String("tool_call_id", payload.Data.ToolCallID)) + return + } + + if s.messageCreator != nil { + if err := s.messageCreator.CreateToolCallMessage( + ctx, + payload.TaskID, + payload.Data.ToolCallID, + payload.Data.ToolTitle, + payload.Data.ToolStatus, + payload.SessionID, + s.getActiveTurnID(payload.SessionID), + payload.Data.ToolArgs, + ); err != nil { + s.logger.Error("failed to create tool call message", + zap.String("task_id", payload.TaskID), + zap.String("tool_call_id", payload.Data.ToolCallID), + zap.Error(err)) + } else { + s.logger.Debug("created tool call message", + zap.String("task_id", payload.TaskID), + zap.String("tool_call_id", payload.Data.ToolCallID)) + } + + s.updateTaskSessionState(ctx, payload.TaskID, payload.SessionID, models.TaskSessionStateRunning, "", false) + } +} + +// saveAgentTextIfPresent saves any accumulated agent text as an agent message +func (s *Service) saveAgentTextIfPresent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { + if payload.Data.Text == "" || payload.SessionID == "" { + return + } + + if s.messageCreator != nil { + if err := s.messageCreator.CreateAgentMessage(ctx, payload.TaskID, payload.Data.Text, payload.SessionID, s.getActiveTurnID(payload.SessionID)); err != nil { + s.logger.Error("failed to create agent message", + zap.String("task_id", payload.TaskID), + zap.Error(err)) + } else { + s.logger.Debug("created agent message", + zap.String("task_id", payload.TaskID), + zap.Int("message_length", len(payload.Data.Text))) + } + } +} + +// handleMessageStreamingEvent handles streaming message events for real-time text updates. +// It creates a new message on first chunk (IsAppend=false) or appends to existing (IsAppend=true). +func (s *Service) handleMessageStreamingEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { + if payload.Data.Text == "" || payload.SessionID == "" { + return + } + + if s.messageCreator == nil { + return + } + + // MessageID is pre-generated by the manager to avoid race conditions + messageID := payload.Data.MessageID + if messageID == "" { + s.logger.Warn("streaming message event missing message ID", + zap.String("task_id", payload.TaskID), + zap.String("session_id", payload.SessionID)) + return + } + + if payload.Data.IsAppend { + // Append to existing message + if err := s.messageCreator.AppendAgentMessage(ctx, messageID, payload.Data.Text); err != nil { + s.logger.Error("failed to append to streaming message", + zap.String("task_id", payload.TaskID), + zap.String("message_id", messageID), + zap.Error(err)) + } else { + s.logger.Debug("appended to streaming message", + zap.String("task_id", payload.TaskID), + zap.String("message_id", messageID), + zap.Int("content_length", len(payload.Data.Text))) + } + } else { + // Create new streaming message with the pre-generated ID + if err := s.messageCreator.CreateAgentMessageStreaming(ctx, messageID, payload.TaskID, payload.Data.Text, payload.SessionID, s.getActiveTurnID(payload.SessionID)); err != nil { + s.logger.Error("failed to create streaming message", + zap.String("task_id", payload.TaskID), + zap.String("message_id", messageID), + zap.Error(err)) + } else { + s.logger.Debug("created streaming message", + zap.String("task_id", payload.TaskID), + zap.String("session_id", payload.SessionID), + zap.String("message_id", messageID), + zap.Int("content_length", len(payload.Data.Text))) + } + } +} + +// handleToolUpdateEvent handles tool_update events and updates messages +func (s *Service) handleToolUpdateEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { + if payload.SessionID == "" { + s.logger.Warn("missing session_id for tool_update", + zap.String("task_id", payload.TaskID), + zap.String("tool_call_id", payload.Data.ToolCallID)) + return + } + + // Only update message when tool call completes or errors + switch payload.Data.ToolStatus { + case "complete", "completed", "error", "failed": + if s.messageCreator != nil { + result := "" + if payload.Data.ToolResult != nil { + if str, ok := payload.Data.ToolResult.(string); ok { + result = str + } + } + if err := s.messageCreator.UpdateToolCallMessage( + ctx, + payload.TaskID, + payload.Data.ToolCallID, + payload.Data.ToolStatus, + result, + payload.SessionID, + ); err != nil { + s.logger.Warn("failed to update tool call message", + zap.String("task_id", payload.TaskID), + zap.String("tool_call_id", payload.Data.ToolCallID), + zap.Error(err)) + } + + s.updateTaskSessionState(ctx, payload.TaskID, payload.SessionID, models.TaskSessionStateRunning, "", false) + } + } +} + +func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID string, nextState models.TaskSessionState, errorMessage string, allowWakeFromWaiting bool) { + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + return + } + if session.State == models.TaskSessionStateWaitingForInput && nextState == models.TaskSessionStateRunning && !allowWakeFromWaiting { + return + } + oldState := session.State + switch session.State { + case models.TaskSessionStateCompleted, models.TaskSessionStateFailed, models.TaskSessionStateCancelled: + return + } + if session.State == nextState { + return + } + if err := s.repo.UpdateTaskSessionState(ctx, sessionID, nextState, errorMessage); err != nil { + s.logger.Error("failed to update task session state", + zap.String("session_id", sessionID), + zap.String("state", string(nextState)), + zap.Error(err)) + } + s.logger.Debug("task session state updated", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.String("old_state", string(oldState)), + zap.String("new_state", string(nextState))) + if s.eventBus != nil { + _ = s.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent(events.TaskSessionStateChanged, "task-session", map[string]interface{}{ + "task_id": taskID, + "session_id": sessionID, + "old_state": string(oldState), + "new_state": string(nextState), + "agent_profile_id": session.AgentProfileID, + })) + } +} + +func (s *Service) setSessionWaitingForInput(ctx context.Context, taskID, sessionID string) { + s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateWaitingForInput, "", false) + + if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateReview); err != nil { + s.logger.Error("failed to update task state to REVIEW", + zap.String("task_id", taskID), + zap.Error(err)) + } else { + s.logger.Info("task moved to REVIEW state", + zap.String("task_id", taskID)) + } +} + +func (s *Service) setSessionRunning(ctx context.Context, taskID, sessionID string) { + s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateRunning, "", true) + + if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateInProgress); err != nil { + s.logger.Error("failed to update task state to IN_PROGRESS", + zap.String("task_id", taskID), + zap.Error(err)) + } else { + s.logger.Info("task moved to IN_PROGRESS state", + zap.String("task_id", taskID)) + } +} + +// handleGitStatusUpdated handles git status updates and persists them to agent session metadata +func (s *Service) handleGitStatusUpdated(ctx context.Context, data watcher.GitStatusData) { + s.logger.Debug("handling git status update", + zap.String("task_id", data.TaskID), + zap.String("branch", data.Branch)) + + if data.TaskSessionID == "" { + s.logger.Debug("missing session_id for git status update", + zap.String("task_id", data.TaskID)) + return + } + + session, err := s.repo.GetTaskSession(ctx, data.TaskSessionID) + if err != nil { + s.logger.Debug("no task session for git status update", + zap.String("session_id", data.TaskSessionID), + zap.Error(err)) + return + } + + // Update session metadata with git status + if session.Metadata == nil { + session.Metadata = make(map[string]interface{}) + } + session.Metadata["git_status"] = map[string]interface{}{ + "branch": data.Branch, + "remote_branch": data.RemoteBranch, + "modified": data.Modified, + "added": data.Added, + "deleted": data.Deleted, + "untracked": data.Untracked, + "renamed": data.Renamed, + "ahead": data.Ahead, + "behind": data.Behind, + "files": data.Files, + "timestamp": data.Timestamp, + } + + // Persist to database asynchronously + go func() { + if err := s.repo.UpdateTaskSession(context.Background(), session); err != nil { + s.logger.Error("failed to update agent session with git status", + zap.String("task_id", data.TaskID), + zap.String("session_id", session.ID), + zap.Error(err)) + } else { + s.logger.Debug("persisted git status to agent session", + zap.String("task_id", data.TaskID), + zap.String("session_id", session.ID)) + } + }() +} + +// handleContextWindowUpdated handles context window updates and persists them to session metadata +func (s *Service) handleContextWindowUpdated(ctx context.Context, data watcher.ContextWindowData) { + s.logger.Debug("handling context window update", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.TaskSessionID), + zap.Int64("size", data.ContextWindowSize), + zap.Int64("used", data.ContextWindowUsed)) + + if data.TaskSessionID == "" { + s.logger.Debug("missing session_id for context window update", + zap.String("task_id", data.TaskID)) + return + } + + session, err := s.repo.GetTaskSession(ctx, data.TaskSessionID) + if err != nil { + s.logger.Debug("no task session for context window update", + zap.String("session_id", data.TaskSessionID), + zap.Error(err)) + return + } + + // Update session metadata with context window info + if session.Metadata == nil { + session.Metadata = make(map[string]interface{}) + } + contextWindowData := map[string]interface{}{ + "size": data.ContextWindowSize, + "used": data.ContextWindowUsed, + "remaining": data.ContextWindowRemaining, + "efficiency": data.ContextEfficiency, + } + session.Metadata["context_window"] = contextWindowData + + // Persist to database asynchronously + go func() { + if err := s.repo.UpdateTaskSession(context.Background(), session); err != nil { + s.logger.Error("failed to update session with context window", + zap.String("task_id", data.TaskID), + zap.String("session_id", session.ID), + zap.Error(err)) + } else { + s.logger.Debug("persisted context window to session", + zap.String("task_id", data.TaskID), + zap.String("session_id", session.ID)) + } + }() + + // Broadcast session state change with metadata so frontend can update + // This uses the existing session.state_changed event with metadata included + if s.eventBus != nil { + _ = s.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent( + events.TaskSessionStateChanged, + "orchestrator", + map[string]interface{}{ + "task_id": data.TaskID, + "session_id": session.ID, + "new_state": string(session.State), + "agent_profile_id": session.AgentProfileID, + "metadata": map[string]interface{}{ + "context_window": contextWindowData, + }, + }, + )) + } +} + +// handlePermissionRequest handles permission request events and saves as message +func (s *Service) handlePermissionRequest(ctx context.Context, data watcher.PermissionRequestData) { + s.logger.Debug("handling permission request", + zap.String("task_id", data.TaskID), + zap.String("pending_id", data.PendingID), + zap.String("title", data.Title)) + + if data.TaskSessionID == "" { + s.logger.Warn("missing session_id for permission_request", + zap.String("task_id", data.TaskID), + zap.String("pending_id", data.PendingID)) + return + } + + s.setSessionWaitingForInput(ctx, data.TaskID, data.TaskSessionID) + + if s.messageCreator != nil { + _, err := s.messageCreator.CreatePermissionRequestMessage( + ctx, + data.TaskID, + data.TaskSessionID, + data.PendingID, + data.ToolCallID, + data.Title, + s.getActiveTurnID(data.TaskSessionID), + data.Options, + data.ActionType, + data.ActionDetails, + ) + if err != nil { + s.logger.Error("failed to create permission request message", + zap.String("task_id", data.TaskID), + zap.String("pending_id", data.PendingID), + zap.Error(err)) + } else { + s.logger.Debug("created permission request message", + zap.String("task_id", data.TaskID), + zap.String("pending_id", data.PendingID)) + } + } +} diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index 8781e8dc55..9bf2048783 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -42,12 +42,6 @@ type AgentManagerClient interface { // StopAgent stops a running agent StopAgent(ctx context.Context, agentExecutionID string, force bool) error - // GetAgentStatus returns the status of an agent execution - GetAgentStatus(ctx context.Context, agentExecutionID string) (*v1.AgentExecution, error) - - // ListAgentTypes returns available agent types - ListAgentTypes(ctx context.Context) ([]*v1.AgentType, error) - // PromptAgent sends a prompt to a running agent // Returns PromptResult indicating if the agent needs input PromptAgent(ctx context.Context, agentExecutionID string, prompt string) (*PromptResult, error) @@ -58,20 +52,10 @@ type AgentManagerClient interface { // RespondToPermission sends a response to a permission request RespondToPermissionBySessionID(ctx context.Context, sessionID, pendingID, optionID string, cancelled bool) error - // GetRecoveredExecutions returns executions recovered from Docker during startup - GetRecoveredExecutions() []RecoveredExecutionInfo - // IsAgentRunningForSession checks if an agent is actually running for a session // This probes the actual agent (Docker container or standalone process) rather than relying on cached state IsAgentRunningForSession(ctx context.Context, sessionID string) bool - // CleanupStaleExecutionBySessionID removes a stale agent execution from tracking without trying to stop it. - // This is used when we detect the agent process has stopped but the execution is still tracked. - CleanupStaleExecutionBySessionID(ctx context.Context, sessionID string) error - - // GetAgentType returns the agent type configuration - GetAgentType(ctx context.Context, agentID string) (*v1.AgentType, error) - // ResolveAgentProfile resolves an agent profile ID to profile information ResolveAgentProfile(ctx context.Context, profileID string) (*AgentProfileInfo, error) } @@ -197,11 +181,6 @@ func NewExecutor(agentManager AgentManagerClient, repo repository.Repository, lo } } -// SetWorktreeEnabled enables or disables worktree mode -func (e *Executor) SetWorktreeEnabled(enabled bool) { - e.worktreeEnabled = enabled -} - func (e *Executor) applyPreferredShellEnv(ctx context.Context, env map[string]string) map[string]string { if e.shellPrefs == nil { return env @@ -222,15 +201,6 @@ func (e *Executor) applyPreferredShellEnv(ctx context.Context, env map[string]st return env } -// RecoveredExecutionInfo contains info about an execution recovered from Docker -type RecoveredExecutionInfo struct { - ExecutionID string - TaskID string - SessionID string - ContainerID string - AgentProfileID string -} - // Execute starts agent execution for a task func (e *Executor) Execute(ctx context.Context, task *v1.Task) (*TaskExecution, error) { return e.ExecuteWithProfile(ctx, task, "", task.Description) @@ -979,30 +949,6 @@ func (e *Executor) RespondToPermission(ctx context.Context, sessionID, pendingID return e.agentManager.RespondToPermissionBySessionID(ctx, sessionID, pendingID, optionID, cancelled) } -// GetExecution returns the current execution state for a task (returns most recent active session) -func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) { - ctx := context.Background() - const startupGracePeriod = 30 * time.Second - - // Load from database - session, err := e.repo.GetActiveTaskSessionByTaskID(ctx, taskID) - if err != nil { - return nil, false - } - - // Verify the agent is actually running by probing the lifecycle manager - // This handles the case where backend was restarted and DB has stale "running" sessions - if session.ID == "" || !e.agentManager.IsAgentRunningForSession(ctx, session.ID) { - if (session.State == models.TaskSessionStateStarting || session.State == models.TaskSessionStateRunning) && - time.Since(session.UpdatedAt) < startupGracePeriod { - return FromTaskSession(session), true - } - return nil, false - } - - return FromTaskSession(session), true -} - // GetExecutionBySession returns the execution state for a specific session func (e *Executor) GetExecutionBySession(sessionID string) (*TaskExecution, bool) { ctx := context.Background() @@ -1029,33 +975,6 @@ func (e *Executor) GetExecutionBySession(sessionID string) (*TaskExecution, bool return FromTaskSession(session), true } -// GetExecutionWithContext returns the current execution state for a task with context (returns most recent active session) -// This is a read-only check - it does NOT cancel sessions as a side effect. -func (e *Executor) GetExecutionWithContext(ctx context.Context, taskID string) (*TaskExecution, bool) { - const startupGracePeriod = 30 * time.Second - - // Load from database - session, err := e.repo.GetActiveTaskSessionByTaskID(ctx, taskID) - if err != nil { - return nil, false - } - if session.ID == "" { - return nil, false - } - - // Verify the agent is actually running by probing the lifecycle manager - if !e.agentManager.IsAgentRunningForSession(ctx, session.ID) { - // Allow a grace period for sessions that are starting up - if (session.State == models.TaskSessionStateStarting || session.State == models.TaskSessionStateRunning) && - time.Since(session.UpdatedAt) < startupGracePeriod { - return FromTaskSession(session), true - } - return nil, false - } - - return FromTaskSession(session), true -} - // ListExecutions returns all active executions func (e *Executor) ListExecutions() []*TaskExecution { ctx := context.Background() @@ -1087,29 +1006,6 @@ func (e *Executor) CanExecute() bool { return true } -// UpdateProgressBySession updates the progress of an execution by session ID -func (e *Executor) UpdateProgressBySession(ctx context.Context, sessionID string, progress int, state v1.TaskSessionState) { - e.logger.Debug("updating execution progress", - zap.String("session_id", sessionID), - zap.Int("progress", progress), - zap.String("state", string(state))) - - // Update database asynchronously (don't block on this) - go func() { - session, err := e.repo.GetTaskSession(ctx, sessionID) - if err != nil { - return - } - session.Progress = progress - session.State = models.TaskSessionState(state) - if err := e.repo.UpdateTaskSession(ctx, session); err != nil { - e.logger.Error("failed to update agent session progress in database", - zap.String("session_id", sessionID), - zap.Error(err)) - } - }() -} - // MarkCompletedBySession marks an execution as completed by session ID func (e *Executor) MarkCompletedBySession(ctx context.Context, sessionID string, state v1.TaskSessionState) { e.logger.Info("execution completed", @@ -1164,141 +1060,3 @@ func cloneMetadata(src map[string]interface{}) map[string]interface{} { } return out } - -// MockAgentManagerClient is a placeholder implementation -type MockAgentManagerClient struct { - logger *logger.Logger -} - -// NewMockAgentManagerClient creates a new mock agent manager client -func NewMockAgentManagerClient(log *logger.Logger) *MockAgentManagerClient { - return &MockAgentManagerClient{ - logger: log.WithFields(zap.String("component", "mock_agent_manager")), - } -} - -// LaunchAgent mocks launching an agent container -func (m *MockAgentManagerClient) LaunchAgent(ctx context.Context, req *LaunchAgentRequest) (*LaunchAgentResponse, error) { - m.logger.Info("mock: launching agent", - zap.String("task_id", req.TaskID), - zap.String("agent_profile_id", req.AgentProfileID), - zap.String("repository_url", req.RepositoryURL), - zap.String("branch", req.Branch)) - - return &LaunchAgentResponse{ - AgentExecutionID: uuid.New().String(), - ContainerID: "mock-container-" + uuid.New().String()[:8], - Status: v1.AgentStatusStarting, - }, nil -} - -// StartAgentProcess mocks starting the agent subprocess -func (m *MockAgentManagerClient) StartAgentProcess(ctx context.Context, agentExecutionID string) error { - m.logger.Info("mock: starting agent process", - zap.String("agent_execution_id", agentExecutionID)) - return nil -} - -// StopAgent mocks stopping an agent -func (m *MockAgentManagerClient) StopAgent(ctx context.Context, agentExecutionID string, force bool) error { - m.logger.Info("mock: stopping agent", - zap.String("agent_execution_id", agentExecutionID), - zap.Bool("force", force)) - return nil -} - -// GetAgentStatus mocks getting agent status -func (m *MockAgentManagerClient) GetAgentStatus(ctx context.Context, agentExecutionID string) (*v1.AgentExecution, error) { - m.logger.Info("mock: getting agent status", - zap.String("agent_execution_id", agentExecutionID)) - - return &v1.AgentExecution{ - ID: agentExecutionID, - Status: v1.AgentStatusRunning, - AgentProfileID: "mock-agent", - }, nil -} - -// ListAgentTypes mocks listing available agent types -func (m *MockAgentManagerClient) ListAgentTypes(ctx context.Context) ([]*v1.AgentType, error) { - m.logger.Info("mock: listing agent types") - - return []*v1.AgentType{ - { - ID: "auggie-cli", - Name: "Auggie CLI Agent", - Description: "CLI-based agent for code tasks", - DockerImage: "auggie-cli", - DockerTag: "latest", - Enabled: true, - }, - }, nil -} - -// PromptAgent mocks sending a prompt to an agent -func (m *MockAgentManagerClient) PromptAgent(ctx context.Context, agentExecutionID string, prompt string) (*PromptResult, error) { - m.logger.Info("mock: prompting agent", - zap.String("agent_execution_id", agentExecutionID), - zap.Int("prompt_length", len(prompt))) - return &PromptResult{StopReason: "end_turn"}, nil -} - -// RespondToPermissionBySessionID mocks responding to a permission request -func (m *MockAgentManagerClient) RespondToPermissionBySessionID(ctx context.Context, sessionID, pendingID, optionID string, cancelled bool) error { - m.logger.Info("mock: responding to permission", - zap.String("session_id", sessionID), - zap.String("pending_id", pendingID), - zap.String("option_id", optionID), - zap.Bool("cancelled", cancelled)) - return nil -} - -// CancelAgent mocks cancelling the current agent turn -func (m *MockAgentManagerClient) CancelAgent(ctx context.Context, sessionID string) error { - m.logger.Info("mock: cancelling agent turn", - zap.String("session_id", sessionID)) - return nil -} - -// GetRecoveredExecutions mocks getting recovered executions (returns empty for mock) -func (m *MockAgentManagerClient) GetRecoveredExecutions() []RecoveredExecutionInfo { - return nil -} - -// IsAgentRunningForSession mocks checking if an agent is running (always returns false for mock) -func (m *MockAgentManagerClient) IsAgentRunningForSession(ctx context.Context, sessionID string) bool { - return false -} - -// CleanupStaleExecutionBySessionID mocks cleaning up a stale execution (no-op for mock) -func (m *MockAgentManagerClient) CleanupStaleExecutionBySessionID(ctx context.Context, sessionID string) error { - return nil -} - -// GetAgentType mocks getting an agent type configuration -func (m *MockAgentManagerClient) GetAgentType(ctx context.Context, agentID string) (*v1.AgentType, error) { - m.logger.Info("mock: getting agent type", - zap.String("agent_id", agentID)) - - return &v1.AgentType{ - ID: agentID, - Name: "Mock Agent", - Description: "Mock agent type for testing", - Enabled: true, - }, nil -} - -// ResolveAgentProfile mocks resolving an agent profile -func (m *MockAgentManagerClient) ResolveAgentProfile(ctx context.Context, profileID string) (*AgentProfileInfo, error) { - m.logger.Info("mock: resolving agent profile", - zap.String("profile_id", profileID)) - - return &AgentProfileInfo{ - ProfileID: profileID, - ProfileName: "Mock Profile", - AgentID: "mock-agent", - AgentName: "mock", - Model: "mock-model", - AutoApprove: false, - }, nil -} diff --git a/apps/backend/internal/orchestrator/handlers/handlers.go b/apps/backend/internal/orchestrator/handlers/handlers.go index 7374647427..90d5c47bcc 100644 --- a/apps/backend/internal/orchestrator/handlers/handlers.go +++ b/apps/backend/internal/orchestrator/handlers/handlers.go @@ -35,7 +35,6 @@ func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { d.RegisterFunc(ws.ActionOrchestratorPrompt, h.wsPromptTask) d.RegisterFunc(ws.ActionOrchestratorComplete, h.wsCompleteTask) d.RegisterFunc(ws.ActionPermissionRespond, h.wsRespondToPermission) - d.RegisterFunc(ws.ActionTaskExecution, h.wsGetTaskExecution) d.RegisterFunc(ws.ActionTaskSessionResume, h.wsResumeTaskSession) d.RegisterFunc(ws.ActionTaskSessionStatus, h.wsGetTaskSessionStatus) d.RegisterFunc(ws.ActionAgentCancel, h.wsCancelAgent) @@ -269,30 +268,6 @@ func (h *Handlers) wsRespondToPermission(ctx context.Context, msg *ws.Message) ( return ws.NewResponse(msg.ID, msg.Action, resp) } -type wsGetTaskExecutionRequest struct { - TaskID string `json:"task_id"` -} - -func (h *Handlers) wsGetTaskExecution(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - var req wsGetTaskExecutionRequest - if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid request payload", nil) - } - - if req.TaskID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "task_id is required", nil) - } - - resp, err := h.controller.GetTaskExecution(ctx, dto.GetTaskExecutionRequest{ - TaskID: req.TaskID, - }) - if err != nil { - h.logger.Error("failed to get task execution", zap.String("task_id", req.TaskID), zap.Error(err)) - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to get task execution: "+err.Error(), nil) - } - return ws.NewResponse(msg.ID, msg.Action, resp) -} - type wsGetTaskSessionStatusRequest struct { TaskID string `json:"task_id"` TaskSessionID string `json:"session_id"` diff --git a/apps/backend/internal/orchestrator/queue/queue.go b/apps/backend/internal/orchestrator/queue/queue.go index 2b446e30f1..7819564da1 100644 --- a/apps/backend/internal/orchestrator/queue/queue.go +++ b/apps/backend/internal/orchestrator/queue/queue.go @@ -121,17 +121,6 @@ func (q *TaskQueue) Dequeue() *QueuedTask { return qt } -// Peek returns the highest priority task without removing it -func (q *TaskQueue) Peek() *QueuedTask { - q.mu.RLock() - defer q.mu.RUnlock() - - if len(q.heap) == 0 { - return nil - } - return q.heap[0] -} - // Remove removes a specific task from the queue func (q *TaskQueue) Remove(taskID string) bool { q.mu.Lock() @@ -147,30 +136,6 @@ func (q *TaskQueue) Remove(taskID string) bool { return true } -// UpdatePriority updates the priority of a task in the queue -func (q *TaskQueue) UpdatePriority(taskID string, newPriority int) bool { - q.mu.Lock() - defer q.mu.Unlock() - - qt, exists := q.taskMap[taskID] - if !exists { - return false - } - - qt.Priority = newPriority - heap.Fix(&q.heap, qt.index) - return true -} - -// Contains checks if a task is in the queue -func (q *TaskQueue) Contains(taskID string) bool { - q.mu.RLock() - defer q.mu.RUnlock() - - _, exists := q.taskMap[taskID] - return exists -} - // Len returns the number of tasks in the queue func (q *TaskQueue) Len() int { q.mu.RLock() @@ -196,13 +161,3 @@ func (q *TaskQueue) List() []*QueuedTask { copy(result, q.heap) return result } - -// Clear removes all tasks from the queue -func (q *TaskQueue) Clear() { - q.mu.Lock() - defer q.mu.Unlock() - - q.heap = make(taskHeap, 0) - q.taskMap = make(map[string]*QueuedTask) - heap.Init(&q.heap) -} diff --git a/apps/backend/internal/orchestrator/queue/queue_test.go b/apps/backend/internal/orchestrator/queue/queue_test.go index 4da52eca35..882488aa5a 100644 --- a/apps/backend/internal/orchestrator/queue/queue_test.go +++ b/apps/backend/internal/orchestrator/queue/queue_test.go @@ -121,30 +121,6 @@ func TestPriorityOrdering(t *testing.T) { } } -func TestPeek(t *testing.T) { - q := NewTaskQueue(10) - - // Peek on empty queue - peeked := q.Peek() - if peeked != nil { - t.Errorf("expected nil from Peek on empty queue") - } - - _ = q.Enqueue(createTestTask("task-1", 5)) - peeked = q.Peek() - - if peeked == nil { - t.Fatal("Peek returned nil on non-empty queue") - } - if peeked.TaskID != "task-1" { - t.Errorf("expected TaskID = task-1, got %s", peeked.TaskID) - } - // Verify it didn't remove the task - if q.Len() != 1 { - t.Errorf("Peek should not remove task from queue") - } -} - func TestRemove(t *testing.T) { q := NewTaskQueue(10) @@ -158,7 +134,8 @@ func TestRemove(t *testing.T) { if q.Len() != 1 { t.Errorf("expected Len() = 1 after remove, got %d", q.Len()) } - if q.Contains("task-1") { + // Verify task was removed by trying to remove again + if q.Remove("task-1") { t.Error("queue should not contain removed task") } } @@ -171,52 +148,6 @@ func TestRemoveNonExistent(t *testing.T) { } } -func TestUpdatePriority(t *testing.T) { - q := NewTaskQueue(10) - - _ = q.Enqueue(createTestTask("task-1", 1)) - _ = q.Enqueue(createTestTask("task-2", 10)) - - // task-2 should be first initially - peeked := q.Peek() - if peeked.TaskID != "task-2" { - t.Errorf("expected task-2 first initially") - } - - // Update task-1 to have higher priority - updated := q.UpdatePriority("task-1", 20) - if !updated { - t.Error("UpdatePriority should return true for existing task") - } - - // Now task-1 should be first - peeked = q.Peek() - if peeked.TaskID != "task-1" { - t.Errorf("expected task-1 first after priority update") - } -} - -func TestUpdatePriorityNonExistent(t *testing.T) { - q := NewTaskQueue(10) - updated := q.UpdatePriority("non-existent", 10) - if updated { - t.Error("UpdatePriority should return false for non-existent task") - } -} - -func TestContains(t *testing.T) { - q := NewTaskQueue(10) - - _ = q.Enqueue(createTestTask("task-1", 5)) - - if !q.Contains("task-1") { - t.Error("Contains should return true for existing task") - } - if q.Contains("task-2") { - t.Error("Contains should return false for non-existent task") - } -} - func TestIsFull(t *testing.T) { q := NewTaskQueue(2) @@ -248,21 +179,6 @@ func TestList(t *testing.T) { } } -func TestClear(t *testing.T) { - q := NewTaskQueue(10) - - _ = q.Enqueue(createTestTask("task-1", 5)) - _ = q.Enqueue(createTestTask("task-2", 3)) - - q.Clear() - if q.Len() != 0 { - t.Errorf("expected Len() = 0 after Clear, got %d", q.Len()) - } - if q.Contains("task-1") || q.Contains("task-2") { - t.Error("Clear should remove all tasks") - } -} - func TestUnlimitedQueue(t *testing.T) { // maxSize of 0 means unlimited q := NewTaskQueue(0) diff --git a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go index 84be87b547..77e942c959 100644 --- a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go +++ b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go @@ -51,17 +51,6 @@ func (m *mockAgentManager) StopAgent(ctx context.Context, agentExecutionID strin return nil } -func (m *mockAgentManager) GetAgentStatus(ctx context.Context, agentExecutionID string) (*v1.AgentExecution, error) { - return &v1.AgentExecution{ - ID: agentExecutionID, - Status: v1.AgentStatusRunning, - }, nil -} - -func (m *mockAgentManager) ListAgentTypes(ctx context.Context) ([]*v1.AgentType, error) { - return []*v1.AgentType{}, nil -} - func (m *mockAgentManager) PromptAgent(ctx context.Context, agentExecutionID string, prompt string) (*executor.PromptResult, error) { return &executor.PromptResult{StopReason: "end_turn"}, nil } @@ -70,30 +59,14 @@ func (m *mockAgentManager) RespondToPermissionBySessionID(ctx context.Context, s return nil } -func (m *mockAgentManager) GetRecoveredExecutions() []executor.RecoveredExecutionInfo { - return nil -} - func (m *mockAgentManager) IsAgentRunningForSession(ctx context.Context, sessionID string) bool { return false } -func (m *mockAgentManager) CleanupStaleExecutionBySessionID(ctx context.Context, sessionID string) error { - return nil -} - func (m *mockAgentManager) CancelAgent(ctx context.Context, sessionID string) error { return nil } -func (m *mockAgentManager) GetAgentType(ctx context.Context, agentID string) (*v1.AgentType, error) { - return &v1.AgentType{ - ID: agentID, - Name: "Mock Agent", - Enabled: true, - }, nil -} - func (m *mockAgentManager) ResolveAgentProfile(ctx context.Context, profileID string) (*executor.AgentProfileInfo, error) { return &executor.AgentProfileInfo{ ProfileID: profileID, @@ -402,10 +375,10 @@ func TestPriorityOrdering(t *testing.T) { _ = s.EnqueueTask(createTestTask("high", 10)) _ = s.EnqueueTask(createTestTask("medium", 5)) - // Verify the underlying queue has correct ordering - peeked := q.Peek() - if peeked.TaskID != "high" { - t.Errorf("expected highest priority task (high) first, got %s", peeked.TaskID) + // Verify the underlying queue has correct ordering by dequeueing + first := q.Dequeue() + if first == nil || first.TaskID != "high" { + t.Errorf("expected highest priority task (high) first, got %v", first) } } diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 407dea55b5..21d0b1eb44 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -4,25 +4,19 @@ package orchestrator import ( "context" "errors" - "fmt" - "os" "sync" "time" "go.uber.org/zap" - "github.com/kandev/kandev/internal/agent/lifecycle" "github.com/kandev/kandev/internal/common/logger" - "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" "github.com/kandev/kandev/internal/orchestrator/scheduler" "github.com/kandev/kandev/internal/orchestrator/watcher" "github.com/kandev/kandev/internal/task/models" "github.com/kandev/kandev/internal/task/repository" - v1 "github.com/kandev/kandev/pkg/api/v1" ) // Common errors @@ -82,11 +76,7 @@ type Service struct { queue *queue.TaskQueue executor *executor.Executor scheduler *scheduler.Scheduler - watcher *watcher.Watcher - - // Agent stream event handlers (for WebSocket streaming) - streamHandlers []func(payload *lifecycle.AgentStreamEventPayload) - streamMu sync.RWMutex + watcher *watcher.Watcher // Message creator for saving agent responses messageCreator MessageCreator @@ -148,15 +138,12 @@ func NewService( repo: repo, agentManager: agentManager, queue: taskQueue, - executor: exec, - scheduler: sched, - streamHandlers: make([]func(payload *lifecycle.AgentStreamEventPayload), 0), + executor: exec, + scheduler: sched, } // Create the watcher with event handlers that wire everything together handlers := watcher.EventHandlers{ - OnTaskStateChanged: s.handleTaskStateChanged, - OnAgentReady: s.handleAgentReady, OnAgentCompleted: s.handleAgentCompleted, OnAgentFailed: s.handleAgentFailed, OnAgentStreamEvent: s.handleAgentStreamEvent, @@ -440,18 +427,6 @@ func (s *Service) resumeExecutorsOnStartup(ctx context.Context) { } } -func validateSessionWorktrees(session *models.TaskSession) error { - for _, wt := range session.Worktrees { - if wt.WorktreePath == "" { - continue - } - if _, err := os.Stat(wt.WorktreePath); err != nil { - return fmt.Errorf("worktree path not found: %w", err) - } - } - return nil -} - // IsRunning returns true if the service is running func (s *Service) IsRunning() bool { s.mu.RLock() @@ -484,1038 +459,4 @@ func (s *Service) GetStatus() *Status { } } -// EnqueueTask manually adds a task to the queue -func (s *Service) EnqueueTask(ctx context.Context, task *v1.Task) error { - s.logger.Debug("manually enqueueing task", - zap.String("task_id", task.ID), - zap.String("title", task.Title)) - return s.scheduler.EnqueueTask(task) -} -// StartTask manually starts agent execution for a task -func (s *Service) StartTask(ctx context.Context, taskID string, agentProfileID string, priority int, prompt string) (*executor.TaskExecution, error) { - s.logger.Debug("manually starting task", - zap.String("task_id", taskID), - zap.String("agent_profile_id", agentProfileID), - zap.Int("priority", priority), - zap.Int("prompt_length", len(prompt))) - - 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)) - } - - // Fetch the task from the repository to get complete task info - task, err := s.scheduler.GetTask(ctx, taskID) - if err != nil { - s.logger.Error("failed to fetch task for manual start", - zap.String("task_id", taskID), - zap.Error(err)) - return nil, err - } - - // Override priority if provided in the request - if priority > 0 { - task.Priority = priority - } - - // Use provided prompt, fall back to task description - effectivePrompt := prompt - if effectivePrompt == "" { - effectivePrompt = task.Description - } - - execution, err := s.executor.ExecuteWithProfile(ctx, task, agentProfileID, effectivePrompt) - if err != nil { - return nil, err - } - - if execution.SessionID != "" { - s.updateTaskSessionState(ctx, taskID, execution.SessionID, models.TaskSessionStateRunning, "", true) - // Create user message for the initial prompt - if s.messageCreator != nil && effectivePrompt != "" { - if err := s.messageCreator.CreateUserMessage(ctx, taskID, effectivePrompt, execution.SessionID, s.getActiveTurnID(execution.SessionID)); err != nil { - s.logger.Error("failed to create initial user message", - zap.String("task_id", taskID), - zap.Error(err)) - } - } - } - - if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateInProgress); err != nil { - s.logger.Warn("failed to update task state to IN_PROGRESS", - zap.String("task_id", taskID), - zap.Error(err)) - } - - return execution, nil -} - -// ResumeTaskSession restarts a specific task session using its stored worktree. -func (s *Service) ResumeTaskSession(ctx context.Context, taskID, sessionID string) (*executor.TaskExecution, error) { - s.logger.Debug("resuming task session", - zap.String("task_id", taskID), - zap.String("session_id", sessionID)) - - if exec, ok := s.executor.GetExecutionWithContext(ctx, taskID); ok && exec != nil { - if exec.SessionID != "" && exec.SessionID != sessionID { - return nil, executor.ErrExecutionAlreadyRunning - } - } - - session, err := s.repo.GetTaskSession(ctx, sessionID) - if err != nil { - return nil, err - } - if session.TaskID != taskID { - return nil, fmt.Errorf("task session does not belong to task") - } - running, err := s.repo.GetExecutorRunningBySessionID(ctx, sessionID) - if err != nil || running == nil || running.ResumeToken == "" || !running.Resumable { - return nil, fmt.Errorf("session is not resumable") - } - if err := validateSessionWorktrees(session); err != nil { - return nil, err - } - - execution, err := s.executor.ResumeSession(ctx, session, true) - if err != nil { - _ = s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateFailed, err.Error()) - return nil, err - } - // Preserve persisted task/session state; resume should not mutate state/columns. - execution.SessionState = v1.TaskSessionState(session.State) - - s.logger.Debug("task session resumed and ready for input", - zap.String("task_id", taskID), - zap.String("session_id", sessionID)) - - return execution, nil -} - -// GetTaskSessionStatus returns the status of a task session including whether it's resumable -func (s *Service) GetTaskSessionStatus(ctx context.Context, taskID, sessionID string) (dto.TaskSessionStatusResponse, error) { - s.logger.Debug("checking task session status", - zap.String("task_id", taskID), - zap.String("session_id", sessionID)) - - resp := dto.TaskSessionStatusResponse{ - SessionID: sessionID, - TaskID: taskID, - } - - // 1. Load session from database - session, err := s.repo.GetTaskSession(ctx, sessionID) - if err != nil { - resp.Error = "session not found" - return resp, nil - } - - if session.TaskID != taskID { - resp.Error = "session does not belong to task" - return resp, nil - } - - resp.State = string(session.State) - resp.AgentProfileID = session.AgentProfileID - - // Extract resume token from executor runtime state. - var resumeToken string - if running, err := s.repo.GetExecutorRunningBySessionID(ctx, sessionID); err == nil && running != nil { - resumeToken = running.ResumeToken - resp.ACPSessionID = resumeToken - if running.Resumable { - resp.IsResumable = true - } - } - - // Extract worktree info - if len(session.Worktrees) > 0 { - wt := session.Worktrees[0] - if wt.WorktreePath != "" { - resp.WorktreePath = &wt.WorktreePath - } - if wt.WorktreeBranch != "" { - resp.WorktreeBranch = &wt.WorktreeBranch - } - } - - // 2. Check if we have an active execution in memory and agent is running - if exec, ok := s.executor.GetExecutionWithContext(ctx, taskID); ok && exec != nil { - resp.IsAgentRunning = true - resp.NeedsResume = false - return resp, nil - } - - // 3. Session can be resumed if it has a resume token - if resumeToken == "" { - resp.IsAgentRunning = false - resp.IsResumable = false - resp.NeedsResume = false - return resp, nil - } - - // 4. Additional validations for resumption - if session.AgentProfileID == "" { - resp.Error = "session missing agent profile" - resp.IsResumable = false - return resp, nil - } - - // Check if worktree exists (if one was used) - if len(session.Worktrees) > 0 && session.Worktrees[0].WorktreePath != "" { - if _, err := os.Stat(session.Worktrees[0].WorktreePath); err != nil { - resp.Error = "worktree not found" - resp.IsResumable = false - return resp, nil - } - } - - resp.IsAgentRunning = false - resp.NeedsResume = true - resp.ResumeReason = "agent_not_running" - - return resp, nil -} - -// StopTask stops agent execution for a task (stops all active sessions for the task) -func (s *Service) StopTask(ctx context.Context, taskID string, reason string, force bool) error { - s.logger.Info("stopping task execution", - zap.String("task_id", taskID), - zap.String("reason", reason), - zap.Bool("force", force)) - return s.executor.StopByTaskID(ctx, taskID, reason, force) -} - -// StopSession stops agent execution for a specific session -func (s *Service) StopSession(ctx context.Context, sessionID string, reason string, force bool) error { - s.logger.Info("stopping session execution", - zap.String("session_id", sessionID), - zap.String("reason", reason), - zap.Bool("force", force)) - return s.executor.Stop(ctx, sessionID, reason, force) -} - -// PromptResult contains the result of a prompt operation -type PromptResult struct { - StopReason string // The reason the agent stopped (e.g., "end_turn") - AgentMessage string // The agent's accumulated response message -} - -// PromptTask sends a follow-up prompt to a running agent for a task session. -func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prompt string, model string) (*PromptResult, error) { - s.logger.Debug("sending prompt to task agent", - zap.String("task_id", taskID), - zap.String("session_id", sessionID), - zap.Int("prompt_length", len(prompt))) - if sessionID == "" { - return nil, fmt.Errorf("session_id is required") - } - - // Check if model switching is requested - if model != "" { - // Get the current session to check current model - session, err := s.repo.GetTaskSession(ctx, sessionID) - if err != nil { - return nil, fmt.Errorf("failed to get session for model switch: %w", err) - } - - // Get current model from agent profile snapshot - if session.AgentProfileSnapshot != nil { - if currentModel, ok := session.AgentProfileSnapshot["model"].(string); ok && currentModel != model { - // Model switch needed - delegate to executor - s.logger.Debug("model switch requested", - zap.String("task_id", taskID), - zap.String("session_id", sessionID), - zap.String("current_model", currentModel), - zap.String("new_model", model)) - - // Start a new turn for this prompt (model switch case) - s.startTurnForSession(ctx, sessionID) - - // SwitchModel will stop agent, rebuild with new model, restart, and send prompt - switchResult, err := s.executor.SwitchModel(ctx, taskID, sessionID, model, prompt) - if err != nil { - return nil, err - } - return &PromptResult{ - StopReason: switchResult.StopReason, - AgentMessage: switchResult.AgentMessage, - }, nil - } - } - } - - s.setSessionRunning(ctx, taskID, sessionID) - // Start a new turn for this prompt - s.startTurnForSession(ctx, sessionID) - result, err := s.executor.Prompt(ctx, taskID, sessionID, prompt) - if err != nil { - return nil, err - } - return &PromptResult{ - StopReason: result.StopReason, - AgentMessage: result.AgentMessage, - }, nil -} - -// RespondToPermission sends a response to a permission request for a session -func (s *Service) RespondToPermission(ctx context.Context, sessionID, pendingID, optionID string, cancelled bool) error { - s.logger.Debug("responding to permission request", - zap.String("session_id", sessionID), - zap.String("pending_id", pendingID), - zap.String("option_id", optionID), - zap.Bool("cancelled", cancelled)) - - // Respond to the permission via agentctl - if err := s.executor.RespondToPermission(ctx, sessionID, pendingID, optionID, cancelled); err != nil { - return err - } - - // Determine status based on response - status := "approved" - if cancelled { - status = "rejected" - } - - // Update the permission message with the new status - if s.messageCreator != nil { - if err := s.messageCreator.UpdatePermissionMessage(ctx, sessionID, pendingID, status); err != nil { - s.logger.Warn("failed to update permission message status", - zap.String("session_id", sessionID), - zap.String("pending_id", pendingID), - zap.String("status", status), - zap.Error(err)) - // Don't fail the whole operation if message update fails - } - } - - if !cancelled { - session, err := s.repo.GetTaskSession(ctx, sessionID) - if err != nil { - s.logger.Warn("failed to load task session after permission response", - zap.String("session_id", sessionID), - zap.Error(err)) - return nil - } - s.setSessionRunning(ctx, session.TaskID, sessionID) - } - - return nil -} - -// CancelAgent interrupts the current agent turn without terminating the process, -// allowing the user to send a new prompt. -func (s *Service) CancelAgent(ctx context.Context, sessionID string) error { - s.logger.Debug("cancelling agent turn", zap.String("session_id", sessionID)) - - // Fetch session for state updates and message creation - session, err := s.repo.GetTaskSession(ctx, sessionID) - if err != nil { - s.logger.Warn("failed to get session for cancel", - zap.String("session_id", sessionID), - zap.Error(err)) - } - - if err := s.agentManager.CancelAgent(ctx, sessionID); err != nil { - return fmt.Errorf("cancel agent: %w", err) - } - - // Transition to WAITING_FOR_INPUT so the user can send a new prompt - if session != nil { - s.updateTaskSessionState(ctx, session.TaskID, sessionID, models.TaskSessionStateWaitingForInput, "", true) - } - - // Record cancellation in the message history - if s.messageCreator != nil && session != nil { - metadata := map[string]interface{}{ - "cancelled": true, - "variant": "warning", - } - if err := s.messageCreator.CreateSessionMessage( - ctx, - session.TaskID, - "Turn cancelled by user", - sessionID, - string(v1.MessageTypeStatus), - s.getActiveTurnID(sessionID), - metadata, - false, - ); err != nil { - s.logger.Warn("failed to create cancel message", - zap.String("session_id", sessionID), - zap.Error(err)) - } - } - - // Complete the turn since the agent was cancelled - s.completeTurnForSession(ctx, sessionID) - - s.logger.Debug("agent turn cancelled", zap.String("session_id", sessionID)) - return nil -} - -// CompleteTask explicitly completes a task and stops all its agents -func (s *Service) CompleteTask(ctx context.Context, taskID string) error { - s.logger.Info("completing task", - zap.String("task_id", taskID)) - - // Stop all agents for this task (which will trigger AgentCompleted events and update session states) - if err := s.executor.StopByTaskID(ctx, taskID, "task completed by user", false); err != nil { - // If agents are already stopped, just update the task state directly - s.logger.Warn("failed to stop agents, updating task state directly", - zap.String("task_id", taskID), - zap.Error(err)) - } - - // Update task state to COMPLETED - if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateCompleted); err != nil { - return fmt.Errorf("failed to update task state: %w", err) - } - - s.logger.Info("task marked as COMPLETED", - zap.String("task_id", taskID)) - return nil -} - -// GetTaskExecution returns the current execution state for a task -func (s *Service) GetTaskExecution(taskID string) (*executor.TaskExecution, bool) { - return s.executor.GetExecution(taskID) -} - -// GetQueuedTasks returns tasks in the queue -func (s *Service) GetQueuedTasks() []*queue.QueuedTask { - return s.queue.List() -} - -// RegisterStreamHandler registers a handler for agent stream events (used by WebSocket streaming) -func (s *Service) RegisterStreamHandler(handler func(payload *lifecycle.AgentStreamEventPayload)) { - s.streamMu.Lock() - defer s.streamMu.Unlock() - s.streamHandlers = append(s.streamHandlers, handler) - s.logger.Debug("registered stream handler", zap.Int("total_handlers", len(s.streamHandlers))) -} - - - -// broadcastStreamEvent broadcasts an agent stream event to all registered handlers -func (s *Service) broadcastStreamEvent(payload *lifecycle.AgentStreamEventPayload) { - s.streamMu.RLock() - handlers := make([]func(payload *lifecycle.AgentStreamEventPayload), len(s.streamHandlers)) - copy(handlers, s.streamHandlers) - s.streamMu.RUnlock() - - for _, handler := range handlers { - handler(payload) - } -} - -// Event handlers - -// handleTaskStateChanged handles task state change events -func (s *Service) handleTaskStateChanged(ctx context.Context, data watcher.TaskEventData) { - s.logger.Debug("handling task state changed", - zap.String("task_id", data.TaskID)) - - // No auto-enqueue on TODO; sessions are started explicitly with agent profile. -} - -// handleAgentReady handles agent ready events (prompt completed, ready for follow-up) -// Now that both ACP and Codex Prompt() calls are synchronous, this event fires after -// the "complete" stream event. The state transition is already handled by handleAgentStreamEvent -// on the "complete" event, so this handler just logs the event for debugging. -func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventData) { - s.logger.Debug("agent ready event received (state transition handled by complete event)", - zap.String("task_id", data.TaskID), - zap.String("agent_execution_id", data.AgentExecutionID)) -} - -func (s *Service) handleACPSessionCreated(ctx context.Context, data watcher.ACPSessionEventData) { - if data.TaskID == "" || data.ACPSessionID == "" { - return - } - - sessionID := "" - if exec, ok := s.executor.GetExecution(data.TaskID); ok && exec != nil { - sessionID = exec.SessionID - } - if sessionID == "" { - if session, err := s.repo.GetTaskSessionByTaskID(ctx, data.TaskID); err == nil && session != nil { - sessionID = session.ID - } - } - if sessionID == "" { - s.logger.Warn("no task session found to store ACP session id", - zap.String("task_id", data.TaskID)) - return - } - - session, err := s.repo.GetTaskSession(ctx, sessionID) - if err != nil { - s.logger.Warn("failed to load task session for ACP session update", - zap.String("task_id", data.TaskID), - zap.String("session_id", sessionID), - zap.Error(err)) - return - } - - resumable := true - if session.ExecutorID != "" { - if executor, err := s.repo.GetExecutor(ctx, session.ExecutorID); err == nil && executor != nil { - resumable = executor.Resumable - } - } - - running := &models.ExecutorRunning{ - ID: session.ID, - SessionID: session.ID, - TaskID: session.TaskID, - ExecutorID: session.ExecutorID, - Status: "ready", - Resumable: resumable, - ResumeToken: data.ACPSessionID, - AgentExecutionID: session.AgentExecutionID, - ContainerID: session.ContainerID, - } - if len(session.Worktrees) > 0 { - running.WorktreeID = session.Worktrees[0].WorktreeID - running.WorktreePath = session.Worktrees[0].WorktreePath - running.WorktreeBranch = session.Worktrees[0].WorktreeBranch - } - - if err := s.repo.UpsertExecutorRunning(ctx, running); err != nil { - s.logger.Warn("failed to persist resume token for session", - zap.String("task_id", data.TaskID), - zap.String("session_id", sessionID), - zap.Error(err)) - return - } - - s.logger.Debug("stored resume token for task session", - zap.String("task_id", data.TaskID), - zap.String("session_id", sessionID), - zap.String("resume_token", data.ACPSessionID)) -} - -// handleAgentCompleted handles agent completion events -func (s *Service) handleAgentCompleted(ctx context.Context, data watcher.AgentEventData) { - s.logger.Info("handling agent completed", - zap.String("task_id", data.TaskID), - zap.String("agent_execution_id", data.AgentExecutionID)) - - // Update scheduler and remove from queue - s.scheduler.HandleTaskCompleted(data.TaskID, true) - s.scheduler.RemoveTask(data.TaskID) - - // Move task to REVIEW state for user review - // The user can then send a follow-up (moves back to IN_PROGRESS) or mark as COMPLETED - if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateReview); err != nil { - s.logger.Error("failed to update task state to REVIEW", - zap.String("task_id", data.TaskID), - zap.Error(err)) - } else { - s.logger.Info("task moved to REVIEW state after agent completion", - zap.String("task_id", data.TaskID)) - } -} - -// handleAgentFailed handles agent failure events -func (s *Service) handleAgentFailed(ctx context.Context, data watcher.AgentEventData) { - s.logger.Warn("handling agent failed", - zap.String("task_id", data.TaskID), - zap.String("agent_execution_id", data.AgentExecutionID), - zap.String("error_message", data.ErrorMessage)) - - // Trigger retry logic - s.scheduler.HandleTaskCompleted(data.TaskID, false) - if !s.scheduler.RetryTask(data.TaskID) { - s.logger.Error("task failed and retry limit exceeded", - zap.String("task_id", data.TaskID)) - - // Move task to REVIEW state even on failure - user can decide to retry or close - // This maintains the review cycle: user reviews the failure and decides next steps - if err := s.taskRepo.UpdateTaskState(ctx, data.TaskID, v1.TaskStateReview); err != nil { - s.logger.Error("failed to update task state to REVIEW after failure", - zap.String("task_id", data.TaskID), - zap.Error(err)) - } else { - s.logger.Info("task moved to REVIEW state after failure (for user review)", - zap.String("task_id", data.TaskID)) - } - } -} - -// handleAgentStreamEvent handles agent stream events (tool calls, message chunks, etc.) -func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { - if payload == nil || payload.Data == nil { - return - } - - taskID := payload.TaskID - sessionID := payload.SessionID - eventType := payload.Data.Type - - s.logger.Debug("handling agent stream event", - zap.String("task_id", taskID), - zap.String("session_id", sessionID), - zap.String("event_type", eventType)) - - // Handle different event types - switch eventType { - case "message_streaming": - // Handle streaming message events (incremental text updates) - s.handleMessageStreamingEvent(ctx, payload) - - case "tool_call": - // Create tool call message when a tool call starts - // If there's accumulated text, save it as an agent message first - s.saveAgentTextIfPresent(ctx, payload) - s.handleToolCallEvent(ctx, payload) - - case "tool_update": - // Update tool call message when status changes - s.handleToolUpdateEvent(ctx, payload) - - case "complete": - // Save any accumulated text as an agent message - s.saveAgentTextIfPresent(ctx, payload) - // Complete the turn for this session - s.completeTurnForSession(ctx, sessionID) - // Now that both ACP and Codex Prompt() calls are synchronous (return when turn is done), - // we can directly finalize the agent ready state here. - // The "complete" event is now the single source of truth for turn completion. - s.finalizeAgentReady(taskID, sessionID) - - case "error": - // Handle error events - if sessionID != "" && s.messageCreator != nil { - if err := s.messageCreator.CreateSessionMessage( - ctx, - taskID, - payload.Data.Error, - sessionID, - string(v1.MessageTypeError), - s.getActiveTurnID(sessionID), - map[string]interface{}{ - "provider": "agent", - "provider_agent": payload.AgentID, - }, - false, - ); err != nil { - s.logger.Error("failed to create error message", - zap.String("task_id", taskID), - zap.Error(err)) - } - } - // Complete the turn since the agent errored - s.completeTurnForSession(ctx, sessionID) - - case "session_status": - // Handle session status events (resumed vs new session) - if sessionID != "" && s.messageCreator != nil { - var statusMsg string - if payload.Data.SessionStatus == "resumed" { - statusMsg = "Session resumed" - } else { - statusMsg = "New session started" - } - if err := s.messageCreator.CreateSessionMessage( - ctx, - taskID, - statusMsg, - sessionID, - string(v1.MessageTypeStatus), - s.getActiveTurnID(sessionID), - nil, - false, - ); err != nil { - s.logger.Error("failed to create session status message", - zap.String("task_id", taskID), - zap.String("session_id", sessionID), - zap.Error(err)) - } - } - } - - // Broadcast to all registered handlers (WebSocket streaming) - s.broadcastStreamEvent(payload) -} - -// handleToolCallEvent handles tool_call events and creates messages -func (s *Service) handleToolCallEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { - if payload.SessionID == "" { - s.logger.Warn("missing session_id for tool_call", - zap.String("task_id", payload.TaskID), - zap.String("tool_call_id", payload.Data.ToolCallID)) - return - } - - if s.messageCreator != nil { - if err := s.messageCreator.CreateToolCallMessage( - ctx, - payload.TaskID, - payload.Data.ToolCallID, - payload.Data.ToolTitle, - payload.Data.ToolStatus, - payload.SessionID, - s.getActiveTurnID(payload.SessionID), - payload.Data.ToolArgs, - ); err != nil { - s.logger.Error("failed to create tool call message", - zap.String("task_id", payload.TaskID), - zap.String("tool_call_id", payload.Data.ToolCallID), - zap.Error(err)) - } else { - s.logger.Debug("created tool call message", - zap.String("task_id", payload.TaskID), - zap.String("tool_call_id", payload.Data.ToolCallID)) - } - - s.updateTaskSessionState(ctx, payload.TaskID, payload.SessionID, models.TaskSessionStateRunning, "", false) - } -} - -// saveAgentTextIfPresent saves any accumulated agent text as an agent message -func (s *Service) saveAgentTextIfPresent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { - if payload.Data.Text == "" || payload.SessionID == "" { - return - } - - if s.messageCreator != nil { - if err := s.messageCreator.CreateAgentMessage(ctx, payload.TaskID, payload.Data.Text, payload.SessionID, s.getActiveTurnID(payload.SessionID)); err != nil { - s.logger.Error("failed to create agent message", - zap.String("task_id", payload.TaskID), - zap.Error(err)) - } else { - s.logger.Debug("created agent message", - zap.String("task_id", payload.TaskID), - zap.Int("message_length", len(payload.Data.Text))) - } - } -} - -// handleMessageStreamingEvent handles streaming message events for real-time text updates. -// It creates a new message on first chunk (IsAppend=false) or appends to existing (IsAppend=true). -func (s *Service) handleMessageStreamingEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { - if payload.Data.Text == "" || payload.SessionID == "" { - return - } - - if s.messageCreator == nil { - return - } - - // MessageID is pre-generated by the manager to avoid race conditions - messageID := payload.Data.MessageID - if messageID == "" { - s.logger.Warn("streaming message event missing message ID", - zap.String("task_id", payload.TaskID), - zap.String("session_id", payload.SessionID)) - return - } - - if payload.Data.IsAppend { - // Append to existing message - if err := s.messageCreator.AppendAgentMessage(ctx, messageID, payload.Data.Text); err != nil { - s.logger.Error("failed to append to streaming message", - zap.String("task_id", payload.TaskID), - zap.String("message_id", messageID), - zap.Error(err)) - } else { - s.logger.Debug("appended to streaming message", - zap.String("task_id", payload.TaskID), - zap.String("message_id", messageID), - zap.Int("content_length", len(payload.Data.Text))) - } - } else { - // Create new streaming message with the pre-generated ID - if err := s.messageCreator.CreateAgentMessageStreaming(ctx, messageID, payload.TaskID, payload.Data.Text, payload.SessionID, s.getActiveTurnID(payload.SessionID)); err != nil { - s.logger.Error("failed to create streaming message", - zap.String("task_id", payload.TaskID), - zap.String("message_id", messageID), - zap.Error(err)) - } else { - s.logger.Debug("created streaming message", - zap.String("task_id", payload.TaskID), - zap.String("session_id", payload.SessionID), - zap.String("message_id", messageID), - zap.Int("content_length", len(payload.Data.Text))) - } - } -} - -// handleToolUpdateEvent handles tool_update events and updates messages -func (s *Service) handleToolUpdateEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { - if payload.SessionID == "" { - s.logger.Warn("missing session_id for tool_update", - zap.String("task_id", payload.TaskID), - zap.String("tool_call_id", payload.Data.ToolCallID)) - return - } - - // Only update message when tool call completes or errors - switch payload.Data.ToolStatus { - case "complete", "completed", "error", "failed": - if s.messageCreator != nil { - result := "" - if payload.Data.ToolResult != nil { - if str, ok := payload.Data.ToolResult.(string); ok { - result = str - } - } - if err := s.messageCreator.UpdateToolCallMessage( - ctx, - payload.TaskID, - payload.Data.ToolCallID, - payload.Data.ToolStatus, - result, - payload.SessionID, - ); err != nil { - s.logger.Warn("failed to update tool call message", - zap.String("task_id", payload.TaskID), - zap.String("tool_call_id", payload.Data.ToolCallID), - zap.Error(err)) - } - - s.updateTaskSessionState(ctx, payload.TaskID, payload.SessionID, models.TaskSessionStateRunning, "", false) - } - } -} - -func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID string, nextState models.TaskSessionState, errorMessage string, allowWakeFromWaiting bool) { - session, err := s.repo.GetTaskSession(ctx, sessionID) - if err != nil { - return - } - if session.State == models.TaskSessionStateWaitingForInput && nextState == models.TaskSessionStateRunning && !allowWakeFromWaiting { - return - } - oldState := session.State - switch session.State { - case models.TaskSessionStateCompleted, models.TaskSessionStateFailed, models.TaskSessionStateCancelled: - return - } - if session.State == nextState { - return - } - if err := s.repo.UpdateTaskSessionState(ctx, sessionID, nextState, errorMessage); err != nil { - s.logger.Error("failed to update task session state", - zap.String("session_id", sessionID), - zap.String("state", string(nextState)), - zap.Error(err)) - } - s.logger.Debug("task session state updated", - zap.String("task_id", taskID), - zap.String("session_id", sessionID), - zap.String("old_state", string(oldState)), - zap.String("new_state", string(nextState))) - if s.eventBus != nil { - _ = s.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent(events.TaskSessionStateChanged, "task-session", map[string]interface{}{ - "task_id": taskID, - "session_id": sessionID, - "old_state": string(oldState), - "new_state": string(nextState), - "agent_profile_id": session.AgentProfileID, - })) - } -} - -func (s *Service) setSessionWaitingForInput(ctx context.Context, taskID, sessionID string) { - s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateWaitingForInput, "", false) - - if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateReview); err != nil { - s.logger.Error("failed to update task state to REVIEW", - zap.String("task_id", taskID), - zap.Error(err)) - } else { - s.logger.Info("task moved to REVIEW state", - zap.String("task_id", taskID)) - } -} - -func (s *Service) setSessionRunning(ctx context.Context, taskID, sessionID string) { - s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateRunning, "", true) - - if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateInProgress); err != nil { - s.logger.Error("failed to update task state to IN_PROGRESS", - zap.String("task_id", taskID), - zap.Error(err)) - } else { - s.logger.Info("task moved to IN_PROGRESS state", - zap.String("task_id", taskID)) - } -} - -func (s *Service) finalizeAgentReady(taskID, sessionID string) { - ctx := context.Background() - if sessionID == "" { - s.logger.Warn("missing session_id for finalizeAgentReady", - zap.String("task_id", taskID)) - return - } - - s.setSessionWaitingForInput(ctx, taskID, sessionID) -} - -// handleGitStatusUpdated handles git status updates and persists them to agent session metadata -func (s *Service) handleGitStatusUpdated(ctx context.Context, data watcher.GitStatusData) { - s.logger.Debug("handling git status update", - zap.String("task_id", data.TaskID), - zap.String("branch", data.Branch)) - - if data.TaskSessionID == "" { - s.logger.Debug("missing session_id for git status update", - zap.String("task_id", data.TaskID)) - return - } - - session, err := s.repo.GetTaskSession(ctx, data.TaskSessionID) - if err != nil { - s.logger.Debug("no task session for git status update", - zap.String("session_id", data.TaskSessionID), - zap.Error(err)) - return - } - - // Update session metadata with git status - if session.Metadata == nil { - session.Metadata = make(map[string]interface{}) - } - session.Metadata["git_status"] = map[string]interface{}{ - "branch": data.Branch, - "remote_branch": data.RemoteBranch, - "modified": data.Modified, - "added": data.Added, - "deleted": data.Deleted, - "untracked": data.Untracked, - "renamed": data.Renamed, - "ahead": data.Ahead, - "behind": data.Behind, - "files": data.Files, - "timestamp": data.Timestamp, - } - - // Persist to database asynchronously - go func() { - if err := s.repo.UpdateTaskSession(context.Background(), session); err != nil { - s.logger.Error("failed to update agent session with git status", - zap.String("task_id", data.TaskID), - zap.String("session_id", session.ID), - zap.Error(err)) - } else { - s.logger.Debug("persisted git status to agent session", - zap.String("task_id", data.TaskID), - zap.String("session_id", session.ID)) - } - }() -} - -// handleContextWindowUpdated handles context window updates and persists them to session metadata -func (s *Service) handleContextWindowUpdated(ctx context.Context, data watcher.ContextWindowData) { - s.logger.Debug("handling context window update", - zap.String("task_id", data.TaskID), - zap.String("session_id", data.TaskSessionID), - zap.Int64("size", data.ContextWindowSize), - zap.Int64("used", data.ContextWindowUsed)) - - if data.TaskSessionID == "" { - s.logger.Debug("missing session_id for context window update", - zap.String("task_id", data.TaskID)) - return - } - - session, err := s.repo.GetTaskSession(ctx, data.TaskSessionID) - if err != nil { - s.logger.Debug("no task session for context window update", - zap.String("session_id", data.TaskSessionID), - zap.Error(err)) - return - } - - // Update session metadata with context window info - if session.Metadata == nil { - session.Metadata = make(map[string]interface{}) - } - contextWindowData := map[string]interface{}{ - "size": data.ContextWindowSize, - "used": data.ContextWindowUsed, - "remaining": data.ContextWindowRemaining, - "efficiency": data.ContextEfficiency, - } - session.Metadata["context_window"] = contextWindowData - - // Persist to database asynchronously - go func() { - if err := s.repo.UpdateTaskSession(context.Background(), session); err != nil { - s.logger.Error("failed to update session with context window", - zap.String("task_id", data.TaskID), - zap.String("session_id", session.ID), - zap.Error(err)) - } else { - s.logger.Debug("persisted context window to session", - zap.String("task_id", data.TaskID), - zap.String("session_id", session.ID)) - } - }() - - // Broadcast session state change with metadata so frontend can update - // This uses the existing session.state_changed event with metadata included - if s.eventBus != nil { - _ = s.eventBus.Publish(ctx, events.TaskSessionStateChanged, bus.NewEvent( - events.TaskSessionStateChanged, - "orchestrator", - map[string]interface{}{ - "task_id": data.TaskID, - "session_id": session.ID, - "new_state": string(session.State), - "agent_profile_id": session.AgentProfileID, - "metadata": map[string]interface{}{ - "context_window": contextWindowData, - }, - }, - )) - } -} - -// handlePermissionRequest handles permission request events and saves as message -func (s *Service) handlePermissionRequest(ctx context.Context, data watcher.PermissionRequestData) { - s.logger.Debug("handling permission request", - zap.String("task_id", data.TaskID), - zap.String("pending_id", data.PendingID), - zap.String("title", data.Title)) - - if data.TaskSessionID == "" { - s.logger.Warn("missing session_id for permission_request", - zap.String("task_id", data.TaskID), - zap.String("pending_id", data.PendingID)) - return - } - - s.setSessionWaitingForInput(ctx, data.TaskID, data.TaskSessionID) - - if s.messageCreator != nil { - _, err := s.messageCreator.CreatePermissionRequestMessage( - ctx, - data.TaskID, - data.TaskSessionID, - data.PendingID, - data.ToolCallID, - data.Title, - s.getActiveTurnID(data.TaskSessionID), - data.Options, - data.ActionType, - data.ActionDetails, - ) - if err != nil { - s.logger.Error("failed to create permission request message", - zap.String("task_id", data.TaskID), - zap.String("pending_id", data.PendingID), - zap.Error(err)) - } else { - s.logger.Debug("created permission request message", - zap.String("task_id", data.TaskID), - zap.String("pending_id", data.PendingID)) - } - } -} diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go new file mode 100644 index 0000000000..09110b7111 --- /dev/null +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -0,0 +1,423 @@ +// Package orchestrator provides the main orchestrator service that ties all components together. +package orchestrator + +import ( + "context" + "fmt" + "os" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/orchestrator/dto" + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator/queue" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// PromptResult contains the result of a prompt operation +type PromptResult struct { + StopReason string // The reason the agent stopped (e.g., "end_turn") + AgentMessage string // The agent's accumulated response message +} + +func validateSessionWorktrees(session *models.TaskSession) error { + for _, wt := range session.Worktrees { + if wt.WorktreePath == "" { + continue + } + if _, err := os.Stat(wt.WorktreePath); err != nil { + return fmt.Errorf("worktree path not found: %w", err) + } + } + return nil +} + +// EnqueueTask manually adds a task to the queue +func (s *Service) EnqueueTask(ctx context.Context, task *v1.Task) error { + s.logger.Debug("manually enqueueing task", + zap.String("task_id", task.ID), + zap.String("title", task.Title)) + return s.scheduler.EnqueueTask(task) +} + +// StartTask manually starts agent execution for a task +func (s *Service) StartTask(ctx context.Context, taskID string, agentProfileID string, priority int, prompt string) (*executor.TaskExecution, error) { + s.logger.Debug("manually starting task", + zap.String("task_id", taskID), + zap.String("agent_profile_id", agentProfileID), + zap.Int("priority", priority), + zap.Int("prompt_length", len(prompt))) + + 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)) + } + + // Fetch the task from the repository to get complete task info + task, err := s.scheduler.GetTask(ctx, taskID) + if err != nil { + s.logger.Error("failed to fetch task for manual start", + zap.String("task_id", taskID), + zap.Error(err)) + return nil, err + } + + // Override priority if provided in the request + if priority > 0 { + task.Priority = priority + } + + // Use provided prompt, fall back to task description + effectivePrompt := prompt + if effectivePrompt == "" { + effectivePrompt = task.Description + } + + execution, err := s.executor.ExecuteWithProfile(ctx, task, agentProfileID, effectivePrompt) + if err != nil { + return nil, err + } + + if execution.SessionID != "" { + s.updateTaskSessionState(ctx, taskID, execution.SessionID, models.TaskSessionStateRunning, "", true) + // Create user message for the initial prompt + if s.messageCreator != nil && effectivePrompt != "" { + if err := s.messageCreator.CreateUserMessage(ctx, taskID, effectivePrompt, execution.SessionID, s.getActiveTurnID(execution.SessionID)); err != nil { + s.logger.Error("failed to create initial user message", + zap.String("task_id", taskID), + zap.Error(err)) + } + } + } + + if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateInProgress); err != nil { + s.logger.Warn("failed to update task state to IN_PROGRESS", + zap.String("task_id", taskID), + zap.Error(err)) + } + + return execution, nil +} + +// ResumeTaskSession restarts a specific task session using its stored worktree. +func (s *Service) ResumeTaskSession(ctx context.Context, taskID, sessionID string) (*executor.TaskExecution, error) { + s.logger.Debug("resuming task session", + zap.String("task_id", taskID), + zap.String("session_id", sessionID)) + + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + return nil, err + } + if session.TaskID != taskID { + return nil, fmt.Errorf("task session does not belong to task") + } + running, err := s.repo.GetExecutorRunningBySessionID(ctx, sessionID) + if err != nil || running == nil || running.ResumeToken == "" || !running.Resumable { + return nil, fmt.Errorf("session is not resumable") + } + if err := validateSessionWorktrees(session); err != nil { + return nil, err + } + + execution, err := s.executor.ResumeSession(ctx, session, true) + if err != nil { + _ = s.repo.UpdateTaskSessionState(ctx, sessionID, models.TaskSessionStateFailed, err.Error()) + return nil, err + } + // Preserve persisted task/session state; resume should not mutate state/columns. + execution.SessionState = v1.TaskSessionState(session.State) + + s.logger.Debug("task session resumed and ready for input", + zap.String("task_id", taskID), + zap.String("session_id", sessionID)) + + return execution, nil +} + +// GetTaskSessionStatus returns the status of a task session including whether it's resumable +func (s *Service) GetTaskSessionStatus(ctx context.Context, taskID, sessionID string) (dto.TaskSessionStatusResponse, error) { + s.logger.Debug("checking task session status", + zap.String("task_id", taskID), + zap.String("session_id", sessionID)) + + resp := dto.TaskSessionStatusResponse{ + SessionID: sessionID, + TaskID: taskID, + } + + // 1. Load session from database + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + resp.Error = "session not found" + return resp, nil + } + + if session.TaskID != taskID { + resp.Error = "session does not belong to task" + return resp, nil + } + + resp.State = string(session.State) + resp.AgentProfileID = session.AgentProfileID + + // Extract resume token from executor runtime state. + var resumeToken string + if running, err := s.repo.GetExecutorRunningBySessionID(ctx, sessionID); err == nil && running != nil { + resumeToken = running.ResumeToken + resp.ACPSessionID = resumeToken + if running.Resumable { + resp.IsResumable = true + } + } + + // Extract worktree info + if len(session.Worktrees) > 0 { + wt := session.Worktrees[0] + if wt.WorktreePath != "" { + resp.WorktreePath = &wt.WorktreePath + } + if wt.WorktreeBranch != "" { + resp.WorktreeBranch = &wt.WorktreeBranch + } + } + + // 2. Check if this session's agent is running + if exec, ok := s.executor.GetExecutionBySession(sessionID); ok && exec != nil { + resp.IsAgentRunning = true + resp.NeedsResume = false + return resp, nil + } + + // 3. Session can be resumed if it has a resume token + if resumeToken == "" { + resp.IsAgentRunning = false + resp.IsResumable = false + resp.NeedsResume = false + return resp, nil + } + + // 4. Additional validations for resumption + if session.AgentProfileID == "" { + resp.Error = "session missing agent profile" + resp.IsResumable = false + return resp, nil + } + + // Check if worktree exists (if one was used) + if len(session.Worktrees) > 0 && session.Worktrees[0].WorktreePath != "" { + if _, err := os.Stat(session.Worktrees[0].WorktreePath); err != nil { + resp.Error = "worktree not found" + resp.IsResumable = false + return resp, nil + } + } + + resp.IsAgentRunning = false + resp.NeedsResume = true + resp.ResumeReason = "agent_not_running" + + return resp, nil +} + +// StopTask stops agent execution for a task (stops all active sessions for the task) +func (s *Service) StopTask(ctx context.Context, taskID string, reason string, force bool) error { + s.logger.Info("stopping task execution", + zap.String("task_id", taskID), + zap.String("reason", reason), + zap.Bool("force", force)) + return s.executor.StopByTaskID(ctx, taskID, reason, force) +} + +// StopSession stops agent execution for a specific session +func (s *Service) StopSession(ctx context.Context, sessionID string, reason string, force bool) error { + s.logger.Info("stopping session execution", + zap.String("session_id", sessionID), + zap.String("reason", reason), + zap.Bool("force", force)) + return s.executor.Stop(ctx, sessionID, reason, force) +} + +// PromptTask sends a follow-up prompt to a running agent for a task session. +func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prompt string, model string) (*PromptResult, error) { + s.logger.Debug("sending prompt to task agent", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.Int("prompt_length", len(prompt))) + if sessionID == "" { + return nil, fmt.Errorf("session_id is required") + } + + // Check if model switching is requested + if model != "" { + // Get the current session to check current model + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + return nil, fmt.Errorf("failed to get session for model switch: %w", err) + } + + // Get current model from agent profile snapshot + if session.AgentProfileSnapshot != nil { + if currentModel, ok := session.AgentProfileSnapshot["model"].(string); ok && currentModel != model { + // Model switch needed - delegate to executor + s.logger.Debug("model switch requested", + zap.String("task_id", taskID), + zap.String("session_id", sessionID), + zap.String("current_model", currentModel), + zap.String("new_model", model)) + + // Start a new turn for this prompt (model switch case) + s.startTurnForSession(ctx, sessionID) + + // SwitchModel will stop agent, rebuild with new model, restart, and send prompt + switchResult, err := s.executor.SwitchModel(ctx, taskID, sessionID, model, prompt) + if err != nil { + return nil, err + } + return &PromptResult{ + StopReason: switchResult.StopReason, + AgentMessage: switchResult.AgentMessage, + }, nil + } + } + } + + s.setSessionRunning(ctx, taskID, sessionID) + // Start a new turn for this prompt + s.startTurnForSession(ctx, sessionID) + result, err := s.executor.Prompt(ctx, taskID, sessionID, prompt) + if err != nil { + return nil, err + } + return &PromptResult{ + StopReason: result.StopReason, + AgentMessage: result.AgentMessage, + }, nil +} + +// RespondToPermission sends a response to a permission request for a session +func (s *Service) RespondToPermission(ctx context.Context, sessionID, pendingID, optionID string, cancelled bool) error { + s.logger.Debug("responding to permission request", + zap.String("session_id", sessionID), + zap.String("pending_id", pendingID), + zap.String("option_id", optionID), + zap.Bool("cancelled", cancelled)) + + // Respond to the permission via agentctl + if err := s.executor.RespondToPermission(ctx, sessionID, pendingID, optionID, cancelled); err != nil { + return err + } + + // Determine status based on response + status := "approved" + if cancelled { + status = "rejected" + } + + // Update the permission message with the new status + if s.messageCreator != nil { + if err := s.messageCreator.UpdatePermissionMessage(ctx, sessionID, pendingID, status); err != nil { + s.logger.Warn("failed to update permission message status", + zap.String("session_id", sessionID), + zap.String("pending_id", pendingID), + zap.String("status", status), + zap.Error(err)) + // Don't fail the whole operation if message update fails + } + } + + if !cancelled { + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + s.logger.Warn("failed to load task session after permission response", + zap.String("session_id", sessionID), + zap.Error(err)) + return nil + } + s.setSessionRunning(ctx, session.TaskID, sessionID) + } + + return nil +} + +// CancelAgent interrupts the current agent turn without terminating the process, +// allowing the user to send a new prompt. +func (s *Service) CancelAgent(ctx context.Context, sessionID string) error { + s.logger.Debug("cancelling agent turn", zap.String("session_id", sessionID)) + + // Fetch session for state updates and message creation + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil { + s.logger.Warn("failed to get session for cancel", + zap.String("session_id", sessionID), + zap.Error(err)) + } + + if err := s.agentManager.CancelAgent(ctx, sessionID); err != nil { + return fmt.Errorf("cancel agent: %w", err) + } + + // Transition to WAITING_FOR_INPUT so the user can send a new prompt + if session != nil { + s.updateTaskSessionState(ctx, session.TaskID, sessionID, models.TaskSessionStateWaitingForInput, "", true) + } + + // Record cancellation in the message history + if s.messageCreator != nil && session != nil { + metadata := map[string]interface{}{ + "cancelled": true, + "variant": "warning", + } + if err := s.messageCreator.CreateSessionMessage( + ctx, + session.TaskID, + "Turn cancelled by user", + sessionID, + string(v1.MessageTypeStatus), + s.getActiveTurnID(sessionID), + metadata, + false, + ); err != nil { + s.logger.Warn("failed to create cancel message", + zap.String("session_id", sessionID), + zap.Error(err)) + } + } + + // Complete the turn since the agent was cancelled + s.completeTurnForSession(ctx, sessionID) + + s.logger.Debug("agent turn cancelled", zap.String("session_id", sessionID)) + return nil +} + +// CompleteTask explicitly completes a task and stops all its agents +func (s *Service) CompleteTask(ctx context.Context, taskID string) error { + s.logger.Info("completing task", + zap.String("task_id", taskID)) + + // Stop all agents for this task (which will trigger AgentCompleted events and update session states) + if err := s.executor.StopByTaskID(ctx, taskID, "task completed by user", false); err != nil { + // If agents are already stopped, just update the task state directly + s.logger.Warn("failed to stop agents, updating task state directly", + zap.String("task_id", taskID), + zap.Error(err)) + } + + // Update task state to COMPLETED + if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateCompleted); err != nil { + return fmt.Errorf("failed to update task state: %w", err) + } + + s.logger.Info("task marked as COMPLETED", + zap.String("task_id", taskID)) + return nil +} + +// GetQueuedTasks returns tasks in the queue +func (s *Service) GetQueuedTasks() []*queue.QueuedTask { + return s.queue.List() +} + diff --git a/apps/backend/internal/orchestrator/watcher/watcher.go b/apps/backend/internal/orchestrator/watcher/watcher.go index 18b0649feb..350766f6c1 100644 --- a/apps/backend/internal/orchestrator/watcher/watcher.go +++ b/apps/backend/internal/orchestrator/watcher/watcher.go @@ -35,6 +35,7 @@ type AgentEventData struct { // ACPSessionEventData contains data from ACP session events type ACPSessionEventData struct { TaskID string `json:"task_id"` + SessionID string `json:"session_id"` AgentExecutionID string `json:"agent_execution_id"` ACPSessionID string `json:"acp_session_id"` } diff --git a/apps/backend/internal/task/repository/sqlite_test.go b/apps/backend/internal/task/repository/sqlite_test.go index 56d74b398d..81bd0f35d8 100644 --- a/apps/backend/internal/task/repository/sqlite_test.go +++ b/apps/backend/internal/task/repository/sqlite_test.go @@ -6,7 +6,7 @@ import ( "testing" "time" - "github.com/kandev/kandev/internal/agent/worktree" + "github.com/kandev/kandev/internal/worktree" "github.com/kandev/kandev/internal/db" "github.com/kandev/kandev/internal/task/models" v1 "github.com/kandev/kandev/pkg/api/v1" diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index f1a3262743..bc47c2d32c 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -13,7 +13,7 @@ import ( v1 "github.com/kandev/kandev/pkg/api/v1" "go.uber.org/zap" - "github.com/kandev/kandev/internal/agent/worktree" + "github.com/kandev/kandev/internal/worktree" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" diff --git a/apps/backend/internal/task/service/service_test.go b/apps/backend/internal/task/service/service_test.go index e07c5673ac..8150c2cb89 100644 --- a/apps/backend/internal/task/service/service_test.go +++ b/apps/backend/internal/task/service/service_test.go @@ -7,7 +7,7 @@ import ( "testing" "time" - "github.com/kandev/kandev/internal/agent/worktree" + "github.com/kandev/kandev/internal/worktree" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/db" "github.com/kandev/kandev/internal/events/bus" diff --git a/apps/backend/internal/agent/worktree/config.go b/apps/backend/internal/worktree/config.go similarity index 100% rename from apps/backend/internal/agent/worktree/config.go rename to apps/backend/internal/worktree/config.go diff --git a/apps/backend/internal/agent/worktree/errors.go b/apps/backend/internal/worktree/errors.go similarity index 100% rename from apps/backend/internal/agent/worktree/errors.go rename to apps/backend/internal/worktree/errors.go diff --git a/apps/backend/internal/agent/worktree/manager.go b/apps/backend/internal/worktree/manager.go similarity index 100% rename from apps/backend/internal/agent/worktree/manager.go rename to apps/backend/internal/worktree/manager.go diff --git a/apps/backend/internal/agent/worktree/manager_test.go b/apps/backend/internal/worktree/manager_test.go similarity index 100% rename from apps/backend/internal/agent/worktree/manager_test.go rename to apps/backend/internal/worktree/manager_test.go diff --git a/apps/backend/internal/agent/worktree/provider.go b/apps/backend/internal/worktree/provider.go similarity index 100% rename from apps/backend/internal/agent/worktree/provider.go rename to apps/backend/internal/worktree/provider.go diff --git a/apps/backend/internal/agent/worktree/store.go b/apps/backend/internal/worktree/store.go similarity index 100% rename from apps/backend/internal/agent/worktree/store.go rename to apps/backend/internal/worktree/store.go diff --git a/apps/backend/internal/agent/worktree/worktree.go b/apps/backend/internal/worktree/worktree.go similarity index 100% rename from apps/backend/internal/agent/worktree/worktree.go rename to apps/backend/internal/worktree/worktree.go diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index c6b7908aaf..452bd8c502 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -62,7 +62,7 @@ const ( ActionTaskDelete = "task.delete" ActionTaskMove = "task.move" ActionTaskState = "task.state" - ActionTaskExecution = "task.execution" + ActionTaskSessionList = "task.session.list" ActionTaskSessionResume = "task.session.resume" ActionTaskSessionStatus = "task.session.status" diff --git a/apps/web/hooks/use-session-agent.ts b/apps/web/hooks/use-session-agent.ts index b1cb4760de..b10751261b 100644 --- a/apps/web/hooks/use-session-agent.ts +++ b/apps/web/hooks/use-session-agent.ts @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useMemo, useState } from 'react'; +import { useCallback, useMemo, useState } from 'react'; import { getWebSocketClient } from '@/lib/ws/connection'; import { useAppStore } from '@/components/state-provider'; import type { TaskSessionState, Task, TaskSession } from '@/lib/types/http'; @@ -17,13 +17,8 @@ interface UseSessionAgentReturn { } export function useSessionAgent(task: Task | null): UseSessionAgentReturn { - const [isAgentRunning, setIsAgentRunning] = useState(false); const [isAgentLoading, setIsAgentLoading] = useState(false); - const [taskSessionId, setAgentSessionId] = useState(null); - const [taskSessionState, setTaskSessionState] = useState(null); - const [worktreePath, setWorktreePath] = useState(null); - const [worktreeBranch, setWorktreeBranch] = useState(null); - const connectionStatus = useAppStore((state) => state.connection.status); + const activeSessionId = useAppStore((state) => state.tasks.activeSessionId); const activeSession = useAppStore((state) => activeSessionId ? state.taskSessions.items[activeSessionId] ?? null : null @@ -31,79 +26,23 @@ export function useSessionAgent(task: Task | null): UseSessionAgentReturn { const sessionsForTask = useAppStore((state) => task?.id ? state.taskSessionsByTask.itemsByTaskId[task.id] ?? EMPTY_SESSIONS : EMPTY_SESSIONS ); - const wsTaskSessionState = useMemo(() => { - if (!task?.id) return undefined; + + // Derive session from store: prefer active session if it belongs to this task, otherwise first session for task + const currentSession = useMemo(() => { + if (!task?.id) return null; if (activeSession && activeSession.task_id === task.id) { - return activeSession.state; + return activeSession; } - return sessionsForTask[0]?.state; + return sessionsForTask[0] ?? null; }, [activeSession, sessionsForTask, task?.id]); - // Get worktree info from active session - useEffect(() => { - if (activeSession) { - setWorktreePath(activeSession.worktree_path ?? null); - setWorktreeBranch(activeSession.worktree_branch ?? null); - } - }, [activeSession]); + const taskSessionId = currentSession?.id ?? null; + const taskSessionState = currentSession?.state ?? null; + const worktreePath = currentSession?.worktree_path ?? null; + const worktreeBranch = currentSession?.worktree_branch ?? null; - useEffect(() => { - setIsAgentRunning(false); - setAgentSessionId(null); - setTaskSessionState(null); - setWorktreePath(null); - setWorktreeBranch(null); - }, [task?.id]); - - useEffect(() => { - if (!wsTaskSessionState) return; - setTaskSessionState(wsTaskSessionState as TaskSessionState); - }, [wsTaskSessionState]); - - // Fetch task execution status from orchestrator on mount - useEffect(() => { - if (!task?.id) return; - if (connectionStatus !== 'connected') return; - - const checkExecution = async () => { - const client = getWebSocketClient(); - if (!client) return; - - try { - const response = await client.request<{ - has_execution: boolean; - task_id: string; - state?: string; - session_id?: string; - }>('task.execution', { task_id: task.id }); - - if (response.has_execution) { - setIsAgentRunning(true); - if (response.state) { - setTaskSessionState(response.state as TaskSessionState); - } - if (response.session_id) { - setAgentSessionId(response.session_id); - } - } else { - setIsAgentRunning(false); - setAgentSessionId(null); - setTaskSessionState(null); - } - } catch { - // Failed to check task execution - } - }; - - checkExecution(); - const interval = setInterval(() => { - if (connectionStatus === 'connected') { - checkExecution(); - } - }, 2000); - - return () => clearInterval(interval); - }, [connectionStatus, task?.id]); + // Agent is running if session state is STARTING or RUNNING + const isAgentRunning = taskSessionState === 'STARTING' || taskSessionState === 'RUNNING'; const handleStartAgent = useCallback(async (agentProfileId: string, prompt?: string) => { if (!task?.id) return; @@ -114,31 +53,12 @@ export function useSessionAgent(task: Task | null): UseSessionAgentReturn { setIsAgentLoading(true); try { - interface StartResponse { - success: boolean; - task_id: string; - agent_instance_id: string; - session_id?: string; - state: string; - worktree_path?: string; - worktree_branch?: string; - } - const response = await client.request('orchestrator.start', { + // 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); - setIsAgentRunning(true); - setTaskSessionState(response.state as TaskSessionState); - if (response?.session_id) { - setAgentSessionId(response.session_id); - } - - // Update worktree info from response - if (response?.worktree_path) { - setWorktreePath(response.worktree_path); - setWorktreeBranch(response.worktree_branch ?? null); - } } catch { // Failed to start agent } finally { @@ -154,10 +74,8 @@ export function useSessionAgent(task: Task | null): UseSessionAgentReturn { setIsAgentLoading(true); try { + // Store will be updated via WebSocket notifications when session stops await client.request('orchestrator.stop', { task_id: task.id }, 15000); - setIsAgentRunning(false); - setAgentSessionId(null); - setTaskSessionState(null); } catch { // Failed to stop agent } finally { diff --git a/apps/web/hooks/use-task-chat-session.ts b/apps/web/hooks/use-task-chat-session.ts index 730ee0c4c2..d7681b6730 100644 --- a/apps/web/hooks/use-task-chat-session.ts +++ b/apps/web/hooks/use-task-chat-session.ts @@ -1,7 +1,8 @@ -import { useEffect, useMemo, useState } from 'react'; -import { getWebSocketClient } from '@/lib/ws/connection'; -import { useAppStore, useAppStoreApi } from '@/components/state-provider'; -import type { TaskSessionState } from '@/lib/types/http'; +import { useMemo } from 'react'; +import { useAppStore } from '@/components/state-provider'; +import type { TaskSessionState, TaskSession } from '@/lib/types/http'; + +const EMPTY_SESSIONS: TaskSession[] = []; type UseTaskChatSessionReturn = { taskSessionId: string | null; @@ -10,73 +11,17 @@ type UseTaskChatSessionReturn = { }; export function useTaskChatSession(taskId: string | null): UseTaskChatSessionReturn { - const store = useAppStoreApi(); - const [taskSessionIdsByTaskId, setTaskSessionIdsByTaskId] = useState>({}); - const connectionStatus = useAppStore((state) => state.connection.status); - const taskSessionId = useMemo( - () => (taskId ? taskSessionIdsByTaskId[taskId] ?? null : null), - [taskId, taskSessionIdsByTaskId] - ); - const taskSessionState = useAppStore((state) => - taskSessionId ? state.taskSessions.items[taskSessionId]?.state ?? null : null + const sessionsForTask = useAppStore((state) => + taskId ? state.taskSessionsByTask.itemsByTaskId[taskId] ?? EMPTY_SESSIONS : EMPTY_SESSIONS ); - useEffect(() => { - if (!taskId) return; - if (connectionStatus !== 'connected') return; - - const checkExecution = async () => { - const client = getWebSocketClient(); - if (!client) return; - - try { - const response = await client.request<{ - has_execution: boolean; - task_id: string; - state?: string; - session_id?: string; - }>('task.execution', { task_id: taskId }); - - setTaskSessionIdsByTaskId((prev) => { - if (!taskId) return prev; - if (!response.has_execution || !response.session_id) { - if (!prev[taskId]) return prev; - const next = { ...prev }; - delete next[taskId]; - return next; - } - if (prev[taskId] === response.session_id) { - return prev; - } - return { ...prev, [taskId]: response.session_id }; - }); - - // Store partial session object in the store so useSession can access it - if (response.has_execution && response.session_id && response.state) { - store.getState().setTaskSession({ - id: response.session_id, - task_id: taskId, - state: response.state as TaskSessionState, - progress: 0, - started_at: '', - updated_at: '', - }); - } - } catch { - // Failed to check task execution - } - }; - - checkExecution(); - const interval = setInterval(() => { - if (connectionStatus === 'connected') { - checkExecution(); - } - }, 2000); - - return () => clearInterval(interval); - }, [connectionStatus, taskId, store]); + // Get the first (most recent) session for this task from the store + const currentSession = useMemo(() => { + return sessionsForTask[0] ?? null; + }, [sessionsForTask]); + const taskSessionId = currentSession?.id ?? null; + const taskSessionState = currentSession?.state ?? null; const isTaskSessionWorking = taskSessionState === 'STARTING' || taskSessionState === 'RUNNING'; return {