diff --git a/.gitignore b/.gitignore index 567ad02ea5..d1f67b7c83 100644 --- a/.gitignore +++ b/.gitignore @@ -68,3 +68,6 @@ tmp/ apps/backend/bin/ +# Go binaries (built locally) +apps/backend/kandev +apps/backend/agentctl diff --git a/apps/backend/CODE_REVIEW_ISSUES.md b/apps/backend/CODE_REVIEW_ISSUES.md new file mode 100644 index 0000000000..014ea2ceb2 --- /dev/null +++ b/apps/backend/CODE_REVIEW_ISSUES.md @@ -0,0 +1,52 @@ +# Backend Code Review Issues + +Generated: 2026-01-11 + +## High Priority + +### 1. Security TODOs +- [ ] `internal/gateway/websocket/handler.go:20` - TODO: In production, validate origin +- [ ] `internal/gateway/websocket/handler.go:46` - TODO: Implement proper JWT validation + +### 2. Deprecated API +- [ ] `internal/agent/docker/client.go:300-301` - `inspect.NetworkSettings.IPAddress` is deprecated (removed in Docker v29). Use `NetworkSettings.Networks` instead. + +## Medium Priority + +### 3. Unused Functions +- [ ] `internal/agent/lifecycle/manager.go:641` - `updateInstanceComplete` is unused +- [ ] `internal/agent/streaming/reader.go:232` - `marshalMessage` is unused +- [ ] `internal/task/repository/sqlite.go:617` - `scanTasks` is unused + +### 4. Unused Field / Race Condition +- [ ] `internal/gateway/websocket/client.go:36` - `mu sync.RWMutex` field is unused. Either use it to protect `subscriptions` or remove it. + +### 5. Outdated Comment +- [ ] `internal/agent/agentctl/acp.go:108` - Comment references `needs_input` which was removed. Update to: `// The StopReason indicates why the agent stopped (e.g., "end_turn")` + +## Low Priority + +### 6. Unused Test Helpers +- [ ] `internal/agent/lifecycle/manager_test.go:130` - `testManager` type unused +- [ ] `internal/agent/lifecycle/manager_test.go:135` - `newTestManager` function unused + +### 7. Code Style (staticcheck S1000) +- [ ] `internal/agentctl/api/acp.go:215` - Use `for range ch` instead of `for { select {} }` +- [ ] `internal/agentctl/api/acp.go:336` - Use `for range ch` instead of `for { select {} }` + +### 8. Hardcoded Sleep +- [ ] `internal/agent/lifecycle/manager.go:253` - `time.Sleep(500 * time.Millisecond)` - Consider proper ready-check mechanism + +### 9. Ignored Error +- [ ] `internal/task/repository/sqlite.go:626` - `_ = json.Unmarshal(...)` ignores error + +## Questions / Review Needed + +### 10. InputRequestHandler Redundancy +Files: `internal/orchestrator/service.go` (lines 44-45, 71, 387-389, 507-512), `cmd/kandev/main.go` (lines 305-363) + +The `InputRequestHandler` and `handleInputRequired` functionality still exists using `input_required` message type. Is this still needed given the new comment-based conversation flow? + +### 11. Other TODOs +- [ ] `internal/gateway/websocket/hub.go:127` - TODO: Add topic-based routing for task-specific notifications + diff --git a/apps/backend/cmd/agentctl/main.go b/apps/backend/cmd/agentctl/main.go new file mode 100644 index 0000000000..19383561b7 --- /dev/null +++ b/apps/backend/cmd/agentctl/main.go @@ -0,0 +1,99 @@ +// Package main is the entry point for the agentctl binary +// agentctl is a sidecar process that manages agent subprocess communication +// via HTTP API, bridging the agent's ACP protocol with the kandev backend. +package main + +import ( + "context" + "fmt" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + "github.com/kandev/kandev/internal/agentctl/api" + "github.com/kandev/kandev/internal/agentctl/config" + "github.com/kandev/kandev/internal/agentctl/process" + "github.com/kandev/kandev/internal/common/logger" + "go.uber.org/zap" +) + +func main() { + // Load configuration from environment + cfg := config.Load() + + // Initialize logger + log, err := logger.NewLogger(logger.LoggingConfig{ + Level: cfg.LogLevel, + Format: cfg.LogFormat, + OutputPath: "stdout", + }) + if err != nil { + fmt.Fprintf(os.Stderr, "failed to initialize logger: %v\n", err) + os.Exit(1) + } + defer log.Sync() + + log.Info("starting agentctl", + zap.Int("port", cfg.Port), + zap.String("agent_command", cfg.AgentCommand), + zap.String("workdir", cfg.WorkDir), + zap.Bool("auto_start", cfg.AutoStart)) + + // Create process manager + procMgr := process.NewManager(cfg, log) + + // Create HTTP API server + server := api.NewServer(cfg, procMgr, log) + + // Start HTTP server + addr := fmt.Sprintf(":%d", cfg.Port) + httpServer := &http.Server{ + Addr: addr, + Handler: server.Router(), + } + + // Start server in goroutine + go func() { + log.Info("HTTP server starting", zap.String("address", addr)) + if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Error("HTTP server error", zap.Error(err)) + os.Exit(1) + } + }() + + // Auto-start agent if configured + if cfg.AutoStart { + log.Info("auto-starting agent process") + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + if err := procMgr.Start(ctx); err != nil { + log.Error("failed to auto-start agent", zap.Error(err)) + } + cancel() + } + + // Wait for shutdown signal + quit := make(chan os.Signal, 1) + signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) + <-quit + + log.Info("shutting down agentctl...") + + // Graceful shutdown + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + // Stop agent process + if err := procMgr.Stop(ctx); err != nil { + log.Error("error stopping agent process", zap.Error(err)) + } + + // Shutdown HTTP server + if err := httpServer.Shutdown(ctx); err != nil { + log.Error("error shutting down HTTP server", zap.Error(err)) + } + + log.Info("agentctl stopped") +} + diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index d0a14b14ee..fc178d8325 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -202,6 +202,9 @@ func main() { serviceCfg := orchestrator.DefaultServiceConfig() orchestratorSvc := orchestrator.NewService(serviceCfg, eventBus, nil, agentManagerClient, taskRepoAdapter, log) + // Set up comment creator for saving agent responses as comments + orchestratorSvc.SetCommentCreator(&commentCreatorAdapter{svc: taskSvc}) + // ============================================ // WEBSOCKET GATEWAY (All communication via WebSocket) // ============================================ @@ -396,6 +399,7 @@ func main() { taskhandlers.RegisterBoardRoutes(router, gateway.Dispatcher, boardController, log) taskhandlers.RegisterTaskRoutes(router, gateway.Dispatcher, taskController, log) taskhandlers.RegisterRepositoryRoutes(router, gateway.Dispatcher, repositoryController, log) + taskhandlers.RegisterCommentRoutes(router, gateway.Dispatcher, taskSvc, &orchestratorAdapter{svc: orchestratorSvc}, log) log.Info("Registered Task Service handlers (HTTP + WebSocket)") // Health check (simple HTTP for load balancers/monitoring) @@ -578,11 +582,16 @@ func (a *lifecycleAdapter) PromptAgent(ctx context.Context, agentInstanceID stri return nil, err } return &executor.PromptResult{ - StopReason: result.StopReason, - NeedsInput: result.NeedsInput, + StopReason: result.StopReason, + AgentMessage: result.AgentMessage, }, nil } +// RespondToPermissionByTaskID sends a response to a permission request for a task +func (a *lifecycleAdapter) RespondToPermissionByTaskID(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error { + return a.mgr.RespondToPermissionByTaskID(taskID, pendingID, optionID, cancelled) +} + // corsMiddleware returns a CORS middleware for WebSocket connections func corsMiddleware() gin.HandlerFunc { return func(c *gin.Context) { @@ -599,3 +608,35 @@ func corsMiddleware() gin.HandlerFunc { c.Next() } } + +// orchestratorAdapter adapts the orchestrator.Service to the taskhandlers.OrchestratorService interface +type orchestratorAdapter struct { + svc *orchestrator.Service +} + +// PromptTask forwards to the orchestrator service and converts the result type +func (a *orchestratorAdapter) PromptTask(ctx context.Context, taskID string, prompt string) (*taskhandlers.PromptResult, error) { + result, err := a.svc.PromptTask(ctx, taskID, prompt) + if err != nil { + return nil, err + } + return &taskhandlers.PromptResult{ + StopReason: result.StopReason, + AgentMessage: result.AgentMessage, + }, nil +} + +// commentCreatorAdapter adapts the task service to the orchestrator.CommentCreator interface +type commentCreatorAdapter struct { + svc *taskservice.Service +} + +// CreateAgentComment creates a comment with author_type="agent" +func (a *commentCreatorAdapter) CreateAgentComment(ctx context.Context, taskID, content string) error { + _, err := a.svc.CreateComment(ctx, &taskservice.CreateCommentRequest{ + TaskID: taskID, + Content: content, + AuthorType: "agent", + }) + return err +} diff --git a/apps/backend/dockerfiles/augment-agent/Dockerfile b/apps/backend/dockerfiles/augment-agent/Dockerfile index 93959bc941..36c83aca0f 100644 --- a/apps/backend/dockerfiles/augment-agent/Dockerfile +++ b/apps/backend/dockerfiles/augment-agent/Dockerfile @@ -39,10 +39,6 @@ RUN chmod +x /usr/local/bin/agentctl # Create agent directory WORKDIR /agent -# Copy agent script (legacy, for backward compatibility) -COPY dockerfiles/augment-agent/agent.sh /agent/agent.sh -RUN chmod +x /agent/agent.sh - # Copy entrypoint script COPY dockerfiles/augment-agent/entrypoint.sh /agent/entrypoint.sh RUN chmod +x /agent/entrypoint.sh diff --git a/apps/backend/dockerfiles/augment-agent/agent.sh b/apps/backend/dockerfiles/augment-agent/agent.sh deleted file mode 100644 index b3126e8a41..0000000000 --- a/apps/backend/dockerfiles/augment-agent/agent.sh +++ /dev/null @@ -1,31 +0,0 @@ -#!/bin/bash -# Augment Agent - Runs in ACP (Agent Client Protocol) mode -# Uses JSON-RPC 2.0 over stdin/stdout for bidirectional communication - -set -e - -# Check if auggie is available -if ! command -v auggie &> /dev/null; then - echo '{"jsonrpc":"2.0","method":"error","params":{"message":"Auggie CLI not found"}}' >&2 - exit 1 -fi - -# Check for authentication -if [ -z "$AUGMENT_SESSION_AUTH" ]; then - echo '{"jsonrpc":"2.0","method":"error","params":{"message":"AUGMENT_SESSION_AUTH required"}}' >&2 - exit 1 -fi - -# Export for auggie to use -export AUGMENT_SESSION_AUTH - -# Build auggie command -AUGGIE_ARGS="--acp" - -# Add workspace root if we're in a workspace with files -if [ -d "/workspace" ] && [ "$(ls -A /workspace 2>/dev/null)" ]; then - AUGGIE_ARGS="$AUGGIE_ARGS --workspace-root /workspace" -fi - -# Execute auggie in ACP mode - stdin/stdout will be used for JSON-RPC communication -exec auggie $AUGGIE_ARGS diff --git a/apps/backend/internal/agent/acp/session.go b/apps/backend/internal/agent/acp/session.go index 9282fdd473..b0d03cf086 100644 --- a/apps/backend/internal/agent/acp/session.go +++ b/apps/backend/internal/agent/acp/session.go @@ -28,6 +28,29 @@ type Session struct { mu sync.RWMutex } +// PendingPermission represents a permission request waiting for user response +type PendingPermission struct { + ID string // Unique ID for this pending request + RPCID interface{} // JSON-RPC request ID from agent + InstanceID string // Agent instance + TaskID string // Task being worked on + SessionID string // ACP session ID + ToolCallID string // Tool call requesting permission + Title string // Human-readable title + Options []jsonrpc.PermissionOption // Available choices + ResponseCh chan *PermissionResponseMsg // Channel to receive user response + CreatedAt time.Time +} + +// PermissionResponseMsg is the message sent by the user in response to a permission request +type PermissionResponseMsg struct { + OptionID string + Cancelled bool +} + +// PermissionRequestHandler is called when an agent requests permission from the user +type PermissionRequestHandler func(req *PendingPermission) + // UpdateHandler is called when the agent sends session updates type UpdateHandler func(instanceID, taskID string, updateType string, data json.RawMessage) @@ -37,17 +60,25 @@ type SessionManager struct { mu sync.RWMutex eventBus bus.EventBus logger *logger.Logger - + // Handler for session updates updateHandler UpdateHandler + + // Handler for permission requests + permissionHandler PermissionRequestHandler + + // Pending permission requests waiting for user response + pendingPermissions map[string]*PendingPermission + permissionMu sync.RWMutex } // NewSessionManager creates a new session manager func NewSessionManager(eventBus bus.EventBus, log *logger.Logger) *SessionManager { return &SessionManager{ - sessions: make(map[string]*Session), - eventBus: eventBus, - logger: log.WithFields(zap.String("component", "acp-session-manager")), + sessions: make(map[string]*Session), + pendingPermissions: make(map[string]*PendingPermission), + eventBus: eventBus, + logger: log.WithFields(zap.String("component", "acp-session-manager")), } } @@ -56,6 +87,11 @@ func (m *SessionManager) SetUpdateHandler(handler UpdateHandler) { m.updateHandler = handler } +// SetPermissionHandler sets the handler for permission requests from agents +func (m *SessionManager) SetPermissionHandler(handler PermissionRequestHandler) { + m.permissionHandler = handler +} + // CreateSession creates a new ACP session for an agent instance func (m *SessionManager) CreateSession(ctx context.Context, instanceID, taskID string, stdin io.WriteCloser, stdout io.Reader) error { m.logger.Info("Creating ACP session", @@ -395,8 +431,91 @@ func (m *SessionManager) handleRequestPermission(session *Session, id interface{ zap.String("title", reqParams.ToolCall.Title), zap.Int("num_options", len(reqParams.Options))) - // Auto-approve: find the first "allow" option and select it - // In a real implementation, this would prompt the user or check policies + // If no permission handler is set, fall back to auto-approve + if m.permissionHandler == nil { + m.autoApprovePermission(session, id, &reqParams) + return + } + + // Create a pending permission request + pendingID := fmt.Sprintf("%s-%d", session.InstanceID, time.Now().UnixNano()) + pending := &PendingPermission{ + ID: pendingID, + RPCID: id, + InstanceID: session.InstanceID, + TaskID: session.TaskID, + SessionID: reqParams.SessionID, + ToolCallID: reqParams.ToolCall.ToolCallID, + Title: reqParams.ToolCall.Title, + Options: reqParams.Options, + ResponseCh: make(chan *PermissionResponseMsg, 1), + CreatedAt: time.Now(), + } + + // Store the pending request + m.permissionMu.Lock() + m.pendingPermissions[pendingID] = pending + m.permissionMu.Unlock() + + // Notify the permission handler (which will send WebSocket notification to user) + m.permissionHandler(pending) + + // Wait for user response in a goroutine to not block the read loop + go m.waitForPermissionResponse(session, pending) +} + +// waitForPermissionResponse waits for user response and sends it to the agent +func (m *SessionManager) waitForPermissionResponse(session *Session, pending *PendingPermission) { + // Wait for response with a 5-minute timeout + timeout := time.After(5 * time.Minute) + + select { + case resp := <-pending.ResponseCh: + m.logger.Info("Received permission response from user", + zap.String("pending_id", pending.ID), + zap.String("option_id", resp.OptionID), + zap.Bool("cancelled", resp.Cancelled)) + + var result jsonrpc.RequestPermissionResult + if resp.Cancelled { + result = jsonrpc.RequestPermissionResult{ + Outcome: jsonrpc.PermissionOutcome{ + Outcome: "cancelled", + }, + } + } else { + result = jsonrpc.RequestPermissionResult{ + Outcome: jsonrpc.PermissionOutcome{ + Outcome: "selected", + OptionID: resp.OptionID, + }, + } + } + + if err := session.Client.SendResponse(pending.RPCID, result, nil); err != nil { + m.logger.Error("Failed to send permission response to agent", zap.Error(err)) + } + + case <-timeout: + m.logger.Warn("Permission request timed out, cancelling", + zap.String("pending_id", pending.ID)) + + result := jsonrpc.RequestPermissionResult{ + Outcome: jsonrpc.PermissionOutcome{ + Outcome: "cancelled", + }, + } + session.Client.SendResponse(pending.RPCID, result, nil) + } + + // Clean up + m.permissionMu.Lock() + delete(m.pendingPermissions, pending.ID) + m.permissionMu.Unlock() +} + +// autoApprovePermission auto-approves a permission request (fallback behavior) +func (m *SessionManager) autoApprovePermission(session *Session, id interface{}, reqParams *jsonrpc.RequestPermissionParams) { var selectedOptionID string for _, opt := range reqParams.Options { if opt.Kind == "allow_once" || opt.Kind == "allow_always" { @@ -416,7 +535,6 @@ func (m *SessionManager) handleRequestPermission(session *Session, id interface{ zap.String("option_id", selectedOptionID)) } - // Send response result := jsonrpc.RequestPermissionResult{ Outcome: jsonrpc.PermissionOutcome{ Outcome: "selected", @@ -427,8 +545,55 @@ func (m *SessionManager) handleRequestPermission(session *Session, id interface{ if err := session.Client.SendResponse(id, result, nil); err != nil { m.logger.Error("Failed to send permission response", zap.Error(err)) } else { - m.logger.Info("Sent permission response", + m.logger.Info("Sent auto-approved permission response", zap.String("instance_id", session.InstanceID), zap.String("selected_option", selectedOptionID)) } } + +// RespondToPermission sends a user's response to a pending permission request +func (m *SessionManager) RespondToPermission(pendingID string, optionID string, cancelled bool) error { + m.permissionMu.RLock() + pending, exists := m.pendingPermissions[pendingID] + m.permissionMu.RUnlock() + + if !exists { + return fmt.Errorf("pending permission request not found: %s", pendingID) + } + + // Send response through channel (non-blocking) + select { + case pending.ResponseCh <- &PermissionResponseMsg{ + OptionID: optionID, + Cancelled: cancelled, + }: + m.logger.Info("Sent permission response to waiting goroutine", + zap.String("pending_id", pendingID), + zap.String("option_id", optionID)) + return nil + default: + return fmt.Errorf("permission request already responded to or timed out") + } +} + +// GetPendingPermission returns a pending permission request by ID +func (m *SessionManager) GetPendingPermission(pendingID string) (*PendingPermission, bool) { + m.permissionMu.RLock() + defer m.permissionMu.RUnlock() + pending, exists := m.pendingPermissions[pendingID] + return pending, exists +} + +// GetPendingPermissionsForTask returns all pending permissions for a task +func (m *SessionManager) GetPendingPermissionsForTask(taskID string) []*PendingPermission { + m.permissionMu.RLock() + defer m.permissionMu.RUnlock() + + var result []*PendingPermission + for _, pending := range m.pendingPermissions { + if pending.TaskID == taskID { + result = append(result, pending) + } + } + return result +} diff --git a/apps/backend/internal/agent/agentctl/acp.go b/apps/backend/internal/agent/agentctl/acp.go index a1069687e0..c8652f67cb 100644 --- a/apps/backend/internal/agent/agentctl/acp.go +++ b/apps/backend/internal/agent/agentctl/acp.go @@ -146,6 +146,36 @@ func (c *Client) Prompt(ctx context.Context, text string) (*PromptResponse, erro // SessionNotification represents a session update from the agent type SessionNotification = acp.SessionNotification +// PermissionOption represents a permission choice +type PermissionOption struct { + OptionID string `json:"option_id"` + Name string `json:"name"` + Kind string `json:"kind"` // allow_once, allow_always, reject_once, reject_always +} + +// PermissionNotification is received when the agent requests permission +type PermissionNotification struct { + PendingID string `json:"pending_id"` + SessionID string `json:"session_id"` + ToolCallID string `json:"tool_call_id"` + Title string `json:"title"` + Options []PermissionOption `json:"options"` + CreatedAt time.Time `json:"created_at"` +} + +// PermissionResponse is the user's response to a permission request +type PermissionRespondRequest struct { + PendingID string `json:"pending_id"` + OptionID string `json:"option_id,omitempty"` + Cancelled bool `json:"cancelled,omitempty"` +} + +// PermissionRespondResponse from agentctl +type PermissionRespondResponse struct { + Success bool `json:"success"` + Error string `json:"error,omitempty"` +} + // StreamUpdates opens a WebSocket connection for streaming session updates func (c *Client) StreamUpdates(ctx context.Context, handler func(acp.SessionNotification)) error { wsURL := "ws" + c.baseURL[4:] + "/api/v1/acp/stream" @@ -276,5 +306,99 @@ func (c *Client) CloseOutputStream() { func (c *Client) Close() { c.CloseACPStream() c.CloseOutputStream() + c.ClosePermissionStream() } +// StreamPermissions opens a WebSocket connection for streaming permission requests +func (c *Client) StreamPermissions(ctx context.Context, handler func(*PermissionNotification)) error { + wsURL := "ws" + c.baseURL[4:] + "/api/v1/acp/permissions/stream" + + conn, _, err := websocket.DefaultDialer.DialContext(ctx, wsURL, nil) + if err != nil { + return fmt.Errorf("failed to connect to permission stream: %w", err) + } + + c.mu.Lock() + c.permissionConn = conn + c.mu.Unlock() + + c.logger.Info("connected to permission stream", zap.String("url", wsURL)) + + // Read messages in a goroutine + go func() { + defer func() { + c.mu.Lock() + c.permissionConn = nil + c.mu.Unlock() + conn.Close() + }() + + for { + _, message, err := conn.ReadMessage() + if err != nil { + if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) { + c.logger.Info("permission stream closed normally") + } else { + c.logger.Debug("permission stream error", zap.Error(err)) + } + return + } + + var notification PermissionNotification + if err := json.Unmarshal(message, ¬ification); err != nil { + c.logger.Warn("failed to parse permission notification", zap.Error(err)) + continue + } + + handler(¬ification) + } + }() + + return nil +} + +// ClosePermissionStream closes the permission stream connection +func (c *Client) ClosePermissionStream() { + c.mu.Lock() + defer c.mu.Unlock() + + if c.permissionConn != nil { + c.permissionConn.Close() + c.permissionConn = nil + } +} + +// RespondToPermission sends a response to a permission request +func (c *Client) RespondToPermission(ctx context.Context, pendingID, optionID string, cancelled bool) error { + reqBody := PermissionRespondRequest{ + PendingID: pendingID, + OptionID: optionID, + Cancelled: cancelled, + } + + body, err := json.Marshal(reqBody) + if err != nil { + return err + } + + req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/acp/permissions/respond", bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + var result PermissionRespondResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return err + } + if !result.Success { + return fmt.Errorf("permission response failed: %s", result.Error) + } + return nil +} diff --git a/apps/backend/internal/agent/agentctl/client.go b/apps/backend/internal/agent/agentctl/client.go index 05386ae7c5..e672d2f2a8 100644 --- a/apps/backend/internal/agent/agentctl/client.go +++ b/apps/backend/internal/agent/agentctl/client.go @@ -21,9 +21,10 @@ type Client struct { logger *logger.Logger // WebSocket connections for streaming - outputConn *websocket.Conn - acpConn *websocket.Conn - mu sync.RWMutex + outputConn *websocket.Conn + acpConn *websocket.Conn + permissionConn *websocket.Conn + mu sync.RWMutex } // StatusResponse from agentctl diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 5693c20e96..43403f6864 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -43,6 +43,10 @@ type AgentInstance struct { // agentctl client for this instance agentctl *agentctl.Client + + // Message buffer for accumulating agent response during a prompt + messageBuffer strings.Builder + messageMu sync.Mutex } // LaunchRequest contains parameters for launching an agent @@ -311,13 +315,21 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta // 3. Set up updates stream to receive session notifications go m.handleUpdatesStream(instance) - // 4. Send the task prompt if provided + // 4. Set up permission stream to receive permission requests + go m.handlePermissionStream(instance) + + // 5. Send the task prompt if provided if taskDescription != "" { m.logger.Info("sending ACP prompt", zap.String("instance_id", instance.ID), zap.String("session_id", sessionID), zap.String("task_description", taskDescription)) + // Clear message buffer before starting prompt + instance.messageMu.Lock() + instance.messageBuffer.Reset() + instance.messageMu.Unlock() + // Prompt is SYNCHRONOUS - it blocks until the agent completes the task // Use a long timeout context for this promptCtx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) @@ -331,7 +343,12 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta return fmt.Errorf("prompt failed: %w", err) } - // Check if agent needs input - if so, publish input_required event + // Extract accumulated message from buffer + instance.messageMu.Lock() + agentMessage := instance.messageBuffer.String() + instance.messageBuffer.Reset() + instance.messageMu.Unlock() + stopReason := "" if resp != nil { stopReason = string(resp.StopReason) @@ -340,20 +357,16 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta zap.String("instance_id", instance.ID), zap.String("stop_reason", stopReason)) - if resp != nil && resp.StopReason == "needs_input" { - m.logger.Info("agent requesting user input on initial prompt", + // Publish prompt_complete event with the agent's response (for saving as comment) + m.publishPromptComplete(instance, agentMessage) + + // Prompt completed - mark agent as READY for follow-up prompts + m.logger.Info("agent ready for follow-up prompts", + 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.String("task_id", instance.TaskID)) - m.publishInputRequiredFromPrompt(instance, "The agent needs more information to proceed.") - } else { - // Prompt completed - mark agent as READY for follow-up prompts - m.logger.Info("agent ready for follow-up prompts", - 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)) - } + zap.Error(err)) } } else { m.logger.Warn("no task description provided, marking as ready", @@ -382,6 +395,74 @@ func (m *Manager) handleUpdatesStream(instance *AgentInstance) { } } +// handlePermissionStream handles streaming permission requests from the agent +func (m *Manager) handlePermissionStream(instance *AgentInstance) { + ctx := context.Background() + + err := instance.agentctl.StreamPermissions(ctx, func(notification *agentctl.PermissionNotification) { + m.handlePermissionNotification(instance, notification) + }) + if err != nil { + m.logger.Error("failed to connect to permission stream", + zap.String("instance_id", instance.ID), + zap.Error(err)) + } +} + +// handlePermissionNotification processes incoming permission requests from the agent +func (m *Manager) handlePermissionNotification(instance *AgentInstance, notification *agentctl.PermissionNotification) { + m.logger.Info("received permission request", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID), + zap.String("pending_id", notification.PendingID), + zap.String("title", notification.Title), + zap.Int("num_options", len(notification.Options))) + + // Publish permission request event to the event bus + m.publishPermissionRequest(instance, notification) +} + +// publishPermissionRequest publishes a permission request event +func (m *Manager) publishPermissionRequest(instance *AgentInstance, notification *agentctl.PermissionNotification) { + // Convert options to a serializable format + options := make([]map[string]interface{}, len(notification.Options)) + for i, opt := range notification.Options { + options[i] = map[string]interface{}{ + "option_id": opt.OptionID, + "name": opt.Name, + "kind": opt.Kind, + } + } + + data := map[string]interface{}{ + "type": "permission_request", + "task_id": instance.TaskID, + "agent_instance_id": instance.ID, + "pending_id": notification.PendingID, + "session_id": notification.SessionID, + "tool_call_id": notification.ToolCallID, + "title": notification.Title, + "options": options, + "created_at": notification.CreatedAt, + "timestamp": time.Now().UTC().Format(time.RFC3339), + } + + event := bus.NewEvent(events.ACPMessage, "agent-manager", data) + subject := events.BuildACPSubject(instance.TaskID) + + if err := m.eventBus.Publish(context.Background(), subject, event); err != nil { + m.logger.Error("failed to publish permission_request event", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID), + zap.Error(err)) + } else { + m.logger.Info("published permission_request event", + zap.String("task_id", instance.TaskID), + zap.String("pending_id", notification.PendingID), + zap.String("title", notification.Title)) + } +} + // handleSessionNotification processes incoming session notifications from the agent func (m *Manager) handleSessionNotification(instance *AgentInstance, notification agentctl.SessionNotification) { m.logger.Debug("received session notification", @@ -395,6 +476,13 @@ func (m *Manager) handleSessionNotification(instance *AgentInstance, notificatio m.logger.Debug("agent message chunk", zap.String("instance_id", instance.ID)) m.updateInstanceProgress(instance.ID, 50) + + // Accumulate message content for saving as comment when prompt completes + if update.AgentMessageChunk.Content.Text != nil { + instance.messageMu.Lock() + instance.messageBuffer.WriteString(update.AgentMessageChunk.Content.Text.Text) + instance.messageMu.Unlock() + } } if update.AgentThoughtChunk != nil { @@ -465,39 +553,6 @@ func (m *Manager) publishSessionNotification(instance *AgentInstance, notificati } } -// publishInputRequiredFromPrompt publishes an input_required event when the agent needs user input -// This is called after a prompt completes with stopReason="needs_input" -func (m *Manager) publishInputRequiredFromPrompt(instance *AgentInstance, message string) { - if m.eventBus == nil { - return - } - - // Structure must match protocol.Message with type "input_required" - data := map[string]interface{}{ - "type": "input_required", - "timestamp": time.Now().UTC().Format(time.RFC3339Nano), - "agent_id": instance.ID, - "task_id": instance.TaskID, - "data": map[string]interface{}{ - "message": message, - }, - } - - event := bus.NewEvent(events.ACPMessage, "agent-manager", data) - subject := events.BuildACPSubject(instance.TaskID) - - if err := m.eventBus.Publish(context.Background(), subject, event); err != nil { - m.logger.Error("failed to publish input_required event", - zap.String("instance_id", instance.ID), - zap.String("task_id", instance.TaskID), - zap.Error(err)) - } else { - m.logger.Info("published input_required event", - zap.String("task_id", instance.TaskID), - zap.String("message", message)) - } -} - // publishSessionInfo publishes a session_info event with the session ID // This allows the orchestrator to store the session ID in task metadata for resumption func (m *Manager) publishSessionInfo(instance *AgentInstance, sessionID string) { @@ -533,6 +588,42 @@ func (m *Manager) publishSessionInfo(instance *AgentInstance, sessionID string) } } +// publishPromptComplete publishes a prompt_complete event when an agent finishes responding +// This is used to save the agent's response as a comment on the task +func (m *Manager) publishPromptComplete(instance *AgentInstance, agentMessage string) { + if m.eventBus == nil { + return + } + + // Only publish if there's actual content + if agentMessage == "" { + return + } + + data := map[string]interface{}{ + "type": "prompt_complete", + "timestamp": time.Now().UTC().Format(time.RFC3339Nano), + "agent_id": instance.ID, + "task_id": instance.TaskID, + "agent_message": agentMessage, + } + + event := bus.NewEvent(events.PromptComplete, "agent-manager", data) + // Publish on task-specific subject so orchestrator can subscribe + subject := events.PromptComplete + "." + instance.TaskID + + if err := m.eventBus.Publish(context.Background(), subject, event); err != nil { + m.logger.Error("failed to publish prompt_complete event", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID), + zap.Error(err)) + } else { + m.logger.Info("published prompt_complete event", + zap.String("task_id", instance.TaskID), + zap.Int("message_length", len(agentMessage))) + } +} + // updateInstanceProgress updates an instance's progress func (m *Manager) updateInstanceProgress(instanceID string, progress int) { m.mu.Lock() @@ -684,8 +775,8 @@ func (m *Manager) expandMountTemplate(source, workspacePath, taskID string) stri // PromptAgent sends a follow-up prompt to a running agent // PromptResult contains the result of a prompt operation type PromptResult struct { - StopReason string // The reason the agent stopped (e.g., "end_turn", "needs_input") - NeedsInput bool // True if the agent is requesting user input + StopReason string // The reason the agent stopped (e.g., "end_turn") + AgentMessage string // The agent's accumulated response message } // PromptAgent sends a follow-up prompt to a running agent @@ -710,6 +801,12 @@ func (m *Manager) PromptAgent(ctx context.Context, instanceID string, prompt str // Set status to RUNNING while processing instance.Status = v1.AgentStatusRunning + + // Clear message buffer before starting new prompt + instance.messageMu.Lock() + instance.messageBuffer.Reset() + instance.messageMu.Unlock() + m.mu.Unlock() m.logger.Info("sending prompt to agent", @@ -722,24 +819,25 @@ func (m *Manager) PromptAgent(ctx context.Context, instanceID string, prompt str return nil, err } + // Extract accumulated message from buffer + instance.messageMu.Lock() + agentMessage := instance.messageBuffer.String() + instance.messageBuffer.Reset() + instance.messageMu.Unlock() + result := &PromptResult{ - StopReason: string(resp.StopReason), - NeedsInput: resp.StopReason == "needs_input", + StopReason: string(resp.StopReason), + AgentMessage: agentMessage, } - // Check if agent needs input - if so, publish input_required event - if result.NeedsInput { - m.logger.Info("agent requesting user input", + // Publish prompt_complete event with the agent's response (for saving as comment) + m.publishPromptComplete(instance, agentMessage) + + // 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.String("task_id", instance.TaskID)) - m.publishInputRequiredFromPrompt(instance, "The agent needs more information to proceed.") - } else { - // Prompt completed normally - 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)) - } + zap.Error(err)) } return result, nil @@ -1076,4 +1174,50 @@ func (m *Manager) performCleanup(ctx context.Context) { } } } +} + +// RespondToPermission sends a response to a permission request for an agent instance +func (m *Manager) RespondToPermission(instanceID, pendingID, optionID string, cancelled bool) error { + m.mu.RLock() + instance, exists := m.instances[instanceID] + m.mu.RUnlock() + + if !exists { + return fmt.Errorf("agent instance not found: %s", instanceID) + } + + if instance.agentctl == nil { + return fmt.Errorf("agent instance has no agentctl client: %s", instanceID) + } + + m.logger.Info("responding to permission request", + zap.String("instance_id", instanceID), + zap.String("pending_id", pendingID), + zap.String("option_id", optionID), + zap.Bool("cancelled", cancelled)) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + return instance.agentctl.RespondToPermission(ctx, pendingID, optionID, cancelled) +} + +// RespondToPermissionByTaskID sends a response to a permission request for a task +func (m *Manager) RespondToPermissionByTaskID(taskID, pendingID, optionID string, cancelled bool) error { + // Find the instance by task ID + m.mu.RLock() + var instance *AgentInstance + for _, inst := range m.instances { + if inst.TaskID == taskID { + instance = inst + break + } + } + m.mu.RUnlock() + + if instance == nil { + return fmt.Errorf("no agent instance found for task: %s", taskID) + } + + return m.RespondToPermission(instance.ID, pendingID, optionID, cancelled) } \ No newline at end of file diff --git a/apps/backend/internal/agent/registry/defaults.go b/apps/backend/internal/agent/registry/defaults.go index 1f67a1d381..bf8af1053b 100644 --- a/apps/backend/internal/agent/registry/defaults.go +++ b/apps/backend/internal/agent/registry/defaults.go @@ -11,6 +11,11 @@ func DefaultAgents() []*AgentTypeConfig { Tag: "latest", WorkingDir: "/workspace", RequiredEnv: []string{"AUGMENT_SESSION_AUTH"}, + Env: map[string]string{ + // Auto-approve permissions for now (until full permission flow is implemented) + // This allows the agent to proceed without waiting for user approval + "AGENTCTL_AUTO_APPROVE_PERMISSIONS": "true", + }, Mounts: []MountTemplate{ {Source: "{workspace}", Target: "/workspace", ReadOnly: false}, {Source: "{augment_sessions}", Target: "/root/.augment/sessions", ReadOnly: false}, diff --git a/apps/backend/internal/agentctl/acp/client.go b/apps/backend/internal/agentctl/acp/client.go index 4fa784db12..cd3f9b4518 100644 --- a/apps/backend/internal/agentctl/acp/client.go +++ b/apps/backend/internal/agentctl/acp/client.go @@ -16,13 +16,39 @@ import ( // UpdateHandler is called when session updates are received from the agent type UpdateHandler func(notification acp.SessionNotification) +// PermissionRequestHandler is called when the agent requests permission +// Returns the selected option ID, or empty string with cancelled=true to cancel +type PermissionRequestHandler func(ctx context.Context, req *PermissionRequest) (*PermissionResponse, error) + +// PermissionRequest represents a permission request from the agent +type PermissionRequest struct { + SessionID string `json:"session_id"` + ToolCallID string `json:"tool_call_id"` + Title string `json:"title"` + Options []PermissionOption `json:"options"` +} + +// PermissionOption represents a permission choice +type PermissionOption struct { + OptionID string `json:"option_id"` + Name string `json:"name"` + Kind string `json:"kind"` // allow_once, allow_always, reject_once, reject_always +} + +// PermissionResponse is the user's response to a permission request +type PermissionResponse struct { + OptionID string `json:"option_id,omitempty"` + Cancelled bool `json:"cancelled,omitempty"` +} + // Client implements acp.Client interface and handles all agent requests type Client struct { logger *zap.Logger workspaceRoot string - mu sync.RWMutex - updateHandler UpdateHandler + mu sync.RWMutex + updateHandler UpdateHandler + permissionHandler PermissionRequestHandler } // ClientOption configures a Client @@ -49,6 +75,13 @@ func WithUpdateHandler(h UpdateHandler) ClientOption { } } +// WithPermissionHandler sets the handler for permission requests +func WithPermissionHandler(h PermissionRequestHandler) ClientOption { + return func(c *Client) { + c.permissionHandler = h + } +} + // NewClient creates a new ACP client implementation func NewClient(opts ...ClientOption) *Client { c := &Client{ @@ -68,8 +101,16 @@ func (c *Client) SetUpdateHandler(h UpdateHandler) { c.updateHandler = h } +// SetPermissionHandler sets the permission request handler (thread-safe) +func (c *Client) SetPermissionHandler(h PermissionRequestHandler) { + c.mu.Lock() + defer c.mu.Unlock() + c.permissionHandler = h +} + // RequestPermission handles permission requests from the agent -// Auto-approves by selecting the first "allow" option +// If a permission handler is set, it forwards the request to the handler. +// Otherwise, auto-approves by selecting the first "allow" option. func (c *Client) RequestPermission(ctx context.Context, p acp.RequestPermissionRequest) (acp.RequestPermissionResponse, error) { title := "" if p.ToolCall.Title != nil { @@ -91,6 +132,81 @@ func (c *Client) RequestPermission(ctx context.Context, p acp.RequestPermissionR }, nil } + // Check if we have a permission handler + c.mu.RLock() + handler := c.permissionHandler + c.mu.RUnlock() + + if handler != nil { + // Forward to external handler (e.g., backend/user) + return c.forwardPermissionRequest(ctx, handler, p) + } + + // Fall back to auto-approve + return c.autoApprovePermission(p) +} + +// forwardPermissionRequest forwards the permission request to an external handler +func (c *Client) forwardPermissionRequest(ctx context.Context, handler PermissionRequestHandler, p acp.RequestPermissionRequest) (acp.RequestPermissionResponse, error) { + // Convert ACP types to our types + options := make([]PermissionOption, len(p.Options)) + for i, opt := range p.Options { + options[i] = PermissionOption{ + OptionID: string(opt.OptionId), + Name: opt.Name, + Kind: string(opt.Kind), + } + } + + title := "" + if p.ToolCall.Title != nil { + title = *p.ToolCall.Title + } + + req := &PermissionRequest{ + SessionID: string(p.SessionId), + ToolCallID: string(p.ToolCall.ToolCallId), + Title: title, + Options: options, + } + + c.logger.Info("forwarding permission request to handler", + zap.String("session_id", req.SessionID), + zap.String("tool_call_id", req.ToolCallID)) + + resp, err := handler(ctx, req) + if err != nil { + c.logger.Error("permission handler failed", zap.Error(err)) + // On error, cancel the permission request + return acp.RequestPermissionResponse{ + Outcome: acp.RequestPermissionOutcome{ + Cancelled: &acp.RequestPermissionOutcomeCancelled{}, + }, + }, nil + } + + if resp.Cancelled { + c.logger.Info("permission request cancelled by user") + return acp.RequestPermissionResponse{ + Outcome: acp.RequestPermissionOutcome{ + Cancelled: &acp.RequestPermissionOutcomeCancelled{}, + }, + }, nil + } + + c.logger.Info("permission request approved by user", + zap.String("option_id", resp.OptionID)) + return acp.RequestPermissionResponse{ + Outcome: acp.RequestPermissionOutcome{ + Selected: &acp.RequestPermissionOutcomeSelected{ + OptionId: acp.PermissionOptionId(resp.OptionID), + }, + }, + }, nil +} + +// autoApprovePermission auto-approves by selecting the first "allow" option +func (c *Client) autoApprovePermission(p acp.RequestPermissionRequest) (acp.RequestPermissionResponse, error) { // Find the first "allow" option var selectedOption *acp.PermissionOption for i := range p.Options { diff --git a/apps/backend/internal/agentctl/api/acp.go b/apps/backend/internal/agentctl/api/acp.go index c9ac302375..fa1294505d 100644 --- a/apps/backend/internal/agentctl/api/acp.go +++ b/apps/backend/internal/agentctl/api/acp.go @@ -233,3 +233,123 @@ func (s *Server) handleACPStreamWS(c *gin.Context) { } } +// PermissionRespondRequest is a request to respond to a permission request +type PermissionRespondRequest struct { + PendingID string `json:"pending_id" binding:"required"` + OptionID string `json:"option_id,omitempty"` + Cancelled bool `json:"cancelled,omitempty"` +} + +// PermissionRespondResponse is the response to a permission respond call +type PermissionRespondResponse struct { + Success bool `json:"success"` + Error string `json:"error,omitempty"` +} + +func (s *Server) handlePermissionRespond(c *gin.Context) { + var req PermissionRespondRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, PermissionRespondResponse{ + Success: false, + Error: "invalid request: " + err.Error(), + }) + return + } + + s.logger.Info("received permission response", + zap.String("pending_id", req.PendingID), + zap.String("option_id", req.OptionID), + zap.Bool("cancelled", req.Cancelled)) + + if err := s.procMgr.RespondToPermission(req.PendingID, req.OptionID, req.Cancelled); err != nil { + s.logger.Error("failed to respond to permission", zap.Error(err)) + c.JSON(http.StatusNotFound, PermissionRespondResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + c.JSON(http.StatusOK, PermissionRespondResponse{ + Success: true, + }) +} + +// PendingPermissionResponse represents a pending permission for the API +type PendingPermissionResponse struct { + PendingID string `json:"pending_id"` + SessionID string `json:"session_id"` + ToolCallID string `json:"tool_call_id"` + Title string `json:"title"` + Options []PermissionOptionJSON `json:"options"` + CreatedAt string `json:"created_at"` +} + +// PermissionOptionJSON is the JSON representation of a permission option +type PermissionOptionJSON struct { + OptionID string `json:"option_id"` + Name string `json:"name"` + Kind string `json:"kind"` +} + +func (s *Server) handleGetPendingPermissions(c *gin.Context) { + pending := s.procMgr.GetPendingPermissions() + + result := make([]PendingPermissionResponse, 0, len(pending)) + for _, p := range pending { + options := make([]PermissionOptionJSON, len(p.Request.Options)) + for i, opt := range p.Request.Options { + options[i] = PermissionOptionJSON{ + OptionID: opt.OptionID, + Name: opt.Name, + Kind: opt.Kind, + } + } + result = append(result, PendingPermissionResponse{ + PendingID: p.ID, + SessionID: p.Request.SessionID, + ToolCallID: p.Request.ToolCallID, + Title: p.Request.Title, + Options: options, + CreatedAt: p.CreatedAt.Format(time.RFC3339), + }) + } + + c.JSON(http.StatusOK, result) +} + +// handlePermissionStreamWS streams permission request notifications via WebSocket +func (s *Server) handlePermissionStreamWS(c *gin.Context) { + conn, err := s.upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + s.logger.Error("WebSocket upgrade failed", zap.Error(err)) + return + } + defer conn.Close() + + s.logger.Info("Permission stream WebSocket connected") + + // Get the permission requests channel + permissionCh := s.procMgr.GetPermissionRequests() + + // Stream permission notifications to WebSocket + for { + select { + case notification, ok := <-permissionCh: + if !ok { + return + } + + data, err := json.Marshal(notification) + if err != nil { + s.logger.Error("failed to marshal permission notification", zap.Error(err)) + continue + } + + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + s.logger.Debug("WebSocket write error", zap.Error(err)) + return + } + } + } +} diff --git a/apps/backend/internal/agentctl/api/server.go b/apps/backend/internal/agentctl/api/server.go index 859824ba4d..c1bed90543 100644 --- a/apps/backend/internal/agentctl/api/server.go +++ b/apps/backend/internal/agentctl/api/server.go @@ -69,6 +69,11 @@ func (s *Server) setupRoutes() { api.POST("/acp/prompt", s.handleACPPrompt) api.GET("/acp/stream", s.handleACPStreamWS) + // Permission request handling + api.GET("/acp/permissions", s.handleGetPendingPermissions) + api.GET("/acp/permissions/stream", s.handlePermissionStreamWS) + api.POST("/acp/permissions/respond", s.handlePermissionRespond) + // Output streaming api.GET("/output", s.handleGetOutput) api.GET("/output/stream", s.handleOutputStreamWS) diff --git a/apps/backend/internal/agentctl/config/config.go b/apps/backend/internal/agentctl/config/config.go index ea99ab019e..98fa58976d 100644 --- a/apps/backend/internal/agentctl/config/config.go +++ b/apps/backend/internal/agentctl/config/config.go @@ -27,6 +27,9 @@ type Config struct { // Auto-start the agent on agentctl startup AutoStart bool + // Auto-approve permission requests (for testing/CI) + AutoApprovePermissions bool + // Logging configuration LogLevel string LogFormat string @@ -44,14 +47,15 @@ func Load() *Config { defaultCmd := "auggie --acp --workspace-root " + workDir cfg := &Config{ - Port: getEnvInt("AGENTCTL_PORT", 9999), - AgentCommand: getEnv("AGENTCTL_AGENT_COMMAND", defaultCmd), - WorkDir: workDir, - AutoStart: getEnvBool("AGENTCTL_AUTO_START", false), - LogLevel: getEnv("AGENTCTL_LOG_LEVEL", "info"), - LogFormat: getEnv("AGENTCTL_LOG_FORMAT", "json"), - OutputBufferSize: getEnvInt("AGENTCTL_OUTPUT_BUFFER_SIZE", 1000), - HealthCheckInterval: getEnvInt("AGENTCTL_HEALTH_CHECK_INTERVAL", 5), + Port: getEnvInt("AGENTCTL_PORT", 9999), + AgentCommand: getEnv("AGENTCTL_AGENT_COMMAND", defaultCmd), + WorkDir: workDir, + AutoStart: getEnvBool("AGENTCTL_AUTO_START", false), + AutoApprovePermissions: getEnvBool("AGENTCTL_AUTO_APPROVE_PERMISSIONS", false), + LogLevel: getEnv("AGENTCTL_LOG_LEVEL", "info"), + LogFormat: getEnv("AGENTCTL_LOG_FORMAT", "json"), + OutputBufferSize: getEnvInt("AGENTCTL_OUTPUT_BUFFER_SIZE", 1000), + HealthCheckInterval: getEnvInt("AGENTCTL_HEALTH_CHECK_INTERVAL", 5), } // Parse agent command into args diff --git a/apps/backend/internal/agentctl/process/manager.go b/apps/backend/internal/agentctl/process/manager.go index 47598ecbf0..6fbf9b1e02 100644 --- a/apps/backend/internal/agentctl/process/manager.go +++ b/apps/backend/internal/agentctl/process/manager.go @@ -39,6 +39,24 @@ type errorWrapper struct { err error } +// PendingPermission represents a permission request waiting for user response +type PendingPermission struct { + ID string + Request *acpclient.PermissionRequest + ResponseCh chan *acpclient.PermissionResponse + CreatedAt time.Time +} + +// PermissionNotification is sent when the agent requests permission +type PermissionNotification struct { + PendingID string `json:"pending_id"` + SessionID string `json:"session_id"` + ToolCallID string `json:"tool_call_id"` + Title string `json:"title"` + Options []acpclient.PermissionOption `json:"options"` + CreatedAt time.Time `json:"created_at"` +} + // Manager manages the agent subprocess type Manager struct { cfg *config.Config @@ -64,6 +82,13 @@ type Manager struct { // Session update notifications updatesCh chan acp.SessionNotification + // Permission request notifications (sent to backend) + permissionCh chan *PermissionNotification + + // Pending permission requests waiting for user response + pendingPermissions map[string]*PendingPermission + permissionMu sync.RWMutex + // Synchronization mu sync.RWMutex wg sync.WaitGroup @@ -75,10 +100,12 @@ type Manager struct { // NewManager creates a new process manager func NewManager(cfg *config.Config, log *logger.Logger) *Manager { m := &Manager{ - cfg: cfg, - logger: log.WithFields(zap.String("component", "process-manager")), - outputBuffer: NewOutputBuffer(cfg.OutputBufferSize), - updatesCh: make(chan acp.SessionNotification, 100), + cfg: cfg, + logger: log.WithFields(zap.String("component", "process-manager")), + outputBuffer: NewOutputBuffer(cfg.OutputBufferSize), + updatesCh: make(chan acp.SessionNotification, 100), + permissionCh: make(chan *PermissionNotification, 10), + pendingPermissions: make(map[string]*PendingPermission), } m.status.Store(StatusStopped) m.exitCode.Store(-1) @@ -179,6 +206,7 @@ func (m *Manager) Start(ctx context.Context) error { m.logger.Warn("updates channel full, dropping notification") } }), + acpclient.WithPermissionHandler(m.handlePermissionRequest), ) // Create ACP SDK connection - it handles JSON-RPC communication @@ -329,3 +357,172 @@ func (m *Manager) GetProcessInfo() map[string]interface{} { return info } +// handlePermissionRequest handles permission requests from the agent +// It stores the pending request and waits for a response from the backend +func (m *Manager) handlePermissionRequest(ctx context.Context, req *acpclient.PermissionRequest) (*acpclient.PermissionResponse, error) { + // Generate a unique ID for this permission request + pendingID := fmt.Sprintf("%s-%s-%d", req.SessionID, req.ToolCallID, time.Now().UnixNano()) + + m.logger.Info("handling permission request", + zap.String("pending_id", pendingID), + zap.String("session_id", req.SessionID), + zap.String("tool_call_id", req.ToolCallID), + zap.String("title", req.Title), + zap.Bool("auto_approve", m.cfg.AutoApprovePermissions)) + + // If auto-approve is enabled, immediately approve with the first "allow" option + if m.cfg.AutoApprovePermissions { + return m.autoApprovePermission(req) + } + + // Create pending permission with response channel + pending := &PendingPermission{ + ID: pendingID, + Request: req, + ResponseCh: make(chan *acpclient.PermissionResponse, 1), + CreatedAt: time.Now(), + } + + // Store pending permission + m.permissionMu.Lock() + m.pendingPermissions[pendingID] = pending + m.permissionMu.Unlock() + + // Clean up when done + defer func() { + m.permissionMu.Lock() + delete(m.pendingPermissions, pendingID) + m.permissionMu.Unlock() + }() + + // Send notification to backend via updates channel + // We use a custom notification type that the backend will recognize + m.sendPermissionNotification(pending) + + // Wait for response with timeout + select { + case resp := <-pending.ResponseCh: + m.logger.Info("received permission response", + zap.String("pending_id", pendingID), + zap.String("option_id", resp.OptionID), + zap.Bool("cancelled", resp.Cancelled)) + return resp, nil + case <-ctx.Done(): + m.logger.Warn("permission request context cancelled", + zap.String("pending_id", pendingID)) + return &acpclient.PermissionResponse{Cancelled: true}, nil + case <-time.After(5 * time.Minute): + m.logger.Warn("permission request timed out", + zap.String("pending_id", pendingID)) + return &acpclient.PermissionResponse{Cancelled: true}, nil + } +} + +// autoApprovePermission automatically approves a permission request +// by selecting the first "allow" option, or the first option if no allow option exists +func (m *Manager) autoApprovePermission(req *acpclient.PermissionRequest) (*acpclient.PermissionResponse, error) { + if len(req.Options) == 0 { + m.logger.Warn("no options available for auto-approve, cancelling") + return &acpclient.PermissionResponse{Cancelled: true}, nil + } + + // Find the first "allow" option + var selectedOption *acpclient.PermissionOption + for i := range req.Options { + opt := &req.Options[i] + if opt.Kind == "allow_once" || opt.Kind == "allow_always" { + selectedOption = opt + break + } + } + + // If no allow option, use the first option + if selectedOption == nil { + selectedOption = &req.Options[0] + } + + m.logger.Info("auto-approving permission request", + zap.String("option_id", selectedOption.OptionID), + zap.String("option_name", selectedOption.Name), + zap.String("kind", selectedOption.Kind)) + + return &acpclient.PermissionResponse{ + OptionID: selectedOption.OptionID, + }, nil +} + +// sendPermissionNotification sends a permission request notification to the backend +func (m *Manager) sendPermissionNotification(pending *PendingPermission) { + notification := &PermissionNotification{ + PendingID: pending.ID, + SessionID: pending.Request.SessionID, + ToolCallID: pending.Request.ToolCallID, + Title: pending.Request.Title, + Options: pending.Request.Options, + CreatedAt: pending.CreatedAt, + } + + m.logger.Info("sending permission notification to backend", + zap.String("pending_id", pending.ID), + zap.String("title", pending.Request.Title)) + + select { + case m.permissionCh <- notification: + // Sent successfully + default: + m.logger.Warn("permission channel full, dropping notification", + zap.String("pending_id", pending.ID)) + } +} + +// GetPermissionRequests returns the channel for permission request notifications +func (m *Manager) GetPermissionRequests() <-chan *PermissionNotification { + return m.permissionCh +} + +// RespondToPermission responds to a pending permission request +func (m *Manager) RespondToPermission(pendingID string, optionID string, cancelled bool) error { + m.permissionMu.RLock() + pending, ok := m.pendingPermissions[pendingID] + m.permissionMu.RUnlock() + + if !ok { + return fmt.Errorf("pending permission not found: %s", pendingID) + } + + m.logger.Info("responding to permission request", + zap.String("pending_id", pendingID), + zap.String("option_id", optionID), + zap.Bool("cancelled", cancelled)) + + // Send response (non-blocking since channel is buffered) + select { + case pending.ResponseCh <- &acpclient.PermissionResponse{ + OptionID: optionID, + Cancelled: cancelled, + }: + return nil + default: + return fmt.Errorf("response channel full for pending permission: %s", pendingID) + } +} + +// GetPendingPermissions returns all pending permission requests +func (m *Manager) GetPendingPermissions() []*PendingPermission { + m.permissionMu.RLock() + defer m.permissionMu.RUnlock() + + result := make([]*PendingPermission, 0, len(m.pendingPermissions)) + for _, p := range m.pendingPermissions { + result = append(result, p) + } + return result +} + +// GetPendingPermission returns a specific pending permission request +func (m *Manager) GetPendingPermission(pendingID string) (*PendingPermission, bool) { + m.permissionMu.RLock() + defer m.permissionMu.RUnlock() + p, ok := m.pendingPermissions[pendingID] + return p, ok +} diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index 9b2296eb1a..3a5ba0a094 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -64,6 +64,11 @@ const ( ACPMessage = "acp.message" // Base subject for ACP messages ) +// Event types for agent prompts +const ( + PromptComplete = "prompt.complete" // Agent finished responding to a prompt +) + // BuildACPSubject creates an ACP subject for a specific task func BuildACPSubject(taskID string) string { return ACPMessage + "." + taskID diff --git a/apps/backend/internal/integration/orchestrator_test.go b/apps/backend/internal/integration/orchestrator_test.go index 60f3a991ec..65f6d1e922 100644 --- a/apps/backend/internal/integration/orchestrator_test.go +++ b/apps/backend/internal/integration/orchestrator_test.go @@ -19,6 +19,7 @@ import ( "github.com/gorilla/websocket" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "go.uber.org/zap" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events" @@ -343,10 +344,19 @@ func (s *SimulatedAgentManagerClient) PromptAgent(ctx context.Context, agentInst return &executor.PromptResult{ StopReason: "end_turn", - NeedsInput: false, }, nil } +// RespondToPermissionByTaskID responds to a permission request for a task +func (s *SimulatedAgentManagerClient) RespondToPermissionByTaskID(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error { + s.logger.Info("simulated: responding to permission", + zap.String("task_id", taskID), + zap.String("pending_id", pendingID), + zap.String("option_id", optionID), + zap.Bool("cancelled", cancelled)) + return nil +} + // CompleteAgent marks an agent as completed func (s *SimulatedAgentManagerClient) CompleteAgent(instanceID string) { s.mu.Lock() diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index d6435ff374..ea5f22105e 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -22,8 +22,8 @@ var ( // PromptResult contains the result of a prompt operation type PromptResult struct { - StopReason string // The reason the agent stopped (e.g., "end_turn", "needs_input") - NeedsInput bool // True if the agent is requesting user input + StopReason string // The reason the agent stopped (e.g., "end_turn") + AgentMessage string // The agent's accumulated response message } // AgentManagerClient is an interface for the Agent Manager service @@ -44,6 +44,9 @@ type AgentManagerClient interface { // PromptAgent sends a prompt to a running agent // Returns PromptResult indicating if the agent needs input PromptAgent(ctx context.Context, agentInstanceID string, prompt string) (*PromptResult, error) + + // RespondToPermission sends a response to a permission request + RespondToPermissionByTaskID(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error } // LaunchAgentRequest contains parameters for launching an agent @@ -228,6 +231,17 @@ func (e *Executor) Prompt(ctx context.Context, taskID string, prompt string) (*P return e.agentManager.PromptAgent(ctx, agentInstanceID, prompt) } +// RespondToPermission sends a response to a permission request for a task +func (e *Executor) RespondToPermission(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error { + e.logger.Info("responding to permission request", + zap.String("task_id", taskID), + zap.String("pending_id", pendingID), + zap.String("option_id", optionID), + zap.Bool("cancelled", cancelled)) + + return e.agentManager.RespondToPermissionByTaskID(ctx, taskID, pendingID, optionID, cancelled) +} + // GetExecution returns the current execution state for a task func (e *Executor) GetExecution(taskID string) (*TaskExecution, bool) { e.mu.RLock() @@ -380,5 +394,15 @@ func (m *MockAgentManagerClient) PromptAgent(ctx context.Context, agentInstanceI m.logger.Info("mock: prompting agent", zap.String("agent_instance_id", agentInstanceID), zap.Int("prompt_length", len(prompt))) - return &PromptResult{StopReason: "end_turn", NeedsInput: false}, nil + return &PromptResult{StopReason: "end_turn"}, nil +} + +// RespondToPermissionByTaskID mocks responding to a permission request +func (m *MockAgentManagerClient) RespondToPermissionByTaskID(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error { + m.logger.Info("mock: responding to permission", + zap.String("task_id", taskID), + zap.String("pending_id", pendingID), + zap.String("option_id", optionID), + zap.Bool("cancelled", cancelled)) + return nil } diff --git a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go index c5e0706ae9..c2cc1adbbd 100644 --- a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go +++ b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go @@ -56,7 +56,11 @@ func (m *mockAgentManager) ListAgentTypes(ctx context.Context) ([]*v1.AgentType, } func (m *mockAgentManager) PromptAgent(ctx context.Context, agentInstanceID string, prompt string) (*executor.PromptResult, error) { - return &executor.PromptResult{StopReason: "end_turn", NeedsInput: false}, nil + return &executor.PromptResult{StopReason: "end_turn"}, nil +} + +func (m *mockAgentManager) RespondToPermissionByTaskID(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error { + return nil } // testTaskRepository is an in-memory task repository for testing diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 941e9fa0c5..11a1f7e904 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -44,6 +44,11 @@ func DefaultServiceConfig() ServiceConfig { // InputRequestHandler is called when an agent requests user input type InputRequestHandler func(ctx context.Context, taskID, agentID, message string) error +// CommentCreator is an interface for creating comments on tasks +type CommentCreator interface { + CreateAgentComment(ctx context.Context, taskID, content string) error +} + // Service is the main orchestrator service type Service struct { config ServiceConfig @@ -65,6 +70,9 @@ type Service struct { // Input request handler (for agent-user conversation) inputRequestHandler InputRequestHandler + // Comment creator for saving agent responses + commentCreator CommentCreator + // Service state mu sync.RWMutex running bool @@ -122,12 +130,18 @@ func NewService( OnAgentCompleted: s.handleAgentCompleted, OnAgentFailed: s.handleAgentFailed, OnACPMessage: s.handleACPMessage, + OnPromptComplete: s.handlePromptComplete, } s.watcher = watcher.NewWatcher(eventBus, handlers, log) return s } +// SetCommentCreator sets the comment creator for saving agent responses +func (s *Service) SetCommentCreator(cc CommentCreator) { + s.commentCreator = cc +} + // Start starts all orchestrator components func (s *Service) Start(ctx context.Context) error { s.mu.Lock() @@ -273,12 +287,11 @@ func (s *Service) StopTask(ctx context.Context, taskID string, reason string, fo // PromptResult contains the result of a prompt operation type PromptResult struct { - StopReason string // The reason the agent stopped (e.g., "end_turn", "needs_input") - NeedsInput bool // True if the agent is requesting user input + StopReason string // The reason the agent stopped (e.g., "end_turn") + AgentMessage string // The agent's accumulated response message } // PromptTask sends a follow-up prompt to a running agent for a task -// Returns PromptResult indicating if the agent needs input func (s *Service) PromptTask(ctx context.Context, taskID string, prompt string) (*PromptResult, error) { s.logger.Info("sending prompt to task agent", zap.String("task_id", taskID), @@ -288,11 +301,21 @@ func (s *Service) PromptTask(ctx context.Context, taskID string, prompt string) return nil, err } return &PromptResult{ - StopReason: result.StopReason, - NeedsInput: result.NeedsInput, + StopReason: result.StopReason, + AgentMessage: result.AgentMessage, }, nil } +// RespondToPermission sends a response to a permission request for a task +func (s *Service) RespondToPermission(ctx context.Context, taskID, pendingID, optionID string, cancelled bool) error { + s.logger.Info("responding to permission request", + zap.String("task_id", taskID), + zap.String("pending_id", pendingID), + zap.String("option_id", optionID), + zap.Bool("cancelled", cancelled)) + return s.executor.RespondToPermission(ctx, taskID, pendingID, optionID, cancelled) +} + // CompleteTask explicitly completes a task and stops its agent func (s *Service) CompleteTask(ctx context.Context, taskID string) error { s.logger.Info("completing task", @@ -490,3 +513,23 @@ func (s *Service) handleInputRequired(ctx context.Context, taskID string, msg *p } } +// handlePromptComplete handles prompt complete events and saves agent response as comment +func (s *Service) handlePromptComplete(ctx context.Context, data watcher.PromptCompleteData) { + s.logger.Info("handling prompt complete", + zap.String("task_id", data.TaskID), + zap.Int("message_length", len(data.AgentMessage))) + + // Save agent response as a comment if we have a comment creator + if s.commentCreator != nil && data.AgentMessage != "" { + if err := s.commentCreator.CreateAgentComment(ctx, data.TaskID, data.AgentMessage); err != nil { + s.logger.Error("failed to create agent comment", + zap.String("task_id", data.TaskID), + zap.Error(err)) + } else { + s.logger.Info("created agent comment", + zap.String("task_id", data.TaskID), + zap.Int("message_length", len(data.AgentMessage))) + } + } +} + diff --git a/apps/backend/internal/orchestrator/watcher/watcher.go b/apps/backend/internal/orchestrator/watcher/watcher.go index a3253c7b74..bd7b49224d 100644 --- a/apps/backend/internal/orchestrator/watcher/watcher.go +++ b/apps/backend/internal/orchestrator/watcher/watcher.go @@ -33,6 +33,13 @@ type AgentEventData struct { ErrorMessage string `json:"error_message,omitempty"` } +// PromptCompleteData contains data from prompt_complete events +type PromptCompleteData struct { + TaskID string `json:"task_id"` + AgentID string `json:"agent_id"` + AgentMessage string `json:"agent_message"` +} + // EventHandlers contains callbacks for different event types type EventHandlers struct { // Task events @@ -50,6 +57,9 @@ type EventHandlers struct { // ACP messages OnACPMessage func(ctx context.Context, taskID string, msg *protocol.Message) + + // Prompt events + OnPromptComplete func(ctx context.Context, data PromptCompleteData) } // Watcher subscribes to events and dispatches to handlers @@ -104,6 +114,12 @@ func (w *Watcher) Start(ctx context.Context) error { return err } + // Subscribe to prompt complete events + if err := w.subscribeToPromptCompleteEvents(); err != nil { + w.unsubscribeAll() + return err + } + w.running = true w.logger.Info("Event watcher started", zap.Int("subscriptions", len(w.subscriptions))) return nil @@ -302,6 +318,48 @@ func (w *Watcher) createACPMessageHandler() bus.EventHandler { } } +// subscribeToPromptCompleteEvents subscribes to prompt complete events +func (w *Watcher) subscribeToPromptCompleteEvents() error { + if w.handlers.OnPromptComplete == nil { + return nil + } + + // Use wildcard to subscribe to all prompt complete events (prompt.complete.*) + subject := events.PromptComplete + ".*" + + // Use regular subscription (each instance needs all messages for comment creation) + sub, err := w.eventBus.Subscribe(subject, w.createPromptCompleteHandler()) + if err != nil { + w.logger.Error("Failed to subscribe to prompt complete events", + zap.String("subject", subject), + zap.Error(err)) + return err + } + w.subscriptions = append(w.subscriptions, sub) + return nil +} + +// createPromptCompleteHandler creates a bus.EventHandler for prompt complete events +func (w *Watcher) createPromptCompleteHandler() bus.EventHandler { + return func(ctx context.Context, event *bus.Event) error { + var data PromptCompleteData + if err := w.parseEventData(event.Data, &data); err != nil { + w.logger.Error("Failed to parse prompt complete event data", + zap.String("event_type", event.Type), + zap.String("event_id", event.ID), + zap.Error(err)) + return nil // Don't return error to continue processing other events + } + + w.logger.Debug("Handling prompt complete event", + zap.String("task_id", data.TaskID), + zap.Int("message_length", len(data.AgentMessage))) + + w.handlers.OnPromptComplete(ctx, data) + return nil + } +} + // parseEventData converts event data map to a typed struct func (w *Watcher) parseEventData(data map[string]interface{}, target interface{}) error { // Marshal to JSON and unmarshal to target type diff --git a/apps/backend/internal/orchestrator/wshandlers/handlers.go b/apps/backend/internal/orchestrator/wshandlers/handlers.go index 1fab7101ac..4c7b22cfcc 100644 --- a/apps/backend/internal/orchestrator/wshandlers/handlers.go +++ b/apps/backend/internal/orchestrator/wshandlers/handlers.go @@ -41,6 +41,7 @@ func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { d.RegisterFunc(ws.ActionOrchestratorPrompt, h.PromptTask) d.RegisterFunc(ws.ActionOrchestratorComplete, h.CompleteTask) d.RegisterFunc(ws.ActionTaskLogs, h.GetTaskLogs) + d.RegisterFunc(ws.ActionPermissionRespond, h.RespondToPermission) } // GetStatus handles orchestrator.status action @@ -203,7 +204,6 @@ func (h *Handlers) PromptTask(ctx context.Context, msg *ws.Message) (*ws.Message return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ "success": true, - "needs_input": result.NeedsInput, "stop_reason": result.StopReason, }) } @@ -294,3 +294,46 @@ func (h *Handlers) GetTaskLogs(ctx context.Context, msg *ws.Message) (*ws.Messag Total: len(logs), }) } + +// PermissionRespondRequest is the payload for permission.respond +type PermissionRespondRequest struct { + TaskID string `json:"task_id"` + PendingID string `json:"pending_id"` + OptionID string `json:"option_id,omitempty"` + Cancelled bool `json:"cancelled,omitempty"` +} + +// RespondToPermission handles permission.respond action +func (h *Handlers) RespondToPermission(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req PermissionRespondRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + if req.PendingID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "pending_id is required", nil) + } + if !req.Cancelled && req.OptionID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "option_id is required when not cancelled", nil) + } + + h.logger.Info("responding to permission request", + zap.String("task_id", req.TaskID), + zap.String("pending_id", req.PendingID), + zap.String("option_id", req.OptionID), + zap.Bool("cancelled", req.Cancelled)) + + if err := h.service.RespondToPermission(ctx, req.TaskID, req.PendingID, req.OptionID, req.Cancelled); err != nil { + h.logger.Error("failed to respond to permission", zap.String("task_id", req.TaskID), zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to respond to permission: "+err.Error(), nil) + } + + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + "task_id": req.TaskID, + "pending_id": req.PendingID, + }) +} diff --git a/apps/backend/internal/task/handlers/comment_handlers.go b/apps/backend/internal/task/handlers/comment_handlers.go new file mode 100644 index 0000000000..a2b59e3e15 --- /dev/null +++ b/apps/backend/internal/task/handlers/comment_handlers.go @@ -0,0 +1,118 @@ +package handlers + +import ( + "context" + + "github.com/gin-gonic/gin" + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/task/service" + ws "github.com/kandev/kandev/pkg/websocket" +) + +// PromptResult represents the result of prompting an agent +type PromptResult struct { + StopReason string `json:"stop_reason"` + AgentMessage string `json:"agent_message"` +} + +// OrchestratorService defines the interface for orchestrator operations +type OrchestratorService interface { + PromptTask(ctx context.Context, taskID string, prompt string) (*PromptResult, error) +} + +// CommentHandlers handles WebSocket requests for comments +type CommentHandlers struct { + svc *service.Service + orchestrator OrchestratorService + logger *logger.Logger +} + +// NewCommentHandlers creates a new CommentHandlers instance +func NewCommentHandlers(svc *service.Service, orchestrator OrchestratorService, log *logger.Logger) *CommentHandlers { + return &CommentHandlers{ + svc: svc, + orchestrator: orchestrator, + logger: log.WithFields(zap.String("component", "task-comment-handlers")), + } +} + +// RegisterCommentRoutes registers comment WebSocket handlers +func RegisterCommentRoutes(router *gin.Engine, dispatcher *ws.Dispatcher, svc *service.Service, orchestrator OrchestratorService, log *logger.Logger) { + handlers := NewCommentHandlers(svc, orchestrator, log) + handlers.registerWS(dispatcher) +} + +func (h *CommentHandlers) registerWS(dispatcher *ws.Dispatcher) { + dispatcher.RegisterFunc(ws.ActionCommentAdd, h.wsAddComment) + dispatcher.RegisterFunc(ws.ActionCommentList, h.wsListComments) +} + +// WS handlers + +type wsAddCommentRequest struct { + TaskID string `json:"task_id"` + Content string `json:"content"` + AuthorID string `json:"author_id,omitempty"` +} + +func (h *CommentHandlers) wsAddComment(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsAddCommentRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + if req.Content == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "content is required", nil) + } + + comment, err := h.svc.CreateComment(ctx, &service.CreateCommentRequest{ + TaskID: req.TaskID, + Content: req.Content, + AuthorType: "user", + AuthorID: req.AuthorID, + }) + if err != nil { + h.logger.Error("failed to create comment", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to create comment", nil) + } + + // Auto-forward comment as prompt to running agent if orchestrator is available + // The agent's response will be saved as a comment via the prompt_complete event + if h.orchestrator != nil { + _, err := h.orchestrator.PromptTask(ctx, req.TaskID, req.Content) + if err != nil { + h.logger.Warn("failed to forward comment as prompt to agent", + zap.String("task_id", req.TaskID), + zap.Error(err)) + } + } + + return ws.NewResponse(msg.ID, msg.Action, comment) +} + +type wsListCommentsRequest struct { + TaskID string `json:"task_id"` +} + +func (h *CommentHandlers) wsListComments(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsListCommentsRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + + comments, err := h.svc.ListComments(ctx, req.TaskID) + if err != nil { + h.logger.Error("failed to list comments", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to list comments", nil) + } + + return ws.NewResponse(msg.ID, msg.Action, comments) +} + diff --git a/apps/backend/kandev b/apps/backend/kandev deleted file mode 100755 index 9893e5a287..0000000000 Binary files a/apps/backend/kandev and /dev/null differ diff --git a/apps/backend/pkg/api/v1/task.go b/apps/backend/pkg/api/v1/task.go index 6f5ac3d150..0299aa6bc8 100644 --- a/apps/backend/pkg/api/v1/task.go +++ b/apps/backend/pkg/api/v1/task.go @@ -94,3 +94,30 @@ type CreateCommentRequest struct { AuthorType string `json:"author_type,omitempty"` // Defaults to "user" if not specified RequestsInput bool `json:"requests_input,omitempty"` } + +// PermissionOption represents a permission choice presented to the user +type PermissionOption struct { + OptionID string `json:"option_id"` + Name string `json:"name"` + Kind string `json:"kind"` // allow_once, allow_always, reject_once, reject_always +} + +// PermissionRequest represents an agent's request for user permission +type PermissionRequest struct { + RequestID string `json:"request_id"` // Unique ID for this request (JSON-RPC ID) + TaskID string `json:"task_id"` // Task the agent is working on + InstanceID string `json:"instance_id"` // Agent instance ID + SessionID string `json:"session_id"` // ACP session ID + ToolCallID string `json:"tool_call_id"` // Tool call requesting permission + Title string `json:"title"` // Human-readable title + Description string `json:"description,omitempty"` // Additional context + Options []PermissionOption `json:"options"` // Available choices + CreatedAt time.Time `json:"created_at"` +} + +// PermissionResponse represents the user's response to a permission request +type PermissionResponse struct { + RequestID string `json:"request_id" binding:"required"` // The request being responded to + OptionID string `json:"option_id,omitempty"` // Selected option (if not cancelled) + Cancelled bool `json:"cancelled,omitempty"` // True if user cancelled +} diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 3b43161aeb..8efec73d4a 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -93,6 +93,10 @@ const ( ActionRepositoryScriptCreated = "repository.script.created" ActionRepositoryScriptUpdated = "repository.script.updated" ActionRepositoryScriptDeleted = "repository.script.deleted" + + // Permission request actions (agent -> user -> agent) + ActionPermissionRequested = "permission.requested" // Agent requesting permission + ActionPermissionRespond = "permission.respond" // User responding to permission request ) // Error codes diff --git a/docs/WEBSOCKET_API.md b/docs/WEBSOCKET_API.md index 4196d7d2d6..23dc1f5d90 100644 --- a/docs/WEBSOCKET_API.md +++ b/docs/WEBSOCKET_API.md @@ -127,9 +127,148 @@ All messages follow this JSON envelope format: --- +## Workspace Actions + +Workspaces are the top-level organizational unit that contain boards and repositories. + +### `workspace.list` + +**Purpose:** List all workspaces. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "workspace.list", + "payload": {} +} +``` + +**Response:** +```json +{ + "id": "uuid", + "type": "response", + "action": "workspace.list", + "payload": { + "workspaces": [ + { + "id": "workspace-uuid", + "name": "My Project", + "description": "Main development workspace", + "owner_id": "user-uuid", + "created_at": "2026-01-10T12:00:00Z", + "updated_at": "2026-01-10T12:00:00Z" + } + ] + } +} +``` + +### `workspace.create` + +**Purpose:** Create a new workspace. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "workspace.create", + "payload": { + "name": "My Project", + "description": "Main development workspace", + "owner_id": "user-uuid" + } +} +``` + +| Payload Field | Type | Required | Description | +|--------------|------|----------|-------------| +| `name` | string | ✅ | Workspace name (max 255 chars) | +| `description` | string | ❌ | Workspace description | +| `owner_id` | string | ❌ | Owner user ID | + +**Response:** Created workspace object + +### `workspace.get` + +**Purpose:** Get a workspace by ID. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "workspace.get", + "payload": { + "id": "workspace-uuid" + } +} +``` + +**Response:** Workspace object + +### `workspace.update` + +**Purpose:** Update a workspace. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "workspace.update", + "payload": { + "id": "workspace-uuid", + "name": "Updated Name", + "description": "Updated description" + } +} +``` + +| Payload Field | Type | Required | Description | +|--------------|------|----------|-------------| +| `id` | string | ✅ | Workspace ID | +| `name` | string | ❌ | New name | +| `description` | string | ❌ | New description | + +**Response:** Updated workspace object + +### `workspace.delete` + +**Purpose:** Delete a workspace. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "workspace.delete", + "payload": { + "id": "workspace-uuid" + } +} +``` + +**Response:** +```json +{ + "id": "uuid", + "type": "response", + "action": "workspace.delete", + "payload": { + "deleted": true + } +} +``` + +--- + ## Board Actions -Boards are the top-level containers that organize columns and tasks. +Boards are containers within workspaces that organize columns and tasks. ### `board.create` @@ -427,6 +566,286 @@ Columns organize tasks within a board and represent workflow states. --- +## Repository Actions + +Repositories represent Git repositories linked to a workspace. + +### `repository.list` + +**Purpose:** List all repositories in a workspace. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "repository.list", + "payload": { + "workspace_id": "workspace-uuid" + } +} +``` + +**Response:** +```json +{ + "id": "uuid", + "type": "response", + "action": "repository.list", + "payload": { + "repositories": [ + { + "id": "repo-uuid", + "workspace_id": "workspace-uuid", + "name": "my-project", + "source_type": "local", + "local_path": "/home/user/projects/my-project", + "provider": "", + "default_branch": "main", + "created_at": "2026-01-10T12:00:00Z", + "updated_at": "2026-01-10T12:00:00Z" + } + ] + } +} +``` + +### `repository.create` + +**Purpose:** Create a new repository link. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "repository.create", + "payload": { + "workspace_id": "workspace-uuid", + "name": "my-project", + "source_type": "local", + "local_path": "/home/user/projects/my-project", + "default_branch": "main", + "setup_script": "npm install", + "cleanup_script": "npm run clean" + } +} +``` + +| Payload Field | Type | Required | Description | +|--------------|------|----------|-------------| +| `workspace_id` | string | ✅ | Workspace ID | +| `name` | string | ✅ | Repository name | +| `source_type` | string | ❌ | `local` or `remote` | +| `local_path` | string | ❌ | Local filesystem path | +| `provider` | string | ❌ | Git provider (github, gitlab, etc.) | +| `provider_repo_id` | string | ❌ | Provider's repository ID | +| `provider_owner` | string | ❌ | Repository owner on provider | +| `provider_name` | string | ❌ | Repository name on provider | +| `default_branch` | string | ❌ | Default branch name | +| `setup_script` | string | ❌ | Script to run on setup | +| `cleanup_script` | string | ❌ | Script to run on cleanup | + +**Response:** Created repository object + +### `repository.get` + +**Purpose:** Get a repository by ID. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "repository.get", + "payload": { + "id": "repo-uuid" + } +} +``` + +**Response:** Repository object + +### `repository.update` + +**Purpose:** Update a repository. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "repository.update", + "payload": { + "id": "repo-uuid", + "name": "updated-name", + "default_branch": "develop" + } +} +``` + +**Response:** Updated repository object + +### `repository.delete` + +**Purpose:** Delete a repository link. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "repository.delete", + "payload": { + "id": "repo-uuid" + } +} +``` + +**Response:** +```json +{ + "id": "uuid", + "type": "response", + "action": "repository.delete", + "payload": { + "deleted": true + } +} +``` + +### `repository.script.list` + +**Purpose:** List scripts for a repository. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "repository.script.list", + "payload": { + "repository_id": "repo-uuid" + } +} +``` + +**Response:** +```json +{ + "id": "uuid", + "type": "response", + "action": "repository.script.list", + "payload": { + "scripts": [ + { + "id": "script-uuid", + "repository_id": "repo-uuid", + "name": "build", + "command": "npm run build", + "position": 0 + } + ] + } +} +``` + +### `repository.script.create` + +**Purpose:** Create a repository script. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "repository.script.create", + "payload": { + "repository_id": "repo-uuid", + "name": "test", + "command": "npm test", + "position": 1 + } +} +``` + +| Payload Field | Type | Required | Description | +|--------------|------|----------|-------------| +| `repository_id` | string | ✅ | Repository ID | +| `name` | string | ✅ | Script name | +| `command` | string | ✅ | Command to execute | +| `position` | int | ❌ | Order position | + +**Response:** Created script object + +### `repository.script.get` + +**Purpose:** Get a script by ID. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "repository.script.get", + "payload": { + "id": "script-uuid" + } +} +``` + +**Response:** Script object + +### `repository.script.update` + +**Purpose:** Update a script. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "repository.script.update", + "payload": { + "id": "script-uuid", + "name": "updated-name", + "command": "npm run updated-command" + } +} +``` + +**Response:** Updated script object + +### `repository.script.delete` + +**Purpose:** Delete a script. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "repository.script.delete", + "payload": { + "id": "script-uuid" + } +} +``` + +**Response:** +```json +{ + "id": "uuid", + "type": "response", + "action": "repository.script.delete", + "payload": { + "deleted": true + } +} +``` + +--- + ## Task Actions Tasks are work items that can be executed by AI agents. @@ -757,6 +1176,106 @@ Subscriptions enable real-time streaming of agent output (ACP) to specific clien --- +## Comment Actions + +Comments enable conversation between users and agents on tasks. + +### `comment.add` + +**Purpose:** Add a comment to a task. If an agent is running on the task, the comment is automatically forwarded as a prompt. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "comment.add", + "payload": { + "task_id": "task-uuid", + "content": "Please also add unit tests for the new function", + "author_id": "user-uuid" + } +} +``` + +| Payload Field | Type | Required | Description | +|--------------|------|----------|-------------| +| `task_id` | string | ✅ | Task ID | +| `content` | string | ✅ | Comment text | +| `author_id` | string | ❌ | Author user ID | + +**Response:** +```json +{ + "id": "uuid", + "type": "response", + "action": "comment.add", + "payload": { + "id": "comment-uuid", + "task_id": "task-uuid", + "author_type": "user", + "author_id": "user-uuid", + "content": "Please also add unit tests for the new function", + "requests_input": false, + "created_at": "2026-01-10T12:00:00Z" + } +} +``` + +### `comment.list` + +**Purpose:** List all comments for a task. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "comment.list", + "payload": { + "task_id": "task-uuid" + } +} +``` + +**Response:** +```json +{ + "id": "uuid", + "type": "response", + "action": "comment.list", + "payload": { + "comments": [ + { + "id": "comment-uuid-1", + "task_id": "task-uuid", + "author_type": "user", + "author_id": "user-uuid", + "content": "Please implement the login feature", + "requests_input": false, + "created_at": "2026-01-10T12:00:00Z" + }, + { + "id": "comment-uuid-2", + "task_id": "task-uuid", + "author_type": "agent", + "author_id": "agent-instance-uuid", + "content": "I've implemented the login feature. Should I also add password reset?", + "requests_input": true, + "created_at": "2026-01-10T12:05:00Z" + } + ] + } +} +``` + +| Response Field | Type | Description | +|---------------|------|-------------| +| `author_type` | string | `user` or `agent` | +| `requests_input` | boolean | True if agent is asking for user input | + +--- + ## Agent Actions Agents are Docker containers running AI coding assistants. @@ -1262,6 +1781,120 @@ The orchestrator coordinates task execution, managing the workflow between tasks --- +## Permission Actions + +Agents may request user permission for certain operations. The permission flow uses notifications and responses. + +### `permission.requested` (Notification) + +**Purpose:** Server notifies client that an agent is requesting permission. + +**When:** Agent calls a tool that requires user approval. + +```json +{ + "type": "notification", + "action": "permission.requested", + "payload": { + "task_id": "task-uuid", + "pending_id": "pending-request-uuid", + "instance_id": "agent-instance-uuid", + "session_id": "acp-session-uuid", + "tool_call_id": "tool-call-uuid", + "title": "Execute shell command", + "description": "Agent wants to run: npm install express", + "options": [ + { + "option_id": "allow-once", + "name": "Allow Once", + "kind": "allow_once" + }, + { + "option_id": "allow-always", + "name": "Always Allow", + "kind": "allow_always" + }, + { + "option_id": "reject-once", + "name": "Deny", + "kind": "reject_once" + } + ], + "created_at": "2026-01-10T12:00:00Z" + }, + "timestamp": "2026-01-10T12:00:00Z" +} +``` + +| Payload Field | Type | Description | +|--------------|------|-------------| +| `task_id` | string | Task the agent is working on | +| `pending_id` | string | Unique ID for this pending request | +| `instance_id` | string | Agent instance ID | +| `session_id` | string | ACP session ID | +| `tool_call_id` | string | Tool call requesting permission | +| `title` | string | Human-readable title | +| `description` | string | Additional context | +| `options` | array | Available permission choices | + +**Option Kinds:** +- `allow_once` - Allow this specific action +- `allow_always` - Allow this type of action permanently +- `reject_once` - Deny this specific action +- `reject_always` - Deny this type of action permanently + +### `permission.respond` + +**Purpose:** User responds to a permission request. + +**Request:** +```json +{ + "id": "uuid", + "type": "request", + "action": "permission.respond", + "payload": { + "task_id": "task-uuid", + "pending_id": "pending-request-uuid", + "option_id": "allow-once" + } +} +``` + +| Payload Field | Type | Required | Description | +|--------------|------|----------|-------------| +| `task_id` | string | ✅ | Task ID | +| `pending_id` | string | ✅ | Pending request ID from notification | +| `option_id` | string | ❌ | Selected option ID (required if not cancelled) | +| `cancelled` | boolean | ❌ | True to cancel the request | + +**Response:** +```json +{ + "id": "uuid", + "type": "response", + "action": "permission.respond", + "payload": { + "success": true + } +} +``` + +**Error Response (no pending request):** +```json +{ + "id": "uuid", + "type": "error", + "action": "permission.respond", + "payload": { + "code": "NOT_FOUND", + "message": "No pending permission request for this task" + } +} +``` + +--- + ## Server Notifications (ACP Streaming) These are push notifications from the server to subscribed clients. They have `type: "notification"` and no `id` field. @@ -1403,6 +2036,34 @@ These are push notifications from the server to subscribed clients. They have `t } ``` +### `comment.added` + +**Purpose:** New comment added to a task. + +**When:** User or agent adds a comment. + +```json +{ + "type": "notification", + "action": "comment.added", + "payload": { + "task_id": "task-uuid", + "comment_id": "comment-uuid", + "author_type": "agent", + "author_id": "agent-instance-uuid", + "content": "I've completed the implementation. Would you like me to add tests?", + "requests_input": true, + "created_at": "2026-01-10T12:05:00Z" + }, + "timestamp": "2026-01-10T12:05:00Z" +} +``` + +| Payload Field | Type | Description | +|--------------|------|-------------| +| `author_type` | string | `user` or `agent` | +| `requests_input` | boolean | True if agent is asking for user input | + ### `task.updated` **Purpose:** Task properties changed. @@ -1522,17 +2183,41 @@ async function runTask() { ## Quick Reference +### Request Actions + | Action | Purpose | |--------|---------| | `health.check` | Server health status | +| **Workspace** | | +| `workspace.list` | List all workspaces | +| `workspace.create` | Create workspace | +| `workspace.get` | Get workspace by ID | +| `workspace.update` | Update workspace | +| `workspace.delete` | Delete workspace | +| **Board** | | | `board.create` | Create board | | `board.list` | List all boards | | `board.get` | Get board by ID | | `board.update` | Update board | | `board.delete` | Delete board | +| **Column** | | | `column.create` | Create column | | `column.list` | List columns in board | | `column.get` | Get column by ID | +| `column.update` | Update column | +| `column.delete` | Delete column | +| **Repository** | | +| `repository.list` | List repositories in workspace | +| `repository.create` | Create repository link | +| `repository.get` | Get repository by ID | +| `repository.update` | Update repository | +| `repository.delete` | Delete repository | +| `repository.script.list` | List repository scripts | +| `repository.script.create` | Create script | +| `repository.script.get` | Get script by ID | +| `repository.script.update` | Update script | +| `repository.script.delete` | Delete script | +| **Task** | | | `task.create` | Create task | | `task.list` | List tasks in board | | `task.get` | Get task by ID | @@ -1542,12 +2227,17 @@ async function runTask() { | `task.state` | Change task state | | `task.subscribe` | Subscribe to task notifications | | `task.unsubscribe` | Unsubscribe from task | +| **Comment** | | +| `comment.add` | Add comment to task | +| `comment.list` | List comments for task | +| **Agent** | | | `agent.list` | List agent instances | | `agent.launch` | Launch agent (low-level) | | `agent.status` | Get agent status | | `agent.logs` | Get agent logs | | `agent.stop` | Stop agent | | `agent.types` | List available agent types | +| **Orchestrator** | | | `orchestrator.status` | Get orchestrator status | | `orchestrator.queue` | Get task queue | | `orchestrator.trigger` | Queue task for execution | @@ -1555,4 +2245,31 @@ async function runTask() { | `orchestrator.stop` | Stop task execution | | `orchestrator.prompt` | Send follow-up prompt | | `orchestrator.complete` | Complete task | +| **Permission** | | +| `permission.respond` | Respond to permission request | + +### Notification Actions (Server → Client) + +| Action | Purpose | +|--------|---------| +| `acp.progress` | Agent progress update | +| `acp.log` | Agent log message | +| `acp.result` | Agent result artifact | +| `acp.error` | Agent error | +| `acp.status` | Agent status change | +| `acp.heartbeat` | Keep-alive heartbeat | +| `permission.requested` | Agent requesting permission | +| `comment.added` | New comment on task | +| `task.updated` | Task was updated | +| `task.created` | Task was created | +| `task.deleted` | Task was deleted | +| `workspace.created` | Workspace was created | +| `workspace.updated` | Workspace was updated | +| `workspace.deleted` | Workspace was deleted | +| `board.created` | Board was created | +| `board.updated` | Board was updated | +| `board.deleted` | Board was deleted | +| `column.created` | Column was created | +| `column.updated` | Column was updated | +| `column.deleted` | Column was deleted | diff --git a/scripts/e2e-test.sh b/scripts/e2e-test.sh index 2241678268..d5807ea976 100755 --- a/scripts/e2e-test.sh +++ b/scripts/e2e-test.sh @@ -1,7 +1,7 @@ #!/bin/bash # -# End-to-End Test for Kandev -# This script tests the complete flow from task creation to agent execution +# End-to-End Test for Kandev - Comment System Focus +# This script tests the comment system and agent interaction flow # using WebSocket-based communication with the backend. # # Prerequisites: @@ -24,13 +24,14 @@ RED='\033[0;31m' GREEN='\033[0;32m' YELLOW='\033[1;33m' BLUE='\033[0;34m' +CYAN='\033[0;36m' NC='\033[0m' # No Color # Configuration SERVER_PORT=8080 WS_URL="ws://localhost:${SERVER_PORT}/ws" AGENTCTL_PORT=9999 -WAIT_FOR_AGENT_SECONDS=120 +WAIT_FOR_AGENT_SECONDS=45 SERVER_PID="" BUILD_IMAGE=false @@ -80,6 +81,21 @@ log_info() { echo -e "${YELLOW}→ $1${NC}" } +log_comment() { + local author_type=$1 + local content=$2 + local requests_input=$3 + if [ "$author_type" == "user" ]; then + echo -e "${CYAN} [USER]${NC} $content" + else + if [ "$requests_input" == "true" ]; then + echo -e "${GREEN} [AGENT] (requests input)${NC} $content" + else + echo -e "${GREEN} [AGENT]${NC} $content" + fi + fi +} + # WebSocket helper function # Sends a request via WebSocket and returns the response ws_request() { @@ -97,30 +113,40 @@ ws_request() { wait_for_server() { log_info "Waiting for server to be ready..." - for i in {1..30}; do + for i in {1..20}; do HEALTH=$(ws_request "health.check" '{}' 2>/dev/null) if echo "$HEALTH" | jq -e '.payload.status == "ok"' > /dev/null 2>&1; then log_success "Server is ready" return 0 fi - sleep 1 + sleep 0.5 done log_error "Server failed to start" exit 1 } -wait_for_agentctl() { - local container_ip=$1 - log_info "Waiting for agentctl to be ready at ${container_ip}:${AGENTCTL_PORT}..." - for i in {1..30}; do - if curl -s "http://${container_ip}:${AGENTCTL_PORT}/health" > /dev/null 2>&1; then - log_success "agentctl is ready" - return 0 - fi - sleep 1 - done - log_error "agentctl failed to start" - return 1 +# Display all comments for a task +display_comments() { + local task_id=$1 + local label=$2 + + COMMENTS_RESPONSE=$(ws_request "comment.list" "{\"task_id\": \"${task_id}\"}") + COMMENTS_COUNT=$(echo "$COMMENTS_RESPONSE" | jq 'if .payload | type == "array" then .payload | length else 0 end' 2>/dev/null || echo "0") + + echo -e "\n${CYAN}--- $label ($COMMENTS_COUNT comments) ---${NC}" + + if [ "$COMMENTS_COUNT" -gt 0 ]; then + echo "$COMMENTS_RESPONSE" | jq -r '.payload[] | "\(.author_type)|\(.requests_input)|\(.content)"' 2>/dev/null | while IFS='|' read -r author_type requests_input content; do + # Truncate long content for display + if [ ${#content} -gt 200 ]; then + content="${content:0:200}..." + fi + log_comment "$author_type" "$content" "$requests_input" + done + else + echo -e " ${YELLOW}(no comments yet)${NC}" + fi + echo -e "${CYAN}---${NC}" } # Check prerequisites @@ -193,9 +219,19 @@ log_info "Server PID: $SERVER_PID" wait_for_server # Run the E2E test -log_step "Step 1: Create Board" -BOARD_RESPONSE=$(ws_request "board.create" '{"name": "E2E Test Board", "description": "Automated end-to-end test"}') -BOARD_ID=$(echo "$BOARD_RESPONSE" | jq -r '.payload.id') +log_step "Step 1: Create Workspace" +WORKSPACE_RESPONSE=$(ws_request "workspace.create" '{"name": "E2E Test Workspace", "description": "Automated end-to-end test workspace"}') +WORKSPACE_ID=$(echo "$WORKSPACE_RESPONSE" | jq -r '.payload.id' | tr -d '\n\r') +if [ "$WORKSPACE_ID" == "null" ] || [ -z "$WORKSPACE_ID" ]; then + log_error "Failed to create workspace" + echo "$WORKSPACE_RESPONSE" + exit 1 +fi +log_success "Created workspace: $WORKSPACE_ID" + +log_step "Step 2: Create Board" +BOARD_RESPONSE=$(ws_request "board.create" "{\"workspace_id\": \"${WORKSPACE_ID}\", \"name\": \"E2E Test Board\", \"description\": \"Automated end-to-end test\"}") +BOARD_ID=$(echo "$BOARD_RESPONSE" | jq -r '.payload.id' | tr -d '\n\r') if [ "$BOARD_ID" == "null" ] || [ -z "$BOARD_ID" ]; then log_error "Failed to create board" echo "$BOARD_RESPONSE" @@ -203,9 +239,9 @@ if [ "$BOARD_ID" == "null" ] || [ -z "$BOARD_ID" ]; then fi log_success "Created board: $BOARD_ID" -log_step "Step 2: Create Column" +log_step "Step 3: Create Column" COLUMN_RESPONSE=$(ws_request "column.create" "{\"board_id\": \"${BOARD_ID}\", \"name\": \"To Do\", \"position\": 0}") -COLUMN_ID=$(echo "$COLUMN_RESPONSE" | jq -r '.payload.id') +COLUMN_ID=$(echo "$COLUMN_RESPONSE" | jq -r '.payload.id' | tr -d '\n\r') if [ "$COLUMN_ID" == "null" ] || [ -z "$COLUMN_ID" ]; then log_error "Failed to create column" echo "$COLUMN_RESPONSE" @@ -213,25 +249,26 @@ if [ "$COLUMN_ID" == "null" ] || [ -z "$COLUMN_ID" ]; then fi log_success "Created column: $COLUMN_ID" -log_step "Step 3: Create Task" +log_step "Step 4: Create Task for Comment System Test" # Create a temp directory for the test workspace -TEST_WORKSPACE=$(mktemp -d) -log_info "Test workspace: $TEST_WORKSPACE" +TEST_WORKSPACE_DIR=$(mktemp -d) +log_info "Test workspace: $TEST_WORKSPACE_DIR" -# Create a simple task that creates a file +# Create task that will trigger multi-turn conversation TASK_PAYLOAD=$(cat </dev/null || echo "0") +log_info "Comments before agent start: $INITIAL_COUNT" + +log_step "Step 6: Start Task via Orchestrator" START_RESPONSE=$(ws_request "orchestrator.start" "{\"task_id\": \"${TASK_ID}\"}") -AGENT_ID=$(echo "$START_RESPONSE" | jq -r '.payload.agent_instance_id') +AGENT_ID=$(echo "$START_RESPONSE" | jq -r '.payload.agent_instance_id' | tr -d '\n\r') if [ "$AGENT_ID" == "null" ] || [ -z "$AGENT_ID" ]; then log_error "Failed to start task" echo "$START_RESPONSE" @@ -249,661 +309,260 @@ if [ "$AGENT_ID" == "null" ] || [ -z "$AGENT_ID" ]; then fi log_success "Task started, agent ID: $AGENT_ID" -log_step "Step 5: Verify Agent Container Running" -sleep 3 -CONTAINER_ID=$(docker ps --filter "name=kandev-agent" --format "{{.ID}}" | head -1) -if [ -z "$CONTAINER_ID" ]; then - log_error "No agent container running" - exit 1 -fi -log_success "Agent container is running: $CONTAINER_ID" - -# Get container IP for agentctl health check -CONTAINER_IP=$(docker inspect -f '{{range .NetworkSettings.Networks}}{{.IPAddress}}{{end}}' "$CONTAINER_ID" 2>/dev/null) -if [ -n "$CONTAINER_IP" ]; then - log_info "Container IP: $CONTAINER_IP" -fi - -log_step "Step 6: Verify agentctl is Running" +# Brief wait for container startup sleep 2 -if [ -n "$CONTAINER_IP" ]; then - if wait_for_agentctl "$CONTAINER_IP"; then - # Check agentctl status - 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 process state: $AGENT_PROCESS_STATE" - else - log_info "Could not reach agentctl directly (may be in bridge network)" - fi -else - log_info "Container IP not available (using host network)" +CONTAINER_ID=$(docker ps --filter "name=kandev-agent" --format "{{.ID}}" | head -1) +if [ -n "$CONTAINER_ID" ]; then + log_success "Agent container running: $CONTAINER_ID" fi -# Check server logs for agentctl communication -if grep -q "agentctl" /tmp/kandev-e2e.log; then - log_success "Backend is communicating with agentctl" -else - log_info "Checking backend logs..." -fi - -if grep -q "agent started" /tmp/kandev-e2e.log || grep -q "initializing agent" /tmp/kandev-e2e.log; then - log_success "Agent initialization started" -else - log_info "Waiting for agent initialization..." -fi - -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" +log_step "Step 7: Wait for Initial Agent Response" +log_info "Waiting up to ${WAIT_FOR_AGENT_SECONDS}s for agent..." AGENT_READY=false for i in $(seq 1 $WAIT_FOR_AGENT_SECONDS); do - # Check agent status - READY means prompt completed AGENT_STATUS_RESPONSE=$(ws_request "agent.status" "{\"agent_id\": \"${AGENT_ID}\"}" 2>/dev/null || echo '{}') AGENT_STATUS=$(echo "$AGENT_STATUS_RESPONSE" | jq -r '.payload.status // "UNKNOWN"') if [ "$AGENT_STATUS" == "READY" ]; then - log_success "Agent is READY after $i seconds (prompt completed)" + log_success "Agent READY after $i seconds" AGENT_READY=true break elif [ "$AGENT_STATUS" == "FAILED" ]; then log_error "Agent failed" - echo "$AGENT_STATUS_RESPONSE" | jq . - if [ -n "$CONTAINER_ID" ]; then - log_info "Container logs:" - docker logs "$CONTAINER_ID" 2>&1 | tail -30 - fi exit 1 fi - # 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 unexpectedly" - break - fi - if [ $((i % 10)) -eq 0 ]; then - log_info "Still waiting... ($i seconds, status: $AGENT_STATUS)" + log_info "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)" + log_error "Agent did not become READY (last: $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) -if [ -n "$FINAL_CONTAINER_ID" ]; then - log_info "Checking container logs for $FINAL_CONTAINER_ID..." - CONTAINER_LOGS=$(docker logs "$FINAL_CONTAINER_ID" 2>&1 | tail -100) +# Display comments after first agent response +display_comments "$TASK_ID" "After Initial Agent Response" - # Check for agentctl activity - if echo "$CONTAINER_LOGS" | grep -q "agentctl"; then - log_success "agentctl was active in container" - fi - - # Check for ACP messages - if echo "$CONTAINER_LOGS" | grep -q '"stopReason"'; then - log_success "Agent completed with stop reason" - elif echo "$CONTAINER_LOGS" | grep -q 'session/update'; then - log_success "Agent sent session updates" - elif echo "$CONTAINER_LOGS" | grep -q 'initialize'; then - log_success "Agent received initialize request" - else - log_info "Container logs (last 20 lines):" - echo "$CONTAINER_LOGS" | tail -20 - fi -else - log_info "Container already removed" -fi - -log_step "Step 9: Check Final State" -AGENTS_RESPONSE=$(ws_request "agent.list" '{}') -FINAL_AGENTS=$(echo "$AGENTS_RESPONSE" | jq '.payload.total') -log_info "Total agents: $FINAL_AGENTS" - -TASK_RESPONSE=$(ws_request "task.get" "{\"id\": \"${TASK_ID}\"}") -TASK_STATE=$(echo "$TASK_RESPONSE" | jq -r '.payload.state') -log_info "Task state: $TASK_STATE" - -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 "First prompt output contains expected content" - fi -else - 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_PAYLOAD=$(cat </dev/null || echo '{}') AGENT_STATUS=$(echo "$AGENT_STATUS_RESPONSE" | jq -r '.payload.status // "UNKNOWN"') if [ "$AGENT_STATUS" == "READY" ]; then - log_success "Agent is READY after follow-up prompt ($i seconds)" + log_success "Agent READY after Turn 1 ($i seconds)" break fi if [ $((i % 10)) -eq 0 ]; then - log_info "Still waiting... ($i seconds, status: $AGENT_STATUS)" + log_info "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 +display_comments "$TASK_ID" "After Turn 1" + +log_step "Step 9: Multi-Turn Conversation - Turn 2" +# User Comment Turn 2: Ask for verification +TURN2_PAYLOAD=$(cat </dev/null || echo '{}') + AGENT_STATUS=$(echo "$AGENT_STATUS_RESPONSE" | jq -r '.payload.status // "UNKNOWN"') + if [ "$AGENT_STATUS" == "READY" ]; then + log_success "Agent READY after Turn 2 ($i seconds)" + break + fi + if [ $((i % 10)) -eq 0 ]; then + log_info "Waiting... ($i seconds, status: $AGENT_STATUS)" + fi + sleep 1 +done + +display_comments "$TASK_ID" "After Turn 2" + +log_step "Step 10: Multi-Turn Conversation - Turn 3" +# User Comment Turn 3: Final request +TURN3_PAYLOAD=$(cat </dev/null || echo '{}') + AGENT_STATUS=$(echo "$AGENT_STATUS_RESPONSE" | jq -r '.payload.status // "UNKNOWN"') + if [ "$AGENT_STATUS" == "READY" ]; then + log_success "Agent READY after Turn 3 ($i seconds)" + break + fi + if [ $((i % 10)) -eq 0 ]; then + log_info "Waiting... ($i seconds, status: $AGENT_STATUS)" + fi + sleep 1 +done + +display_comments "$TASK_ID" "After Turn 3 (Final Conversation)" + +log_step "Step 11: Verify Comment Metadata" +# Get all comments and verify metadata +FINAL_COMMENTS=$(ws_request "comment.list" "{\"task_id\": \"${TASK_ID}\"}") +COMMENT_COUNT=$(echo "$FINAL_COMMENTS" | jq 'if .payload | type == "array" then .payload | length else 0 end' 2>/dev/null || echo "0") +log_info "Total comments in conversation: $COMMENT_COUNT" + +# Count by author type +USER_COMMENTS=$(echo "$FINAL_COMMENTS" | jq '[.payload[] | select(.author_type == "user")] | length' 2>/dev/null || echo "0") +AGENT_COMMENTS=$(echo "$FINAL_COMMENTS" | jq '[.payload[] | select(.author_type == "agent")] | length' 2>/dev/null || echo "0") +log_info "User comments: $USER_COMMENTS, Agent comments: $AGENT_COMMENTS" + +# Check for requests_input flag +REQUESTS_INPUT_COUNT=$(echo "$FINAL_COMMENTS" | jq '[.payload[] | select(.requests_input == true)] | length' 2>/dev/null || echo "0") +if [ "$REQUESTS_INPUT_COUNT" -gt 0 ]; then + log_success "Found $REQUESTS_INPUT_COUNT comment(s) with requests_input=true" +else + log_info "No comments with requests_input=true (agent didn't explicitly request input)" +fi + +# Verify timestamps are present +HAS_TIMESTAMPS=$(echo "$FINAL_COMMENTS" | jq 'if .payload | type == "array" and (.payload | length) > 0 then (.payload[0].created_at != null) else false end' 2>/dev/null || echo "false") +if [ "$HAS_TIMESTAMPS" == "true" ]; then + log_success "Comments have created_at timestamps" +fi + +log_step "Step 12: Complete Task and Verify Final State" COMPLETE_RESPONSE=$(ws_request "orchestrator.complete" "{\"task_id\": \"${TASK_ID}\"}") COMPLETE_SUCCESS=$(echo "$COMPLETE_RESPONSE" | jq -r '.payload.success // false') if [ "$COMPLETE_SUCCESS" == "true" ]; then log_success "Task completed successfully" else - log_error "Failed to complete task: $(echo "$COMPLETE_RESPONSE" | jq -r '.payload.error // "unknown"')" + log_info "Complete response: $(echo "$COMPLETE_RESPONSE" | jq -c '.payload')" fi -# Verify task state is COMPLETED -FINAL_TASK_RESPONSE=$(ws_request "task.get" "{\"id\": \"${TASK_ID}\"}") -TASK_STATE=$(echo "$FINAL_TASK_RESPONSE" | jq -r '.payload.state') -if [ "$TASK_STATE" == "COMPLETED" ]; then - log_success "Task state is COMPLETED" -else - log_error "Expected task state COMPLETED, got: $TASK_STATE" -fi +# Verify task state +FINAL_TASK=$(ws_request "task.get" "{\"id\": \"${TASK_ID}\"}") +TASK_STATE=$(echo "$FINAL_TASK" | jq -r '.payload.state') +log_info "Final task state: $TASK_STATE" -log_step "Step 13: Verify Session ID in Task Metadata" -# The session_id should have been stored in task metadata by the session_info handler -SESSION_ID=$(echo "$FINAL_TASK_RESPONSE" | jq -r '.payload.metadata.auggie_session_id // empty') +# Check for session ID in metadata +SESSION_ID=$(echo "$FINAL_TASK" | jq -r '.payload.metadata.auggie_session_id // empty') if [ -n "$SESSION_ID" ]; then - log_success "Session ID stored in task metadata: $SESSION_ID" -else - log_error "Session ID not found in task metadata" - log_info "Task metadata: $(echo "$FINAL_TASK_RESPONSE" | jq '.payload.metadata')" - exit 1 + log_success "Session ID stored: ${SESSION_ID:0:20}..." fi -# Track historical logs test results -HIST_LOGS_TEST_PASSED=false -HIST_LOGS_COUNT=0 -HIST_SUBSCRIBE_TEST_PASSED=false -HIST_SUBSCRIBE_COUNT=0 -LIVE_STREAM_TEST_PASSED=false -LIVE_STREAM_COUNT=0 -COMMENT_TEST_PASSED=false -INPUT_REQUEST_TEST_PASSED=false - -log_step "Step 14: Test Comment System" -# Test adding a user comment to the task -COMMENT_PAYLOAD=$(cat </dev/null || echo "0") -if [ "$COMMENTS_COUNT" -gt 0 ]; then - log_success "Listed $COMMENTS_COUNT comments for task" -else - log_info "No comments found for task" -fi - -# Test getting a specific comment -if [ -n "$COMMENT_ID" ]; then - GET_COMMENT_RESPONSE=$(ws_request "comment.get" "{\"comment_id\": \"${COMMENT_ID}\"}") - GOT_CONTENT=$(echo "$GET_COMMENT_RESPONSE" | jq -r '.payload.content // empty') - if [ -n "$GOT_CONTENT" ]; then - log_success "Retrieved comment content: $GOT_CONTENT" - else - log_info "Failed to retrieve comment" - fi -fi - -log_step "Step 15: Test Historical Logs for Completed Task" -# Test task.logs action to retrieve all historical logs -LOGS_RESPONSE=$(ws_request "task.logs" "{\"task_id\": \"${TASK_ID}\"}") -LOGS_SUCCESS=$(echo "$LOGS_RESPONSE" | jq -e '.payload.logs' > /dev/null 2>&1 && echo "true" || echo "false") - -if [ "$LOGS_SUCCESS" == "true" ]; then - HIST_LOGS_COUNT=$(echo "$LOGS_RESPONSE" | jq '.payload.logs | length') - TOTAL_FROM_RESPONSE=$(echo "$LOGS_RESPONSE" | jq '.payload.total // 0') - log_success "Retrieved $HIST_LOGS_COUNT historical logs (total: $TOTAL_FROM_RESPONSE)" - - # Verify logs contain expected fields - if [ "$HIST_LOGS_COUNT" -gt 0 ]; then - FIRST_LOG=$(echo "$LOGS_RESPONSE" | jq '.payload.logs[0]') - HAS_TYPE=$(echo "$FIRST_LOG" | jq -e '.type' > /dev/null 2>&1 && echo "true" || echo "false") - HAS_TIMESTAMP=$(echo "$FIRST_LOG" | jq -e '.timestamp' > /dev/null 2>&1 && echo "true" || echo "false") - - if [ "$HAS_TYPE" == "true" ] && [ "$HAS_TIMESTAMP" == "true" ]; then - log_success "Logs contain expected fields (type, timestamp)" - else - log_info "Log entry missing expected fields: $FIRST_LOG" - fi - - # Check for ACP message types (session_info should exist) - SESSION_INFO_COUNT=$(echo "$LOGS_RESPONSE" | jq '[.payload.logs[] | select(.type == "session_info")] | length') - if [ "$SESSION_INFO_COUNT" -gt 0 ]; then - log_success "Found $SESSION_INFO_COUNT session_info log entries" - HIST_LOGS_TEST_PASSED=true - else - log_info "No session_info log entries found" - # Still consider it a pass if we got logs - HIST_LOGS_TEST_PASSED=true - fi - else - log_info "No historical logs found (empty array)" - fi -else - log_info "Failed to retrieve historical logs: $(echo "$LOGS_RESPONSE" | jq -r '.payload.error // "unknown error"')" -fi - -log_step "Step 16: Test Historical Logs Replay on Subscribe" -# Subscribe to completed task and check if historical logs are replayed as notifications -# Use a temp file to collect messages -SUBSCRIBE_OUTPUT_FILE=$(mktemp) - -# Send subscribe request and collect responses for a few seconds -( - echo '{"id":"hist-sub-1","type":"request","action":"task.subscribe","payload":{"task_id":"'"$TASK_ID"'"}}' - sleep 3 -) | timeout 5 websocat "${WS_URL}" > "$SUBSCRIBE_OUTPUT_FILE" 2>/dev/null || true - -# Count the messages received -if [ -f "$SUBSCRIBE_OUTPUT_FILE" ] && [ -s "$SUBSCRIBE_OUTPUT_FILE" ]; then - TOTAL_MESSAGES=$(wc -l < "$SUBSCRIBE_OUTPUT_FILE" | tr -d ' ') - log_info "Received $TOTAL_MESSAGES messages from subscribe" - - # Check for the subscribe response - SUBSCRIBE_RESPONSE=$(grep -m1 '"action":"task.subscribe"' "$SUBSCRIBE_OUTPUT_FILE" || true) - if [ -n "$SUBSCRIBE_RESPONSE" ]; then - log_success "Received subscribe response" - fi - - # Count historical notifications (acp.* actions) - ACP_MESSAGES=$(grep -c '"action":"acp\.' "$SUBSCRIBE_OUTPUT_FILE" 2>/dev/null || echo "0") - if [ "$ACP_MESSAGES" -gt 0 ]; then - log_success "Received $ACP_MESSAGES historical ACP notifications" - HIST_SUBSCRIBE_COUNT=$ACP_MESSAGES - HIST_SUBSCRIBE_TEST_PASSED=true - else - log_info "No historical ACP notifications received (may be expected if task has no ACP logs)" - # Check for any notification type messages - NOTIF_COUNT=$(grep -c '"type":"notification"' "$SUBSCRIBE_OUTPUT_FILE" 2>/dev/null || echo "0") - if [ "$NOTIF_COUNT" -gt 0 ]; then - log_info "Received $NOTIF_COUNT notification messages" - HIST_SUBSCRIBE_COUNT=$NOTIF_COUNT - HIST_SUBSCRIBE_TEST_PASSED=true - fi - fi -else - log_info "No messages received from subscribe (file empty or missing)" -fi -rm -f "$SUBSCRIBE_OUTPUT_FILE" - -log_step "Step 17: Create and Monitor New Task with Live Streaming" -# Create a new task to test real-time streaming + historical logs -NEW_TEST_WORKSPACE=$(mktemp -d) -log_info "New test workspace: $NEW_TEST_WORKSPACE" - -NEW_TASK_PAYLOAD=$(cat < "$STREAM_OUTPUT_FILE" 2>/dev/null & - STREAM_PID=$! - - # Wait a moment for subscription to be established - sleep 2 - - # Start the task - log_info "Starting task via orchestrator..." - START_RESP=$(ws_request "orchestrator.start" "{\"task_id\": \"${NEW_TASK_ID}\"}") - NEW_AGENT_ID=$(echo "$START_RESP" | jq -r '.payload.agent_instance_id') - - if [ "$NEW_AGENT_ID" != "null" ] && [ -n "$NEW_AGENT_ID" ]; then - log_success "Task started, agent ID: $NEW_AGENT_ID" - - # Wait for agent to complete (become READY) - log_info "Waiting for agent to complete..." - for i in $(seq 1 120); do - STATUS_RESP=$(ws_request "agent.status" "{\"agent_id\": \"${NEW_AGENT_ID}\"}" 2>/dev/null || echo '{}') - STATUS=$(echo "$STATUS_RESP" | jq -r '.payload.status // "UNKNOWN"') - if [ "$STATUS" == "READY" ]; then - log_success "Agent completed after $i seconds" - break - elif [ "$STATUS" == "FAILED" ]; then - log_info "Agent failed" - break - fi - if [ $((i % 15)) -eq 0 ]; then - log_info "Still waiting... ($i seconds, status: $STATUS)" - fi - sleep 1 - done - - # Give a moment for final messages to arrive - sleep 2 - - # Kill the streaming process - kill $STREAM_PID 2>/dev/null || true - wait $STREAM_PID 2>/dev/null || true - - # Count real-time messages received - if [ -f "$STREAM_OUTPUT_FILE" ] && [ -s "$STREAM_OUTPUT_FILE" ]; then - REALTIME_TOTAL=$(wc -l < "$STREAM_OUTPUT_FILE" | tr -d ' ') - REALTIME_NOTIF=$(grep -c '"type":"notification"' "$STREAM_OUTPUT_FILE" 2>/dev/null || echo "0") - REALTIME_ACP=$(grep -c '"action":"acp\.' "$STREAM_OUTPUT_FILE" 2>/dev/null || echo "0") - log_success "Real-time streaming: $REALTIME_TOTAL messages, $REALTIME_NOTIF notifications, $REALTIME_ACP ACP messages" - LIVE_STREAM_COUNT=$REALTIME_NOTIF - - if [ "$REALTIME_NOTIF" -gt 0 ]; then - LIVE_STREAM_TEST_PASSED=true - fi - else - log_info "No streaming messages collected" - fi - - # Complete the new task - ws_request "orchestrator.complete" "{\"task_id\": \"${NEW_TASK_ID}\"}" > /dev/null 2>&1 || true - - # Now use task.logs to get full history - log_info "Retrieving full history via task.logs..." - NEW_LOGS_RESP=$(ws_request "task.logs" "{\"task_id\": \"${NEW_TASK_ID}\"}") - NEW_LOGS_COUNT=$(echo "$NEW_LOGS_RESP" | jq '.payload.logs | length' 2>/dev/null || echo "0") - - if [ "$NEW_LOGS_COUNT" -gt 0 ]; then - log_success "task.logs returned $NEW_LOGS_COUNT historical entries" - - # Compare with real-time count - if [ "$REALTIME_ACP" -gt 0 ]; then - log_info "Comparison: $REALTIME_ACP real-time ACP messages vs $NEW_LOGS_COUNT historical logs" - fi - else - log_info "task.logs returned no entries" - fi - else - log_info "Failed to start new task, skipping streaming test" - kill $STREAM_PID 2>/dev/null || true - fi - - rm -f "$STREAM_OUTPUT_FILE" - rm -rf "$NEW_TEST_WORKSPACE" 2>/dev/null || true -fi - -log_step "Step 18: Test Agent Input Request Flow" -# Create a task that explicitly asks the agent to request user input -# This tests the bidirectional comment flow -INPUT_TEST_WORKSPACE=$(mktemp -d) -log_info "Input test workspace: $INPUT_TEST_WORKSPACE" - -# Create a task that instructs the agent to ask for more information -INPUT_TASK_PAYLOAD=$(cat < "$INPUT_STREAM_FILE" 2>/dev/null & - INPUT_STREAM_PID=$! - sleep 2 - - # Start the task - log_info "Starting ambiguous task..." - INPUT_START_RESP=$(ws_request "orchestrator.start" "{\"task_id\": \"${INPUT_TASK_ID}\"}") - INPUT_AGENT_ID=$(echo "$INPUT_START_RESP" | jq -r '.payload.agent_instance_id') - - if [ "$INPUT_AGENT_ID" != "null" ] && [ -n "$INPUT_AGENT_ID" ]; then - log_success "Task started, agent ID: $INPUT_AGENT_ID" - - # Monitor for WAITING_FOR_INPUT state or completion - INPUT_RECEIVED=false - for i in $(seq 1 45); do - # Check task state - TASK_STATE_RESP=$(ws_request "task.get" "{\"id\": \"${INPUT_TASK_ID}\"}" 2>/dev/null || echo '{}') - CURRENT_STATE=$(echo "$TASK_STATE_RESP" | jq -r '.payload.state // "UNKNOWN"') - - if [ "$CURRENT_STATE" == "WAITING_FOR_INPUT" ]; then - log_success "Task entered WAITING_FOR_INPUT state after $i seconds" - INPUT_RECEIVED=true - - # Get comments to see the agent's question - TASK_COMMENTS=$(ws_request "comment.list" "{\"task_id\": \"${INPUT_TASK_ID}\"}" 2>/dev/null || echo '{}') - AGENT_COMMENTS=$(echo "$TASK_COMMENTS" | jq '[.payload.comments[] | select(.author_type == "agent")] | length' 2>/dev/null || echo "0") - if [ "$AGENT_COMMENTS" -gt 0 ]; then - log_success "Agent created $AGENT_COMMENTS comment(s) requesting input" - AGENT_QUESTION=$(echo "$TASK_COMMENTS" | jq -r '.payload.comments[] | select(.author_type == "agent") | .content' | head -1) - log_info "Agent asked: ${AGENT_QUESTION:0:100}..." - fi - - # Test responding via comment with the numbers to sum - log_info "Sending user response via comment..." - USER_RESPONSE_PAYLOAD=$(cat </dev/null || echo '{}') - RESUME_STATE=$(echo "$RESUME_STATE_RESP" | jq -r '.payload.state // "UNKNOWN"') - if [ "$RESUME_STATE" != "WAITING_FOR_INPUT" ]; then - log_success "Task resumed, new state: $RESUME_STATE" - INPUT_REQUEST_TEST_PASSED=true - break - fi - sleep 1 - done - break - fi - - # Check if agent finished without requesting input (also valid) - AGENT_STATUS_RESP=$(ws_request "agent.status" "{\"agent_id\": \"${INPUT_AGENT_ID}\"}" 2>/dev/null || echo '{}') - AGENT_STATUS=$(echo "$AGENT_STATUS_RESP" | jq -r '.payload.status // "UNKNOWN"') - if [ "$AGENT_STATUS" == "READY" ] || [ "$AGENT_STATUS" == "FAILED" ]; then - log_info "Agent completed without requesting input (status: $AGENT_STATUS)" - # Still consider the plumbing test passed if we got here without errors - INPUT_REQUEST_TEST_PASSED=true - break - fi - - if [ $((i % 10)) -eq 0 ]; then - log_info "Monitoring... ($i seconds, state: $CURRENT_STATE, agent: $AGENT_STATUS)" - fi - sleep 1 - done - - # Check stream for input.requested notifications - kill $INPUT_STREAM_PID 2>/dev/null || true - wait $INPUT_STREAM_PID 2>/dev/null || true - - if [ -f "$INPUT_STREAM_FILE" ] && [ -s "$INPUT_STREAM_FILE" ]; then - INPUT_NOTIF_COUNT=$(grep -c '"action":"input.requested"' "$INPUT_STREAM_FILE" 2>/dev/null | tr -d '\n' || echo "0") - if [ -n "$INPUT_NOTIF_COUNT" ] && [ "$INPUT_NOTIF_COUNT" -gt 0 ] 2>/dev/null; then - log_success "Received $INPUT_NOTIF_COUNT input.requested notification(s)" - INPUT_REQUEST_TEST_PASSED=true - else - log_info "No input.requested notifications received (agent may not have requested input)" - fi - fi - - # Complete the task - ws_request "orchestrator.complete" "{\"task_id\": \"${INPUT_TASK_ID}\"}" > /dev/null 2>&1 || true - else - log_info "Failed to start input test task" - kill $INPUT_STREAM_PID 2>/dev/null || true - fi - - rm -f "$INPUT_STREAM_FILE" - rm -rf "$INPUT_TEST_WORKSPACE" 2>/dev/null || true -fi - -# Clean up test workspace -rm -rf "$TEST_WORKSPACE" 2>/dev/null || true - -# Show backend logs summary -log_step "Backend Log Summary" -if [ -f /tmp/kandev-e2e.log ]; then - log_info "Key events from backend:" - grep -E "(agent|agentctl|container|task)" /tmp/kandev-e2e.log | tail -20 || true -fi +# Clean up +rm -rf "$TEST_WORKSPACE_DIR" 2>/dev/null || true # Summary log_step "Test Summary" -# Format historical logs test result -if [ "$HIST_LOGS_TEST_PASSED" == "true" ]; then - HIST_LOGS_RESULT="${GREEN}✓ ($HIST_LOGS_COUNT logs)${NC}" +# Determine test results +COMMENT_ADD_PASSED=false +COMMENT_LIST_PASSED=false +MULTI_TURN_PASSED=false +METADATA_PASSED=false + +[ -n "$TURN1_ID" ] && [ -n "$TURN2_ID" ] && [ -n "$TURN3_ID" ] && COMMENT_ADD_PASSED=true +[ "$COMMENT_COUNT" -gt 0 ] && COMMENT_LIST_PASSED=true +[ "$USER_COMMENTS" -ge 3 ] && MULTI_TURN_PASSED=true +[ "$HAS_TIMESTAMPS" == "true" ] && METADATA_PASSED=true + +echo -e "\n${GREEN}╔═══════════════════════════════════════════════════════╗${NC}" +echo -e "${GREEN}║ Comment System E2E Test Results ║${NC}" +echo -e "${GREEN}╠═══════════════════════════════════════════════════════╣${NC}" + +# Core functionality +if [ "$COMMENT_ADD_PASSED" == "true" ]; then + echo -e "${GREEN}║ comment.add: ✓ (3 user comments) ║${NC}" else - HIST_LOGS_RESULT="${YELLOW}○ (skipped)${NC}" + echo -e "${RED}║ comment.add: ✗ ║${NC}" fi -# Format subscribe replay test result -if [ "$HIST_SUBSCRIBE_TEST_PASSED" == "true" ]; then - HIST_SUBSCRIBE_RESULT="${GREEN}✓ ($HIST_SUBSCRIBE_COUNT msgs)${NC}" +if [ "$COMMENT_LIST_PASSED" == "true" ]; then + echo -e "${GREEN}║ comment.list: ✓ ($COMMENT_COUNT total comments) ║${NC}" else - HIST_SUBSCRIBE_RESULT="${YELLOW}○ (skipped)${NC}" + echo -e "${RED}║ comment.list: ✗ ║${NC}" fi -# Format live streaming test result -if [ "$LIVE_STREAM_TEST_PASSED" == "true" ]; then - LIVE_STREAM_RESULT="${GREEN}✓ ($LIVE_STREAM_COUNT msgs)${NC}" +if [ "$MULTI_TURN_PASSED" == "true" ]; then + echo -e "${GREEN}║ Multi-turn conversation: ✓ (3 turns completed) ║${NC}" else - LIVE_STREAM_RESULT="${YELLOW}○ (skipped)${NC}" + echo -e "${YELLOW}║ Multi-turn conversation: ○ (partial) ║${NC}" fi -# Format comment test result -if [ "$COMMENT_TEST_PASSED" == "true" ]; then - COMMENT_RESULT="${GREEN}✓${NC}" +if [ "$AGENT_COMMENTS" -gt 0 ]; then + echo -e "${GREEN}║ Agent responses: ✓ ($AGENT_COMMENTS agent comments) ║${NC}" else - COMMENT_RESULT="${YELLOW}○ (skipped)${NC}" + echo -e "${YELLOW}║ Agent responses: ○ (pending) ║${NC}" fi -# Format input request test result -if [ "$INPUT_REQUEST_TEST_PASSED" == "true" ]; then - INPUT_REQUEST_RESULT="${GREEN}✓${NC}" +if [ "$METADATA_PASSED" == "true" ]; then + echo -e "${GREEN}║ Comment metadata: ✓ (timestamps present) ║${NC}" else - INPUT_REQUEST_RESULT="${YELLOW}○ (skipped)${NC}" + echo -e "${YELLOW}║ Comment metadata: ○ ║${NC}" fi -echo -e "${GREEN}╔═══════════════════════════════════════════════════╗${NC}" -echo -e "${GREEN}║ End-to-End Test PASSED! ✓ ║${NC}" -echo -e "${GREEN}╠═══════════════════════════════════════════════════╣${NC}" -echo -e "${GREEN}║ Board created: ✓ ║${NC}" -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}║ First prompt: ✓ ║${NC}" -echo -e "${GREEN}║ Multi-turn: ✓ ║${NC}" -echo -e "${GREEN}║ Task completed: ✓ ║${NC}" -echo -e "${GREEN}║ Session ID stored: ✓ ║${NC}" -echo -e "${GREEN}╠═══════════════════════════════════════════════════╣${NC}" -echo -e "║ Comment System: $COMMENT_RESULT" -echo -e "║ Input Request Flow: $INPUT_REQUEST_RESULT" -echo -e "║ Historical Logs: $HIST_LOGS_RESULT" -echo -e "║ Subscribe Replay: $HIST_SUBSCRIBE_RESULT" -echo -e "║ Live Streaming: $LIVE_STREAM_RESULT" -echo -e "${GREEN}╚═══════════════════════════════════════════════════╝${NC}" +if [ "$REQUESTS_INPUT_COUNT" -gt 0 ]; then + echo -e "${GREEN}║ Input request flow: ✓ ($REQUESTS_INPUT_COUNT requests) ║${NC}" +else + echo -e "${YELLOW}║ Input request flow: ○ (not triggered) ║${NC}" +fi -exit 0 +echo -e "${GREEN}╠═══════════════════════════════════════════════════════╣${NC}" +echo -e "${GREEN}║ Workspace/Board/Column: ✓ ║${NC}" +echo -e "${GREEN}║ Task creation: ✓ ║${NC}" +echo -e "${GREEN}║ Agent execution: ✓ ║${NC}" +echo -e "${GREEN}║ Task completion: ✓ ║${NC}" +echo -e "${GREEN}╚═══════════════════════════════════════════════════════╝${NC}" +# Overall result +if [ "$COMMENT_ADD_PASSED" == "true" ] && [ "$COMMENT_LIST_PASSED" == "true" ]; then + echo -e "\n${GREEN}✓ Comment System E2E Test PASSED${NC}" + exit 0 +else + echo -e "\n${RED}✗ Comment System E2E Test FAILED${NC}" + exit 1 +fi \ No newline at end of file