diff --git a/apps/backend/cmd/agentctl/main.go b/apps/backend/cmd/agentctl/main.go index a3f3044648..f8a86946d2 100644 --- a/apps/backend/cmd/agentctl/main.go +++ b/apps/backend/cmd/agentctl/main.go @@ -128,7 +128,8 @@ func containsHelper(s, substr string) bool { return false } -// runDockerMode runs agentctl in Docker mode (single instance, same as legacy behavior) +// runDockerMode runs agentctl in Docker mode (single instance) +// Uses the unified ControlServer with embedded process manager func runDockerMode(cfg *config.MultiConfig, log *logger.Logger) { // Load full config from environment (including AutoApprovePermissions) singleCfg := config.Load() @@ -158,14 +159,14 @@ func runDockerMode(cfg *config.MultiConfig, log *logger.Logger) { // Create process manager procMgr := process.NewManager(singleCfg, log) - // Create HTTP API server - server := api.NewServer(singleCfg, procMgr, log) + // Create unified ControlServer with embedded single-instance support + controlServer := api.NewSingleInstanceServer(cfg, singleCfg, procMgr, log) // Start HTTP server addr := fmt.Sprintf(":%d", singleCfg.Port) httpServer := &http.Server{ Addr: addr, - Handler: server.Router(), + Handler: controlServer.Router(), } go func() { diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 3e5390affe..169fae5475 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -359,7 +359,7 @@ func main() { } serviceCfg := orchestrator.DefaultServiceConfig() - orchestratorSvc := orchestrator.NewService(serviceCfg, eventBus, nil, agentManagerClient, taskRepoAdapter, taskRepo, log) + orchestratorSvc := orchestrator.NewService(serviceCfg, eventBus, agentManagerClient, taskRepoAdapter, taskRepo, log) // Set up comment creator for saving agent responses as comments orchestratorSvc.SetCommentCreator(&commentCreatorAdapter{svc: taskSvc}) @@ -1040,71 +1040,4 @@ func (a *commentCreatorAdapter) UpdateToolCallComment(ctx context.Context, taskI return a.svc.UpdateToolCallComment(ctx, taskID, toolCallID, status, result) } -// mapACPTypeToCommentType maps ACP message types to comment types -func mapACPTypeToCommentType(msgType protocol.MessageType) string { - switch msgType { - case protocol.MessageTypeProgress: - return "progress" - case protocol.MessageTypeLog: - return "content" // Log messages are treated as content - case protocol.MessageTypeResult: - return "status" - case protocol.MessageTypeError: - return "error" - case protocol.MessageTypeStatus: - return "status" - case protocol.MessageTypeInputRequired: - return "content" - case protocol.MessageTypeSessionInfo: - return "status" - default: - return "content" - } -} -// extractACPContent extracts a human-readable message from ACP data -func extractACPContent(msg *protocol.Message) string { - if msg.Data == nil { - return "" - } - - // Check for nested content.text structure (used by agent_message_chunk) - if content, ok := msg.Data["content"].(map[string]interface{}); ok { - if text, ok := content["text"].(string); ok && text != "" { - return text - } - } - - // Try common message fields - if message, ok := msg.Data["message"].(string); ok && message != "" { - return message - } - if text, ok := msg.Data["text"].(string); ok && text != "" { - return text - } - if errMsg, ok := msg.Data["error"].(string); ok && errMsg != "" { - return errMsg - } - if summary, ok := msg.Data["summary"].(string); ok && summary != "" { - return summary - } - if prompt, ok := msg.Data["prompt"].(string); ok && prompt != "" { - return prompt - } - - // For progress updates, generate a message - if msg.Type == protocol.MessageTypeProgress { - if progress, ok := msg.Data["progress"].(float64); ok { - return fmt.Sprintf("Progress: %d%%", int(progress)) - } - } - - // For status updates - if msg.Type == protocol.MessageTypeStatus { - if status, ok := msg.Data["status"].(string); ok && status != "" { - return fmt.Sprintf("Status: %s", status) - } - } - - return "" -} diff --git a/apps/backend/internal/agent/acp/session.go b/apps/backend/internal/agent/acp/session.go deleted file mode 100644 index b0d03cf086..0000000000 --- a/apps/backend/internal/agent/acp/session.go +++ /dev/null @@ -1,599 +0,0 @@ -// Package acp provides ACP (Agent Client Protocol) session management -package acp - -import ( - "context" - "encoding/json" - "fmt" - "io" - "sync" - "time" - - "github.com/kandev/kandev/internal/common/logger" - "github.com/kandev/kandev/internal/events/bus" - "github.com/kandev/kandev/pkg/acp/jsonrpc" - "go.uber.org/zap" -) - -// Session represents an active ACP session with an agent -type Session struct { - InstanceID string - TaskID string - SessionID string // ACP session ID (from agent) - Client *jsonrpc.Client - Stdin io.WriteCloser - Stdout io.Reader - CreatedAt time.Time - Status string // initializing, ready, prompting, complete, error - mu sync.RWMutex -} - -// PendingPermission represents a permission request waiting for user response -type PendingPermission struct { - ID string // Unique ID for this pending request - RPCID interface{} // JSON-RPC request ID from agent - InstanceID string // Agent instance - TaskID string // Task being worked on - SessionID string // ACP session ID - ToolCallID string // Tool call requesting permission - Title string // Human-readable title - Options []jsonrpc.PermissionOption // Available choices - ResponseCh chan *PermissionResponseMsg // Channel to receive user response - CreatedAt time.Time -} - -// PermissionResponseMsg is the message sent by the user in response to a permission request -type PermissionResponseMsg struct { - OptionID string - Cancelled bool -} - -// PermissionRequestHandler is called when an agent requests permission from the user -type PermissionRequestHandler func(req *PendingPermission) - -// UpdateHandler is called when the agent sends session updates -type UpdateHandler func(instanceID, taskID string, updateType string, data json.RawMessage) - -// SessionManager manages ACP sessions for multiple agents -type SessionManager struct { - sessions map[string]*Session // by instance ID - mu sync.RWMutex - eventBus bus.EventBus - logger *logger.Logger - - // Handler for session updates - updateHandler UpdateHandler - - // Handler for permission requests - permissionHandler PermissionRequestHandler - - // Pending permission requests waiting for user response - pendingPermissions map[string]*PendingPermission - permissionMu sync.RWMutex -} - -// NewSessionManager creates a new session manager -func NewSessionManager(eventBus bus.EventBus, log *logger.Logger) *SessionManager { - return &SessionManager{ - sessions: make(map[string]*Session), - pendingPermissions: make(map[string]*PendingPermission), - eventBus: eventBus, - logger: log.WithFields(zap.String("component", "acp-session-manager")), - } -} - -// SetUpdateHandler sets the handler for session updates -func (m *SessionManager) SetUpdateHandler(handler UpdateHandler) { - m.updateHandler = handler -} - -// SetPermissionHandler sets the handler for permission requests from agents -func (m *SessionManager) SetPermissionHandler(handler PermissionRequestHandler) { - m.permissionHandler = handler -} - -// CreateSession creates a new ACP session for an agent instance -func (m *SessionManager) CreateSession(ctx context.Context, instanceID, taskID string, stdin io.WriteCloser, stdout io.Reader) error { - m.logger.Info("Creating ACP session", - zap.String("instance_id", instanceID), - zap.String("task_id", taskID)) - - // Create JSON-RPC client - client := jsonrpc.NewClient(stdin, stdout, m.logger) - - session := &Session{ - InstanceID: instanceID, - TaskID: taskID, - Client: client, - Stdin: stdin, - Stdout: stdout, - CreatedAt: time.Now(), - Status: "initializing", - } - - // Set notification handler - client.SetNotificationHandler(func(method string, params json.RawMessage) { - m.handleNotification(session, method, params) - }) - - // Set request handler for agent-to-client requests (like session/request_permission) - client.SetRequestHandler(func(id interface{}, method string, params json.RawMessage) { - m.handleRequest(session, id, method, params) - }) - - // Start the client read loop - client.Start(ctx) - - // Store session - m.mu.Lock() - m.sessions[instanceID] = session - m.mu.Unlock() - - return nil -} - -// Initialize performs the ACP initialize handshake -func (m *SessionManager) Initialize(ctx context.Context, instanceID string) error { - session, err := m.getSession(instanceID) - if err != nil { - return err - } - - m.logger.Info("Initializing ACP session", zap.String("instance_id", instanceID)) - - params := jsonrpc.InitializeParams{ - ProtocolVersion: 1, - ClientInfo: jsonrpc.ClientInfo{ - Name: "kandev", - Version: "0.1.0", - }, - Capabilities: jsonrpc.ClientCapabilities{ - Streaming: true, - }, - } - - resp, err := session.Client.Call(ctx, jsonrpc.MethodInitialize, params) - if err != nil { - return fmt.Errorf("initialize failed: %w", err) - } - - if resp.Error != nil { - return fmt.Errorf("initialize error: %s (code %d)", resp.Error.Message, resp.Error.Code) - } - - session.mu.Lock() - session.Status = "ready" - session.mu.Unlock() - - m.logger.Info("ACP session initialized", zap.String("instance_id", instanceID)) - return nil -} - -// NewSession creates a new ACP session (session/new) -func (m *SessionManager) NewSession(ctx context.Context, instanceID, cwd string) (string, error) { - session, err := m.getSession(instanceID) - if err != nil { - return "", err - } - - m.logger.Info("Creating new ACP session", - zap.String("instance_id", instanceID), - zap.String("cwd", cwd)) - - params := jsonrpc.SessionNewParams{ - Cwd: cwd, - McpServers: []jsonrpc.McpServer{}, // Empty array, no MCP servers for now - } - - resp, err := session.Client.Call(ctx, jsonrpc.MethodSessionNew, params) - if err != nil { - return "", fmt.Errorf("session/new failed: %w", err) - } - - if resp.Error != nil { - return "", fmt.Errorf("session/new error: %s", resp.Error.Message) - } - - var result jsonrpc.SessionNewResult - if err := json.Unmarshal(resp.Result, &result); err != nil { - return "", fmt.Errorf("failed to parse session/new result: %w", err) - } - - session.mu.Lock() - session.SessionID = result.SessionID - session.mu.Unlock() - - m.logger.Info("ACP session created", - zap.String("instance_id", instanceID), - zap.String("session_id", result.SessionID)) - - return result.SessionID, nil -} - -// LoadSession resumes an existing ACP session (session/load) -func (m *SessionManager) LoadSession(ctx context.Context, instanceID, sessionID string) error { - session, err := m.getSession(instanceID) - if err != nil { - return err - } - - m.logger.Info("Loading ACP session", - zap.String("instance_id", instanceID), - zap.String("session_id", sessionID)) - - params := jsonrpc.SessionLoadParams{ - SessionID: sessionID, - } - - resp, err := session.Client.Call(ctx, jsonrpc.MethodSessionLoad, params) - if err != nil { - return fmt.Errorf("session/load failed: %w", err) - } - - if resp.Error != nil { - return fmt.Errorf("session/load error: %s", resp.Error.Message) - } - - session.mu.Lock() - session.SessionID = sessionID - session.mu.Unlock() - - m.logger.Info("ACP session loaded", zap.String("instance_id", instanceID)) - return nil -} - -// Prompt sends a prompt to the agent (session/prompt) -func (m *SessionManager) Prompt(ctx context.Context, instanceID, message string) error { - session, err := m.getSession(instanceID) - if err != nil { - return err - } - - session.mu.RLock() - sessionID := session.SessionID - session.mu.RUnlock() - - if sessionID == "" { - return fmt.Errorf("no session ID available, call NewSession first") - } - - m.logger.Info("Sending prompt to agent", - zap.String("instance_id", instanceID), - zap.String("session_id", sessionID), - zap.Int("message_length", len(message))) - - session.mu.Lock() - session.Status = "prompting" - session.mu.Unlock() - - // ACP protocol requires prompt to be an array of ContentBlock - params := jsonrpc.SessionPromptParams{ - SessionID: sessionID, - Prompt: []jsonrpc.ContentBlock{ - { - Type: "text", - Text: message, - }, - }, - } - - resp, err := session.Client.Call(ctx, jsonrpc.MethodSessionPrompt, params) - if err != nil { - return fmt.Errorf("session/prompt failed: %w", err) - } - - if resp.Error != nil { - return fmt.Errorf("session/prompt error: %s", resp.Error.Message) - } - - return nil -} - -// Cancel cancels the current operation (session/cancel) -func (m *SessionManager) Cancel(ctx context.Context, instanceID, reason string) error { - session, err := m.getSession(instanceID) - if err != nil { - return err - } - - m.logger.Info("Cancelling agent operation", - zap.String("instance_id", instanceID), - zap.String("reason", reason)) - - params := jsonrpc.SessionCancelParams{ - Reason: reason, - } - - // Cancel is a notification, not a request - return session.Client.Notify(jsonrpc.MethodSessionCancel, params) -} - -// CloseSession closes an ACP session -func (m *SessionManager) CloseSession(instanceID string) error { - m.mu.Lock() - session, exists := m.sessions[instanceID] - if exists { - delete(m.sessions, instanceID) - } - m.mu.Unlock() - - if !exists { - return fmt.Errorf("session not found: %s", instanceID) - } - - m.logger.Info("Closing ACP session", zap.String("instance_id", instanceID)) - - session.Client.Stop() - if session.Stdin != nil { - session.Stdin.Close() - } - - return nil -} - -// GetSession returns a session by instance ID -func (m *SessionManager) GetSession(instanceID string) (*Session, bool) { - m.mu.RLock() - defer m.mu.RUnlock() - session, exists := m.sessions[instanceID] - return session, exists -} - -// getSession returns a session or error if not found -func (m *SessionManager) getSession(instanceID string) (*Session, error) { - m.mu.RLock() - defer m.mu.RUnlock() - session, exists := m.sessions[instanceID] - if !exists { - return nil, fmt.Errorf("session not found: %s", instanceID) - } - return session, nil -} - -// handleNotification handles incoming notifications from the agent -func (m *SessionManager) handleNotification(session *Session, method string, params json.RawMessage) { - m.logger.Debug("Received notification", - zap.String("instance_id", session.InstanceID), - zap.String("method", method)) - - switch method { - case jsonrpc.NotificationSessionUpdate: - var update jsonrpc.SessionUpdate - if err := json.Unmarshal(params, &update); err != nil { - m.logger.Error("Failed to parse session update", zap.Error(err)) - return - } - - // Handle completion - if update.Type == "complete" { - session.mu.Lock() - session.Status = "complete" - session.mu.Unlock() - } - - // Forward to update handler - if m.updateHandler != nil { - m.updateHandler(session.InstanceID, session.TaskID, update.Type, update.Data) - } - - default: - m.logger.Warn("Unknown notification method", zap.String("method", method)) - } -} - -// GetSessionID returns the ACP session ID for an instance -func (m *SessionManager) GetSessionID(instanceID string) (string, bool) { - session, exists := m.GetSession(instanceID) - if !exists { - return "", false - } - session.mu.RLock() - defer session.mu.RUnlock() - return session.SessionID, session.SessionID != "" -} - -// handleRequest handles incoming requests from the agent (e.g., session/request_permission) -func (m *SessionManager) handleRequest(session *Session, id interface{}, method string, params json.RawMessage) { - m.logger.Info("Received request from agent", - zap.String("instance_id", session.InstanceID), - zap.String("method", method), - zap.Any("id", id)) - - switch method { - case jsonrpc.MethodRequestPermission: - m.handleRequestPermission(session, id, params) - default: - m.logger.Warn("Unknown request method from agent", zap.String("method", method)) - // Send error response for unknown methods - session.Client.SendResponse(id, nil, &jsonrpc.Error{ - Code: jsonrpc.MethodNotFound, - Message: fmt.Sprintf("Unknown method: %s", method), - }) - } -} - -// handleRequestPermission handles session/request_permission requests from the agent -func (m *SessionManager) handleRequestPermission(session *Session, id interface{}, params json.RawMessage) { - var reqParams jsonrpc.RequestPermissionParams - if err := json.Unmarshal(params, &reqParams); err != nil { - m.logger.Error("Failed to parse request_permission params", zap.Error(err)) - session.Client.SendResponse(id, nil, &jsonrpc.Error{ - Code: jsonrpc.InvalidParams, - Message: "Invalid params", - }) - return - } - - m.logger.Info("Agent requesting permission", - zap.String("instance_id", session.InstanceID), - zap.String("session_id", reqParams.SessionID), - zap.String("tool_call_id", reqParams.ToolCall.ToolCallID), - zap.String("title", reqParams.ToolCall.Title), - zap.Int("num_options", len(reqParams.Options))) - - // If no permission handler is set, fall back to auto-approve - if m.permissionHandler == nil { - m.autoApprovePermission(session, id, &reqParams) - return - } - - // Create a pending permission request - pendingID := fmt.Sprintf("%s-%d", session.InstanceID, time.Now().UnixNano()) - pending := &PendingPermission{ - ID: pendingID, - RPCID: id, - InstanceID: session.InstanceID, - TaskID: session.TaskID, - SessionID: reqParams.SessionID, - ToolCallID: reqParams.ToolCall.ToolCallID, - Title: reqParams.ToolCall.Title, - Options: reqParams.Options, - ResponseCh: make(chan *PermissionResponseMsg, 1), - CreatedAt: time.Now(), - } - - // Store the pending request - m.permissionMu.Lock() - m.pendingPermissions[pendingID] = pending - m.permissionMu.Unlock() - - // Notify the permission handler (which will send WebSocket notification to user) - m.permissionHandler(pending) - - // Wait for user response in a goroutine to not block the read loop - go m.waitForPermissionResponse(session, pending) -} - -// waitForPermissionResponse waits for user response and sends it to the agent -func (m *SessionManager) waitForPermissionResponse(session *Session, pending *PendingPermission) { - // Wait for response with a 5-minute timeout - timeout := time.After(5 * time.Minute) - - select { - case resp := <-pending.ResponseCh: - m.logger.Info("Received permission response from user", - zap.String("pending_id", pending.ID), - zap.String("option_id", resp.OptionID), - zap.Bool("cancelled", resp.Cancelled)) - - var result jsonrpc.RequestPermissionResult - if resp.Cancelled { - result = jsonrpc.RequestPermissionResult{ - Outcome: jsonrpc.PermissionOutcome{ - Outcome: "cancelled", - }, - } - } else { - result = jsonrpc.RequestPermissionResult{ - Outcome: jsonrpc.PermissionOutcome{ - Outcome: "selected", - OptionID: resp.OptionID, - }, - } - } - - if err := session.Client.SendResponse(pending.RPCID, result, nil); err != nil { - m.logger.Error("Failed to send permission response to agent", zap.Error(err)) - } - - case <-timeout: - m.logger.Warn("Permission request timed out, cancelling", - zap.String("pending_id", pending.ID)) - - result := jsonrpc.RequestPermissionResult{ - Outcome: jsonrpc.PermissionOutcome{ - Outcome: "cancelled", - }, - } - session.Client.SendResponse(pending.RPCID, result, nil) - } - - // Clean up - m.permissionMu.Lock() - delete(m.pendingPermissions, pending.ID) - m.permissionMu.Unlock() -} - -// autoApprovePermission auto-approves a permission request (fallback behavior) -func (m *SessionManager) autoApprovePermission(session *Session, id interface{}, reqParams *jsonrpc.RequestPermissionParams) { - var selectedOptionID string - for _, opt := range reqParams.Options { - if opt.Kind == "allow_once" || opt.Kind == "allow_always" { - selectedOptionID = opt.OptionID - m.logger.Info("Auto-approving permission request", - zap.String("option_id", selectedOptionID), - zap.String("option_name", opt.Name), - zap.String("kind", opt.Kind)) - break - } - } - - // If no allow option found, use the first option - if selectedOptionID == "" && len(reqParams.Options) > 0 { - selectedOptionID = reqParams.Options[0].OptionID - m.logger.Info("No allow option found, using first option", - zap.String("option_id", selectedOptionID)) - } - - result := jsonrpc.RequestPermissionResult{ - Outcome: jsonrpc.PermissionOutcome{ - Outcome: "selected", - OptionID: selectedOptionID, - }, - } - - if err := session.Client.SendResponse(id, result, nil); err != nil { - m.logger.Error("Failed to send permission response", zap.Error(err)) - } else { - m.logger.Info("Sent auto-approved permission response", - zap.String("instance_id", session.InstanceID), - zap.String("selected_option", selectedOptionID)) - } -} - -// RespondToPermission sends a user's response to a pending permission request -func (m *SessionManager) RespondToPermission(pendingID string, optionID string, cancelled bool) error { - m.permissionMu.RLock() - pending, exists := m.pendingPermissions[pendingID] - m.permissionMu.RUnlock() - - if !exists { - return fmt.Errorf("pending permission request not found: %s", pendingID) - } - - // Send response through channel (non-blocking) - select { - case pending.ResponseCh <- &PermissionResponseMsg{ - OptionID: optionID, - Cancelled: cancelled, - }: - m.logger.Info("Sent permission response to waiting goroutine", - zap.String("pending_id", pendingID), - zap.String("option_id", optionID)) - return nil - default: - return fmt.Errorf("permission request already responded to or timed out") - } -} - -// GetPendingPermission returns a pending permission request by ID -func (m *SessionManager) GetPendingPermission(pendingID string) (*PendingPermission, bool) { - m.permissionMu.RLock() - defer m.permissionMu.RUnlock() - pending, exists := m.pendingPermissions[pendingID] - return pending, exists -} - -// GetPendingPermissionsForTask returns all pending permissions for a task -func (m *SessionManager) GetPendingPermissionsForTask(taskID string) []*PendingPermission { - m.permissionMu.RLock() - defer m.permissionMu.RUnlock() - - var result []*PendingPermission - for _, pending := range m.pendingPermissions { - if pending.TaskID == taskID { - result = append(result, pending) - } - } - return result -} diff --git a/apps/backend/internal/agent/agentctl/launcher/launcher.go b/apps/backend/internal/agent/agentctl/launcher/launcher.go index c8144907f3..435ba7140a 100644 --- a/apps/backend/internal/agent/agentctl/launcher/launcher.go +++ b/apps/backend/internal/agent/agentctl/launcher/launcher.go @@ -221,32 +221,7 @@ func (l *Launcher) Stop(ctx context.Context) error { } } -// Wait blocks until the agentctl process exits. -// Returns the exit error if any. -func (l *Launcher) Wait() error { - <-l.exited - if l.cmd != nil && l.cmd.ProcessState != nil { - if !l.cmd.ProcessState.Success() { - return fmt.Errorf("agentctl exited with code %d", l.cmd.ProcessState.ExitCode()) - } - } - return nil -} -// Running returns true if agentctl is currently running. -func (l *Launcher) Running() bool { - select { - case <-l.exited: - return false - default: - return l.cmd != nil && l.cmd.Process != nil - } -} - -// Port returns the control port agentctl is listening on. -func (l *Launcher) Port() int { - return l.port -} // checkPortAvailable verifies the port is not in use. func (l *Launcher) checkPortAvailable() error { diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 593e3a7ce8..51f163605e 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -1491,21 +1491,7 @@ func (m *Manager) updateInstanceProgress(instanceID string, progress int) { instance.Progress = progress } -// updateInstanceComplete marks an instance as completed -func (m *Manager) updateInstanceComplete(instanceID string) { - m.mu.Lock() - defer m.mu.Unlock() - instance, exists := m.instances[instanceID] - if !exists { - return - } - - instance.Status = v1.AgentStatusCompleted - instance.Progress = 100 - now := time.Now() - instance.FinishedAt = &now -} // updateInstanceError updates an instance with an error func (m *Manager) updateInstanceError(instanceID, errorMsg string) { diff --git a/apps/backend/internal/agent/streaming/manager.go b/apps/backend/internal/agent/streaming/manager.go deleted file mode 100644 index 3821555f91..0000000000 --- a/apps/backend/internal/agent/streaming/manager.go +++ /dev/null @@ -1,145 +0,0 @@ -package streaming - -import ( - "context" - "fmt" - "io" - "sync" - - "go.uber.org/zap" - - "github.com/kandev/kandev/internal/common/logger" - "github.com/kandev/kandev/internal/events/bus" -) - -// Manager manages ACP stream readers for multiple agent instances -type Manager struct { - eventBus bus.EventBus - logger *logger.Logger - - readers map[string]*StreamReader // by instanceID - mu sync.RWMutex -} - -// NewManager creates a new streaming manager -func NewManager(eventBus bus.EventBus, log *logger.Logger) *Manager { - return &Manager{ - eventBus: eventBus, - logger: log.WithFields(zap.String("component", "streaming-manager")), - readers: make(map[string]*StreamReader), - } -} - -// StartStreaming starts streaming for an agent instance -func (m *Manager) StartStreaming(ctx context.Context, instanceID, taskID string, reader io.ReadCloser) error { - m.mu.Lock() - defer m.mu.Unlock() - - // Check if already streaming for this instance - if existing, exists := m.readers[instanceID]; exists { - if existing.IsRunning() { - return fmt.Errorf("streaming already active for instance %s", instanceID) - } - // Previous reader exists but not running, replace it - } - - // Create new stream reader - streamReader := NewStreamReader(instanceID, taskID, m.eventBus, m.logger) - m.readers[instanceID] = streamReader - - // Start reading - if err := streamReader.Start(ctx, reader); err != nil { - delete(m.readers, instanceID) - return fmt.Errorf("failed to start streaming: %w", err) - } - - m.logger.Info("started streaming for instance", - zap.String("instance_id", instanceID), - zap.String("task_id", taskID)) - - return nil -} - -// StopStreaming stops streaming for an agent instance -func (m *Manager) StopStreaming(instanceID string) error { - m.mu.Lock() - reader, exists := m.readers[instanceID] - if !exists { - m.mu.Unlock() - return nil // Not streaming, no error - } - delete(m.readers, instanceID) - m.mu.Unlock() - - if err := reader.Stop(); err != nil { - m.logger.Warn("error stopping stream reader", - zap.String("instance_id", instanceID), - zap.Error(err)) - return err - } - - m.logger.Info("stopped streaming for instance", - zap.String("instance_id", instanceID)) - - return nil -} - -// StopAll stops all stream readers -func (m *Manager) StopAll() { - m.mu.Lock() - readers := make([]*StreamReader, 0, len(m.readers)) - instanceIDs := make([]string, 0, len(m.readers)) - for id, reader := range m.readers { - readers = append(readers, reader) - instanceIDs = append(instanceIDs, id) - } - m.readers = make(map[string]*StreamReader) - m.mu.Unlock() - - // Stop all readers outside the lock - for i, reader := range readers { - if err := reader.Stop(); err != nil { - m.logger.Warn("error stopping stream reader during shutdown", - zap.String("instance_id", instanceIDs[i]), - zap.Error(err)) - } - } - - m.logger.Info("stopped all stream readers", zap.Int("count", len(readers))) -} - -// IsStreaming returns true if streaming is active for an instance -func (m *Manager) IsStreaming(instanceID string) bool { - m.mu.RLock() - defer m.mu.RUnlock() - - reader, exists := m.readers[instanceID] - if !exists { - return false - } - return reader.IsRunning() -} - -// GetActiveCount returns the number of active stream readers -func (m *Manager) GetActiveCount() int { - m.mu.RLock() - defer m.mu.RUnlock() - - count := 0 - for _, reader := range m.readers { - if reader.IsRunning() { - count++ - } - } - return count -} - -// GetReader returns the stream reader for an instance, if it exists -func (m *Manager) GetReader(instanceID string) (*StreamReader, bool) { - m.mu.RLock() - defer m.mu.RUnlock() - - reader, exists := m.readers[instanceID] - return reader, exists -} - diff --git a/apps/backend/internal/agent/streaming/reader.go b/apps/backend/internal/agent/streaming/reader.go deleted file mode 100644 index 168f656ba3..0000000000 --- a/apps/backend/internal/agent/streaming/reader.go +++ /dev/null @@ -1,239 +0,0 @@ -// Package streaming provides ACP message streaming from agent containers. -package streaming - -import ( - "bufio" - "context" - "encoding/json" - "io" - "strings" - "sync" - - "go.uber.org/zap" - - "github.com/kandev/kandev/internal/common/logger" - "github.com/kandev/kandev/internal/events" - "github.com/kandev/kandev/internal/events/bus" - "github.com/kandev/kandev/pkg/acp/protocol" -) - -// StreamReader reads ACP messages from a container's output -type StreamReader struct { - instanceID string - taskID string - eventBus bus.EventBus - logger *logger.Logger - - cancel context.CancelFunc - wg sync.WaitGroup - running bool - mu sync.Mutex -} - -// NewStreamReader creates a new stream reader -func NewStreamReader( - instanceID string, - taskID string, - eventBus bus.EventBus, - log *logger.Logger, -) *StreamReader { - return &StreamReader{ - instanceID: instanceID, - taskID: taskID, - eventBus: eventBus, - logger: log.WithFields( - zap.String("component", "stream-reader"), - zap.String("instance_id", instanceID), - zap.String("task_id", taskID), - ), - } -} - -// Start begins reading from the given reader (container logs) -func (r *StreamReader) Start(ctx context.Context, reader io.ReadCloser) error { - r.mu.Lock() - if r.running { - r.mu.Unlock() - return nil // Already running - } - r.running = true - - // Create a cancellable context - ctx, r.cancel = context.WithCancel(ctx) - r.mu.Unlock() - - r.logger.Info("starting stream reader") - - r.wg.Add(1) - go r.readLoop(ctx, reader) - - return nil -} - -// Stop stops the stream reader -func (r *StreamReader) Stop() error { - r.mu.Lock() - if !r.running { - r.mu.Unlock() - return nil - } - - if r.cancel != nil { - r.cancel() - } - r.mu.Unlock() - - r.wg.Wait() - - r.mu.Lock() - r.running = false - r.mu.Unlock() - - r.logger.Info("stream reader stopped") - return nil -} - -// IsRunning returns true if the reader is active -func (r *StreamReader) IsRunning() bool { - r.mu.Lock() - defer r.mu.Unlock() - return r.running -} - -// readLoop reads lines from the reader and parses ACP messages -func (r *StreamReader) readLoop(ctx context.Context, reader io.ReadCloser) { - defer r.wg.Done() - defer reader.Close() - - scanner := bufio.NewScanner(reader) - // Increase buffer size for potentially large JSON messages - buf := make([]byte, 64*1024) - scanner.Buffer(buf, 1024*1024) - - for { - select { - case <-ctx.Done(): - r.logger.Debug("read loop cancelled") - return - default: - if !scanner.Scan() { - if err := scanner.Err(); err != nil { - r.logger.Error("scanner error", zap.Error(err)) - } else { - r.logger.Debug("EOF reached") - } - return - } - - line := scanner.Text() - r.processLine(ctx, line) - } - } -} - -// processLine processes a single line of output -func (r *StreamReader) processLine(ctx context.Context, line string) { - // Skip empty lines - if len(line) == 0 { - return - } - - // Handle Docker multiplexed output prefix (8 bytes header) - // Docker log format: [8-byte header][payload] - // The header contains stream type and size - cleanedLine := r.stripDockerHeader(line) - if len(cleanedLine) == 0 { - return - } - - // Try to parse as JSON ACP message - msg, err := protocol.Parse([]byte(cleanedLine)) - if err != nil { - // Not valid JSON, treat as regular log line - r.logger.Debug("non-ACP output", zap.String("line", cleanedLine)) - return - } - - // Check if it's a valid ACP message - if !msg.IsValid() { - r.logger.Debug("invalid ACP message (missing required fields)", - zap.String("line", cleanedLine)) - return - } - - // Publish the message - if err := r.publishMessage(ctx, msg); err != nil { - r.logger.Error("failed to publish ACP message", - zap.String("message_type", string(msg.Type)), - zap.Error(err)) - } -} - -// stripDockerHeader removes Docker multiplexed output header and timestamps if present -// Docker log format: [8-byte header][payload] or [timestamp][space][payload] -// Header format: [stream_type(1)][0(3)][size(4)] -// Timestamp format: 2026-01-10T00:42:29.114697835Z {json...} -func (r *StreamReader) stripDockerHeader(line string) string { - // Check if line starts with a Docker stream header - // Stream types: 0=stdin, 1=stdout, 2=stderr - if len(line) < 8 { - return line - } - - // Check if first byte is a valid stream type (0, 1, or 2) - // and next 3 bytes are zeros - if (line[0] == 0 || line[0] == 1 || line[0] == 2) && - line[1] == 0 && line[2] == 0 && line[3] == 0 { - // This looks like a Docker header, skip first 8 bytes - return line[8:] - } - - // Handle Docker timestamp format: "2026-01-10T00:42:29.114697835Z {...}" - // The timestamp is typically 30-35 characters followed by a space and JSON - // Look for JSON start anywhere in the line (for timestamp-prefixed logs) - if idx := strings.Index(line, "{"); idx > 0 { - return line[idx:] - } - - return line -} - -// publishMessage publishes an ACP message to NATS -func (r *StreamReader) publishMessage(ctx context.Context, msg *protocol.Message) error { - if r.eventBus == nil { - return nil - } - - // Build the event data from the ACP message - data := map[string]interface{}{ - "type": string(msg.Type), - "timestamp": msg.Timestamp, - "agent_id": msg.AgentID, - "task_id": msg.TaskID, - "data": msg.Data, - } - - // Create and publish event - event := bus.NewEvent(events.ACPMessage, "agent-manager", data) - subject := events.BuildACPSubject(r.taskID) - - if err := r.eventBus.Publish(ctx, subject, event); err != nil { - return err - } - - r.logger.Debug("published ACP message", - zap.String("subject", subject), - zap.String("message_type", string(msg.Type))) - - return nil -} - -// marshalMessage marshals an ACP message for logging/debugging -func (r *StreamReader) marshalMessage(msg *protocol.Message) string { - data, err := json.Marshal(msg) - if err != nil { - return "" - } - return string(data) -} - diff --git a/apps/backend/internal/agent/worktree/config.go b/apps/backend/internal/agent/worktree/config.go index 960e918578..7f5963ed63 100644 --- a/apps/backend/internal/agent/worktree/config.go +++ b/apps/backend/internal/agent/worktree/config.go @@ -27,15 +27,7 @@ type Config struct { BranchPrefix string `mapstructure:"branch_prefix"` } -// DefaultConfig returns the default worktree configuration. -func DefaultConfig() Config { - return Config{ - Enabled: true, - BasePath: "~/.kandev/worktrees", - MaxPerRepo: 10, - BranchPrefix: "kandev/", - } -} + // Validate validates the configuration and returns an error if invalid. func (c *Config) Validate() error { diff --git a/apps/backend/internal/agentctl/api/control_server.go b/apps/backend/internal/agentctl/api/control_server.go index 43d9905c43..06cc48aaaf 100644 --- a/apps/backend/internal/agentctl/api/control_server.go +++ b/apps/backend/internal/agentctl/api/control_server.go @@ -2,22 +2,33 @@ package api import ( + "context" + "encoding/json" "net/http" "time" "github.com/gin-gonic/gin" + "github.com/gorilla/websocket" "github.com/kandev/kandev/internal/agentctl/config" "github.com/kandev/kandev/internal/agentctl/instance" + "github.com/kandev/kandev/internal/agentctl/process" "github.com/kandev/kandev/internal/common/logger" "go.uber.org/zap" ) // ControlServer provides instance CRUD endpoints on the control port (9999). +// In single-instance mode, it also embeds the agent API routes directly. type ControlServer struct { cfg *config.MultiConfig instMgr *instance.Manager logger *logger.Logger router *gin.Engine + + // Single-instance mode: embedded process manager and config + singleMode bool + singleCfg *config.Config + singleProcMgr *process.Manager + upgrader websocket.Upgrader } // NewControlServer creates a new ControlServer for multi-instance management. @@ -35,6 +46,29 @@ func NewControlServer(cfg *config.MultiConfig, instMgr *instance.Manager, log *l return cs } +// NewSingleInstanceServer creates a ControlServer for single-instance (Docker) mode. +// It embeds the agent API routes directly on the control server. +func NewSingleInstanceServer(cfg *config.MultiConfig, singleCfg *config.Config, procMgr *process.Manager, log *logger.Logger) *ControlServer { + gin.SetMode(gin.ReleaseMode) + + cs := &ControlServer{ + cfg: cfg, + logger: log.WithFields(zap.String("component", "control-server")), + router: gin.New(), + singleMode: true, + singleCfg: singleCfg, + singleProcMgr: procMgr, + upgrader: websocket.Upgrader{ + CheckOrigin: func(r *http.Request) bool { + return true // Allow all origins for container-local communication + }, + }, + } + + cs.setupRoutes() + return cs +} + // Router returns the HTTP handler for the control server. func (m *ControlServer) Router() http.Handler { return m.router @@ -44,9 +78,13 @@ func (m *ControlServer) setupRoutes() { // Health check m.router.GET("/health", m.handleHealth) - // Instance management API api := m.router.Group("/api/v1") - { + + if m.singleMode { + // Single-instance mode: mount agent API routes directly + m.setupSingleInstanceRoutes(api) + } else { + // Multi-instance mode: instance management API api.POST("/instances", m.handleCreateInstance) api.GET("/instances", m.handleListInstances) api.GET("/instances/:id", m.handleGetInstance) @@ -54,6 +92,38 @@ func (m *ControlServer) setupRoutes() { } } +// setupSingleInstanceRoutes adds the agent API routes for single-instance mode. +// These are the same routes that api.Server provides. +func (m *ControlServer) setupSingleInstanceRoutes(api *gin.RouterGroup) { + // Status and info + api.GET("/status", m.handleStatus) + api.GET("/info", m.handleInfo) + + // Process control + api.POST("/start", m.handleStart) + api.POST("/stop", m.handleStop) + + // ACP high-level methods + api.POST("/acp/initialize", m.handleACPInitialize) + api.POST("/acp/session/new", m.handleACPNewSession) + api.POST("/acp/prompt", m.handleACPPrompt) + api.GET("/acp/stream", m.handleACPStreamWS) + + // Permission request handling + api.GET("/acp/permissions", m.handleGetPendingPermissions) + api.GET("/acp/permissions/stream", m.handlePermissionStreamWS) + api.POST("/acp/permissions/respond", m.handlePermissionRespond) + + // Workspace monitoring (git status, files) + api.GET("/workspace/git-status/stream", m.handleGitStatusStreamWS) + api.GET("/workspace/files/stream", m.handleFilesStreamWS) + api.GET("/workspace/file-changes/stream", m.handleFileChangesStreamWS) + + // Workspace file operations (simple HTTP) + api.GET("/workspace/tree", m.handleFileTree) + api.GET("/workspace/file/content", m.handleFileContent) +} + func (m *ControlServer) handleHealth(c *gin.Context) { c.JSON(http.StatusOK, gin.H{ "status": "ok", @@ -136,3 +206,473 @@ func (m *ControlServer) handleDeleteInstance(c *gin.Context) { }) } +// ============================================================================= +// Single-instance mode handlers +// These mirror the handlers in server.go but use singleProcMgr +// ============================================================================= + +func (m *ControlServer) handleStatus(c *gin.Context) { + c.JSON(http.StatusOK, StatusResponse{ + AgentStatus: string(m.singleProcMgr.Status()), + ProcessInfo: m.singleProcMgr.GetProcessInfo(), + }) +} + +func (m *ControlServer) handleInfo(c *gin.Context) { + c.JSON(http.StatusOK, InfoResponse{ + Version: "0.1.0", + AgentCommand: m.singleCfg.AgentCommand, + WorkDir: m.singleCfg.WorkDir, + Port: m.singleCfg.Port, + }) +} + +func (m *ControlServer) handleStart(c *gin.Context) { + if err := m.singleProcMgr.Start(c.Request.Context()); err != nil { + m.logger.Error("failed to start agent", zap.Error(err)) + c.JSON(http.StatusInternalServerError, StartResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + c.JSON(http.StatusOK, StartResponse{ + Success: true, + Message: "agent started", + }) +} + +func (m *ControlServer) handleStop(c *gin.Context) { + if err := m.singleProcMgr.Stop(c.Request.Context()); err != nil { + m.logger.Error("failed to stop agent", zap.Error(err)) + c.JSON(http.StatusInternalServerError, StopResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + c.JSON(http.StatusOK, StopResponse{ + Success: true, + Message: "agent stopped", + }) +} + + +// ACP handlers for single-instance mode + +func (m *ControlServer) handleACPInitialize(c *gin.Context) { + var req InitializeRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, InitializeResponse{ + Success: false, + Error: "invalid request: " + err.Error(), + }) + return + } + + ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) + defer cancel() + + adapter := m.singleProcMgr.GetAdapter() + if adapter == nil { + c.JSON(http.StatusServiceUnavailable, InitializeResponse{ + Success: false, + Error: "agent not running", + }) + return + } + + err := adapter.Initialize(ctx) + if err != nil { + m.logger.Error("initialize failed", zap.Error(err)) + c.JSON(http.StatusInternalServerError, InitializeResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + var agentInfoResp *AgentInfoResponse + if info := adapter.GetAgentInfo(); info != nil { + agentInfoResp = &AgentInfoResponse{ + Name: info.Name, + Version: info.Version, + } + } + + c.JSON(http.StatusOK, InitializeResponse{ + Success: true, + AgentInfo: agentInfoResp, + }) +} + +func (m *ControlServer) handleACPNewSession(c *gin.Context) { + var req NewSessionRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, NewSessionResponse{ + Success: false, + Error: "invalid request: " + err.Error(), + }) + return + } + + ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) + defer cancel() + + adapter := m.singleProcMgr.GetAdapter() + if adapter == nil { + c.JSON(http.StatusServiceUnavailable, NewSessionResponse{ + Success: false, + Error: "agent not running", + }) + return + } + + sessionID, err := adapter.NewSession(ctx) + if err != nil { + m.logger.Error("new session failed", zap.Error(err)) + c.JSON(http.StatusInternalServerError, NewSessionResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + c.JSON(http.StatusOK, NewSessionResponse{ + Success: true, + SessionID: sessionID, + }) +} + +func (m *ControlServer) handleACPPrompt(c *gin.Context) { + var req PromptRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, PromptResponse{ + Success: false, + Error: "invalid request: " + err.Error(), + }) + return + } + + adapter := m.singleProcMgr.GetAdapter() + if adapter == nil { + c.JSON(http.StatusServiceUnavailable, PromptResponse{ + Success: false, + Error: "agent not running", + }) + return + } + + sessionID := m.singleProcMgr.GetSessionID() + if sessionID == "" { + c.JSON(http.StatusBadRequest, PromptResponse{ + Success: false, + Error: "no active session - call new_session first", + }) + return + } + + ctx, cancel := context.WithTimeout(c.Request.Context(), 10*time.Minute) + defer cancel() + + err := adapter.Prompt(ctx, req.Text) + if err != nil { + m.logger.Error("prompt failed", zap.Error(err)) + c.JSON(http.StatusInternalServerError, PromptResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + m.logger.Info("prompt completed") + c.JSON(http.StatusOK, PromptResponse{ + Success: true, + }) +} + + +func (m *ControlServer) handleACPStreamWS(c *gin.Context) { + conn, err := m.upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + m.logger.Error("WebSocket upgrade failed", zap.Error(err)) + return + } + defer conn.Close() + + m.logger.Info("ACP stream WebSocket connected") + + updatesCh := m.singleProcMgr.GetUpdates() + + for { + select { + case notification, ok := <-updatesCh: + if !ok { + return + } + + data, err := json.Marshal(notification) + if err != nil { + m.logger.Error("failed to marshal notification", zap.Error(err)) + continue + } + + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + return + } + } + } +} + +func (m *ControlServer) handleGetPendingPermissions(c *gin.Context) { + pending := m.singleProcMgr.GetPendingPermissions() + + result := make([]PendingPermissionResponse, 0, len(pending)) + for _, p := range pending { + options := make([]PermissionOptionJSON, len(p.Request.Options)) + for i, opt := range p.Request.Options { + options[i] = PermissionOptionJSON{ + OptionID: opt.OptionID, + Name: opt.Name, + Kind: opt.Kind, + } + } + result = append(result, PendingPermissionResponse{ + PendingID: p.ID, + SessionID: p.Request.SessionID, + ToolCallID: p.Request.ToolCallID, + Title: p.Request.Title, + Options: options, + CreatedAt: p.CreatedAt.Format(time.RFC3339), + }) + } + + c.JSON(http.StatusOK, result) +} + +func (m *ControlServer) handlePermissionStreamWS(c *gin.Context) { + conn, err := m.upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + m.logger.Error("WebSocket upgrade failed", zap.Error(err)) + return + } + defer conn.Close() + + m.logger.Info("Permission stream WebSocket connected") + + permissionCh := m.singleProcMgr.GetPermissionRequests() + + for { + select { + case notification, ok := <-permissionCh: + if !ok { + return + } + + data, err := json.Marshal(notification) + if err != nil { + m.logger.Error("failed to marshal permission notification", zap.Error(err)) + continue + } + + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + m.logger.Debug("WebSocket write error", zap.Error(err)) + return + } + } + } +} + +func (m *ControlServer) handlePermissionRespond(c *gin.Context) { + var req PermissionRespondRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, PermissionRespondResponse{ + Success: false, + Error: "invalid request: " + err.Error(), + }) + return + } + + m.logger.Info("received permission response", + zap.String("pending_id", req.PendingID), + zap.String("option_id", req.OptionID), + zap.Bool("cancelled", req.Cancelled)) + + if err := m.singleProcMgr.RespondToPermission(req.PendingID, req.OptionID, req.Cancelled); err != nil { + m.logger.Error("failed to respond to permission", zap.Error(err)) + c.JSON(http.StatusNotFound, PermissionRespondResponse{ + Success: false, + Error: err.Error(), + }) + return + } + + c.JSON(http.StatusOK, PermissionRespondResponse{ + Success: true, + }) +} + + +// Workspace handlers for single-instance mode + +func (m *ControlServer) handleGitStatusStreamWS(c *gin.Context) { + conn, err := m.upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + m.logger.Error("WebSocket upgrade failed", zap.Error(err)) + return + } + defer conn.Close() + + sub := m.singleProcMgr.GetWorkspaceTracker().SubscribeGitStatus() + defer m.singleProcMgr.GetWorkspaceTracker().UnsubscribeGitStatus(sub) + + closeCh := make(chan struct{}) + go func() { + for { + _, _, err := conn.ReadMessage() + if err != nil { + close(closeCh) + return + } + } + }() + + for { + select { + case update, ok := <-sub: + if !ok { + return + } + data, err := json.Marshal(update) + if err != nil { + m.logger.Error("failed to marshal git status update", zap.Error(err)) + continue + } + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + return + } + case <-closeCh: + return + } + } +} + +func (m *ControlServer) handleFilesStreamWS(c *gin.Context) { + conn, err := m.upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + m.logger.Error("WebSocket upgrade failed", zap.Error(err)) + return + } + defer conn.Close() + + sub := m.singleProcMgr.GetWorkspaceTracker().SubscribeFiles() + defer m.singleProcMgr.GetWorkspaceTracker().UnsubscribeFiles(sub) + + closeCh := make(chan struct{}) + go func() { + for { + _, _, err := conn.ReadMessage() + if err != nil { + close(closeCh) + return + } + } + }() + + for { + select { + case update, ok := <-sub: + if !ok { + return + } + data, err := json.Marshal(update) + if err != nil { + m.logger.Error("failed to marshal file update", zap.Error(err)) + continue + } + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + return + } + case <-closeCh: + return + } + } +} + +func (m *ControlServer) handleFileChangesStreamWS(c *gin.Context) { + conn, err := m.upgrader.Upgrade(c.Writer, c.Request, nil) + if err != nil { + m.logger.Error("WebSocket upgrade failed", zap.Error(err)) + return + } + defer conn.Close() + + sub := m.singleProcMgr.GetWorkspaceTracker().SubscribeFileChanges() + defer m.singleProcMgr.GetWorkspaceTracker().UnsubscribeFileChanges(sub) + + closeCh := make(chan struct{}) + go func() { + for { + _, _, err := conn.ReadMessage() + if err != nil { + close(closeCh) + return + } + } + }() + + for { + select { + case notification, ok := <-sub: + if !ok { + return + } + data, err := json.Marshal(notification) + if err != nil { + m.logger.Error("failed to marshal file change notification", zap.Error(err)) + continue + } + if err := conn.WriteMessage(websocket.TextMessage, data); err != nil { + return + } + case <-closeCh: + return + } + } +} + +func (m *ControlServer) handleFileTree(c *gin.Context) { + path := c.Query("path") + depth := 1 + if d := c.Query("depth"); d != "" { + if _, err := json.Number(d).Int64(); err == nil { + depth = int(mustParseInt(d)) + } + } + + tree, err := m.singleProcMgr.GetWorkspaceTracker().GetFileTree(path, depth) + if err != nil { + c.JSON(400, process.FileTreeResponse{Error: err.Error()}) + return + } + + c.JSON(200, process.FileTreeResponse{Root: tree}) +} + +func (m *ControlServer) handleFileContent(c *gin.Context) { + path := c.Query("path") + if path == "" { + c.JSON(400, process.FileContentResponse{Error: "path is required"}) + return + } + + content, size, err := m.singleProcMgr.GetWorkspaceTracker().GetFileContent(path) + if err != nil { + c.JSON(400, process.FileContentResponse{Path: path, Error: err.Error(), Size: size}) + return + } + + c.JSON(200, process.FileContentResponse{Path: path, Content: content, Size: size}) +} \ No newline at end of file diff --git a/apps/backend/internal/agentctl/instance/port_allocator.go b/apps/backend/internal/agentctl/instance/port_allocator.go index 64cca92a77..edf4ed5001 100644 --- a/apps/backend/internal/agentctl/instance/port_allocator.go +++ b/apps/backend/internal/agentctl/instance/port_allocator.go @@ -53,12 +53,3 @@ func (p *PortAllocator) Release(port int) { delete(p.allocated, port) } -// IsAvailable checks if a specific port is available for allocation. -func (p *PortAllocator) IsAvailable(port int) bool { - p.mu.Lock() - defer p.mu.Unlock() - - _, exists := p.allocated[port] - return !exists -} - diff --git a/apps/backend/internal/agentctl/process/manager.go b/apps/backend/internal/agentctl/process/manager.go index 37f379764b..141a3057fc 100644 --- a/apps/backend/internal/agentctl/process/manager.go +++ b/apps/backend/internal/agentctl/process/manager.go @@ -563,10 +563,4 @@ func (m *Manager) GetPendingPermissions() []*PendingPermission { return result } -// GetPendingPermission returns a specific pending permission request -func (m *Manager) GetPendingPermission(pendingID string) (*PendingPermission, bool) { - m.permissionMu.RLock() - defer m.permissionMu.RUnlock() - p, ok := m.pendingPermissions[pendingID] - return p, ok -} + diff --git a/apps/backend/internal/agentctl/process/workspace_tracker.go b/apps/backend/internal/agentctl/process/workspace_tracker.go index 0845cbccc2..f3e4f31362 100644 --- a/apps/backend/internal/agentctl/process/workspace_tracker.go +++ b/apps/backend/internal/agentctl/process/workspace_tracker.go @@ -558,19 +558,7 @@ func (wt *WorkspaceTracker) notifyFilesSubscribers(update FileListUpdate) { } } -// GetCurrentGitStatus returns the current git status -func (wt *WorkspaceTracker) GetCurrentGitStatus() GitStatusUpdate { - wt.mu.RLock() - defer wt.mu.RUnlock() - return wt.currentStatus -} -// GetCurrentFiles returns the current file list -func (wt *WorkspaceTracker) GetCurrentFiles() FileListUpdate { - wt.mu.RLock() - defer wt.mu.RUnlock() - return wt.currentFiles -} // SubscribeFileChanges creates a new file change subscriber func (wt *WorkspaceTracker) SubscribeFileChanges() FileChangeSubscriber { diff --git a/apps/backend/internal/common/database/database.go b/apps/backend/internal/common/database/database.go deleted file mode 100644 index c74743784e..0000000000 --- a/apps/backend/internal/common/database/database.go +++ /dev/null @@ -1,152 +0,0 @@ -// Package database provides PostgreSQL connection pooling and database operations. -package database - -import ( - "context" - "fmt" - "time" - - "github.com/jackc/pgx/v5" - "github.com/jackc/pgx/v5/pgconn" - "github.com/jackc/pgx/v5/pgxpool" - - "github.com/kandev/kandev/internal/common/config" -) - -// DB wraps a pgxpool.Pool and provides helper methods for database operations. -type DB struct { - pool *pgxpool.Pool -} - -// NewDB creates a new database connection pool using the provided configuration. -// It builds the connection string from config, configures pool settings, -// establishes the connection, and verifies it with a ping. -func NewDB(ctx context.Context, cfg config.DatabaseConfig) (*DB, error) { - connString := cfg.DSN() - - poolConfig, err := pgxpool.ParseConfig(connString) - if err != nil { - return nil, fmt.Errorf("failed to parse database config: %w", err) - } - - // Configure pool settings - poolConfig.MaxConns = int32(cfg.MaxConns) - poolConfig.MinConns = int32(cfg.MinConns) - - // Set reasonable connection timeouts - poolConfig.ConnConfig.ConnectTimeout = 10 * time.Second - - // Create the connection pool - pool, err := pgxpool.NewWithConfig(ctx, poolConfig) - if err != nil { - return nil, fmt.Errorf("failed to create connection pool: %w", err) - } - - // Verify the connection - if err := pool.Ping(ctx); err != nil { - pool.Close() - return nil, fmt.Errorf("failed to ping database: %w", err) - } - - return &DB{pool: pool}, nil -} - -// Pool returns the underlying pgxpool.Pool. -func (db *DB) Pool() *pgxpool.Pool { - return db.pool -} - -// Close closes the connection pool. -func (db *DB) Close() { - if db.pool != nil { - db.pool.Close() - } -} - -// Ping verifies the database connection is still alive. -func (db *DB) Ping(ctx context.Context) error { - return db.pool.Ping(ctx) -} - -// Exec executes a query that doesn't return rows. -func (db *DB) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) { - return db.pool.Exec(ctx, sql, args...) -} - -// Query executes a query that returns rows. -func (db *DB) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) { - return db.pool.Query(ctx, sql, args...) -} - -// QueryRow executes a query that returns at most one row. -func (db *DB) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row { - return db.pool.QueryRow(ctx, sql, args...) -} - -// BeginTx starts a transaction with the given options. -func (db *DB) BeginTx(ctx context.Context, opts pgx.TxOptions) (pgx.Tx, error) { - return db.pool.BeginTx(ctx, opts) -} - -// WithTx executes the given function within a transaction. -// If the function returns an error, the transaction is rolled back. -// If the function succeeds, the transaction is committed. -func (db *DB) WithTx(ctx context.Context, fn func(tx pgx.Tx) error) error { - tx, err := db.pool.Begin(ctx) - if err != nil { - return fmt.Errorf("failed to begin transaction: %w", err) - } - - defer func() { - if p := recover(); p != nil { - // Rollback on panic - _ = tx.Rollback(ctx) - panic(p) - } - }() - - if err := fn(tx); err != nil { - if rbErr := tx.Rollback(ctx); rbErr != nil { - return fmt.Errorf("tx failed: %w, rollback failed: %v", err, rbErr) - } - return err - } - - if err := tx.Commit(ctx); err != nil { - return fmt.Errorf("failed to commit transaction: %w", err) - } - - return nil -} - -// WithTxOptions executes the given function within a transaction with custom options. -// If the function returns an error, the transaction is rolled back. -// If the function succeeds, the transaction is committed. -func (db *DB) WithTxOptions(ctx context.Context, opts pgx.TxOptions, fn func(tx pgx.Tx) error) error { - tx, err := db.pool.BeginTx(ctx, opts) - if err != nil { - return fmt.Errorf("failed to begin transaction: %w", err) - } - - defer func() { - if p := recover(); p != nil { - // Rollback on panic - _ = tx.Rollback(ctx) - panic(p) - } - }() - - if err := fn(tx); err != nil { - if rbErr := tx.Rollback(ctx); rbErr != nil { - return fmt.Errorf("tx failed: %w, rollback failed: %v", err, rbErr) - } - return err - } - - if err := tx.Commit(ctx); err != nil { - return fmt.Errorf("failed to commit transaction: %w", err) - } - - return nil -} - diff --git a/apps/backend/internal/common/errors/errors.go b/apps/backend/internal/common/errors/errors.go deleted file mode 100644 index fa3e33d7be..0000000000 --- a/apps/backend/internal/common/errors/errors.go +++ /dev/null @@ -1,169 +0,0 @@ -// Package errors provides custom error types for the Kandev application. -package errors - -import ( - "errors" - "fmt" - "net/http" -) - -// Error codes as constants -const ( - ErrCodeNotFound = "NOT_FOUND" - ErrCodeBadRequest = "BAD_REQUEST" - ErrCodeUnauthorized = "UNAUTHORIZED" - ErrCodeForbidden = "FORBIDDEN" - ErrCodeInternalError = "INTERNAL_ERROR" - ErrCodeConflict = "CONFLICT" - ErrCodeValidationError = "VALIDATION_ERROR" - ErrCodeServiceUnavailable = "SERVICE_UNAVAILABLE" -) - -// AppError represents an application-specific error with additional context. -type AppError struct { - Code string `json:"code"` - Message string `json:"message"` - HTTPStatus int `json:"http_status"` - Err error `json:"-"` -} - -// Error implements the error interface. -func (e *AppError) Error() string { - if e.Err != nil { - return fmt.Sprintf("%s: %s: %v", e.Code, e.Message, e.Err) - } - return fmt.Sprintf("%s: %s", e.Code, e.Message) -} - -// Unwrap returns the wrapped error for use with errors.Is and errors.As. -func (e *AppError) Unwrap() error { - return e.Err -} - -// NotFound creates a new not found error for a resource. -func NotFound(resource string, id string) *AppError { - return &AppError{ - Code: ErrCodeNotFound, - Message: fmt.Sprintf("%s with id '%s' not found", resource, id), - HTTPStatus: http.StatusNotFound, - } -} - -// BadRequest creates a new bad request error. -func BadRequest(message string) *AppError { - return &AppError{ - Code: ErrCodeBadRequest, - Message: message, - HTTPStatus: http.StatusBadRequest, - } -} - -// Unauthorized creates a new unauthorized error. -func Unauthorized(message string) *AppError { - return &AppError{ - Code: ErrCodeUnauthorized, - Message: message, - HTTPStatus: http.StatusUnauthorized, - } -} - -// Forbidden creates a new forbidden error. -func Forbidden(message string) *AppError { - return &AppError{ - Code: ErrCodeForbidden, - Message: message, - HTTPStatus: http.StatusForbidden, - } -} - -// InternalError creates a new internal server error with a wrapped underlying error. -func InternalError(message string, err error) *AppError { - return &AppError{ - Code: ErrCodeInternalError, - Message: message, - HTTPStatus: http.StatusInternalServerError, - Err: err, - } -} - -// Conflict creates a new conflict error. -func Conflict(message string) *AppError { - return &AppError{ - Code: ErrCodeConflict, - Message: message, - HTTPStatus: http.StatusConflict, - } -} - -// ValidationError creates a new validation error for a specific field. -func ValidationError(field string, message string) *AppError { - return &AppError{ - Code: ErrCodeValidationError, - Message: fmt.Sprintf("validation failed for field '%s': %s", field, message), - HTTPStatus: http.StatusBadRequest, - } -} - -// ServiceUnavailable creates a new service unavailable error. -func ServiceUnavailable(service string) *AppError { - return &AppError{ - Code: ErrCodeServiceUnavailable, - Message: fmt.Sprintf("service '%s' is currently unavailable", service), - HTTPStatus: http.StatusServiceUnavailable, - } -} - -// Wrap wraps an existing error with additional context, returning an AppError. -func Wrap(err error, message string) *AppError { - if err == nil { - return nil - } - - // If the error is already an AppError, preserve its code and status - var appErr *AppError - if errors.As(err, &appErr) { - return &AppError{ - Code: appErr.Code, - Message: fmt.Sprintf("%s: %s", message, appErr.Message), - HTTPStatus: appErr.HTTPStatus, - Err: err, - } - } - - // Otherwise, wrap as an internal error - return &AppError{ - Code: ErrCodeInternalError, - Message: message, - HTTPStatus: http.StatusInternalServerError, - Err: err, - } -} - -// IsNotFound checks if the error is a not found error. -func IsNotFound(err error) bool { - var appErr *AppError - if errors.As(err, &appErr) { - return appErr.Code == ErrCodeNotFound - } - return false -} - -// IsBadRequest checks if the error is a bad request error. -func IsBadRequest(err error) bool { - var appErr *AppError - if errors.As(err, &appErr) { - return appErr.Code == ErrCodeBadRequest || appErr.Code == ErrCodeValidationError - } - return false -} - -// GetHTTPStatus returns the HTTP status code for an error. -// Returns 500 Internal Server Error if the error is not an AppError. -func GetHTTPStatus(err error) int { - var appErr *AppError - if errors.As(err, &appErr) { - return appErr.HTTPStatus - } - return http.StatusInternalServerError -} - diff --git a/apps/backend/internal/common/stringutil/stringutil.go b/apps/backend/internal/common/stringutil/stringutil.go deleted file mode 100644 index 2b3e5fd6d9..0000000000 --- a/apps/backend/internal/common/stringutil/stringutil.go +++ /dev/null @@ -1,26 +0,0 @@ -// Package stringutil provides common string utility functions. -package stringutil - -// TruncateString truncates a string to a maximum length. -// If the string is shorter than maxLen, it returns the original string. -// If the string is longer, it returns the first maxLen characters. -func TruncateString(s string, maxLen int) string { - if len(s) <= maxLen { - return s - } - return s[:maxLen] -} - -// TruncateStringWithEllipsis truncates a string to a maximum length and adds "..." suffix. -// If the string is shorter than maxLen, it returns the original string. -// If the string is longer, it returns the first (maxLen-3) characters followed by "...". -func TruncateStringWithEllipsis(s string, maxLen int) string { - if maxLen < 4 { - return TruncateString(s, maxLen) - } - if len(s) <= maxLen { - return s - } - return s[:maxLen-3] + "..." -} - diff --git a/apps/backend/internal/gateway/websocket/handler.go b/apps/backend/internal/gateway/websocket/handler.go index b97cfc3f82..e24e48c405 100644 --- a/apps/backend/internal/gateway/websocket/handler.go +++ b/apps/backend/internal/gateway/websocket/handler.go @@ -72,13 +72,7 @@ func (h *Handler) HandleConnection(c *gin.Context) { client.ReadPump(c.Request.Context()) } -// SetupRoutes adds the WebSocket route to the router -func SetupRoutes(router *gin.Engine, hub *Hub, log *logger.Logger) { - handler := NewHandler(hub, log) - // Main WebSocket endpoint - router.GET("/ws", handler.HandleConnection) -} // RegisterHealthHandler registers the health check handler func RegisterHealthHandler(d *ws.Dispatcher) { diff --git a/apps/backend/internal/integration/integration_test.go b/apps/backend/internal/integration/integration_test.go index efcc93a6b9..85c5696deb 100644 --- a/apps/backend/internal/integration/integration_test.go +++ b/apps/backend/internal/integration/integration_test.go @@ -55,8 +55,13 @@ func NewTestServer(t *testing.T) *TestServer { // Initialize event bus eventBus := bus.NewMemoryEventBus(log) - // Initialize task repository (in-memory for tests) - taskRepo := repository.NewMemoryRepository() + // Initialize task repository (SQLite for tests) + tmpDir := t.TempDir() + taskRepo, err := repository.NewSQLiteRepository(tmpDir + "/test.db") + if err != nil { + t.Fatalf("failed to create test repository: %v", err) + } + t.Cleanup(func() { taskRepo.Close() }) // Initialize task service taskSvc := taskservice.NewService(taskRepo, eventBus, log, taskservice.RepositoryDiscoveryConfig{}) @@ -728,8 +733,10 @@ func TestMultipleClients(t *testing.T) { require.NoError(t, err) boardID := boardPayload["id"].(string) - // Client 2 can see the board - listResp, err := client2.SendRequest("c2-list-1", ws.ActionBoardList, map[string]interface{}{}) + // Client 2 can see the board (filter by workspace to avoid default board) + listResp, err := client2.SendRequest("c2-list-1", ws.ActionBoardList, map[string]interface{}{ + "workspace_id": workspaceID, + }) require.NoError(t, err) var listPayload map[string]interface{} diff --git a/apps/backend/internal/integration/orchestrator_test.go b/apps/backend/internal/integration/orchestrator_test.go index a9d69b2763..ca3af69e45 100644 --- a/apps/backend/internal/integration/orchestrator_test.go +++ b/apps/backend/internal/integration/orchestrator_test.go @@ -468,8 +468,13 @@ func NewOrchestratorTestServer(t *testing.T) *OrchestratorTestServer { // Initialize event bus eventBus := bus.NewMemoryEventBus(log) - // Initialize task repository (in-memory for tests) - taskRepo := repository.NewMemoryRepository() + // Initialize task repository (SQLite for tests) + tmpDir := t.TempDir() + taskRepo, err := repository.NewSQLiteRepository(tmpDir + "/test.db") + if err != nil { + t.Fatalf("failed to create test repository: %v", err) + } + t.Cleanup(func() { taskRepo.Close() }) // Initialize task service taskSvc := taskservice.NewService(taskRepo, eventBus, log, taskservice.RepositoryDiscoveryConfig{}) @@ -483,7 +488,7 @@ func NewOrchestratorTestServer(t *testing.T) *OrchestratorTestServer { // Create orchestrator service cfg := orchestrator.DefaultServiceConfig() cfg.Scheduler.ProcessInterval = 50 * time.Millisecond // Faster for tests - orchestratorSvc := orchestrator.NewService(cfg, eventBus, nil, agentManager, taskRepoAdapter, taskRepo, log) + orchestratorSvc := orchestrator.NewService(cfg, eventBus, agentManager, taskRepoAdapter, taskRepo, log) // Create WebSocket gateway gateway := gateways.NewGateway(log) diff --git a/apps/backend/internal/orchestrator/dto/dto.go b/apps/backend/internal/orchestrator/dto/dto.go index fb7a118b66..c6dfdcdd03 100644 --- a/apps/backend/internal/orchestrator/dto/dto.go +++ b/apps/backend/internal/orchestrator/dto/dto.go @@ -1,8 +1,6 @@ // Package dto provides Data Transfer Objects for the orchestrator module. package dto -import "time" - // GetStatusRequest is the request for orchestrator.status type GetStatusRequest struct{} @@ -109,14 +107,7 @@ type PermissionRespondResponse struct { PendingID string `json:"pending_id"` } -// FromQueuedTask converts a queue.QueuedTask to QueuedTaskDTO -func FromQueuedTask(qt interface{ GetTaskID() string; GetPriority() int; GetQueuedAt() time.Time }) QueuedTaskDTO { - return QueuedTaskDTO{ - TaskID: qt.GetTaskID(), - Priority: qt.GetPriority(), - QueuedAt: qt.GetQueuedAt().Format("2006-01-02T15:04:05Z"), - } -} + // GetTaskExecutionRequest is the payload for task.execution type GetTaskExecutionRequest struct { diff --git a/apps/backend/internal/orchestrator/scheduler/scheduler.go b/apps/backend/internal/orchestrator/scheduler/scheduler.go index 123d395af8..fb75d4053b 100644 --- a/apps/backend/internal/orchestrator/scheduler/scheduler.go +++ b/apps/backend/internal/orchestrator/scheduler/scheduler.go @@ -327,61 +327,3 @@ func (s *Scheduler) processTasks(ctx context.Context) { } // MockTaskRepository is a placeholder implementation -type MockTaskRepository struct { - tasks map[string]*v1.Task - mu sync.RWMutex - logger *logger.Logger -} - -// NewMockTaskRepository creates a new mock task repository -func NewMockTaskRepository(log *logger.Logger) *MockTaskRepository { - return &MockTaskRepository{ - tasks: make(map[string]*v1.Task), - logger: log.WithFields(zap.String("component", "mock_task_repo")), - } -} - -// GetTask retrieves a task by ID -func (r *MockTaskRepository) GetTask(ctx context.Context, taskID string) (*v1.Task, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - task, exists := r.tasks[taskID] - if !exists { - r.logger.Debug("task not found", zap.String("task_id", taskID)) - return nil, ErrTaskNotFound - } - - // Return a copy - taskCopy := *task - return &taskCopy, nil -} - -// UpdateTaskState updates the state of a task -func (r *MockTaskRepository) UpdateTaskState(ctx context.Context, taskID string, state v1.TaskState) error { - r.mu.Lock() - defer r.mu.Unlock() - - task, exists := r.tasks[taskID] - if !exists { - r.logger.Debug("task not found for state update", zap.String("task_id", taskID)) - return ErrTaskNotFound - } - - task.State = state - task.UpdatedAt = time.Now() - r.logger.Info("task state updated", - zap.String("task_id", taskID), - zap.String("state", string(state))) - return nil -} - -// AddTask adds a task to the mock repository -func (r *MockTaskRepository) AddTask(task *v1.Task) { - r.mu.Lock() - defer r.mu.Unlock() - - r.tasks[task.ID] = task - r.logger.Debug("task added to mock repository", zap.String("task_id", task.ID)) -} - diff --git a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go index 814f0d8ffe..abc6eb00f9 100644 --- a/apps/backend/internal/orchestrator/scheduler/scheduler_test.go +++ b/apps/backend/internal/orchestrator/scheduler/scheduler_test.go @@ -120,8 +120,13 @@ func createTestLogger() *logger.Logger { return log } -func createTestExecutor(agentMgr *mockAgentManager, log *logger.Logger, maxConcurrent int) *executor.Executor { - repo := repository.NewMemoryRepository() +func createTestExecutor(t *testing.T, agentMgr *mockAgentManager, log *logger.Logger, maxConcurrent int) *executor.Executor { + tmpDir := t.TempDir() + repo, err := repository.NewSQLiteRepository(tmpDir + "/test.db") + if err != nil { + t.Fatalf("failed to create test repository: %v", err) + } + t.Cleanup(func() { repo.Close() }) cfg := executor.ExecutorConfig{ MaxConcurrent: maxConcurrent, WorktreeEnabled: true, @@ -147,7 +152,7 @@ func TestNewScheduler(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() s := NewScheduler(q, exec, taskRepo, log, DefaultSchedulerConfig()) @@ -180,7 +185,7 @@ func TestStartStop(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() s := NewScheduler(q, exec, taskRepo, log, DefaultSchedulerConfig()) @@ -206,7 +211,7 @@ func TestStartAlreadyRunning(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() s := NewScheduler(q, exec, taskRepo, log, DefaultSchedulerConfig()) @@ -224,7 +229,7 @@ func TestStopNotRunning(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() s := NewScheduler(q, exec, taskRepo, log, DefaultSchedulerConfig()) @@ -239,7 +244,7 @@ func TestEnqueueTask(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() s := NewScheduler(q, exec, taskRepo, log, DefaultSchedulerConfig()) @@ -259,7 +264,7 @@ func TestRemoveTask(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() s := NewScheduler(q, exec, taskRepo, log, DefaultSchedulerConfig()) @@ -281,7 +286,7 @@ func TestRemoveNonExistentTask(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() s := NewScheduler(q, exec, taskRepo, log, DefaultSchedulerConfig()) @@ -296,7 +301,7 @@ func TestGetQueueStatus(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() cfg := DefaultSchedulerConfig() @@ -322,7 +327,7 @@ func TestHandleTaskCompleted(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() s := NewScheduler(q, exec, taskRepo, log, DefaultSchedulerConfig()) @@ -351,7 +356,7 @@ func TestPriorityOrdering(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() s := NewScheduler(q, exec, taskRepo, log, DefaultSchedulerConfig()) @@ -372,7 +377,7 @@ func TestIsRunning(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() s := NewScheduler(q, exec, taskRepo, log, DefaultSchedulerConfig()) @@ -396,7 +401,7 @@ func TestSchedulerWithContextCancellation(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() cfg := DefaultSchedulerConfig() @@ -425,7 +430,7 @@ func TestEnqueueDuplicateTask(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() s := NewScheduler(q, exec, taskRepo, log, DefaultSchedulerConfig()) @@ -443,7 +448,7 @@ func TestRetryTaskExceedsLimit(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() cfg := DefaultSchedulerConfig() @@ -477,7 +482,7 @@ func TestRetryTaskNotFound(t *testing.T) { q := queue.NewTaskQueue(100) agentMgr := newMockAgentManager() log := createTestLogger() - exec := createTestExecutor(agentMgr, log, 5) + exec := createTestExecutor(t, agentMgr, log, 5) taskRepo := newTestTaskRepository() cfg := DefaultSchedulerConfig() diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 217a837adb..661952bc84 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -10,7 +10,6 @@ import ( "go.uber.org/zap" - "github.com/kandev/kandev/internal/common/database" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/orchestrator/executor" @@ -59,7 +58,6 @@ type Service struct { config ServiceConfig logger *logger.Logger eventBus bus.EventBus - db *database.DB taskRepo scheduler.TaskRepository repo repository.Repository // Full repository for agent sessions agentManager executor.AgentManagerClient @@ -101,7 +99,6 @@ type Status struct { func NewService( cfg ServiceConfig, eventBus bus.EventBus, - db *database.DB, agentManager executor.AgentManagerClient, taskRepo scheduler.TaskRepository, repo repository.Repository, @@ -127,7 +124,6 @@ func NewService( config: cfg, logger: svcLogger, eventBus: eventBus, - db: db, taskRepo: taskRepo, repo: repo, agentManager: agentManager, diff --git a/apps/backend/internal/task/repository/memory.go b/apps/backend/internal/task/repository/memory.go deleted file mode 100644 index e9593681a0..0000000000 --- a/apps/backend/internal/task/repository/memory.go +++ /dev/null @@ -1,987 +0,0 @@ -package repository - -import ( - "context" - "fmt" - "sync" - "time" - - "github.com/google/uuid" - "github.com/kandev/kandev/internal/task/models" - v1 "github.com/kandev/kandev/pkg/api/v1" -) - -// MemoryRepository provides in-memory task storage operations -type MemoryRepository struct { - workspaces map[string]*models.Workspace - tasks map[string]*models.Task - boards map[string]*models.Board - columns map[string]*models.Column - comments map[string]*models.Comment - repositories map[string]*models.Repository - repoScripts map[string]*models.RepositoryScript - agentSessions map[string]*models.AgentSession - executors map[string]*models.Executor - environments map[string]*models.Environment - taskBoards map[string]map[string]struct{} - taskPlacements map[string]map[string]*taskPlacement - mu sync.RWMutex -} - -// Ensure MemoryRepository implements Repository interface -var _ Repository = (*MemoryRepository)(nil) - -// NewMemoryRepository creates a new in-memory task repository -func NewMemoryRepository() *MemoryRepository { - repo := &MemoryRepository{ - workspaces: make(map[string]*models.Workspace), - tasks: make(map[string]*models.Task), - boards: make(map[string]*models.Board), - columns: make(map[string]*models.Column), - comments: make(map[string]*models.Comment), - repositories: make(map[string]*models.Repository), - repoScripts: make(map[string]*models.RepositoryScript), - agentSessions: make(map[string]*models.AgentSession), - executors: make(map[string]*models.Executor), - environments: make(map[string]*models.Environment), - taskBoards: make(map[string]map[string]struct{}), - taskPlacements: make(map[string]map[string]*taskPlacement), - } - repo.seedDefaults() - return repo -} - -func (r *MemoryRepository) seedDefaults() { - now := time.Now().UTC() - r.executors[models.ExecutorIDLocalPC] = &models.Executor{ - ID: models.ExecutorIDLocalPC, - Name: "Local PC", - Type: models.ExecutorTypeLocalPC, - Status: models.ExecutorStatusActive, - IsSystem: true, - Config: map[string]string{}, - CreatedAt: now, - UpdatedAt: now, - } - r.executors[models.ExecutorIDLocalDocker] = &models.Executor{ - ID: models.ExecutorIDLocalDocker, - Name: "Local Docker", - Type: models.ExecutorTypeLocalDocker, - Status: models.ExecutorStatusActive, - IsSystem: false, - Config: map[string]string{"docker_host": "unix:///var/run/docker.sock"}, - CreatedAt: now, - UpdatedAt: now, - } - r.executors[models.ExecutorIDRemoteDocker] = &models.Executor{ - ID: models.ExecutorIDRemoteDocker, - Name: "Remote Docker", - Type: models.ExecutorTypeRemoteDocker, - Status: models.ExecutorStatusDisabled, - IsSystem: false, - Config: map[string]string{}, - CreatedAt: now, - UpdatedAt: now, - } - r.environments[models.EnvironmentIDLocal] = &models.Environment{ - ID: models.EnvironmentIDLocal, - Name: "Local", - Kind: models.EnvironmentKindLocalPC, - WorktreeRoot: "~/kandev", - CreatedAt: now, - UpdatedAt: now, - } -} - -type taskPlacement struct { - boardID string - columnID string - position int -} - -// Close is a no-op for in-memory repository -func (r *MemoryRepository) Close() error { - return nil -} - -// Workspace operations - -// CreateWorkspace creates a new workspace -func (r *MemoryRepository) CreateWorkspace(ctx context.Context, workspace *models.Workspace) error { - r.mu.Lock() - defer r.mu.Unlock() - - if workspace.ID == "" { - workspace.ID = uuid.New().String() - } - now := time.Now().UTC() - workspace.CreatedAt = now - workspace.UpdatedAt = now - - r.workspaces[workspace.ID] = workspace - return nil -} - -// GetWorkspace retrieves a workspace by ID -func (r *MemoryRepository) GetWorkspace(ctx context.Context, id string) (*models.Workspace, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - workspace, ok := r.workspaces[id] - if !ok { - return nil, fmt.Errorf("workspace not found: %s", id) - } - return workspace, nil -} - -// UpdateWorkspace updates an existing workspace -func (r *MemoryRepository) UpdateWorkspace(ctx context.Context, workspace *models.Workspace) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.workspaces[workspace.ID]; !ok { - return fmt.Errorf("workspace not found: %s", workspace.ID) - } - workspace.UpdatedAt = time.Now().UTC() - r.workspaces[workspace.ID] = workspace - return nil -} - -// DeleteWorkspace deletes a workspace by ID -func (r *MemoryRepository) DeleteWorkspace(ctx context.Context, id string) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.workspaces[id]; !ok { - return fmt.Errorf("workspace not found: %s", id) - } - delete(r.workspaces, id) - return nil -} - -// ListWorkspaces returns all workspaces -func (r *MemoryRepository) ListWorkspaces(ctx context.Context) ([]*models.Workspace, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - result := make([]*models.Workspace, 0, len(r.workspaces)) - for _, workspace := range r.workspaces { - result = append(result, workspace) - } - return result, nil -} - -// Task operations - -// CreateTask creates a new task -func (r *MemoryRepository) CreateTask(ctx context.Context, task *models.Task) error { - r.mu.Lock() - defer r.mu.Unlock() - - if task.ID == "" { - task.ID = uuid.New().String() - } - now := time.Now().UTC() - task.CreatedAt = now - task.UpdatedAt = now - - r.tasks[task.ID] = task - if task.BoardID != "" { - r.addTaskPlacement(task.ID, task.BoardID, task.ColumnID, task.Position) - } - return nil -} - -// GetTask retrieves a task by ID -func (r *MemoryRepository) GetTask(ctx context.Context, id string) (*models.Task, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - task, ok := r.tasks[id] - if !ok { - return nil, fmt.Errorf("task not found: %s", id) - } - return task, nil -} - -// UpdateTask updates an existing task -func (r *MemoryRepository) UpdateTask(ctx context.Context, task *models.Task) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.tasks[task.ID]; !ok { - return fmt.Errorf("task not found: %s", task.ID) - } - task.UpdatedAt = time.Now().UTC() - r.tasks[task.ID] = task - if task.BoardID != "" { - r.addTaskPlacement(task.ID, task.BoardID, task.ColumnID, task.Position) - } - return nil -} - -// DeleteTask deletes a task by ID -func (r *MemoryRepository) DeleteTask(ctx context.Context, id string) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.tasks[id]; !ok { - return fmt.Errorf("task not found: %s", id) - } - delete(r.tasks, id) - delete(r.taskBoards, id) - delete(r.taskPlacements, id) - return nil -} - -// ListTasks returns all tasks for a board -func (r *MemoryRepository) ListTasks(ctx context.Context, boardID string) ([]*models.Task, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - var result []*models.Task - for _, task := range r.tasks { - placement, ok := r.getTaskPlacement(task.ID, boardID) - if !ok { - continue - } - copy := *task - copy.BoardID = placement.boardID - copy.ColumnID = placement.columnID - copy.Position = placement.position - result = append(result, ©) - } - return result, nil -} - -// ListTasksByColumn returns all tasks in a column -func (r *MemoryRepository) ListTasksByColumn(ctx context.Context, columnID string) ([]*models.Task, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - var result []*models.Task - for _, task := range r.tasks { - placements, ok := r.taskPlacements[task.ID] - if !ok { - continue - } - for _, placement := range placements { - if placement.columnID != columnID { - continue - } - copy := *task - copy.BoardID = placement.boardID - copy.ColumnID = placement.columnID - copy.Position = placement.position - result = append(result, ©) - } - } - return result, nil -} - -// UpdateTaskState updates the state of a task -func (r *MemoryRepository) UpdateTaskState(ctx context.Context, id string, state v1.TaskState) error { - r.mu.Lock() - defer r.mu.Unlock() - - task, ok := r.tasks[id] - if !ok { - return fmt.Errorf("task not found: %s", id) - } - task.State = state - task.UpdatedAt = time.Now().UTC() - return nil -} - -// AddTaskToBoard adds a task to a board with placement -func (r *MemoryRepository) AddTaskToBoard(ctx context.Context, taskID, boardID, columnID string, position int) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.tasks[taskID]; !ok { - return fmt.Errorf("task not found: %s", taskID) - } - r.addTaskPlacement(taskID, boardID, columnID, position) - return nil -} - -// RemoveTaskFromBoard removes a task from a board -func (r *MemoryRepository) RemoveTaskFromBoard(ctx context.Context, taskID, boardID string) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.tasks[taskID]; !ok { - return fmt.Errorf("task not found: %s", taskID) - } - if placements, ok := r.taskPlacements[taskID]; ok { - delete(placements, boardID) - } - if boards, ok := r.taskBoards[taskID]; ok { - delete(boards, boardID) - } - return nil -} - -// Board operations - -// CreateBoard creates a new board -func (r *MemoryRepository) CreateBoard(ctx context.Context, board *models.Board) error { - r.mu.Lock() - defer r.mu.Unlock() - - if board.ID == "" { - board.ID = uuid.New().String() - } - now := time.Now().UTC() - board.CreatedAt = now - board.UpdatedAt = now - - r.boards[board.ID] = board - return nil -} - -// GetBoard retrieves a board by ID -func (r *MemoryRepository) GetBoard(ctx context.Context, id string) (*models.Board, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - board, ok := r.boards[id] - if !ok { - return nil, fmt.Errorf("board not found: %s", id) - } - return board, nil -} - -// UpdateBoard updates an existing board -func (r *MemoryRepository) UpdateBoard(ctx context.Context, board *models.Board) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.boards[board.ID]; !ok { - return fmt.Errorf("board not found: %s", board.ID) - } - board.UpdatedAt = time.Now().UTC() - r.boards[board.ID] = board - return nil -} - -// DeleteBoard deletes a board by ID -func (r *MemoryRepository) DeleteBoard(ctx context.Context, id string) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.boards[id]; !ok { - return fmt.Errorf("board not found: %s", id) - } - delete(r.boards, id) - return nil -} - -// ListBoards returns all boards -func (r *MemoryRepository) ListBoards(ctx context.Context, workspaceID string) ([]*models.Board, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - result := make([]*models.Board, 0, len(r.boards)) - for _, board := range r.boards { - if workspaceID != "" && board.WorkspaceID != workspaceID { - continue - } - result = append(result, board) - } - return result, nil -} - -// Column operations - -// CreateColumn creates a new column -func (r *MemoryRepository) CreateColumn(ctx context.Context, column *models.Column) error { - r.mu.Lock() - defer r.mu.Unlock() - - if column.ID == "" { - column.ID = uuid.New().String() - } - now := time.Now().UTC() - column.CreatedAt = now - column.UpdatedAt = now - - r.columns[column.ID] = column - return nil -} - -// GetColumn retrieves a column by ID -func (r *MemoryRepository) GetColumn(ctx context.Context, id string) (*models.Column, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - column, ok := r.columns[id] - if !ok { - return nil, fmt.Errorf("column not found: %s", id) - } - return column, nil -} - -// GetColumnByState retrieves a column by board ID and state -func (r *MemoryRepository) GetColumnByState(ctx context.Context, boardID string, state v1.TaskState) (*models.Column, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - for _, column := range r.columns { - if column.BoardID == boardID && column.State == state { - return column, nil - } - } - return nil, fmt.Errorf("column not found for board %s with state %s", boardID, state) -} - -// UpdateColumn updates an existing column -func (r *MemoryRepository) UpdateColumn(ctx context.Context, column *models.Column) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.columns[column.ID]; !ok { - return fmt.Errorf("column not found: %s", column.ID) - } - column.UpdatedAt = time.Now().UTC() - r.columns[column.ID] = column - return nil -} - -// DeleteColumn deletes a column by ID -func (r *MemoryRepository) DeleteColumn(ctx context.Context, id string) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.columns[id]; !ok { - return fmt.Errorf("column not found: %s", id) - } - delete(r.columns, id) - return nil -} - -// ListColumns returns all columns for a board -func (r *MemoryRepository) ListColumns(ctx context.Context, boardID string) ([]*models.Column, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - var result []*models.Column - for _, column := range r.columns { - if column.BoardID == boardID { - result = append(result, column) - } - } - return result, nil -} - -// Repository operations - -func (r *MemoryRepository) CreateRepository(ctx context.Context, repository *models.Repository) error { - r.mu.Lock() - defer r.mu.Unlock() - - if repository.ID == "" { - repository.ID = uuid.New().String() - } - now := time.Now().UTC() - repository.CreatedAt = now - repository.UpdatedAt = now - r.repositories[repository.ID] = repository - return nil -} - -func (r *MemoryRepository) GetRepository(ctx context.Context, id string) (*models.Repository, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - repository, ok := r.repositories[id] - if !ok { - return nil, fmt.Errorf("repository not found: %s", id) - } - return repository, nil -} - -func (r *MemoryRepository) UpdateRepository(ctx context.Context, repository *models.Repository) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.repositories[repository.ID]; !ok { - return fmt.Errorf("repository not found: %s", repository.ID) - } - repository.UpdatedAt = time.Now().UTC() - r.repositories[repository.ID] = repository - return nil -} - -func (r *MemoryRepository) DeleteRepository(ctx context.Context, id string) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.repositories[id]; !ok { - return fmt.Errorf("repository not found: %s", id) - } - delete(r.repositories, id) - for scriptID, script := range r.repoScripts { - if script.RepositoryID == id { - delete(r.repoScripts, scriptID) - } - } - return nil -} - -func (r *MemoryRepository) ListRepositories(ctx context.Context, workspaceID string) ([]*models.Repository, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - result := make([]*models.Repository, 0) - for _, repository := range r.repositories { - if repository.WorkspaceID == workspaceID { - result = append(result, repository) - } - } - return result, nil -} - -// Repository script operations - -func (r *MemoryRepository) CreateRepositoryScript(ctx context.Context, script *models.RepositoryScript) error { - r.mu.Lock() - defer r.mu.Unlock() - - if script.ID == "" { - script.ID = uuid.New().String() - } - now := time.Now().UTC() - script.CreatedAt = now - script.UpdatedAt = now - r.repoScripts[script.ID] = script - return nil -} - -func (r *MemoryRepository) GetRepositoryScript(ctx context.Context, id string) (*models.RepositoryScript, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - script, ok := r.repoScripts[id] - if !ok { - return nil, fmt.Errorf("repository script not found: %s", id) - } - return script, nil -} - -func (r *MemoryRepository) UpdateRepositoryScript(ctx context.Context, script *models.RepositoryScript) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.repoScripts[script.ID]; !ok { - return fmt.Errorf("repository script not found: %s", script.ID) - } - script.UpdatedAt = time.Now().UTC() - r.repoScripts[script.ID] = script - return nil -} - -func (r *MemoryRepository) DeleteRepositoryScript(ctx context.Context, id string) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.repoScripts[id]; !ok { - return fmt.Errorf("repository script not found: %s", id) - } - delete(r.repoScripts, id) - return nil -} - -func (r *MemoryRepository) ListRepositoryScripts(ctx context.Context, repositoryID string) ([]*models.RepositoryScript, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - result := make([]*models.RepositoryScript, 0) - for _, script := range r.repoScripts { - if script.RepositoryID == repositoryID { - result = append(result, script) - } - } - return result, nil -} - -func (r *MemoryRepository) addTaskPlacement(taskID, boardID, columnID string, position int) { - if _, ok := r.taskBoards[taskID]; !ok { - r.taskBoards[taskID] = make(map[string]struct{}) - } - r.taskBoards[taskID][boardID] = struct{}{} - - if _, ok := r.taskPlacements[taskID]; !ok { - r.taskPlacements[taskID] = make(map[string]*taskPlacement) - } - r.taskPlacements[taskID][boardID] = &taskPlacement{ - boardID: boardID, - columnID: columnID, - position: position, - } -} - -func (r *MemoryRepository) getTaskPlacement(taskID, boardID string) (*taskPlacement, bool) { - placements, ok := r.taskPlacements[taskID] - if !ok { - return nil, false - } - placement, ok := placements[boardID] - return placement, ok -} - -// Comment operations - -// CreateComment creates a new comment -func (r *MemoryRepository) CreateComment(ctx context.Context, comment *models.Comment) error { - r.mu.Lock() - defer r.mu.Unlock() - - if comment.ID == "" { - comment.ID = uuid.New().String() - } - if comment.CreatedAt.IsZero() { - comment.CreatedAt = time.Now().UTC() - } - if comment.AuthorType == "" { - comment.AuthorType = models.CommentAuthorUser - } - if comment.Type == "" { - comment.Type = models.CommentTypeMessage - } - - r.comments[comment.ID] = comment - return nil -} - -// GetComment retrieves a comment by ID -func (r *MemoryRepository) GetComment(ctx context.Context, id string) (*models.Comment, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - comment, ok := r.comments[id] - if !ok { - return nil, fmt.Errorf("comment not found: %s", id) - } - return comment, nil -} - -// ListComments returns all comments for a task ordered by creation time -func (r *MemoryRepository) ListComments(ctx context.Context, taskID string) ([]*models.Comment, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - var result []*models.Comment - for _, comment := range r.comments { - if comment.TaskID == taskID { - result = append(result, comment) - } - } - // Sort by CreatedAt - for i := 0; i < len(result); i++ { - for j := i + 1; j < len(result); j++ { - if result[i].CreatedAt.After(result[j].CreatedAt) { - result[i], result[j] = result[j], result[i] - } - } - } - return result, nil -} - -// DeleteComment deletes a comment by ID -func (r *MemoryRepository) DeleteComment(ctx context.Context, id string) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.comments[id]; !ok { - return fmt.Errorf("comment not found: %s", id) - } - delete(r.comments, id) - return nil -} - -// GetCommentByToolCallID retrieves a comment by task ID and tool_call_id in metadata -func (r *MemoryRepository) GetCommentByToolCallID(ctx context.Context, taskID, toolCallID string) (*models.Comment, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - for _, comment := range r.comments { - if comment.TaskID == taskID && comment.Metadata != nil { - if tcID, ok := comment.Metadata["tool_call_id"].(string); ok && tcID == toolCallID { - return comment, nil - } - } - } - return nil, fmt.Errorf("comment not found for tool_call_id: %s", toolCallID) -} - -// UpdateComment updates an existing comment -func (r *MemoryRepository) UpdateComment(ctx context.Context, comment *models.Comment) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.comments[comment.ID]; !ok { - return fmt.Errorf("comment not found: %s", comment.ID) - } - r.comments[comment.ID] = comment - return nil -} - -// AgentSession operations - -// CreateAgentSession creates a new agent session -func (r *MemoryRepository) CreateAgentSession(ctx context.Context, session *models.AgentSession) error { - r.mu.Lock() - defer r.mu.Unlock() - - if session.ID == "" { - session.ID = uuid.New().String() - } - now := time.Now().UTC() - if session.StartedAt.IsZero() { - session.StartedAt = now - } - session.UpdatedAt = now - - r.agentSessions[session.ID] = session - return nil -} - -// GetAgentSession retrieves an agent session by ID -func (r *MemoryRepository) GetAgentSession(ctx context.Context, id string) (*models.AgentSession, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - session, ok := r.agentSessions[id] - if !ok { - return nil, fmt.Errorf("agent session not found: %s", id) - } - return session, nil -} - -// GetAgentSessionByTaskID retrieves the most recent agent session for a task -func (r *MemoryRepository) GetAgentSessionByTaskID(ctx context.Context, taskID string) (*models.AgentSession, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - var latest *models.AgentSession - for _, session := range r.agentSessions { - if session.TaskID == taskID { - if latest == nil || session.StartedAt.After(latest.StartedAt) { - latest = session - } - } - } - if latest == nil { - return nil, fmt.Errorf("no agent session found for task: %s", taskID) - } - return latest, nil -} - -// GetActiveAgentSessionByTaskID retrieves the active agent session for a task -func (r *MemoryRepository) GetActiveAgentSessionByTaskID(ctx context.Context, taskID string) (*models.AgentSession, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - for _, session := range r.agentSessions { - if session.TaskID == taskID && (session.Status == models.AgentSessionStatusRunning || session.Status == models.AgentSessionStatusWaiting || session.Status == models.AgentSessionStatusPending) { - return session, nil - } - } - return nil, fmt.Errorf("no active agent session found for task: %s", taskID) -} - -// UpdateAgentSession updates an existing agent session -func (r *MemoryRepository) UpdateAgentSession(ctx context.Context, session *models.AgentSession) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.agentSessions[session.ID]; !ok { - return fmt.Errorf("agent session not found: %s", session.ID) - } - session.UpdatedAt = time.Now().UTC() - r.agentSessions[session.ID] = session - return nil -} - -// UpdateAgentSessionStatus updates the status of an agent session -func (r *MemoryRepository) UpdateAgentSessionStatus(ctx context.Context, id string, status models.AgentSessionStatus, errorMessage string) error { - r.mu.Lock() - defer r.mu.Unlock() - - session, ok := r.agentSessions[id] - if !ok { - return fmt.Errorf("agent session not found: %s", id) - } - session.Status = status - session.ErrorMessage = errorMessage - session.UpdatedAt = time.Now().UTC() - if status == models.AgentSessionStatusCompleted || status == models.AgentSessionStatusFailed || status == models.AgentSessionStatusStopped { - now := time.Now().UTC() - session.CompletedAt = &now - } - return nil -} - -// ListAgentSessions returns all agent sessions for a task -func (r *MemoryRepository) ListAgentSessions(ctx context.Context, taskID string) ([]*models.AgentSession, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - var result []*models.AgentSession - for _, session := range r.agentSessions { - if session.TaskID == taskID { - result = append(result, session) - } - } - return result, nil -} - -// ListActiveAgentSessions returns all active agent sessions -func (r *MemoryRepository) ListActiveAgentSessions(ctx context.Context) ([]*models.AgentSession, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - var result []*models.AgentSession - for _, session := range r.agentSessions { - if session.Status == models.AgentSessionStatusRunning || session.Status == models.AgentSessionStatusWaiting || session.Status == models.AgentSessionStatusPending { - result = append(result, session) - } - } - return result, nil -} - -// DeleteAgentSession deletes an agent session by ID -func (r *MemoryRepository) DeleteAgentSession(ctx context.Context, id string) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.agentSessions[id]; !ok { - return fmt.Errorf("agent session not found: %s", id) - } - delete(r.agentSessions, id) - return nil -} - -// Executor operations - -func (r *MemoryRepository) CreateExecutor(ctx context.Context, executor *models.Executor) error { - r.mu.Lock() - defer r.mu.Unlock() - - if executor.ID == "" { - executor.ID = uuid.New().String() - } - now := time.Now().UTC() - executor.CreatedAt = now - executor.UpdatedAt = now - - r.executors[executor.ID] = executor - return nil -} - -func (r *MemoryRepository) GetExecutor(ctx context.Context, id string) (*models.Executor, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - executor, ok := r.executors[id] - if !ok { - return nil, fmt.Errorf("executor not found: %s", id) - } - return executor, nil -} - -func (r *MemoryRepository) UpdateExecutor(ctx context.Context, executor *models.Executor) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.executors[executor.ID]; !ok { - return fmt.Errorf("executor not found: %s", executor.ID) - } - executor.UpdatedAt = time.Now().UTC() - r.executors[executor.ID] = executor - return nil -} - -func (r *MemoryRepository) DeleteExecutor(ctx context.Context, id string) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.executors[id]; !ok { - return fmt.Errorf("executor not found: %s", id) - } - delete(r.executors, id) - return nil -} - -func (r *MemoryRepository) ListExecutors(ctx context.Context) ([]*models.Executor, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - result := make([]*models.Executor, 0, len(r.executors)) - for _, executor := range r.executors { - result = append(result, executor) - } - return result, nil -} - -// Environment operations - -func (r *MemoryRepository) CreateEnvironment(ctx context.Context, environment *models.Environment) error { - r.mu.Lock() - defer r.mu.Unlock() - - if environment.ID == "" { - environment.ID = uuid.New().String() - } - now := time.Now().UTC() - environment.CreatedAt = now - environment.UpdatedAt = now - r.environments[environment.ID] = environment - return nil -} - -func (r *MemoryRepository) GetEnvironment(ctx context.Context, id string) (*models.Environment, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - environment, ok := r.environments[id] - if !ok { - return nil, fmt.Errorf("environment not found: %s", id) - } - return environment, nil -} - -func (r *MemoryRepository) UpdateEnvironment(ctx context.Context, environment *models.Environment) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.environments[environment.ID]; !ok { - return fmt.Errorf("environment not found: %s", environment.ID) - } - environment.UpdatedAt = time.Now().UTC() - r.environments[environment.ID] = environment - return nil -} - -func (r *MemoryRepository) DeleteEnvironment(ctx context.Context, id string) error { - r.mu.Lock() - defer r.mu.Unlock() - - if _, ok := r.environments[id]; !ok { - return fmt.Errorf("environment not found: %s", id) - } - delete(r.environments, id) - return nil -} - -func (r *MemoryRepository) ListEnvironments(ctx context.Context) ([]*models.Environment, error) { - r.mu.RLock() - defer r.mu.RUnlock() - - result := make([]*models.Environment, 0, len(r.environments)) - for _, environment := range r.environments { - result = append(result, environment) - } - return result, nil -} diff --git a/apps/backend/internal/task/repository/memory_test.go b/apps/backend/internal/task/repository/memory_test.go deleted file mode 100644 index cbd9dd22f6..0000000000 --- a/apps/backend/internal/task/repository/memory_test.go +++ /dev/null @@ -1,465 +0,0 @@ -package repository - -import ( - "context" - "testing" - - "github.com/kandev/kandev/internal/task/models" - v1 "github.com/kandev/kandev/pkg/api/v1" -) - -func TestNewMemoryRepository(t *testing.T) { - repo := NewMemoryRepository() - if repo == nil { - t.Fatal("expected non-nil repository") - } - if repo.tasks == nil { - t.Error("expected tasks map to be initialized") - } - if repo.workspaces == nil { - t.Error("expected workspaces map to be initialized") - } - if repo.boards == nil { - t.Error("expected boards map to be initialized") - } - if repo.columns == nil { - t.Error("expected columns map to be initialized") - } -} - -func TestMemoryRepository_Close(t *testing.T) { - repo := NewMemoryRepository() - err := repo.Close() - if err != nil { - t.Errorf("expected no error on close, got %v", err) - } -} - -// Board CRUD tests - -func TestMemoryRepository_BoardCRUD(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - workspace := &models.Workspace{ID: "ws-1", Name: "Workspace"} - if err := repo.CreateWorkspace(ctx, workspace); err != nil { - t.Fatalf("failed to create workspace: %v", err) - } - - // Create - board := &models.Board{WorkspaceID: workspace.ID, Name: "Test Board", Description: "A test board"} - if err := repo.CreateBoard(ctx, board); err != nil { - t.Fatalf("failed to create board: %v", err) - } - if board.ID == "" { - t.Error("expected board ID to be set") - } - if board.CreatedAt.IsZero() { - t.Error("expected CreatedAt to be set") - } - - // Get - retrieved, err := repo.GetBoard(ctx, board.ID) - if err != nil { - t.Fatalf("failed to get board: %v", err) - } - if retrieved.Name != "Test Board" { - t.Errorf("expected name 'Test Board', got %s", retrieved.Name) - } - - // Update - board.Name = "Updated Name" - if err := repo.UpdateBoard(ctx, board); err != nil { - t.Fatalf("failed to update board: %v", err) - } - retrieved, _ = repo.GetBoard(ctx, board.ID) - if retrieved.Name != "Updated Name" { - t.Errorf("expected name 'Updated Name', got %s", retrieved.Name) - } - - // Delete - if err := repo.DeleteBoard(ctx, board.ID); err != nil { - t.Fatalf("failed to delete board: %v", err) - } - _, err = repo.GetBoard(ctx, board.ID) - if err == nil { - t.Error("expected board to be deleted") - } -} - -func TestMemoryRepository_BoardNotFound(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - _, err := repo.GetBoard(ctx, "nonexistent") - if err == nil { - t.Error("expected error for nonexistent board") - } - - err = repo.UpdateBoard(ctx, &models.Board{ID: "nonexistent", Name: "Test"}) - if err == nil { - t.Error("expected error for updating nonexistent board") - } - - err = repo.DeleteBoard(ctx, "nonexistent") - if err == nil { - t.Error("expected error for deleting nonexistent board") - } -} - -func TestMemoryRepository_ListBoards(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - _ = repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace 1"}) - _ = repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-2", Name: "Workspace 2"}) - - _ = repo.CreateBoard(ctx, &models.Board{ID: "board-1", WorkspaceID: "ws-1", Name: "Board 1"}) - _ = repo.CreateBoard(ctx, &models.Board{ID: "board-2", WorkspaceID: "ws-1", Name: "Board 2"}) - _ = repo.CreateBoard(ctx, &models.Board{ID: "board-3", WorkspaceID: "ws-2", Name: "Board 3"}) - - boards, err := repo.ListBoards(ctx, "ws-1") - if err != nil { - t.Fatalf("failed to list boards: %v", err) - } - if len(boards) != 2 { - t.Errorf("expected 2 boards, got %d", len(boards)) - } -} - -// Column CRUD tests - -func TestMemoryRepository_ColumnCRUD(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - // Create - column := &models.Column{BoardID: "board-123", Name: "To Do", Position: 0, State: v1.TaskStateTODO} - if err := repo.CreateColumn(ctx, column); err != nil { - t.Fatalf("failed to create column: %v", err) - } - if column.ID == "" { - t.Error("expected column ID to be set") - } - - // Get - retrieved, err := repo.GetColumn(ctx, column.ID) - if err != nil { - t.Fatalf("failed to get column: %v", err) - } - if retrieved.Name != "To Do" { - t.Errorf("expected name 'To Do', got %s", retrieved.Name) - } - - // Update - column.Name = "Done" - column.State = v1.TaskStateCompleted - if err := repo.UpdateColumn(ctx, column); err != nil { - t.Fatalf("failed to update column: %v", err) - } - retrieved, _ = repo.GetColumn(ctx, column.ID) - if retrieved.Name != "Done" { - t.Errorf("expected name 'Done', got %s", retrieved.Name) - } - - // Delete - if err := repo.DeleteColumn(ctx, column.ID); err != nil { - t.Fatalf("failed to delete column: %v", err) - } - _, err = repo.GetColumn(ctx, column.ID) - if err == nil { - t.Error("expected column to be deleted") - } -} - -func TestMemoryRepository_ColumnNotFound(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - _, err := repo.GetColumn(ctx, "nonexistent") - if err == nil { - t.Error("expected error for nonexistent column") - } - - err = repo.UpdateColumn(ctx, &models.Column{ID: "nonexistent", Name: "Test"}) - if err == nil { - t.Error("expected error for updating nonexistent column") - } - - err = repo.DeleteColumn(ctx, "nonexistent") - if err == nil { - t.Error("expected error for deleting nonexistent column") - } -} - -func TestMemoryRepository_ListColumns(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - _ = repo.CreateColumn(ctx, &models.Column{ID: "col-1", BoardID: "board-123", Name: "To Do"}) - _ = repo.CreateColumn(ctx, &models.Column{ID: "col-2", BoardID: "board-123", Name: "In Progress"}) - _ = repo.CreateColumn(ctx, &models.Column{ID: "col-3", BoardID: "board-456", Name: "Done"}) - - columns, err := repo.ListColumns(ctx, "board-123") - if err != nil { - t.Fatalf("failed to list columns: %v", err) - } - if len(columns) != 2 { - t.Errorf("expected 2 columns for board-123, got %d", len(columns)) - } -} - -// Task CRUD tests - -func TestMemoryRepository_TaskCRUD(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - // Create - task := &models.Task{ - WorkspaceID: "ws-1", - BoardID: "board-123", - ColumnID: "col-123", - Title: "Test Task", - Description: "A test task", - State: v1.TaskStateTODO, - Priority: 5, - } - if err := repo.CreateTask(ctx, task); err != nil { - t.Fatalf("failed to create task: %v", err) - } - if task.ID == "" { - t.Error("expected task ID to be set") - } - if task.CreatedAt.IsZero() { - t.Error("expected CreatedAt to be set") - } - - // Get - retrieved, err := repo.GetTask(ctx, task.ID) - if err != nil { - t.Fatalf("failed to get task: %v", err) - } - if retrieved.Title != "Test Task" { - t.Errorf("expected title 'Test Task', got %s", retrieved.Title) - } - - // Update - task.Title = "Updated Task" - task.Description = "Updated description" - if err := repo.UpdateTask(ctx, task); err != nil { - t.Fatalf("failed to update task: %v", err) - } - retrieved, _ = repo.GetTask(ctx, task.ID) - if retrieved.Title != "Updated Task" { - t.Errorf("expected title 'Updated Task', got %s", retrieved.Title) - } - - // Delete - if err := repo.DeleteTask(ctx, task.ID); err != nil { - t.Fatalf("failed to delete task: %v", err) - } - _, err = repo.GetTask(ctx, task.ID) - if err == nil { - t.Error("expected task to be deleted") - } -} - -func TestMemoryRepository_TaskNotFound(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - _, err := repo.GetTask(ctx, "nonexistent") - if err == nil { - t.Error("expected error for nonexistent task") - } - - err = repo.UpdateTask(ctx, &models.Task{ID: "nonexistent", Title: "Test"}) - if err == nil { - t.Error("expected error for updating nonexistent task") - } - - err = repo.DeleteTask(ctx, "nonexistent") - if err == nil { - t.Error("expected error for deleting nonexistent task") - } -} - -func TestMemoryRepository_UpdateTaskState(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - task := &models.Task{ID: "task-123", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-123", Title: "Test", State: v1.TaskStateTODO} - _ = repo.CreateTask(ctx, task) - - err := repo.UpdateTaskState(ctx, "task-123", v1.TaskStateInProgress) - if err != nil { - t.Fatalf("failed to update task state: %v", err) - } - - retrieved, _ := repo.GetTask(ctx, "task-123") - if retrieved.State != v1.TaskStateInProgress { - t.Errorf("expected state IN_PROGRESS, got %s", retrieved.State) - } -} - -func TestMemoryRepository_UpdateTaskStateNotFound(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - err := repo.UpdateTaskState(ctx, "nonexistent", v1.TaskStateInProgress) - if err == nil { - t.Error("expected error for nonexistent task") - } -} - -func TestMemoryRepository_ListTasks(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - _ = repo.CreateTask(ctx, &models.Task{ID: "task-1", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-1", Title: "Task 1"}) - _ = repo.CreateTask(ctx, &models.Task{ID: "task-2", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-1", Title: "Task 2"}) - _ = repo.CreateTask(ctx, &models.Task{ID: "task-3", WorkspaceID: "ws-1", BoardID: "board-456", ColumnID: "col-2", Title: "Task 3"}) - - tasks, err := repo.ListTasks(ctx, "board-123") - if err != nil { - t.Fatalf("failed to list tasks: %v", err) - } - if len(tasks) != 2 { - t.Errorf("expected 2 tasks for board-123, got %d", len(tasks)) - } -} - -func TestMemoryRepository_ListTasksByColumn(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - _ = repo.CreateTask(ctx, &models.Task{ID: "task-1", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-1", Title: "Task 1"}) - _ = repo.CreateTask(ctx, &models.Task{ID: "task-2", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-1", Title: "Task 2"}) - _ = repo.CreateTask(ctx, &models.Task{ID: "task-3", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-2", Title: "Task 3"}) - - tasks, err := repo.ListTasksByColumn(ctx, "col-1") - if err != nil { - t.Fatalf("failed to list tasks by column: %v", err) - } - if len(tasks) != 2 { - t.Errorf("expected 2 tasks for col-1, got %d", len(tasks)) - } -} - -// Comment CRUD tests - -func TestMemoryRepository_CommentCRUD(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - // Create a task first - task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"} - _ = repo.CreateTask(ctx, task) - - // Create comment - comment := &models.Comment{ - TaskID: "task-123", - AuthorType: models.CommentAuthorUser, - AuthorID: "user-123", - Content: "This is a test comment", - RequestsInput: false, - } - if err := repo.CreateComment(ctx, comment); err != nil { - t.Fatalf("failed to create comment: %v", err) - } - if comment.ID == "" { - t.Error("expected comment ID to be set") - } - if comment.CreatedAt.IsZero() { - t.Error("expected CreatedAt to be set") - } - - // Get comment - retrieved, err := repo.GetComment(ctx, comment.ID) - if err != nil { - t.Fatalf("failed to get comment: %v", err) - } - if retrieved.Content != "This is a test comment" { - t.Errorf("expected content 'This is a test comment', got %s", retrieved.Content) - } - if retrieved.AuthorType != models.CommentAuthorUser { - t.Errorf("expected author type 'user', got %s", retrieved.AuthorType) - } - - // Delete comment - if err := repo.DeleteComment(ctx, comment.ID); err != nil { - t.Fatalf("failed to delete comment: %v", err) - } - _, err = repo.GetComment(ctx, comment.ID) - if err == nil { - t.Error("expected comment to be deleted") - } -} - -func TestMemoryRepository_CommentNotFound(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - _, err := repo.GetComment(ctx, "nonexistent") - if err == nil { - t.Error("expected error for nonexistent comment") - } - - err = repo.DeleteComment(ctx, "nonexistent") - if err == nil { - t.Error("expected error for deleting nonexistent comment") - } -} - -func TestMemoryRepository_ListComments(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - // Create a task - task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"} - _ = repo.CreateTask(ctx, task) - - // Create multiple comments - _ = repo.CreateComment(ctx, &models.Comment{ID: "comment-1", TaskID: "task-123", AuthorType: models.CommentAuthorUser, Content: "Comment 1"}) - _ = repo.CreateComment(ctx, &models.Comment{ID: "comment-2", TaskID: "task-123", AuthorType: models.CommentAuthorAgent, Content: "Comment 2"}) - _ = repo.CreateComment(ctx, &models.Comment{ID: "comment-3", TaskID: "task-456", AuthorType: models.CommentAuthorUser, Content: "Comment 3"}) - - comments, err := repo.ListComments(ctx, "task-123") - if err != nil { - t.Fatalf("failed to list comments: %v", err) - } - if len(comments) != 2 { - t.Errorf("expected 2 comments for task-123, got %d", len(comments)) - } -} - -func TestMemoryRepository_CommentWithRequestsInput(t *testing.T) { - repo := NewMemoryRepository() - ctx := context.Background() - - // Create a task - task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"} - _ = repo.CreateTask(ctx, task) - - // Create agent comment requesting input - comment := &models.Comment{ - TaskID: "task-123", - AuthorType: models.CommentAuthorAgent, - AuthorID: "agent-123", - Content: "What should I do next?", - RequestsInput: true, - ACPSessionID: "session-abc", - } - if err := repo.CreateComment(ctx, comment); err != nil { - t.Fatalf("failed to create comment: %v", err) - } - - retrieved, _ := repo.GetComment(ctx, comment.ID) - if !retrieved.RequestsInput { - t.Error("expected RequestsInput to be true") - } - if retrieved.ACPSessionID != "session-abc" { - t.Errorf("expected ACPSessionID 'session-abc', got %s", retrieved.ACPSessionID) - } -} diff --git a/apps/backend/internal/task/repository/sqlite.go b/apps/backend/internal/task/repository/sqlite.go index 789781c667..ea8211fd79 100644 --- a/apps/backend/internal/task/repository/sqlite.go +++ b/apps/backend/internal/task/repository/sqlite.go @@ -916,26 +916,7 @@ func (r *SQLiteRepository) ListTasksByColumn(ctx context.Context, columnID strin return r.scanTasksWithPlacement(rows) } -// scanTasks scans multiple task rows -func (r *SQLiteRepository) scanTasks(rows *sql.Rows) ([]*models.Task, error) { - var result []*models.Task - for rows.Next() { - task := &models.Task{} - var metadata string - var agentProfileID, repositoryURL, branch, assignedTo sql.NullString - err := rows.Scan(&task.ID, &task.WorkspaceID, &task.BoardID, &task.ColumnID, &task.Title, &task.Description, &task.State, &task.Priority, &agentProfileID, &repositoryURL, &branch, &assignedTo, &task.Position, &metadata, &task.CreatedAt, &task.UpdatedAt) - if err != nil { - return nil, err - } - task.AgentProfileID = agentProfileID.String - task.RepositoryURL = repositoryURL.String - task.Branch = branch.String - task.AssignedTo = assignedTo.String - _ = json.Unmarshal([]byte(metadata), &task.Metadata) - result = append(result, task) - } - return result, rows.Err() -} + func (r *SQLiteRepository) scanTasksWithPlacement(rows *sql.Rows) ([]*models.Task, error) { var result []*models.Task diff --git a/apps/backend/internal/task/service/repository_discovery_test.go b/apps/backend/internal/task/service/repository_discovery_test.go index b30838bbb9..9e88e993db 100644 --- a/apps/backend/internal/task/service/repository_discovery_test.go +++ b/apps/backend/internal/task/service/repository_discovery_test.go @@ -272,7 +272,12 @@ func writeRef(t *testing.T, path string) { func newDiscoveryService(t *testing.T, root string) *Service { t.Helper() - repo := repository.NewMemoryRepository() + tmpDir := t.TempDir() + repo, err := repository.NewSQLiteRepository(tmpDir + "/test.db") + if err != nil { + t.Fatalf("failed to create test repository: %v", err) + } + t.Cleanup(func() { repo.Close() }) log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json", OutputPath: "stdout"}) eventBus := bus.NewMemoryEventBus(log) return NewService(repo, eventBus, log, RepositoryDiscoveryConfig{ diff --git a/apps/backend/internal/task/service/service_test.go b/apps/backend/internal/task/service/service_test.go index dc60e71936..c8bb9933aa 100644 --- a/apps/backend/internal/task/service/service_test.go +++ b/apps/backend/internal/task/service/service_test.go @@ -67,9 +67,14 @@ func (m *MockEventBus) ClearEvents() { m.publishedEvents = make([]*bus.Event, 0) } -func createTestService(t *testing.T) (*Service, *MockEventBus, *repository.MemoryRepository) { +func createTestService(t *testing.T) (*Service, *MockEventBus, repository.Repository) { t.Helper() - repo := repository.NewMemoryRepository() + tmpDir := t.TempDir() + repo, err := repository.NewSQLiteRepository(tmpDir + "/test.db") + if err != nil { + t.Fatalf("failed to create test repository: %v", err) + } + t.Cleanup(func() { repo.Close() }) eventBus := NewMockEventBus() log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json", OutputPath: "stdout"}) svc := NewService(repo, eventBus, log, RepositoryDiscoveryConfig{}) @@ -128,9 +133,8 @@ func TestService_GetTask(t *testing.T) { svc, _, repo := createTestService(t) ctx := context.Background() - // Create a task directly in repo - task := &models.Task{ID: "task-123", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"} - _ = repo.CreateTask(ctx, task) + // Create required entities + setupTestTask(t, repo) retrieved, err := svc.GetTask(ctx, "task-123") if err != nil { @@ -155,8 +159,10 @@ func TestService_UpdateTask(t *testing.T) { svc, eventBus, repo := createTestService(t) ctx := context.Background() - task := &models.Task{ID: "task-123", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-123", Title: "Original"} - _ = repo.CreateTask(ctx, task) + _ = repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}) + _ = repo.CreateBoard(ctx, &models.Board{ID: "board-123", WorkspaceID: "ws-1", Name: "Board"}) + _ = repo.CreateColumn(ctx, &models.Column{ID: "col-123", BoardID: "board-123", Name: "Column", State: v1.TaskStateTODO}) + _ = repo.CreateTask(ctx, &models.Task{ID: "task-123", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-123", Title: "Original"}) eventBus.ClearEvents() newTitle := "Updated Title" @@ -184,8 +190,10 @@ func TestService_DeleteTask(t *testing.T) { svc, eventBus, repo := createTestService(t) ctx := context.Background() - task := &models.Task{ID: "task-123", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-123", Title: "Test"} - _ = repo.CreateTask(ctx, task) + _ = repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}) + _ = repo.CreateBoard(ctx, &models.Board{ID: "board-123", WorkspaceID: "ws-1", Name: "Board"}) + _ = repo.CreateColumn(ctx, &models.Column{ID: "col-123", BoardID: "board-123", Name: "Column", State: v1.TaskStateTODO}) + _ = repo.CreateTask(ctx, &models.Task{ID: "task-123", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-123", Title: "Test"}) eventBus.ClearEvents() err := svc.DeleteTask(ctx, "task-123") @@ -210,6 +218,9 @@ func TestService_ListTasks(t *testing.T) { svc, _, repo := createTestService(t) ctx := context.Background() + _ = repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}) + _ = repo.CreateBoard(ctx, &models.Board{ID: "board-123", WorkspaceID: "ws-1", Name: "Board"}) + _ = repo.CreateColumn(ctx, &models.Column{ID: "col-123", BoardID: "board-123", Name: "Column", State: v1.TaskStateTODO}) _ = repo.CreateTask(ctx, &models.Task{ID: "task-1", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-123", Title: "Task 1"}) _ = repo.CreateTask(ctx, &models.Task{ID: "task-2", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-123", Title: "Task 2"}) @@ -227,6 +238,8 @@ func TestService_UpdateTaskState(t *testing.T) { ctx := context.Background() _ = repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}) + _ = repo.CreateBoard(ctx, &models.Board{ID: "board-123", WorkspaceID: "ws-1", Name: "Board"}) + _ = repo.CreateColumn(ctx, &models.Column{ID: "col-123", BoardID: "board-123", Name: "Column", State: v1.TaskStateTODO}) task := &models.Task{ID: "task-123", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-123", Title: "Test", State: v1.TaskStateTODO} _ = repo.CreateTask(ctx, task) eventBus.ClearEvents() @@ -256,9 +269,10 @@ func TestService_MoveTask(t *testing.T) { _ = repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}) _ = repo.CreateBoard(ctx, &models.Board{ID: "board-123", WorkspaceID: "ws-1", Name: "Board"}) - // Create column with different state - column := &models.Column{ID: "col-done", BoardID: "board-123", Name: "Done", State: v1.TaskStateCompleted} - _ = repo.CreateColumn(ctx, column) + // Create source column + _ = repo.CreateColumn(ctx, &models.Column{ID: "col-123", BoardID: "board-123", Name: "Todo", State: v1.TaskStateTODO}) + // Create destination column with different state + _ = repo.CreateColumn(ctx, &models.Column{ID: "col-done", BoardID: "board-123", Name: "Done", State: v1.TaskStateCompleted}) task := &models.Task{ID: "task-123", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-123", Title: "Test", State: v1.TaskStateTODO} _ = repo.CreateTask(ctx, task) @@ -440,13 +454,21 @@ func TestService_ListColumns(t *testing.T) { // Comment tests +func setupTestTask(t *testing.T, repo repository.Repository) { + t.Helper() + ctx := context.Background() + _ = repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}) + _ = repo.CreateBoard(ctx, &models.Board{ID: "board-123", WorkspaceID: "ws-1", Name: "Board"}) + _ = repo.CreateColumn(ctx, &models.Column{ID: "col-123", BoardID: "board-123", Name: "Column", State: v1.TaskStateTODO}) + _ = repo.CreateTask(ctx, &models.Task{ID: "task-123", WorkspaceID: "ws-1", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"}) +} + func TestService_CreateComment(t *testing.T) { svc, eventBus, repo := createTestService(t) ctx := context.Background() // Create a task first - task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"} - _ = repo.CreateTask(ctx, task) + setupTestTask(t, repo) eventBus.ClearEvents() req := &CreateCommentRequest{ @@ -485,8 +507,7 @@ func TestService_CreateAgentComment(t *testing.T) { svc, _, repo := createTestService(t) ctx := context.Background() - task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"} - _ = repo.CreateTask(ctx, task) + setupTestTask(t, repo) req := &CreateCommentRequest{ TaskID: "task-123", @@ -532,8 +553,7 @@ func TestService_GetComment(t *testing.T) { svc, _, repo := createTestService(t) ctx := context.Background() - task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"} - _ = repo.CreateTask(ctx, task) + setupTestTask(t, repo) comment := &models.Comment{ID: "comment-123", TaskID: "task-123", AuthorType: models.CommentAuthorUser, Content: "Test"} _ = repo.CreateComment(ctx, comment) @@ -551,8 +571,7 @@ func TestService_ListComments(t *testing.T) { svc, _, repo := createTestService(t) ctx := context.Background() - task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"} - _ = repo.CreateTask(ctx, task) + setupTestTask(t, repo) _ = repo.CreateComment(ctx, &models.Comment{ID: "comment-1", TaskID: "task-123", AuthorType: models.CommentAuthorUser, Content: "Comment 1"}) _ = repo.CreateComment(ctx, &models.Comment{ID: "comment-2", TaskID: "task-123", AuthorType: models.CommentAuthorAgent, Content: "Comment 2"}) @@ -570,8 +589,7 @@ func TestService_DeleteComment(t *testing.T) { svc, _, repo := createTestService(t) ctx := context.Background() - task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"} - _ = repo.CreateTask(ctx, task) + setupTestTask(t, repo) comment := &models.Comment{ID: "comment-123", TaskID: "task-123", AuthorType: models.CommentAuthorUser, Content: "Test"} _ = repo.CreateComment(ctx, comment) diff --git a/apps/backend/pkg/acp/jsonrpc/client.go b/apps/backend/pkg/acp/jsonrpc/client.go deleted file mode 100644 index 6e5d528e3a..0000000000 --- a/apps/backend/pkg/acp/jsonrpc/client.go +++ /dev/null @@ -1,296 +0,0 @@ -package jsonrpc - -import ( - "bufio" - "context" - "encoding/json" - "fmt" - "io" - "sync" - "sync/atomic" - - "github.com/kandev/kandev/internal/common/logger" - "go.uber.org/zap" -) - -// Client handles JSON-RPC 2.0 communication over stdin/stdout streams -type Client struct { - stdin io.Writer - stdout io.Reader - - requestID atomic.Int64 - pending map[interface{}]chan *Response - mu sync.Mutex - - // Notification handler - onNotification func(method string, params json.RawMessage) - - // Request handler for agent-to-client requests (like session/request_permission) - onRequest func(id interface{}, method string, params json.RawMessage) - - logger *logger.Logger - done chan struct{} -} - -// NewClient creates a new JSON-RPC client -func NewClient(stdin io.Writer, stdout io.Reader, log *logger.Logger) *Client { - return &Client{ - stdin: stdin, - stdout: stdout, - pending: make(map[interface{}]chan *Response), - logger: log.WithFields(zap.String("component", "jsonrpc-client")), - done: make(chan struct{}), - } -} - -// SetNotificationHandler sets the handler for incoming notifications -func (c *Client) SetNotificationHandler(handler func(method string, params json.RawMessage)) { - c.onNotification = handler -} - -// SetRequestHandler sets the handler for incoming requests from the agent -// (e.g., session/request_permission). The handler should call SendResponse to reply. -func (c *Client) SetRequestHandler(handler func(id interface{}, method string, params json.RawMessage)) { - c.onRequest = handler -} - -// SendResponse sends a response to an agent request -func (c *Client) SendResponse(id interface{}, result interface{}, err *Error) error { - var resultJSON json.RawMessage - if result != nil && err == nil { - var marshalErr error - resultJSON, marshalErr = json.Marshal(result) - if marshalErr != nil { - return fmt.Errorf("failed to marshal result: %w", marshalErr) - } - } - - resp := &Response{ - JSONRPC: "2.0", - ID: id, - Result: resultJSON, - Error: err, - } - - return c.send(resp) -} - -// Start begins reading responses from stdout -func (c *Client) Start(ctx context.Context) { - go c.readLoop(ctx) -} - -// Stop stops the client -func (c *Client) Stop() { - close(c.done) -} - -// Call sends a request and waits for a response -func (c *Client) Call(ctx context.Context, method string, params interface{}) (*Response, error) { - id := c.requestID.Add(1) - - var paramsJSON json.RawMessage - if params != nil { - var err error - paramsJSON, err = json.Marshal(params) - if err != nil { - return nil, fmt.Errorf("failed to marshal params: %w", err) - } - } - - req := &Request{ - JSONRPC: "2.0", - ID: id, - Method: method, - Params: paramsJSON, - } - - // Create response channel - respCh := make(chan *Response, 1) - c.mu.Lock() - c.pending[id] = respCh - c.mu.Unlock() - - defer func() { - c.mu.Lock() - delete(c.pending, id) - c.mu.Unlock() - }() - - // Send request - if err := c.send(req); err != nil { - return nil, err - } - - // Wait for response - select { - case resp := <-respCh: - return resp, nil - case <-ctx.Done(): - return nil, ctx.Err() - case <-c.done: - return nil, fmt.Errorf("client closed") - } -} - -// Notify sends a notification (no response expected) -func (c *Client) Notify(method string, params interface{}) error { - var paramsJSON json.RawMessage - if params != nil { - var err error - paramsJSON, err = json.Marshal(params) - if err != nil { - return fmt.Errorf("failed to marshal params: %w", err) - } - } - - notif := &Notification{ - JSONRPC: "2.0", - Method: method, - Params: paramsJSON, - } - - return c.send(notif) -} - -func (c *Client) send(msg interface{}) error { - data, err := json.Marshal(msg) - if err != nil { - return fmt.Errorf("failed to marshal message: %w", err) - } - - data = append(data, '\n') - _, err = c.stdin.Write(data) - if err != nil { - return fmt.Errorf("failed to write message: %w", err) - } - - c.logger.Debug("sent message", zap.String("data", string(data))) - return nil -} - -func (c *Client) readLoop(ctx context.Context) { - scanner := bufio.NewScanner(c.stdout) - // Increase buffer size for large messages - buf := make([]byte, 0, 64*1024) - scanner.Buffer(buf, 1024*1024) - - for scanner.Scan() { - select { - case <-ctx.Done(): - return - case <-c.done: - return - default: - } - - line := scanner.Bytes() - if len(line) == 0 { - continue - } - - c.logger.Debug("received message", zap.String("data", string(line))) - - // Parse the message to determine its type - // JSON-RPC 2.0 message types: - // - Response: has "id" + ("result" OR "error"), no "method" - // - Request: has "id" + "method" - // - Notification: has "method", no "id" - var msg struct { - ID interface{} `json:"id"` - Method string `json:"method"` - Result json.RawMessage `json:"result"` - Error *Error `json:"error"` - Params json.RawMessage `json:"params"` - } - if err := json.Unmarshal(line, &msg); err != nil { - c.logger.Warn("failed to parse message", zap.Error(err), zap.String("data", string(line))) - continue - } - - // Determine message type based on fields present - hasID := msg.ID != nil - hasMethod := msg.Method != "" - hasResult := msg.Result != nil - hasError := msg.Error != nil - - if hasID && !hasMethod && (hasResult || hasError) { - // This is a response to our request - resp := &Response{ - JSONRPC: "2.0", - ID: msg.ID, - Result: msg.Result, - Error: msg.Error, - } - c.handleResponse(resp) - } else if hasID && hasMethod { - // This is a request FROM the agent (e.g., session/request_permission) - c.handleRequest(msg.ID, msg.Method, msg.Params) - } else if hasMethod && !hasID { - // This is a notification - notif := &Notification{ - JSONRPC: "2.0", - Method: msg.Method, - Params: msg.Params, - } - c.handleNotification(notif) - } else { - c.logger.Warn("received unknown message format", zap.String("data", string(line))) - } - } - - if err := scanner.Err(); err != nil { - c.logger.Error("read loop error", zap.Error(err)) - } -} - -func (c *Client) handleResponse(resp *Response) { - // Normalize the ID - JSON unmarshals numbers as float64, but we store as int64 - id := normalizeID(resp.ID) - - c.mu.Lock() - ch, ok := c.pending[id] - c.mu.Unlock() - - if ok { - ch <- resp - } else { - c.logger.Warn("received response for unknown request", zap.Any("id", resp.ID)) - } -} - -// normalizeID converts JSON unmarshaled IDs to a consistent type for map lookup. -// JSON numbers are unmarshaled as float64, but we store request IDs as int64. -func normalizeID(id interface{}) interface{} { - switch v := id.(type) { - case float64: - return int64(v) - case json.Number: - if i, err := v.Int64(); err == nil { - return i - } - } - return id -} - -func (c *Client) handleNotification(notif *Notification) { - if c.onNotification != nil { - c.onNotification(notif.Method, notif.Params) - } -} - -func (c *Client) handleRequest(id interface{}, method string, params json.RawMessage) { - if c.onRequest != nil { - c.onRequest(id, method, params) - } else { - // No handler registered, send error response - c.logger.Warn("received request but no handler registered", - zap.Any("id", id), - zap.String("method", method)) - c.SendResponse(id, nil, &Error{ - Code: MethodNotFound, - Message: "Method not found", - }) - } -} - diff --git a/apps/backend/pkg/acp/protocol/helpers.go b/apps/backend/pkg/acp/protocol/helpers.go deleted file mode 100644 index 958eee880d..0000000000 --- a/apps/backend/pkg/acp/protocol/helpers.go +++ /dev/null @@ -1,71 +0,0 @@ -package protocol - -// NewProgressMessage creates a new progress message -func NewProgressMessage(agentID, taskID string, data ProgressData) *Message { - return NewMessage(MessageTypeProgress, agentID, taskID, map[string]interface{}{ - "progress": data.Progress, - "message": data.Message, - "current_file": data.CurrentFile, - "files_processed": data.FilesProcessed, - "total_files": data.TotalFiles, - }) -} - -// NewLogMessage creates a new log message -func NewLogMessage(agentID, taskID string, data LogData) *Message { - return NewMessage(MessageTypeLog, agentID, taskID, map[string]interface{}{ - "level": data.Level, - "message": data.Message, - "metadata": data.Metadata, - }) -} - -// NewResultMessage creates a new result message -func NewResultMessage(agentID, taskID string, data ResultData) *Message { - // Convert artifacts to interface slice - artifacts := make([]interface{}, len(data.Artifacts)) - for i, a := range data.Artifacts { - artifacts[i] = map[string]interface{}{ - "type": a.Type, - "path": a.Path, - "url": a.URL, - } - } - - return NewMessage(MessageTypeResult, agentID, taskID, map[string]interface{}{ - "status": data.Status, - "summary": data.Summary, - "artifacts": artifacts, - }) -} - -// NewErrorMessage creates a new error message -func NewErrorMessage(agentID, taskID string, data ErrorData) *Message { - return NewMessage(MessageTypeError, agentID, taskID, map[string]interface{}{ - "error": data.Error, - "file": data.File, - "details": data.Details, - }) -} - -// NewStatusMessage creates a new status message -func NewStatusMessage(agentID, taskID string, data StatusData) *Message { - return NewMessage(MessageTypeStatus, agentID, taskID, map[string]interface{}{ - "status": data.Status, - "message": data.Message, - }) -} - -// NewHeartbeatMessage creates a new heartbeat message -func NewHeartbeatMessage(agentID, taskID string) *Message { - return NewMessage(MessageTypeHeartbeat, agentID, taskID, map[string]interface{}{}) -} - -// NewControlMessage creates a new control message -func NewControlMessage(agentID, taskID string, data ControlData) *Message { - return NewMessage(MessageTypeControl, agentID, taskID, map[string]interface{}{ - "action": data.Action, - "reason": data.Reason, - }) -} - diff --git a/apps/backend/pkg/acp/protocol/helpers_test.go b/apps/backend/pkg/acp/protocol/helpers_test.go deleted file mode 100644 index c1eae261ab..0000000000 --- a/apps/backend/pkg/acp/protocol/helpers_test.go +++ /dev/null @@ -1,275 +0,0 @@ -package protocol - -import ( - "testing" -) - -func TestNewProgressMessage(t *testing.T) { - data := ProgressData{ - Progress: 50, - Message: "Processing files", - CurrentFile: "main.go", - FilesProcessed: 5, - TotalFiles: 10, - } - - msg := NewProgressMessage("agent-1", "task-1", data) - - if msg.Type != MessageTypeProgress { - t.Errorf("Expected type %s, got %s", MessageTypeProgress, msg.Type) - } - if msg.AgentID != "agent-1" { - t.Errorf("Expected agentID 'agent-1', got %s", msg.AgentID) - } - if msg.TaskID != "task-1" { - t.Errorf("Expected taskID 'task-1', got %s", msg.TaskID) - } - if msg.Data["progress"] != 50 { - t.Errorf("Expected progress 50, got %v", msg.Data["progress"]) - } - if msg.Data["message"] != "Processing files" { - t.Errorf("Expected message 'Processing files', got %v", msg.Data["message"]) - } - if msg.Data["current_file"] != "main.go" { - t.Errorf("Expected current_file 'main.go', got %v", msg.Data["current_file"]) - } - if msg.Data["files_processed"] != 5 { - t.Errorf("Expected files_processed 5, got %v", msg.Data["files_processed"]) - } - if msg.Data["total_files"] != 10 { - t.Errorf("Expected total_files 10, got %v", msg.Data["total_files"]) - } -} - -func TestNewLogMessage(t *testing.T) { - data := LogData{ - Level: "info", - Message: "Task started", - Metadata: map[string]interface{}{ - "component": "scanner", - }, - } - - msg := NewLogMessage("agent-2", "task-2", data) - - if msg.Type != MessageTypeLog { - t.Errorf("Expected type %s, got %s", MessageTypeLog, msg.Type) - } - if msg.AgentID != "agent-2" { - t.Errorf("Expected agentID 'agent-2', got %s", msg.AgentID) - } - if msg.TaskID != "task-2" { - t.Errorf("Expected taskID 'task-2', got %s", msg.TaskID) - } - if msg.Data["level"] != "info" { - t.Errorf("Expected level 'info', got %v", msg.Data["level"]) - } - if msg.Data["message"] != "Task started" { - t.Errorf("Expected message 'Task started', got %v", msg.Data["message"]) - } - metadata, ok := msg.Data["metadata"].(map[string]interface{}) - if !ok { - t.Fatal("Expected metadata to be a map") - } - if metadata["component"] != "scanner" { - t.Errorf("Expected component 'scanner', got %v", metadata["component"]) - } -} - -func TestNewResultMessage(t *testing.T) { - data := ResultData{ - Status: "completed", - Summary: "All checks passed", - Artifacts: []Artifact{ - {Type: "report", Path: "/reports/scan.json", URL: "https://example.com/report"}, - {Type: "log", Path: "/logs/scan.log", URL: ""}, - }, - } - - msg := NewResultMessage("agent-3", "task-3", data) - - if msg.Type != MessageTypeResult { - t.Errorf("Expected type %s, got %s", MessageTypeResult, msg.Type) - } - if msg.AgentID != "agent-3" { - t.Errorf("Expected agentID 'agent-3', got %s", msg.AgentID) - } - if msg.TaskID != "task-3" { - t.Errorf("Expected taskID 'task-3', got %s", msg.TaskID) - } - if msg.Data["status"] != "completed" { - t.Errorf("Expected status 'completed', got %v", msg.Data["status"]) - } - if msg.Data["summary"] != "All checks passed" { - t.Errorf("Expected summary 'All checks passed', got %v", msg.Data["summary"]) - } - - artifacts, ok := msg.Data["artifacts"].([]interface{}) - if !ok { - t.Fatal("Expected artifacts to be a slice") - } - if len(artifacts) != 2 { - t.Errorf("Expected 2 artifacts, got %d", len(artifacts)) - } - - artifact1, ok := artifacts[0].(map[string]interface{}) - if !ok { - t.Fatal("Expected first artifact to be a map") - } - if artifact1["type"] != "report" { - t.Errorf("Expected first artifact type 'report', got %v", artifact1["type"]) - } - if artifact1["path"] != "/reports/scan.json" { - t.Errorf("Expected first artifact path '/reports/scan.json', got %v", artifact1["path"]) - } -} - -func TestNewErrorMessage(t *testing.T) { - data := ErrorData{ - Error: "Failed to parse file", - File: "config.yaml", - Details: "Invalid YAML syntax on line 42", - } - - msg := NewErrorMessage("agent-4", "task-4", data) - - if msg.Type != MessageTypeError { - t.Errorf("Expected type %s, got %s", MessageTypeError, msg.Type) - } - if msg.AgentID != "agent-4" { - t.Errorf("Expected agentID 'agent-4', got %s", msg.AgentID) - } - if msg.TaskID != "task-4" { - t.Errorf("Expected taskID 'task-4', got %s", msg.TaskID) - } - if msg.Data["error"] != "Failed to parse file" { - t.Errorf("Expected error 'Failed to parse file', got %v", msg.Data["error"]) - } - if msg.Data["file"] != "config.yaml" { - t.Errorf("Expected file 'config.yaml', got %v", msg.Data["file"]) - } - if msg.Data["details"] != "Invalid YAML syntax on line 42" { - t.Errorf("Expected details, got %v", msg.Data["details"]) - } -} - -func TestNewStatusMessage(t *testing.T) { - data := StatusData{ - Status: "running", - Message: "Processing task", - } - - msg := NewStatusMessage("agent-5", "task-5", data) - - if msg.Type != MessageTypeStatus { - t.Errorf("Expected type %s, got %s", MessageTypeStatus, msg.Type) - } - if msg.Data["status"] != "running" { - t.Errorf("Expected status 'running', got %v", msg.Data["status"]) - } - if msg.Data["message"] != "Processing task" { - t.Errorf("Expected message 'Processing task', got %v", msg.Data["message"]) - } -} - -func TestNewHeartbeatMessage(t *testing.T) { - msg := NewHeartbeatMessage("agent-6", "task-6") - - if msg.Type != MessageTypeHeartbeat { - t.Errorf("Expected type %s, got %s", MessageTypeHeartbeat, msg.Type) - } - if msg.AgentID != "agent-6" { - t.Errorf("Expected agentID 'agent-6', got %s", msg.AgentID) - } - if msg.TaskID != "task-6" { - t.Errorf("Expected taskID 'task-6', got %s", msg.TaskID) - } - if len(msg.Data) != 0 { - t.Errorf("Expected empty data map, got %v", msg.Data) - } -} - -func TestNewControlMessage(t *testing.T) { - data := ControlData{ - Action: "pause", - Reason: "User requested pause", - } - - msg := NewControlMessage("agent-7", "task-7", data) - - if msg.Type != MessageTypeControl { - t.Errorf("Expected type %s, got %s", MessageTypeControl, msg.Type) - } - if msg.AgentID != "agent-7" { - t.Errorf("Expected agentID 'agent-7', got %s", msg.AgentID) - } - if msg.TaskID != "task-7" { - t.Errorf("Expected taskID 'task-7', got %s", msg.TaskID) - } - if msg.Data["action"] != "pause" { - t.Errorf("Expected action 'pause', got %v", msg.Data["action"]) - } - if msg.Data["reason"] != "User requested pause" { - t.Errorf("Expected reason 'User requested pause', got %v", msg.Data["reason"]) - } -} - -func TestNewProgressMessage_ZeroValues(t *testing.T) { - data := ProgressData{ - Progress: 0, - Message: "", - } - - msg := NewProgressMessage("agent-1", "task-1", data) - - if msg.Type != MessageTypeProgress { - t.Errorf("Expected type %s, got %s", MessageTypeProgress, msg.Type) - } - if msg.Data["progress"] != 0 { - t.Errorf("Expected progress 0, got %v", msg.Data["progress"]) - } - if msg.Data["message"] != "" { - t.Errorf("Expected empty message, got %v", msg.Data["message"]) - } -} - -func TestNewResultMessage_NoArtifacts(t *testing.T) { - data := ResultData{ - Status: "completed", - Summary: "No files generated", - Artifacts: []Artifact{}, - } - - msg := NewResultMessage("agent-1", "task-1", data) - - if msg.Type != MessageTypeResult { - t.Errorf("Expected type %s, got %s", MessageTypeResult, msg.Type) - } - artifacts, ok := msg.Data["artifacts"].([]interface{}) - if !ok { - t.Fatal("Expected artifacts to be a slice") - } - if len(artifacts) != 0 { - t.Errorf("Expected 0 artifacts, got %d", len(artifacts)) - } -} - -func TestNewLogMessage_AllLevels(t *testing.T) { - levels := []string{"debug", "info", "warn", "error"} - - for _, level := range levels { - t.Run(level, func(t *testing.T) { - data := LogData{ - Level: level, - Message: "Test message", - } - - msg := NewLogMessage("agent-1", "task-1", data) - - if msg.Data["level"] != level { - t.Errorf("Expected level '%s', got %v", level, msg.Data["level"]) - } - }) - } -} - diff --git a/apps/backend/pkg/acp/protocol/message.go b/apps/backend/pkg/acp/protocol/message.go index 77e258ba88..7d16f4b593 100644 --- a/apps/backend/pkg/acp/protocol/message.go +++ b/apps/backend/pkg/acp/protocol/message.go @@ -33,17 +33,6 @@ type Message struct { Data map[string]interface{} `json:"data"` } -// NewMessage creates a new ACP message with the current timestamp -func NewMessage(msgType MessageType, agentID, taskID string, data map[string]interface{}) *Message { - return &Message{ - Type: msgType, - Timestamp: time.Now().UTC(), - AgentID: agentID, - TaskID: taskID, - Data: data, - } -} - // MarshalJSON implements custom JSON marshaling func (m *Message) MarshalJSON() ([]byte, error) { type Alias Message @@ -56,15 +45,6 @@ func (m *Message) MarshalJSON() ([]byte, error) { }) } -// Parse parses a JSON string into an ACP message -func Parse(data []byte) (*Message, error) { - var msg Message - if err := json.Unmarshal(data, &msg); err != nil { - return nil, err - } - return &msg, nil -} - // IsValid checks if the message has required fields func (m *Message) IsValid() bool { if m.Type == "" { diff --git a/apps/backend/pkg/acp/protocol/message_test.go b/apps/backend/pkg/acp/protocol/message_test.go deleted file mode 100644 index a8f6327aab..0000000000 --- a/apps/backend/pkg/acp/protocol/message_test.go +++ /dev/null @@ -1,186 +0,0 @@ -package protocol - -import ( - "encoding/json" - "testing" - "time" -) - -func TestMessageType_Constants(t *testing.T) { - tests := []struct { - msgType MessageType - expected string - }{ - {MessageTypeProgress, "progress"}, - {MessageTypeLog, "log"}, - {MessageTypeResult, "result"}, - {MessageTypeError, "error"}, - {MessageTypeStatus, "status"}, - {MessageTypeHeartbeat, "heartbeat"}, - {MessageTypeControl, "control"}, - } - - for _, tt := range tests { - if string(tt.msgType) != tt.expected { - t.Errorf("Expected MessageType %s, got %s", tt.expected, tt.msgType) - } - } -} - -func TestNewMessage(t *testing.T) { - agentID := "agent-123" - taskID := "task-456" - data := map[string]interface{}{"key": "value"} - - before := time.Now().UTC() - msg := NewMessage(MessageTypeProgress, agentID, taskID, data) - after := time.Now().UTC() - - if msg.Type != MessageTypeProgress { - t.Errorf("Expected type %s, got %s", MessageTypeProgress, msg.Type) - } - if msg.AgentID != agentID { - t.Errorf("Expected agentID %s, got %s", agentID, msg.AgentID) - } - if msg.TaskID != taskID { - t.Errorf("Expected taskID %s, got %s", taskID, msg.TaskID) - } - if msg.Data["key"] != "value" { - t.Error("Expected data to contain key=value") - } - if msg.Timestamp.Before(before) || msg.Timestamp.After(after) { - t.Error("Expected timestamp to be set to current time") - } -} - -func TestMessage_MarshalJSON(t *testing.T) { - fixedTime := time.Date(2024, 1, 15, 10, 30, 0, 0, time.UTC) - msg := &Message{ - Type: MessageTypeLog, - Timestamp: fixedTime, - AgentID: "agent-001", - TaskID: "task-001", - Data: map[string]interface{}{"level": "info"}, - } - - data, err := json.Marshal(msg) - if err != nil { - t.Fatalf("MarshalJSON failed: %v", err) - } - - var result map[string]interface{} - if err := json.Unmarshal(data, &result); err != nil { - t.Fatalf("Failed to unmarshal result: %v", err) - } - - if result["type"] != "log" { - t.Errorf("Expected type 'log', got %v", result["type"]) - } - if result["agent_id"] != "agent-001" { - t.Errorf("Expected agent_id 'agent-001', got %v", result["agent_id"]) - } - if result["task_id"] != "task-001" { - t.Errorf("Expected task_id 'task-001', got %v", result["task_id"]) - } - - // Check timestamp is in RFC3339Nano format - ts, ok := result["timestamp"].(string) - if !ok { - t.Fatal("Expected timestamp to be a string") - } - _, err = time.Parse(time.RFC3339Nano, ts) - if err != nil { - t.Errorf("Timestamp not in RFC3339Nano format: %v", err) - } -} - -func TestParse(t *testing.T) { - jsonData := []byte(`{ - "type": "progress", - "timestamp": "2024-01-15T10:30:00Z", - "agent_id": "agent-test", - "task_id": "task-test", - "data": {"progress": 50, "message": "Processing..."} - }`) - - msg, err := Parse(jsonData) - if err != nil { - t.Fatalf("Parse failed: %v", err) - } - - if msg.Type != MessageTypeProgress { - t.Errorf("Expected type %s, got %s", MessageTypeProgress, msg.Type) - } - if msg.AgentID != "agent-test" { - t.Errorf("Expected agentID 'agent-test', got %s", msg.AgentID) - } - if msg.TaskID != "task-test" { - t.Errorf("Expected taskID 'task-test', got %s", msg.TaskID) - } - if msg.Data["progress"] != float64(50) { - t.Errorf("Expected progress 50, got %v", msg.Data["progress"]) - } -} - -func TestParse_InvalidJSON(t *testing.T) { - invalidJSON := []byte(`{invalid json}`) - - _, err := Parse(invalidJSON) - if err == nil { - t.Error("Expected error for invalid JSON") - } -} - -func TestMessage_IsValid(t *testing.T) { - tests := []struct { - name string - msg *Message - expected bool - }{ - { - name: "valid message", - msg: &Message{ - Type: MessageTypeProgress, - AgentID: "agent-1", - TaskID: "task-1", - }, - expected: true, - }, - { - name: "missing type", - msg: &Message{ - Type: "", - AgentID: "agent-1", - TaskID: "task-1", - }, - expected: false, - }, - { - name: "missing agent_id", - msg: &Message{ - Type: MessageTypeProgress, - AgentID: "", - TaskID: "task-1", - }, - expected: false, - }, - { - name: "missing task_id", - msg: &Message{ - Type: MessageTypeProgress, - AgentID: "agent-1", - TaskID: "", - }, - expected: false, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - if got := tt.msg.IsValid(); got != tt.expected { - t.Errorf("IsValid() = %v, expected %v", got, tt.expected) - } - }) - } -} - diff --git a/apps/backend/pkg/codex/client.go b/apps/backend/pkg/codex/client.go index cafe761866..076dabc280 100644 --- a/apps/backend/pkg/codex/client.go +++ b/apps/backend/pkg/codex/client.go @@ -232,18 +232,3 @@ func (c *Client) handleRequest(id interface{}, method string, params json.RawMes } } -// extractMethod extracts the method field from a JSON message for logging -func extractMethod(data []byte) string { - var msg struct { - Method string `json:"method"` - ID int `json:"id"` - } - if err := json.Unmarshal(data, &msg); err != nil { - return "parse-error" - } - if msg.Method != "" { - return msg.Method - } - return "response" -} - diff --git a/apps/backend/pkg/websocket/message.go b/apps/backend/pkg/websocket/message.go index 254a80eb73..9b7fe6ec2d 100644 --- a/apps/backend/pkg/websocket/message.go +++ b/apps/backend/pkg/websocket/message.go @@ -57,6 +57,8 @@ type ErrorPayload struct { Details map[string]interface{} `json:"details,omitempty"` } + + // NewRequest creates a new request message func NewRequest(id, action string, payload interface{}) (*Message, error) { data, err := json.Marshal(payload)