refactor: unify WebSocket patterns to Pattern A (handlers + controller + dto)
- Add missing repository action constants to pkg/websocket/actions.go - Update repository_handlers.go to use action constants instead of string literals - Refactor orchestrator module: create dto/, controller/, handlers/ packages - Refactor agent module: create dto/, controller/, handlers/ packages - Update main.go and integration tests to use new Pattern A wiring - Remove old wshandlers/ directories from both modules All modules now follow the same architecture: handlers/ → Parse WS messages, validate, call controller controller/ → Business logic coordination, convert to/from DTOs dto/ → Centralized request/response Data Transfer Objects service/ → Core business logic (unchanged)
This commit is contained in:
jcfs
@@ -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")
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
)
|
||||
|
||||
@@ -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"),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
})
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user