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/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 cbc9fd6227..903952756b 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -573,6 +573,11 @@ func (a *lifecycleAdapter) PromptAgent(ctx context.Context, agentInstanceID stri }, 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) { 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..2ba6095f98 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -311,7 +311,10 @@ 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), @@ -382,6 +385,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", @@ -1076,4 +1147,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/integration/orchestrator_test.go b/apps/backend/internal/integration/orchestrator_test.go index 52626c0cb8..195af17898 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" @@ -347,6 +348,16 @@ func (s *SimulatedAgentManagerClient) PromptAgent(ctx context.Context, agentInst }, 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..e719ddda13 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -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() @@ -382,3 +396,13 @@ func (m *MockAgentManagerClient) PromptAgent(ctx context.Context, agentInstanceI zap.Int("prompt_length", len(prompt))) return &PromptResult{StopReason: "end_turn", NeedsInput: false}, 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..ccf300d0be 100644 --- a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go +++ b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go @@ -59,6 +59,10 @@ func (m *mockAgentManager) PromptAgent(ctx context.Context, agentInstanceID stri return &executor.PromptResult{StopReason: "end_turn", NeedsInput: false}, 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 type testTaskRepository struct { tasks map[string]*v1.Task diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 941e9fa0c5..1cb7688322 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -293,6 +293,16 @@ func (s *Service) PromptTask(ctx context.Context, taskID string, prompt string) }, 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", diff --git a/apps/backend/internal/orchestrator/wshandlers/handlers.go b/apps/backend/internal/orchestrator/wshandlers/handlers.go index 1fab7101ac..7841b1d6f3 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 @@ -294,3 +295,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/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 1fad66b6b2..a9a44dabcb 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -87,6 +87,10 @@ const ( ActionColumnDeleted = "column.deleted" ActionCommentAdded = "comment.added" ActionInputRequested = "input.requested" + + // 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/scripts/e2e-test.sh b/scripts/e2e-test.sh index 2241678268..61bad3bed4 100755 --- a/scripts/e2e-test.sh +++ b/scripts/e2e-test.sh @@ -193,9 +193,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 +213,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 +223,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" # 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 TASK_PAYLOAD=$(cat < /dev/null 2>&1 && echo "true" || echo "false") @@ -544,7 +555,7 @@ 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" +log_step "Step 17: 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) @@ -587,7 +598,7 @@ else fi rm -f "$SUBSCRIBE_OUTPUT_FILE" -log_step "Step 17: Create and Monitor New Task with Live Streaming" +log_step "Step 18: 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" @@ -596,6 +607,7 @@ NEW_TASK_PAYLOAD=$(cat </dev/null || true fi -log_step "Step 18: Test Agent Input Request Flow" +log_step "Step 19: 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) @@ -713,6 +725,7 @@ INPUT_TASK_PAYLOAD=$(cat </dev/null || true fi +log_step "Step 20: Test Permission Response WebSocket Action" +# Test that the permission.respond action is properly registered and validates input +# We send a request with a non-existent pending_id to verify the handler works +PERMISSION_TEST_PASSED=false + +# Test 1: Missing task_id should return validation error +PERM_RESP1=$(ws_request "permission.respond" '{"pending_id": "test-pending-id", "option_id": "test-option"}') +PERM_ERR1=$(echo "$PERM_RESP1" | jq -r '.payload.code // empty') +if [ "$PERM_ERR1" == "VALIDATION_ERROR" ]; then + log_success "permission.respond correctly validates missing task_id" +else + log_info "permission.respond validation test 1 result: $PERM_RESP1" +fi + +# Test 2: Missing pending_id should return validation error +PERM_RESP2=$(ws_request "permission.respond" '{"task_id": "test-task-id", "option_id": "test-option"}') +PERM_ERR2=$(echo "$PERM_RESP2" | jq -r '.payload.code // empty') +if [ "$PERM_ERR2" == "VALIDATION_ERROR" ]; then + log_success "permission.respond correctly validates missing pending_id" +else + log_info "permission.respond validation test 2 result: $PERM_RESP2" +fi + +# Test 3: Missing option_id (when not cancelled) should return validation error +PERM_RESP3=$(ws_request "permission.respond" '{"task_id": "test-task-id", "pending_id": "test-pending-id"}') +PERM_ERR3=$(echo "$PERM_RESP3" | jq -r '.payload.code // empty') +if [ "$PERM_ERR3" == "VALIDATION_ERROR" ]; then + log_success "permission.respond correctly validates missing option_id" + PERMISSION_TEST_PASSED=true +else + log_info "permission.respond validation test 3 result: $PERM_RESP3" +fi + +# Test 4: Valid request with non-existent task should return internal error (task not found) +PERM_RESP4=$(ws_request "permission.respond" '{"task_id": "non-existent-task", "pending_id": "test-pending-id", "option_id": "test-option"}') +PERM_ERR4=$(echo "$PERM_RESP4" | jq -r '.payload.code // empty') +if [ "$PERM_ERR4" == "INTERNAL_ERROR" ]; then + log_success "permission.respond correctly handles non-existent task" +else + log_info "permission.respond test 4 result: $PERM_RESP4" +fi + # Clean up test workspace -rm -rf "$TEST_WORKSPACE" 2>/dev/null || true +rm -rf "$TEST_WORKSPACE_DIR" 2>/dev/null || true # Show backend logs summary log_step "Backend Log Summary" @@ -885,6 +940,13 @@ else INPUT_REQUEST_RESULT="${YELLOW}○ (skipped)${NC}" fi +# Format permission test result +if [ "$PERMISSION_TEST_PASSED" == "true" ]; then + PERMISSION_RESULT="${GREEN}✓${NC}" +else + PERMISSION_RESULT="${YELLOW}○ (skipped)${NC}" +fi + echo -e "${GREEN}╔═══════════════════════════════════════════════════╗${NC}" echo -e "${GREEN}║ End-to-End Test PASSED! ✓ ║${NC}" echo -e "${GREEN}╠═══════════════════════════════════════════════════╣${NC}" @@ -900,6 +962,7 @@ 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 "║ Permission Response: $PERMISSION_RESULT" echo -e "║ Historical Logs: $HIST_LOGS_RESULT" echo -e "║ Subscribe Replay: $HIST_SUBSCRIBE_RESULT" echo -e "║ Live Streaming: $LIVE_STREAM_RESULT"