diff --git a/apps/backend/cmd/kandev/adapters.go b/apps/backend/cmd/kandev/adapters.go index 6730b73858..461d4457e7 100644 --- a/apps/backend/cmd/kandev/adapters.go +++ b/apps/backend/cmd/kandev/adapters.go @@ -9,6 +9,7 @@ import ( "github.com/kandev/kandev/internal/agent/lifecycle" "github.com/kandev/kandev/internal/agent/registry" + "github.com/kandev/kandev/internal/clarification" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/orchestrator" "github.com/kandev/kandev/internal/orchestrator/executor" @@ -339,6 +340,53 @@ func (a *messageCreatorAdapter) UpdatePermissionMessage(ctx context.Context, ses return a.svc.UpdatePermissionMessage(ctx, sessionID, pendingID, status) } +// CreateClarificationRequestMessage creates a message for a clarification request. +// This allows clarification requests to appear in the chat as messages. +func (a *messageCreatorAdapter) CreateClarificationRequestMessage(ctx context.Context, taskID, sessionID, pendingID string, question clarification.Question, clarificationContext string) (string, error) { + // Convert question options to interface{} for metadata storage + options := make([]interface{}, len(question.Options)) + for j, opt := range question.Options { + options[j] = map[string]interface{}{ + "option_id": opt.ID, + "label": opt.Label, + "description": opt.Description, + } + } + + questionData := map[string]interface{}{ + "id": question.ID, + "title": question.Title, + "prompt": question.Prompt, + "options": options, + } + + metadata := map[string]interface{}{ + "pending_id": pendingID, + "question": questionData, + "context": clarificationContext, + "status": "pending", + } + + msg, err := a.svc.CreateMessage(ctx, &taskservice.CreateMessageRequest{ + TaskSessionID: sessionID, + TaskID: taskID, + Content: question.Prompt, + AuthorType: "agent", + Type: "clarification_request", + Metadata: metadata, + RequestsInput: true, // This marks the session as waiting for input + }) + if err != nil { + return "", err + } + return msg.ID, nil +} + +// UpdateClarificationMessage updates a clarification message's status and response +func (a *messageCreatorAdapter) UpdateClarificationMessage(ctx context.Context, sessionID, pendingID, status string, answer *clarification.Answer) error { + return a.svc.UpdateClarificationMessage(ctx, sessionID, pendingID, status, answer) +} + // CreateAgentMessageStreaming creates a new agent message with a pre-generated ID. // This is used for real-time streaming where content arrives incrementally. func (a *messageCreatorAdapter) CreateAgentMessageStreaming(ctx context.Context, messageID, taskID, content, agentSessionID, turnID string) error { diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 89f0a2d9d5..3774ac17da 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -40,6 +40,8 @@ import ( promptcontroller "github.com/kandev/kandev/internal/prompts/controller" prompthandlers "github.com/kandev/kandev/internal/prompts/handlers" + "github.com/kandev/kandev/internal/clarification" + // Task Service packages taskcontroller "github.com/kandev/kandev/internal/task/controller" taskhandlers "github.com/kandev/kandev/internal/task/handlers" @@ -230,6 +232,11 @@ func main() { if cfg.Agent.McpServerURL == "" { cfg.Agent.McpServerURL = mcpServerURL } + // Set MCP server URL on controller and ensure default MCP config for all profiles + agentSettingsController.SetMcpServerURL(mcpServerURL) + if err := agentSettingsController.EnsureDefaultMcpConfig(ctx); err != nil { + log.Warn("Failed to ensure default MCP config", zap.Error(err)) + } } // ============================================ @@ -266,7 +273,7 @@ func main() { // ============================================ log.Info("Initializing Orchestrator...") - orchestratorSvc, err := provideOrchestrator(log, eventBus, taskRepo, taskSvc, userSvc, lifecycleMgr, agentRegistry, services.Workflow) + orchestratorSvc, msgCreator, err := provideOrchestrator(log, eventBus, taskRepo, taskSvc, userSvc, lifecycleMgr, agentRegistry, services.Workflow) if err != nil { log.Fatal("Failed to initialize orchestrator", zap.Error(err)) } @@ -551,6 +558,11 @@ func main() { prompthandlers.RegisterRoutes(router, promptCtrl, log) log.Debug("Registered Prompts handlers (HTTP)") + // Clarification routes for agent questions + clarificationStore := clarification.NewStore(10 * time.Minute) + clarification.RegisterRoutes(router, clarificationStore, gateway.Hub, msgCreator, taskRepo, log) + log.Debug("Registered Clarification handlers (HTTP)") + // Health check (simple HTTP for load balancers/monitoring) router.GET("/health", func(c *gin.Context) { c.JSON(http.StatusOK, gin.H{ diff --git a/apps/backend/cmd/kandev/orchestrator.go b/apps/backend/cmd/kandev/orchestrator.go index edc90b7980..3840aa02d9 100644 --- a/apps/backend/cmd/kandev/orchestrator.go +++ b/apps/backend/cmd/kandev/orchestrator.go @@ -24,9 +24,9 @@ func provideOrchestrator( lifecycleMgr *lifecycle.Manager, agentRegistry *registry.Registry, workflowSvc *workflowservice.Service, -) (*orchestrator.Service, error) { +) (*orchestrator.Service, *messageCreatorAdapter, error) { if lifecycleMgr == nil { - return nil, errors.New("lifecycle manager is required: configure agent runtime (docker or standalone)") + return nil, nil, errors.New("lifecycle manager is required: configure agent runtime (docker or standalone)") } taskRepoAdapter := &taskRepositoryAdapter{repo: taskRepo, svc: taskSvc} @@ -46,7 +46,7 @@ func provideOrchestrator( orchestratorSvc.SetWorkflowStepGetter(&orchestratorWorkflowStepGetterAdapter{svc: workflowSvc}) } - return orchestratorSvc, nil + return orchestratorSvc, msgCreator, nil } // orchestratorWorkflowStepGetterAdapter adapts workflow service to orchestrator's WorkflowStepGetter interface. diff --git a/apps/backend/internal/agent/lifecycle/session.go b/apps/backend/internal/agent/lifecycle/session.go index 52493efdb3..896445332d 100644 --- a/apps/backend/internal/agent/lifecycle/session.go +++ b/apps/backend/internal/agent/lifecycle/session.go @@ -181,6 +181,15 @@ func (sm *SessionManager) getResumeContextPrompt(agentConfig *registry.AgentType return resumePrompt } +// injectSessionContext prepends session context information to the prompt. +// This allows the agent to know which session ID to use when calling MCP tools like ask_user_question. +func (sm *SessionManager) injectSessionContext(sessionID, prompt string) string { + if sessionID == "" { + return prompt + } + return fmt.Sprintf("\nKandev Session ID: %s\nUse this session ID when calling Kandev MCP tools that require a session_id parameter.\n\n\n%s", sessionID, prompt) +} + // loadSession loads an existing session via ACP session/load func (sm *SessionManager) loadSession( ctx context.Context, @@ -308,8 +317,11 @@ func (sm *SessionManager) InitializeAndPrompt( // For resumed sessions (fork_session pattern): If we have session history but no new task description, // we still need to send a resume context prompt so the agent knows about the prior conversation. if taskDescription != "" { + // Inject session ID context so the agent knows which session to use for MCP tools + promptWithSessionContext := sm.injectSessionContext(execution.SessionID, taskDescription) + // Check if we need to inject resume context (fork_session pattern for ACP agents) - effectivePrompt := sm.getResumeContextPrompt(agentConfig, execution.SessionID, taskDescription) + effectivePrompt := sm.getResumeContextPrompt(agentConfig, execution.SessionID, promptWithSessionContext) if effectivePrompt != taskDescription { sm.logger.Info("injecting resume context into initial prompt", zap.String("execution_id", execution.ID), diff --git a/apps/backend/internal/agent/settings/controller/controller.go b/apps/backend/internal/agent/settings/controller/controller.go index 2577587cd6..2c222e870d 100644 --- a/apps/backend/internal/agent/settings/controller/controller.go +++ b/apps/backend/internal/agent/settings/controller/controller.go @@ -37,6 +37,7 @@ type Controller struct { mcpService *mcpconfig.Service modelFetcher *modelfetcher.Fetcher logger *logger.Logger + mcpServerURL string // URL of the embedded Kandev MCP server } type SessionChecker interface { @@ -56,6 +57,80 @@ func NewController(repo store.Repository, discoveryRegistry *discovery.Registry, } } +// SetMcpServerURL sets the URL of the embedded Kandev MCP server. +// This should be called after the MCP server is started. +func (c *Controller) SetMcpServerURL(url string) { + c.mcpServerURL = url +} + +// EnsureDefaultMcpConfig ensures all agent profiles that support MCP have the +// Kandev MCP server configured and enabled by default. +func (c *Controller) EnsureDefaultMcpConfig(ctx context.Context) error { + if c.mcpServerURL == "" { + c.logger.Debug("MCP server URL not set, skipping default MCP config") + return nil + } + + agents, err := c.repo.ListAgents(ctx) + if err != nil { + return err + } + + for _, agent := range agents { + if !agent.SupportsMCP { + continue + } + + profiles, err := c.repo.ListAgentProfiles(ctx, agent.ID) + if err != nil { + return err + } + + for _, profile := range profiles { + // Check if MCP config already exists + existingConfig, err := c.repo.GetAgentProfileMcpConfig(ctx, profile.ID) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + + // Skip if config already exists (don't overwrite user settings) + if existingConfig != nil { + continue + } + + // Create default MCP config with Kandev server enabled + config := &models.AgentProfileMcpConfig{ + ProfileID: profile.ID, + Enabled: true, + Servers: map[string]interface{}{ + "kandev": map[string]interface{}{ + "type": "sse", + "url": c.mcpServerURL, + }, + }, + Meta: map[string]interface{}{}, + CreatedAt: time.Now().UTC(), + UpdatedAt: time.Now().UTC(), + } + + if err := c.repo.UpsertAgentProfileMcpConfig(ctx, config); err != nil { + c.logger.Warn("failed to create default MCP config for profile", + zap.String("profile_id", profile.ID), + zap.String("profile_name", profile.Name), + zap.Error(err)) + continue + } + + c.logger.Info("created default MCP config for profile", + zap.String("profile_id", profile.ID), + zap.String("profile_name", profile.Name), + zap.String("mcp_server_url", c.mcpServerURL)) + } + } + + return nil +} + func (c *Controller) ListDiscovery(ctx context.Context) (*dto.ListDiscoveryResponse, error) { results, err := c.discovery.Detect(ctx) if err != nil { diff --git a/apps/backend/internal/agentctl/server/adapter/opencode_adapter.go b/apps/backend/internal/agentctl/server/adapter/opencode_adapter.go index 32946c31c9..5989db8f5a 100644 --- a/apps/backend/internal/agentctl/server/adapter/opencode_adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/opencode_adapter.go @@ -6,6 +6,8 @@ import ( "encoding/json" "fmt" "io" + "os" + "path/filepath" "strings" "sync" "time" @@ -106,10 +108,25 @@ func NewOpenCodeAdapter(cfg *Config, log *logger.Logger) *OpenCodeAdapter { } // PrepareEnvironment performs protocol-specific setup before the agent process starts. -// For OpenCode, we return environment variables for server authentication. +// For OpenCode, we write MCP servers to ~/.config/opencode/opencode.json and return +// environment variables for server authentication. func (a *OpenCodeAdapter) PrepareEnvironment() (map[string]string, error) { a.logger.Info("PrepareEnvironment called", zap.Int("mcp_server_count", len(a.cfg.McpServers))) + for i, srv := range a.cfg.McpServers { + a.logger.Info("MCP server config", + zap.Int("index", i), + zap.String("name", srv.Name), + zap.String("url", srv.URL), + zap.String("type", srv.Type), + zap.String("command", srv.Command)) + } + + // Write MCP servers to OpenCode config file + if err := WriteOpenCodeMcpConfig(a.cfg.McpServers, "", a.logger); err != nil { + a.logger.Warn("failed to write OpenCode MCP config", zap.Error(err)) + // Continue - MCP servers are optional + } env := map[string]string{ "OPENCODE_SERVER_PASSWORD": a.password, @@ -856,3 +873,106 @@ func (a *OpenCodeAdapter) RespondToPermission(ctx context.Context, requestID str return a.client.ReplyPermission(ctx, requestID, reply, msg) } +// WriteOpenCodeMcpConfig writes MCP server configuration to OpenCode's config file. +// It merges with existing config, preserving other settings and existing MCP servers. +// OpenCode reads MCP servers from ~/.config/opencode/opencode.json at startup time. +// This function should be called before starting the OpenCode process. +// If homeDir is empty, it uses os.UserHomeDir(). +func WriteOpenCodeMcpConfig(mcpServers []McpServerConfig, homeDir string, log *logger.Logger) error { + if len(mcpServers) == 0 { + return nil // Nothing to configure + } + + // Determine config directory + if homeDir == "" { + var err error + homeDir, err = os.UserHomeDir() + if err != nil { + return fmt.Errorf("failed to get user home directory: %w", err) + } + } + configDir := filepath.Join(homeDir, ".config", "opencode") + configPath := filepath.Join(configDir, "opencode.json") + + // Create config directory if it doesn't exist + if err := os.MkdirAll(configDir, 0755); err != nil { + return fmt.Errorf("failed to create opencode config directory: %w", err) + } + + // Read existing config if it exists + existingData, err := os.ReadFile(configPath) + if err != nil && !os.IsNotExist(err) { + return fmt.Errorf("failed to read existing opencode config: %w", err) + } + + // Parse existing config into a generic map to preserve all fields + var rawConfig map[string]interface{} + if len(existingData) > 0 { + if err := json.Unmarshal(existingData, &rawConfig); err != nil { + if log != nil { + log.Warn("failed to parse existing opencode config, will create new", + zap.String("path", configPath), + zap.Error(err)) + } + rawConfig = make(map[string]interface{}) + } + } else { + rawConfig = make(map[string]interface{}) + } + + // Get or create mcp section + mcpSection, ok := rawConfig["mcp"].(map[string]interface{}) + if !ok { + mcpSection = make(map[string]interface{}) + } + + // Add/update our MCP servers + for _, server := range mcpServers { + serverConfig := make(map[string]interface{}) + + if server.Type == "sse" || server.Type == "http" || server.Type == "remote" { + // Remote transport - use url field + serverConfig["type"] = "remote" + if server.URL != "" { + serverConfig["url"] = server.URL + } + } else { + // Local/stdio transport - use command and args + serverConfig["type"] = "local" + if server.Command != "" { + // OpenCode expects command as an array: ["command", "arg1", "arg2"] + cmdArray := []string{server.Command} + cmdArray = append(cmdArray, server.Args...) + serverConfig["command"] = cmdArray + } + } + + // Always enable the server + serverConfig["enabled"] = true + + mcpSection[server.Name] = serverConfig + } + + // Update the mcp section in the config + rawConfig["mcp"] = mcpSection + + // Marshal back to JSON with indentation + output, err := json.MarshalIndent(rawConfig, "", " ") + if err != nil { + return fmt.Errorf("failed to marshal opencode config: %w", err) + } + + // Write config file + if err := os.WriteFile(configPath, output, 0644); err != nil { + return fmt.Errorf("failed to write opencode config: %w", err) + } + + if log != nil { + log.Info("wrote opencode MCP config", + zap.String("path", configPath), + zap.Int("server_count", len(mcpServers))) + } + + return nil +} + diff --git a/apps/backend/internal/clarification/handlers.go b/apps/backend/internal/clarification/handlers.go new file mode 100644 index 0000000000..3a11d98332 --- /dev/null +++ b/apps/backend/internal/clarification/handlers.go @@ -0,0 +1,222 @@ +// Package clarification provides types and services for agent clarification requests. +package clarification + +import ( + "context" + "fmt" + "net/http" + "time" + + "github.com/gin-gonic/gin" + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/task/repository" + wsmsg "github.com/kandev/kandev/pkg/websocket" + "go.uber.org/zap" +) + +// Broadcaster interface for sending WebSocket notifications +type Broadcaster interface { + BroadcastToSession(sessionID string, msg *wsmsg.Message) +} + +// MessageCreator interface for creating messages in the database +type MessageCreator interface { + // CreateClarificationRequestMessage creates a message for a clarification request + CreateClarificationRequestMessage(ctx context.Context, taskID, sessionID, pendingID string, question Question, clarificationContext string) (string, error) + // UpdateClarificationMessage updates a clarification message's status and response + UpdateClarificationMessage(ctx context.Context, sessionID, pendingID, status string, answer *Answer) error +} + +// Handlers provides HTTP handlers for clarification requests. +type Handlers struct { + store *Store + hub Broadcaster + messageCreator MessageCreator + repo repository.Repository + logger *logger.Logger +} + +// NewHandlers creates new clarification handlers. +func NewHandlers(store *Store, hub Broadcaster, messageCreator MessageCreator, repo repository.Repository, log *logger.Logger) *Handlers { + return &Handlers{ + store: store, + hub: hub, + messageCreator: messageCreator, + repo: repo, + logger: log.WithFields(zap.String("component", "clarification-handlers")), + } +} + +// RegisterRoutes registers clarification HTTP routes. +func RegisterRoutes(router *gin.Engine, store *Store, hub Broadcaster, messageCreator MessageCreator, repo repository.Repository, log *logger.Logger) { + h := NewHandlers(store, hub, messageCreator, repo, log) + api := router.Group("/api/v1/clarification") + api.POST("/request", h.httpCreateRequest) + api.GET("/:id", h.httpGetRequest) + api.GET("/:id/wait", h.httpWaitForResponse) + api.POST("/:id/respond", h.httpRespond) +} + +// CreateRequestBody is the request body for creating a clarification request. +type CreateRequestBody struct { + SessionID string `json:"session_id" binding:"required"` + TaskID string `json:"task_id"` + Question Question `json:"question" binding:"required"` + Context string `json:"context"` +} + +// CreateRequestResponse is the response for creating a clarification request. +type CreateRequestResponse struct { + PendingID string `json:"pending_id"` +} + +func (h *Handlers) httpCreateRequest(c *gin.Context) { + var body CreateRequestBody + if err := c.ShouldBindJSON(&body); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid payload: " + err.Error()}) + return + } + + // Validate question has ID, generate if missing + if body.Question.ID == "" { + body.Question.ID = "q1" + } + // Validate options have IDs + for j := range body.Question.Options { + if body.Question.Options[j].ID == "" { + body.Question.Options[j].ID = generateOptionID(0, j) + } + } + + // Look up the task ID for this session + sessionID := body.SessionID + taskID := body.TaskID + if taskID == "" { + session, err := h.repo.GetTaskSession(c.Request.Context(), sessionID) + if err != nil { + h.logger.Warn("failed to look up session", + zap.String("session_id", sessionID), + zap.Error(err)) + } else { + taskID = session.TaskID + } + } + + req := &Request{ + SessionID: sessionID, + TaskID: taskID, + Question: body.Question, + Context: body.Context, + } + + pendingID := h.store.CreateRequest(req) + + // Create a message in the database for the clarification request. + // This triggers the session.message.added WebSocket event which the frontend listens to. + if h.messageCreator != nil { + _, err := h.messageCreator.CreateClarificationRequestMessage( + c.Request.Context(), + taskID, + sessionID, + pendingID, + body.Question, + body.Context, + ) + if err != nil { + h.logger.Error("failed to create clarification request message", + zap.String("pending_id", pendingID), + zap.String("session_id", sessionID), + zap.Error(err)) + // Don't fail the request - the clarification is still created in the store + } + } + + c.JSON(http.StatusOK, CreateRequestResponse{PendingID: pendingID}) +} + +func (h *Handlers) httpGetRequest(c *gin.Context) { + pendingID := c.Param("id") + + req, ok := h.store.GetRequest(pendingID) + if !ok { + c.JSON(http.StatusNotFound, gin.H{"error": "clarification request not found"}) + return + } + + c.JSON(http.StatusOK, req) +} + +func (h *Handlers) httpWaitForResponse(c *gin.Context) { + pendingID := c.Param("id") + resp, err := h.store.WaitForResponse(c.Request.Context(), pendingID) + if err != nil { + c.JSON(http.StatusGatewayTimeout, gin.H{"error": err.Error()}) + return + } + + c.JSON(http.StatusOK, resp) +} + +// RespondBody is the request body for responding to a clarification request. +type RespondBody struct { + Answers []Answer `json:"answers"` // Keep as array for frontend compatibility, but only first is used + Rejected bool `json:"rejected"` + RejectReason string `json:"reject_reason"` +} + +func (h *Handlers) httpRespond(c *gin.Context) { + pendingID := c.Param("id") + + var body RespondBody + if err := c.ShouldBindJSON(&body); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid payload: " + err.Error()}) + return + } + + // Get the pending request to find the session ID + pending, ok := h.store.GetRequest(pendingID) + if !ok { + c.JSON(http.StatusNotFound, gin.H{"error": "clarification request not found"}) + return + } + + // Extract the single answer (first one if provided) + var answer *Answer + if len(body.Answers) > 0 { + answer = &body.Answers[0] + } + + resp := &Response{ + PendingID: pendingID, + Answer: answer, + Rejected: body.Rejected, + RejectReason: body.RejectReason, + RespondedAt: time.Now(), + } + + if err := h.store.Respond(pendingID, resp); err != nil { + h.logger.Warn("failed to respond to clarification", + zap.String("pending_id", pendingID), + zap.Error(err)) + c.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + return + } + + // Update the message in the database with status and answer + status := "answered" + if body.Rejected { + status = "rejected" + } + if err := h.messageCreator.UpdateClarificationMessage(c.Request.Context(), pending.SessionID, pendingID, status, answer); err != nil { + h.logger.Warn("failed to update clarification message", + zap.String("pending_id", pendingID), + zap.Error(err)) + } + + c.JSON(http.StatusOK, gin.H{"success": true}) +} + +func generateOptionID(questionIndex, optionIndex int) string { + return fmt.Sprintf("q%d_opt%d", questionIndex+1, optionIndex+1) +} + diff --git a/apps/backend/internal/clarification/store.go b/apps/backend/internal/clarification/store.go new file mode 100644 index 0000000000..9d4234424e --- /dev/null +++ b/apps/backend/internal/clarification/store.go @@ -0,0 +1,184 @@ +// Package clarification provides types and services for agent clarification requests. +package clarification + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/google/uuid" +) + +// Store manages pending clarification requests. +// It provides thread-safe storage and notification when responses arrive. +type Store struct { + mu sync.RWMutex + pending map[string]*PendingClarification + timeout time.Duration +} + +// NewStore creates a new clarification store. +func NewStore(timeout time.Duration) *Store { + if timeout == 0 { + timeout = 10 * time.Minute // Default timeout + } + return &Store{ + pending: make(map[string]*PendingClarification), + timeout: timeout, + } +} + +// CreateRequest creates a new clarification request and returns its pending ID. +// The request will be stored until a response is received or it times out. +func (s *Store) CreateRequest(req *Request) string { + s.mu.Lock() + defer s.mu.Unlock() + + if req.PendingID == "" { + req.PendingID = uuid.New().String() + } + req.CreatedAt = time.Now() + + s.pending[req.PendingID] = &PendingClarification{ + Request: req, + ResponseCh: make(chan *Response, 1), + CreatedAt: time.Now(), + } + + return req.PendingID +} + +// GetRequest returns a pending clarification request by ID. +func (s *Store) GetRequest(pendingID string) (*Request, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + + pending, ok := s.pending[pendingID] + if !ok { + return nil, false + } + return pending.Request, true +} + +// WaitForResponse blocks until a response is received or the context is cancelled. +// Returns the response or an error if cancelled/timed out. +func (s *Store) WaitForResponse(ctx context.Context, pendingID string) (*Response, error) { + s.mu.RLock() + pending, ok := s.pending[pendingID] + s.mu.RUnlock() + + if !ok { + return nil, fmt.Errorf("clarification request not found: %s", pendingID) + } + + // Create timeout context + timeoutCtx, cancel := context.WithTimeout(ctx, s.timeout) + defer cancel() + + select { + case resp := <-pending.ResponseCh: + // Clean up after receiving response + s.mu.Lock() + delete(s.pending, pendingID) + s.mu.Unlock() + return resp, nil + case <-timeoutCtx.Done(): + // Clean up on timeout + s.mu.Lock() + delete(s.pending, pendingID) + s.mu.Unlock() + if ctx.Err() != nil { + return nil, ctx.Err() + } + return nil, fmt.Errorf("clarification request timed out: %s", pendingID) + } +} + +// Respond submits a response to a pending clarification request. +// Returns an error if the request is not found. +func (s *Store) Respond(pendingID string, resp *Response) error { + s.mu.RLock() + pending, ok := s.pending[pendingID] + s.mu.RUnlock() + + if !ok { + return fmt.Errorf("clarification request not found: %s", pendingID) + } + + resp.PendingID = pendingID + resp.RespondedAt = time.Now() + + // Non-blocking send (channel has buffer of 1) + select { + case pending.ResponseCh <- resp: + return nil + default: + return fmt.Errorf("response already submitted for: %s", pendingID) + } +} + +// Cancel cancels a pending clarification request. +func (s *Store) Cancel(pendingID string) error { + s.mu.Lock() + defer s.mu.Unlock() + + pending, ok := s.pending[pendingID] + if !ok { + return fmt.Errorf("clarification request not found: %s", pendingID) + } + + // Send a cancelled response + select { + case pending.ResponseCh <- &Response{ + PendingID: pendingID, + Rejected: true, + RejectReason: "cancelled", + RespondedAt: time.Now(), + }: + default: + } + + delete(s.pending, pendingID) + return nil +} + +// CleanupExpired removes expired pending requests. +// Returns the number of requests cleaned up. +func (s *Store) CleanupExpired() int { + s.mu.Lock() + defer s.mu.Unlock() + + count := 0 + now := time.Now() + for id, pending := range s.pending { + if now.Sub(pending.CreatedAt) > s.timeout { + // Send timeout response + select { + case pending.ResponseCh <- &Response{ + PendingID: id, + Rejected: true, + RejectReason: "expired", + RespondedAt: now, + }: + default: + } + delete(s.pending, id) + count++ + } + } + return count +} + +// ListPending returns all pending clarification requests. +func (s *Store) ListPending() []*Request { + s.mu.RLock() + defer s.mu.RUnlock() + + requests := make([]*Request, 0, len(s.pending)) + for _, pending := range s.pending { + requests = append(requests, pending.Request) + } + return requests +} + diff --git a/apps/backend/internal/clarification/types.go b/apps/backend/internal/clarification/types.go new file mode 100644 index 0000000000..92f0256271 --- /dev/null +++ b/apps/backend/internal/clarification/types.go @@ -0,0 +1,67 @@ +// Package clarification provides types and services for agent clarification requests. +// This allows agents to ask structured questions to users and wait for responses. +package clarification + +import ( + "time" +) + +// Option represents a single choice option for a question. +type Option struct { + ID string `json:"option_id"` + Label string `json:"label"` // Concise 1-5 words + Description string `json:"description"` // Explanation of the option +} + +// Question represents a single question with multiple choice options. +type Question struct { + ID string `json:"id"` + Title string `json:"title"` // Short label (max 12 chars) + Prompt string `json:"prompt"` // Full question text + Options []Option `json:"options"` // 2-4 options +} + +// Request represents a clarification request from an agent. +type Request struct { + PendingID string `json:"pending_id"` + SessionID string `json:"session_id"` + TaskID string `json:"task_id"` + Question Question `json:"question"` // Single question + Context string `json:"context,omitempty"` // Optional context for the question + CreatedAt time.Time `json:"created_at"` +} + +// Answer represents the user's answer to a single question. +type Answer struct { + QuestionID string `json:"question_id"` + SelectedOptions []string `json:"selected_options,omitempty"` // Option IDs + CustomText string `json:"custom_text,omitempty"` // Free-text input +} + +// Response represents the user's response to a clarification request. +type Response struct { + PendingID string `json:"pending_id"` + Answer *Answer `json:"answer,omitempty"` + Rejected bool `json:"rejected,omitempty"` + RejectReason string `json:"reject_reason,omitempty"` // If rejected + RespondedAt time.Time `json:"responded_at"` +} + +// PendingClarification represents a clarification request waiting for a response. +type PendingClarification struct { + Request *Request + ResponseCh chan *Response + CreatedAt time.Time +} + +// Status represents the status of a clarification request. +type Status string + +const ( + StatusPending Status = "pending" + StatusAnswered Status = "answered" + StatusRejected Status = "rejected" + StatusExpired Status = "expired" + StatusCancelled Status = "cancelled" +) + diff --git a/apps/backend/internal/mcpserver/tools.go b/apps/backend/internal/mcpserver/tools.go index e65ece6f05..6ce9b93763 100644 --- a/apps/backend/internal/mcpserver/tools.go +++ b/apps/backend/internal/mcpserver/tools.go @@ -6,6 +6,7 @@ import ( "encoding/json" "fmt" "net/http" + "strings" "github.com/kandev/kandev/internal/common/logger" "github.com/mark3labs/mcp-go/mcp" @@ -106,7 +107,39 @@ func registerTools(s *server.MCPServer, cfg Config, log *logger.Logger) { updateTaskHandler(cfg, log), ) - log.Info("registered MCP tools", zap.Int("count", 6)) + // Ask User Question tool + s.AddTool( + mcp.NewTool("ask_user_question", + mcp.WithDescription( + "Ask the user a clarifying question when you need more information to proceed. "+ + "Use this tool when you need to:\n"+ + "1. Gather user preferences or requirements\n"+ + "2. Clarify ambiguous instructions\n"+ + "3. Get decisions on implementation choices\n"+ + "4. Offer choices about what direction to take\n\n"+ + "The question includes 2-4 multiple-choice options. Users can also provide custom text input.\n"+ + "Returns the user's answer.", + ), + mcp.WithString("session_id", + mcp.Required(), + mcp.Description("The session ID for this clarification request"), + ), + mcp.WithString("prompt", + mcp.Required(), + mcp.Description("The question to ask the user"), + ), + mcp.WithArray("options", + mcp.Required(), + mcp.Description("2-4 options for the user to choose from. Each option should have: label (concise 1-5 words) and description (explanation of the option)"), + ), + mcp.WithString("context", + mcp.Description("Optional context explaining why this question is being asked"), + ), + ), + askUserQuestionHandler(cfg, log), + ) + + log.Info("registered MCP tools", zap.Int("count", 7)) } func listWorkspacesHandler(cfg Config, log *logger.Logger) server.ToolHandlerFunc { @@ -311,3 +344,190 @@ func updateTaskHandler(cfg Config, log *logger.Logger) server.ToolHandlerFunc { } } +func askUserQuestionHandler(cfg Config, log *logger.Logger) server.ToolHandlerFunc { + return func(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + sessionID, err := req.RequireString("session_id") + if err != nil || sessionID == "" { + return mcp.NewToolResultError("session_id is required - this tool can only be used within a Kandev task session"), nil + } + + prompt, err := req.RequireString("prompt") + if err != nil { + return mcp.NewToolResultError("prompt is required"), nil + } + + args := req.GetArguments() + optionsRaw, ok := args["options"] + if !ok { + return mcp.NewToolResultError("options is required"), nil + } + + // Parse options from the raw interface + optionsJSON, err := json.Marshal(optionsRaw) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("Failed to parse options: %v", err)), nil + } + + var options []map[string]interface{} + if err := json.Unmarshal(optionsJSON, &options); err != nil { + return mcp.NewToolResultError(fmt.Sprintf("Failed to parse options: %v", err)), nil + } + + if len(options) < 2 || len(options) > 4 { + return mcp.NewToolResultError("Must provide 2-4 options"), nil + } + + // Generate option IDs to ensure consistency between request and response + // The backend also generates these, but we need them here for matching + for i := range options { + options[i]["option_id"] = fmt.Sprintf("q1_opt%d", i+1) + } + + context := req.GetString("context", "") + + // Build the single question + question := map[string]interface{}{ + "id": "q1", + "title": "Question", + "prompt": prompt, + "options": options, + } + + // Create the clarification request (backend still expects questions array for now) + payload := map[string]interface{}{ + "session_id": sessionID, + "question": question, + "context": context, + } + + body, _ := json.Marshal(payload) + url := fmt.Sprintf("%s/api/v1/clarification/request", cfg.KandevURL) + + log.Debug("creating clarification request", + zap.String("session_id", sessionID)) + + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("Failed to create request: %v", err)), nil + } + httpReq.Header.Set("Content-Type", "application/json") + + resp, err := http.DefaultClient.Do(httpReq) + if err != nil { + log.Error("failed to create clarification request", zap.Error(err)) + return mcp.NewToolResultError(fmt.Sprintf("Failed to create clarification request: %v", err)), nil + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode >= 400 { + var errBody json.RawMessage + _ = json.NewDecoder(resp.Body).Decode(&errBody) + return mcp.NewToolResultError(fmt.Sprintf("API error (%d): %s", resp.StatusCode, string(errBody))), nil + } + + var createResp struct { + PendingID string `json:"pending_id"` + } + if err := json.NewDecoder(resp.Body).Decode(&createResp); err != nil { + return mcp.NewToolResultError(fmt.Sprintf("Failed to parse response: %v", err)), nil + } + + // Now wait for the user's response + waitURL := fmt.Sprintf("%s/api/v1/clarification/%s/wait", cfg.KandevURL, createResp.PendingID) + + log.Debug("waiting for user response", + zap.String("pending_id", createResp.PendingID)) + + waitReq, err := http.NewRequestWithContext(ctx, http.MethodGet, waitURL, nil) + if err != nil { + return mcp.NewToolResultError(fmt.Sprintf("Failed to create wait request: %v", err)), nil + } + + waitResp, err := http.DefaultClient.Do(waitReq) + if err != nil { + log.Error("failed to wait for response", zap.Error(err)) + return mcp.NewToolResultError(fmt.Sprintf("Failed to wait for response: %v", err)), nil + } + defer func() { _ = waitResp.Body.Close() }() + + if waitResp.StatusCode >= 400 { + var errBody json.RawMessage + _ = json.NewDecoder(waitResp.Body).Decode(&errBody) + return mcp.NewToolResultError(fmt.Sprintf("Wait error (%d): %s", waitResp.StatusCode, string(errBody))), nil + } + + var userResponse struct { + PendingID string `json:"pending_id"` + Answer struct { + SelectedOptions []string `json:"selected_options"` + CustomText string `json:"custom_text"` + } `json:"answer"` + Rejected bool `json:"rejected"` + RejectReason string `json:"reject_reason"` + } + if err := json.NewDecoder(waitResp.Body).Decode(&userResponse); err != nil { + return mcp.NewToolResultError(fmt.Sprintf("Failed to parse user response: %v", err)), nil + } + + // Format the response for the agent + result := formatClarificationResponse(prompt, options, userResponse.Answer, userResponse.Rejected, userResponse.RejectReason) + + log.Debug("clarification response received", + zap.String("pending_id", createResp.PendingID), + zap.Bool("rejected", userResponse.Rejected)) + + return mcp.NewToolResultText(result), nil + } +} + +func formatClarificationResponse(prompt string, options []map[string]interface{}, answer struct { + SelectedOptions []string `json:"selected_options"` + CustomText string `json:"custom_text"` +}, rejected bool, rejectReason string) string { + if rejected { + if rejectReason != "" { + return fmt.Sprintf("The user declined to answer and said: %s\n\nProceed with your best judgment based on the available information.", rejectReason) + } + return "The user declined to answer this question. Proceed with your best judgment based on the available information." + } + + var result strings.Builder + result.WriteString("# User Response\n\n") + result.WriteString(fmt.Sprintf("**Question asked:** %s\n\n", prompt)) + + hasAnswer := false + + if len(answer.SelectedOptions) > 0 { + // Find the label and description for the selected option + for _, opt := range options { + optID, _ := opt["option_id"].(string) + if optID == "" { + optID, _ = opt["id"].(string) + } + for _, selected := range answer.SelectedOptions { + if optID == selected { + label, _ := opt["label"].(string) + description, _ := opt["description"].(string) + result.WriteString(fmt.Sprintf("**User's choice:** %s\n", label)) + if description != "" { + result.WriteString(fmt.Sprintf("**Choice description:** %s\n", description)) + } + hasAnswer = true + } + } + } + } + + if answer.CustomText != "" { + result.WriteString(fmt.Sprintf("\n**Additional user input:** %s\n", answer.CustomText)) + hasAnswer = true + } + + if !hasAnswer { + result.WriteString("The user did not provide an answer.\n") + } + + result.WriteString("\nProceed with the task based on the user's response above.") + + return result.String() +} diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 63d7a64c3e..c76761d2e3 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -108,6 +108,8 @@ const ( MessageTypeStatus MessageType = "status" // MessageTypePermissionRequest is for agent permission requests MessageTypePermissionRequest MessageType = "permission_request" + // MessageTypeClarificationRequest is for agent clarification questions + MessageTypeClarificationRequest MessageType = "clarification_request" // MessageTypeScriptExecution is for setup/cleanup script execution messages MessageTypeScriptExecution MessageType = "script_execution" // MessageTypeThinking is for agent thinking/reasoning content diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index 6b70436a9c..1423b099b8 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -2149,6 +2149,73 @@ func (s *Service) UpdatePermissionMessage(ctx context.Context, sessionID, pendin return nil } +// UpdateClarificationMessage updates a clarification request message's status and response. +// It includes retry logic to handle race conditions. +// The answers parameter should be a slice of answer objects with question_id, selected_options, and custom_text. +func (s *Service) UpdateClarificationMessage(ctx context.Context, sessionID, pendingID, status string, answers interface{}) 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("clarification 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("clarification 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 answers != nil { + message.Metadata["response"] = answers + } + + if err := s.repo.UpdateMessage(ctx, message); err != nil { + s.logger.Error("failed to update clarification 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("clarification message updated", + zap.String("message_id", message.ID), + zap.String("pending_id", pendingID), + zap.String("status", status)) + + return nil +} + // Turn operations // StartTurn creates a new turn for a session and publishes the turn.started event. diff --git a/apps/backend/internal/workflow/repository/sqlite.go b/apps/backend/internal/workflow/repository/sqlite.go index 3c5e19799b..316067f468 100644 --- a/apps/backend/internal/workflow/repository/sqlite.go +++ b/apps/backend/internal/workflow/repository/sqlite.go @@ -266,7 +266,7 @@ func (r *Repository) getDevTemplate(now time.Time) *models.WorkflowTemplate { TaskState: "IN_PROGRESS", AutoStartAgent: true, PlanMode: true, - PromptPrefix: "[PLANNING PHASE]\nAnalyze this task and create a detailed implementation plan.\nDo NOT make any code changes yet - only analyze and plan.\n\nCreate a plan that includes:\n1. Understanding of the requirements\n2. Files that need to be modified or created\n3. Step-by-step implementation approach\n4. Potential risks or considerations\n\nIMPORTANT: Save your plan to ONLY this file: `.kandev/plans/{task_id}.md`\nDo not create any other files during this phase - only the plan file above.", + PromptPrefix: "[PLANNING PHASE]\nAnalyze this task and create a detailed implementation plan.\nDo NOT make any code changes yet - only analyze and plan.\n\nBefore creating the plan, ask the user clarifying questions if anything is unclear or ambiguous about the requirements. Use the ask_user_question_kandev tool to get answers before proceeding.\n\nCreate a plan that includes:\n1. Understanding of the requirements\n2. Files that need to be modified or created\n3. Step-by-step implementation approach\n4. Potential risks or considerations\n\nIMPORTANT: Save your plan to ONLY this file: `.kandev/plans/{task_id}.md`\nDo not create any other files during this phase - only the plan file above.", OnCompleteStepID: "review-plan", AllowManualMove: true, }, @@ -349,7 +349,7 @@ func (r *Repository) getArchitectureTemplate(now time.Time) *models.WorkflowTemp TaskState: "IN_PROGRESS", AutoStartAgent: true, PlanMode: true, - PromptPrefix: "[ARCHITECTURE PHASE]\nAnalyze and design the architecture for this task.\n\nIMPORTANT: Document your design in ONLY this file: `.kandev/designs/{task_id}.md`\nDo not create any other files during this phase - only the design file above.", + PromptPrefix: "[ARCHITECTURE PHASE]\nAnalyze and design the architecture for this task.\n\nBefore creating the design, ask the user clarifying questions if anything is unclear or ambiguous about the requirements. Use the ask_user_question_kandev tool to get answers before proceeding.\n\nIMPORTANT: Document your design in ONLY this file: `.kandev/designs/{task_id}.md`\nDo not create any other files during this phase - only the design file above.", OnCompleteStepID: "review", AllowManualMove: true, }, diff --git a/apps/web/components/task/chat/chat-input-container.tsx b/apps/web/components/task/chat/chat-input-container.tsx index f982530c78..98af91b5c9 100644 --- a/apps/web/components/task/chat/chat-input-container.tsx +++ b/apps/web/components/task/chat/chat-input-container.tsx @@ -26,8 +26,10 @@ import { ModelSelector } from '@/components/task/model-selector'; import { RichTextInput, type RichTextInputHandle } from './rich-text-input'; import { MentionMenu } from './mention-menu'; import { SlashCommandMenu } from './slash-command-menu'; +import { ClarificationInputOverlay } from './clarification-input-overlay'; import { useInlineMention } from '@/hooks/use-inline-mention'; import { useInlineSlash } from '@/hooks/use-inline-slash'; +import type { Message } from '@/lib/types/http'; type ChatInputContainerProps = { onSubmit: (message: string) => void; @@ -44,6 +46,8 @@ type ChatInputContainerProps = { showApproveButton?: boolean; onApprove?: () => void; placeholder?: string; + pendingClarification?: Message | null; + onClarificationResolved?: () => void; }; export function ChatInputContainer({ @@ -61,6 +65,8 @@ export function ChatInputContainer({ showApproveButton, onApprove, placeholder, + pendingClarification, + onClarificationResolved, }: ChatInputContainerProps) { // Keep input state local to avoid re-rendering parent on every keystroke const [value, setValue] = useState(''); @@ -108,12 +114,14 @@ export function ChatInputContainer({ }, [mention.isOpen, slash.isOpen, value, onSubmit]); const isDisabled = isStarting || isSending; + const hasPendingClarification = pendingClarification && onClarificationResolved; return (
{/* Popup menus */} @@ -138,17 +146,27 @@ export function ChatInputContainer({ setSelectedIndex={slash.setSelectedIndex} /> - {/* Input area */} - + {/* Clarification inline (replaces input area when pending) */} + {hasPendingClarification ? ( +
+ +
+ ) : ( + /* Normal input area */ + + )} {/* Integrated toolbar */}
diff --git a/apps/web/components/task/chat/clarification-input-overlay.tsx b/apps/web/components/task/chat/clarification-input-overlay.tsx new file mode 100644 index 0000000000..cc5c1c9c3e --- /dev/null +++ b/apps/web/components/task/chat/clarification-input-overlay.tsx @@ -0,0 +1,177 @@ +'use client'; + +import { useCallback, useState } from 'react'; +import { IconX, IconMessageQuestion } from '@tabler/icons-react'; +import { Tooltip, TooltipContent, TooltipTrigger } from '@kandev/ui/tooltip'; +import { cn } from '@/lib/utils'; +import { getBackendConfig } from '@/lib/config'; +import type { Message, ClarificationRequestMetadata, ClarificationAnswer } from '@/lib/types/http'; + +type ClarificationInputOverlayProps = { + message: Message; + onResolved: () => void; +}; + +/** + * Inline clarification UI - simple numbered text options like Conductor. + */ +export function ClarificationInputOverlay({ message, onResolved }: ClarificationInputOverlayProps) { + const metadata = message.metadata as ClarificationRequestMetadata | undefined; + const [customText, setCustomText] = useState(''); + const [isSubmitting, setIsSubmitting] = useState(false); + + const handleSubmitOption = useCallback(async (optionId: string) => { + if (!metadata?.pending_id || isSubmitting) return; + setIsSubmitting(true); + try { + const answer: ClarificationAnswer = { + question_id: metadata.question.id, + selected_options: [optionId], + }; + + const { apiBaseUrl } = getBackendConfig(); + const response = await fetch(`${apiBaseUrl}/api/v1/clarification/${metadata.pending_id}/respond`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ answers: [answer], rejected: false }), + }); + if (response.ok) { + onResolved(); + } + } catch (error) { + console.error('Failed to submit clarification response:', error); + } finally { + setIsSubmitting(false); + } + }, [metadata, isSubmitting, onResolved]); + + const handleSubmitCustom = useCallback(async () => { + if (!metadata?.pending_id || isSubmitting || !customText.trim()) return; + setIsSubmitting(true); + try { + const answer: ClarificationAnswer = { + question_id: metadata.question.id, + selected_options: [], + custom_text: customText.trim(), + }; + + const { apiBaseUrl } = getBackendConfig(); + const response = await fetch(`${apiBaseUrl}/api/v1/clarification/${metadata.pending_id}/respond`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ answers: [answer], rejected: false }), + }); + if (response.ok) { + onResolved(); + } + } catch (error) { + console.error('Failed to submit clarification response:', error); + } finally { + setIsSubmitting(false); + } + }, [metadata, isSubmitting, customText, onResolved]); + + const handleSkip = useCallback(async () => { + if (!metadata?.pending_id || isSubmitting) return; + setIsSubmitting(true); + try { + const { apiBaseUrl } = getBackendConfig(); + const response = await fetch(`${apiBaseUrl}/api/v1/clarification/${metadata.pending_id}/respond`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ answers: [], rejected: true, reject_reason: 'User skipped' }), + }); + if (response.ok) { + onResolved(); + } + } catch (error) { + console.error('Failed to skip clarification:', error); + } finally { + setIsSubmitting(false); + } + }, [metadata, isSubmitting, onResolved]); + + if (!metadata?.question) { + return null; + } + + const question = metadata.question; + + return ( +
+ {/* Close/skip button */} + + + {/* Content */} +
+ {/* Question text */} +
+ +

{question.prompt}

+
+ + {/* Options with bullets - indented to align with question text */} +
+ {question.options.map((option) => { + const button = ( + + ); + + if (option.description) { + return ( + + {button} + + {option.description} + + + ); + } + + return button; + })} +
+ + {/* Custom input row - indented to align with options */} +
+ • + setCustomText(e.target.value)} + disabled={isSubmitting} + className="flex-1 text-sm bg-transparent placeholder:text-muted-foreground focus:outline-none" + onKeyDown={(e) => { + if (e.key === 'Enter' && !e.shiftKey && customText.trim()) { + e.preventDefault(); + handleSubmitCustom(); + } + }} + /> +
+
+
+ ); +} + diff --git a/apps/web/components/task/chat/message-renderer.tsx b/apps/web/components/task/chat/message-renderer.tsx index 7caf504d3b..2842970733 100644 --- a/apps/web/components/task/chat/message-renderer.tsx +++ b/apps/web/components/task/chat/message-renderer.tsx @@ -9,6 +9,7 @@ import { ToolCallMessage } from '@/components/task/chat/messages/tool-call-messa import { ThinkingMessage } from '@/components/task/chat/messages/thinking-message'; import { TodoMessage } from '@/components/task/chat/messages/todo-message'; import { ScriptExecutionMessage } from '@/components/task/chat/messages/script-execution-message'; +import { ClarificationRequestMessage } from '@/components/task/chat/messages/clarification-request-message'; type AdapterContext = { isTaskDescription: boolean; @@ -47,6 +48,10 @@ const adapters: MessageAdapter[] = [ matches: (comment) => comment.type === 'permission_request', render: (comment) => , }, + { + matches: (comment) => comment.type === 'clarification_request', + render: (comment) => , + }, { matches: (comment) => comment.type === 'script_execution', render: (comment) => , diff --git a/apps/web/components/task/chat/messages/clarification-request-message.tsx b/apps/web/components/task/chat/messages/clarification-request-message.tsx new file mode 100644 index 0000000000..4a12acbe7c --- /dev/null +++ b/apps/web/components/task/chat/messages/clarification-request-message.tsx @@ -0,0 +1,89 @@ +'use client'; + +import { IconMessageQuestion, IconCheck, IconX } from '@tabler/icons-react'; +import { cn } from '@/lib/utils'; +import type { Message, ClarificationRequestMetadata } from '@/lib/types/http'; + +type ClarificationRequestMessageProps = { + comment: Message; +}; + +/** + * Displays a resolved clarification request in the chat history. + * Pending clarifications are shown in the input area instead. + */ +export function ClarificationRequestMessage({ comment }: ClarificationRequestMessageProps) { + const metadata = comment.metadata as ClarificationRequestMetadata | undefined; + + if (!metadata?.question) { + return null; + } + + const question = metadata.question; + const status = metadata.status; + const isAnswered = status === 'answered'; + const isSkipped = status === 'rejected'; + + const getStatusIndicator = () => { + if (isAnswered) { + return ; + } + if (isSkipped) { + return ; + } + return null; + }; + + // Get the answer summary for display + const getAnswerSummary = () => { + const response = metadata.response; + if (!response) return 'No selection'; + + const parts: string[] = []; + + // Get selected option labels + if (response.selected_options?.length) { + for (const optionId of response.selected_options) { + const option = question.options.find((o) => o.option_id === optionId); + if (option) { + parts.push(option.label); + } + } + } + + // Add custom text + if (response.custom_text) { + parts.push(`"${response.custom_text}"`); + } + + return parts.length > 0 ? parts.join(', ') : 'No selection'; + }; + + return ( +
+
+ {/* Icon */} +
+ +
+ + {/* Content */} +
+ {/* Header with question and answer */} +
+ + {question.prompt} + + {getStatusIndicator()} + {isAnswered && ( + {getAnswerSummary()} + )} + {isSkipped && ( + Skipped + )} +
+
+
+
+ ); +} diff --git a/apps/web/components/task/task-changes-panel.tsx b/apps/web/components/task/task-changes-panel.tsx index f64d19237f..f26fdf64d4 100644 --- a/apps/web/components/task/task-changes-panel.tsx +++ b/apps/web/components/task/task-changes-panel.tsx @@ -246,7 +246,8 @@ const TaskChangesPanel = memo(function TaskChangesPanel({ {isSingleDiffSelected && selectedDiffContent ? ( // Render single file diff (from commit, snapshot, or uncommitted) -
+ // Key includes content length to force re-render when diff updates +
{ if (!file.diff) return null; const fileLang = getFileLang(file.path); + // Include diff length in key to force re-render when content changes + const diffKey = `uncommitted-${file.path}-${file.diff.length}`; return ( -
+
{file.path} @@ -312,8 +315,10 @@ const TaskChangesPanel = memo(function TaskChangesPanel({ {committedFiles.map((file) => { if (!file.diff) return null; const fileLang = getFileLang(file.path); + // Include diff length in key to force re-render when content changes + const diffKey = `committed-${file.path}-${file.diff.length}`; return ( -
+
{file.path} diff --git a/apps/web/components/task/task-chat-panel.tsx b/apps/web/components/task/task-chat-panel.tsx index ba74a27ada..6010cbbe64 100644 --- a/apps/web/components/task/task-chat-panel.tsx +++ b/apps/web/components/task/task-chat-panel.tsx @@ -46,13 +46,20 @@ export function TaskChatPanel({ const { messages, isLoading: messagesLoading } = useSessionMessages(resolvedSessionId); // Process messages (filtering, todos, etc.) - const { allMessages, permissionsByToolCallId, todoItems, agentMessageCount } = useProcessedMessages( + const { allMessages, permissionsByToolCallId, todoItems, agentMessageCount, pendingClarification } = useProcessedMessages( messages, taskId, resolvedSessionId, taskDescription ); + // Track clarification resolved state to trigger re-render after submit + const [clarificationKey, setClarificationKey] = useState(0); + const handleClarificationResolved = useCallback(() => { + // Force re-render which will update messages from backend + setClarificationKey((k) => k + 1); + }, []); + // Model management const { sessionModel, activeModel } = useSessionModel( resolvedSessionId, @@ -165,6 +172,7 @@ export function TaskChatPanel({
{todoItems.length > 0 && }
diff --git a/apps/web/hooks/domains/session/use-cumulative-diff.ts b/apps/web/hooks/domains/session/use-cumulative-diff.ts index 8a65a86cd2..413103d53d 100644 --- a/apps/web/hooks/domains/session/use-cumulative-diff.ts +++ b/apps/web/hooks/domains/session/use-cumulative-diff.ts @@ -1,10 +1,21 @@ -import { useEffect, useCallback, useState } from 'react'; +import { useEffect, useCallback, useState, useRef } from 'react'; import { getWebSocketClient } from '@/lib/ws/connection'; import type { CumulativeDiff } from '@/lib/state/slices/session-runtime/types'; // Local state cache for cumulative diffs (not stored in global store since it's on-demand) const cumulativeDiffCache: Record = {}; const loadingState: Record = {}; +// Version counter to trigger refetch when cache is invalidated +const cacheVersion: Record = {}; + +/** + * Invalidate the cumulative diff cache for a session. + * Call this when git status changes (commits, snapshots, etc.) + */ +export function invalidateCumulativeDiffCache(sessionId: string) { + delete cumulativeDiffCache[sessionId]; + cacheVersion[sessionId] = (cacheVersion[sessionId] ?? 0) + 1; +} /** * Hook to fetch cumulative diff for a session. @@ -18,6 +29,8 @@ export function useCumulativeDiff(sessionId: string | null) { sessionId ? loadingState[sessionId] ?? false : false ); const [error, setError] = useState(null); + // Track the version we last fetched for + const lastFetchedVersionRef = useRef(null); const fetchCumulativeDiff = useCallback(async () => { if (!sessionId) return; @@ -38,6 +51,7 @@ export function useCumulativeDiff(sessionId: string | null) { if (response?.cumulative_diff) { cumulativeDiffCache[sessionId] = response.cumulative_diff; setDiff(response.cumulative_diff); + lastFetchedVersionRef.current = cacheVersion[sessionId] ?? 0; } } catch (err) { console.error('Failed to fetch cumulative diff:', err); @@ -48,21 +62,44 @@ export function useCumulativeDiff(sessionId: string | null) { } }, [sessionId]); - // Fetch on mount if not cached + // Fetch on mount if not cached, or refetch if cache was invalidated useEffect(() => { - if (sessionId && !diff && !loading) { + if (!sessionId) return; + + const currentVersion = cacheVersion[sessionId] ?? 0; + const cachedDiff = cumulativeDiffCache[sessionId]; + const needsFetch = !cachedDiff || lastFetchedVersionRef.current !== currentVersion; + + if (needsFetch && !loading) { fetchCumulativeDiff(); } - }, [sessionId, diff, loading, fetchCumulativeDiff]); + }, [sessionId, loading, fetchCumulativeDiff]); + + // Poll for cache invalidation (check every 500ms if cache was invalidated) + useEffect(() => { + if (!sessionId) return; + + const interval = setInterval(() => { + const currentVersion = cacheVersion[sessionId] ?? 0; + if (lastFetchedVersionRef.current !== currentVersion && !loadingState[sessionId]) { + // Cache was invalidated, clear local state to trigger refetch + setDiff(null); + } + }, 500); + + return () => clearInterval(interval); + }, [sessionId]); // Update local state when sessionId changes useEffect(() => { if (sessionId) { setDiff(cumulativeDiffCache[sessionId] ?? null); setLoading(loadingState[sessionId] ?? false); + lastFetchedVersionRef.current = cacheVersion[sessionId] ?? 0; } else { setDiff(null); setLoading(false); + lastFetchedVersionRef.current = null; } }, [sessionId]); diff --git a/apps/web/hooks/use-processed-messages.ts b/apps/web/hooks/use-processed-messages.ts index 271b185767..4796295e70 100644 --- a/apps/web/hooks/use-processed-messages.ts +++ b/apps/web/hooks/use-processed-messages.ts @@ -1,5 +1,5 @@ import { useMemo } from 'react'; -import type { Message } from '@/lib/types/http'; +import type { Message, ClarificationRequestMetadata } from '@/lib/types/http'; export function useProcessedMessages( messages: Message[], @@ -35,11 +35,36 @@ export function useProcessedMessages( return map; }, [messages]); + // Find pending clarification request (the most recent one that is still pending) + // This will be rendered in the input area, not in the message list + const pendingClarification = useMemo(() => { + for (let i = messages.length - 1; i >= 0; i--) { + const message = messages[i]; + if (message.type === 'clarification_request') { + const metadata = message.metadata as ClarificationRequestMetadata | undefined; + // A clarification is pending if it has no status or status is 'pending' + if (!metadata?.status || metadata.status === 'pending') { + return message; + } + } + } + return null; + }, [messages]); + // Filter to only show chat-relevant types. // 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. + // Pending clarification requests are also hidden (shown in input area instead). const visibleMessages = useMemo(() => { return messages.filter((message) => { + // Hide pending clarification request (it's shown in input area) + if (message.type === 'clarification_request') { + const metadata = message.metadata as ClarificationRequestMetadata | undefined; + const isPending = !metadata?.status || metadata.status === 'pending'; + // Only show clarification requests that are resolved (not pending) + return !isPending; + } + // Standard visible types if ( !message.type || @@ -112,5 +137,6 @@ export function useProcessedMessages( permissionsByToolCallId, todoItems, agentMessageCount, + pendingClarification, }; } diff --git a/apps/web/lib/types/http.ts b/apps/web/lib/types/http.ts index ec94e9dc76..b858ce9b85 100644 --- a/apps/web/lib/types/http.ts +++ b/apps/web/lib/types/http.ts @@ -467,6 +467,7 @@ export type MessageType = | 'thinking' | 'todo' | 'permission_request' + | 'clarification_request' | 'script_execution'; export type Message = { @@ -625,3 +626,40 @@ export type ListAvailableAgentsResponse = { agents: AvailableAgent[]; total: number; }; + +// Clarification request types (for ask_user_question feature) +export type ClarificationOption = { + option_id: string; + label: string; + description: string; +}; + +export type ClarificationQuestion = { + id: string; + title: string; + prompt: string; + options: ClarificationOption[]; +}; + +export type ClarificationRequestMetadata = { + pending_id: string; + session_id: string; + task_id?: string; + question: ClarificationQuestion; + context?: string; + status?: 'pending' | 'answered' | 'rejected' | 'expired'; + response?: ClarificationAnswer; +}; + +export type ClarificationAnswer = { + question_id: string; + selected_options?: string[]; + custom_text?: string; +}; + +export type ClarificationResponse = { + pending_id: string; + answers: ClarificationAnswer[]; + rejected?: boolean; + reject_reason?: string; +}; diff --git a/apps/web/lib/ws/handlers/git-status.ts b/apps/web/lib/ws/handlers/git-status.ts index 0ffc9ba133..cadd557618 100644 --- a/apps/web/lib/ws/handlers/git-status.ts +++ b/apps/web/lib/ws/handlers/git-status.ts @@ -8,6 +8,7 @@ import type { GitCommitsResetEvent, GitSnapshotCreatedEvent, } from '@/lib/types/git-events'; +import { invalidateCumulativeDiffCache } from '@/hooks/domains/session/use-cumulative-diff'; // Handler functions for each event type type GitEventHandlers = { @@ -32,6 +33,8 @@ const gitEventHandlers: GitEventHandlers = { files: event.status.files, timestamp: event.timestamp, }); + // Invalidate cumulative diff cache when files change + invalidateCumulativeDiffCache(event.session_id); }, commit_created: (store, event) => { @@ -49,11 +52,15 @@ const gitEventHandlers: GitEventHandlers = { committed_at: event.commit.committed_at, created_at: event.commit.created_at ?? event.timestamp, }); + // Invalidate cumulative diff cache when new commit is created + invalidateCumulativeDiffCache(event.session_id); }, commits_reset: (store, event) => { // Clear commits to trigger refetch in useSessionCommits hook store.getState().clearSessionCommits(event.session_id); + // Invalidate cumulative diff cache when commits are reset + invalidateCumulativeDiffCache(event.session_id); }, snapshot_created: (store, event) => { @@ -71,6 +78,8 @@ const gitEventHandlers: GitEventHandlers = { triggered_by: event.snapshot.triggered_by, created_at: event.snapshot.created_at, }); + // Invalidate cumulative diff cache when snapshot is created + invalidateCumulativeDiffCache(event.session_id); }, };