diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index d9ad32f574..4703869f85 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -322,13 +322,13 @@ func main() { var containerMgr *lifecycle.ContainerManager switch cfg.Agent.Runtime { case "standalone": - standaloneCtl := agentctl.NewStandaloneCtl( + controlClient := agentctl.NewControlClient( cfg.Agent.StandaloneHost, cfg.Agent.StandalonePort, log, ) agentRuntime = lifecycle.NewStandaloneRuntime( - standaloneCtl, + controlClient, cfg.Agent.StandaloneHost, cfg.Agent.StandalonePort, log, @@ -421,7 +421,8 @@ func main() { taskSvc.SetExecutionStopper(orchestratorSvc) // Set up comment creator for saving agent responses as comments - orchestratorSvc.SetMessageCreator(&messageCreatorAdapter{svc: taskSvc}) + msgCreator := &messageCreatorAdapter{svc: taskSvc} + orchestratorSvc.SetMessageCreator(msgCreator) // ============================================ // WEBSOCKET GATEWAY (All communication via WebSocket) @@ -554,44 +555,8 @@ func main() { }) log.Info("Historical logs provider configured for task subscriptions (using messages and git status)") - // Set up pending permissions provider for task subscriptions - gateway.Hub.SetPendingPermissionsProvider(func(ctx context.Context, taskID string) ([]*ws.Message, error) { - if lifecycleMgr == nil { - return nil, nil - } - session, err := taskRepo.GetActiveTaskSessionByTaskID(ctx, taskID) - if err != nil || session == nil || session.ID == "" { - return nil, nil - } - permissions, err := lifecycleMgr.GetPendingPermissionsForSession(ctx, session.ID) - if err != nil { - return nil, err - } - - result := make([]*ws.Message, 0, len(permissions)) - for _, p := range permissions { - // Build payload matching the permission.requested format - payload := map[string]interface{}{ - "type": "permission_request", - "task_id": taskID, - "pending_id": p.PendingID, - "session_id": p.SessionID, - "tool_call_id": p.ToolCallID, - "title": p.Title, - "options": p.Options, - "action_type": p.ActionType, - "action_details": p.ActionDetails, - "created_at": p.CreatedAt.Format(time.RFC3339), - } - notification, err := ws.NewNotification(ws.ActionPermissionRequested, payload) - if err != nil { - continue - } - result = append(result, notification) - } - return result, nil - }) - log.Info("Pending permissions provider configured for task subscriptions") + // NOTE: Pending permissions are now stored as messages and retrieved via the historical logs provider. + // The separate pending permissions provider is no longer needed. // NOTE: We no longer create comments for each ACP message chunk. // Instead, the lifecycle manager accumulates message chunks and: @@ -600,35 +565,9 @@ func main() { // This is handled via the prompt.complete and step.complete events. // Wire ACP handler to broadcast to WebSocket clients as notifications + // NOTE: Permission requests are now handled via message creation in the lifecycle manager + // and broadcast automatically via the session.message.added event orchestratorSvc.RegisterACPHandler(func(taskID string, msg *protocol.Message) { - // Check for permission_request events (type is in msg.Type, payload in msg.Data) - if msg.Type == "permission_request" { - sessionID := msg.SessionID - if sessionID == "" { - log.Warn("permission_request missing session_id, skipping", - zap.String("task_id", taskID)) - return - } - // Build payload for frontend with all necessary fields - payload := map[string]interface{}{ - "type": "permission_request", - "task_id": taskID, - "agent_instance_id": msg.AgentID, - "timestamp": msg.Timestamp, - } - // Copy all fields from msg.Data (pending_id, options, action_type, etc.) - for k, v := range msg.Data { - payload[k] = v - } - // Broadcast permission.requested notification to task subscribers - notification, _ := ws.NewNotification(ws.ActionPermissionRequested, payload) - gateway.Hub.BroadcastToSession(sessionID, notification) - log.Info("broadcasted permission.requested notification", - zap.String("task_id", taskID), - zap.Any("pending_id", payload["pending_id"])) - return - } - // Only broadcast message types that the frontend handles // Skip internal message types like session/update switch msg.Type { @@ -1311,3 +1250,32 @@ func (a *messageCreatorAdapter) CreateSessionMessage(ctx context.Context, taskID }) return err } + +// CreatePermissionRequestMessage creates a message for a permission request +func (a *messageCreatorAdapter) CreatePermissionRequestMessage(ctx context.Context, taskID, sessionID, pendingID, toolCallID, title string, options []map[string]interface{}, actionType string, actionDetails map[string]interface{}) (string, error) { + metadata := map[string]interface{}{ + "pending_id": pendingID, + "tool_call_id": toolCallID, + "options": options, + "action_type": actionType, + "action_details": actionDetails, + } + + msg, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{ + TaskSessionID: sessionID, + TaskID: taskID, + Content: title, + AuthorType: "agent", + Type: "permission_request", + Metadata: metadata, + }) + if err != nil { + return "", err + } + return msg.ID, nil +} + +// UpdatePermissionMessage updates a permission message's status +func (a *messageCreatorAdapter) UpdatePermissionMessage(ctx context.Context, sessionID, pendingID, status string) error { + return a.svc.UpdatePermissionMessage(ctx, sessionID, pendingID, status) +} diff --git a/apps/backend/internal/agent/lifecycle/container.go b/apps/backend/internal/agent/lifecycle/container.go index 4e6f9501b9..666934f47f 100644 --- a/apps/backend/internal/agent/lifecycle/container.go +++ b/apps/backend/internal/agent/lifecycle/container.go @@ -78,8 +78,8 @@ func (cm *ContainerManager) LaunchContainer(ctx context.Context, config Containe containerIP = "127.0.0.1" } - // Create StandaloneCtl to communicate with the container's ControlServer - ctl := agentctl.NewStandaloneCtl(containerIP, AgentCtlPort, cm.logger) + // Create ControlClient to communicate with the container's control server + ctl := agentctl.NewControlClient(containerIP, AgentCtlPort, cm.logger) // Wait for agentctl to be healthy if err := cm.waitForHealth(ctx, ctl); err != nil { @@ -115,7 +115,7 @@ func (cm *ContainerManager) LaunchContainer(ctx context.Context, config Containe } // waitForHealth waits for agentctl to be healthy with retries -func (cm *ContainerManager) waitForHealth(ctx context.Context, ctl *agentctl.StandaloneCtl) error { +func (cm *ContainerManager) waitForHealth(ctx context.Context, ctl *agentctl.ControlClient) error { const maxRetries = 30 const retryDelay = 500 * time.Millisecond diff --git a/apps/backend/internal/agent/lifecycle/events.go b/apps/backend/internal/agent/lifecycle/events.go index 06d567278f..d10902dc4b 100644 --- a/apps/backend/internal/agent/lifecycle/events.go +++ b/apps/backend/internal/agent/lifecycle/events.go @@ -114,63 +114,64 @@ func (p *EventPublisher) PublishACPSessionCreated(execution *AgentExecution, ses } } -// PublishSessionUpdate publishes a session update to the event bus for WebSocket streaming. -func (p *EventPublisher) PublishSessionUpdate(execution *AgentExecution, update agentctl.SessionUpdate) { +// PublishAgentStreamEvent publishes an agent stream event to the event bus for WebSocket streaming. +// This is different from PublishAgentEvent which publishes lifecycle events (started, stopped, etc.). +func (p *EventPublisher) PublishAgentStreamEvent(execution *AgentExecution, event agentctl.AgentEvent) { if p.eventBus == nil { return } - // Build the update data - our SessionUpdate type marshals cleanly - updateData := map[string]interface{}{ - "type": update.Type, + // Build the event data - our AgentEvent type marshals cleanly + eventData := map[string]interface{}{ + "type": event.Type, } - if update.SessionID != "" { - updateData["session_id"] = update.SessionID + if event.SessionID != "" { + eventData["session_id"] = event.SessionID } - if update.Text != "" { - updateData["text"] = update.Text + if event.Text != "" { + eventData["text"] = event.Text } - if update.ToolCallID != "" { - updateData["tool_call_id"] = update.ToolCallID + if event.ToolCallID != "" { + eventData["tool_call_id"] = event.ToolCallID } - if update.ToolName != "" { - updateData["tool_name"] = update.ToolName + if event.ToolName != "" { + eventData["tool_name"] = event.ToolName } - if update.ToolTitle != "" { - updateData["tool_title"] = update.ToolTitle + if event.ToolTitle != "" { + eventData["tool_title"] = event.ToolTitle } - if update.ToolStatus != "" { - updateData["tool_status"] = update.ToolStatus + if event.ToolStatus != "" { + eventData["tool_status"] = event.ToolStatus } - if update.ToolArgs != nil { - updateData["tool_args"] = update.ToolArgs + if event.ToolArgs != nil { + eventData["tool_args"] = event.ToolArgs } - if update.ToolResult != nil { - updateData["tool_result"] = update.ToolResult + if event.ToolResult != nil { + eventData["tool_result"] = event.ToolResult } - if update.Error != "" { - updateData["error"] = update.Error + if event.Error != "" { + eventData["error"] = event.Error } - if update.Data != nil { - updateData["data"] = update.Data + if event.Data != nil { + eventData["data"] = event.Data } - // Build ACP message data + // Build agent event message data data := map[string]interface{}{ - "type": "session/update", + "type": "agent/event", "timestamp": time.Now().UTC().Format(time.RFC3339Nano), "agent_id": execution.ID, "task_id": execution.TaskID, - "session_id": update.SessionID, - "data": updateData, + "session_id": event.SessionID, + "data": eventData, } - event := bus.NewEvent(events.ACPMessage, "agent-manager", data) + busEvent := bus.NewEvent(events.ACPMessage, "agent-manager", data) subject := events.BuildACPSubject(execution.TaskID) - if err := p.eventBus.Publish(context.Background(), subject, event); err != nil { - p.logger.Error("failed to publish session update", + if err := p.eventBus.Publish(context.Background(), subject, busEvent); err != nil { + p.logger.Error("failed to publish agent stream event", zap.String("instance_id", execution.ID), zap.String("task_id", execution.TaskID), zap.Error(err)) @@ -320,8 +321,8 @@ func (p *EventPublisher) PublishToolCall(execution *AgentExecution, toolCallID, } } -// PublishToolCallComplete publishes a tool call completion event from a SessionUpdate. -func (p *EventPublisher) PublishToolCallComplete(execution *AgentExecution, update agentctl.SessionUpdate) { +// PublishToolCallComplete publishes a tool call completion event from an AgentEvent. +func (p *EventPublisher) PublishToolCallComplete(execution *AgentExecution, event agentctl.AgentEvent) { if p.eventBus == nil { return } @@ -334,14 +335,14 @@ func (p *EventPublisher) PublishToolCallComplete(execution *AgentExecution, upda "agent_id": execution.ID, "task_id": execution.TaskID, "session_id": sessionID, - "tool_call_id": update.ToolCallID, - "status": update.ToolStatus, + "tool_call_id": event.ToolCallID, + "status": event.ToolStatus, } - event := bus.NewEvent(events.ToolCallComplete, "agent-manager", data) + busEvent := bus.NewEvent(events.ToolCallComplete, "agent-manager", data) subject := events.ToolCallComplete + "." + execution.TaskID - if err := p.eventBus.Publish(context.Background(), subject, event); err != nil { + if err := p.eventBus.Publish(context.Background(), subject, busEvent); err != nil { p.logger.Error("failed to publish tool_call_complete event", zap.String("instance_id", execution.ID), zap.String("task_id", execution.TaskID), @@ -349,16 +350,15 @@ func (p *EventPublisher) PublishToolCallComplete(execution *AgentExecution, upda } } -// PublishPermissionRequest publishes a permission request event. -// The event is structured as a protocol.Message so it can be parsed by the watcher. -func (p *EventPublisher) PublishPermissionRequest(execution *AgentExecution, notification *agentctl.PermissionNotification) { +// PublishPermissionRequest publishes a permission request event to the event bus. +func (p *EventPublisher) PublishPermissionRequest(execution *AgentExecution, event agentctl.AgentEvent) { if p.eventBus == nil { return } // Convert options to a serializable format - options := make([]map[string]interface{}, len(notification.Options)) - for i, opt := range notification.Options { + options := make([]map[string]interface{}, len(event.PermissionOptions)) + for i, opt := range event.PermissionOptions { options[i] = map[string]interface{}{ "option_id": opt.OptionID, "name": opt.Name, @@ -366,32 +366,24 @@ func (p *EventPublisher) PublishPermissionRequest(execution *AgentExecution, not } } - // Build payload data (goes into msg.Data when parsed as protocol.Message) - payloadData := map[string]interface{}{ - "pending_id": notification.PendingID, - "session_id": notification.SessionID, - "tool_call_id": notification.ToolCallID, - "title": notification.Title, - "options": options, - "action_type": notification.ActionType, - "action_details": notification.ActionDetails, - "created_at": notification.CreatedAt, - } - - // Structure as protocol.Message so watcher can parse it correctly data := map[string]interface{}{ - "type": "permission_request", - "timestamp": time.Now().UTC().Format(time.RFC3339Nano), - "agent_id": execution.ID, - "task_id": execution.TaskID, - "session_id": notification.SessionID, - "data": payloadData, + "type": "permission_request", + "timestamp": time.Now().UTC().Format(time.RFC3339Nano), + "agent_id": execution.ID, + "task_id": execution.TaskID, + "session_id": execution.SessionID, + "pending_id": event.PendingID, + "tool_call_id": event.ToolCallID, + "title": event.PermissionTitle, + "options": options, + "action_type": event.ActionType, + "action_details": event.ActionDetails, } - event := bus.NewEvent(events.ACPMessage, "agent-manager", data) - subject := events.BuildACPSubject(execution.TaskID) + busEvent := bus.NewEvent(events.PermissionRequestReceived, "agent-manager", data) + subject := events.PermissionRequestReceived + "." + execution.TaskID - if err := p.eventBus.Publish(context.Background(), subject, event); err != nil { + if err := p.eventBus.Publish(context.Background(), subject, busEvent); err != nil { p.logger.Error("failed to publish permission_request event", zap.String("instance_id", execution.ID), zap.String("task_id", execution.TaskID), @@ -399,7 +391,7 @@ func (p *EventPublisher) PublishPermissionRequest(execution *AgentExecution, not } else { p.logger.Debug("published permission_request event", zap.String("task_id", execution.TaskID), - zap.String("pending_id", notification.PendingID), - zap.String("title", notification.Title)) + zap.String("pending_id", event.PendingID), + zap.String("title", event.PermissionTitle)) } } diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 800d45ec20..bd11666458 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -99,10 +99,9 @@ func NewManager( // Initialize stream manager with callbacks that delegate to manager methods mgr.streamManager = NewStreamManager(log, StreamCallbacks{ - OnSessionUpdate: mgr.handleSessionUpdate, - OnPermission: mgr.handlePermissionNotification, - OnGitStatus: mgr.handleGitStatusUpdate, - OnFileChange: mgr.handleFileChangeNotification, + OnAgentEvent: mgr.handleAgentEvent, + OnGitStatus: mgr.handleGitStatusUpdate, + OnFileChange: mgr.handleFileChangeNotification, }) // Set session manager dependencies for full orchestration @@ -708,34 +707,33 @@ func (m *Manager) initializeACPSession(ctx context.Context, execution *AgentExec return m.sessionManager.InitializeAndPrompt(ctx, execution, agentConfig, taskDescription, m.MarkReady) } -// handlePermissionNotification processes incoming permission requests from the agent -func (m *Manager) handlePermissionNotification(execution *AgentExecution, notification *agentctl.PermissionNotification) { - // Publish permission request event to the event bus (logged at Debug level there) - m.eventPublisher.PublishPermissionRequest(execution, notification) +// handlePermissionRequestEvent processes permission requests from the unified agent event stream +func (m *Manager) handlePermissionRequestEvent(execution *AgentExecution, event agentctl.AgentEvent) { + m.eventPublisher.PublishPermissionRequest(execution, event) } -// handleSessionUpdate processes incoming session updates from the agent -func (m *Manager) handleSessionUpdate(execution *AgentExecution, update agentctl.SessionUpdate) { - // Handle different update types based on the Type field - switch update.Type { +// handleAgentEvent processes incoming agent events from the agent +func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.AgentEvent) { + // Handle different event types based on the Type field + switch event.Type { case "message_chunk": m.updateExecutionProgress(execution.ID, 50) // Accumulate message content for saving as comment when a step completes - if update.Text != "" { + if event.Text != "" { execution.messageMu.Lock() - execution.messageBuffer.WriteString(update.Text) + execution.messageBuffer.WriteString(event.Text) execution.messageMu.Unlock() } case "reasoning": // Accumulate reasoning/thinking content execution.messageMu.Lock() - if update.ReasoningText != "" { - execution.reasoningBuffer.WriteString(update.ReasoningText) + if event.ReasoningText != "" { + execution.reasoningBuffer.WriteString(event.ReasoningText) } - if update.ReasoningSummary != "" { - execution.summaryBuffer.WriteString(update.ReasoningSummary) + if event.ReasoningSummary != "" { + execution.summaryBuffer.WriteString(event.ReasoningSummary) } execution.messageMu.Unlock() @@ -746,21 +744,21 @@ func (m *Manager) handleSessionUpdate(execution *AgentExecution, update agentctl m.logger.Debug("tool call started", zap.String("execution_id", execution.ID), - zap.String("tool_call_id", update.ToolCallID), - zap.String("tool_name", update.ToolName)) + zap.String("tool_call_id", event.ToolCallID), + zap.String("tool_name", event.ToolName)) m.updateExecutionProgress(execution.ID, 60) // Publish tool call as a comment so it appears in the chat - m.eventPublisher.PublishToolCall(execution, update.ToolCallID, update.ToolTitle, update.ToolStatus, update.ToolArgs) + m.eventPublisher.PublishToolCall(execution, event.ToolCallID, event.ToolTitle, event.ToolStatus, event.ToolArgs) case "tool_update": // Check if tool call completed - switch update.ToolStatus { + switch event.ToolStatus { case "complete", "completed": m.updateExecutionProgress(execution.ID, 80) - m.eventPublisher.PublishToolCallComplete(execution, update) + m.eventPublisher.PublishToolCallComplete(execution, event) case "error", "failed": - m.eventPublisher.PublishToolCallComplete(execution, update) + m.eventPublisher.PublishToolCallComplete(execution, event) } case "plan": @@ -770,12 +768,12 @@ func (m *Manager) handleSessionUpdate(execution *AgentExecution, update agentctl case "error": m.logger.Error("agent error", zap.String("execution_id", execution.ID), - zap.String("error", update.Error)) + zap.String("error", event.Error)) case "complete": m.logger.Info("agent turn complete", zap.String("execution_id", execution.ID), - zap.String("session_id", update.SessionID)) + zap.String("session_id", event.SessionID)) // Flush accumulated message buffer as a comment m.flushMessageBufferAsComment(execution) @@ -786,10 +784,19 @@ func (m *Manager) handleSessionUpdate(execution *AgentExecution, update agentctl zap.String("execution_id", execution.ID), zap.Error(err)) } + + case "permission_request": + m.logger.Debug("permission request received", + zap.String("execution_id", execution.ID), + zap.String("pending_id", event.PendingID), + zap.String("title", event.PermissionTitle)) + + // Handle permission request inline + m.handlePermissionRequestEvent(execution, event) } - // Publish session update to event bus for WebSocket streaming - m.eventPublisher.PublishSessionUpdate(execution, update) + // Publish agent stream event to event bus for WebSocket streaming + m.eventPublisher.PublishAgentStreamEvent(execution, event) } // handleGitStatusUpdate processes git status updates from the workspace tracker @@ -1195,17 +1202,4 @@ func (m *Manager) RespondToPermissionBySessionID(sessionID, pendingID, optionID return m.RespondToPermission(execution.ID, pendingID, optionID, cancelled) } -// GetPendingPermissionsForSession returns all pending permissions for a session -func (m *Manager) GetPendingPermissionsForSession(ctx context.Context, sessionID string) ([]agentctl.PermissionNotification, error) { - execution, exists := m.executionStore.GetBySessionID(sessionID) - if !exists { - // No active execution for this session, so no pending permissions - return nil, nil - } - if execution.agentctl == nil { - return nil, nil - } - - return execution.agentctl.GetPendingPermissions(ctx) -} diff --git a/apps/backend/internal/agent/lifecycle/runtime_docker.go b/apps/backend/internal/agent/lifecycle/runtime_docker.go index 1087958641..45b423b10f 100644 --- a/apps/backend/internal/agent/lifecycle/runtime_docker.go +++ b/apps/backend/internal/agent/lifecycle/runtime_docker.go @@ -154,8 +154,8 @@ func (r *DockerRuntime) RecoverInstances(ctx context.Context) ([]*RuntimeInstanc containerIP = "127.0.0.1" } - // Query the ControlServer to get the instance port - ctl := agentctl.NewStandaloneCtl(containerIP, AgentCtlPort, r.logger) + // Query the control server to get the instance port + ctl := agentctl.NewControlClient(containerIP, AgentCtlPort, r.logger) instanceInfo, err := ctl.GetInstance(ctx, instanceID) if err != nil { r.logger.Warn("failed to get instance info from container", diff --git a/apps/backend/internal/agent/lifecycle/runtime_standalone.go b/apps/backend/internal/agent/lifecycle/runtime_standalone.go index 63e482f8af..546e6ab075 100644 --- a/apps/backend/internal/agent/lifecycle/runtime_standalone.go +++ b/apps/backend/internal/agent/lifecycle/runtime_standalone.go @@ -14,14 +14,14 @@ import ( // StandaloneRuntime implements Runtime for standalone agentctl execution. // In this mode, a single agentctl control server manages multiple agent instances. type StandaloneRuntime struct { - ctl *agentctl.StandaloneCtl + ctl *agentctl.ControlClient host string port int logger *logger.Logger } // NewStandaloneRuntime creates a new standalone runtime. -func NewStandaloneRuntime(ctl *agentctl.StandaloneCtl, host string, port int, log *logger.Logger) *StandaloneRuntime { +func NewStandaloneRuntime(ctl *agentctl.ControlClient, host string, port int, log *logger.Logger) *StandaloneRuntime { return &StandaloneRuntime{ ctl: ctl, host: host, diff --git a/apps/backend/internal/agent/lifecycle/streams.go b/apps/backend/internal/agent/lifecycle/streams.go index 12505b23f2..0efca0a64d 100644 --- a/apps/backend/internal/agent/lifecycle/streams.go +++ b/apps/backend/internal/agent/lifecycle/streams.go @@ -12,10 +12,9 @@ import ( // StreamCallbacks defines callbacks for stream events type StreamCallbacks struct { - OnSessionUpdate func(execution *AgentExecution, update agentctl.SessionUpdate) - OnPermission func(execution *AgentExecution, notification *agentctl.PermissionNotification) - OnGitStatus func(execution *AgentExecution, update *agentctl.GitStatusUpdate) - OnFileChange func(execution *AgentExecution, notification *agentctl.FileChangeNotification) + OnAgentEvent func(execution *AgentExecution, event agentctl.AgentEvent) + OnGitStatus func(execution *AgentExecution, update *agentctl.GitStatusUpdate) + OnFileChange func(execution *AgentExecution, notification *agentctl.FileChangeNotification) } // StreamManager manages WebSocket streams to agent executions @@ -36,7 +35,6 @@ func NewStreamManager(log *logger.Logger, callbacks StreamCallbacks) *StreamMana // If ready is non-nil, it will be closed when the updates stream is connected. func (sm *StreamManager) ConnectAll(execution *AgentExecution, ready chan<- struct{}) { go sm.connectUpdatesStream(execution, ready) - go sm.connectPermissionStream(execution) go sm.connectGitStatusStream(execution) go sm.connectFileChangesStream(execution) } @@ -74,9 +72,9 @@ func (sm *StreamManager) ReconnectAll(execution *AgentExecution) { func (sm *StreamManager) connectUpdatesStream(execution *AgentExecution, ready chan<- struct{}) { ctx := context.Background() - err := execution.agentctl.StreamUpdates(ctx, func(update agentctl.SessionUpdate) { - if sm.callbacks.OnSessionUpdate != nil { - sm.callbacks.OnSessionUpdate(execution, update) + err := execution.agentctl.StreamUpdates(ctx, func(event agentctl.AgentEvent) { + if sm.callbacks.OnAgentEvent != nil { + sm.callbacks.OnAgentEvent(execution, event) } }) @@ -94,28 +92,7 @@ func (sm *StreamManager) connectUpdatesStream(execution *AgentExecution, ready c } } -// connectPermissionStream handles the permission WebSocket stream -func (sm *StreamManager) connectPermissionStream(execution *AgentExecution) { - ctx := context.Background() - sm.logger.Debug("connecting to permission stream", - zap.String("instance_id", execution.ID), - zap.String("task_id", execution.TaskID)) - - err := execution.agentctl.StreamPermissions(ctx, func(notification *agentctl.PermissionNotification) { - sm.logger.Debug("received permission notification from stream", - zap.String("pending_id", notification.PendingID), - zap.String("title", notification.Title)) - if sm.callbacks.OnPermission != nil { - sm.callbacks.OnPermission(execution, notification) - } - }) - if err != nil { - sm.logger.Error("failed to connect to permission stream", - zap.String("instance_id", execution.ID), - zap.Error(err)) - } -} // connectGitStatusStream handles git status stream with retry logic func (sm *StreamManager) connectGitStatusStream(execution *AgentExecution) { diff --git a/apps/backend/internal/agentctl/client/acp.go b/apps/backend/internal/agentctl/client/acp.go index a2a66dfc85..6bf072f415 100644 --- a/apps/backend/internal/agentctl/client/acp.go +++ b/apps/backend/internal/agentctl/client/acp.go @@ -10,9 +10,19 @@ import ( "github.com/coder/acp-go-sdk" "github.com/gorilla/websocket" + "github.com/kandev/kandev/internal/agentctl/types/streams" "go.uber.org/zap" ) +// Re-export stream types for convenience. +type ( + AgentEvent = streams.AgentEvent + PermissionNotification = streams.PermissionNotification + PermissionOption = streams.PermissionOption + PermissionRespondRequest = streams.PermissionRespondRequest + PermissionRespondResponse = streams.PermissionRespondResponse +) + // AgentInfo contains information about the connected agent. type AgentInfo struct { Name string `json:"name"` @@ -237,70 +247,12 @@ func (c *Client) Prompt(ctx context.Context, text string) (*PromptResponse, erro return &result, nil } -// SessionUpdate represents a session update from the agent. -// This matches adapter.SessionUpdate that agentctl sends over WebSocket. -type SessionUpdate struct { - Type string `json:"type"` - SessionID string `json:"session_id,omitempty"` - OperationID string `json:"operation_id,omitempty"` // Turn/operation ID for cancellation +// Note: AgentEvent, PermissionNotification, PermissionOption, and +// PermissionRespondRequest are re-exported from streams package at the top of this file. - // Message fields - Text string `json:"text,omitempty"` - - // Reasoning fields (for "reasoning" type) - ReasoningText string `json:"reasoning_text,omitempty"` - ReasoningSummary string `json:"reasoning_summary,omitempty"` - - // Tool call fields - ToolCallID string `json:"tool_call_id,omitempty"` - ToolName string `json:"tool_name,omitempty"` - ToolTitle string `json:"tool_title,omitempty"` - ToolStatus string `json:"tool_status,omitempty"` - ToolArgs map[string]interface{} `json:"tool_args,omitempty"` - ToolResult interface{} `json:"tool_result,omitempty"` - - // Diff for file changes - Diff string `json:"diff,omitempty"` - - // Error and raw data - Error string `json:"error,omitempty"` - Data map[string]interface{} `json:"data,omitempty"` -} - -// 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"` - ActionType string `json:"action_type,omitempty"` // command, file_write, file_read, network, mcp_tool, other - ActionDetails map[string]interface{} `json:"action_details,omitempty"` // Additional details about the action - 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(SessionUpdate)) error { +// StreamUpdates opens a WebSocket connection for streaming agent events. +// Events include message chunks, reasoning, tool calls, plan updates, and completion/error. +func (c *Client) StreamUpdates(ctx context.Context, handler func(AgentEvent)) error { wsURL := "ws" + c.baseURL[4:] + "/api/v1/acp/stream" conn, _, err := websocket.DefaultDialer.DialContext(ctx, wsURL, nil) @@ -336,99 +288,33 @@ func (c *Client) StreamUpdates(ctx context.Context, handler func(SessionUpdate)) return } - var update SessionUpdate - if err := json.Unmarshal(message, &update); err != nil { - c.logger.Warn("failed to parse session update", zap.Error(err)) + var event AgentEvent + if err := json.Unmarshal(message, &event); err != nil { + c.logger.Warn("failed to parse agent event", zap.Error(err)) continue } - handler(update) + handler(event) } }() return nil } -// CloseUpdatesStream closes the updates stream connection +// CloseUpdatesStream closes the agent events stream connection. func (c *Client) CloseUpdatesStream() { - c.CloseACPStream() -} - -// CloseACPStream closes the ACP stream connection -func (c *Client) CloseACPStream() { c.mu.Lock() defer c.mu.Unlock() if c.acpConn != nil { if err := c.acpConn.Close(); err != nil { - c.logger.Debug("failed to close ACP stream", zap.Error(err)) + c.logger.Debug("failed to close agent events stream", zap.Error(err)) } c.acpConn = nil } } -// 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() - if err := conn.Close(); err != nil { - c.logger.Debug("failed to close permissions websocket", zap.Error(err)) - } - }() - - 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 { - if err := c.permissionConn.Close(); err != nil { - c.logger.Debug("failed to close permission stream", zap.Error(err)) - } - c.permissionConn = nil - } -} // RespondToPermission sends a response to a permission request func (c *Client) RespondToPermission(ctx context.Context, pendingID, optionID string, cancelled bool) error { @@ -478,35 +364,4 @@ func (c *Client) RespondToPermission(ctx context.Context, pendingID, optionID st return nil } -// GetPendingPermissions retrieves all pending permission requests from agentctl -func (c *Client) GetPendingPermissions(ctx context.Context) ([]PermissionNotification, error) { - req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/api/v1/acp/permissions", nil) - if err != nil { - return nil, err - } - resp, err := c.httpClient.Do(req) - if err != nil { - return nil, err - } - defer func() { - if err := resp.Body.Close(); err != nil { - c.logger.Debug("failed to close permissions response body", zap.Error(err)) - } - }() - - respBody, err := readResponseBody(resp) - if err != nil { - return nil, fmt.Errorf("failed to read response body: %w", err) - } - - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return nil, fmt.Errorf("get pending permissions failed with status %d: %s", resp.StatusCode, string(respBody)) - } - - var result []PermissionNotification - if err := json.Unmarshal(respBody, &result); err != nil { - return nil, fmt.Errorf("failed to parse permissions response: %w", err) - } - return result, nil -} diff --git a/apps/backend/internal/agentctl/client/client.go b/apps/backend/internal/agentctl/client/client.go index fc23fbd43f..d8d17b49be 100644 --- a/apps/backend/internal/agentctl/client/client.go +++ b/apps/backend/internal/agentctl/client/client.go @@ -24,7 +24,6 @@ type Client struct { // WebSocket connections for streaming acpConn *websocket.Conn - permissionConn *websocket.Conn gitStatusConn *websocket.Conn fileChangesConn *websocket.Conn shellConn *websocket.Conn @@ -251,7 +250,8 @@ func (c *Client) BaseURL() string { return c.baseURL } -// Re-export types from types package for convenience +// Re-export types from types package for convenience. +// These types are defined in the streams subpackage and re-exported through types. type ( GitStatusUpdate = types.GitStatusUpdate FileInfo = types.FileInfo @@ -263,6 +263,9 @@ type ( FileContentRequest = types.FileContentRequest FileContentResponse = types.FileContentResponse FileChangeNotification = types.FileChangeNotification + ShellMessage = types.ShellMessage + ShellStatusResponse = types.ShellStatusResponse + ShellBufferResponse = types.ShellBufferResponse ) // StreamGitStatus opens a WebSocket connection for streaming git status updates @@ -453,23 +456,16 @@ func (c *Client) CloseFileChangesStream() { } } -// Close closes all connections (ACP, permissions, git status, file changes, shell) +// Close closes all connections func (c *Client) Close() { - c.CloseACPStream() - c.ClosePermissionStream() + c.CloseUpdatesStream() c.CloseGitStatusStream() c.CloseFileChangesStream() c.CloseShellStream() } -// ShellStatusResponse from agentctl shell status endpoint -type ShellStatusResponse struct { - Running bool `json:"running"` - Pid int `json:"pid"` - Shell string `json:"shell"` - Cwd string `json:"cwd"` - StartedAt string `json:"started_at"` -} +// Note: ShellStatusResponse, ShellMessage, ShellBufferResponse are re-exported +// from streams package at the top of this file in the type block. // ShellStatus gets the status of the embedded shell session func (c *Client) ShellStatus(ctx context.Context) (*ShellStatusResponse, error) { @@ -496,18 +492,6 @@ func (c *Client) ShellStatus(ctx context.Context) (*ShellStatusResponse, error) return &result, nil } -// ShellMessage represents a message to/from the shell -type ShellMessage struct { - Type string `json:"type"` // "input", "output", "ping", "pong", "exit" - Data string `json:"data,omitempty"` - Code int `json:"code,omitempty"` // For exit type -} - -// ShellBufferResponse is the response from the shell buffer endpoint -type ShellBufferResponse struct { - Data string `json:"data"` -} - // ShellBuffer returns the buffered shell output func (c *Client) ShellBuffer(ctx context.Context) (string, error) { req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/api/v1/shell/buffer", nil) diff --git a/apps/backend/internal/agentctl/client/standalone.go b/apps/backend/internal/agentctl/client/control.go similarity index 78% rename from apps/backend/internal/agentctl/client/standalone.go rename to apps/backend/internal/agentctl/client/control.go index 942bcba510..3c2b9bc55f 100644 --- a/apps/backend/internal/agentctl/client/standalone.go +++ b/apps/backend/internal/agentctl/client/control.go @@ -1,5 +1,5 @@ // Package agentctl provides a client for communicating with agentctl. -// This file contains the StandaloneCtl client for the multi-instance control API. +// This file contains the ControlClient for the agentctl control server API. package client import ( @@ -14,9 +14,10 @@ import ( "go.uber.org/zap" ) -// StandaloneCtl is a client for the agentctl control API in standalone mode. +// ControlClient is a client for the agentctl control server API. // It manages creation and deletion of agent instances. -type StandaloneCtl struct { +// Used by both Docker and standalone runtimes. +type ControlClient struct { baseURL string httpClient *http.Client logger *logger.Logger @@ -50,25 +51,25 @@ type InstanceInfo struct { CreatedAt time.Time `json:"created_at"` } -// NewStandaloneCtl creates a new StandaloneCtl client. -func NewStandaloneCtl(host string, port int, log *logger.Logger) *StandaloneCtl { - return &StandaloneCtl{ +// NewControlClient creates a new ControlClient for the agentctl control server. +func NewControlClient(host string, port int, log *logger.Logger) *ControlClient { + return &ControlClient{ baseURL: fmt.Sprintf("http://%s:%d", host, port), httpClient: &http.Client{ Timeout: 30 * time.Second, }, - logger: log.WithFields(zap.String("component", "standalone-agentctl")), + logger: log.WithFields(zap.String("component", "agentctl-control")), } } -// Health checks if the standalone agentctl is healthy. -func (s *StandaloneCtl) Health(ctx context.Context) error { - req, err := http.NewRequestWithContext(ctx, "GET", s.baseURL+"/health", nil) +// Health checks if the agentctl control server is healthy. +func (c *ControlClient) Health(ctx context.Context) error { + req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/health", nil) if err != nil { return err } - resp, err := s.httpClient.Do(req) + resp, err := c.httpClient.Do(req) if err != nil { return err } @@ -83,19 +84,19 @@ func (s *StandaloneCtl) Health(ctx context.Context) error { } // CreateInstance creates a new agent instance. -func (s *StandaloneCtl) CreateInstance(ctx context.Context, req *CreateInstanceRequest) (*CreateInstanceResponse, error) { +func (c *ControlClient) CreateInstance(ctx context.Context, req *CreateInstanceRequest) (*CreateInstanceResponse, error) { body, err := json.Marshal(req) if err != nil { return nil, fmt.Errorf("failed to marshal request: %w", err) } - httpReq, err := http.NewRequestWithContext(ctx, "POST", s.baseURL+"/api/v1/instances", bytes.NewReader(body)) + httpReq, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/api/v1/instances", bytes.NewReader(body)) if err != nil { return nil, err } httpReq.Header.Set("Content-Type", "application/json") - resp, err := s.httpClient.Do(httpReq) + resp, err := c.httpClient.Do(httpReq) if err != nil { return nil, fmt.Errorf("failed to create instance: %w", err) } @@ -116,7 +117,7 @@ func (s *StandaloneCtl) CreateInstance(ctx context.Context, req *CreateInstanceR return nil, fmt.Errorf("failed to decode response: %w", err) } - s.logger.Info("created agent instance", + c.logger.Info("created agent instance", zap.String("instance_id", result.ID), zap.Int("port", result.Port)) @@ -124,13 +125,13 @@ func (s *StandaloneCtl) CreateInstance(ctx context.Context, req *CreateInstanceR } // DeleteInstance stops and removes an agent instance. -func (s *StandaloneCtl) DeleteInstance(ctx context.Context, instanceID string) error { - req, err := http.NewRequestWithContext(ctx, "DELETE", s.baseURL+"/api/v1/instances/"+instanceID, nil) +func (c *ControlClient) DeleteInstance(ctx context.Context, instanceID string) error { + req, err := http.NewRequestWithContext(ctx, "DELETE", c.baseURL+"/api/v1/instances/"+instanceID, nil) if err != nil { return err } - resp, err := s.httpClient.Do(req) + resp, err := c.httpClient.Do(req) if err != nil { return fmt.Errorf("failed to delete instance: %w", err) } @@ -148,18 +149,18 @@ func (s *StandaloneCtl) DeleteInstance(ctx context.Context, instanceID string) e return fmt.Errorf("failed to delete instance: %s (status %d)", errResp.Error, resp.StatusCode) } - s.logger.Info("deleted agent instance", zap.String("instance_id", instanceID)) + c.logger.Info("deleted agent instance", zap.String("instance_id", instanceID)) return nil } // GetInstance gets information about a specific instance. -func (s *StandaloneCtl) GetInstance(ctx context.Context, instanceID string) (*InstanceInfo, error) { - req, err := http.NewRequestWithContext(ctx, "GET", s.baseURL+"/api/v1/instances/"+instanceID, nil) +func (c *ControlClient) GetInstance(ctx context.Context, instanceID string) (*InstanceInfo, error) { + req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/api/v1/instances/"+instanceID, nil) if err != nil { return nil, err } - resp, err := s.httpClient.Do(req) + resp, err := c.httpClient.Do(req) if err != nil { return nil, fmt.Errorf("failed to get instance: %w", err) } @@ -180,13 +181,13 @@ func (s *StandaloneCtl) GetInstance(ctx context.Context, instanceID string) (*In } // ListInstances lists all running agent instances. -func (s *StandaloneCtl) ListInstances(ctx context.Context) ([]*InstanceInfo, error) { - req, err := http.NewRequestWithContext(ctx, "GET", s.baseURL+"/api/v1/instances", nil) +func (c *ControlClient) ListInstances(ctx context.Context) ([]*InstanceInfo, error) { + req, err := http.NewRequestWithContext(ctx, "GET", c.baseURL+"/api/v1/instances", nil) if err != nil { return nil, err } - resp, err := s.httpClient.Do(req) + resp, err := c.httpClient.Do(req) if err != nil { return nil, fmt.Errorf("failed to list instances: %w", err) } diff --git a/apps/backend/internal/agentctl/server/adapter/acp_adapter.go b/apps/backend/internal/agentctl/server/adapter/acp_adapter.go index fd59c6a46e..04886676c7 100644 --- a/apps/backend/internal/agentctl/server/adapter/acp_adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/acp_adapter.go @@ -34,7 +34,7 @@ type ACPAdapter struct { agentInfo *AgentInfo // Update channel - updatesCh chan SessionUpdate + updatesCh chan AgentEvent // Permission handler permissionHandler PermissionHandler @@ -52,7 +52,7 @@ func NewACPAdapter(stdin io.Writer, stdout io.Reader, cfg *Config, log *logger.L logger: log.WithFields(zap.String("adapter", "acp")), stdin: stdin, stdout: stdout, - updatesCh: make(chan SessionUpdate, 100), + updatesCh: make(chan AgentEvent, 100), } } @@ -62,7 +62,7 @@ func (a *ACPAdapter) Initialize(ctx context.Context) error { a.logger.Info("initializing ACP adapter", zap.String("workdir", a.cfg.WorkDir)) - // Create ACP client with update handler that converts to SessionUpdate + // Create ACP client with update handler that converts to AgentEvent a.acpClient = acpclient.NewClient( acpclient.WithLogger(a.logger.Zap()), acpclient.WithWorkspaceRoot(a.cfg.WorkDir), @@ -190,8 +190,8 @@ func (a *ACPAdapter) Cancel(ctx context.Context) error { }) } -// Updates returns the channel for session updates. -func (a *ACPAdapter) Updates() <-chan SessionUpdate { +// Updates returns the channel for agent events. +func (a *ACPAdapter) Updates() <-chan AgentEvent { return a.updatesCh } @@ -244,20 +244,20 @@ func (a *ACPAdapter) GetACPConnection() *acp.ClientSideConnection { return a.acpConn } -// handleACPUpdate converts ACP SessionNotification to protocol-agnostic SessionUpdate. +// handleACPUpdate converts ACP SessionNotification to protocol-agnostic AgentEvent. func (a *ACPAdapter) handleACPUpdate(n acp.SessionNotification) { - update := a.convertNotification(n) - if update != nil { + event := a.convertNotification(n) + if event != nil { select { - case a.updatesCh <- *update: + case a.updatesCh <- *event: default: a.logger.Warn("updates channel full, dropping notification") } } } -// convertNotification converts an ACP SessionNotification to a SessionUpdate. -func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpdate { +// convertNotification converts an ACP SessionNotification to an AgentEvent. +func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *AgentEvent { u := n.Update // Log what update types we're receiving @@ -271,8 +271,8 @@ func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpda switch { case u.AgentMessageChunk != nil: if u.AgentMessageChunk.Content.Text != nil { - return &SessionUpdate{ - Type: UpdateTypeMessageChunk, + return &AgentEvent{ + Type: EventTypeMessageChunk, SessionID: string(n.SessionId), Text: u.AgentMessageChunk.Content.Text.Text, } @@ -281,8 +281,8 @@ func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpda case u.AgentThoughtChunk != nil: // Agent thinking/reasoning - map to the new reasoning type if u.AgentThoughtChunk.Content.Text != nil { - return &SessionUpdate{ - Type: UpdateTypeReasoning, + return &AgentEvent{ + Type: EventTypeReasoning, SessionID: string(n.SessionId), ReasoningText: u.AgentThoughtChunk.Content.Text.Text, } @@ -324,8 +324,8 @@ func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpda status = "running" } - return &SessionUpdate{ - Type: UpdateTypeToolCall, + return &AgentEvent{ + Type: EventTypeToolCall, SessionID: string(n.SessionId), ToolCallID: string(u.ToolCall.ToolCallId), ToolName: string(u.ToolCall.Kind), // Kind is effectively the tool name @@ -343,8 +343,8 @@ func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpda if status == "completed" { status = "complete" } - return &SessionUpdate{ - Type: UpdateTypeToolUpdate, + return &AgentEvent{ + Type: EventTypeToolUpdate, SessionID: string(n.SessionId), ToolCallID: string(u.ToolCallUpdate.ToolCallId), ToolStatus: status, @@ -359,8 +359,8 @@ func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpda Priority: string(e.Priority), } } - return &SessionUpdate{ - Type: UpdateTypePlan, + return &AgentEvent{ + Type: EventTypePlan, SessionID: string(n.SessionId), PlanEntries: entries, } diff --git a/apps/backend/internal/agentctl/server/adapter/adapter.go b/apps/backend/internal/agentctl/server/adapter/adapter.go index d7b3ca1ebe..eede0a154a 100644 --- a/apps/backend/internal/agentctl/server/adapter/adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/adapter.go @@ -6,6 +6,7 @@ import ( "context" "github.com/kandev/kandev/internal/agentctl/types" + "github.com/kandev/kandev/internal/agentctl/types/streams" ) // Re-export permission types from the shared types package for convenience. @@ -14,71 +15,27 @@ import ( type ( PermissionRequest = types.PermissionRequest PermissionResponse = types.PermissionResponse - PermissionOption = types.PermissionOption + PermissionOption = streams.PermissionOption PermissionHandler = types.PermissionHandler ) -// SessionUpdate type constants -const ( - UpdateTypeMessageChunk = "message_chunk" // Agent text streaming - UpdateTypeReasoning = "reasoning" // Chain-of-thought/thinking - UpdateTypeToolCall = "tool_call" // Tool invocation started - UpdateTypeToolUpdate = "tool_update" // Tool status update - UpdateTypePlan = "plan" // Agent plan updates - UpdateTypeComplete = "complete" // Turn/session complete - UpdateTypeError = "error" // Error occurred +// Re-export stream types for convenience. +type ( + AgentEvent = streams.AgentEvent + PlanEntry = streams.PlanEntry ) -// SessionUpdate is a protocol-agnostic update from the agent. -// All protocol adapters normalize their updates to this format. -type SessionUpdate struct { - // Type identifies the update type. Use UpdateType* constants. - Type string `json:"type"` - - // SessionID is the current session identifier - SessionID string `json:"session_id,omitempty"` - - // OperationID identifies the current in-flight operation (turn, prompt, etc.) - // Used to target specific operations for cancellation or status updates. - // For Codex this is the turn ID, for other protocols it may be empty. - OperationID string `json:"operation_id,omitempty"` - - // Message fields (for "message_chunk" type) - Text string `json:"text,omitempty"` - - // Reasoning fields (for "reasoning" type) - // Used for chain-of-thought, thinking traces, etc. - ReasoningText string `json:"reasoning_text,omitempty"` // Full reasoning content - ReasoningSummary string `json:"reasoning_summary,omitempty"` // Summarized version (if available) - - // Tool call fields (for "tool_call" and "tool_update" types) - ToolCallID string `json:"tool_call_id,omitempty"` - ToolName string `json:"tool_name,omitempty"` - ToolTitle string `json:"tool_title,omitempty"` - ToolStatus string `json:"tool_status,omitempty"` // "started", "running", "completed", "error" - ToolArgs map[string]interface{} `json:"tool_args,omitempty"` - ToolResult interface{} `json:"tool_result,omitempty"` - - // Diff contains unified diff content for file changes. - // Populated when tools modify files, providing the aggregated diff. - Diff string `json:"diff,omitempty"` - - // Plan fields (for "plan" type) - PlanEntries []PlanEntry `json:"plan_entries,omitempty"` - - // Error fields (for "error" type) - Error string `json:"error,omitempty"` - - // Raw data for protocol-specific extensions - Data map[string]interface{} `json:"data,omitempty"` -} - -// PlanEntry represents an entry in the agent's execution plan -type PlanEntry struct { - Description string `json:"description,omitempty"` // Content/description of the task - Status string `json:"status,omitempty"` // "pending", "in_progress", "completed", "failed" - Priority string `json:"priority,omitempty"` // Relative importance -} +// Re-export agent event type constants from streams package. +const ( + EventTypeMessageChunk = streams.EventTypeMessageChunk + EventTypeReasoning = streams.EventTypeReasoning + EventTypeToolCall = streams.EventTypeToolCall + EventTypeToolUpdate = streams.EventTypeToolUpdate + EventTypePlan = streams.EventTypePlan + EventTypeComplete = streams.EventTypeComplete + EventTypeError = streams.EventTypeError + EventTypePermissionRequest = streams.EventTypePermissionRequest +) // AgentInfo contains information about the connected agent. type AgentInfo struct { @@ -88,7 +45,7 @@ type AgentInfo struct { // AgentAdapter defines the interface for protocol adapters. // Each adapter translates a specific protocol (ACP, REST, MCP, etc.) into the -// normalized SessionUpdate format that agentctl exposes via its HTTP API. +// normalized AgentEvent format that agentctl exposes via its HTTP API. type AgentAdapter interface { // Initialize establishes the connection with the agent and exchanges capabilities. // For subprocess-based agents (ACP), this sends the initialize request. @@ -112,9 +69,9 @@ type AgentAdapter interface { // Cancel cancels the current operation. Cancel(ctx context.Context) error - // Updates returns a channel that receives session updates. + // Updates returns a channel that receives agent events. // The channel is closed when the adapter is closed. - Updates() <-chan SessionUpdate + Updates() <-chan AgentEvent // GetSessionID returns the current session ID. GetSessionID() string diff --git a/apps/backend/internal/agentctl/server/adapter/adapter_test.go b/apps/backend/internal/agentctl/server/adapter/adapter_test.go index 07a2d9a04e..60ea8428a3 100644 --- a/apps/backend/internal/agentctl/server/adapter/adapter_test.go +++ b/apps/backend/internal/agentctl/server/adapter/adapter_test.go @@ -6,7 +6,7 @@ import ( ) func TestSessionUpdate_Fields(t *testing.T) { - update := SessionUpdate{ + update := AgentEvent{ Type: "tool_call", SessionID: "session-123", Text: "test message", @@ -56,7 +56,7 @@ func TestSessionUpdate_Fields(t *testing.T) { } func TestSessionUpdate_JSONSerialization(t *testing.T) { - update := SessionUpdate{ + update := AgentEvent{ Type: "message_chunk", SessionID: "session-abc", Text: "Hello, world!", @@ -68,7 +68,7 @@ func TestSessionUpdate_JSONSerialization(t *testing.T) { t.Fatalf("failed to marshal SessionUpdate: %v", err) } - var decoded SessionUpdate + var decoded AgentEvent if err := json.Unmarshal(data, &decoded); err != nil { t.Fatalf("failed to unmarshal SessionUpdate: %v", err) } @@ -86,7 +86,7 @@ func TestSessionUpdate_JSONSerialization(t *testing.T) { func TestSessionUpdate_JSONOmitEmpty(t *testing.T) { // Test that omitempty fields are not included when empty - update := SessionUpdate{ + update := AgentEvent{ Type: "error", } @@ -257,7 +257,7 @@ func TestSessionUpdate_UpdateTypes(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { - update := SessionUpdate{Type: tc.updateType} + update := AgentEvent{Type: tc.updateType} if update.Type != tc.updateType { t.Errorf("expected Type %q, got %q", tc.updateType, update.Type) } @@ -288,7 +288,7 @@ func TestPlanEntry_StatusValues(t *testing.T) { func TestSessionUpdate_WithComplexToolArgs(t *testing.T) { // Test with complex nested ToolArgs - update := SessionUpdate{ + update := AgentEvent{ Type: "tool_call", ToolCallID: "call-123", ToolName: "file_edit", @@ -311,7 +311,7 @@ func TestSessionUpdate_WithComplexToolArgs(t *testing.T) { t.Fatalf("failed to marshal: %v", err) } - var decoded SessionUpdate + var decoded AgentEvent if err := json.Unmarshal(data, &decoded); err != nil { t.Fatalf("failed to unmarshal: %v", err) } @@ -328,4 +328,3 @@ func TestSessionUpdate_WithComplexToolArgs(t *testing.T) { t.Errorf("expected 2 locations, got %d", len(locations)) } } - diff --git a/apps/backend/internal/agentctl/server/adapter/codex_adapter.go b/apps/backend/internal/agentctl/server/adapter/codex_adapter.go index ca29f93019..c30dfb73c6 100644 --- a/apps/backend/internal/agentctl/server/adapter/codex_adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/codex_adapter.go @@ -39,7 +39,7 @@ type CodexAdapter struct { agentInfo *AgentInfo // Update channel - updatesCh chan SessionUpdate + updatesCh chan AgentEvent // Permission handler permissionHandler PermissionHandler @@ -65,7 +65,7 @@ func NewCodexAdapter(stdin io.Writer, stdout io.Reader, cfg *Config, log *logger stdout: stdout, ctx: ctx, cancel: cancel, - updatesCh: make(chan SessionUpdate, 100), + updatesCh: make(chan AgentEvent, 100), } } @@ -292,8 +292,8 @@ func (a *CodexAdapter) Cancel(ctx context.Context) error { return err } -// Updates returns the channel for session updates. -func (a *CodexAdapter) Updates() <-chan SessionUpdate { +// Updates returns the channel for agent events. +func (a *CodexAdapter) Updates() <-chan AgentEvent { return a.updatesCh } @@ -346,8 +346,8 @@ func (a *CodexAdapter) Close() error { return nil } -// sendUpdate safely sends an update to the updates channel. -func (a *CodexAdapter) sendUpdate(update SessionUpdate) { +// sendUpdate safely sends an event to the updates channel. +func (a *CodexAdapter) sendUpdate(update AgentEvent) { select { case a.updatesCh <- update: default: @@ -355,7 +355,7 @@ func (a *CodexAdapter) sendUpdate(update SessionUpdate) { } } -// handleNotification processes Codex notifications and emits SessionUpdates. +// handleNotification processes Codex notifications and emits AgentEvents. func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) { a.mu.RLock() threadID := a.threadID @@ -373,8 +373,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) a.mu.Lock() a.messageBuffer += p.Delta a.mu.Unlock() - a.sendUpdate(SessionUpdate{ - Type: UpdateTypeMessageChunk, + a.sendUpdate(AgentEvent{ + Type: EventTypeMessageChunk, SessionID: threadID, OperationID: turnID, Text: p.Delta, @@ -389,8 +389,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) a.mu.Lock() a.reasoningBuffer += p.Delta a.mu.Unlock() - a.sendUpdate(SessionUpdate{ - Type: UpdateTypeReasoning, + a.sendUpdate(AgentEvent{ + Type: EventTypeReasoning, SessionID: threadID, OperationID: turnID, ReasoningText: p.Delta, @@ -405,8 +405,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) a.mu.Lock() a.summaryBuffer += p.Delta a.mu.Unlock() - a.sendUpdate(SessionUpdate{ - Type: UpdateTypeReasoning, + a.sendUpdate(AgentEvent{ + Type: EventTypeReasoning, SessionID: threadID, OperationID: turnID, ReasoningSummary: p.Delta, @@ -418,13 +418,13 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) a.logger.Warn("failed to parse turn completed", zap.Error(err)) return } - update := SessionUpdate{ - Type: UpdateTypeComplete, + update := AgentEvent{ + Type: EventTypeComplete, SessionID: threadID, OperationID: p.TurnID, } if !p.Success && p.Error != "" { - update.Type = UpdateTypeError + update.Type = EventTypeError update.Error = p.Error } a.sendUpdate(update) @@ -435,8 +435,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) a.logger.Warn("failed to parse turn diff updated", zap.Error(err)) return } - a.sendUpdate(SessionUpdate{ - Type: UpdateTypeMessageChunk, + a.sendUpdate(AgentEvent{ + Type: EventTypeMessageChunk, SessionID: threadID, OperationID: p.TurnID, Diff: p.Diff, @@ -455,8 +455,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) Status: e.Status, } } - a.sendUpdate(SessionUpdate{ - Type: UpdateTypePlan, + a.sendUpdate(AgentEvent{ + Type: EventTypePlan, SessionID: threadID, OperationID: p.TurnID, PlanEntries: entries, @@ -468,8 +468,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) a.logger.Warn("failed to parse error notification", zap.Error(err)) return } - a.sendUpdate(SessionUpdate{ - Type: UpdateTypeError, + a.sendUpdate(AgentEvent{ + Type: EventTypeError, SessionID: threadID, Error: p.Message, }) @@ -487,8 +487,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) // Item types: "userMessage", "agentMessage", "commandExecution", "fileChange", "reasoning" switch p.Item.Type { case "commandExecution": - a.sendUpdate(SessionUpdate{ - Type: UpdateTypeToolCall, + a.sendUpdate(AgentEvent{ + Type: EventTypeToolCall, SessionID: threadID, OperationID: turnID, ToolCallID: p.Item.ID, @@ -509,8 +509,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) title += fmt.Sprintf(" (+%d more)", len(p.Item.Changes)-1) } } - a.sendUpdate(SessionUpdate{ - Type: UpdateTypeToolCall, + a.sendUpdate(AgentEvent{ + Type: EventTypeToolCall, SessionID: threadID, OperationID: turnID, ToolCallID: p.Item.ID, @@ -535,8 +535,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) if p.Item.Status == "failed" { status = "error" } - update := SessionUpdate{ - Type: UpdateTypeToolUpdate, + update := AgentEvent{ + Type: EventTypeToolUpdate, SessionID: threadID, OperationID: turnID, ToolCallID: p.Item.ID, @@ -567,8 +567,8 @@ func (a *CodexAdapter) handleNotification(method string, params json.RawMessage) a.logger.Warn("failed to parse command output delta", zap.Error(err)) return } - a.sendUpdate(SessionUpdate{ - Type: UpdateTypeToolUpdate, + a.sendUpdate(AgentEvent{ + Type: EventTypeToolUpdate, SessionID: threadID, OperationID: turnID, ToolCallID: p.ItemID, diff --git a/apps/backend/internal/agentctl/server/api/acp.go b/apps/backend/internal/agentctl/server/api/acp.go index 1523c5a65d..641ea1c572 100644 --- a/apps/backend/internal/agentctl/server/api/acp.go +++ b/apps/backend/internal/agentctl/server/api/acp.go @@ -319,82 +319,3 @@ func (s *Server) handlePermissionRespond(c *gin.Context) { }) } -// 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 func() { - if err := conn.Close(); err != nil { - s.logger.Debug("failed to close permission websocket", zap.Error(err)) - } - }() - - s.logger.Info("Permission stream WebSocket connected") - - // Get the permission requests channel - permissionCh := s.procMgr.GetPermissionRequests() - - // Stream permission notifications to WebSocket - for notification := range permissionCh { - s.logger.Info("streaming permission notification to backend", - zap.String("pending_id", notification.PendingID), - zap.String("title", notification.Title)) - - 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/server/api/server.go b/apps/backend/internal/agentctl/server/api/server.go index 0293443cbf..74219ce515 100644 --- a/apps/backend/internal/agentctl/server/api/server.go +++ b/apps/backend/internal/agentctl/server/api/server.go @@ -72,8 +72,6 @@ func (s *Server) setupRoutes() { 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) // Workspace monitoring (git status, files) diff --git a/apps/backend/internal/agentctl/server/process/manager.go b/apps/backend/internal/agentctl/server/process/manager.go index c11e3fe175..67def5acc9 100644 --- a/apps/backend/internal/agentctl/server/process/manager.go +++ b/apps/backend/internal/agentctl/server/process/manager.go @@ -15,6 +15,7 @@ import ( "github.com/kandev/kandev/internal/agentctl/server/adapter" "github.com/kandev/kandev/internal/agentctl/server/config" "github.com/kandev/kandev/internal/agentctl/server/shell" + "github.com/kandev/kandev/internal/agentctl/types/streams" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/pkg/agent" "go.uber.org/zap" @@ -80,11 +81,8 @@ type Manager struct { // Protocol adapter for agent communication adapter adapter.AgentAdapter - // Session update notifications (protocol-agnostic) - updatesCh chan adapter.SessionUpdate - - // Permission request notifications (sent to backend) - permissionCh chan *PermissionNotification + // Agent event notifications (protocol-agnostic) + updatesCh chan adapter.AgentEvent // Pending permission requests waiting for user response pendingPermissions map[string]*PendingPermission @@ -104,8 +102,7 @@ func NewManager(cfg *config.InstanceConfig, log *logger.Logger) *Manager { cfg: cfg, logger: log.WithFields(zap.String("component", "process-manager")), workspaceTracker: NewWorkspaceTracker(cfg.WorkDir, log), - updatesCh: make(chan adapter.SessionUpdate, 100), - permissionCh: make(chan *PermissionNotification, 10), + updatesCh: make(chan adapter.AgentEvent, 100), pendingPermissions: make(map[string]*PendingPermission), } m.status.Store(StatusStopped) @@ -344,8 +341,8 @@ func (m *Manager) forwardUpdates() { } } -// GetUpdates returns the channel for session update notifications -func (m *Manager) GetUpdates() <-chan adapter.SessionUpdate { +// GetUpdates returns the channel for agent event notifications +func (m *Manager) GetUpdates() <-chan adapter.AgentEvent { return m.updatesCh } @@ -586,38 +583,39 @@ func (m *Manager) autoApprovePermission(req *adapter.PermissionRequest) (*adapte }, nil } -// sendPermissionNotification sends a permission request notification to the backend +// sendPermissionNotification sends a permission request notification through the updates channel 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, - ActionType: pending.Request.ActionType, - ActionDetails: pending.Request.ActionDetails, - CreatedAt: pending.CreatedAt, + // Convert options to streams.PermissionOption (types.PermissionOption is an alias) + options := make([]streams.PermissionOption, len(pending.Request.Options)) + for i, opt := range pending.Request.Options { + options[i] = streams.PermissionOption(opt) } - m.logger.Info("sending permission notification to backend", + event := adapter.AgentEvent{ + Type: adapter.EventTypePermissionRequest, + SessionID: pending.Request.SessionID, + ToolCallID: pending.Request.ToolCallID, + PendingID: pending.ID, + PermissionTitle: pending.Request.Title, + PermissionOptions: options, + ActionType: pending.Request.ActionType, + ActionDetails: pending.Request.ActionDetails, + } + + m.logger.Info("sending permission notification via updates channel", zap.String("pending_id", pending.ID), zap.String("title", pending.Request.Title), zap.String("action_type", pending.Request.ActionType)) select { - case m.permissionCh <- notification: + case m.updatesCh <- event: // Sent successfully default: - m.logger.Warn("permission channel full, dropping notification", + m.logger.Warn("updates channel full, dropping permission 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() @@ -645,17 +643,7 @@ func (m *Manager) RespondToPermission(pendingID string, optionID string, cancell } } -// 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 -} // Shell returns the embedded shell session, or nil if not available func (m *Manager) Shell() *shell.Session { diff --git a/apps/backend/internal/agentctl/types/streams/agent.go b/apps/backend/internal/agentctl/types/streams/agent.go new file mode 100644 index 0000000000..18f951b97c --- /dev/null +++ b/apps/backend/internal/agentctl/types/streams/agent.go @@ -0,0 +1,129 @@ +package streams + +// AgentEvent type constants define the types of events streamed from the agent. +const ( + // EventTypeMessageChunk indicates streaming text content from the agent. + EventTypeMessageChunk = "message_chunk" + + // EventTypeReasoning indicates chain-of-thought or thinking content. + EventTypeReasoning = "reasoning" + + // EventTypeToolCall indicates a tool invocation has started. + EventTypeToolCall = "tool_call" + + // EventTypeToolUpdate indicates a tool status update (running, completed, etc.). + EventTypeToolUpdate = "tool_update" + + // EventTypePlan indicates agent plan/task list updates. + EventTypePlan = "plan" + + // EventTypeComplete indicates the turn or session has completed. + EventTypeComplete = "complete" + + // EventTypeError indicates an error occurred. + EventTypeError = "error" + + // EventTypePermissionRequest indicates the agent is requesting permission. + EventTypePermissionRequest = "permission_request" +) + +// AgentEvent is the message type streamed from the agent process. +// This represents protocol-agnostic events from the agent, normalized from +// various underlying protocols (ACP, Codex, Claude Code, etc.). +// +// Stream endpoint: ws://.../api/v1/agent/events +type AgentEvent struct { + // Type identifies the event type. Use EventType* constants: + // "message_chunk", "reasoning", "tool_call", "tool_update", "plan", "complete", "error" + Type string `json:"type"` + + // SessionID is the current session identifier. + SessionID string `json:"session_id,omitempty"` + + // OperationID identifies the current in-flight operation (turn, prompt, etc.). + // Used to target specific operations for cancellation or status updates. + // For Codex this is the turn ID, for other protocols it may be empty. + OperationID string `json:"operation_id,omitempty"` + + // --- Message fields (for "message_chunk" type) --- + + // Text contains streaming text content from the agent. + Text string `json:"text,omitempty"` + + // --- Reasoning fields (for "reasoning" type) --- + + // ReasoningText contains full reasoning/chain-of-thought content. + ReasoningText string `json:"reasoning_text,omitempty"` + + // ReasoningSummary contains a summarized version of reasoning (if available). + ReasoningSummary string `json:"reasoning_summary,omitempty"` + + // --- Tool call fields (for "tool_call" and "tool_update" types) --- + + // ToolCallID uniquely identifies the tool invocation. + ToolCallID string `json:"tool_call_id,omitempty"` + + // ToolName is the name of the tool being invoked. + ToolName string `json:"tool_name,omitempty"` + + // ToolTitle is a human-readable title for the tool call. + ToolTitle string `json:"tool_title,omitempty"` + + // ToolStatus indicates the current status: "started", "running", "completed", "error". + ToolStatus string `json:"tool_status,omitempty"` + + // ToolArgs contains the arguments passed to the tool. + ToolArgs map[string]interface{} `json:"tool_args,omitempty"` + + // ToolResult contains the result from the tool execution. + ToolResult interface{} `json:"tool_result,omitempty"` + + // Diff contains unified diff content for file changes. + // Populated when tools modify files, providing the aggregated diff. + Diff string `json:"diff,omitempty"` + + // --- Plan fields (for "plan" type) --- + + // PlanEntries contains the agent's execution plan/task list. + PlanEntries []PlanEntry `json:"plan_entries,omitempty"` + + // --- Error fields (for "error" type) --- + + // Error contains error message when Type is "error". + Error string `json:"error,omitempty"` + + // --- Permission request fields (for "permission_request" type) --- + + // PendingID uniquely identifies this pending permission request. + PendingID string `json:"pending_id,omitempty"` + + // PermissionTitle is a human-readable description of the action requiring permission. + PermissionTitle string `json:"permission_title,omitempty"` + + // PermissionOptions contains the available permission choices. + PermissionOptions []PermissionOption `json:"permission_options,omitempty"` + + // ActionType categorizes the action requiring approval. + // Use ActionType* constants: "command", "file_write", "file_read", "network", "mcp_tool", "other". + ActionType string `json:"action_type,omitempty"` + + // ActionDetails contains structured details about the action. + ActionDetails map[string]interface{} `json:"action_details,omitempty"` + + // --- Extension fields --- + + // Data contains raw protocol-specific extensions. + Data map[string]interface{} `json:"data,omitempty"` +} + +// PlanEntry represents an entry in the agent's execution plan. +type PlanEntry struct { + // Description is the content/description of the task. + Description string `json:"description,omitempty"` + + // Status indicates task status: "pending", "in_progress", "completed", "failed". + Status string `json:"status,omitempty"` + + // Priority indicates relative importance. + Priority string `json:"priority,omitempty"` +} diff --git a/apps/backend/internal/agentctl/types/streams/doc.go b/apps/backend/internal/agentctl/types/streams/doc.go new file mode 100644 index 0000000000..b910959ac0 --- /dev/null +++ b/apps/backend/internal/agentctl/types/streams/doc.go @@ -0,0 +1,51 @@ +// Package streams defines the protocol message types for agentctl WebSocket streams. +// +// The agentctl client produces messages to the following channels/streams: +// +// # Agent Events Stream (/api/v1/agent/events) +// +// Streams real-time events from the agent process including message chunks, +// reasoning/thinking content, tool invocations, plan updates, and completion +// or error notifications. This stream is protocol-agnostic and works with +// any agent backend (ACP, Codex, Claude Code, etc.). +// +// Message type: AgentEvent +// +// Event types (use EventType* constants): +// - message_chunk: Streaming text content from the agent +// - reasoning: Chain-of-thought or thinking content +// - tool_call: A tool invocation has started +// - tool_update: Tool status update (running, completed, error) +// - plan: Agent plan/task list updates +// - complete: The turn or operation has completed +// - error: An error occurred +// +// # Permission Stream (/api/v1/acp/permissions/stream) +// +// Streams permission requests from the agent when it needs approval for +// actions like file writes, shell commands, or network access. +// +// Message type: PermissionNotification +// +// # Git Status Stream (/api/v1/workspace/git-status/stream) +// +// Streams git status updates when the workspace state changes. +// +// Message type: GitStatusUpdate +// +// # File Changes Stream (/api/v1/workspace/file-changes/stream) +// +// Streams file system change notifications when files are created, +// modified, deleted, or renamed in the workspace. +// +// Message type: FileChangeNotification +// +// # Shell Stream (/api/v1/shell/stream) +// +// Bidirectional WebSocket for interactive shell I/O. +// +// Message type: ShellMessage (input/output/ping/pong/exit) +// +// All streams use JSON-encoded messages over WebSocket connections. +package streams + diff --git a/apps/backend/internal/agentctl/types/streams/file.go b/apps/backend/internal/agentctl/types/streams/file.go new file mode 100644 index 0000000000..043776ac5e --- /dev/null +++ b/apps/backend/internal/agentctl/types/streams/file.go @@ -0,0 +1,130 @@ +package streams + +import "time" + +// File change operation constants. +const ( + // FileOpCreate indicates a file was created. + FileOpCreate = "create" + + // FileOpWrite indicates a file was written/modified. + FileOpWrite = "write" + + // FileOpRemove indicates a file was removed. + FileOpRemove = "remove" + + // FileOpRename indicates a file was renamed. + FileOpRename = "rename" + + // FileOpChmod indicates file permissions changed. + FileOpChmod = "chmod" + + // FileOpRefresh indicates a refresh/rescan of the file. + FileOpRefresh = "refresh" +) + +// FileChangeNotification is the message type streamed via the file changes stream. +// Represents a filesystem change notification. +// +// Stream endpoint: ws://.../api/v1/workspace/file-changes/stream +type FileChangeNotification struct { + // Timestamp is when the change was detected. + Timestamp time.Time `json:"timestamp"` + + // Path is the file path relative to workspace root. + Path string `json:"path"` + + // Operation indicates the type of change. Use FileOp* constants: + // "create", "write", "remove", "rename", "chmod", "refresh". + Operation string `json:"operation"` +} + +// FileListUpdate represents a file listing update. +// Used for initial sync or full refresh of workspace files. +type FileListUpdate struct { + // Timestamp is when the listing was captured. + Timestamp time.Time `json:"timestamp"` + + // Files contains the list of files in the workspace. + Files []FileEntry `json:"files"` +} + +// FileEntry represents a file in the workspace. +type FileEntry struct { + // Path is the file path relative to workspace root. + Path string `json:"path"` + + // IsDir indicates if this is a directory. + IsDir bool `json:"is_dir"` + + // Size is the file size in bytes (for files only). + Size int64 `json:"size,omitempty"` +} + +// FileTreeNode represents a node in the file tree. +type FileTreeNode struct { + // Name is the file or directory name. + Name string `json:"name"` + + // Path is the full path relative to workspace root. + Path string `json:"path"` + + // IsDir indicates if this is a directory. + IsDir bool `json:"is_dir"` + + // Size is the file size in bytes (for files only). + Size int64 `json:"size,omitempty"` + + // Children contains child nodes (for directories only). + Children []*FileTreeNode `json:"children,omitempty"` +} + +// FileTreeRequest represents a request for file tree. +// +// HTTP endpoint: GET /api/v1/workspace/tree +type FileTreeRequest struct { + // Path is the path to get tree for (relative to workspace root). + Path string `json:"path"` + + // Depth is the depth to traverse (0 = unlimited, 1 = immediate children only). + Depth int `json:"depth"` +} + +// FileTreeResponse represents a response with file tree. +type FileTreeResponse struct { + // RequestID identifies the request. + RequestID string `json:"request_id"` + + // Root is the root node of the file tree. + Root *FileTreeNode `json:"root"` + + // Error contains error message if the request failed. + Error string `json:"error,omitempty"` +} + +// FileContentRequest represents a request for file content. +// +// HTTP endpoint: GET /api/v1/workspace/file/content +type FileContentRequest struct { + // Path is the file path (relative to workspace root). + Path string `json:"path"` +} + +// FileContentResponse represents a response with file content. +type FileContentResponse struct { + // RequestID identifies the request. + RequestID string `json:"request_id"` + + // Path is the file path. + Path string `json:"path"` + + // Content is the file content. + Content string `json:"content"` + + // Size is the file size in bytes. + Size int64 `json:"size"` + + // Error contains error message if the request failed. + Error string `json:"error,omitempty"` +} + diff --git a/apps/backend/internal/agentctl/types/streams/git.go b/apps/backend/internal/agentctl/types/streams/git.go new file mode 100644 index 0000000000..f9e43ad125 --- /dev/null +++ b/apps/backend/internal/agentctl/types/streams/git.go @@ -0,0 +1,64 @@ +package streams + +import "time" + +// GitStatusUpdate is the message type streamed via the git status stream. +// Represents the current git state of the workspace. +// +// Stream endpoint: ws://.../api/v1/workspace/git-status/stream +type GitStatusUpdate struct { + // Timestamp is when this status was captured. + Timestamp time.Time `json:"timestamp"` + + // Modified contains paths of modified files. + Modified []string `json:"modified"` + + // Added contains paths of added/staged files. + Added []string `json:"added"` + + // Deleted contains paths of deleted files. + Deleted []string `json:"deleted"` + + // Untracked contains paths of untracked files. + Untracked []string `json:"untracked"` + + // Renamed contains paths of renamed files. + Renamed []string `json:"renamed"` + + // Ahead is the number of commits ahead of the remote branch. + Ahead int `json:"ahead"` + + // Behind is the number of commits behind the remote branch. + Behind int `json:"behind"` + + // Branch is the current local branch name. + Branch string `json:"branch"` + + // RemoteBranch is the tracked remote branch (e.g., "origin/main"). + RemoteBranch string `json:"remote_branch,omitempty"` + + // Files contains detailed information about each changed file. + Files map[string]FileInfo `json:"files,omitempty"` +} + +// FileInfo represents detailed information about a file's git status. +type FileInfo struct { + // Path is the file path relative to workspace root. + Path string `json:"path"` + + // Status indicates the file status: "modified", "added", "deleted", "untracked", "renamed". + Status string `json:"status"` + + // Additions is the number of added lines. + Additions int `json:"additions,omitempty"` + + // Deletions is the number of deleted lines. + Deletions int `json:"deletions,omitempty"` + + // OldPath is the original path for renamed files. + OldPath string `json:"old_path,omitempty"` + + // Diff contains the unified diff content for this file. + Diff string `json:"diff,omitempty"` +} + diff --git a/apps/backend/internal/agentctl/types/streams/permission.go b/apps/backend/internal/agentctl/types/streams/permission.go new file mode 100644 index 0000000000..3e0d5c0e33 --- /dev/null +++ b/apps/backend/internal/agentctl/types/streams/permission.go @@ -0,0 +1,103 @@ +package streams + +import "time" + +// Permission action types categorize the kind of action requiring approval. +const ( + // ActionTypeCommand indicates shell command execution. + ActionTypeCommand = "command" + + // ActionTypeFileWrite indicates file modification or creation. + ActionTypeFileWrite = "file_write" + + // ActionTypeFileRead indicates file read (for sensitive files). + ActionTypeFileRead = "file_read" + + // ActionTypeNetwork indicates network access. + ActionTypeNetwork = "network" + + // ActionTypeMCPTool indicates MCP tool invocation. + ActionTypeMCPTool = "mcp_tool" + + // ActionTypeOther indicates other/unknown action type. + ActionTypeOther = "other" +) + +// PermissionNotification is the message type streamed via the permissions stream. +// Received when the agent requests permission for an action. +// +// Stream endpoint: ws://.../api/v1/acp/permissions/stream +type PermissionNotification struct { + // PendingID uniquely identifies this pending permission request. + PendingID string `json:"pending_id"` + + // SessionID is the session making the request. + SessionID string `json:"session_id"` + + // ToolCallID is the tool call that triggered this permission request. + ToolCallID string `json:"tool_call_id"` + + // Title is a human-readable description of the action. + Title string `json:"title"` + + // Options contains the available permission choices. + Options []PermissionOption `json:"options"` + + // ActionType categorizes the action requiring approval. + // Use ActionType* constants: "command", "file_write", "file_read", "network", "mcp_tool", "other". + ActionType string `json:"action_type,omitempty"` + + // ActionDetails contains structured details about the action. + // For commands: {"command": ["ls", "-la"], "cwd": "/path"} + // For files: {"path": "/file.go", "diff": "..."} + // For MCP tools: {"server": "...", "tool": "...", "arguments": {...}} + ActionDetails map[string]interface{} `json:"action_details,omitempty"` + + // CreatedAt is when the permission request was created. + CreatedAt time.Time `json:"created_at"` +} + +// PermissionOption represents a permission choice presented to the user. +type PermissionOption struct { + // OptionID uniquely identifies this option. + OptionID string `json:"option_id"` + + // Name is a human-readable name for the option. + Name string `json:"name"` + + // Kind indicates the type of permission: "allow_once", "allow_always", + // "reject_once", "reject_always". + Kind string `json:"kind"` + + // Metadata contains protocol-specific option data. + // For Codex: {"for_session": true} for session-wide approvals. + Metadata map[string]interface{} `json:"metadata,omitempty"` +} + +// PermissionRespondRequest is sent to respond to a permission request. +// +// HTTP endpoint: POST /api/v1/acp/permissions/respond +type PermissionRespondRequest struct { + // PendingID is the ID of the permission request to respond to. + PendingID string `json:"pending_id"` + + // OptionID is the selected option ID. + OptionID string `json:"option_id,omitempty"` + + // Cancelled indicates if the request was cancelled. + Cancelled bool `json:"cancelled,omitempty"` + + // ResponseMetadata contains protocol-specific response data. + // For Codex: {"accept_settings": {"for_session": true}}. + ResponseMetadata map[string]interface{} `json:"response_metadata,omitempty"` +} + +// PermissionRespondResponse is the response from the permission respond endpoint. +type PermissionRespondResponse struct { + // Success indicates if the response was accepted. + Success bool `json:"success"` + + // Error contains error message if Success is false. + Error string `json:"error,omitempty"` +} + diff --git a/apps/backend/internal/agentctl/types/streams/shell.go b/apps/backend/internal/agentctl/types/streams/shell.go new file mode 100644 index 0000000000..3364265b7f --- /dev/null +++ b/apps/backend/internal/agentctl/types/streams/shell.go @@ -0,0 +1,64 @@ +package streams + +// Shell message type constants. +const ( + // ShellMsgTypeInput indicates user input to the shell. + ShellMsgTypeInput = "input" + + // ShellMsgTypeOutput indicates output from the shell. + ShellMsgTypeOutput = "output" + + // ShellMsgTypePing is a ping message for keepalive. + ShellMsgTypePing = "ping" + + // ShellMsgTypePong is a pong response to ping. + ShellMsgTypePong = "pong" + + // ShellMsgTypeExit indicates the shell has exited. + ShellMsgTypeExit = "exit" +) + +// ShellMessage is the message type for bidirectional shell stream. +// Used for interactive shell I/O over WebSocket. +// +// Stream endpoint: ws://.../api/v1/shell/stream +type ShellMessage struct { + // Type indicates the message type. Use ShellMsgType* constants: + // "input", "output", "ping", "pong", "exit". + Type string `json:"type"` + + // Data contains the shell input or output data. + Data string `json:"data,omitempty"` + + // Code is the exit code (for "exit" type only). + Code int `json:"code,omitempty"` +} + +// ShellStatusResponse is the response from the shell status endpoint. +// +// HTTP endpoint: GET /api/v1/shell/status +type ShellStatusResponse struct { + // Running indicates if the shell is currently running. + Running bool `json:"running"` + + // Pid is the process ID of the shell. + Pid int `json:"pid"` + + // Shell is the shell type (e.g., "/bin/bash"). + Shell string `json:"shell"` + + // Cwd is the current working directory. + Cwd string `json:"cwd"` + + // StartedAt is when the shell was started (ISO 8601 format). + StartedAt string `json:"started_at"` +} + +// ShellBufferResponse is the response from the shell buffer endpoint. +// +// HTTP endpoint: GET /api/v1/shell/buffer +type ShellBufferResponse struct { + // Data contains the buffered shell output. + Data string `json:"data"` +} + diff --git a/apps/backend/internal/agentctl/types/types.go b/apps/backend/internal/agentctl/types/types.go index 0ad05bc02e..c958608e4e 100644 --- a/apps/backend/internal/agentctl/types/types.go +++ b/apps/backend/internal/agentctl/types/types.go @@ -1,23 +1,85 @@ // Package types provides shared types for agentctl packages. // This breaks import cycles between adapter and acp packages. +// +// For stream protocol message types, see the streams subpackage. package types import ( "context" - "time" + + "github.com/kandev/kandev/internal/agentctl/types/streams" ) -// Permission action types - categorize the kind of action requiring approval +// Re-export stream types for convenience. +// New code may import from the streams package directly. +type ( + // Agent event stream types + AgentEvent = streams.AgentEvent + PlanEntry = streams.PlanEntry + + // Permission stream types + PermissionNotification = streams.PermissionNotification + PermissionOption = streams.PermissionOption + PermissionRespondRequest = streams.PermissionRespondRequest + PermissionRespondResponse = streams.PermissionRespondResponse + + // Git stream types + GitStatusUpdate = streams.GitStatusUpdate + FileInfo = streams.FileInfo + + // File stream types + FileChangeNotification = streams.FileChangeNotification + FileListUpdate = streams.FileListUpdate + FileEntry = streams.FileEntry + FileTreeNode = streams.FileTreeNode + FileTreeRequest = streams.FileTreeRequest + FileTreeResponse = streams.FileTreeResponse + FileContentRequest = streams.FileContentRequest + FileContentResponse = streams.FileContentResponse + + // Shell stream types + ShellMessage = streams.ShellMessage + ShellStatusResponse = streams.ShellStatusResponse + ShellBufferResponse = streams.ShellBufferResponse +) + +// Re-export stream constants for convenience. const ( - ActionTypeCommand = "command" // Shell command execution - ActionTypeFileWrite = "file_write" // File modification/creation - ActionTypeFileRead = "file_read" // File read (for sensitive files) - ActionTypeNetwork = "network" // Network access - ActionTypeMCPTool = "mcp_tool" // MCP tool invocation - ActionTypeOther = "other" // Other/unknown action type + // Agent event types (preferred) + EventTypeMessageChunk = streams.EventTypeMessageChunk + EventTypeReasoning = streams.EventTypeReasoning + EventTypeToolCall = streams.EventTypeToolCall + EventTypeToolUpdate = streams.EventTypeToolUpdate + EventTypePlan = streams.EventTypePlan + EventTypeComplete = streams.EventTypeComplete + EventTypeError = streams.EventTypeError + + // Permission action types + ActionTypeCommand = streams.ActionTypeCommand + ActionTypeFileWrite = streams.ActionTypeFileWrite + ActionTypeFileRead = streams.ActionTypeFileRead + ActionTypeNetwork = streams.ActionTypeNetwork + ActionTypeMCPTool = streams.ActionTypeMCPTool + ActionTypeOther = streams.ActionTypeOther + + // File operation types + FileOpCreate = streams.FileOpCreate + FileOpWrite = streams.FileOpWrite + FileOpRemove = streams.FileOpRemove + FileOpRename = streams.FileOpRename + FileOpChmod = streams.FileOpChmod + FileOpRefresh = streams.FileOpRefresh + + // Shell message types + ShellMsgTypeInput = streams.ShellMsgTypeInput + ShellMsgTypeOutput = streams.ShellMsgTypeOutput + ShellMsgTypePing = streams.ShellMsgTypePing + ShellMsgTypePong = streams.ShellMsgTypePong + ShellMsgTypeExit = streams.ShellMsgTypeExit ) -// PermissionRequest represents a permission request from the agent +// PermissionRequest represents a permission request from the agent. +// This is used internally by adapters and is not sent over streams directly. type PermissionRequest struct { SessionID string `json:"session_id"` ToolCallID string `json:"tool_call_id"` @@ -35,18 +97,8 @@ type PermissionRequest struct { ActionDetails map[string]interface{} `json:"action_details,omitempty"` } -// 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 - - // Metadata contains protocol-specific option data. - // For Codex: {"for_session": true} for session-wide approvals - Metadata map[string]interface{} `json:"metadata,omitempty"` -} - -// PermissionResponse is the user's response to a permission request +// PermissionResponse is the user's response to a permission request. +// This is used internally by adapters. type PermissionResponse struct { OptionID string `json:"option_id,omitempty"` Cancelled bool `json:"cancelled,omitempty"` @@ -56,96 +108,17 @@ type PermissionResponse struct { ResponseMetadata map[string]interface{} `json:"response_metadata,omitempty"` } -// PermissionHandler is called when the agent requests permission for an action +// PermissionHandler is called when the agent requests permission for an action. type PermissionHandler func(ctx context.Context, req *PermissionRequest) (*PermissionResponse, error) -// GitStatusUpdate represents a git status update -type GitStatusUpdate struct { - Timestamp time.Time `json:"timestamp"` - Modified []string `json:"modified"` - Added []string `json:"added"` - Deleted []string `json:"deleted"` - Untracked []string `json:"untracked"` - Renamed []string `json:"renamed"` - Ahead int `json:"ahead"` - Behind int `json:"behind"` - Branch string `json:"branch"` - RemoteBranch string `json:"remote_branch,omitempty"` - Files map[string]FileInfo `json:"files,omitempty"` -} +// Subscriber types for internal use. +type ( + // GitStatusSubscriber is a channel that receives git status updates. + GitStatusSubscriber chan GitStatusUpdate -// FileInfo represents information about a file -type FileInfo struct { - Path string `json:"path"` - Status string `json:"status"` // modified, added, deleted, untracked, renamed - Additions int `json:"additions,omitempty"` - Deletions int `json:"deletions,omitempty"` - OldPath string `json:"old_path,omitempty"` // For renamed files - Diff string `json:"diff,omitempty"` -} - -// FileListUpdate represents a file listing update -type FileListUpdate struct { - Timestamp time.Time `json:"timestamp"` - Files []FileEntry `json:"files"` -} - -// FileEntry represents a file in the workspace -type FileEntry struct { - Path string `json:"path"` - IsDir bool `json:"is_dir"` - Size int64 `json:"size,omitempty"` -} - -// FileTreeNode represents a node in the file tree -type FileTreeNode struct { - Name string `json:"name"` - Path string `json:"path"` - IsDir bool `json:"is_dir"` - Size int64 `json:"size,omitempty"` - Children []*FileTreeNode `json:"children,omitempty"` -} - -// FileTreeRequest represents a request for file tree -type FileTreeRequest struct { - Path string `json:"path"` // Path to get tree for (relative to workspace root) - Depth int `json:"depth"` // Depth to traverse (0 = unlimited, 1 = immediate children only) -} - -// FileTreeResponse represents a response with file tree -type FileTreeResponse struct { - RequestID string `json:"request_id"` - Root *FileTreeNode `json:"root"` - Error string `json:"error,omitempty"` -} - -// FileContentRequest represents a request for file content -type FileContentRequest struct { - Path string `json:"path"` // Path to file (relative to workspace root) -} - -// FileContentResponse represents a response with file content -type FileContentResponse struct { - RequestID string `json:"request_id"` - Path string `json:"path"` - Content string `json:"content"` - Size int64 `json:"size"` - Error string `json:"error,omitempty"` -} - -// FileChangeNotification represents a filesystem change notification -type FileChangeNotification struct { - Timestamp time.Time `json:"timestamp"` - Path string `json:"path"` - Operation string `json:"operation"` // create, write, remove, rename, chmod, refresh -} - -// GitStatusSubscriber is a channel that receives git status updates -type GitStatusSubscriber chan GitStatusUpdate - -// FilesSubscriber is a channel that receives file listing updates -type FilesSubscriber chan FileListUpdate - -// FileChangeSubscriber is a channel that receives file change notifications -type FileChangeSubscriber chan FileChangeNotification + // FilesSubscriber is a channel that receives file listing updates. + FilesSubscriber chan FileListUpdate + // FileChangeSubscriber is a channel that receives file change notifications. + FileChangeSubscriber chan FileChangeNotification +) diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index 7a2999781e..99e933de04 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -95,9 +95,10 @@ const ( // Event types for agent prompts const ( - PromptComplete = "prompt.complete" // Agent finished responding to a prompt - ToolCallStarted = "tool_call.started" // Agent started a tool call - ToolCallComplete = "tool_call.complete" // Agent finished a tool call + PromptComplete = "prompt.complete" // Agent finished responding to a prompt + ToolCallStarted = "tool_call.started" // Agent started a tool call + ToolCallComplete = "tool_call.complete" // Agent finished a tool call + PermissionRequestReceived = "permission_request.received" // Agent requested permission ) // Event types for workspace/git status diff --git a/apps/backend/internal/gateway/websocket/client.go b/apps/backend/internal/gateway/websocket/client.go index c23cecdacf..9ad8dee6fb 100644 --- a/apps/backend/internal/gateway/websocket/client.go +++ b/apps/backend/internal/gateway/websocket/client.go @@ -169,11 +169,8 @@ func (c *Client) handleSubscribe(msg *ws.Message) { }) c.sendMessage(resp) - // Send historical logs if available + // Send historical logs if available (now includes pending permission request messages) c.sendHistoricalLogs(req.TaskID) - - // Send pending permissions if available - c.sendPendingPermissions(req.TaskID) } type UserSubscribeRequest struct { @@ -275,30 +272,9 @@ func (c *Client) sendHistoricalLogs(taskID string) { } } -// sendPendingPermissions sends any pending permission requests to the client -func (c *Client) sendPendingPermissions(taskID string) { - ctx := context.Background() - permissions, err := c.hub.GetPendingPermissions(ctx, taskID) - if err != nil { - c.logger.Error("Failed to get pending permissions", - zap.String("task_id", taskID), - zap.Error(err)) - return - } - - if len(permissions) == 0 { - return - } - - c.logger.Debug("Sending pending permissions", - zap.String("task_id", taskID), - zap.Int("count", len(permissions))) - - // Send each pending permission as a notification - for _, permission := range permissions { - c.sendMessage(permission) - } -} +// sendPendingPermissions is deprecated. Pending permissions are now stored as messages +// and sent via sendHistoricalLogs. This function is kept for reference but is unused. +// func (c *Client) sendPendingPermissions(taskID string) {} // handleUnsubscribe handles task.unsubscribe action func (c *Client) handleUnsubscribe(msg *ws.Message) { diff --git a/apps/backend/internal/gateway/websocket/hub.go b/apps/backend/internal/gateway/websocket/hub.go index 4daad9e98a..99d1f92926 100644 --- a/apps/backend/internal/gateway/websocket/hub.go +++ b/apps/backend/internal/gateway/websocket/hub.go @@ -14,8 +14,9 @@ import ( // HistoricalLogsProvider is a function that retrieves historical logs for a task type HistoricalLogsProvider func(ctx context.Context, taskID string) ([]*ws.Message, error) -// PendingPermissionsProvider is a function that retrieves pending permissions for a task -type PendingPermissionsProvider func(ctx context.Context, taskID string) ([]*ws.Message, error) +// PendingPermissionsProvider is deprecated. Pending permissions are now stored as messages +// and retrieved via the HistoricalLogsProvider. +// type PendingPermissionsProvider func(ctx context.Context, taskID string) ([]*ws.Message, error) // Hub manages all WebSocket client connections type Hub struct { @@ -41,8 +42,8 @@ type Hub struct { // Optional provider for historical logs on subscription historicalLogsProvider HistoricalLogsProvider - // Optional provider for pending permissions on subscription - pendingPermissionsProvider PendingPermissionsProvider + // NOTE: pendingPermissionsProvider is deprecated. Pending permissions are now stored + // as messages and retrieved via the historicalLogsProvider. mu sync.RWMutex logger *logger.Logger @@ -392,15 +393,10 @@ func (h *Hub) GetHistoricalLogs(ctx context.Context, taskID string) ([]*ws.Messa return h.historicalLogsProvider(ctx, taskID) } -// SetPendingPermissionsProvider sets the provider for pending permissions on subscription -func (h *Hub) SetPendingPermissionsProvider(provider PendingPermissionsProvider) { - h.pendingPermissionsProvider = provider -} +// SetPendingPermissionsProvider is deprecated. Pending permissions are now stored as messages +// and retrieved via the HistoricalLogsProvider. This method is a no-op for backwards compatibility. +// func (h *Hub) SetPendingPermissionsProvider(provider PendingPermissionsProvider) {} -// GetPendingPermissions retrieves pending permissions for a task if a provider is set -func (h *Hub) GetPendingPermissions(ctx context.Context, taskID string) ([]*ws.Message, error) { - if h.pendingPermissionsProvider == nil { - return nil, nil - } - return h.pendingPermissionsProvider(ctx, taskID) -} +// GetPendingPermissions is deprecated. Pending permissions are now stored as messages +// and retrieved via the HistoricalLogsProvider. This method always returns nil. +// func (h *Hub) GetPendingPermissions(ctx context.Context, taskID string) ([]*ws.Message, error) {} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 739d228465..b4394eae70 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -57,6 +57,8 @@ type MessageCreator interface { CreateToolCallMessage(ctx context.Context, taskID, toolCallID, title, status, agentSessionID string, args map[string]interface{}) error UpdateToolCallMessage(ctx context.Context, taskID, toolCallID, status, result, agentSessionID string) error CreateSessionMessage(ctx context.Context, taskID, content, agentSessionID, messageType string, metadata map[string]interface{}, requestsInput bool) error + CreatePermissionRequestMessage(ctx context.Context, taskID, sessionID, pendingID, toolCallID, title string, options []map[string]interface{}, actionType string, actionDetails map[string]interface{}) (string, error) + UpdatePermissionMessage(ctx context.Context, sessionID, pendingID, status string) error } // Service is the main orchestrator service @@ -152,16 +154,17 @@ func NewService( // Create the watcher with event handlers that wire everything together handlers := watcher.EventHandlers{ - OnTaskStateChanged: s.handleTaskStateChanged, - OnAgentReady: s.handleAgentReady, - OnAgentCompleted: s.handleAgentCompleted, - OnAgentFailed: s.handleAgentFailed, - OnACPMessage: s.handleACPMessage, - OnACPSessionCreated: s.handleACPSessionCreated, - OnPromptComplete: s.handlePromptComplete, - OnToolCallStarted: s.handleToolCallStarted, - OnToolCallComplete: s.handleToolCallComplete, - OnGitStatusUpdated: s.handleGitStatusUpdated, + OnTaskStateChanged: s.handleTaskStateChanged, + OnAgentReady: s.handleAgentReady, + OnAgentCompleted: s.handleAgentCompleted, + OnAgentFailed: s.handleAgentFailed, + OnACPMessage: s.handleACPMessage, + OnACPSessionCreated: s.handleACPSessionCreated, + OnPromptComplete: s.handlePromptComplete, + OnToolCallStarted: s.handleToolCallStarted, + OnToolCallComplete: s.handleToolCallComplete, + OnPermissionRequest: s.handlePermissionRequest, + OnGitStatusUpdated: s.handleGitStatusUpdated, } s.watcher = watcher.NewWatcher(eventBus, handlers, log) @@ -554,7 +557,31 @@ func (s *Service) RespondToPermission(ctx context.Context, sessionID, pendingID, zap.String("pending_id", pendingID), zap.String("option_id", optionID), zap.Bool("cancelled", cancelled)) - return s.executor.RespondToPermission(ctx, sessionID, pendingID, optionID, cancelled) + + // Respond to the permission via agentctl + if err := s.executor.RespondToPermission(ctx, sessionID, pendingID, optionID, cancelled); err != nil { + return err + } + + // Determine status based on response + status := "approved" + if cancelled { + status = "rejected" + } + + // Update the permission message with the new status + if s.messageCreator != nil { + if err := s.messageCreator.UpdatePermissionMessage(ctx, sessionID, pendingID, status); err != nil { + s.logger.Warn("failed to update permission message status", + zap.String("session_id", sessionID), + zap.String("pending_id", pendingID), + zap.String("status", status), + zap.Error(err)) + // Don't fail the whole operation if message update fails + } + } + + return nil } // CompleteTask explicitly completes a task and stops its agent @@ -1180,3 +1207,42 @@ func (s *Service) handleToolCallComplete(ctx context.Context, data watcher.ToolC s.updateTaskSessionState(ctx, data.TaskID, data.TaskSessionID, models.TaskSessionStateRunning, "", false) } } + +// handlePermissionRequest handles permission request events and saves as message +func (s *Service) handlePermissionRequest(ctx context.Context, data watcher.PermissionRequestData) { + s.logger.Info("handling permission request", + zap.String("task_id", data.TaskID), + zap.String("pending_id", data.PendingID), + zap.String("title", data.Title)) + + if data.TaskSessionID == "" { + s.logger.Warn("missing session_id for permission_request", + zap.String("task_id", data.TaskID), + zap.String("pending_id", data.PendingID)) + return + } + + if s.messageCreator != nil { + _, err := s.messageCreator.CreatePermissionRequestMessage( + ctx, + data.TaskID, + data.TaskSessionID, + data.PendingID, + data.ToolCallID, + data.Title, + data.Options, + data.ActionType, + data.ActionDetails, + ) + if err != nil { + s.logger.Error("failed to create permission request message", + zap.String("task_id", data.TaskID), + zap.String("pending_id", data.PendingID), + zap.Error(err)) + } else { + s.logger.Info("created permission request message", + zap.String("task_id", data.TaskID), + zap.String("pending_id", data.PendingID)) + } + } +} diff --git a/apps/backend/internal/orchestrator/watcher/watcher.go b/apps/backend/internal/orchestrator/watcher/watcher.go index af1c6257c3..99438ec523 100644 --- a/apps/backend/internal/orchestrator/watcher/watcher.go +++ b/apps/backend/internal/orchestrator/watcher/watcher.go @@ -69,6 +69,19 @@ type ToolCallCompleteData struct { Result string `json:"result,omitempty"` } +// PermissionRequestData contains data from permission_request events +type PermissionRequestData struct { + TaskID string `json:"task_id"` + TaskSessionID string `json:"session_id"` + AgentID string `json:"agent_id"` + PendingID string `json:"pending_id"` + ToolCallID string `json:"tool_call_id"` + Title string `json:"title"` + Options []map[string]interface{} `json:"options"` + ActionType string `json:"action_type"` + ActionDetails map[string]interface{} `json:"action_details"` +} + // GitStatusData contains data from git status events type GitStatusData struct { TaskID string `json:"task_id"` @@ -113,6 +126,9 @@ type EventHandlers struct { OnToolCallStarted func(ctx context.Context, data ToolCallData) OnToolCallComplete func(ctx context.Context, data ToolCallCompleteData) + // Permission request events + OnPermissionRequest func(ctx context.Context, data PermissionRequestData) + // Git status events OnGitStatusUpdated func(ctx context.Context, data GitStatusData) } @@ -187,6 +203,12 @@ func (w *Watcher) Start(ctx context.Context) error { return err } + // Subscribe to permission request events + if err := w.subscribeToPermissionRequestEvents(); err != nil { + w.unsubscribeAll() + return err + } + // Subscribe to git status events if err := w.subscribeToGitStatusEvents(); err != nil { w.unsubscribeAll() @@ -547,6 +569,46 @@ func (w *Watcher) createToolCallCompleteHandler() bus.EventHandler { } } +// subscribeToPermissionRequestEvents subscribes to permission request events +func (w *Watcher) subscribeToPermissionRequestEvents() error { + if w.handlers.OnPermissionRequest == nil { + return nil + } + + subject := events.PermissionRequestReceived + ".*" + sub, err := w.eventBus.Subscribe(subject, w.createPermissionRequestHandler()) + if err != nil { + w.logger.Error("Failed to subscribe to permission request events", + zap.String("subject", subject), + zap.Error(err)) + return err + } + w.subscriptions = append(w.subscriptions, sub) + return nil +} + +// createPermissionRequestHandler creates a bus.EventHandler for permission request events +func (w *Watcher) createPermissionRequestHandler() bus.EventHandler { + return func(ctx context.Context, event *bus.Event) error { + var data PermissionRequestData + if err := w.parseEventData(event.Data, &data); err != nil { + w.logger.Error("Failed to parse permission request event data", + zap.String("event_type", event.Type), + zap.String("event_id", event.ID), + zap.Error(err)) + return nil + } + + w.logger.Debug("Handling permission request event", + zap.String("task_id", data.TaskID), + zap.String("pending_id", data.PendingID), + zap.String("title", data.Title)) + + w.handlers.OnPermissionRequest(ctx, data) + return nil + } +} + // subscribeToGitStatusEvents subscribes to git status events func (w *Watcher) subscribeToGitStatusEvents() error { if w.handlers.OnGitStatusUpdated == nil { diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 5aae350170..d4f8cb0f60 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -96,6 +96,8 @@ const ( MessageTypeError MessageType = "error" // MessageTypeStatus is for status changes: started, completed, failed MessageTypeStatus MessageType = "status" + // MessageTypePermissionRequest is for agent permission requests + MessageTypePermissionRequest MessageType = "permission_request" ) // Message represents a message in a task session diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index 0dfaa4c8e5..5faba2209a 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -62,6 +62,7 @@ type Repository interface { CreateMessage(ctx context.Context, message *models.Message) error GetMessage(ctx context.Context, id string) (*models.Message, error) GetMessageByToolCallID(ctx context.Context, sessionID, toolCallID string) (*models.Message, error) + GetMessageByPendingID(ctx context.Context, sessionID, pendingID string) (*models.Message, error) UpdateMessage(ctx context.Context, message *models.Message) error ListMessages(ctx context.Context, sessionID string) ([]*models.Message, error) ListMessagesPaginated(ctx context.Context, sessionID string, opts ListMessagesOptions) ([]*models.Message, bool, error) diff --git a/apps/backend/internal/task/repository/sqlite.go b/apps/backend/internal/task/repository/sqlite.go index 8309b073fe..240d767318 100644 --- a/apps/backend/internal/task/repository/sqlite.go +++ b/apps/backend/internal/task/repository/sqlite.go @@ -1546,7 +1546,7 @@ func (r *SQLiteRepository) DeleteMessage(ctx context.Context, id string) error { return nil } -// GetMessageByToolCallID retrieves a message by session ID and tool_call_id in metadata +// GetMessageByToolCallID retrieves a tool_call message by session ID and tool_call_id in metadata func (r *SQLiteRepository) GetMessageByToolCallID(ctx context.Context, sessionID, toolCallID string) (*models.Message, error) { message := &models.Message{} var requestsInput int @@ -1554,7 +1554,7 @@ func (r *SQLiteRepository) GetMessageByToolCallID(ctx context.Context, sessionID var metadataJSON string err := r.db.QueryRowContext(ctx, ` SELECT id, task_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at - FROM task_session_messages WHERE task_session_id = ? AND json_extract(metadata, '$.tool_call_id') = ? + FROM task_session_messages WHERE task_session_id = ? AND type = 'tool_call' AND json_extract(metadata, '$.tool_call_id') = ? `, sessionID, toolCallID).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.AuthorType, &message.AuthorID, &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt) if err != nil { @@ -1568,6 +1568,28 @@ func (r *SQLiteRepository) GetMessageByToolCallID(ctx context.Context, sessionID return message, nil } +// GetMessageByPendingID retrieves a message by session ID and pending_id in metadata +func (r *SQLiteRepository) GetMessageByPendingID(ctx context.Context, sessionID, pendingID string) (*models.Message, error) { + message := &models.Message{} + var requestsInput int + var messageType string + var metadataJSON string + err := r.db.QueryRowContext(ctx, ` + SELECT id, task_session_id, task_id, author_type, author_id, content, requests_input, type, metadata, created_at + FROM task_session_messages WHERE task_session_id = ? AND json_extract(metadata, '$.pending_id') = ? + `, sessionID, pendingID).Scan(&message.ID, &message.TaskSessionID, &message.TaskID, &message.AuthorType, &message.AuthorID, + &message.Content, &requestsInput, &messageType, &metadataJSON, &message.CreatedAt) + if err != nil { + return nil, err + } + message.RequestsInput = requestsInput == 1 + message.Type = models.MessageType(messageType) + if metadataJSON != "" { + _ = json.Unmarshal([]byte(metadataJSON), &message.Metadata) + } + return message, nil +} + // UpdateMessage updates an existing message func (r *SQLiteRepository) UpdateMessage(ctx context.Context, message *models.Message) error { metadataJSON, err := json.Marshal(message.Metadata) diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index 91d0690617..19523fe771 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -1611,6 +1611,69 @@ func (s *Service) UpdateToolCallMessage(ctx context.Context, sessionID, toolCall return nil } +// UpdatePermissionMessage updates a permission request message's status. +// It includes retry logic to handle race conditions. +func (s *Service) UpdatePermissionMessage(ctx context.Context, sessionID, pendingID, status string) error { + const maxRetries = 5 + const retryDelay = 100 * time.Millisecond + + var message *models.Message + var err error + + // Retry loop to handle race condition + for attempt := 0; attempt < maxRetries; attempt++ { + message, err = s.repo.GetMessageByPendingID(ctx, sessionID, pendingID) + if err == nil { + break + } + + if ctx.Err() != nil { + return ctx.Err() + } + + if attempt < maxRetries-1 { + s.logger.Debug("permission message not found, retrying", + zap.String("session_id", sessionID), + zap.String("pending_id", pendingID), + zap.Int("attempt", attempt+1), + zap.Int("max_retries", maxRetries)) + time.Sleep(retryDelay) + } + } + + if err != nil { + s.logger.Warn("permission message not found for update after retries", + zap.String("session_id", sessionID), + zap.String("pending_id", pendingID), + zap.Int("retries", maxRetries), + zap.Error(err)) + return err + } + + if message.Metadata == nil { + message.Metadata = make(map[string]interface{}) + } + message.Metadata["status"] = status + + if err := s.repo.UpdateMessage(ctx, message); err != nil { + s.logger.Error("failed to update permission message", + zap.String("message_id", message.ID), + zap.String("pending_id", pendingID), + zap.Error(err)) + return err + } + + // Publish message.updated event + s.publishMessageEvent(ctx, events.MessageUpdated, message) + + s.logger.Info("permission message updated", + zap.String("message_id", message.ID), + zap.String("pending_id", pendingID), + zap.String("status", status)) + + return nil +} + // publishMessageEvent publishes message events to the event bus func (s *Service) publishMessageEvent(ctx context.Context, eventType string, message *models.Message) { if s.eventBus == nil { diff --git a/apps/web/app/task/[id]/page-client.tsx b/apps/web/app/task/[id]/page-client.tsx index 81d75012b9..be0245c1b8 100644 --- a/apps/web/app/task/[id]/page-client.tsx +++ b/apps/web/app/task/[id]/page-client.tsx @@ -5,7 +5,6 @@ import '@git-diff-view/react/styles/diff-view.css'; import { TooltipProvider } from '@kandev/ui/tooltip'; import { TaskTopBar } from '@/components/task/task-top-bar'; import { TaskLayout } from '@/components/task/task-layout'; -import { CommandApprovalDialog } from '@/components/task/command-approval-dialog'; import { DebugOverlay } from '@/components/debug-overlay'; import type { Repository, Task } from '@/lib/types/http'; import { DEBUG_UI } from '@/lib/config'; @@ -174,9 +173,6 @@ export default function TaskPage({ workspaceId={task?.workspace_id ?? null} boardId={task?.board_id ?? null} /> - - {/* Fallback dialog for permissions without matching tool call (e.g., workspace indexing) */} - {task?.id && } ); diff --git a/apps/web/components/task/chat/message-renderer.tsx b/apps/web/components/task/chat/message-renderer.tsx index e3c17084a6..f5aa9e76f5 100644 --- a/apps/web/components/task/chat/message-renderer.tsx +++ b/apps/web/components/task/chat/message-renderer.tsx @@ -3,6 +3,7 @@ import type { ReactElement } from 'react'; import type { Message } from '@/lib/types/http'; import { ChatMessage } from '@/components/task/chat/messages/chat-message'; +import { PermissionRequestMessage } from '@/components/task/chat/messages/permission-request-message'; import { StatusMessage } from '@/components/task/chat/messages/status-message'; import { ToolCallMessage } from '@/components/task/chat/messages/tool-call-message'; import { ThinkingMessage } from '@/components/task/chat/messages/thinking-message'; @@ -11,6 +12,7 @@ import { TodoMessage } from '@/components/task/chat/messages/todo-message'; type AdapterContext = { isTaskDescription: boolean; taskId?: string; + permissionsByToolCallId?: Map; }; type MessageAdapter = { @@ -29,12 +31,21 @@ const adapters: MessageAdapter[] = [ }, { matches: (comment) => comment.type === 'tool_call', - render: (comment, ctx) => , + render: (comment, ctx) => { + const toolCallId = (comment.metadata as { tool_call_id?: string } | undefined)?.tool_call_id; + const permissionMessage = toolCallId ? ctx.permissionsByToolCallId?.get(toolCallId) : undefined; + return ; + }, }, { matches: (comment) => comment.type === 'error' || comment.type === 'status' || comment.type === 'progress', render: (comment) => , }, + { + // Standalone permission requests (no matching tool call) + matches: (comment) => comment.type === 'permission_request', + render: (comment) => , + }, { matches: () => true, render: (comment, ctx) => { @@ -73,10 +84,11 @@ type MessageRendererProps = { comment: Message; isTaskDescription: boolean; taskId?: string; + permissionsByToolCallId?: Map; }; -export function MessageRenderer({ comment, isTaskDescription, taskId }: MessageRendererProps) { - const ctx = { isTaskDescription, taskId }; +export function MessageRenderer({ comment, isTaskDescription, taskId, permissionsByToolCallId }: MessageRendererProps) { + const ctx = { isTaskDescription, taskId, permissionsByToolCallId }; const adapter = adapters.find((entry) => entry.matches(comment, ctx)) ?? adapters[adapters.length - 1]; return adapter.render(comment, ctx); } diff --git a/apps/web/components/task/chat/messages/permission-request-message.tsx b/apps/web/components/task/chat/messages/permission-request-message.tsx new file mode 100644 index 0000000000..0050d65de8 --- /dev/null +++ b/apps/web/components/task/chat/messages/permission-request-message.tsx @@ -0,0 +1,170 @@ +'use client'; + +import { useCallback, useState } from 'react'; +import { + IconAlertTriangle, + IconCheck, + IconX, + IconLoader2, +} from '@tabler/icons-react'; +import { Button } from '@kandev/ui/button'; +import { cn } from '@/lib/utils'; +import { getWebSocketClient } from '@/lib/ws/connection'; +import type { Message } from '@/lib/types/http'; + +type PermissionOption = { + option_id: string; + name: string; + kind: string; +}; + +type PermissionRequestMetadata = { + pending_id: string; + tool_call_id?: string; + options: PermissionOption[]; + action_type: string; + action_details?: { + command?: string; + path?: string; + cwd?: string; + }; + status?: 'pending' | 'approved' | 'rejected'; +}; + +type PermissionRequestMessageProps = { + comment: Message; +}; + +export function PermissionRequestMessage({ comment }: PermissionRequestMessageProps) { + const [isResponding, setIsResponding] = useState(false); + const metadata = comment.metadata as PermissionRequestMetadata | undefined; + + const isResolved = metadata?.status === 'approved' || metadata?.status === 'rejected'; + const isPending = !isResolved; + + const handleRespond = useCallback( + async (optionId: string, cancelled: boolean = false) => { + if (!metadata) return; + + const client = getWebSocketClient(); + if (!client) { + console.error('WebSocket client not available'); + return; + } + + setIsResponding(true); + try { + await client.request('permission.respond', { + session_id: comment.session_id, + pending_id: metadata.pending_id, + option_id: cancelled ? undefined : optionId, + cancelled, + }); + } catch (error) { + console.error('Failed to respond to permission request:', error); + } finally { + setIsResponding(false); + } + }, + [comment.session_id, metadata] + ); + + const handleApprove = useCallback(() => { + const allowOption = metadata?.options.find( + (opt) => opt.kind === 'allow_once' || opt.kind === 'allow_always' + ); + if (allowOption) { + handleRespond(allowOption.option_id); + } + }, [metadata, handleRespond]); + + const handleReject = useCallback(() => { + const rejectOption = metadata?.options.find( + (opt) => opt.kind === 'reject_once' || opt.kind === 'reject_always' + ); + if (rejectOption) { + handleRespond(rejectOption.option_id); + } else { + handleRespond('', true); + } + }, [metadata, handleRespond]); + + const getStatusBadge = () => { + if (metadata?.status === 'approved') { + return ( + + Approved + + ); + } + if (metadata?.status === 'rejected') { + return ( + + Rejected + + ); + } + return ( + + Pending Approval + + ); + }; + + return ( +
+ {/* Header */} +
+ + + {comment.content || 'Permission Required'} + + {getStatusBadge()} +
+ + {/* Approval buttons - only show when pending */} + {isPending && ( +
+ + Approve this action? + + + +
+ )} +
+ ); +} + diff --git a/apps/web/components/task/chat/messages/tool-call-message.tsx b/apps/web/components/task/chat/messages/tool-call-message.tsx index 968fc013e9..5359721934 100644 --- a/apps/web/components/task/chat/messages/tool-call-message.tsx +++ b/apps/web/components/task/chat/messages/tool-call-message.tsx @@ -16,10 +16,28 @@ import { } from '@tabler/icons-react'; import { Button } from '@kandev/ui/button'; import { cn } from '@/lib/utils'; +import { getWebSocketClient } from '@/lib/ws/connection'; import type { Message } from '@/lib/types/http'; import type { ToolCallMetadata } from '@/components/task/chat/types'; -import { useAppStore } from '@/components/state-provider'; -import { getWebSocketClient } from '@/lib/ws/connection'; + +type PermissionOption = { + option_id: string; + name: string; + kind: string; +}; + +type PermissionRequestMetadata = { + pending_id: string; + tool_call_id: string; + options: PermissionOption[]; + action_type: string; + action_details: { + command?: string; + path?: string; + cwd?: string; + }; + status?: 'pending' | 'approved' | 'rejected'; +}; function getToolIcon(toolName: string | undefined, className: string) { const name = toolName?.toLowerCase() ?? ''; @@ -41,7 +59,15 @@ function getToolIcon(toolName: string | undefined, className: string) { return ; } -function getStatusIcon(status?: string) { +function getStatusIcon(status?: string, permissionStatus?: string) { + // If there's a permission status, show that instead + if (permissionStatus === 'approved') { + return ; + } + if (permissionStatus === 'rejected') { + return ; + } + switch (status) { case 'complete': return ; @@ -56,74 +82,64 @@ function getStatusIcon(status?: string) { type ToolCallMessageProps = { comment: Message; - taskId?: string; + permissionMessage?: Message; }; -export function ToolCallMessage({ comment, taskId }: ToolCallMessageProps) { +export function ToolCallMessage({ comment, permissionMessage }: ToolCallMessageProps) { const [isExpanded, setIsExpanded] = useState(false); const [isResponding, setIsResponding] = useState(false); const metadata = comment.metadata as ToolCallMetadata | undefined; - const pendingPermissions = useAppStore((state) => state.permissions.pending); - const removePendingPermission = useAppStore((state) => state.removePendingPermission); - const activeSessionId = useAppStore((state) => state.tasks.activeSessionId); + const permissionMetadata = permissionMessage?.metadata as PermissionRequestMetadata | undefined; - const toolCallId = metadata?.tool_call_id; const toolName = metadata?.tool_name ?? ''; const title = metadata?.title ?? comment.content ?? 'Tool call'; const status = metadata?.status; const args = metadata?.args; const result = metadata?.result; - // Find pending permission for this tool call (only if we have a valid tool_call_id) - const pendingPermission = pendingPermissions.find( - (p) => - toolCallId && - p.tool_call_id === toolCallId && - (!activeSessionId || p.session_id === activeSessionId) && - (!taskId || p.task_id === taskId) - ); - const isPendingApproval = !!pendingPermission; + // Permission state + const hasPermission = !!permissionMessage; + const permissionStatus = permissionMetadata?.status; + const isPermissionPending = hasPermission && permissionStatus !== 'approved' && permissionStatus !== 'rejected'; const handleRespond = useCallback( async (optionId: string, cancelled: boolean = false) => { - if (!pendingPermission) return; - setIsResponding(true); + if (!permissionMetadata || !permissionMessage) return; const client = getWebSocketClient(); if (!client) { console.error('WebSocket client not available'); - setIsResponding(false); return; } + setIsResponding(true); try { await client.request('permission.respond', { - session_id: pendingPermission.session_id, - pending_id: pendingPermission.pending_id, + session_id: permissionMessage.session_id, + pending_id: permissionMetadata.pending_id, option_id: cancelled ? undefined : optionId, cancelled, }); - removePendingPermission(pendingPermission.pending_id); } catch (error) { console.error('Failed to respond to permission request:', error); } finally { setIsResponding(false); } }, - [pendingPermission, removePendingPermission] + [permissionMessage, permissionMetadata] ); const handleApprove = useCallback(() => { - const allowOption = pendingPermission?.options.find( + const allowOption = permissionMetadata?.options.find( (opt) => opt.kind === 'allow_once' || opt.kind === 'allow_always' ); if (allowOption) { handleRespond(allowOption.option_id); } - }, [pendingPermission, handleRespond]); + }, [permissionMetadata, handleRespond]); const handleReject = useCallback(() => { - const rejectOption = pendingPermission?.options.find( + const rejectOption = permissionMetadata?.options.find( (opt) => opt.kind === 'reject_once' || opt.kind === 'reject_always' ); if (rejectOption) { @@ -131,9 +147,12 @@ export function ToolCallMessage({ comment, taskId }: ToolCallMessageProps) { } else { handleRespond('', true); } - }, [pendingPermission, handleRespond]); + }, [permissionMetadata, handleRespond]); - const toolIcon = getToolIcon(toolName, 'h-4 w-4 text-amber-600 dark:text-amber-400 flex-shrink-0'); + const toolIcon = getToolIcon(toolName, cn( + 'h-4 w-4 flex-shrink-0', + isPermissionPending ? 'text-amber-600 dark:text-amber-400' : 'text-amber-600 dark:text-amber-400' + )); const hasDetails = args && Object.keys(args).length > 0; let filePath: string | undefined; @@ -151,9 +170,7 @@ export function ToolCallMessage({ comment, taskId }: ToolCallMessageProps) { return (
- {/* Inline approval buttons when permission is pending */} - {isPendingApproval && ( -
- - Approve this action? - - - -
- )} - {isExpanded && hasDetails && (
{args && Object.entries(args).map(([key, value]) => { @@ -243,6 +227,39 @@ export function ToolCallMessage({ comment, taskId }: ToolCallMessageProps) { )}
)} + + {/* Inline permission approval UI */} + {isPermissionPending && ( +
+ + Approve this action? + + + +
+ )}
); } diff --git a/apps/web/components/task/command-approval-dialog.tsx b/apps/web/components/task/command-approval-dialog.tsx deleted file mode 100644 index b3e2b06b2c..0000000000 --- a/apps/web/components/task/command-approval-dialog.tsx +++ /dev/null @@ -1,187 +0,0 @@ -'use client'; - -import { useCallback, useMemo, useState } from 'react'; -import { IconTerminal2, IconAlertTriangle, IconLoader2 } from '@tabler/icons-react'; -import { - AlertDialog, - AlertDialogAction, - AlertDialogCancel, - AlertDialogContent, - AlertDialogDescription, - AlertDialogFooter, - AlertDialogHeader, - AlertDialogTitle, -} from '@kandev/ui/alert-dialog'; -import { useAppStore } from '@/components/state-provider'; -import { getWebSocketClient } from '@/lib/ws/connection'; -import type { Message } from '@/lib/types/http'; - -const EMPTY_MESSAGES: Message[] = []; - -type CommandApprovalDialogProps = { - taskId: string; - /** Only show permissions that don't have a matching tool call (standalone permissions like workspace indexing) */ - standaloneOnly?: boolean; -}; - -export function CommandApprovalDialog({ taskId, standaloneOnly = false }: CommandApprovalDialogProps) { - const [isResponding, setIsResponding] = useState(false); - const pendingPermissions = useAppStore((state) => state.permissions.pending); - const removePendingPermission = useAppStore((state) => state.removePendingPermission); - const activeSessionId = useAppStore((state) => state.tasks.activeSessionId); - const messages = useAppStore((state) => - activeSessionId ? state.messages.bySession[activeSessionId] ?? EMPTY_MESSAGES : EMPTY_MESSAGES - ); - - // Get tool_call_ids from all tool call messages - const toolCallIds = useMemo(() => { - const ids = new Set(); - for (const msg of messages) { - const tcid = msg.metadata?.tool_call_id; - if (msg.type === 'tool_call' && typeof tcid === 'string' && tcid) { - ids.add(tcid); - } - } - return ids; - }, [messages]); - - // Get the first pending permission for this task - // If standaloneOnly, filter out permissions that have a matching tool call - const permission = useMemo(() => { - return pendingPermissions.find((p) => { - if (activeSessionId) { - if (p.session_id !== activeSessionId) return false; - } else if (p.task_id !== taskId) { - return false; - } - if (standaloneOnly && p.tool_call_id && toolCallIds.has(p.tool_call_id)) { - return false; // This permission will be handled by inline UI - } - return true; - }); - }, [pendingPermissions, activeSessionId, taskId, standaloneOnly, toolCallIds]); - - const handleRespond = useCallback( - async (optionId: string, cancelled: boolean = false) => { - if (!permission) return; - setIsResponding(true); - - const client = getWebSocketClient(); - if (!client) { - console.error('WebSocket client not available'); - setIsResponding(false); - return; - } - - try { - await client.request('permission.respond', { - session_id: permission.session_id, - pending_id: permission.pending_id, - option_id: cancelled ? undefined : optionId, - cancelled, - }); - removePendingPermission(permission.pending_id); - } catch (error) { - console.error('Failed to respond to permission request:', error); - } finally { - setIsResponding(false); - } - }, - [permission, removePendingPermission] - ); - - const handleApprove = useCallback(() => { - // Find the "allow" option (allow_once or allow_always) - const allowOption = permission?.options.find( - (opt) => opt.kind === 'allow_once' || opt.kind === 'allow_always' - ); - if (allowOption) { - handleRespond(allowOption.option_id); - } - }, [permission, handleRespond]); - - const handleReject = useCallback(() => { - // Find the "reject" option (reject_once or reject_always) - const rejectOption = permission?.options.find( - (opt) => opt.kind === 'reject_once' || opt.kind === 'reject_always' - ); - if (rejectOption) { - handleRespond(rejectOption.option_id); - } else { - // If no reject option, cancel the request - handleRespond('', true); - } - }, [permission, handleRespond]); - - if (!permission) { - return null; - } - - // Extract command details from action_details - const actionDetails = permission.action_details || {}; - const command = actionDetails.command as string | undefined; - const cwd = actionDetails.cwd as string | undefined; - const reasoning = actionDetails.reasoning as string | undefined; - - return ( - - - -
-
- -
- - {permission.title || 'Command Execution Approval'} - -
- - The agent is requesting permission to execute a command. - - - -
- {command && ( -
-
- - Command -
- - {command} - -
- )} - - {cwd && ( -
- Working directory: {cwd} -
- )} - - {reasoning && ( -
- Reason: {reasoning} -
- )} -
- - - - Reject - - - {isResponding ? ( - <> - - Approving... - - ) : ( - 'Approve' - )} - - - - - ); -} diff --git a/apps/web/components/task/task-chat-panel.tsx b/apps/web/components/task/task-chat-panel.tsx index b36f430bf3..0921e3e7bc 100644 --- a/apps/web/components/task/task-chat-panel.tsx +++ b/apps/web/components/task/task-chat-panel.tsx @@ -1,6 +1,6 @@ 'use client'; -import { FormEvent, useCallback, useEffect, useRef, useState } from 'react'; +import { FormEvent, useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { flushSync } from 'react-dom'; import { useVirtualizer } from '@tanstack/react-virtual'; import { IconBrain, IconListCheck, IconLoader2, IconPaperclip } from '@tabler/icons-react'; @@ -68,18 +68,62 @@ export function TaskChatPanel({ const showLoadingState = (messagesLoading || isInitialLoading) && !isWorking; const { loadMore, hasMore, isLoading: isLoadingMore } = useLazyLoadMessages(resolvedSessionId); + // Build a set of tool_call_ids that exist in the messages + const toolCallIds = useMemo(() => { + const set = new Set(); + for (const message of messages) { + if (message.type === 'tool_call') { + const toolCallId = (message.metadata as { tool_call_id?: string } | undefined)?.tool_call_id; + if (toolCallId) { + set.add(toolCallId); + } + } + } + return set; + }, [messages]); + + // Build a lookup map of tool_call_id → permission_request message + const permissionsByToolCallId = useMemo(() => { + const map = new Map(); + for (const message of messages) { + if (message.type === 'permission_request') { + const toolCallId = (message.metadata as { tool_call_id?: string } | undefined)?.tool_call_id; + if (toolCallId) { + map.set(toolCallId, message); + } + } + } + return map; + }, [messages]); + // Filter to only show chat-relevant types. - const visibleMessages = messages.filter((message) => - !message.type || - message.type === 'message' || - message.type === 'content' || - message.type === 'tool_call' || - message.type === 'progress' || - message.type === 'status' || - message.type === 'error' || - message.type === 'thinking' || - message.type === 'todo' - ); + // Permission requests are only hidden if they have a matching tool_call in the messages + // (so they can be displayed inline). Standalone permissions are shown as separate messages. + const visibleMessages = messages.filter((message) => { + // Standard visible types + if ( + !message.type || + message.type === 'message' || + message.type === 'content' || + message.type === 'tool_call' || + message.type === 'progress' || + message.type === 'status' || + message.type === 'error' || + message.type === 'thinking' || + message.type === 'todo' + ) { + return true; + } + + // Permission requests: show if no matching tool_call exists (standalone permission) + if (message.type === 'permission_request') { + const toolCallId = (message.metadata as { tool_call_id?: string } | undefined)?.tool_call_id; + // Show as standalone if no tool_call_id or no matching tool_call message exists + return !toolCallId || !toolCallIds.has(toolCallId); + } + + return false; + }); // Create a synthetic "user" message for the task description const taskDescriptionMessage: Message | null = @@ -248,6 +292,7 @@ export function TaskChatPanel({ comment={message} isTaskDescription={message.id === 'task-description'} taskId={taskId ?? undefined} + permissionsByToolCallId={permissionsByToolCallId} /> diff --git a/apps/web/lib/state/store.ts b/apps/web/lib/state/store.ts index 76d5deb2c1..2ef4bf3e5d 100644 --- a/apps/web/lib/state/store.ts +++ b/apps/web/lib/state/store.ts @@ -175,30 +175,6 @@ export type DiffState = { files: Array<{ path: string; status: 'A' | 'M' | 'D'; plus: number; minus: number }>; }; -export type PermissionOption = { - option_id: string; - name: string; - kind: string; // allow_once, allow_always, reject_once, reject_always -}; - -export type PendingPermission = { - pending_id: string; - task_id: string; - agent_instance_id: string; - session_id: string; - tool_call_id: string; - title: string; - options: PermissionOption[]; - action_type?: string; // command, file_write, file_read, network, mcp_tool, other - action_details?: Record; - created_at: string; - timestamp: string; -}; - -export type PermissionsState = { - pending: PendingPermission[]; -}; - export type ConnectionState = { status: 'disconnected' | 'connecting' | 'connected' | 'error' | 'reconnecting'; error: string | null; @@ -263,7 +239,6 @@ export type AppState = { sessionAgentctl: SessionAgentctlState; worktrees: WorktreesState; sessionWorktreesBySessionId: SessionWorktreesState; - permissions: PermissionsState; hydrate: (state: Partial) => void; setActiveWorkspace: (workspaceId: string | null) => void; setWorkspaces: (workspaces: WorkspaceState['items']) => void; @@ -317,9 +292,6 @@ export type AppState = { setGitStatus: (sessionId: string, gitStatus: GitStatusEntry) => void; clearGitStatus: (sessionId: string) => void; bumpAgentProfilesVersion: () => void; - addPendingPermission: (permission: PendingPermission) => void; - removePendingPermission: (pendingId: string) => void; - clearPendingPermissions: (sessionId?: string) => void; }; export type AppStore = ReturnType; @@ -356,7 +328,6 @@ const defaultState: AppState = { sessionAgentctl: { itemsBySessionId: {} }, worktrees: { items: {} }, sessionWorktreesBySessionId: { itemsBySessionId: {} }, - permissions: { pending: [] }, hydrate: () => undefined, setActiveWorkspace: () => undefined, setWorkspaces: () => undefined, @@ -396,9 +367,6 @@ const defaultState: AppState = { setGitStatus: () => undefined, clearGitStatus: () => undefined, bumpAgentProfilesVersion: () => undefined, - addPendingPermission: () => undefined, - removePendingPermission: () => undefined, - clearPendingPermissions: () => undefined, }; function mergeInitialState( @@ -444,9 +412,6 @@ function mergeInitialState( | 'setGitStatus' | 'clearGitStatus' | 'bumpAgentProfilesVersion' - | 'addPendingPermission' - | 'removePendingPermission' - | 'clearPendingPermissions' > { if (!initialState) return defaultState; return { @@ -478,7 +443,6 @@ function mergeInitialState( ...defaultState.sessionWorktreesBySessionId, ...initialState.sessionWorktreesBySessionId, }, - permissions: { ...defaultState.permissions, ...initialState.permissions }, }; } @@ -521,7 +485,6 @@ export function createAppStore(initialState?: Partial) { if (state.sessionWorktreesBySessionId) { Object.assign(draft.sessionWorktreesBySessionId, state.sessionWorktreesBySessionId); } - if (state.permissions) Object.assign(draft.permissions, state.permissions); }), setActiveWorkspace: (workspaceId) => set((draft) => { @@ -793,26 +756,6 @@ export function createAppStore(initialState?: Partial) { set((draft) => { draft.agentProfiles.version += 1; }), - addPendingPermission: (permission) => - set((draft) => { - // Avoid duplicates - const exists = draft.permissions.pending.some((p) => p.pending_id === permission.pending_id); - if (!exists) { - draft.permissions.pending.push(permission); - } - }), - removePendingPermission: (pendingId) => - set((draft) => { - draft.permissions.pending = draft.permissions.pending.filter((p) => p.pending_id !== pendingId); - }), - clearPendingPermissions: (sessionId) => - set((draft) => { - if (sessionId) { - draft.permissions.pending = draft.permissions.pending.filter((p) => p.session_id !== sessionId); - } else { - draft.permissions.pending = []; - } - }), })) ); } diff --git a/apps/web/lib/types/backend.ts b/apps/web/lib/types/backend.ts index 066f0d51aa..41e7555917 100644 --- a/apps/web/lib/types/backend.ts +++ b/apps/web/lib/types/backend.ts @@ -36,8 +36,7 @@ export type BackendMessageType = | 'agent.profile.updated' | 'user.settings.updated' | 'session.workspace.file.changes' - | 'session.shell.output' - | 'permission.requested'; + | 'session.shell.output'; export type BackendMessage = { id?: string; @@ -254,26 +253,6 @@ export type ShellOutputPayload = { code?: number; }; -export type PermissionOptionPayload = { - option_id: string; - name: string; - kind: string; // allow_once, allow_always, reject_once, reject_always -}; - -export type PermissionRequestedPayload = { - pending_id: string; - task_id: string; - agent_instance_id: string; - session_id: string; - tool_call_id: string; - title: string; - options: PermissionOptionPayload[]; - action_type?: string; // command, file_write, file_read, network, mcp_tool, other - action_details?: Record; - created_at: string; - timestamp: string; -}; - export type BackendMessageMap = { 'kanban.update': BackendMessage<'kanban.update', KanbanUpdatePayload>; 'task.created': BackendMessage<'task.created', TaskEventPayload>; @@ -313,7 +292,6 @@ export type BackendMessageMap = { 'user.settings.updated': BackendMessage<'user.settings.updated', UserSettingsUpdatedPayload>; 'session.workspace.file.changes': BackendMessage<'session.workspace.file.changes', FileChangeNotificationPayload>; 'session.shell.output': BackendMessage<'session.shell.output', ShellOutputPayload>; - 'permission.requested': BackendMessage<'permission.requested', PermissionRequestedPayload>; }; // Workspace file types diff --git a/apps/web/lib/types/http.ts b/apps/web/lib/types/http.ts index 91c89983c2..eab582ba3e 100644 --- a/apps/web/lib/types/http.ts +++ b/apps/web/lib/types/http.ts @@ -307,7 +307,8 @@ export type MessageType = | 'error' | 'status' | 'thinking' - | 'todo'; + | 'todo' + | 'permission_request'; export type Message = { id: string; diff --git a/apps/web/lib/ws/handlers/messages.ts b/apps/web/lib/ws/handlers/messages.ts index 66d1e44ba3..cb19ef0df4 100644 --- a/apps/web/lib/ws/handlers/messages.ts +++ b/apps/web/lib/ws/handlers/messages.ts @@ -1,6 +1,7 @@ import type { StoreApi } from 'zustand'; import type { AppState } from '@/lib/state/store'; import type { WsHandlers } from '@/lib/ws/handlers/types'; +import type { MessageType } from '@/lib/types/http'; export function registerMessagesHandlers(store: StoreApi): WsHandlers { return { @@ -16,7 +17,7 @@ export function registerMessagesHandlers(store: StoreApi): WsHandlers author_type: payload.author_type, author_id: payload.author_id, content: payload.content, - type: (payload.type as 'message' | 'content' | 'tool_call' | 'progress' | 'error' | 'status') || 'message', + type: (payload.type as MessageType) || 'message', metadata: payload.metadata, requests_input: payload.requests_input, created_at: payload.created_at, @@ -34,7 +35,7 @@ export function registerMessagesHandlers(store: StoreApi): WsHandlers author_type: payload.author_type, author_id: payload.author_id, content: payload.content, - type: (payload.type as 'message' | 'content' | 'tool_call' | 'progress' | 'error' | 'status') || 'message', + type: (payload.type as MessageType) || 'message', metadata: payload.metadata, requests_input: payload.requests_input, created_at: payload.created_at, diff --git a/apps/web/lib/ws/handlers/permissions.ts b/apps/web/lib/ws/handlers/permissions.ts deleted file mode 100644 index c924073fbf..0000000000 --- a/apps/web/lib/ws/handlers/permissions.ts +++ /dev/null @@ -1,29 +0,0 @@ -import type { StoreApi } from 'zustand'; -import type { AppState, PendingPermission } from '@/lib/state/store'; -import type { WsHandlers } from '@/lib/ws/handlers/types'; -import type { PermissionRequestedPayload } from '@/lib/types/backend'; - -export function registerPermissionsHandlers(store: StoreApi): WsHandlers { - return { - 'permission.requested': (message) => { - const payload = message.payload as PermissionRequestedPayload; - const state = store.getState(); - // Convert to PendingPermission type (they're compatible) - const permission: PendingPermission = { - pending_id: payload.pending_id, - task_id: payload.task_id, - agent_instance_id: payload.agent_instance_id, - session_id: payload.session_id, - tool_call_id: payload.tool_call_id, - title: payload.title, - options: payload.options, - action_type: payload.action_type, - action_details: payload.action_details, - created_at: payload.created_at, - timestamp: payload.timestamp, - }; - state.addPendingPermission(permission); - }, - }; -} - diff --git a/apps/web/lib/ws/router.ts b/apps/web/lib/ws/router.ts index 38c5bfc387..76f02fa543 100644 --- a/apps/web/lib/ws/router.ts +++ b/apps/web/lib/ws/router.ts @@ -11,7 +11,6 @@ import { registerEnvironmentsHandlers } from '@/lib/ws/handlers/environments'; import { registerExecutorsHandlers } from '@/lib/ws/handlers/executors'; import { registerGitStatusHandlers } from '@/lib/ws/handlers/git-status'; import { registerKanbanHandlers } from '@/lib/ws/handlers/kanban'; -import { registerPermissionsHandlers } from '@/lib/ws/handlers/permissions'; import { registerSystemEventsHandlers } from '@/lib/ws/handlers/system-events'; import { registerTasksHandlers } from '@/lib/ws/handlers/tasks'; import { registerTerminalsHandlers } from '@/lib/ws/handlers/terminals'; @@ -35,7 +34,6 @@ export function registerWsHandlers(store: StoreApi) { ...registerMessagesHandlers(store), ...registerNotificationsHandlers(store), ...registerGitStatusHandlers(store), - ...registerPermissionsHandlers(store), ...registerSystemEventsHandlers(store), }; }