dreampiler/kandev
1
0
Fork 0
Code Issues 43 Pull Requests Actions 6 Packages Projects Releases Wiki Activity

refactor: comprehensive code quality improvements (#69)

Browse Source
This commit is contained in:
João Salavisa
2026-01-21 19:49:36 +00:00
committed by GitHub
parent fba868d3fe
commit 8dd88e8c60
28 changed files with 463 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/cmd/kandev/adapters.go
+17
View File
@@ -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)
}
apps/backend/internal/agent/lifecycle/event_types.go
+6
View File
@@ -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).
apps/backend/internal/agent/lifecycle/events.go
+18
View File
@@ -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 {
apps/backend/internal/agent/lifecycle/manager.go
+116 -3
View File
@@ -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",
apps/backend/internal/agent/lifecycle/session.go
+34 -7
View File
@@ -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)
}
apps/backend/internal/agent/lifecycle/streams.go
+3 -3
View File
@@ -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)
apps/backend/internal/agent/lifecycle/types.go
+4
View File
@@ -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
apps/backend/internal/agentctl/server/process/git.go
+35
View File
@@ -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
}
apps/backend/internal/agentctl/server/process/workspace_tracker.go
+12 -4
View File
@@ -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)
apps/backend/internal/orchestrator/controller/errors.go
-5
View File
@@ -1,5 +0,0 @@
package controller
// This file is reserved for controller-specific errors.
// Currently no custom errors are defined.
apps/backend/internal/orchestrator/executor/executor.go
+2 -2
View File
@@ -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),
apps/backend/internal/orchestrator/service.go
+83 -38
View File
@@ -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))
}
apps/backend/internal/task/service/service.go
+84
View File
@@ -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.
apps/web/components/task-preview-panel.tsx
-7
View File
@@ -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 }
<div className="flex-1 min-h-0 p-4 flex flex-col">
{task ? (
<TaskChatPanel
agents={AGENTS}
onSend={handleSendMessage}
sessionId={activeSessionId}
/>
apps/web/components/task/chat/messages/chat-message.tsx
+26 -8
View File
@@ -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 (
<div className={cn('w-full rounded-lg border px-4 py-3 text-sm', className)}>
<p className="text-[11px] uppercase tracking-wide mb-2 opacity-70">
{label}
{comment.requests_input ? (
<span className="ml-2 rounded-full bg-amber-500/20 px-2 py-0.5 text-[10px] text-amber-300">
Needs input
</span>
) : null}
</p>
<div className="flex items-center justify-between mb-2">
<p className="text-[11px] uppercase tracking-wide opacity-70">
{label}
{comment.requests_input ? (
<span className="ml-2 rounded-full bg-amber-500/20 px-2 py-0.5 text-[10px] text-amber-300">
Needs input
</span>
) : null}
</p>
{formattedTime && (
<span className="text-[10px] text-muted-foreground/60">{formattedTime}</span>
)}
</div>
{isUser ? (
<p className="whitespace-pre-wrap">{comment.content || '(empty)'}</p>
) : (
apps/web/components/task/editors-menu.tsx
-3
View File
@@ -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 (
<div className="inline-flex rounded-md border border-border overflow-hidden">
apps/web/components/task/task-center-panel.tsx
+1 -6
View File
@@ -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({
<TabsContent value="chat" className="mt-3 flex flex-col min-h-0 flex-1">
{activeTaskId ? (
<TaskChatPanel agents={AGENTS} sessionId={sessionId} />
<TaskChatPanel sessionId={sessionId} />
) : (
<div className="flex items-center justify-center h-full text-muted-foreground">
No task selected
apps/web/components/task/task-chat-panel.tsx
+10 -4
View File
@@ -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();
apps/web/hooks/use-session-agent.ts
+6 -13
View File
@@ -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<StartResponse>('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);
}
apps/web/hooks/use-session-messages.ts
-9
View File
@@ -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 },
apps/web/hooks/use-session-resumption.ts
-8
View File
@@ -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<SessionStatus>('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');
}
apps/web/hooks/use-task-chat-session.ts
+2 -2
View File
@@ -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
}
};
apps/web/hooks/use-task-session.ts
+1 -2
View File
@@ -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);
apps/web/hooks/use-task-sessions.ts
+1 -2
View File
@@ -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);
apps/web/lib/state/store.ts
+2 -18
View File
@@ -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<AppState>) {
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<AppState>) {
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<AppState>) {
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) => {
apps/web/lib/ws/client.ts
-22
View File
@@ -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<ReconnectOptions> = {
enabled: true,
maxAttempts: 10,
@@ -45,17 +37,6 @@ export class WebSocketClient {
private subscriptions = new Map<string, number>();
private sessionSubscriptions = new Map<string, number>();
private userSubscriptionCount = 0;
private heartbeatTimer: ReturnType<typeof setTimeout> | null = null;
private heartbeatTimeoutTimer: ReturnType<typeof setTimeout> | 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
apps/web/lib/ws/handlers/git-status.ts
-8
View File
@@ -9,14 +9,6 @@ export function registerGitStatusHandlers(store: StoreApi<AppState>): 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,
apps/web/lib/ws/utils.ts
-7
View File
@@ -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)}`;
}