From e012e070a94e9ec1d2d768a4c0d490aaef787e71 Mon Sep 17 00:00:00 2001 From: jcfs Date: Mon, 12 Jan 2026 03:39:02 +0000 Subject: [PATCH] fix: recover agent state from Docker on backend restart - Add Docker reconciliation on lifecycle manager startup to find running containers - Add container_id field to agent_sessions table for cleanup tracking - Add GetRecoveredInstances to AgentManagerClient interface - Sync executor cache with recovered Docker instances on startup - Add task.execution WebSocket endpoint to check agent running state - Fix Stop Agent button showing Start Agent after page refresh - Ensure containers are not leaked when backend restarts --- apps/backend/cmd/kandev/main.go | 15 ++ apps/backend/internal/agent/docker/client.go | 17 +++ .../internal/agent/lifecycle/manager.go | 140 +++++++++++++++++- .../internal/integration/orchestrator_test.go | 5 + .../orchestrator/controller/controller.go | 21 +++ apps/backend/internal/orchestrator/dto/dto.go | 16 ++ .../orchestrator/executor/executor.go | 58 +++++++- .../orchestrator/handlers/handlers.go | 24 +++ .../orchestrator/scheduler/scheduler_test.go | 4 + apps/backend/internal/orchestrator/service.go | 38 +++-- apps/backend/internal/task/models/models.go | 2 + .../internal/task/repository/sqlite.go | 34 +++-- apps/backend/pkg/websocket/actions.go | 15 +- apps/web/app/task/[id]/page-client.tsx | 30 +++- 14 files changed, 373 insertions(+), 46 deletions(-) diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index ab470a2e4f..def5c1c4a6 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -619,6 +619,21 @@ func (a *lifecycleAdapter) RespondToPermissionByTaskID(ctx context.Context, task return a.mgr.RespondToPermissionByTaskID(taskID, pendingID, optionID, cancelled) } +// GetRecoveredInstances returns instances recovered from Docker during startup +func (a *lifecycleAdapter) GetRecoveredInstances() []executor.RecoveredInstanceInfo { + recovered := a.mgr.GetRecoveredInstances() + result := make([]executor.RecoveredInstanceInfo, len(recovered)) + for i, r := range recovered { + result[i] = executor.RecoveredInstanceInfo{ + InstanceID: r.InstanceID, + TaskID: r.TaskID, + ContainerID: r.ContainerID, + AgentType: r.AgentType, + } + } + return result +} + // corsMiddleware returns a CORS middleware for HTTP and WebSocket connections func corsMiddleware() gin.HandlerFunc { return func(c *gin.Context) { diff --git a/apps/backend/internal/agent/docker/client.go b/apps/backend/internal/agent/docker/client.go index 1c78713fa1..3db21451d5 100644 --- a/apps/backend/internal/agent/docker/client.go +++ b/apps/backend/internal/agent/docker/client.go @@ -316,6 +316,23 @@ func (c *Client) GetContainerIP(ctx context.Context, containerID string) (string return "", fmt.Errorf("no IP address found for container %s", containerID) } +// GetContainerLabels returns the labels of a container +func (c *Client) GetContainerLabels(ctx context.Context, containerID string) (map[string]string, error) { + c.logger.Debug("Getting container labels", zap.String("container_id", containerID)) + + inspect, err := c.cli.ContainerInspect(ctx, containerID) + if err != nil { + c.logger.Error("Failed to inspect container for labels", zap.String("container_id", containerID), zap.Error(err)) + return nil, err + } + + if inspect.Config != nil && inspect.Config.Labels != nil { + return inspect.Config.Labels, nil + } + + return make(map[string]string), nil +} + // GetContainerLogs returns logs from a container. func (c *Client) GetContainerLogs(ctx context.Context, containerID string, follow bool, tail string) (io.ReadCloser, error) { c.logger.Debug("Getting container logs", diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 66ef817257..f4ac2955e8 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -114,6 +114,12 @@ func (m *Manager) SetCredentialsManager(credsMgr CredentialsManager) { func (m *Manager) Start(ctx context.Context) error { m.logger.Info("starting lifecycle manager") + // Reconcile with Docker to find running containers from previous runs + if _, err := m.reconcileWithDocker(ctx); err != nil { + m.logger.Warn("failed to reconcile with Docker", zap.Error(err)) + // Continue anyway - this is best-effort + } + // Start cleanup loop m.wg.Add(1) go m.cleanupLoop(ctx) @@ -121,6 +127,131 @@ func (m *Manager) Start(ctx context.Context) error { return nil } +// RecoveredInstance contains info about an instance recovered from Docker +type RecoveredInstance struct { + InstanceID string + TaskID string + ContainerID string + AgentType string +} + +// reconcileWithDocker finds running kandev containers and re-populates tracking maps +// Returns a list of recovered instances so the caller can sync with the database +func (m *Manager) reconcileWithDocker(ctx context.Context) ([]RecoveredInstance, error) { + // Skip reconciliation if docker client is not available (e.g., in tests) + if m.docker == nil { + m.logger.Debug("skipping Docker reconciliation - no docker client") + return nil, nil + } + + m.logger.Info("reconciling with Docker - looking for running containers") + + // List all containers with our label + containers, err := m.docker.ListContainers(ctx, map[string]string{ + "kandev.managed": "true", + }) + if err != nil { + return nil, fmt.Errorf("failed to list containers: %w", err) + } + + m.mu.Lock() + defer m.mu.Unlock() + + var recovered []RecoveredInstance + for _, ctr := range containers { + // Only recover running containers + if ctr.State != "running" { + m.logger.Debug("skipping non-running container", + zap.String("container_id", ctr.ID), + zap.String("state", ctr.State)) + continue + } + + // Get container details to extract labels + labels, err := m.docker.GetContainerLabels(ctx, ctr.ID) + if err != nil { + m.logger.Warn("failed to get container labels", + zap.String("container_id", ctr.ID), + zap.Error(err)) + continue + } + + instanceID := labels["kandev.instance_id"] + taskID := labels["kandev.task_id"] + agentType := labels["kandev.agent_type"] + + if instanceID == "" || taskID == "" { + m.logger.Warn("container missing required labels", + zap.String("container_id", ctr.ID)) + continue + } + + // Get container IP + containerIP, err := m.docker.GetContainerIP(ctx, ctr.ID) + if err != nil { + m.logger.Warn("failed to get container IP", + zap.String("container_id", ctr.ID), + zap.Error(err)) + containerIP = "localhost" + } + + // Create instance record + instance := &AgentInstance{ + ID: instanceID, + TaskID: taskID, + AgentType: agentType, + ContainerID: ctr.ID, + ContainerIP: containerIP, + Status: v1.AgentStatusRunning, + StartedAt: time.Now(), // We don't know exact start time + } + + // Create agentctl client + instance.agentctl = agentctl.NewClient(containerIP, AgentCtlPort, m.logger) + + // Add to tracking maps + m.instances[instanceID] = instance + m.byTask[taskID] = instanceID + m.byContainer[ctr.ID] = instanceID + + recovered = append(recovered, RecoveredInstance{ + InstanceID: instanceID, + TaskID: taskID, + ContainerID: ctr.ID, + AgentType: agentType, + }) + + m.logger.Info("recovered running container", + zap.String("instance_id", instanceID), + zap.String("task_id", taskID), + zap.String("container_id", ctr.ID)) + } + + m.logger.Info("reconciliation complete", + zap.Int("containers_found", len(containers)), + zap.Int("recovered", len(recovered))) + + return recovered, nil +} + +// GetRecoveredInstances returns a snapshot of all currently tracked instances +// This can be used by the orchestrator to sync with the database +func (m *Manager) GetRecoveredInstances() []RecoveredInstance { + m.mu.RLock() + defer m.mu.RUnlock() + + result := make([]RecoveredInstance, 0, len(m.instances)) + for _, inst := range m.instances { + result = append(result, RecoveredInstance{ + InstanceID: inst.ID, + TaskID: inst.TaskID, + ContainerID: inst.ContainerID, + AgentType: inst.AgentType, + }) + } + return result +} + // Stop stops the lifecycle manager func (m *Manager) Stop() error { m.logger.Info("stopping lifecycle manager") @@ -994,13 +1125,20 @@ func (m *Manager) StopAgent(ctx context.Context, instanceID string, force bool) return fmt.Errorf("failed to stop container: %w", err) } - // Update instance status + // Update instance status and remove from tracking maps m.mu.Lock() instance.Status = v1.AgentStatusStopped now := time.Now() instance.FinishedAt = &now + delete(m.instances, instanceID) + delete(m.byTask, instance.TaskID) + delete(m.byContainer, instance.ContainerID) m.mu.Unlock() + m.logger.Info("agent stopped and removed from tracking", + zap.String("instance_id", instanceID), + zap.String("task_id", instance.TaskID)) + // Publish stopped event m.publishEvent(ctx, events.AgentStopped, instance) diff --git a/apps/backend/internal/integration/orchestrator_test.go b/apps/backend/internal/integration/orchestrator_test.go index 0d13c8e4fb..aaa9dc8230 100644 --- a/apps/backend/internal/integration/orchestrator_test.go +++ b/apps/backend/internal/integration/orchestrator_test.go @@ -396,6 +396,11 @@ func (s *SimulatedAgentManagerClient) Close() { close(s.stopCh) } +// GetRecoveredInstances returns recovered instances (none for simulated agent) +func (s *SimulatedAgentManagerClient) GetRecoveredInstances() []executor.RecoveredInstanceInfo { + return nil +} + // ============================================ // ORCHESTRATOR TEST SERVER // ============================================ diff --git a/apps/backend/internal/orchestrator/controller/controller.go b/apps/backend/internal/orchestrator/controller/controller.go index 6b553bcdda..96e7522887 100644 --- a/apps/backend/internal/orchestrator/controller/controller.go +++ b/apps/backend/internal/orchestrator/controller/controller.go @@ -127,3 +127,24 @@ func (c *Controller) RespondToPermission(ctx context.Context, req dto.Permission }, 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, + AgentInstanceID: execution.AgentInstanceID, + AgentType: execution.AgentType, + Status: string(execution.Status), + 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 c63efdbbd3..c2ae4eb53c 100644 --- a/apps/backend/internal/orchestrator/dto/dto.go +++ b/apps/backend/internal/orchestrator/dto/dto.go @@ -116,3 +116,19 @@ func FromQueuedTask(qt interface{ GetTaskID() string; GetPriority() int; GetQueu } } +// 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"` + AgentInstanceID string `json:"agent_instance_id,omitempty"` + AgentType string `json:"agent_type,omitempty"` + Status string `json:"status,omitempty"` + Progress int `json:"progress,omitempty"` + StartedAt string `json:"started_at,omitempty"` +} + diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index afe6253201..e942473b56 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -50,6 +50,9 @@ type AgentManagerClient interface { // RespondToPermission sends a response to a permission request RespondToPermissionByTaskID(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error + + // GetRecoveredInstances returns instances recovered from Docker during startup + GetRecoveredInstances() []RecoveredInstanceInfo } // LaunchAgentRequest contains parameters for launching an agent @@ -192,6 +195,43 @@ func (e *Executor) LoadActiveSessionsFromDB(ctx context.Context) error { return nil } +// RecoveredInstanceInfo contains info about an instance recovered from Docker +type RecoveredInstanceInfo struct { + InstanceID string + TaskID string + ContainerID string + AgentType string +} + +// SyncWithRecoveredInstances ensures the executor's cache is in sync with +// instances recovered from Docker by the lifecycle manager. +// For each recovered instance, if not already in cache, add it. +func (e *Executor) SyncWithRecoveredInstances(ctx context.Context, instances []RecoveredInstanceInfo) { + e.mu.Lock() + defer e.mu.Unlock() + + for _, inst := range instances { + if _, exists := e.executions[inst.TaskID]; exists { + continue // Already have this execution + } + + // Add to cache - this instance is running in Docker but wasn't in DB as active + e.executions[inst.TaskID] = &TaskExecution{ + TaskID: inst.TaskID, + AgentInstanceID: inst.InstanceID, + AgentType: inst.AgentType, + StartedAt: time.Now(), // We don't know exact start time + Status: v1.AgentStatusRunning, + Progress: 0, + LastUpdate: time.Now(), + } + + e.logger.Info("synced recovered instance to executor cache", + zap.String("task_id", inst.TaskID), + zap.String("instance_id", inst.InstanceID)) + } +} + // Execute starts agent execution for a task func (e *Executor) Execute(ctx context.Context, task *v1.Task) (*TaskExecution, error) { // Check if max concurrent limit is reached @@ -243,6 +283,7 @@ func (e *Executor) Execute(ctx context.Context, task *v1.Task) (*TaskExecution, ID: sessionID, TaskID: task.ID, AgentInstanceID: resp.AgentInstanceID, + ContainerID: resp.ContainerID, AgentType: *task.AgentType, Status: v1StatusToAgentSessionStatus(resp.Status), Progress: 0, @@ -297,27 +338,29 @@ func (e *Executor) Stop(ctx context.Context, taskID string, reason string, force err = e.agentManager.StopAgent(ctx, execution.AgentInstanceID, force) if err != nil { - e.logger.Error("failed to stop agent", + // Log the error but continue to clean up execution state + // The agent instance may already be gone (container stopped externally) + e.logger.Warn("failed to stop agent (may already be stopped)", zap.String("task_id", taskID), zap.Error(err)) - return err } - // Update execution status in memory and database + // Update execution status in memory and database regardless of agent stop result e.mu.Lock() if exec, ok := e.executions[taskID]; ok { exec.Status = v1.AgentStatusStopped exec.LastUpdate = time.Now() } + delete(e.executions, taskID) // Remove from cache so GetExecution returns false e.mu.Unlock() // Update database if execution.SessionID != "" { - if err := e.repo.UpdateAgentSessionStatus(ctx, execution.SessionID, models.AgentSessionStatusStopped, reason); err != nil { + if dbErr := e.repo.UpdateAgentSessionStatus(ctx, execution.SessionID, models.AgentSessionStatusStopped, reason); dbErr != nil { e.logger.Error("failed to update agent session status in database", zap.String("task_id", taskID), zap.String("session_id", execution.SessionID), - zap.Error(err)) + zap.Error(dbErr)) } } @@ -651,3 +694,8 @@ func (m *MockAgentManagerClient) RespondToPermissionByTaskID(ctx context.Context zap.Bool("cancelled", cancelled)) return nil } + +// GetRecoveredInstances mocks getting recovered instances (returns empty for mock) +func (m *MockAgentManagerClient) GetRecoveredInstances() []RecoveredInstanceInfo { + return nil +} diff --git a/apps/backend/internal/orchestrator/handlers/handlers.go b/apps/backend/internal/orchestrator/handlers/handlers.go index 6eb70b39d6..0bc68eae77 100644 --- a/apps/backend/internal/orchestrator/handlers/handlers.go +++ b/apps/backend/internal/orchestrator/handlers/handlers.go @@ -35,6 +35,7 @@ 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) } // WS handlers @@ -222,3 +223,26 @@ 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) +} diff --git a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go index 581017dac6..e6a17157ca 100644 --- a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go +++ b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go @@ -64,6 +64,10 @@ func (m *mockAgentManager) RespondToPermissionByTaskID(ctx context.Context, task return nil } +func (m *mockAgentManager) GetRecoveredInstances() []executor.RecoveredInstanceInfo { + return nil +} + // testTaskRepository is an in-memory task repository for testing type testTaskRepository struct { tasks map[string]*v1.Task diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 557a790ee9..a3244d3127 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -53,11 +53,12 @@ type CommentCreator interface { // Service is the main orchestrator service type Service struct { - config ServiceConfig - logger *logger.Logger - eventBus bus.EventBus - db *database.DB - taskRepo scheduler.TaskRepository + config ServiceConfig + logger *logger.Logger + eventBus bus.EventBus + db *database.DB + taskRepo scheduler.TaskRepository + agentManager executor.AgentManagerClient // Components queue *queue.TaskQueue @@ -115,15 +116,16 @@ func NewService( // Create the service (watcher will be created after we have handlers) s := &Service{ - config: cfg, - logger: svcLogger, - eventBus: eventBus, - db: db, - taskRepo: taskRepo, - queue: taskQueue, - executor: exec, - scheduler: sched, - acpHandlers: make([]func(taskID string, msg *protocol.Message), 0), + config: cfg, + logger: svcLogger, + eventBus: eventBus, + db: db, + taskRepo: taskRepo, + agentManager: agentManager, + queue: taskQueue, + executor: exec, + scheduler: sched, + acpHandlers: make([]func(taskID string, msg *protocol.Message), 0), } // Create the watcher with event handlers that wire everything together @@ -164,6 +166,14 @@ func (s *Service) Start(ctx context.Context) error { s.logger.Warn("failed to load active sessions from database (non-fatal)", zap.Error(err)) } + // Sync with instances recovered from Docker by the lifecycle manager + // This ensures we track containers that are running but weren't in the DB as active + recovered := s.agentManager.GetRecoveredInstances() + if len(recovered) > 0 { + s.logger.Info("syncing with recovered Docker instances", zap.Int("count", len(recovered))) + s.executor.SyncWithRecoveredInstances(ctx, recovered) + } + // Start the watcher first to begin receiving events if err := s.watcher.Start(ctx); err != nil { s.mu.Lock() diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index a8e594b342..decc33a9c6 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -148,6 +148,7 @@ type AgentSession struct { ID string `json:"id"` TaskID string `json:"task_id"` AgentInstanceID string `json:"agent_instance_id"` // Docker container/agent instance + ContainerID string `json:"container_id"` // Docker container ID for cleanup AgentType string `json:"agent_type"` // e.g., "augment-agent" ACPSessionID string `json:"acp_session_id"` // ACP protocol session for resumption Status AgentSessionStatus `json:"status"` @@ -166,6 +167,7 @@ func (s *AgentSession) ToAPI() map[string]interface{} { "id": s.ID, "task_id": s.TaskID, "agent_instance_id": s.AgentInstanceID, + "container_id": s.ContainerID, "agent_type": s.AgentType, "acp_session_id": s.ACPSessionID, "status": string(s.Status), diff --git a/apps/backend/internal/task/repository/sqlite.go b/apps/backend/internal/task/repository/sqlite.go index 3d70994bf3..e21e16e1a7 100644 --- a/apps/backend/internal/task/repository/sqlite.go +++ b/apps/backend/internal/task/repository/sqlite.go @@ -214,6 +214,7 @@ func (r *SQLiteRepository) initSchema() error { id TEXT PRIMARY KEY, task_id TEXT NOT NULL, agent_instance_id TEXT NOT NULL DEFAULT '', + container_id TEXT NOT NULL DEFAULT '', agent_type TEXT NOT NULL, acp_session_id TEXT DEFAULT '', status TEXT NOT NULL DEFAULT 'pending', @@ -256,6 +257,11 @@ func (r *SQLiteRepository) initSchema() error { return err } + // Ensure container_id column exists for existing databases + if err := r.ensureColumn("agent_sessions", "container_id", "TEXT NOT NULL DEFAULT ''"); err != nil { + return err + } + if err := r.ensureDefaultWorkspace(); err != nil { return err } @@ -1292,10 +1298,10 @@ func (r *SQLiteRepository) CreateAgentSession(ctx context.Context, session *mode _, err = r.db.ExecContext(ctx, ` INSERT INTO agent_sessions ( - id, task_id, agent_instance_id, agent_type, acp_session_id, status, progress, + id, task_id, agent_instance_id, container_id, agent_type, acp_session_id, status, progress, error_message, metadata, started_at, completed_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - `, session.ID, session.TaskID, session.AgentInstanceID, session.AgentType, session.ACPSessionID, + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, session.ID, session.TaskID, session.AgentInstanceID, session.ContainerID, session.AgentType, session.ACPSessionID, string(session.Status), session.Progress, session.ErrorMessage, string(metadataJSON), session.StartedAt, session.CompletedAt, session.UpdatedAt) @@ -1310,11 +1316,11 @@ func (r *SQLiteRepository) GetAgentSession(ctx context.Context, id string) (*mod var completedAt sql.NullTime err := r.db.QueryRowContext(ctx, ` - SELECT id, task_id, agent_instance_id, agent_type, acp_session_id, status, progress, + SELECT id, task_id, agent_instance_id, container_id, agent_type, acp_session_id, status, progress, error_message, metadata, started_at, completed_at, updated_at FROM agent_sessions WHERE id = ? `, id).Scan( - &session.ID, &session.TaskID, &session.AgentInstanceID, &session.AgentType, + &session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentType, &session.ACPSessionID, &status, &session.Progress, &session.ErrorMessage, &metadataJSON, &session.StartedAt, &completedAt, &session.UpdatedAt, ) @@ -1347,11 +1353,11 @@ func (r *SQLiteRepository) GetAgentSessionByTaskID(ctx context.Context, taskID s var completedAt sql.NullTime err := r.db.QueryRowContext(ctx, ` - SELECT id, task_id, agent_instance_id, agent_type, acp_session_id, status, progress, + SELECT id, task_id, agent_instance_id, container_id, agent_type, acp_session_id, status, progress, error_message, metadata, started_at, completed_at, updated_at FROM agent_sessions WHERE task_id = ? ORDER BY started_at DESC LIMIT 1 `, taskID).Scan( - &session.ID, &session.TaskID, &session.AgentInstanceID, &session.AgentType, + &session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentType, &session.ACPSessionID, &status, &session.Progress, &session.ErrorMessage, &metadataJSON, &session.StartedAt, &completedAt, &session.UpdatedAt, ) @@ -1384,13 +1390,13 @@ func (r *SQLiteRepository) GetActiveAgentSessionByTaskID(ctx context.Context, ta var completedAt sql.NullTime err := r.db.QueryRowContext(ctx, ` - SELECT id, task_id, agent_instance_id, agent_type, acp_session_id, status, progress, + SELECT id, task_id, agent_instance_id, container_id, agent_type, acp_session_id, status, progress, error_message, metadata, started_at, completed_at, updated_at FROM agent_sessions WHERE task_id = ? AND status IN ('pending', 'running', 'waiting') ORDER BY started_at DESC LIMIT 1 `, taskID).Scan( - &session.ID, &session.TaskID, &session.AgentInstanceID, &session.AgentType, + &session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentType, &session.ACPSessionID, &status, &session.Progress, &session.ErrorMessage, &metadataJSON, &session.StartedAt, &completedAt, &session.UpdatedAt, ) @@ -1427,10 +1433,10 @@ func (r *SQLiteRepository) UpdateAgentSession(ctx context.Context, session *mode result, err := r.db.ExecContext(ctx, ` UPDATE agent_sessions SET - agent_instance_id = ?, agent_type = ?, acp_session_id = ?, status = ?, progress = ?, + agent_instance_id = ?, container_id = ?, agent_type = ?, acp_session_id = ?, status = ?, progress = ?, error_message = ?, metadata = ?, completed_at = ?, updated_at = ? WHERE id = ? - `, session.AgentInstanceID, session.AgentType, session.ACPSessionID, string(session.Status), + `, session.AgentInstanceID, session.ContainerID, session.AgentType, session.ACPSessionID, string(session.Status), session.Progress, session.ErrorMessage, string(metadataJSON), session.CompletedAt, session.UpdatedAt, session.ID) if err != nil { @@ -1470,7 +1476,7 @@ func (r *SQLiteRepository) UpdateAgentSessionStatus(ctx context.Context, id stri // ListAgentSessions returns all agent sessions for a task func (r *SQLiteRepository) ListAgentSessions(ctx context.Context, taskID string) ([]*models.AgentSession, error) { rows, err := r.db.QueryContext(ctx, ` - SELECT id, task_id, agent_instance_id, agent_type, acp_session_id, status, progress, + SELECT id, task_id, agent_instance_id, container_id, agent_type, acp_session_id, status, progress, error_message, metadata, started_at, completed_at, updated_at FROM agent_sessions WHERE task_id = ? ORDER BY started_at DESC `, taskID) @@ -1485,7 +1491,7 @@ func (r *SQLiteRepository) ListAgentSessions(ctx context.Context, taskID string) // ListActiveAgentSessions returns all active agent sessions across all tasks func (r *SQLiteRepository) ListActiveAgentSessions(ctx context.Context) ([]*models.AgentSession, error) { rows, err := r.db.QueryContext(ctx, ` - SELECT id, task_id, agent_instance_id, agent_type, acp_session_id, status, progress, + SELECT id, task_id, agent_instance_id, container_id, agent_type, acp_session_id, status, progress, error_message, metadata, started_at, completed_at, updated_at FROM agent_sessions WHERE status IN ('pending', 'running', 'waiting') ORDER BY started_at DESC `) @@ -1507,7 +1513,7 @@ func (r *SQLiteRepository) scanAgentSessions(rows *sql.Rows) ([]*models.AgentSes var completedAt sql.NullTime err := rows.Scan( - &session.ID, &session.TaskID, &session.AgentInstanceID, &session.AgentType, + &session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentType, &session.ACPSessionID, &status, &session.Progress, &session.ErrorMessage, &metadataJSON, &session.StartedAt, &completedAt, &session.UpdatedAt, ) diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index e3e819da48..d2365ae814 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -41,13 +41,14 @@ const ( ActionRepositoryScriptDelete = "repository.script.delete" // Task actions - ActionTaskList = "task.list" - ActionTaskCreate = "task.create" - ActionTaskGet = "task.get" - ActionTaskUpdate = "task.update" - ActionTaskDelete = "task.delete" - ActionTaskMove = "task.move" - ActionTaskState = "task.state" + ActionTaskList = "task.list" + ActionTaskCreate = "task.create" + ActionTaskGet = "task.get" + ActionTaskUpdate = "task.update" + ActionTaskDelete = "task.delete" + ActionTaskMove = "task.move" + ActionTaskState = "task.state" + ActionTaskExecution = "task.execution" // Agent actions ActionAgentList = "agent.list" diff --git a/apps/web/app/task/[id]/page-client.tsx b/apps/web/app/task/[id]/page-client.tsx index c9388ed9fc..c47ee7b8e2 100644 --- a/apps/web/app/task/[id]/page-client.tsx +++ b/apps/web/app/task/[id]/page-client.tsx @@ -47,12 +47,32 @@ export default function TaskPage({ task }: TaskPageClientProps) { state.kanban.tasks.find((t) => t.id === task?.id) ); - // Update isAgentRunning based on task state + // Fetch task execution status from orchestrator on mount useEffect(() => { - // Check if task state indicates agent is working - const taskState = task?.state; - setIsAgentRunning(taskState === 'IN_PROGRESS'); - }, [task?.state]); + if (!task?.id) return; + + const checkExecution = async () => { + const client = getWebSocketClient(); + if (!client) return; + + try { + const response = await client.request<{ + has_execution: boolean; + task_id: string; + status?: string; + }>('task.execution', { task_id: task.id }); + + console.log('[TaskPage] Task execution check:', response); + if (response.has_execution) { + setIsAgentRunning(true); + } + } catch (err) { + console.error('[TaskPage] Failed to check task execution:', err); + } + }; + + checkExecution(); + }, [task?.id]); const fetchBranches = useCallback(async (workspaceId: string, repoPath: string) => { const response = await listRepositories(getBackendConfig().apiBaseUrl, workspaceId);