diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 06a5a089d0..1bcfffb8fb 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -27,16 +27,18 @@ import ( gateways "github.com/kandev/kandev/internal/gateway/websocket" // Agent Manager packages + agentcontroller "github.com/kandev/kandev/internal/agent/controller" "github.com/kandev/kandev/internal/agent/credentials" "github.com/kandev/kandev/internal/agent/docker" + agenthandlers "github.com/kandev/kandev/internal/agent/handlers" "github.com/kandev/kandev/internal/agent/lifecycle" "github.com/kandev/kandev/internal/agent/registry" - agentwshandlers "github.com/kandev/kandev/internal/agent/wshandlers" // Orchestrator packages "github.com/kandev/kandev/internal/orchestrator" + orchestratorcontroller "github.com/kandev/kandev/internal/orchestrator/controller" "github.com/kandev/kandev/internal/orchestrator/executor" - orchestratorwshandlers "github.com/kandev/kandev/internal/orchestrator/wshandlers" + orchestratorhandlers "github.com/kandev/kandev/internal/orchestrator/handlers" // Task Service packages taskcontroller "github.com/kandev/kandev/internal/task/controller" @@ -219,14 +221,18 @@ func main() { taskController := taskcontroller.NewTaskController(taskSvc) repositoryController := taskcontroller.NewRepositoryController(taskSvc) - orchestratorWSHandlers := orchestratorwshandlers.NewHandlers(orchestratorSvc, log) - orchestratorWSHandlers.SetACPHandler(acpHandler) - orchestratorWSHandlers.RegisterHandlers(gateway.Dispatcher) + // Create orchestrator controller and handlers (Pattern A) + orchestratorCtrl := orchestratorcontroller.NewController(orchestratorSvc) + orchestratorCtrl.SetACPHandler(acpHandler) + orchestratorHandlers := orchestratorhandlers.NewHandlers(orchestratorCtrl, log) + orchestratorHandlers.RegisterHandlers(gateway.Dispatcher) log.Info("Registered Orchestrator WebSocket handlers") + // Create agent controller and handlers (Pattern A) if lifecycleMgr != nil && agentRegistry != nil { - agentWSHandlers := agentwshandlers.NewHandlers(lifecycleMgr, agentRegistry, log) - agentWSHandlers.RegisterHandlers(gateway.Dispatcher) + agentCtrl := agentcontroller.NewController(lifecycleMgr, agentRegistry) + agentHandlers := agenthandlers.NewHandlers(agentCtrl, log) + agentHandlers.RegisterHandlers(gateway.Dispatcher) log.Info("Registered Agent Manager WebSocket handlers") } diff --git a/apps/backend/internal/agent/controller/controller.go b/apps/backend/internal/agent/controller/controller.go new file mode 100644 index 0000000000..6b843c67ca --- /dev/null +++ b/apps/backend/internal/agent/controller/controller.go @@ -0,0 +1,164 @@ +// Package controller provides the business logic coordination layer for the agent module. +package controller + +import ( + "context" + + "github.com/kandev/kandev/internal/agent/dto" + "github.com/kandev/kandev/internal/agent/lifecycle" + "github.com/kandev/kandev/internal/agent/registry" +) + +// Controller coordinates agent business logic +type Controller struct { + lifecycle *lifecycle.Manager + registry *registry.Registry +} + +// NewController creates a new agent controller +func NewController(lm *lifecycle.Manager, reg *registry.Registry) *Controller { + return &Controller{ + lifecycle: lm, + registry: reg, + } +} + +// ListAgents returns all agent instances +func (c *Controller) ListAgents(ctx context.Context, req dto.ListAgentsRequest) (dto.ListAgentsResponse, error) { + if c.lifecycle == nil { + return dto.ListAgentsResponse{}, ErrLifecycleManagerNotAvailable + } + + agents := c.lifecycle.ListInstances() + resp := dto.ListAgentsResponse{ + Agents: make([]dto.AgentDTO, 0, len(agents)), + Total: len(agents), + } + + for _, a := range agents { + agent := dto.FromAgentInstance(&dto.AgentInstanceData{ + ID: a.ID, + TaskID: a.TaskID, + AgentType: a.AgentType, + ContainerID: a.ContainerID, + Status: string(a.Status), + Progress: a.Progress, + StartedAt: a.StartedAt, + FinishedAt: a.FinishedAt, + ExitCode: a.ExitCode, + ErrorMessage: a.ErrorMessage, + }) + resp.Agents = append(resp.Agents, agent) + } + + return resp, nil +} + +// LaunchAgent launches a new agent instance +func (c *Controller) LaunchAgent(ctx context.Context, req dto.LaunchAgentRequest) (dto.LaunchAgentResponse, error) { + if c.lifecycle == nil { + return dto.LaunchAgentResponse{}, ErrLifecycleManagerNotAvailable + } + + launchReq := &lifecycle.LaunchRequest{ + TaskID: req.TaskID, + AgentType: req.AgentType, + WorkspacePath: req.WorkspacePath, + Env: req.Env, + } + + instance, err := c.lifecycle.Launch(ctx, launchReq) + if err != nil { + return dto.LaunchAgentResponse{}, err + } + + return dto.LaunchAgentResponse{ + Success: true, + AgentID: instance.ID, + TaskID: req.TaskID, + }, nil +} + +// GetAgentStatus returns the status of a specific agent +func (c *Controller) GetAgentStatus(ctx context.Context, req dto.GetAgentStatusRequest) (dto.AgentDTO, error) { + if c.lifecycle == nil { + return dto.AgentDTO{}, ErrLifecycleManagerNotAvailable + } + + agent, found := c.lifecycle.GetInstance(req.AgentID) + if !found { + return dto.AgentDTO{}, ErrAgentNotFound + } + + return dto.FromAgentInstance(&dto.AgentInstanceData{ + ID: agent.ID, + TaskID: agent.TaskID, + AgentType: agent.AgentType, + ContainerID: agent.ContainerID, + Status: string(agent.Status), + Progress: agent.Progress, + StartedAt: agent.StartedAt, + FinishedAt: agent.FinishedAt, + ExitCode: agent.ExitCode, + ErrorMessage: agent.ErrorMessage, + }), nil +} + +// GetAgentLogs returns logs for a specific agent +func (c *Controller) GetAgentLogs(ctx context.Context, req dto.GetAgentLogsRequest) (dto.GetAgentLogsResponse, error) { + if c.lifecycle == nil { + return dto.GetAgentLogsResponse{}, ErrLifecycleManagerNotAvailable + } + + _, found := c.lifecycle.GetInstance(req.AgentID) + if !found { + return dto.GetAgentLogsResponse{}, ErrAgentNotFound + } + + // Note: Log retrieval requires Docker client access which is not available here + return dto.GetAgentLogsResponse{ + AgentID: req.AgentID, + Logs: []string{}, + Message: "Use HTTP API for full log retrieval", + }, nil +} + +// StopAgent stops a running agent +func (c *Controller) StopAgent(ctx context.Context, req dto.StopAgentRequest) (dto.SuccessResponse, error) { + if c.lifecycle == nil { + return dto.SuccessResponse{}, ErrLifecycleManagerNotAvailable + } + + if err := c.lifecycle.StopAgent(ctx, req.AgentID, false); err != nil { + return dto.SuccessResponse{}, err + } + + return dto.SuccessResponse{Success: true}, nil +} + +// ListAgentTypes returns all available agent types +func (c *Controller) ListAgentTypes(ctx context.Context, req dto.ListAgentTypesRequest) (dto.ListAgentTypesResponse, error) { + if c.registry == nil { + return dto.ListAgentTypesResponse{}, ErrRegistryNotAvailable + } + + types := c.registry.List() + resp := dto.ListAgentTypesResponse{ + Types: make([]dto.AgentTypeDTO, 0, len(types)), + Total: len(types), + } + + for _, t := range types { + resp.Types = append(resp.Types, dto.FromAgentType(&dto.AgentTypeData{ + ID: t.ID, + Name: t.Name, + Description: t.Description, + Image: t.Image, + Capabilities: t.Capabilities, + Enabled: t.Enabled, + })) + } + + return resp, nil +} + diff --git a/apps/backend/internal/agent/controller/errors.go b/apps/backend/internal/agent/controller/errors.go new file mode 100644 index 0000000000..eb01f54a1d --- /dev/null +++ b/apps/backend/internal/agent/controller/errors.go @@ -0,0 +1,15 @@ +package controller + +import "errors" + +var ( + // ErrLifecycleManagerNotAvailable is returned when lifecycle manager is nil + ErrLifecycleManagerNotAvailable = errors.New("agent lifecycle manager not available") + + // ErrRegistryNotAvailable is returned when registry is nil + ErrRegistryNotAvailable = errors.New("agent registry not available") + + // ErrAgentNotFound is returned when agent is not found + ErrAgentNotFound = errors.New("agent not found") +) + diff --git a/apps/backend/internal/agent/dto/dto.go b/apps/backend/internal/agent/dto/dto.go new file mode 100644 index 0000000000..6d599cc21c --- /dev/null +++ b/apps/backend/internal/agent/dto/dto.go @@ -0,0 +1,149 @@ +// Package dto provides Data Transfer Objects for the agent module. +package dto + +import "time" + +// ListAgentsRequest is the request for agent.list +type ListAgentsRequest struct{} + +// AgentDTO represents an agent instance +type AgentDTO struct { + ID string `json:"id"` + TaskID string `json:"task_id"` + AgentType string `json:"agent_type"` + ContainerID string `json:"container_id,omitempty"` + Status string `json:"status"` + Progress int `json:"progress"` + StartedAt string `json:"started_at"` + FinishedAt string `json:"finished_at,omitempty"` + ExitCode *int `json:"exit_code,omitempty"` + Error string `json:"error,omitempty"` + Metadata map[string]string `json:"metadata,omitempty"` +} + +// ListAgentsResponse is the response for agent.list +type ListAgentsResponse struct { + Agents []AgentDTO `json:"agents"` + Total int `json:"total"` +} + +// LaunchAgentRequest is the payload for agent.launch +type LaunchAgentRequest struct { + TaskID string `json:"task_id"` + AgentType string `json:"agent_type"` + WorkspacePath string `json:"workspace_path"` + Env map[string]string `json:"env,omitempty"` +} + +// LaunchAgentResponse is the response for agent.launch +type LaunchAgentResponse struct { + Success bool `json:"success"` + AgentID string `json:"agent_id"` + TaskID string `json:"task_id"` +} + +// GetAgentStatusRequest is the payload for agent.status +type GetAgentStatusRequest struct { + AgentID string `json:"agent_id"` +} + +// GetAgentLogsRequest is the payload for agent.logs +type GetAgentLogsRequest struct { + AgentID string `json:"agent_id"` +} + +// GetAgentLogsResponse is the response for agent.logs +type GetAgentLogsResponse struct { + AgentID string `json:"agent_id"` + Logs []string `json:"logs"` + Message string `json:"message,omitempty"` +} + +// StopAgentRequest is the payload for agent.stop +type StopAgentRequest struct { + AgentID string `json:"agent_id"` +} + +// SuccessResponse is a generic success response +type SuccessResponse struct { + Success bool `json:"success"` +} + +// ListAgentTypesRequest is the request for agent.types +type ListAgentTypesRequest struct{} + +// AgentTypeDTO represents an agent type +type AgentTypeDTO struct { + ID string `json:"id"` + Name string `json:"name"` + Description string `json:"description,omitempty"` + Image string `json:"image"` + Capabilities []string `json:"capabilities,omitempty"` + Enabled bool `json:"enabled"` +} + +// ListAgentTypesResponse is the response for agent.types +type ListAgentTypesResponse struct { + Types []AgentTypeDTO `json:"types"` + Total int `json:"total"` +} + +// FromAgentInstance converts a lifecycle.AgentInstance to AgentDTO +func FromAgentInstance(a *AgentInstanceData) AgentDTO { + agent := AgentDTO{ + ID: a.ID, + TaskID: a.TaskID, + AgentType: a.AgentType, + ContainerID: a.ContainerID, + Status: a.Status, + Progress: a.Progress, + StartedAt: a.StartedAt.Format("2006-01-02T15:04:05Z"), + } + if a.FinishedAt != nil && !a.FinishedAt.IsZero() { + agent.FinishedAt = a.FinishedAt.Format("2006-01-02T15:04:05Z") + } + if a.ExitCode != nil { + agent.ExitCode = a.ExitCode + } + if a.ErrorMessage != "" { + agent.Error = a.ErrorMessage + } + return agent +} + +// AgentInstanceData is the input data for FromAgentInstance (avoids circular import) +type AgentInstanceData struct { + ID string + TaskID string + AgentType string + ContainerID string + Status string + Progress int + StartedAt time.Time + FinishedAt *time.Time + ExitCode *int + ErrorMessage string +} + +// FromAgentType converts a registry.AgentType to AgentTypeDTO +func FromAgentType(t *AgentTypeData) AgentTypeDTO { + return AgentTypeDTO{ + ID: t.ID, + Name: t.Name, + Description: t.Description, + Image: t.Image, + Capabilities: t.Capabilities, + Enabled: t.Enabled, + } +} + +// AgentTypeData is the input data for FromAgentType (avoids circular import) +type AgentTypeData struct { + ID string + Name string + Description string + Image string + Capabilities []string + Enabled bool +} + diff --git a/apps/backend/internal/agent/handlers/handlers.go b/apps/backend/internal/agent/handlers/handlers.go new file mode 100644 index 0000000000..118de32446 --- /dev/null +++ b/apps/backend/internal/agent/handlers/handlers.go @@ -0,0 +1,156 @@ +// Package handlers provides WebSocket message handlers for the agent module. +package handlers + +import ( + "context" + + "github.com/kandev/kandev/internal/agent/controller" + "github.com/kandev/kandev/internal/agent/dto" + "github.com/kandev/kandev/internal/common/logger" + ws "github.com/kandev/kandev/pkg/websocket" + "go.uber.org/zap" +) + +// Handlers contains WebSocket handlers for the agent API +type Handlers struct { + controller *controller.Controller + logger *logger.Logger +} + +// NewHandlers creates a new WebSocket handlers instance +func NewHandlers(ctrl *controller.Controller, log *logger.Logger) *Handlers { + return &Handlers{ + controller: ctrl, + logger: log.WithFields(zap.String("component", "agent-handlers")), + } +} + +// RegisterHandlers registers all agent handlers with the dispatcher +func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { + d.RegisterFunc(ws.ActionAgentList, h.wsListAgents) + d.RegisterFunc(ws.ActionAgentLaunch, h.wsLaunchAgent) + d.RegisterFunc(ws.ActionAgentStatus, h.wsGetAgentStatus) + d.RegisterFunc(ws.ActionAgentLogs, h.wsGetAgentLogs) + d.RegisterFunc(ws.ActionAgentStop, h.wsStopAgent) + d.RegisterFunc(ws.ActionAgentTypes, h.wsListAgentTypes) +} + +// WS handlers + +func (h *Handlers) wsListAgents(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + resp, err := h.controller.ListAgents(ctx, dto.ListAgentsRequest{}) + if err != nil { + h.logger.Error("failed to list agents", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +type wsLaunchAgentRequest struct { + TaskID string `json:"task_id"` + AgentType string `json:"agent_type"` + WorkspacePath string `json:"workspace_path"` + Env map[string]string `json:"env,omitempty"` +} + +func (h *Handlers) wsLaunchAgent(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsLaunchAgentRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + if req.AgentType == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_type is required", nil) + } + if req.WorkspacePath == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "workspace_path is required", nil) + } + + resp, err := h.controller.LaunchAgent(ctx, dto.LaunchAgentRequest{ + TaskID: req.TaskID, + AgentType: req.AgentType, + WorkspacePath: req.WorkspacePath, + Env: req.Env, + }) + if err != nil { + h.logger.Error("failed to launch agent", + zap.String("task_id", req.TaskID), + zap.String("agent_type", req.AgentType), + zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to launch agent: "+err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +type wsAgentIDRequest struct { + AgentID string `json:"agent_id"` +} + +func (h *Handlers) wsGetAgentStatus(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsAgentIDRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.AgentID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_id is required", nil) + } + + resp, err := h.controller.GetAgentStatus(ctx, dto.GetAgentStatusRequest{AgentID: req.AgentID}) + if err != nil { + if err == controller.ErrAgentNotFound { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "Agent not found", nil) + } + h.logger.Error("failed to get agent status", zap.String("agent_id", req.AgentID), zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +func (h *Handlers) wsGetAgentLogs(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsAgentIDRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.AgentID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_id is required", nil) + } + + resp, err := h.controller.GetAgentLogs(ctx, dto.GetAgentLogsRequest{AgentID: req.AgentID}) + if err != nil { + if err == controller.ErrAgentNotFound { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "Agent not found", nil) + } + h.logger.Error("failed to get agent logs", zap.String("agent_id", req.AgentID), zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +func (h *Handlers) wsStopAgent(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsAgentIDRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.AgentID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_id is required", nil) + } + + resp, err := h.controller.StopAgent(ctx, dto.StopAgentRequest{AgentID: req.AgentID}) + if err != nil { + h.logger.Error("failed to stop agent", zap.String("agent_id", req.AgentID), zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to stop agent: "+err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +func (h *Handlers) wsListAgentTypes(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + resp, err := h.controller.ListAgentTypes(ctx, dto.ListAgentTypesRequest{}) + if err != nil { + h.logger.Error("failed to list agent types", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + diff --git a/apps/backend/internal/agent/wshandlers/handlers.go b/apps/backend/internal/agent/wshandlers/handlers.go deleted file mode 100644 index a4d0d1df5f..0000000000 --- a/apps/backend/internal/agent/wshandlers/handlers.go +++ /dev/null @@ -1,290 +0,0 @@ -// Package wshandlers provides WebSocket message handlers for the agent manager. -package wshandlers - -import ( - "context" - - "github.com/kandev/kandev/internal/agent/lifecycle" - "github.com/kandev/kandev/internal/agent/registry" - "github.com/kandev/kandev/internal/common/logger" - ws "github.com/kandev/kandev/pkg/websocket" - "go.uber.org/zap" -) - -// Handlers contains WebSocket handlers for the agent API -type Handlers struct { - lifecycle *lifecycle.Manager - registry *registry.Registry - logger *logger.Logger -} - -// NewHandlers creates a new WebSocket handlers instance -func NewHandlers(lm *lifecycle.Manager, reg *registry.Registry, log *logger.Logger) *Handlers { - return &Handlers{ - lifecycle: lm, - registry: reg, - logger: log.WithFields(zap.String("component", "agent-ws-handlers")), - } -} - -// RegisterHandlers registers all agent handlers with the dispatcher -func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { - d.RegisterFunc(ws.ActionAgentList, h.ListAgents) - d.RegisterFunc(ws.ActionAgentLaunch, h.LaunchAgent) - d.RegisterFunc(ws.ActionAgentStatus, h.GetAgentStatus) - d.RegisterFunc(ws.ActionAgentLogs, h.GetAgentLogs) - d.RegisterFunc(ws.ActionAgentStop, h.StopAgent) - d.RegisterFunc(ws.ActionAgentTypes, h.ListAgentTypes) -} - -// AgentResponse represents an agent instance -type AgentResponse struct { - ID string `json:"id"` - TaskID string `json:"task_id"` - AgentType string `json:"agent_type"` - ContainerID string `json:"container_id,omitempty"` - Status string `json:"status"` - Progress int `json:"progress"` - StartedAt string `json:"started_at"` - FinishedAt string `json:"finished_at,omitempty"` - ExitCode *int `json:"exit_code,omitempty"` - Error string `json:"error,omitempty"` - Metadata map[string]string `json:"metadata,omitempty"` -} - -// ListAgentsResponse is the response for agent.list -type ListAgentsResponse struct { - Agents []AgentResponse `json:"agents"` - Total int `json:"total"` -} - -// ListAgents handles agent.list action -func (h *Handlers) ListAgents(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - if h.lifecycle == nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Agent manager not available", nil) - } - - agents := h.lifecycle.ListInstances() - resp := ListAgentsResponse{ - Agents: make([]AgentResponse, 0, len(agents)), - Total: len(agents), - } - - for _, a := range agents { - agent := AgentResponse{ - ID: a.ID, - TaskID: a.TaskID, - AgentType: a.AgentType, - ContainerID: a.ContainerID, - Status: string(a.Status), - Progress: a.Progress, - StartedAt: a.StartedAt.Format("2006-01-02T15:04:05Z"), - } - if a.FinishedAt != nil && !a.FinishedAt.IsZero() { - agent.FinishedAt = a.FinishedAt.Format("2006-01-02T15:04:05Z") - } - if a.ExitCode != nil { - agent.ExitCode = a.ExitCode - } - if a.ErrorMessage != "" { - agent.Error = a.ErrorMessage - } - resp.Agents = append(resp.Agents, agent) - } - - return ws.NewResponse(msg.ID, msg.Action, resp) -} - -// LaunchAgentRequest is the payload for agent.launch -type LaunchAgentRequest struct { - TaskID string `json:"task_id"` - AgentType string `json:"agent_type"` - WorkspacePath string `json:"workspace_path"` - Env map[string]string `json:"env,omitempty"` -} - -// LaunchAgent handles agent.launch action -func (h *Handlers) LaunchAgent(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - var req LaunchAgentRequest - if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) - } - - if req.TaskID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) - } - if req.AgentType == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_type is required", nil) - } - if req.WorkspacePath == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "workspace_path is required", nil) - } - - if h.lifecycle == nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Agent manager not available", nil) - } - - launchReq := &lifecycle.LaunchRequest{ - TaskID: req.TaskID, - AgentType: req.AgentType, - WorkspacePath: req.WorkspacePath, - Env: req.Env, - } - - instance, err := h.lifecycle.Launch(ctx, launchReq) - if err != nil { - h.logger.Error("failed to launch agent", - zap.String("task_id", req.TaskID), - zap.String("agent_type", req.AgentType), - zap.Error(err)) - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to launch agent: "+err.Error(), nil) - } - - return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ - "success": true, - "agent_id": instance.ID, - "task_id": req.TaskID, - }) -} - -// AgentIDRequest is a common request with just agent_id -type AgentIDRequest struct { - AgentID string `json:"agent_id"` -} - -// GetAgentStatus handles agent.status action -func (h *Handlers) GetAgentStatus(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - var req AgentIDRequest - if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) - } - - if req.AgentID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_id is required", nil) - } - - if h.lifecycle == nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Agent manager not available", nil) - } - - agent, found := h.lifecycle.GetInstance(req.AgentID) - if !found { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "Agent not found", nil) - } - - resp := AgentResponse{ - ID: agent.ID, - TaskID: agent.TaskID, - AgentType: agent.AgentType, - ContainerID: agent.ContainerID, - Status: string(agent.Status), - Progress: agent.Progress, - StartedAt: agent.StartedAt.Format("2006-01-02T15:04:05Z"), - } - if agent.FinishedAt != nil && !agent.FinishedAt.IsZero() { - resp.FinishedAt = agent.FinishedAt.Format("2006-01-02T15:04:05Z") - } - if agent.ExitCode != nil { - resp.ExitCode = agent.ExitCode - } - if agent.ErrorMessage != "" { - resp.Error = agent.ErrorMessage - } - - return ws.NewResponse(msg.ID, msg.Action, resp) -} - -// GetAgentLogs handles agent.logs action -func (h *Handlers) GetAgentLogs(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - var req AgentIDRequest - if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) - } - - if req.AgentID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_id is required", nil) - } - - if h.lifecycle == nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Agent manager not available", nil) - } - - _, found := h.lifecycle.GetInstance(req.AgentID) - if !found { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "Agent not found", nil) - } - - // Note: Log retrieval requires Docker client access which is not available in wshandlers - // This handler returns a stub response; full log retrieval should use the HTTP API - return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ - "agent_id": req.AgentID, - "logs": []string{}, - "message": "Use HTTP API for full log retrieval", - }) -} - -// StopAgent handles agent.stop action -func (h *Handlers) StopAgent(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - var req AgentIDRequest - if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) - } - - if req.AgentID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "agent_id is required", nil) - } - - if h.lifecycle == nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Agent manager not available", nil) - } - - if err := h.lifecycle.StopAgent(ctx, req.AgentID, false); err != nil { - h.logger.Error("failed to stop agent", zap.String("agent_id", req.AgentID), zap.Error(err)) - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to stop agent: "+err.Error(), nil) - } - - return ws.NewResponse(msg.ID, msg.Action, map[string]bool{"success": true}) -} - -// AgentTypeResponse represents an agent type -type AgentTypeResponse struct { - ID string `json:"id"` - Name string `json:"name"` - Description string `json:"description,omitempty"` - Image string `json:"image"` - Capabilities []string `json:"capabilities,omitempty"` - Enabled bool `json:"enabled"` -} - -// ListAgentTypesResponse is the response for agent.types -type ListAgentTypesResponse struct { - Types []AgentTypeResponse `json:"types"` - Total int `json:"total"` -} - -// ListAgentTypes handles agent.types action -func (h *Handlers) ListAgentTypes(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - if h.registry == nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Registry not available", nil) - } - - types := h.registry.List() - resp := ListAgentTypesResponse{ - Types: make([]AgentTypeResponse, 0, len(types)), - Total: len(types), - } - - for _, t := range types { - resp.Types = append(resp.Types, AgentTypeResponse{ - ID: t.ID, - Name: t.Name, - Description: t.Description, - Image: t.Image, - Capabilities: t.Capabilities, - Enabled: t.Enabled, - }) - } - - return ws.NewResponse(msg.ID, msg.Action, resp) -} - diff --git a/apps/backend/internal/integration/orchestrator_test.go b/apps/backend/internal/integration/orchestrator_test.go index 65f6d1e922..733ccdc680 100644 --- a/apps/backend/internal/integration/orchestrator_test.go +++ b/apps/backend/internal/integration/orchestrator_test.go @@ -26,8 +26,9 @@ import ( "github.com/kandev/kandev/internal/events/bus" gateways "github.com/kandev/kandev/internal/gateway/websocket" "github.com/kandev/kandev/internal/orchestrator" + orchestratorcontroller "github.com/kandev/kandev/internal/orchestrator/controller" "github.com/kandev/kandev/internal/orchestrator/executor" - orchestratorwshandlers "github.com/kandev/kandev/internal/orchestrator/wshandlers" + orchestratorhandlers "github.com/kandev/kandev/internal/orchestrator/handlers" taskcontroller "github.com/kandev/kandev/internal/task/controller" taskhandlers "github.com/kandev/kandev/internal/task/handlers" "github.com/kandev/kandev/internal/task/repository" @@ -469,9 +470,10 @@ func NewOrchestratorTestServer(t *testing.T) *OrchestratorTestServer { // Create WebSocket gateway gateway := gateways.NewGateway(log) - // Register orchestrator handlers - orchestratorWSHandlers := orchestratorwshandlers.NewHandlers(orchestratorSvc, log) - orchestratorWSHandlers.RegisterHandlers(gateway.Dispatcher) + // Register orchestrator handlers (Pattern A) + orchestratorCtrl := orchestratorcontroller.NewController(orchestratorSvc) + orchestratorHandlers := orchestratorhandlers.NewHandlers(orchestratorCtrl, log) + orchestratorHandlers.RegisterHandlers(gateway.Dispatcher) // Start hub go gateway.Hub.Run(ctx) diff --git a/apps/backend/internal/orchestrator/controller/controller.go b/apps/backend/internal/orchestrator/controller/controller.go new file mode 100644 index 0000000000..5a466e773b --- /dev/null +++ b/apps/backend/internal/orchestrator/controller/controller.go @@ -0,0 +1,169 @@ +// Package controller provides the business logic coordination layer for the orchestrator. +package controller + +import ( + "context" + + "github.com/kandev/kandev/internal/orchestrator" + "github.com/kandev/kandev/internal/orchestrator/acp" + "github.com/kandev/kandev/internal/orchestrator/dto" +) + +// Controller coordinates orchestrator business logic +type Controller struct { + service *orchestrator.Service + acpHandler *acp.Handler +} + +// NewController creates a new orchestrator controller +func NewController(svc *orchestrator.Service) *Controller { + return &Controller{ + service: svc, + } +} + +// SetACPHandler sets the ACP handler for log retrieval +func (c *Controller) SetACPHandler(handler *acp.Handler) { + c.acpHandler = handler +} + +// GetStatus returns the orchestrator status +func (c *Controller) GetStatus(ctx context.Context, req dto.GetStatusRequest) (dto.StatusResponse, error) { + status := c.service.GetStatus() + return dto.StatusResponse{ + Running: status.Running, + ActiveAgents: status.ActiveAgents, + QueuedTasks: status.QueuedTasks, + MaxConcurrent: 0, // Not available from service Status; would need scheduler access + }, nil +} + +// GetQueue returns the queued tasks +func (c *Controller) GetQueue(ctx context.Context, req dto.GetQueueRequest) (dto.QueueResponse, error) { + queuedTasks := c.service.GetQueuedTasks() + + tasks := make([]dto.QueuedTaskDTO, 0, len(queuedTasks)) + for _, qt := range queuedTasks { + tasks = append(tasks, dto.QueuedTaskDTO{ + TaskID: qt.TaskID, + Priority: qt.Priority, + QueuedAt: qt.QueuedAt.Format("2006-01-02T15:04:05Z"), + }) + } + + return dto.QueueResponse{ + Tasks: tasks, + Total: len(tasks), + }, nil +} + +// TriggerTask triggers a task for execution +func (c *Controller) TriggerTask(ctx context.Context, req dto.TriggerTaskRequest) (dto.TriggerTaskResponse, error) { + // For now, just return success - the service would need task details + return dto.TriggerTaskResponse{ + Success: true, + Message: "Task triggered", + TaskID: req.TaskID, + }, nil +} + +// StartTask starts task execution +func (c *Controller) StartTask(ctx context.Context, req dto.StartTaskRequest) (dto.StartTaskResponse, error) { + execution, err := c.service.StartTask(ctx, req.TaskID, req.AgentType, req.Priority) + if err != nil { + return dto.StartTaskResponse{}, err + } + + return dto.StartTaskResponse{ + Success: true, + TaskID: execution.TaskID, + AgentInstanceID: execution.AgentInstanceID, + Status: string(execution.Status), + }, nil +} + +// StopTask stops task execution +func (c *Controller) StopTask(ctx context.Context, req dto.StopTaskRequest) (dto.SuccessResponse, error) { + reason := req.Reason + if reason == "" { + reason = "stopped via API" + } + + if err := c.service.StopTask(ctx, req.TaskID, reason, req.Force); err != nil { + return dto.SuccessResponse{}, err + } + + return dto.SuccessResponse{Success: true}, nil +} + +// PromptTask sends a prompt to a running task +func (c *Controller) PromptTask(ctx context.Context, req dto.PromptTaskRequest) (dto.PromptTaskResponse, error) { + result, err := c.service.PromptTask(ctx, req.TaskID, req.Prompt) + if err != nil { + return dto.PromptTaskResponse{}, err + } + + return dto.PromptTaskResponse{ + Success: true, + StopReason: result.StopReason, + }, nil +} + +// CompleteTask marks a task as complete +func (c *Controller) CompleteTask(ctx context.Context, req dto.CompleteTaskRequest) (dto.CompleteTaskResponse, error) { + if err := c.service.CompleteTask(ctx, req.TaskID); err != nil { + return dto.CompleteTaskResponse{}, err + } + + return dto.CompleteTaskResponse{ + Success: true, + Message: "task completed", + }, nil +} + +// GetTaskLogs retrieves historical logs for a task +func (c *Controller) GetTaskLogs(ctx context.Context, req dto.GetTaskLogsRequest) (dto.GetTaskLogsResponse, error) { + if c.acpHandler == nil { + return dto.GetTaskLogsResponse{}, ErrACPHandlerNotConfigured + } + + messages, err := c.acpHandler.GetAllMessages(ctx, req.TaskID) + if err != nil { + return dto.GetTaskLogsResponse{}, err + } + + // Apply limit if specified + if req.Limit > 0 && len(messages) > req.Limit { + messages = messages[len(messages)-req.Limit:] + } + + logs := make([]dto.LogEntryDTO, 0, len(messages)) + for _, m := range messages { + logs = append(logs, dto.LogEntryDTO{ + Type: string(m.Type), + Timestamp: m.Timestamp.Format("2006-01-02T15:04:05Z"), + AgentID: m.AgentID, + Data: m.Data, + }) + } + + return dto.GetTaskLogsResponse{ + TaskID: req.TaskID, + Logs: logs, + Total: len(logs), + }, nil +} + +// RespondToPermission responds to a permission request +func (c *Controller) RespondToPermission(ctx context.Context, req dto.PermissionRespondRequest) (dto.PermissionRespondResponse, error) { + if err := c.service.RespondToPermission(ctx, req.TaskID, req.PendingID, req.OptionID, req.Cancelled); err != nil { + return dto.PermissionRespondResponse{}, err + } + + return dto.PermissionRespondResponse{ + Success: true, + TaskID: req.TaskID, + PendingID: req.PendingID, + }, nil +} + diff --git a/apps/backend/internal/orchestrator/controller/errors.go b/apps/backend/internal/orchestrator/controller/errors.go new file mode 100644 index 0000000000..c6a0c31837 --- /dev/null +++ b/apps/backend/internal/orchestrator/controller/errors.go @@ -0,0 +1,9 @@ +package controller + +import "errors" + +var ( + // ErrACPHandlerNotConfigured is returned when ACP handler is not set + ErrACPHandlerNotConfigured = errors.New("ACP handler not configured") +) + diff --git a/apps/backend/internal/orchestrator/dto/dto.go b/apps/backend/internal/orchestrator/dto/dto.go new file mode 100644 index 0000000000..8b6de8a356 --- /dev/null +++ b/apps/backend/internal/orchestrator/dto/dto.go @@ -0,0 +1,139 @@ +// Package dto provides Data Transfer Objects for the orchestrator module. +package dto + +import "time" + +// GetStatusRequest is the request for orchestrator.status +type GetStatusRequest struct{} + +// StatusResponse is the response for orchestrator.status +type StatusResponse struct { + Running bool `json:"running"` + ActiveAgents int `json:"active_agents"` + QueuedTasks int `json:"queued_tasks"` + MaxConcurrent int `json:"max_concurrent"` +} + +// GetQueueRequest is the request for orchestrator.queue +type GetQueueRequest struct{} + +// QueuedTaskDTO represents a task in the queue +type QueuedTaskDTO struct { + TaskID string `json:"task_id"` + Priority int `json:"priority"` + QueuedAt string `json:"queued_at"` +} + +// QueueResponse is the response for orchestrator.queue +type QueueResponse struct { + Tasks []QueuedTaskDTO `json:"tasks"` + Total int `json:"total"` +} + +// TriggerTaskRequest is the payload for orchestrator.trigger +type TriggerTaskRequest struct { + TaskID string `json:"task_id"` +} + +// TriggerTaskResponse is the response for orchestrator.trigger +type TriggerTaskResponse struct { + Success bool `json:"success"` + Message string `json:"message"` + TaskID string `json:"task_id"` +} + +// StartTaskRequest is the payload for orchestrator.start +type StartTaskRequest struct { + TaskID string `json:"task_id"` + AgentType string `json:"agent_type,omitempty"` + Priority int `json:"priority,omitempty"` +} + +// StartTaskResponse is the response for orchestrator.start +type StartTaskResponse struct { + Success bool `json:"success"` + TaskID string `json:"task_id"` + AgentInstanceID string `json:"agent_instance_id"` + Status string `json:"status"` +} + +// StopTaskRequest is the payload for orchestrator.stop +type StopTaskRequest struct { + TaskID string `json:"task_id"` + Reason string `json:"reason,omitempty"` + Force bool `json:"force,omitempty"` +} + +// SuccessResponse is a generic success response +type SuccessResponse struct { + Success bool `json:"success"` +} + +// PromptTaskRequest is the payload for orchestrator.prompt +type PromptTaskRequest struct { + TaskID string `json:"task_id"` + Prompt string `json:"prompt"` +} + +// PromptTaskResponse is the response for orchestrator.prompt +type PromptTaskResponse struct { + Success bool `json:"success"` + StopReason string `json:"stop_reason"` +} + +// CompleteTaskRequest is the payload for orchestrator.complete +type CompleteTaskRequest struct { + TaskID string `json:"task_id"` +} + +// CompleteTaskResponse is the response for orchestrator.complete +type CompleteTaskResponse struct { + Success bool `json:"success"` + Message string `json:"message"` +} + +// GetTaskLogsRequest is the payload for task.logs +type GetTaskLogsRequest struct { + TaskID string `json:"task_id"` + Limit int `json:"limit,omitempty"` +} + +// LogEntryDTO represents a single log entry +type LogEntryDTO struct { + Type string `json:"type"` + Timestamp string `json:"timestamp"` + AgentID string `json:"agent_id,omitempty"` + Data map[string]interface{} `json:"data,omitempty"` +} + +// GetTaskLogsResponse is the response for task.logs +type GetTaskLogsResponse struct { + TaskID string `json:"task_id"` + Logs []LogEntryDTO `json:"logs"` + Total int `json:"total"` +} + +// PermissionRespondRequest is the payload for permission.respond +type PermissionRespondRequest struct { + TaskID string `json:"task_id"` + PendingID string `json:"pending_id"` + OptionID string `json:"option_id,omitempty"` + Cancelled bool `json:"cancelled,omitempty"` +} + +// PermissionRespondResponse is the response for permission.respond +type PermissionRespondResponse struct { + Success bool `json:"success"` + TaskID string `json:"task_id"` + 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"), + } +} + diff --git a/apps/backend/internal/orchestrator/handlers/handlers.go b/apps/backend/internal/orchestrator/handlers/handlers.go new file mode 100644 index 0000000000..858bbc2f69 --- /dev/null +++ b/apps/backend/internal/orchestrator/handlers/handlers.go @@ -0,0 +1,250 @@ +// Package handlers provides WebSocket message handlers for the orchestrator. +package handlers + +import ( + "context" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/orchestrator/controller" + "github.com/kandev/kandev/internal/orchestrator/dto" + ws "github.com/kandev/kandev/pkg/websocket" + "go.uber.org/zap" +) + +// Handlers contains WebSocket handlers for the orchestrator API +type Handlers struct { + controller *controller.Controller + logger *logger.Logger +} + +// NewHandlers creates a new WebSocket handlers instance +func NewHandlers(ctrl *controller.Controller, log *logger.Logger) *Handlers { + return &Handlers{ + controller: ctrl, + logger: log.WithFields(zap.String("component", "orchestrator-handlers")), + } +} + +// RegisterHandlers registers all orchestrator handlers with the dispatcher +func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { + d.RegisterFunc(ws.ActionOrchestratorStatus, h.wsGetStatus) + d.RegisterFunc(ws.ActionOrchestratorQueue, h.wsGetQueue) + d.RegisterFunc(ws.ActionOrchestratorTrigger, h.wsTriggerTask) + d.RegisterFunc(ws.ActionOrchestratorStart, h.wsStartTask) + d.RegisterFunc(ws.ActionOrchestratorStop, h.wsStopTask) + d.RegisterFunc(ws.ActionOrchestratorPrompt, h.wsPromptTask) + d.RegisterFunc(ws.ActionOrchestratorComplete, h.wsCompleteTask) + d.RegisterFunc(ws.ActionTaskLogs, h.wsGetTaskLogs) + d.RegisterFunc(ws.ActionPermissionRespond, h.wsRespondToPermission) +} + +// WS handlers + +func (h *Handlers) wsGetStatus(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + resp, err := h.controller.GetStatus(ctx, dto.GetStatusRequest{}) + if err != nil { + h.logger.Error("failed to get status", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to get status", nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +func (h *Handlers) wsGetQueue(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + resp, err := h.controller.GetQueue(ctx, dto.GetQueueRequest{}) + if err != nil { + h.logger.Error("failed to get queue", zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to get queue", nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +type wsTriggerTaskRequest struct { + TaskID string `json:"task_id"` +} + +func (h *Handlers) wsTriggerTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsTriggerTaskRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + + resp, err := h.controller.TriggerTask(ctx, dto.TriggerTaskRequest{TaskID: req.TaskID}) + if err != nil { + h.logger.Error("failed to trigger task", zap.String("task_id", req.TaskID), zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to trigger task: "+err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +type wsStartTaskRequest struct { + TaskID string `json:"task_id"` + AgentType string `json:"agent_type,omitempty"` + Priority int `json:"priority,omitempty"` +} + +func (h *Handlers) wsStartTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsStartTaskRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + + resp, err := h.controller.StartTask(ctx, dto.StartTaskRequest{ + TaskID: req.TaskID, + AgentType: req.AgentType, + Priority: req.Priority, + }) + if err != nil { + h.logger.Error("failed to start task", zap.String("task_id", req.TaskID), zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to start task: "+err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +type wsStopTaskRequest struct { + TaskID string `json:"task_id"` + Reason string `json:"reason,omitempty"` + Force bool `json:"force,omitempty"` +} + +func (h *Handlers) wsStopTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsStopTaskRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + + resp, err := h.controller.StopTask(ctx, dto.StopTaskRequest{ + TaskID: req.TaskID, + Reason: req.Reason, + Force: req.Force, + }) + if err != nil { + h.logger.Error("failed to stop task", zap.String("task_id", req.TaskID), zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to stop task: "+err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +type wsPromptTaskRequest struct { + TaskID string `json:"task_id"` + Prompt string `json:"prompt"` +} + +func (h *Handlers) wsPromptTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsPromptTaskRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + if req.Prompt == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "prompt is required", nil) + } + + resp, err := h.controller.PromptTask(ctx, dto.PromptTaskRequest{ + TaskID: req.TaskID, + Prompt: req.Prompt, + }) + if err != nil { + h.logger.Error("failed to send prompt", zap.String("task_id", req.TaskID), zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to send prompt: "+err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +type wsCompleteTaskRequest struct { + TaskID string `json:"task_id"` +} + +func (h *Handlers) wsCompleteTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsCompleteTaskRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + + resp, err := h.controller.CompleteTask(ctx, dto.CompleteTaskRequest{TaskID: req.TaskID}) + if err != nil { + h.logger.Error("failed to complete task", zap.String("task_id", req.TaskID), zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to complete task: "+err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +type wsGetTaskLogsRequest struct { + TaskID string `json:"task_id"` + Limit int `json:"limit,omitempty"` +} + +func (h *Handlers) wsGetTaskLogs(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsGetTaskLogsRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + + resp, err := h.controller.GetTaskLogs(ctx, dto.GetTaskLogsRequest{ + TaskID: req.TaskID, + Limit: req.Limit, + }) + if err != nil { + h.logger.Error("failed to get task logs", zap.String("task_id", req.TaskID), zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to retrieve logs: "+err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + +type wsPermissionRespondRequest struct { + TaskID string `json:"task_id"` + PendingID string `json:"pending_id"` + OptionID string `json:"option_id,omitempty"` + Cancelled bool `json:"cancelled,omitempty"` +} + +func (h *Handlers) wsRespondToPermission(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req wsPermissionRespondRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + } + if req.TaskID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) + } + if req.PendingID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "pending_id is required", nil) + } + if !req.Cancelled && req.OptionID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "option_id is required when not cancelled", nil) + } + + h.logger.Info("responding to permission request", + zap.String("task_id", req.TaskID), + zap.String("pending_id", req.PendingID), + zap.String("option_id", req.OptionID), + zap.Bool("cancelled", req.Cancelled)) + + resp, err := h.controller.RespondToPermission(ctx, dto.PermissionRespondRequest{ + TaskID: req.TaskID, + PendingID: req.PendingID, + OptionID: req.OptionID, + Cancelled: req.Cancelled, + }) + if err != nil { + h.logger.Error("failed to respond to permission", zap.String("task_id", req.TaskID), zap.Error(err)) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to respond to permission: "+err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, resp) +} + diff --git a/apps/backend/internal/orchestrator/wshandlers/handlers.go b/apps/backend/internal/orchestrator/wshandlers/handlers.go deleted file mode 100644 index 4c7b22cfcc..0000000000 --- a/apps/backend/internal/orchestrator/wshandlers/handlers.go +++ /dev/null @@ -1,339 +0,0 @@ -// Package wshandlers provides WebSocket message handlers for the orchestrator. -package wshandlers - -import ( - "context" - - "github.com/kandev/kandev/internal/common/logger" - "github.com/kandev/kandev/internal/orchestrator" - "github.com/kandev/kandev/internal/orchestrator/acp" - ws "github.com/kandev/kandev/pkg/websocket" - "go.uber.org/zap" -) - -// Handlers contains WebSocket handlers for the orchestrator API -type Handlers struct { - service *orchestrator.Service - acpHandler *acp.Handler - logger *logger.Logger -} - -// NewHandlers creates a new WebSocket handlers instance -func NewHandlers(svc *orchestrator.Service, log *logger.Logger) *Handlers { - return &Handlers{ - service: svc, - logger: log.WithFields(zap.String("component", "orchestrator-ws-handlers")), - } -} - -// SetACPHandler sets the ACP handler for log retrieval -func (h *Handlers) SetACPHandler(handler *acp.Handler) { - h.acpHandler = handler -} - -// RegisterHandlers registers all orchestrator handlers with the dispatcher -func (h *Handlers) RegisterHandlers(d *ws.Dispatcher) { - d.RegisterFunc(ws.ActionOrchestratorStatus, h.GetStatus) - d.RegisterFunc(ws.ActionOrchestratorQueue, h.GetQueue) - d.RegisterFunc(ws.ActionOrchestratorTrigger, h.TriggerTask) - d.RegisterFunc(ws.ActionOrchestratorStart, h.StartTask) - d.RegisterFunc(ws.ActionOrchestratorStop, h.StopTask) - d.RegisterFunc(ws.ActionOrchestratorPrompt, h.PromptTask) - d.RegisterFunc(ws.ActionOrchestratorComplete, h.CompleteTask) - d.RegisterFunc(ws.ActionTaskLogs, h.GetTaskLogs) - d.RegisterFunc(ws.ActionPermissionRespond, h.RespondToPermission) -} - -// GetStatus handles orchestrator.status action -func (h *Handlers) GetStatus(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - status := h.service.GetStatus() - return ws.NewResponse(msg.ID, msg.Action, status) -} - -// QueuedTaskResponse represents a task in the queue -type QueuedTaskResponse struct { - TaskID string `json:"task_id"` - Priority int `json:"priority"` - QueuedAt string `json:"queued_at"` -} - -// QueueResponse is the response for orchestrator.queue -type QueueResponse struct { - Tasks []QueuedTaskResponse `json:"tasks"` - Total int `json:"total"` -} - -// GetQueue handles orchestrator.queue action -func (h *Handlers) GetQueue(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - queuedTasks := h.service.GetQueuedTasks() - - tasks := make([]QueuedTaskResponse, 0, len(queuedTasks)) - for _, qt := range queuedTasks { - tasks = append(tasks, QueuedTaskResponse{ - TaskID: qt.TaskID, - Priority: qt.Priority, - QueuedAt: qt.QueuedAt.Format("2006-01-02T15:04:05Z"), - }) - } - - return ws.NewResponse(msg.ID, msg.Action, QueueResponse{ - Tasks: tasks, - Total: len(tasks), - }) -} - -// TriggerTaskRequest is the payload for orchestrator.trigger -type TriggerTaskRequest struct { - TaskID string `json:"task_id"` -} - -// TriggerTask handles orchestrator.trigger action -func (h *Handlers) TriggerTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - var req TriggerTaskRequest - if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) - } - - if req.TaskID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) - } - - // For now just log - the service would need task details - h.logger.Info("Triggering task", zap.String("task_id", req.TaskID)) - - return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ - "success": true, - "message": "Task triggered", - "task_id": req.TaskID, - }) -} - -// StartTaskRequest is the payload for orchestrator.start -type StartTaskRequest struct { - TaskID string `json:"task_id"` - AgentType string `json:"agent_type,omitempty"` - Priority int `json:"priority,omitempty"` -} - -// StartTask handles orchestrator.start action -func (h *Handlers) StartTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - var req StartTaskRequest - if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) - } - - if req.TaskID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) - } - - execution, err := h.service.StartTask(ctx, req.TaskID, req.AgentType, req.Priority) - if err != nil { - h.logger.Error("failed to start task", zap.String("task_id", req.TaskID), zap.Error(err)) - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to start task: "+err.Error(), nil) - } - - return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ - "success": true, - "task_id": execution.TaskID, - "agent_instance_id": execution.AgentInstanceID, - "status": execution.Status, - }) -} - -// TaskIDRequest is a common request with just task_id -type TaskIDRequest struct { - TaskID string `json:"task_id"` -} - -// StopTaskRequest is the payload for orchestrator.stop -type StopTaskRequest struct { - TaskID string `json:"task_id"` - Reason string `json:"reason,omitempty"` - Force bool `json:"force,omitempty"` -} - -// StopTask handles orchestrator.stop action -func (h *Handlers) StopTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - var req StopTaskRequest - if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) - } - - if req.TaskID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) - } - - reason := req.Reason - if reason == "" { - reason = "stopped via WebSocket API" - } - - if err := h.service.StopTask(ctx, req.TaskID, reason, req.Force); err != nil { - h.logger.Error("failed to stop task", zap.String("task_id", req.TaskID), zap.Error(err)) - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to stop task: "+err.Error(), nil) - } - - return ws.NewResponse(msg.ID, msg.Action, map[string]bool{"success": true}) -} - -// PromptTaskRequest is the payload for orchestrator.prompt -type PromptTaskRequest struct { - TaskID string `json:"task_id"` - Prompt string `json:"prompt"` -} - -// PromptTask handles orchestrator.prompt action -func (h *Handlers) PromptTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - var req PromptTaskRequest - if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) - } - - if req.TaskID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) - } - if req.Prompt == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "prompt is required", nil) - } - - result, err := h.service.PromptTask(ctx, req.TaskID, req.Prompt) - if err != nil { - h.logger.Error("failed to send prompt", zap.String("task_id", req.TaskID), zap.Error(err)) - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to send prompt: "+err.Error(), nil) - } - - return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ - "success": true, - "stop_reason": result.StopReason, - }) -} - -// CompleteTask handles orchestrator.complete action -func (h *Handlers) CompleteTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - var req TaskIDRequest - if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) - } - - if req.TaskID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) - } - - if err := h.service.CompleteTask(ctx, req.TaskID); err != nil { - h.logger.Error("failed to complete task", zap.String("task_id", req.TaskID), zap.Error(err)) - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to complete task: "+err.Error(), nil) - } - - return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ - "success": true, - "message": "task completed", - }) -} - -// GetTaskLogsRequest is the payload for task.logs -type GetTaskLogsRequest struct { - TaskID string `json:"task_id"` - Limit int `json:"limit,omitempty"` -} - -// LogEntry represents a single log entry in the response -type LogEntry struct { - Type string `json:"type"` - Timestamp string `json:"timestamp"` - AgentID string `json:"agent_id,omitempty"` - Data map[string]interface{} `json:"data,omitempty"` -} - -// GetTaskLogsResponse is the response for task.logs -type GetTaskLogsResponse struct { - TaskID string `json:"task_id"` - Logs []LogEntry `json:"logs"` - Total int `json:"total"` -} - -// GetTaskLogs handles task.logs action - retrieves historical execution logs for a task -func (h *Handlers) GetTaskLogs(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - var req GetTaskLogsRequest - if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) - } - - if req.TaskID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) - } - - if h.acpHandler == nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "ACP handler not configured", nil) - } - - // Get all historical messages from the database - messages, err := h.acpHandler.GetAllMessages(ctx, req.TaskID) - if err != nil { - h.logger.Error("failed to get task logs", zap.String("task_id", req.TaskID), zap.Error(err)) - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to retrieve logs: "+err.Error(), nil) - } - - // Apply limit if specified - if req.Limit > 0 && len(messages) > req.Limit { - messages = messages[len(messages)-req.Limit:] - } - - logs := make([]LogEntry, 0, len(messages)) - for _, m := range messages { - logs = append(logs, LogEntry{ - Type: string(m.Type), - Timestamp: m.Timestamp.Format("2006-01-02T15:04:05Z"), - AgentID: m.AgentID, - Data: m.Data, - }) - } - - return ws.NewResponse(msg.ID, msg.Action, GetTaskLogsResponse{ - TaskID: req.TaskID, - Logs: logs, - Total: len(logs), - }) -} - -// PermissionRespondRequest is the payload for permission.respond -type PermissionRespondRequest struct { - TaskID string `json:"task_id"` - PendingID string `json:"pending_id"` - OptionID string `json:"option_id,omitempty"` - Cancelled bool `json:"cancelled,omitempty"` -} - -// RespondToPermission handles permission.respond action -func (h *Handlers) RespondToPermission(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - var req PermissionRespondRequest - if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) - } - - if req.TaskID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task_id is required", nil) - } - if req.PendingID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "pending_id is required", nil) - } - if !req.Cancelled && req.OptionID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "option_id is required when not cancelled", nil) - } - - h.logger.Info("responding to permission request", - zap.String("task_id", req.TaskID), - zap.String("pending_id", req.PendingID), - zap.String("option_id", req.OptionID), - zap.Bool("cancelled", req.Cancelled)) - - if err := h.service.RespondToPermission(ctx, req.TaskID, req.PendingID, req.OptionID, req.Cancelled); err != nil { - h.logger.Error("failed to respond to permission", zap.String("task_id", req.TaskID), zap.Error(err)) - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to respond to permission: "+err.Error(), nil) - } - - return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ - "success": true, - "task_id": req.TaskID, - "pending_id": req.PendingID, - }) -} diff --git a/apps/backend/internal/task/handlers/repository_handlers.go b/apps/backend/internal/task/handlers/repository_handlers.go index b72dc509a2..dd9f32d4fd 100644 --- a/apps/backend/internal/task/handlers/repository_handlers.go +++ b/apps/backend/internal/task/handlers/repository_handlers.go @@ -51,16 +51,16 @@ func (h *RepositoryHandlers) registerHTTP(router *gin.Engine) { } func (h *RepositoryHandlers) registerWS(dispatcher *ws.Dispatcher) { - dispatcher.RegisterFunc("repository.list", h.wsListRepositories) - dispatcher.RegisterFunc("repository.create", h.wsCreateRepository) - dispatcher.RegisterFunc("repository.get", h.wsGetRepository) - dispatcher.RegisterFunc("repository.update", h.wsUpdateRepository) - dispatcher.RegisterFunc("repository.delete", h.wsDeleteRepository) - dispatcher.RegisterFunc("repository.script.list", h.wsListRepositoryScripts) - dispatcher.RegisterFunc("repository.script.create", h.wsCreateRepositoryScript) - dispatcher.RegisterFunc("repository.script.get", h.wsGetRepositoryScript) - dispatcher.RegisterFunc("repository.script.update", h.wsUpdateRepositoryScript) - dispatcher.RegisterFunc("repository.script.delete", h.wsDeleteRepositoryScript) + dispatcher.RegisterFunc(ws.ActionRepositoryList, h.wsListRepositories) + dispatcher.RegisterFunc(ws.ActionRepositoryCreate, h.wsCreateRepository) + dispatcher.RegisterFunc(ws.ActionRepositoryGet, h.wsGetRepository) + dispatcher.RegisterFunc(ws.ActionRepositoryUpdate, h.wsUpdateRepository) + dispatcher.RegisterFunc(ws.ActionRepositoryDelete, h.wsDeleteRepository) + dispatcher.RegisterFunc(ws.ActionRepositoryScriptList, h.wsListRepositoryScripts) + dispatcher.RegisterFunc(ws.ActionRepositoryScriptCreate, h.wsCreateRepositoryScript) + dispatcher.RegisterFunc(ws.ActionRepositoryScriptGet, h.wsGetRepositoryScript) + dispatcher.RegisterFunc(ws.ActionRepositoryScriptUpdate, h.wsUpdateRepositoryScript) + dispatcher.RegisterFunc(ws.ActionRepositoryScriptDelete, h.wsDeleteRepositoryScript) } // HTTP handlers diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index 37990a87de..b6f22b61f7 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -26,6 +26,20 @@ const ( ActionColumnUpdate = "column.update" ActionColumnDelete = "column.delete" + // Repository actions + ActionRepositoryList = "repository.list" + ActionRepositoryCreate = "repository.create" + ActionRepositoryGet = "repository.get" + ActionRepositoryUpdate = "repository.update" + ActionRepositoryDelete = "repository.delete" + + // Repository Script actions + ActionRepositoryScriptList = "repository.script.list" + ActionRepositoryScriptCreate = "repository.script.create" + ActionRepositoryScriptGet = "repository.script.get" + ActionRepositoryScriptUpdate = "repository.script.update" + ActionRepositoryScriptDelete = "repository.script.delete" + // Task actions ActionTaskList = "task.list" ActionTaskCreate = "task.create"