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
This commit is contained in:
@@ -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) {
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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
|
||||
// ============================================
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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"`
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user