diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 80540c9391..7d09fa349b 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -422,3 +422,8 @@ func (a *lifecycleAdapter) ListAgentTypes(ctx context.Context) ([]*v1.AgentType, } return result, nil } + +// PromptAgent sends a follow-up prompt to a running agent +func (a *lifecycleAdapter) PromptAgent(ctx context.Context, agentInstanceID string, prompt string) error { + return a.mgr.PromptAgent(ctx, agentInstanceID, prompt) +} diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 6d854a34f8..a27408b8a5 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -328,17 +328,22 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta return fmt.Errorf("prompt failed: %w", err) } - // Prompt completed successfully - mark agent as completed - m.logger.Info("ACP prompt completed successfully", + // Prompt completed - mark agent as READY for follow-up prompts + m.logger.Info("ACP prompt completed, agent ready for follow-up prompts", zap.String("instance_id", instance.ID)) - if err := m.MarkCompleted(instance.ID, 0, ""); err != nil { - m.logger.Error("failed to mark instance as completed", + if err := m.MarkReady(instance.ID); err != nil { + m.logger.Error("failed to mark instance as ready", zap.String("instance_id", instance.ID), zap.Error(err)) } } else { - m.logger.Warn("no task description provided, skipping prompt", + m.logger.Warn("no task description provided, marking as ready", zap.String("instance_id", instance.ID)) + if err := m.MarkReady(instance.ID); err != nil { + m.logger.Error("failed to mark instance as ready", + zap.String("instance_id", instance.ID), + zap.Error(err)) + } } return nil @@ -559,6 +564,49 @@ func (m *Manager) expandMountTemplate(source, workspacePath, taskID string) stri return result } +// PromptAgent sends a follow-up prompt to a running agent +func (m *Manager) PromptAgent(ctx context.Context, instanceID string, prompt string) error { + m.mu.Lock() + instance, exists := m.instances[instanceID] + if !exists { + m.mu.Unlock() + return fmt.Errorf("instance %q not found", instanceID) + } + + if instance.agentctl == nil { + m.mu.Unlock() + return fmt.Errorf("instance %q has no agentctl client", instanceID) + } + + // Accept both RUNNING (initial) and READY (after first prompt) states + if instance.Status != v1.AgentStatusRunning && instance.Status != v1.AgentStatusReady { + m.mu.Unlock() + return fmt.Errorf("instance %q is not ready for prompts (status: %s)", instanceID, instance.Status) + } + + // Set status to RUNNING while processing + instance.Status = v1.AgentStatusRunning + m.mu.Unlock() + + m.logger.Info("sending prompt to agent", + zap.String("instance_id", instanceID), + zap.Int("prompt_length", len(prompt))) + + // Prompt is synchronous - blocks until agent completes + if err := instance.agentctl.Prompt(ctx, prompt); err != nil { + return err + } + + // Prompt completed - mark as READY for next prompt + if err := m.MarkReady(instanceID); err != nil { + m.logger.Error("failed to mark instance as ready after prompt", + zap.String("instance_id", instanceID), + zap.Error(err)) + } + + return nil +} + // StopAgent stops an agent instance func (m *Manager) StopAgent(ctx context.Context, instanceID string, force bool) error { m.mu.RLock() @@ -706,6 +754,27 @@ func (m *Manager) UpdateProgress(instanceID string, progress int) error { return nil } +// MarkReady marks an instance as ready for follow-up prompts +func (m *Manager) MarkReady(instanceID string) error { + m.mu.Lock() + instance, exists := m.instances[instanceID] + if !exists { + m.mu.Unlock() + return fmt.Errorf("instance %q not found", instanceID) + } + + instance.Status = v1.AgentStatusReady + m.mu.Unlock() + + m.logger.Info("instance ready for follow-up prompts", + zap.String("instance_id", instanceID)) + + // Publish ready event + m.publishEvent(context.Background(), events.AgentReady, instance) + + return nil +} + // MarkCompleted marks an instance as completed func (m *Manager) MarkCompleted(instanceID string, exitCode int, errorMessage string) error { m.mu.Lock() diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index af721c107f..6d5a3bd9c3 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -13,6 +13,7 @@ const ( const ( AgentStarted = "agent.started" AgentRunning = "agent.running" + AgentReady = "agent.ready" // Agent finished prompt, ready for follow-up AgentCompleted = "agent.completed" AgentFailed = "agent.failed" AgentStopped = "agent.stopped" diff --git a/apps/backend/internal/orchestrator/api/handlers.go b/apps/backend/internal/orchestrator/api/handlers.go index c5ef8b6a49..ef2e1a2aa8 100644 --- a/apps/backend/internal/orchestrator/api/handlers.go +++ b/apps/backend/internal/orchestrator/api/handlers.go @@ -145,6 +145,57 @@ func (h *Handler) StopTask(c *gin.Context) { }) } +// PromptTask sends a follow-up prompt to a running agent +// POST /api/v1/orchestrator/tasks/:taskId/prompt +func (h *Handler) PromptTask(c *gin.Context) { + taskID := c.Param("taskId") + if taskID == "" { + appErr := errors.BadRequest("taskId is required") + c.JSON(appErr.HTTPStatus, appErr) + return + } + + var req PromptTaskRequest + if err := c.ShouldBindJSON(&req); err != nil { + appErr := errors.ValidationError("request", err.Error()) + c.JSON(appErr.HTTPStatus, appErr) + return + } + + if err := h.service.PromptTask(c.Request.Context(), taskID, req.Prompt); err != nil { + h.logger.Error("failed to prompt task", zap.String("task_id", taskID), zap.Error(err)) + c.JSON(http.StatusInternalServerError, PromptTaskResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + c.JSON(http.StatusOK, PromptTaskResponse{Success: true}) +} + +// CompleteTask explicitly completes a running agent and marks the task as completed +// POST /api/v1/orchestrator/tasks/:taskId/complete +func (h *Handler) CompleteTask(c *gin.Context) { + taskID := c.Param("taskId") + if taskID == "" { + appErr := errors.BadRequest("taskId is required") + c.JSON(appErr.HTTPStatus, appErr) + return + } + + if err := h.service.CompleteTask(c.Request.Context(), taskID); err != nil { + h.logger.Error("failed to complete task", zap.String("task_id", taskID), zap.Error(err)) + c.JSON(http.StatusInternalServerError, gin.H{ + "success": false, + "error": err.Error(), + }) + return + } + + c.JSON(http.StatusOK, gin.H{"success": true, "message": "task completed"}) +} + // PauseTask pauses agent execution (if supported) // POST /api/v1/orchestrator/tasks/:taskId/pause func (h *Handler) PauseTask(c *gin.Context) { diff --git a/apps/backend/internal/orchestrator/api/requests.go b/apps/backend/internal/orchestrator/api/requests.go index aa225cd9ef..d3bada4906 100644 --- a/apps/backend/internal/orchestrator/api/requests.go +++ b/apps/backend/internal/orchestrator/api/requests.go @@ -21,6 +21,22 @@ type StopTaskRequest struct { Force bool `json:"force"` } +// PromptTaskRequest for sending a follow-up prompt to a running agent +type PromptTaskRequest struct { + Prompt string `json:"prompt" binding:"required"` +} + +// PromptTaskResponse for prompt endpoint +type PromptTaskResponse struct { + Success bool `json:"success"` + Error string `json:"error,omitempty"` +} + +// CompleteTaskRequest for explicitly completing a task +type CompleteTaskRequest struct { + Message string `json:"message,omitempty"` +} + // TaskStatusResponse for task execution status type TaskStatusResponse struct { TaskID string `json:"task_id"` diff --git a/apps/backend/internal/orchestrator/api/router.go b/apps/backend/internal/orchestrator/api/router.go index a3338998f5..281a4cdfdb 100644 --- a/apps/backend/internal/orchestrator/api/router.go +++ b/apps/backend/internal/orchestrator/api/router.go @@ -20,6 +20,8 @@ func SetupRoutes(router *gin.RouterGroup, service *orchestrator.Service, log *lo { tasks.POST("/start", handler.StartTask) tasks.POST("/stop", handler.StopTask) + tasks.POST("/prompt", handler.PromptTask) + tasks.POST("/complete", handler.CompleteTask) tasks.POST("/pause", handler.PauseTask) tasks.POST("/resume", handler.ResumeTask) tasks.GET("/status", handler.GetTaskStatus) diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index b817cbe21f..486ba5f9b9 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -34,6 +34,9 @@ type AgentManagerClient interface { // ListAgentTypes returns available agent types ListAgentTypes(ctx context.Context) ([]*v1.AgentType, error) + + // PromptAgent sends a prompt to a running agent + PromptAgent(ctx context.Context, agentInstanceID string, prompt string) error } // LaunchAgentRequest contains parameters for launching an agent @@ -198,6 +201,25 @@ func (e *Executor) Stop(ctx context.Context, taskID string, reason string, force return nil } +// Prompt sends a follow-up prompt to a running agent for a task +func (e *Executor) Prompt(ctx context.Context, taskID string, prompt string) error { + e.mu.RLock() + execution, exists := e.executions[taskID] + if !exists { + e.mu.RUnlock() + return ErrExecutionNotFound + } + agentInstanceID := execution.AgentInstanceID + e.mu.RUnlock() + + e.logger.Info("sending prompt to agent", + zap.String("task_id", taskID), + zap.String("agent_instance_id", agentInstanceID), + zap.Int("prompt_length", len(prompt))) + + return e.agentManager.PromptAgent(ctx, agentInstanceID, prompt) +} + // GetExecution returns the current execution state for a task func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) { e.mu.RLock() @@ -344,3 +366,11 @@ func (m *MockAgentManagerClient) ListAgentTypes(ctx context.Context) ([]*v1.Agen }, }, nil } + +// PromptAgent mocks sending a prompt to an agent +func (m *MockAgentManagerClient) PromptAgent(ctx context.Context, agentInstanceID string, prompt string) error { + m.logger.Info("mock: prompting agent", + zap.String("agent_instance_id", agentInstanceID), + zap.Int("prompt_length", len(prompt))) + return nil +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 0e1513e310..e9e0ff7ea7 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -4,6 +4,7 @@ package orchestrator import ( "context" "errors" + "fmt" "sync" "time" @@ -111,6 +112,7 @@ func NewService( // 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, OnACPMessage: s.handleACPMessage, @@ -263,6 +265,37 @@ func (s *Service) StopTask(ctx context.Context, taskID string, reason string, fo return s.executor.Stop(ctx, taskID, reason, force) } +// PromptTask sends a follow-up prompt to a running agent for a task +func (s *Service) PromptTask(ctx context.Context, taskID string, prompt string) error { + s.logger.Info("sending prompt to task agent", + zap.String("task_id", taskID), + zap.Int("prompt_length", len(prompt))) + return s.executor.Prompt(ctx, taskID, prompt) +} + +// CompleteTask explicitly completes a task and stops its agent +func (s *Service) CompleteTask(ctx context.Context, taskID string) error { + s.logger.Info("completing task", + zap.String("task_id", taskID)) + + // Stop the agent (which will trigger AgentCompleted event and update task state) + if err := s.executor.Stop(ctx, taskID, "task completed by user", false); err != nil { + // If agent is already stopped, just update the task state directly + s.logger.Warn("failed to stop agent, 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) @@ -331,6 +364,17 @@ func (s *Service) handleTaskStateChanged(ctx context.Context, data watcher.TaskE } } +// handleAgentReady handles agent ready events (prompt completed, ready for follow-up) +func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventData) { + s.logger.Info("handling agent ready", + zap.String("task_id", data.TaskID), + zap.String("agent_instance_id", data.AgentInstanceID)) + + // Agent completed initial prompt but is still running + // Task stays in IN_PROGRESS state - user can send follow-up prompts + // or explicitly complete/stop the task +} + // handleAgentCompleted handles agent completion events func (s *Service) handleAgentCompleted(ctx context.Context, data watcher.AgentEventData) { s.logger.Info("handling agent completed", diff --git a/apps/backend/internal/orchestrator/watcher/watcher.go b/apps/backend/internal/orchestrator/watcher/watcher.go index 82890c6a2b..a3253c7b74 100644 --- a/apps/backend/internal/orchestrator/watcher/watcher.go +++ b/apps/backend/internal/orchestrator/watcher/watcher.go @@ -43,6 +43,7 @@ type EventHandlers struct { // Agent events OnAgentStarted func(ctx context.Context, data AgentEventData) + OnAgentReady func(ctx context.Context, data AgentEventData) OnAgentCompleted func(ctx context.Context, data AgentEventData) OnAgentFailed func(ctx context.Context, data AgentEventData) OnAgentStopped func(ctx context.Context, data AgentEventData) @@ -177,6 +178,7 @@ func (w *Watcher) subscribeToAgentEvents() error { handler func(ctx context.Context, data AgentEventData) }{ {events.AgentStarted, w.handlers.OnAgentStarted}, + {events.AgentReady, w.handlers.OnAgentReady}, {events.AgentCompleted, w.handlers.OnAgentCompleted}, {events.AgentFailed, w.handlers.OnAgentFailed}, {events.AgentStopped, w.handlers.OnAgentStopped}, diff --git a/apps/backend/kandev b/apps/backend/kandev index 36da0a4b2e..7a54d9a500 100755 Binary files a/apps/backend/kandev and b/apps/backend/kandev differ diff --git a/apps/backend/pkg/api/v1/agent.go b/apps/backend/pkg/api/v1/agent.go index 2b9d1ba2a4..4c966b9cae 100644 --- a/apps/backend/pkg/api/v1/agent.go +++ b/apps/backend/pkg/api/v1/agent.go @@ -8,7 +8,8 @@ type AgentStatus string const ( AgentStatusPending AgentStatus = "PENDING" AgentStatusStarting AgentStatus = "STARTING" - AgentStatusRunning AgentStatus = "RUNNING" + AgentStatusRunning AgentStatus = "RUNNING" // Agent is processing a prompt + AgentStatusReady AgentStatus = "READY" // Agent finished prompt, waiting for next AgentStatusCompleted AgentStatus = "COMPLETED" AgentStatusFailed AgentStatus = "FAILED" AgentStatusStopped AgentStatus = "STOPPED" diff --git a/scripts/e2e-test.sh b/scripts/e2e-test.sh index 2840dbea94..932a760984 100755 --- a/scripts/e2e-test.sh +++ b/scripts/e2e-test.sh @@ -269,20 +269,22 @@ else log_info "Waiting for agent initialization..." fi -log_step "Step 7: Wait for Agent to Process" -log_info "Waiting up to ${WAIT_FOR_AGENT_SECONDS} seconds for agent to complete..." -AGENT_STATUS="RUNNING" +log_step "Step 7: Wait for Agent to Complete First Prompt" +log_info "Waiting up to ${WAIT_FOR_AGENT_SECONDS} seconds for agent to become READY..." +AGENT_OUTPUT_FILE="${TEST_WORKSPACE}/agent-result.txt" +AGENT_READY=false for i in $(seq 1 $WAIT_FOR_AGENT_SECONDS); do + # Check agent status - READY means prompt completed AGENT_INFO=$(curl -s "${BASE_URL}/api/v1/agents/${AGENT_ID}/status" 2>/dev/null || echo '{}') AGENT_STATUS=$(echo "$AGENT_INFO" | jq -r '.status // "UNKNOWN"') - if [ "$AGENT_STATUS" == "COMPLETED" ]; then - log_success "Agent completed successfully" + if [ "$AGENT_STATUS" == "READY" ]; then + log_success "Agent is READY after $i seconds (prompt completed)" + AGENT_READY=true break elif [ "$AGENT_STATUS" == "FAILED" ]; then log_error "Agent failed" echo "$AGENT_INFO" | jq . - # Show container logs for debugging if [ -n "$CONTAINER_ID" ]; then log_info "Container logs:" docker logs "$CONTAINER_ID" 2>&1 | tail -30 @@ -293,23 +295,21 @@ for i in $(seq 1 $WAIT_FOR_AGENT_SECONDS); do # Check if container is still running CONTAINER_RUNNING=$(docker ps --filter "id=${CONTAINER_ID}" -q | wc -l) if [ "$CONTAINER_RUNNING" -eq 0 ]; then - log_info "Container stopped, checking for completion..." + log_info "Container stopped unexpectedly" break fi - # Periodically check agentctl status if IP is available - if [ -n "$CONTAINER_IP" ] && [ $((i % 15)) -eq 0 ]; then - AGENTCTL_STATUS=$(curl -s "http://${CONTAINER_IP}:${AGENTCTL_PORT}/api/v1/status" 2>/dev/null || echo '{}') - AGENT_PROCESS_STATE=$(echo "$AGENTCTL_STATUS" | jq -r '.agent_status // "unknown"') - log_info "Agent state via agentctl: $AGENT_PROCESS_STATE" - fi - if [ $((i % 10)) -eq 0 ]; then - log_info "Still waiting... ($i seconds)" + log_info "Still waiting... ($i seconds, status: $AGENT_STATUS)" fi sleep 1 done +if [ "$AGENT_READY" != "true" ]; then + log_error "Agent did not become READY within timeout (last status: $AGENT_STATUS)" + exit 1 +fi + log_step "Step 8: Verify Agent Response" # Get container logs to verify agent processed the request FINAL_CONTAINER_ID=$(docker ps -a --filter "name=kandev-agent" --format "{{.ID}}" | head -1) @@ -345,25 +345,79 @@ FINAL_TASK=$(curl -s "${BASE_URL}/api/v1/tasks/${TASK_ID}") TASK_STATE=$(echo "$FINAL_TASK" | jq -r '.state') log_info "Task state: $TASK_STATE" -log_step "Step 10: Verify Agent Output" -AGENT_OUTPUT_FILE="${TEST_WORKSPACE}/agent-result.txt" +log_step "Step 10: Verify First Prompt Output" if [ -f "$AGENT_OUTPUT_FILE" ]; then AGENT_OUTPUT=$(cat "$AGENT_OUTPUT_FILE") log_success "Agent created file: $AGENT_OUTPUT_FILE" log_info "File content: $AGENT_OUTPUT" if echo "$AGENT_OUTPUT" | grep -qi "hello"; then - log_success "Agent output contains expected content" - else - log_info "Note: Output doesn't contain 'Hello' but file was created" + log_success "First prompt output contains expected content" fi else - log_info "Agent did not create expected file (agent-result.txt)" - log_info "Checking workspace for any created files..." - ls -la "$TEST_WORKSPACE" 2>/dev/null || true + log_error "Agent did not create expected file" + exit 1 +fi + +log_step "Step 11: Test Multi-Turn - Send Follow-up Prompt" +SECOND_OUTPUT_FILE="${TEST_WORKSPACE}/second-result.txt" +log_info "Sending follow-up prompt..." +PROMPT_RESULT=$(curl -s -X POST "${BASE_URL}/api/v1/orchestrator/tasks/${TASK_ID}/prompt" \ + -H "Content-Type: application/json" \ + -d "{\"prompt\": \"Create another file named 'second-result.txt' with the content 'Follow-up successful'. Do nothing else.\"}") +PROMPT_SUCCESS=$(echo "$PROMPT_RESULT" | jq -r '.success // false') +if [ "$PROMPT_SUCCESS" == "true" ]; then + log_success "Follow-up prompt accepted" +else + log_error "Follow-up prompt failed: $(echo "$PROMPT_RESULT" | jq -r '.error // "unknown"')" + exit 1 +fi + +# Wait for agent to become READY again (prompt completed) +log_info "Waiting for agent to become READY again..." +for i in $(seq 1 120); do + AGENT_INFO=$(curl -s "${BASE_URL}/api/v1/agents/${AGENT_ID}/status" 2>/dev/null || echo '{}') + AGENT_STATUS=$(echo "$AGENT_INFO" | jq -r '.status // "UNKNOWN"') + if [ "$AGENT_STATUS" == "READY" ]; then + log_success "Agent is READY after follow-up prompt ($i seconds)" + break + fi + if [ $((i % 10)) -eq 0 ]; then + log_info "Still waiting... ($i seconds, status: $AGENT_STATUS)" + fi + sleep 1 +done + +# Verify second file was created +if [ -f "$SECOND_OUTPUT_FILE" ]; then + SECOND_OUTPUT=$(cat "$SECOND_OUTPUT_FILE") + log_success "Second file created: $SECOND_OUTPUT_FILE" + log_info "Second file content: $SECOND_OUTPUT" +else + log_error "Agent did not create second file" + exit 1 +fi + +log_step "Step 12: Complete Task" +COMPLETE_RESULT=$(curl -s -X POST "${BASE_URL}/api/v1/orchestrator/tasks/${TASK_ID}/complete" \ + -H "Content-Type: application/json") +COMPLETE_SUCCESS=$(echo "$COMPLETE_RESULT" | jq -r '.success // false') +if [ "$COMPLETE_SUCCESS" == "true" ]; then + log_success "Task completed successfully" +else + log_error "Failed to complete task: $(echo "$COMPLETE_RESULT" | jq -r '.error // "unknown"')" +fi + +# Verify task state is COMPLETED +FINAL_TASK=$(curl -s "${BASE_URL}/api/v1/tasks/${TASK_ID}") +TASK_STATE=$(echo "$FINAL_TASK" | jq -r '.state') +if [ "$TASK_STATE" == "COMPLETED" ]; then + log_success "Task state is COMPLETED" +else + log_error "Expected task state COMPLETED, got: $TASK_STATE" fi # Clean up test workspace -#rm -rf "$TEST_WORKSPACE" 2>/dev/null || true +rm -rf "$TEST_WORKSPACE" 2>/dev/null || true # Show backend logs summary log_step "Backend Log Summary" @@ -382,7 +436,9 @@ echo -e "${GREEN}║ Column created: ✓ ║${NC}" echo -e "${GREEN}║ Task created: ✓ ║${NC}" echo -e "${GREEN}║ Agent launched: ✓ ║${NC}" echo -e "${GREEN}║ agentctl running: ✓ ║${NC}" -echo -e "${GREEN}║ Agent processed: ✓ ║${NC}" +echo -e "${GREEN}║ First prompt: ✓ ║${NC}" +echo -e "${GREEN}║ Multi-turn: ✓ ║${NC}" +echo -e "${GREEN}║ Task completed: ✓ ║${NC}" echo -e "${GREEN}╚════════════════════════════════════════╝${NC}" exit 0