From 8dd88e8c601456add1fc24b71329fd797e0a9dfd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Wed, 21 Jan 2026 19:49:36 +0000 Subject: [PATCH] refactor: comprehensive code quality improvements (#69) --- apps/backend/cmd/kandev/adapters.go | 17 +++ .../internal/agent/lifecycle/event_types.go | 6 + .../internal/agent/lifecycle/events.go | 18 +++ .../internal/agent/lifecycle/manager.go | 119 ++++++++++++++++- .../internal/agent/lifecycle/session.go | 41 +++++- .../internal/agent/lifecycle/streams.go | 6 +- .../backend/internal/agent/lifecycle/types.go | 4 + .../internal/agentctl/server/process/git.go | 35 +++++ .../server/process/workspace_tracker.go | 16 ++- .../orchestrator/controller/errors.go | 5 - .../orchestrator/executor/executor.go | 4 +- apps/backend/internal/orchestrator/service.go | 121 ++++++++++++------ apps/backend/internal/task/service/service.go | 84 ++++++++++++ apps/web/components/task-preview-panel.tsx | 7 - .../task/chat/messages/chat-message.tsx | 34 +++-- apps/web/components/task/editors-menu.tsx | 3 - .../web/components/task/task-center-panel.tsx | 7 +- apps/web/components/task/task-chat-panel.tsx | 14 +- apps/web/hooks/use-session-agent.ts | 19 +-- apps/web/hooks/use-session-messages.ts | 9 -- apps/web/hooks/use-session-resumption.ts | 8 -- apps/web/hooks/use-task-chat-session.ts | 4 +- apps/web/hooks/use-task-session.ts | 3 +- apps/web/hooks/use-task-sessions.ts | 3 +- apps/web/lib/state/store.ts | 20 +-- apps/web/lib/ws/client.ts | 22 ---- apps/web/lib/ws/handlers/git-status.ts | 8 -- apps/web/lib/ws/utils.ts | 7 - 28 files changed, 463 insertions(+), 181 deletions(-) delete mode 100644 apps/backend/internal/orchestrator/controller/errors.go diff --git a/apps/backend/cmd/kandev/adapters.go b/apps/backend/cmd/kandev/adapters.go index 8d4a7316a4..8812e4c835 100644 --- a/apps/backend/cmd/kandev/adapters.go +++ b/apps/backend/cmd/kandev/adapters.go @@ -359,3 +359,20 @@ func (a *messageCreatorAdapter) CreatePermissionRequestMessage(ctx context.Conte func (a *messageCreatorAdapter) UpdatePermissionMessage(ctx context.Context, sessionID, pendingID, status string) error { return a.svc.UpdatePermissionMessage(ctx, sessionID, pendingID, status) } + +// 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 string) error { + _, err := a.svc.CreateMessageWithID(ctx, messageID, &taskservice.CreateMessageRequest{ + TaskSessionID: agentSessionID, + TaskID: taskID, + Content: content, + AuthorType: "agent", + }) + return err +} + +// AppendAgentMessage appends additional content to an existing streaming message. +func (a *messageCreatorAdapter) AppendAgentMessage(ctx context.Context, messageID, additionalContent string) error { + return a.svc.AppendMessageContent(ctx, messageID, additionalContent) +} diff --git a/apps/backend/internal/agent/lifecycle/event_types.go b/apps/backend/internal/agent/lifecycle/event_types.go index 84410bde59..b9e3f04f75 100644 --- a/apps/backend/internal/agent/lifecycle/event_types.go +++ b/apps/backend/internal/agent/lifecycle/event_types.go @@ -46,6 +46,12 @@ type AgentStreamEventData struct { Error string `json:"error,omitempty"` SessionStatus string `json:"session_status,omitempty"` // "resumed" or "new" for session_status events Data interface{} `json:"data,omitempty"` + + // Streaming message fields (for "message_streaming" type) + // MessageID is the ID of the message being streamed (empty for first chunk, set for appends) + MessageID string `json:"message_id,omitempty"` + // IsAppend indicates whether this is an append to an existing message (true) or a new message (false) + IsAppend bool `json:"is_append,omitempty"` } // AgentStreamEventPayload is the payload for agent stream events (WebSocket streaming). diff --git a/apps/backend/internal/agent/lifecycle/events.go b/apps/backend/internal/agent/lifecycle/events.go index 066871252d..cb7f31816e 100644 --- a/apps/backend/internal/agent/lifecycle/events.go +++ b/apps/backend/internal/agent/lifecycle/events.go @@ -152,6 +152,24 @@ func (p *EventPublisher) PublishAgentStreamEvent(execution *AgentExecution, even } } +// PublishAgentStreamEventPayload publishes a pre-built agent stream event payload. +// This is used for streaming message events where the payload is constructed by the caller. +func (p *EventPublisher) PublishAgentStreamEventPayload(payload *AgentStreamEventPayload) { + if p.eventBus == nil { + return + } + + busEvent := bus.NewEvent(events.AgentStream, "agent-manager", *payload) + subject := events.BuildAgentStreamSubject(payload.SessionID) + + if err := p.eventBus.Publish(context.Background(), subject, busEvent); err != nil { + p.logger.Error("failed to publish agent stream event payload", + zap.String("task_id", payload.TaskID), + zap.String("session_id", payload.SessionID), + zap.Error(err)) + } +} + // PublishGitStatus publishes a git status update to the event bus. func (p *EventPublisher) PublishGitStatus(execution *AgentExecution, update *agentctl.GitStatusUpdate) { if p.eventBus == nil { diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index a32d344c26..c35452e72d 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -870,11 +870,31 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age case "message_chunk": m.updateExecutionProgress(execution.ID, 50) - // Accumulate message content for saving as comment when a step completes + // Accumulate message content and stream on newline boundaries for real-time feedback if event.Text != "" { execution.messageMu.Lock() execution.messageBuffer.WriteString(event.Text) - execution.messageMu.Unlock() + + // Check if buffer contains newlines - if so, flush up to the last newline + bufContent := execution.messageBuffer.String() + lastNewline := strings.LastIndex(bufContent, "\n") + if lastNewline != -1 { + // Extract content up to and including the last newline + toFlush := bufContent[:lastNewline+1] + remainder := bufContent[lastNewline+1:] + + // Reset buffer with remainder + execution.messageBuffer.Reset() + execution.messageBuffer.WriteString(remainder) + execution.messageMu.Unlock() + + // Publish streaming message event if there's content to flush + if strings.TrimSpace(toFlush) != "" { + m.publishStreamingMessage(execution, toFlush) + } + } else { + execution.messageMu.Unlock() + } } case "reasoning": @@ -917,6 +937,9 @@ func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.Age zap.String("execution_id", execution.ID)) case "error": + // Flush any accumulated content and clear streaming state on error + m.flushMessageBuffer(execution) + m.logger.Error("agent error", zap.String("execution_id", execution.ID), zap.String("error", event.Error)) @@ -1009,17 +1032,99 @@ func (m *Manager) handleShellExit(execution *AgentExecution, code int) { m.eventPublisher.PublishShellExit(execution, code) } +// publishStreamingMessage publishes a streaming message event for real-time text updates. +// It creates a new message on first call (currentMessageID empty) or appends to existing. +// The message ID is generated and set synchronously to avoid race conditions. +func (m *Manager) publishStreamingMessage(execution *AgentExecution, content string) { + execution.messageMu.Lock() + isAppend := execution.currentMessageID != "" + messageID := execution.currentMessageID + + // If this is the first chunk of a new message segment, generate the ID now + if !isAppend { + messageID = uuid.New().String() + execution.currentMessageID = messageID + } + execution.messageMu.Unlock() + + event := AgentStreamEventData{ + Type: "message_streaming", + Text: content, + MessageID: messageID, + IsAppend: isAppend, + } + + // Create payload manually to include streaming-specific fields + payload := &AgentStreamEventPayload{ + Type: "agent/event", + Timestamp: time.Now().UTC().Format(time.RFC3339Nano), + AgentID: execution.ID, + TaskID: execution.TaskID, + SessionID: execution.SessionID, + Data: &event, + } + + m.logger.Debug("publishing streaming message", + zap.String("execution_id", execution.ID), + zap.String("message_id", messageID), + zap.Bool("is_append", isAppend), + zap.Int("content_length", len(content))) + + // Publish the streaming event - orchestrator will handle create/append logic + m.eventPublisher.PublishAgentStreamEventPayload(payload) +} + // flushMessageBuffer extracts any accumulated message from the buffer and returns it. // This is called when a tool use starts or on complete to get the agent's response. +// It also clears the currentMessageID to start fresh for the next message segment. func (m *Manager) flushMessageBuffer(execution *AgentExecution) string { execution.messageMu.Lock() agentMessage := execution.messageBuffer.String() execution.messageBuffer.Reset() execution.reasoningBuffer.Reset() execution.summaryBuffer.Reset() + // Clear the streaming message ID so next segment starts fresh + currentMsgID := execution.currentMessageID + execution.currentMessageID = "" execution.messageMu.Unlock() - return strings.TrimSpace(agentMessage) + // If we have remaining content and an active streaming message, append it + trimmedMessage := strings.TrimSpace(agentMessage) + if trimmedMessage != "" && currentMsgID != "" { + // Publish final append to the streaming message + m.publishStreamingMessageFinal(execution, currentMsgID, trimmedMessage) + // Return empty since we've already handled it via streaming + return "" + } + + return trimmedMessage +} + +// publishStreamingMessageFinal publishes the final chunk of a streaming message. +// This is called during flush to append any remaining buffered content. +func (m *Manager) publishStreamingMessageFinal(execution *AgentExecution, messageID, content string) { + event := AgentStreamEventData{ + Type: "message_streaming", + Text: content, + MessageID: messageID, + IsAppend: true, + } + + payload := &AgentStreamEventPayload{ + Type: "agent/event", + Timestamp: time.Now().UTC().Format(time.RFC3339Nano), + AgentID: execution.ID, + TaskID: execution.TaskID, + SessionID: execution.SessionID, + Data: &event, + } + + m.logger.Debug("publishing final streaming message chunk", + zap.String("execution_id", execution.ID), + zap.String("message_id", messageID), + zap.Int("content_length", len(content))) + + m.eventPublisher.PublishAgentStreamEventPayload(payload) } // updateExecutionProgress updates an execution's progress @@ -1065,6 +1170,14 @@ func (m *Manager) CancelAgent(ctx context.Context, executionID string) error { return fmt.Errorf("failed to cancel agent: %w", err) } + // Clear streaming state after cancel to ensure clean state for next prompt + execution.messageMu.Lock() + execution.messageBuffer.Reset() + execution.reasoningBuffer.Reset() + execution.summaryBuffer.Reset() + execution.currentMessageID = "" + execution.messageMu.Unlock() + // Mark as ready for follow-up prompts after successful cancel if err := m.MarkReady(executionID); err != nil { m.logger.Warn("failed to mark execution as ready after cancel", diff --git a/apps/backend/internal/agent/lifecycle/session.go b/apps/backend/internal/agent/lifecycle/session.go index 879c9ab309..6f86afd191 100644 --- a/apps/backend/internal/agent/lifecycle/session.go +++ b/apps/backend/internal/agent/lifecycle/session.go @@ -291,11 +291,13 @@ func (sm *SessionManager) SendPrompt( } } - // Clear buffers before starting prompt + // Clear buffers and streaming state before starting prompt + // This ensures each prompt starts fresh and doesn't append to previous message execution.messageMu.Lock() execution.messageBuffer.Reset() execution.reasoningBuffer.Reset() execution.summaryBuffer.Reset() + execution.currentMessageID = "" // Clear streaming message ID for new turn execution.messageMu.Unlock() sm.logger.Info("sending prompt to agent", @@ -311,12 +313,14 @@ func (sm *SessionManager) SendPrompt( return nil, fmt.Errorf("prompt failed: %w", err) } - // Extract accumulated content from buffers + // Extract accumulated content from buffers and handle streaming message completion execution.messageMu.Lock() agentMessage := execution.messageBuffer.String() execution.messageBuffer.Reset() execution.reasoningBuffer.Reset() execution.summaryBuffer.Reset() + currentMsgID := execution.currentMessageID + execution.currentMessageID = "" // Clear for next turn execution.messageMu.Unlock() stopReason := "" @@ -324,19 +328,42 @@ func (sm *SessionManager) SendPrompt( stopReason = string(resp.StopReason) } + trimmedMessage := strings.TrimSpace(agentMessage) + sm.logger.Info("ACP prompt completed", zap.String("execution_id", execution.ID), zap.String("stop_reason", stopReason), - zap.Int("message_length", len(agentMessage))) + zap.Int("message_length", len(trimmedMessage)), + zap.Bool("has_streaming_msg", currentMsgID != "")) - // Publish complete event with accumulated agent message - // This is needed for ACP where Prompt() is synchronous and there's no - // separate "complete" notification from the agent + // Handle remaining content from the buffer if sm.eventPublisher != nil { + // If there's remaining content and an active streaming message, + // append it to the streaming message instead of creating a new one + if trimmedMessage != "" && currentMsgID != "" { + // Publish streaming append with the remaining content + sm.eventPublisher.PublishAgentStreamEventPayload(&AgentStreamEventPayload{ + Type: "agent/event", + Timestamp: time.Now().UTC().Format(time.RFC3339), + AgentID: execution.AgentProfileID, + TaskID: execution.TaskID, + SessionID: execution.SessionID, + Data: &AgentStreamEventData{ + Type: "message_streaming", + MessageID: currentMsgID, + IsAppend: true, + Text: trimmedMessage, + }, + }) + // Content was handled via streaming, publish complete with empty text + trimmedMessage = "" + } + + // Publish complete event (with remaining text only if no streaming message was active) completeEvent := streams.AgentEvent{ Type: streams.EventTypeComplete, SessionID: execution.ACPSessionID, - Text: strings.TrimSpace(agentMessage), + Text: trimmedMessage, } sm.eventPublisher.PublishAgentStreamEvent(execution, completeEvent) } diff --git a/apps/backend/internal/agent/lifecycle/streams.go b/apps/backend/internal/agent/lifecycle/streams.go index e55b0c20fb..af69aa1c0a 100644 --- a/apps/backend/internal/agent/lifecycle/streams.go +++ b/apps/backend/internal/agent/lifecycle/streams.go @@ -43,7 +43,7 @@ func (sm *StreamManager) ConnectAll(execution *AgentExecution, ready chan<- stru // ReconnectAll reconnects to all streams (used after backend restart). // This waits for agentctl to be ready before connecting to streams. func (sm *StreamManager) ReconnectAll(execution *AgentExecution) { - sm.logger.Info("reconnecting to agent streams after recovery", + sm.logger.Debug("reconnecting to agent streams after recovery", zap.String("instance_id", execution.ID), zap.String("task_id", execution.TaskID)) @@ -64,7 +64,7 @@ func (sm *StreamManager) ReconnectAll(execution *AgentExecution) { // Reconnect to WebSocket streams sm.ConnectAll(execution, nil) - sm.logger.Info("agent streams reconnected", + sm.logger.Debug("agent streams reconnected", zap.String("instance_id", execution.ID), zap.String("task_id", execution.TaskID)) } @@ -151,7 +151,7 @@ func (sm *StreamManager) connectWorkspaceStream(execution *AgentExecution) { // Store the workspace stream on the execution for shell I/O execution.SetWorkspaceStream(ws) - sm.logger.Info("connected to unified workspace stream", + sm.logger.Debug("connected to unified workspace stream", zap.String("instance_id", execution.ID)) // Wait for the stream to close (it stays open until disconnected) diff --git a/apps/backend/internal/agent/lifecycle/types.go b/apps/backend/internal/agent/lifecycle/types.go index 956ee00ae1..036d0b57e3 100644 --- a/apps/backend/internal/agent/lifecycle/types.go +++ b/apps/backend/internal/agent/lifecycle/types.go @@ -51,6 +51,10 @@ type AgentExecution struct { reasoningBuffer strings.Builder summaryBuffer strings.Builder messageMu sync.Mutex + + // Streaming message tracking - ID of the current in-progress message being streamed + // This is set when we create a streaming message and cleared on tool_call/complete + currentMessageID string } // GetAgentCtlClient returns the agentctl client for this execution diff --git a/apps/backend/internal/agentctl/server/process/git.go b/apps/backend/internal/agentctl/server/process/git.go index 84538d42d1..7b61ac6c37 100644 --- a/apps/backend/internal/agentctl/server/process/git.go +++ b/apps/backend/internal/agentctl/server/process/git.go @@ -9,6 +9,7 @@ import ( "os" "os/exec" "path/filepath" + "regexp" "strings" "sync" @@ -19,6 +20,30 @@ import ( // ErrOperationInProgress is returned when a git operation is already in progress. var ErrOperationInProgress = errors.New("git operation already in progress") +// ErrInvalidBranchName is returned when a branch name contains invalid characters. +var ErrInvalidBranchName = errors.New("invalid branch name") + +// validBranchNameRegex matches safe git branch names. +// Allows alphanumeric, hyphens, underscores, slashes, and dots. +// Disallows: spaces, shell metacharacters, and control characters. +var validBranchNameRegex = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._/-]*$`) + +// isValidBranchName validates that a branch name is safe to use in git commands. +func isValidBranchName(branch string) bool { + if branch == "" || len(branch) > 255 { + return false + } + // Disallow ".." to prevent path traversal + if strings.Contains(branch, "..") { + return false + } + // Disallow ending with ".lock" + if strings.HasSuffix(branch, ".lock") { + return false + } + return validBranchNameRegex.MatchString(branch) +} + // GitOperationResult represents the result of a git operation. type GitOperationResult struct { Success bool `json:"success"` @@ -237,6 +262,11 @@ func (g *GitOperator) Push(ctx context.Context, force bool, setUpstream bool) (* // Rebase performs a git rebase onto the specified base branch. func (g *GitOperator) Rebase(ctx context.Context, baseBranch string) (*GitOperationResult, error) { + // Validate branch name to prevent command injection + if !isValidBranchName(baseBranch) { + return nil, ErrInvalidBranchName + } + if !g.tryLock("rebase") { return nil, ErrOperationInProgress } @@ -279,6 +309,11 @@ func (g *GitOperator) Rebase(ctx context.Context, baseBranch string) (*GitOperat // Merge performs a git merge of the specified base branch. func (g *GitOperator) Merge(ctx context.Context, baseBranch string) (*GitOperationResult, error) { + // Validate branch name to prevent command injection + if !isValidBranchName(baseBranch) { + return nil, ErrInvalidBranchName + } + if !g.tryLock("merge") { return nil, ErrOperationInProgress } diff --git a/apps/backend/internal/agentctl/server/process/workspace_tracker.go b/apps/backend/internal/agentctl/server/process/workspace_tracker.go index 5b2747bf44..cf650e23d6 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_tracker.go +++ b/apps/backend/internal/agentctl/server/process/workspace_tracker.go @@ -641,8 +641,12 @@ func (wt *WorkspaceTracker) addDirectoryRecursive(dir string) error { // GetFileTree returns the file tree for a given path and depth func (wt *WorkspaceTracker) GetFileTree(reqPath string, depth int) (*types.FileTreeNode, error) { - // Resolve the full path - fullPath := filepath.Join(wt.workDir, reqPath) + // Resolve the full path with path traversal protection + fullPath := filepath.Join(wt.workDir, filepath.Clean(reqPath)) + cleanWorkDir := filepath.Clean(wt.workDir) + if !strings.HasPrefix(fullPath, cleanWorkDir+string(os.PathSeparator)) && fullPath != cleanWorkDir { + return nil, fmt.Errorf("path traversal detected") + } // Check if path exists info, err := os.Stat(fullPath) @@ -712,8 +716,12 @@ func (wt *WorkspaceTracker) buildFileTreeNode(fullPath, relPath string, info os. // GetFileContent returns the content of a file func (wt *WorkspaceTracker) GetFileContent(reqPath string) (string, int64, error) { - // Resolve the full path - fullPath := filepath.Join(wt.workDir, reqPath) + // Resolve the full path with path traversal protection + fullPath := filepath.Join(wt.workDir, filepath.Clean(reqPath)) + cleanWorkDir := filepath.Clean(wt.workDir) + if !strings.HasPrefix(fullPath, cleanWorkDir+string(os.PathSeparator)) && fullPath != cleanWorkDir { + return "", 0, fmt.Errorf("path traversal detected") + } // Check if file exists and is a regular file info, err := os.Stat(fullPath) diff --git a/apps/backend/internal/orchestrator/controller/errors.go b/apps/backend/internal/orchestrator/controller/errors.go deleted file mode 100644 index 4137808535..0000000000 --- a/apps/backend/internal/orchestrator/controller/errors.go +++ /dev/null @@ -1,5 +0,0 @@ -package controller - -// This file is reserved for controller-specific errors. -// Currently no custom errors are defined. - diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index cee2b16a3b..8781e8dc55 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -814,7 +814,7 @@ func (e *Executor) Prompt(ctx context.Context, taskID, sessionID string, prompt return nil, ErrExecutionNotFound } - e.logger.Info("sending prompt to agent", + e.logger.Debug("sending prompt to agent", zap.String("task_id", taskID), zap.String("session_id", sessionID), zap.String("agent_execution_id", session.AgentExecutionID), @@ -970,7 +970,7 @@ func (e *Executor) SwitchModel(ctx context.Context, taskID, sessionID, newModel, // RespondToPermission sends a response to a permission request for a session func (e *Executor) RespondToPermission(ctx context.Context, sessionID, pendingID, optionID string, cancelled bool) error { - e.logger.Info("responding to permission request", + e.logger.Debug("responding to permission request", zap.String("session_id", sessionID), zap.String("pending_id", pendingID), zap.String("option_id", optionID), diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index b59b2d3010..4fe76eec36 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -56,6 +56,10 @@ type MessageCreator interface { 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 + // CreateAgentMessageStreaming creates a new agent message with a pre-generated ID for streaming updates + CreateAgentMessageStreaming(ctx context.Context, messageID, taskID, content, agentSessionID string) error + // AppendAgentMessage appends additional content to an existing streaming message + AppendAgentMessage(ctx context.Context, messageID, additionalContent string) error } // Service is the main orchestrator service @@ -124,9 +128,9 @@ func NewService( // Create the service (watcher will be created after we have handlers) s := &Service{ - config: cfg, - logger: svcLogger, - eventBus: eventBus, + config: cfg, + logger: svcLogger, + eventBus: eventBus, taskRepo: taskRepo, repo: repo, agentManager: agentManager, @@ -403,7 +407,7 @@ func (s *Service) GetStatus() *Status { // EnqueueTask manually adds a task to the queue func (s *Service) EnqueueTask(ctx context.Context, task *v1.Task) error { - s.logger.Info("manually enqueueing task", + s.logger.Debug("manually enqueueing task", zap.String("task_id", task.ID), zap.String("title", task.Title)) return s.scheduler.EnqueueTask(task) @@ -411,7 +415,7 @@ func (s *Service) EnqueueTask(ctx context.Context, task *v1.Task) error { // StartTask manually starts agent execution for a task func (s *Service) StartTask(ctx context.Context, taskID string, agentProfileID string, priority int, prompt string) (*executor.TaskExecution, error) { - s.logger.Info("manually starting task", + s.logger.Debug("manually starting task", zap.String("task_id", taskID), zap.String("agent_profile_id", agentProfileID), zap.Int("priority", priority), @@ -471,7 +475,7 @@ func (s *Service) StartTask(ctx context.Context, taskID string, agentProfileID s // ResumeTaskSession restarts a specific task session using its stored worktree. func (s *Service) ResumeTaskSession(ctx context.Context, taskID, sessionID string) (*executor.TaskExecution, error) { - s.logger.Info("resuming task session", + s.logger.Debug("resuming task session", zap.String("task_id", taskID), zap.String("session_id", sessionID)) @@ -504,7 +508,7 @@ func (s *Service) ResumeTaskSession(ctx context.Context, taskID, sessionID strin // Preserve persisted task/session state; resume should not mutate state/columns. execution.SessionState = v1.TaskSessionState(session.State) - s.logger.Info("task session resumed and ready for input", + s.logger.Debug("task session resumed and ready for input", zap.String("task_id", taskID), zap.String("session_id", sessionID)) @@ -513,7 +517,7 @@ func (s *Service) ResumeTaskSession(ctx context.Context, taskID, sessionID strin // GetTaskSessionStatus returns the status of a task session including whether it's resumable func (s *Service) GetTaskSessionStatus(ctx context.Context, taskID, sessionID string) (dto.TaskSessionStatusResponse, error) { - s.logger.Info("checking task session status", + s.logger.Debug("checking task session status", zap.String("task_id", taskID), zap.String("session_id", sessionID)) @@ -622,7 +626,7 @@ type PromptResult struct { // PromptTask sends a follow-up prompt to a running agent for a task session. func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prompt string, model string) (*PromptResult, error) { - s.logger.Info("sending prompt to task agent", + s.logger.Debug("sending prompt to task agent", zap.String("task_id", taskID), zap.String("session_id", sessionID), zap.Int("prompt_length", len(prompt))) @@ -642,7 +646,7 @@ func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prom if session.AgentProfileSnapshot != nil { if currentModel, ok := session.AgentProfileSnapshot["model"].(string); ok && currentModel != model { // Model switch needed - delegate to executor - s.logger.Info("model switch requested", + s.logger.Debug("model switch requested", zap.String("task_id", taskID), zap.String("session_id", sessionID), zap.String("current_model", currentModel), @@ -661,7 +665,7 @@ func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prom } } - s.setSessionRunning(ctx, taskID, sessionID, nil) + s.setSessionRunning(ctx, taskID, sessionID) result, err := s.executor.Prompt(ctx, taskID, sessionID, prompt) if err != nil { return nil, err @@ -674,7 +678,7 @@ func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prom // RespondToPermission sends a response to a permission request for a session func (s *Service) RespondToPermission(ctx context.Context, sessionID, pendingID, optionID string, cancelled bool) error { - s.logger.Info("responding to permission request", + s.logger.Debug("responding to permission request", zap.String("session_id", sessionID), zap.String("pending_id", pendingID), zap.String("option_id", optionID), @@ -711,7 +715,7 @@ func (s *Service) RespondToPermission(ctx context.Context, sessionID, pendingID, zap.Error(err)) return nil } - s.setSessionRunning(ctx, session.TaskID, sessionID, nil) + s.setSessionRunning(ctx, session.TaskID, sessionID) } return nil @@ -720,7 +724,7 @@ func (s *Service) RespondToPermission(ctx context.Context, sessionID, pendingID, // CancelAgent interrupts the current agent turn without terminating the process, // allowing the user to send a new prompt. func (s *Service) CancelAgent(ctx context.Context, sessionID string) error { - s.logger.Info("cancelling agent turn", zap.String("session_id", sessionID)) + s.logger.Debug("cancelling agent turn", zap.String("session_id", sessionID)) // Fetch session for state updates and message creation session, err := s.repo.GetTaskSession(ctx, sessionID) @@ -760,7 +764,7 @@ func (s *Service) CancelAgent(ctx context.Context, sessionID string) error { } } - s.logger.Info("agent turn cancelled", zap.String("session_id", sessionID)) + s.logger.Debug("agent turn cancelled", zap.String("session_id", sessionID)) return nil } @@ -805,20 +809,7 @@ func (s *Service) RegisterStreamHandler(handler func(payload *lifecycle.AgentStr s.logger.Debug("registered stream handler", zap.Int("total_handlers", len(s.streamHandlers))) } -// UnregisterStreamHandler removes a stream handler -func (s *Service) UnregisterStreamHandler(handler func(payload *lifecycle.AgentStreamEventPayload)) { - s.streamMu.Lock() - defer s.streamMu.Unlock() - // Find and remove the handler by comparing function pointers - // Note: This uses a simple approach - in production you might want to use handler IDs - for i := range s.streamHandlers { - // We can't directly compare functions, so we'll keep the handler in the list - // In a real implementation, you'd use unique handler IDs - _ = i - } - s.logger.Debug("unregister stream handler called") -} // broadcastStreamEvent broadcasts an agent stream event to all registered handlers func (s *Service) broadcastStreamEvent(payload *lifecycle.AgentStreamEventPayload) { @@ -913,7 +904,7 @@ func (s *Service) handleACPSessionCreated(ctx context.Context, data watcher.ACPS return } - s.logger.Info("stored resume token for task session", + s.logger.Debug("stored resume token for task session", zap.String("task_id", data.TaskID), zap.String("session_id", sessionID), zap.String("resume_token", data.ACPSessionID)) @@ -984,6 +975,10 @@ func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle // Handle different event types switch eventType { + case "message_streaming": + // Handle streaming message events (incremental text updates) + s.handleMessageStreamingEvent(ctx, payload) + case "tool_call": // Create tool call message when a tool call starts // If there's accumulated text, save it as an agent message first @@ -1077,7 +1072,7 @@ func (s *Service) handleToolCallEvent(ctx context.Context, payload *lifecycle.Ag zap.String("tool_call_id", payload.Data.ToolCallID), zap.Error(err)) } else { - s.logger.Info("created tool call message", + s.logger.Debug("created tool call message", zap.String("task_id", payload.TaskID), zap.String("tool_call_id", payload.Data.ToolCallID)) } @@ -1098,13 +1093,63 @@ func (s *Service) saveAgentTextIfPresent(ctx context.Context, payload *lifecycle zap.String("task_id", payload.TaskID), zap.Error(err)) } else { - s.logger.Info("created agent message", + s.logger.Debug("created agent message", zap.String("task_id", payload.TaskID), zap.Int("message_length", len(payload.Data.Text))) } } } +// handleMessageStreamingEvent handles streaming message events for real-time text updates. +// It creates a new message on first chunk (IsAppend=false) or appends to existing (IsAppend=true). +func (s *Service) handleMessageStreamingEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { + if payload.Data.Text == "" || payload.SessionID == "" { + return + } + + if s.messageCreator == nil { + return + } + + // MessageID is pre-generated by the manager to avoid race conditions + messageID := payload.Data.MessageID + if messageID == "" { + s.logger.Warn("streaming message event missing message ID", + zap.String("task_id", payload.TaskID), + zap.String("session_id", payload.SessionID)) + return + } + + if payload.Data.IsAppend { + // Append to existing message + if err := s.messageCreator.AppendAgentMessage(ctx, messageID, payload.Data.Text); err != nil { + s.logger.Error("failed to append to streaming message", + zap.String("task_id", payload.TaskID), + zap.String("message_id", messageID), + zap.Error(err)) + } else { + s.logger.Debug("appended to streaming message", + zap.String("task_id", payload.TaskID), + zap.String("message_id", messageID), + zap.Int("content_length", len(payload.Data.Text))) + } + } else { + // Create new streaming message with the pre-generated ID + if err := s.messageCreator.CreateAgentMessageStreaming(ctx, messageID, payload.TaskID, payload.Data.Text, payload.SessionID); err != nil { + s.logger.Error("failed to create streaming message", + zap.String("task_id", payload.TaskID), + zap.String("message_id", messageID), + zap.Error(err)) + } else { + s.logger.Debug("created streaming message", + zap.String("task_id", payload.TaskID), + zap.String("session_id", payload.SessionID), + zap.String("message_id", messageID), + zap.Int("content_length", len(payload.Data.Text))) + } + } +} + // handleToolUpdateEvent handles tool_update events and updates messages func (s *Service) handleToolUpdateEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) { if payload.SessionID == "" { @@ -1165,7 +1210,7 @@ func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID zap.String("state", string(nextState)), zap.Error(err)) } - s.logger.Info("task session state updated", + s.logger.Debug("task session state updated", zap.String("task_id", taskID), zap.String("session_id", sessionID), zap.String("old_state", string(oldState)), @@ -1181,7 +1226,7 @@ func (s *Service) updateTaskSessionState(ctx context.Context, taskID, sessionID } } -func (s *Service) setSessionWaitingForInput(ctx context.Context, taskID, sessionID string, _ *v1.TaskState) { +func (s *Service) setSessionWaitingForInput(ctx context.Context, taskID, sessionID string) { s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateWaitingForInput, "", false) if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateReview); err != nil { @@ -1194,7 +1239,7 @@ func (s *Service) setSessionWaitingForInput(ctx context.Context, taskID, session } } -func (s *Service) setSessionRunning(ctx context.Context, taskID, sessionID string, _ *v1.TaskState) { +func (s *Service) setSessionRunning(ctx context.Context, taskID, sessionID string) { s.updateTaskSessionState(ctx, taskID, sessionID, models.TaskSessionStateRunning, "", true) if err := s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateInProgress); err != nil { @@ -1215,7 +1260,7 @@ func (s *Service) finalizeAgentReady(taskID, sessionID string) { return } - s.setSessionWaitingForInput(ctx, taskID, sessionID, nil) + s.setSessionWaitingForInput(ctx, taskID, sessionID) } // handleGitStatusUpdated handles git status updates and persists them to agent session metadata @@ -1340,7 +1385,7 @@ func (s *Service) handleContextWindowUpdated(ctx context.Context, data watcher.C // 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", + s.logger.Debug("handling permission request", zap.String("task_id", data.TaskID), zap.String("pending_id", data.PendingID), zap.String("title", data.Title)) @@ -1352,7 +1397,7 @@ func (s *Service) handlePermissionRequest(ctx context.Context, data watcher.Perm return } - s.setSessionWaitingForInput(ctx, data.TaskID, data.TaskSessionID, nil) + s.setSessionWaitingForInput(ctx, data.TaskID, data.TaskSessionID) if s.messageCreator != nil { _, err := s.messageCreator.CreatePermissionRequestMessage( @@ -1372,7 +1417,7 @@ func (s *Service) handlePermissionRequest(ctx context.Context, data watcher.Perm zap.String("pending_id", data.PendingID), zap.Error(err)) } else { - s.logger.Info("created permission request message", + s.logger.Debug("created permission request message", zap.String("task_id", data.TaskID), zap.String("pending_id", data.PendingID)) } diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index 795a831e2b..9d9008a451 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -1563,6 +1563,58 @@ func (s *Service) CreateMessage(ctx context.Context, req *CreateMessageRequest) return message, nil } +// CreateMessageWithID creates a new message with a pre-generated ID. +// This is used for streaming messages where the ID is generated by the caller. +func (s *Service) CreateMessageWithID(ctx context.Context, id string, req *CreateMessageRequest) (*models.Message, error) { + session, err := s.repo.GetTaskSession(ctx, req.TaskSessionID) + if err != nil { + return nil, err + } + + authorType := models.MessageAuthorUser + if req.AuthorType == "agent" { + authorType = models.MessageAuthorAgent + } + + messageType := models.MessageType(req.Type) + if messageType == "" { + messageType = models.MessageTypeMessage + } + + taskID := req.TaskID + if taskID == "" && session != nil { + taskID = session.TaskID + } + + message := &models.Message{ + ID: id, + TaskSessionID: req.TaskSessionID, + TaskID: taskID, + AuthorType: authorType, + AuthorID: req.AuthorID, + Content: req.Content, + Type: messageType, + Metadata: req.Metadata, + RequestsInput: req.RequestsInput, + CreatedAt: time.Now().UTC(), + } + + if err := s.repo.CreateMessage(ctx, message); err != nil { + s.logger.Error("failed to create message with ID", zap.String("id", id), zap.Error(err)) + return nil, err + } + + // Publish message.added event + s.publishMessageEvent(ctx, events.MessageAdded, message) + + s.logger.Info("message created with ID", + zap.String("message_id", message.ID), + zap.String("session_id", message.TaskSessionID), + zap.String("author_type", string(message.AuthorType))) + + return message, nil +} + // GetMessage retrieves a message by ID func (s *Service) GetMessage(ctx context.Context, id string) (*models.Message, error) { return s.repo.GetMessage(ctx, id) @@ -1601,6 +1653,38 @@ func (s *Service) DeleteMessage(ctx context.Context, id string) error { return nil } +// AppendMessageContent appends additional content to an existing message. +// This is used for streaming agent responses where content arrives incrementally. +func (s *Service) AppendMessageContent(ctx context.Context, messageID, additionalContent string) error { + message, err := s.repo.GetMessage(ctx, messageID) + if err != nil { + s.logger.Error("failed to get message for append", + zap.String("message_id", messageID), + zap.Error(err)) + return err + } + + // Append the new content + message.Content = message.Content + additionalContent + + if err := s.repo.UpdateMessage(ctx, message); err != nil { + s.logger.Error("failed to append message content", + zap.String("message_id", messageID), + zap.Error(err)) + return err + } + + // Publish message.updated event for real-time streaming + s.publishMessageEvent(ctx, events.MessageUpdated, message) + + s.logger.Debug("message content appended", + zap.String("message_id", messageID), + zap.Int("appended_length", len(additionalContent)), + zap.Int("total_length", len(message.Content))) + + return nil +} + // UpdateToolCallMessage updates a tool call message's status. // It includes retry logic to handle race conditions where the complete event // may arrive before the message has been created by the start event. diff --git a/apps/web/components/task-preview-panel.tsx b/apps/web/components/task-preview-panel.tsx index 79ff652ea5..859529df1b 100644 --- a/apps/web/components/task-preview-panel.tsx +++ b/apps/web/components/task-preview-panel.tsx @@ -14,12 +14,6 @@ interface TaskPreviewPanelProps { onMaximize?: (task: Task, sessionId: string) => void; } -// TODO: fix this -const AGENTS = [ - { id: "codex", label: "Codex" }, - { id: "claude", label: "Claude Code" }, -]; - export function TaskPreviewPanel({ task, sessionId = null, onClose, onMaximize }: TaskPreviewPanelProps) { const { taskSessionId } = useTaskChatSession(task?.id ?? null); const activeSessionId = sessionId ?? taskSessionId; @@ -77,7 +71,6 @@ export function TaskPreviewPanel({ task, sessionId = null, onClose, onMaximize }
{task ? ( diff --git a/apps/web/components/task/chat/messages/chat-message.tsx b/apps/web/components/task/chat/messages/chat-message.tsx index 542c781797..1960898829 100644 --- a/apps/web/components/task/chat/messages/chat-message.tsx +++ b/apps/web/components/task/chat/messages/chat-message.tsx @@ -1,5 +1,6 @@ 'use client'; +import { useMemo } from 'react'; import ReactMarkdown from 'react-markdown'; import remarkBreaks from 'remark-breaks'; import remarkGfm from 'remark-gfm'; @@ -14,19 +15,36 @@ type ChatMessageProps = { showRichBlocks?: boolean; }; +function formatMessageTime(dateString: string): string { + if (!dateString) return ''; + try { + const date = new Date(dateString); + if (isNaN(date.getTime())) return ''; + return date.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }); + } catch { + return ''; + } +} + export function ChatMessage({ comment, label, className, showRichBlocks }: ChatMessageProps) { const isUser = comment.author_type === 'user'; + const formattedTime = useMemo(() => formatMessageTime(comment.created_at), [comment.created_at]); return (
-

- {label} - {comment.requests_input ? ( - - Needs input - - ) : null} -

+
+

+ {label} + {comment.requests_input ? ( + + Needs input + + ) : null} +

+ {formattedTime && ( + {formattedTime} + )} +
{isUser ? (

{comment.content || '(empty)'}

) : ( diff --git a/apps/web/components/task/editors-menu.tsx b/apps/web/components/task/editors-menu.tsx index cbcc1d210a..3e6d3ad98a 100644 --- a/apps/web/components/task/editors-menu.tsx +++ b/apps/web/components/task/editors-menu.tsx @@ -17,7 +17,6 @@ export function EditorsMenu({ activeSessionId }: EditorsMenuProps) { const openEditor = useOpenSessionInEditor(activeSessionId ?? null); const { editors } = useEditors(); const defaultEditorId = useAppStore((state) => state.userSettings.defaultEditorId); - console.warn("here[17]: editors-menu.tsx:19: defaultEditorId=", defaultEditorId) const enabledEditors = useMemo( () => @@ -28,7 +27,6 @@ export function EditorsMenu({ activeSessionId }: EditorsMenuProps) { }), [editors] ); - console.warn("here[15]: editors-menu.tsx:21: enabledEditors=", enabledEditors) const resolvedEditorId = useMemo(() => { if (defaultEditorId && enabledEditors.some((editor) => editor.id === defaultEditorId)) { @@ -36,7 +34,6 @@ export function EditorsMenu({ activeSessionId }: EditorsMenuProps) { } return enabledEditors[0]?.id ?? ''; }, [defaultEditorId, enabledEditors]); - console.warn("here[16]: editors-menu.tsx:32: resolvedEditorId=", resolvedEditorId) return (
diff --git a/apps/web/components/task/task-center-panel.tsx b/apps/web/components/task/task-center-panel.tsx index bdba7818e7..1918c5106f 100644 --- a/apps/web/components/task/task-center-panel.tsx +++ b/apps/web/components/task/task-center-panel.tsx @@ -11,11 +11,6 @@ import type { OpenFileTab } from '@/lib/types/backend'; import { FILE_EXTENSION_COLORS } from '@/lib/types/backend'; import { useAppStore } from '@/components/state-provider'; -const AGENTS = [ - { id: 'codex', label: 'Codex' }, - { id: 'claude', label: 'Claude Code' }, -]; - type TaskCenterPanelProps = { selectedDiffPath: string | null; openFileRequest: OpenFileTab | null; @@ -154,7 +149,7 @@ export const TaskCenterPanel = memo(function TaskCenterPanel({ {activeTaskId ? ( - + ) : (
No task selected diff --git a/apps/web/components/task/task-chat-panel.tsx b/apps/web/components/task/task-chat-panel.tsx index 80a977113d..965128c763 100644 --- a/apps/web/components/task/task-chat-panel.tsx +++ b/apps/web/components/task/task-chat-panel.tsx @@ -34,8 +34,6 @@ import { useSessionContextWindow } from '@/hooks/use-session-context-window'; import { ModelSelector } from '@/components/task/model-selector'; type TaskChatPanelProps = { - /** @deprecated No longer used - model selection is now handled by ModelSelector */ - agents?: { id: string; label: string }[]; onSend?: (message: string) => void; sessionId?: string | null; }; @@ -210,13 +208,21 @@ export function TaskChatPanel({ return () => element.removeEventListener('scroll', checkAtBottom); }, [checkAtBottom]); - // Scroll to bottom when new messages arrive or when typing indicator appears + // Track last message content to detect streaming updates + // Use messages from store directly (stable reference) to avoid lint warning + const lastMessageContent = useMemo(() => { + if (messages.length === 0) return ''; + const lastMsg = messages[messages.length - 1]; + return lastMsg?.content ?? ''; + }, [messages]); + + // Scroll to bottom when new messages arrive or when last message content changes (streaming) useEffect(() => { if (itemCount === 0) return; if (wasAtBottomRef.current) { virtualizer.scrollToIndex(itemCount - 1, { align: 'end' }); } - }, [itemCount, virtualizer]); + }, [itemCount, lastMessageContent, virtualizer]); const virtualItems = virtualizer.getVirtualItems(); diff --git a/apps/web/hooks/use-session-agent.ts b/apps/web/hooks/use-session-agent.ts index 78bde25e09..b1cb4760de 100644 --- a/apps/web/hooks/use-session-agent.ts +++ b/apps/web/hooks/use-session-agent.ts @@ -77,7 +77,6 @@ export function useSessionAgent(task: Task | null): UseSessionAgentReturn { session_id?: string; }>('task.execution', { task_id: task.id }); - console.log('[useSessionAgent] Task execution check:', response); if (response.has_execution) { setIsAgentRunning(true); if (response.state) { @@ -91,8 +90,8 @@ export function useSessionAgent(task: Task | null): UseSessionAgentReturn { setAgentSessionId(null); setTaskSessionState(null); } - } catch (err) { - console.error('[useSessionAgent] Failed to check task execution:', err); + } catch { + // Failed to check task execution } }; @@ -124,17 +123,11 @@ export function useSessionAgent(task: Task | null): UseSessionAgentReturn { worktree_path?: string; worktree_branch?: string; } - console.log('[useSessionAgent] orchestrator.start request', { - taskId: task.id, - agentProfileId, - promptLength: prompt?.length ?? 0, - }); const response = await client.request('orchestrator.start', { task_id: task.id, agent_profile_id: agentProfileId, prompt: prompt ?? task.description ?? '', }, 15000); - console.log('[useSessionAgent] orchestrator.start response', response); setIsAgentRunning(true); setTaskSessionState(response.state as TaskSessionState); if (response?.session_id) { @@ -146,8 +139,8 @@ export function useSessionAgent(task: Task | null): UseSessionAgentReturn { setWorktreePath(response.worktree_path); setWorktreeBranch(response.worktree_branch ?? null); } - } catch (error) { - console.error('Failed to start agent:', error); + } catch { + // Failed to start agent } finally { setIsAgentLoading(false); } @@ -165,8 +158,8 @@ export function useSessionAgent(task: Task | null): UseSessionAgentReturn { setIsAgentRunning(false); setAgentSessionId(null); setTaskSessionState(null); - } catch (error) { - console.error('Failed to stop agent:', error); + } catch { + // Failed to stop agent } finally { setIsAgentLoading(false); } diff --git a/apps/web/hooks/use-session-messages.ts b/apps/web/hooks/use-session-messages.ts index 2e7e3e853c..879babd460 100644 --- a/apps/web/hooks/use-session-messages.ts +++ b/apps/web/hooks/use-session-messages.ts @@ -38,7 +38,6 @@ export function useSessionMessages( useEffect(() => { if (!taskSessionId) { - console.log('[useSessionMessages] no session_id yet, clearing messages'); initialFetchStartRef.current = null; lastFetchedSessionIdRef.current = null; setIsWaitingForInitialMessages(false); @@ -52,7 +51,6 @@ export function useSessionMessages( setIsWaitingForInitialMessages(false); return; } - if (messages.length > 0) return; if (initialFetchStartRef.current === null) { initialFetchStartRef.current = Date.now(); setIsWaitingForInitialMessages(true); @@ -63,13 +61,11 @@ export function useSessionMessages( useEffect(() => { if (!taskSessionId) return; if (connectionStatus !== 'connected') { - console.warn('[useSessionMessages] WebSocket not connected yet, waiting to fetch messages'); return; } // Check if messages are already loaded (from SSR or previous fetch) if (messages.length > 0) { - console.log('[useSessionMessages] messages already loaded from SSR or cache, skipping fetch'); lastFetchedSessionIdRef.current = taskSessionId; setIsWaitingForInitialMessages(false); return; @@ -82,7 +78,6 @@ export function useSessionMessages( const fetchMessages = async () => { const client = getWebSocketClient(); if (!client) { - console.warn('[useSessionMessages] WebSocket client not ready'); return; } @@ -94,13 +89,11 @@ export function useSessionMessages( } try { - console.log('[useSessionMessages] requesting message.list', { taskSessionId }); const response = await client.request<{ messages: Message[] }>( 'message.list', { session_id: taskSessionId }, 10000 ); - console.log('[useSessionMessages] message.list response:', JSON.stringify(response, null, 2)); store.getState().setMessages(taskSessionId, response.messages ?? []); lastFetchedSessionIdRef.current = taskSessionId; if ((response.messages ?? []).length > 0) { @@ -147,14 +140,12 @@ export function useSessionMessages( const fetchMessages = async () => { const client = getWebSocketClient(); if (!client) { - console.warn('[useSessionMessages] WebSocket client not ready for state fetch'); return; } setIsLoading(true); store.getState().setMessagesLoading(taskSessionId, true); try { - console.log('[useSessionMessages] requesting message.list after state change', { taskSessionId, taskSessionState }); const response = await client.request<{ messages: Message[] }>( 'message.list', { session_id: taskSessionId }, diff --git a/apps/web/hooks/use-session-resumption.ts b/apps/web/hooks/use-session-resumption.ts index 1b8b0cdddf..0338aecab7 100644 --- a/apps/web/hooks/use-session-resumption.ts +++ b/apps/web/hooks/use-session-resumption.ts @@ -71,12 +71,10 @@ export function useSessionResumption( try { // 1. Check session status - console.log('[useSessionResumption] Checking session status:', { taskId, sessionId }); const status = await client.request('task.session.status', { task_id: taskId, session_id: sessionId, }); - console.log('[useSessionResumption] Session status:', status); setSessionStatus(status); if (status.error) { @@ -101,14 +99,12 @@ export function useSessionResumption( // 2. If agent is already running, we're good if (status.is_agent_running) { - console.log('[useSessionResumption] Agent already running'); setResumptionState('running'); return; } // 3. If session needs resume and is resumable, auto-resume if (status.needs_resume && status.is_resumable) { - console.log('[useSessionResumption] Auto-resuming session'); setResumptionState('resuming'); const resumeResp = await client.request<{ @@ -123,7 +119,6 @@ export function useSessionResumption( }, 30000); if (resumeResp.success) { - console.log('[useSessionResumption] Session resumed successfully'); setResumptionState('resumed'); if (taskId && sessionId && resumeResp.state) { setTaskSession({ @@ -142,20 +137,17 @@ export function useSessionResumption( setWorktreeBranch(resumeResp.worktree_branch); } } else { - console.error('[useSessionResumption] Resume failed:', resumeResp.error); setResumptionState('error'); setError(resumeResp.error ?? 'Failed to resume session'); } } else if (!status.is_resumable) { // Session cannot be resumed (terminal state or missing data) - console.log('[useSessionResumption] Session not resumable'); setResumptionState('idle'); } else { // Session exists but doesn't need resume (already running or other state) setResumptionState('idle'); } } catch (err) { - console.error('[useSessionResumption] Error checking/resuming session:', err); setResumptionState('error'); setError(err instanceof Error ? err.message : 'Unknown error'); } diff --git a/apps/web/hooks/use-task-chat-session.ts b/apps/web/hooks/use-task-chat-session.ts index b2cacb8cca..730ee0c4c2 100644 --- a/apps/web/hooks/use-task-chat-session.ts +++ b/apps/web/hooks/use-task-chat-session.ts @@ -62,8 +62,8 @@ export function useTaskChatSession(taskId: string | null): UseTaskChatSessionRet updated_at: '', }); } - } catch (err) { - console.error('[useTaskChatSession] Failed to check task execution:', err); + } catch { + // Failed to check task execution } }; diff --git a/apps/web/hooks/use-task-session.ts b/apps/web/hooks/use-task-session.ts index d0572e46df..369da94b88 100644 --- a/apps/web/hooks/use-task-session.ts +++ b/apps/web/hooks/use-task-session.ts @@ -55,8 +55,7 @@ export function useTaskSession(taskId: string | null) { setFetchedSessionId(response.sessions[0]?.id ?? null); setIsLoading(false); } - } catch (error) { - console.error('Failed to load task sessions:', error); + } catch { if (isActive) { setFetchedSessionId(null); setIsLoading(false); diff --git a/apps/web/hooks/use-task-sessions.ts b/apps/web/hooks/use-task-sessions.ts index 1c989155b5..993aba2840 100644 --- a/apps/web/hooks/use-task-sessions.ts +++ b/apps/web/hooks/use-task-sessions.ts @@ -25,8 +25,7 @@ export function useTaskSessions(taskId: string | null) { try { const response = await listTaskSessions(taskId, { cache: 'no-store' }); setTaskSessionsForTask(taskId, response.sessions ?? []); - } catch (error) { - console.error('Failed to load task sessions:', error); + } catch { setTaskSessionsForTask(taskId, []); } finally { setTaskSessionsLoading(taskId, false); diff --git a/apps/web/lib/state/store.ts b/apps/web/lib/state/store.ts index bbdc3a2f1b..e57dafe1d0 100644 --- a/apps/web/lib/state/store.ts +++ b/apps/web/lib/state/store.ts @@ -14,6 +14,7 @@ import type { TaskState as TaskStatus, TaskSession, } from '@/lib/types/http'; +import type { FileInfo } from '@/lib/types/backend'; import { createStore } from 'zustand/vanilla'; import { immer } from 'zustand/middleware/immer'; @@ -161,15 +162,7 @@ export type ShellState = { >; }; -export type FileInfo = { - path: string; - status: 'modified' | 'added' | 'deleted' | 'untracked' | 'renamed'; - staged: boolean; - additions?: number; - deletions?: number; - old_path?: string; - diff?: string; -}; +export type { FileInfo }; export type GitStatusEntry = { branch: string | null; @@ -662,9 +655,6 @@ export function createAppStore(initialState?: Partial) { setAvailableAgentsLoading: (loading) => set((draft) => { draft.availableAgents.loading = loading; - if (!loading && !draft.availableAgents.loaded) { - draft.availableAgents.loaded = false; - } }), setAgentProfiles: (profiles) => set((draft) => { @@ -679,9 +669,6 @@ export function createAppStore(initialState?: Partial) { setRepositoriesLoading: (workspaceId, loading) => set((draft) => { draft.repositories.loadingByWorkspaceId[workspaceId] = loading; - if (!loading && !draft.repositories.loadedByWorkspaceId[workspaceId]) { - draft.repositories.loadedByWorkspaceId[workspaceId] = false; - } }), setRepositoryBranches: (repositoryId, branches) => set((draft) => { @@ -692,9 +679,6 @@ export function createAppStore(initialState?: Partial) { setRepositoryBranchesLoading: (repositoryId, loading) => set((draft) => { draft.repositoryBranches.loadingByRepositoryId[repositoryId] = loading; - if (!loading && !draft.repositoryBranches.loadedByRepositoryId[repositoryId]) { - draft.repositoryBranches.loadedByRepositoryId[repositoryId] = false; - } }), setSettingsData: (next) => set((draft) => { diff --git a/apps/web/lib/ws/client.ts b/apps/web/lib/ws/client.ts index 269511a9aa..4a78b3f462 100644 --- a/apps/web/lib/ws/client.ts +++ b/apps/web/lib/ws/client.ts @@ -13,14 +13,6 @@ export interface ReconnectOptions { backoffMultiplier?: number; } -export interface ConnectionMetrics { - reconnectAttempts: number; - lastDisconnectTime: number | null; - lastDisconnectReason: string | null; - totalReconnects: number; - isHealthy: boolean; -} - const DEFAULT_RECONNECT_OPTIONS: Required = { enabled: true, maxAttempts: 10, @@ -45,17 +37,6 @@ export class WebSocketClient { private subscriptions = new Map(); private sessionSubscriptions = new Map(); private userSubscriptionCount = 0; - private heartbeatTimer: ReturnType | null = null; - private heartbeatTimeoutTimer: ReturnType | null = null; - private lastMessageTime: number = Date.now(); - private metrics: ConnectionMetrics = { - reconnectAttempts: 0, - lastDisconnectTime: null, - lastDisconnectReason: null, - totalReconnects: 0, - isHealthy: true, - }; - private onMetricsChange?: (metrics: ConnectionMetrics) => void; constructor( private url: string, @@ -122,13 +103,10 @@ export class WebSocketClient { } const action = (message as { action?: string })?.action as BackendMessageType | undefined; if (!action) continue; - console.log('[WS] Received notification:', action); const handlers = this.handlers.get(action); if (!handlers) { - console.log('[WS] No handlers registered for:', action); continue; } - console.log('[WS] Calling', handlers.size, 'handler(s) for:', action); handlers.forEach((handler) => handler(message)); } catch { // Ignore parse errors for individual messages diff --git a/apps/web/lib/ws/handlers/git-status.ts b/apps/web/lib/ws/handlers/git-status.ts index d872d8fba9..3f6731ac16 100644 --- a/apps/web/lib/ws/handlers/git-status.ts +++ b/apps/web/lib/ws/handlers/git-status.ts @@ -9,14 +9,6 @@ export function registerGitStatusHandlers(store: StoreApi): WsHandlers if (!payload.session_id) { return; } - console.log('[WS] git.status received:', { - session_id: payload.session_id, - branch: payload.branch, - modified: payload.modified.length, - added: payload.added.length, - deleted: payload.deleted.length, - untracked: payload.untracked.length, - }); store.getState().setGitStatus(payload.session_id, { branch: payload.branch, remote_branch: payload.remote_branch ?? null, diff --git a/apps/web/lib/ws/utils.ts b/apps/web/lib/ws/utils.ts index 7f28c31a4a..909b574e4d 100644 --- a/apps/web/lib/ws/utils.ts +++ b/apps/web/lib/ws/utils.ts @@ -24,10 +24,3 @@ export function isValidBackendUrl(url: string): boolean { return false; } } - -/** - * Generate unique message ID - */ -export function generateMessageId(): string { - return `${Date.now()}-${Math.random().toString(36).slice(2, 11)}`; -}