feat: add READY status and multi-turn conversation support

- Add AgentStatusReady status for when agent finishes a prompt
- Add AgentReady event published when prompt completes
- PromptAgent accepts RUNNING/READY states, marks READY when done
- Add /complete endpoint to explicitly complete a task
- E2E test polls for READY status instead of file creation
- E2E test validates multi-turn conversation flow
This commit is contained in:
jcfs
2026-01-10 17:06:50 +00:00
parent 20f5f43fb3
commit 7faef25f3a
12 changed files with 308 additions and 31 deletions
+5
View File
@@ -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)
}
@@ -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()
+1
View File
@@ -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"
@@ -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) {
@@ -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"`
@@ -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)
@@ -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
}
@@ -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",
@@ -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},
Binary file not shown.
+2 -1
View File
@@ -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"
+81 -25
View File
@@ -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