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

chore: Clean up unused code and placeholder directories

Browse Source 
Remove obsolete REST API code and empty placeholder directories:

Removed REST API code:
- apps/backend/internal/agent/api/ (REST handlers)
- apps/backend/internal/orchestrator/api/ (REST handlers)
- apps/backend/internal/orchestrator/streaming/ (REST streaming)
- apps/backend/internal/task/api/ (REST handlers)

Removed obsolete cmd binaries:
- apps/backend/cmd/agent-manager/
- apps/backend/cmd/orchestrator/
- apps/backend/cmd/api-gateway/
- apps/backend/cmd/task-service/
- apps/backend/cmd/agentctl/

Removed empty placeholder directories:
- apps/backend/internal/agent/models/
- apps/backend/internal/agent/monitor/
- apps/backend/internal/common/acp/
- apps/backend/internal/common/auth/
- apps/backend/internal/events/publisher/
- apps/backend/internal/events/subscriber/
- apps/backend/internal/gateway/handlers/
- apps/backend/internal/gateway/middleware/
- apps/backend/internal/gateway/router/
- apps/backend/internal/gateway/streaming/
- apps/backend/pkg/acp/client/
- apps/backend/pkg/acp/server/
- deployments/docker/agents/
- deployments/docker/compose/
- migrations/postgres/
- apps/agentctl/

Fixed:
- Added missing PromptAgent method to mock in scheduler_test.go

All tests pass, build succeeds.
This commit is contained in:
jcfs
2026-01-10 23:33:25 +00:00
parent ec0b97ebfb
commit 4d0efdc3ed
45 changed files with 17 additions and 3895 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/agentctl/go.mod
-41
View File
@@ -1,41 +0,0 @@
module github.com/kandev/agentctl
go 1.21
require (
github.com/gin-gonic/gin v1.10.0
github.com/google/uuid v1.6.0
github.com/gorilla/websocket v1.5.3
go.uber.org/zap v1.27.0
)
require (
github.com/bytedance/sonic v1.11.6 // indirect
github.com/bytedance/sonic/loader v0.1.1 // indirect
github.com/cloudwego/base64x v0.1.4 // indirect
github.com/cloudwego/iasm v0.2.0 // indirect
github.com/gabriel-vasile/mimetype v1.4.3 // indirect
github.com/gin-contrib/sse v0.1.0 // indirect
github.com/go-playground/locales v0.14.1 // indirect
github.com/go-playground/universal-translator v0.18.1 // indirect
github.com/go-playground/validator/v10 v10.20.0 // indirect
github.com/goccy/go-json v0.10.2 // indirect
github.com/json-iterator/go v1.1.12 // indirect
github.com/klauspost/cpuid/v2 v2.2.7 // indirect
github.com/leodido/go-urn v1.4.0 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
github.com/modern-go/reflect2 v1.0.2 // indirect
github.com/pelletier/go-toml/v2 v2.2.2 // indirect
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
github.com/ugorji/go/codec v1.2.12 // indirect
go.uber.org/multierr v1.10.0 // indirect
golang.org/x/arch v0.8.0 // indirect
golang.org/x/crypto v0.23.0 // indirect
golang.org/x/net v0.25.0 // indirect
golang.org/x/sys v0.20.0 // indirect
golang.org/x/text v0.15.0 // indirect
google.golang.org/protobuf v1.34.1 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)
apps/backend/NEXT_STEPS.md
+13 -34
View File
@@ -52,51 +52,30 @@ Track all agent executions:
## API Reference
### Task Management
```
POST /api/v1/boards - Create board
GET /api/v1/boards - List boards
POST /api/v1/boards/:id/columns - Create column
POST /api/v1/tasks - Create task (requires board_id, column_id, title)
GET /api/v1/tasks/:id - Get task (includes metadata.auggie_session_id)
```
All communication happens over WebSocket at `ws://localhost:8080/ws`.
See `docs/WEBSOCKET_API.md` for complete API documentation.
### Agent Management
```
POST /api/v1/agents/launch - Launch agent
GET /api/v1/agents/:id/status - Get agent status
GET /api/v1/agents/:id/logs - Get agent logs
DELETE /api/v1/agents/:id - Stop agent
```
### Key Actions
### Launch Agent Request
```json
{
"task_id": "uuid",
"agent_type": "augment-agent",
"workspace_path": "/path/to/project",
"env": {
"AUGMENT_SESSION_AUTH": "{...session json...}",
"TASK_DESCRIPTION": "Your task here",
"AUGGIE_SESSION_ID": "optional-for-resumption"
}
}
```
| Action | Description |
|--------|-------------|
| `board.create` | Create a new board |
| `task.create` | Create a task |
| `task.subscribe` | Subscribe to ACP streaming for a task |
| `orchestrator.start` | Start executing a task with an agent |
| `agent.prompt` | Send follow-up prompt to running agent |
## Development Commands
```bash
# Build
cd backend && make build
cd apps/backend && go build -o kandev ./cmd/kandev
# Run server
./bin/kandev
./kandev
# Build Docker image
cd dockerfiles/augment-agent && docker build -t kandev/augment-agent:latest .
# Get Augment session auth
cat ~/.augment/session.json
cd apps/backend/dockerfiles/augment-agent && docker build -t kandev/augment-agent:latest .
```
## Key Files
apps/backend/agentctl
BIN
View File
Binary file not shown.
apps/backend/bin/kandev
BIN
View File
Binary file not shown.
apps/backend/cmd/agent-manager/.gitkeep
View File
apps/backend/cmd/agent-manager/main.go
-166
View File
@@ -1,166 +0,0 @@
package main
import (
"context"
"fmt"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"github.com/gin-gonic/gin"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/agent/api"
"github.com/kandev/kandev/internal/agent/credentials"
"github.com/kandev/kandev/internal/agent/docker"
"github.com/kandev/kandev/internal/agent/lifecycle"
"github.com/kandev/kandev/internal/agent/registry"
"github.com/kandev/kandev/internal/agent/streaming"
"github.com/kandev/kandev/internal/common/config"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/events/bus"
)
func main() {
// 1. Load configuration
cfg, err := config.Load()
if err != nil {
fmt.Fprintf(os.Stderr, "Failed to load configuration: %v\n", err)
os.Exit(1)
}
// 2. Initialize logger
log, err := logger.NewLogger(logger.LoggingConfig{
Level: cfg.Logging.Level,
Format: cfg.Logging.Format,
})
if err != nil {
fmt.Fprintf(os.Stderr, "Failed to initialize logger: %v\n", err)
os.Exit(1)
}
defer log.Sync()
logger.SetDefault(log)
log.Info("Starting Agent Manager service...")
// 3. Create context with cancellation
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// 4. Connect to NATS event bus
eventBus, err := bus.NewNATSEventBus(cfg.NATS, log)
if err != nil {
log.Fatal("Failed to connect to NATS", zap.Error(err))
}
defer eventBus.Close()
log.Info("Connected to NATS event bus")
// 5. Initialize Docker client
dockerClient, err := docker.NewClient(cfg.Docker, log)
if err != nil {
log.Fatal("Failed to initialize Docker client", zap.Error(err))
}
defer dockerClient.Close()
// Ping Docker to verify connection
if err := dockerClient.Ping(ctx); err != nil {
log.Fatal("Failed to connect to Docker daemon", zap.Error(err))
}
log.Info("Connected to Docker daemon")
// 6. Initialize Agent Registry
reg := registry.NewRegistry(log)
reg.LoadDefaults()
log.Info("Loaded agent registry", zap.Int("agent_types", len(reg.List())))
// 7. Initialize Credentials Manager
credsMgr := credentials.NewManager(log)
credsMgr.AddProvider(credentials.NewEnvProvider("KANDEV_"))
// Optionally load from file
if credsFile := os.Getenv("KANDEV_CREDENTIALS_FILE"); credsFile != "" {
credsMgr.AddProvider(credentials.NewFileProvider(credsFile))
}
log.Info("Initialized credentials manager")
// 8. Initialize Streaming Manager
streamMgr := streaming.NewManager(eventBus, log)
log.Info("Initialized streaming manager")
// 9. Initialize Lifecycle Manager
lifecycleMgr := lifecycle.NewManager(dockerClient, reg, eventBus, log)
if err := lifecycleMgr.Start(ctx); err != nil {
log.Fatal("Failed to start lifecycle manager", zap.Error(err))
}
log.Info("Started lifecycle manager")
// 10. Setup HTTP server with Gin
if cfg.Logging.Level != "debug" {
gin.SetMode(gin.ReleaseMode)
}
router := gin.New()
router.Use(gin.Recovery())
// 11. Register API routes
v1 := router.Group("/api/v1/agents")
api.SetupRoutes(v1, lifecycleMgr, reg, dockerClient, nil, log)
// Health check endpoint at root level
handler := api.NewHandler(lifecycleMgr, reg, dockerClient, nil, log)
router.GET("/health", handler.HealthCheck)
// 12. Create HTTP server
port := cfg.Server.Port
if port == 0 {
port = 8083 // Default agent manager port
}
server := &http.Server{
Addr: fmt.Sprintf(":%d", port),
Handler: router,
ReadTimeout: cfg.Server.ReadTimeoutDuration(),
WriteTimeout: cfg.Server.WriteTimeoutDuration(),
}
// 13. Start server in goroutine
go func() {
log.Info("HTTP server listening", zap.Int("port", port))
if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Fatal("Failed to start HTTP server", zap.Error(err))
}
}()
// 14. Wait for shutdown signal
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
log.Info("Shutting down Agent Manager service...")
// 15. Graceful shutdown
cancel()
// Shutdown HTTP server
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second)
defer shutdownCancel()
if err := server.Shutdown(shutdownCtx); err != nil {
log.Error("HTTP server shutdown error", zap.Error(err))
}
// Stop streaming manager
streamMgr.StopAll()
// Stop lifecycle manager
if err := lifecycleMgr.Stop(); err != nil {
log.Error("Lifecycle manager stop error", zap.Error(err))
}
// Log that credentials manager is available (for future use)
_ = credsMgr // Used for building env vars when launching agents
log.Info("Agent Manager service stopped")
}
apps/backend/cmd/agentctl/main.go
-103
View File
@@ -1,103 +0,0 @@
// Package main is the entry point for agentctl - a control process that runs
// inside Docker containers to manage AI agent processes.
//
// agentctl provides:
// - HTTP REST API for control operations (start, stop, pause, resume)
// - WebSocket streaming of agent output
// - Process management for the underlying agent (e.g., auggie --acp)
// - ACP message relay between HTTP API and agent subprocess
package main
import (
"context"
"fmt"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"github.com/kandev/kandev/internal/agentctl/api"
"github.com/kandev/kandev/internal/agentctl/config"
"github.com/kandev/kandev/internal/agentctl/process"
"github.com/kandev/kandev/internal/common/logger"
"go.uber.org/zap"
)
func main() {
// Load configuration
cfg := config.Load()
// Initialize logger
log, err := logger.NewLogger(logger.LoggingConfig{
Level: cfg.LogLevel,
Format: cfg.LogFormat,
OutputPath: "stdout",
})
if err != nil {
fmt.Fprintf(os.Stderr, "failed to initialize logger: %v\n", err)
os.Exit(1)
}
defer log.Sync()
log.Info("starting agentctl",
zap.String("version", "0.1.0"),
zap.Int("port", cfg.Port),
zap.String("agent_command", cfg.AgentCommand),
)
// Create process manager
procMgr := process.NewManager(cfg, log)
// Create API server
apiServer := api.NewServer(cfg, procMgr, log)
// Create HTTP server
server := &http.Server{
Addr: fmt.Sprintf(":%d", cfg.Port),
Handler: apiServer.Router(),
ReadTimeout: 30 * time.Second,
WriteTimeout: 30 * time.Second,
}
// Start server in goroutine
go func() {
log.Info("HTTP server listening", zap.String("addr", server.Addr))
if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Fatal("HTTP server error", zap.Error(err))
}
}()
// Auto-start agent if configured
if cfg.AutoStart {
log.Info("auto-starting agent process")
ctx := context.Background()
if err := procMgr.Start(ctx); err != nil {
log.Error("failed to auto-start agent", zap.Error(err))
}
}
// Wait for shutdown signal
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
log.Info("shutting down agentctl")
// Graceful shutdown
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// Stop the agent process
if err := procMgr.Stop(ctx); err != nil {
log.Error("failed to stop agent process", zap.Error(err))
}
// Shutdown HTTP server
if err := server.Shutdown(ctx); err != nil {
log.Error("HTTP server shutdown error", zap.Error(err))
}
log.Info("agentctl stopped")
}
apps/backend/cmd/api-gateway/.gitkeep
View File
apps/backend/cmd/orchestrator/.gitkeep
View File
apps/backend/cmd/orchestrator/main.go
-163
View File
@@ -1,163 +0,0 @@
// Package main is the entry point for the Orchestrator service.
package main
import (
"context"
"fmt"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"github.com/gin-gonic/gin"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/common/config"
"github.com/kandev/kandev/internal/common/database"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/events/bus"
"github.com/kandev/kandev/internal/orchestrator"
"github.com/kandev/kandev/internal/orchestrator/api"
"github.com/kandev/kandev/internal/orchestrator/executor"
"github.com/kandev/kandev/internal/orchestrator/scheduler"
"github.com/kandev/kandev/internal/orchestrator/streaming"
"github.com/kandev/kandev/pkg/acp/protocol"
)
func main() {
// 1. Load configuration
cfg, err := config.Load()
if err != nil {
fmt.Fprintf(os.Stderr, "Failed to load configuration: %v\n", err)
os.Exit(1)
}
// 2. Initialize logger
logCfg := logger.LoggingConfig{
Level: cfg.Logging.Level,
Format: cfg.Logging.Format,
OutputPath: cfg.Logging.OutputPath,
}
log, err := logger.NewLogger(logCfg)
if err != nil {
fmt.Fprintf(os.Stderr, "Failed to initialize logger: %v\n", err)
os.Exit(1)
}
defer log.Sync()
logger.SetDefault(log)
log.Info("Starting Orchestrator service...")
// 3. Create context with cancellation
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// 4. Connect to PostgreSQL
db, err := database.NewDB(ctx, cfg.Database)
if err != nil {
log.Fatal("Failed to connect to database", zap.Error(err))
}
defer db.Close()
log.Info("Connected to PostgreSQL")
// 5. Connect to NATS event bus
eventBus, err := bus.NewNATSEventBus(cfg.NATS, log)
if err != nil {
log.Fatal("Failed to connect to NATS", zap.Error(err))
}
defer eventBus.Close()
log.Info("Connected to NATS event bus")
// 6. Initialize components
// For now, use mock implementations
agentManager := executor.NewMockAgentManagerClient(log)
taskRepo := scheduler.NewMockTaskRepository(log)
// 7. Create orchestrator service
serviceCfg := orchestrator.DefaultServiceConfig()
service := orchestrator.NewService(serviceCfg, eventBus, db, agentManager, taskRepo, log)
// 8. Create WebSocket hub
wsHub := streaming.NewHub(log)
go wsHub.Run(ctx)
// 9. Wire ACP handler to broadcast to WebSocket clients
service.RegisterACPHandler(func(taskID string, msg *protocol.Message) {
wsHub.Broadcast(taskID, msg)
})
// 10. Start orchestrator service
if err := service.Start(ctx); err != nil {
log.Fatal("Failed to start orchestrator service", zap.Error(err))
}
log.Info("Orchestrator service started")
// 11. Setup HTTP server with Gin
if cfg.Logging.Level != "debug" {
gin.SetMode(gin.ReleaseMode)
}
router := gin.New()
router.Use(api.RequestLogger(log))
router.Use(api.Recovery(log))
router.Use(api.CORS())
// 12. Register API routes
v1 := router.Group("/api/v1/orchestrator")
api.SetupRoutes(v1, service, log)
// 13. Register WebSocket routes
wsHandler := streaming.NewWSHandler(wsHub, service, log)
streaming.SetupWebSocketRoutes(v1, wsHandler)
// 14. Health check endpoint
router.GET("/health", func(c *gin.Context) {
c.JSON(http.StatusOK, gin.H{"status": "ok"})
})
// 15. Create HTTP server
port := cfg.Server.Port
if port == 0 {
port = 8082 // Default orchestrator port
}
server := &http.Server{
Addr: fmt.Sprintf(":%d", port),
Handler: router,
ReadTimeout: cfg.Server.ReadTimeoutDuration(),
WriteTimeout: cfg.Server.WriteTimeoutDuration(),
}
// 16. Start server in goroutine
go func() {
log.Info("HTTP server listening", zap.Int("port", port))
if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Fatal("Failed to start HTTP server", zap.Error(err))
}
}()
// 17. Wait for shutdown signal
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
log.Info("Shutting down Orchestrator service...")
// 18. Graceful shutdown
cancel() // Cancel context to stop background goroutines
// Shutdown HTTP server with timeout
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second)
defer shutdownCancel()
if err := server.Shutdown(shutdownCtx); err != nil {
log.Error("HTTP server shutdown error", zap.Error(err))
}
// Stop orchestrator service
if err := service.Stop(); err != nil {
log.Error("Orchestrator service stop error", zap.Error(err))
}
log.Info("Orchestrator service stopped")
}
apps/backend/cmd/task-service/.gitkeep
View File
apps/backend/internal/agent/api/handlers.go
-425
View File
@@ -1,425 +0,0 @@
package api
import (
"bufio"
"context"
"io"
"net/http"
"strings"
"time"
"github.com/gin-gonic/gin"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/agent/lifecycle"
"github.com/kandev/kandev/internal/agent/registry"
"github.com/kandev/kandev/internal/common/errors"
"github.com/kandev/kandev/internal/common/logger"
)
// DockerClient interface for container operations needed by handlers
type DockerClient interface {
GetContainerLogs(ctx context.Context, containerID string, follow bool, tail string) (io.ReadCloser, error)
}
// ACPManager interface for sending commands to agents
type ACPManager interface {
Prompt(ctx context.Context, instanceID, message string) error
Cancel(ctx context.Context, instanceID, reason string) error
GetSessionID(instanceID string) (string, bool)
}
// Handler contains HTTP handlers for the agent manager API
type Handler struct {
lifecycle *lifecycle.Manager
registry *registry.Registry
docker DockerClient
acp ACPManager
logger *logger.Logger
}
// NewHandler creates a new API handler
func NewHandler(
lm *lifecycle.Manager,
reg *registry.Registry,
docker DockerClient,
acp ACPManager,
log *logger.Logger,
) *Handler {
return &Handler{
lifecycle: lm,
registry: reg,
docker: docker,
acp: acp,
logger: log.WithFields(zap.String("component", "agent-api")),
}
}
// LaunchAgent launches a new agent for a task
// POST /api/v1/agents/launch
func (h *Handler) LaunchAgent(c *gin.Context) {
var req LaunchAgentRequest
if err := c.ShouldBindJSON(&req); err != nil {
appErr := errors.BadRequest("invalid request body: " + err.Error())
c.JSON(appErr.HTTPStatus, appErr)
return
}
launchReq := &lifecycle.LaunchRequest{
TaskID: req.TaskID,
AgentType: req.AgentType,
WorkspacePath: req.WorkspacePath,
Env: req.Env,
Metadata: req.Metadata,
}
instance, err := h.lifecycle.Launch(c.Request.Context(), launchReq)
if err != nil {
h.logger.Error("failed to launch agent", zap.Error(err))
if strings.Contains(err.Error(), "not found") || strings.Contains(err.Error(), "disabled") {
appErr := errors.BadRequest(err.Error())
c.JSON(appErr.HTTPStatus, appErr)
return
}
if strings.Contains(err.Error(), "already has an agent running") {
appErr := errors.Conflict(err.Error())
c.JSON(appErr.HTTPStatus, appErr)
return
}
appErr := errors.InternalError("failed to launch agent", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
resp := instanceToResponse(instance)
c.JSON(http.StatusCreated, resp)
}
// StopAgent stops an agent instance
// DELETE /api/v1/agents/:instanceId
func (h *Handler) StopAgent(c *gin.Context) {
instanceID := c.Param("instanceId")
if instanceID == "" {
appErr := errors.BadRequest("instanceId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
var req StopAgentRequest
// Bind JSON body if present, but don't require it
_ = c.ShouldBindJSON(&req)
err := h.lifecycle.StopAgent(c.Request.Context(), instanceID, req.Force)
if err != nil {
h.logger.Error("failed to stop agent", zap.String("instance_id", instanceID), zap.Error(err))
if strings.Contains(err.Error(), "not found") {
appErr := errors.NotFound("agent instance", instanceID)
c.JSON(appErr.HTTPStatus, appErr)
return
}
appErr := errors.InternalError("failed to stop agent", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusOK, gin.H{"message": "agent stopped successfully"})
}
// GetAgentStatus returns agent instance status
// GET /api/v1/agents/:instanceId/status
func (h *Handler) GetAgentStatus(c *gin.Context) {
instanceID := c.Param("instanceId")
if instanceID == "" {
appErr := errors.BadRequest("instanceId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
instance, found := h.lifecycle.GetInstance(instanceID)
if !found {
appErr := errors.NotFound("agent instance", instanceID)
c.JSON(appErr.HTTPStatus, appErr)
return
}
resp := instanceToResponse(instance)
c.JSON(http.StatusOK, resp)
}
// GetAgentLogs returns agent logs
// GET /api/v1/agents/:instanceId/logs
func (h *Handler) GetAgentLogs(c *gin.Context) {
instanceID := c.Param("instanceId")
if instanceID == "" {
appErr := errors.BadRequest("instanceId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
instance, found := h.lifecycle.GetInstance(instanceID)
if !found {
appErr := errors.NotFound("agent instance", instanceID)
c.JSON(appErr.HTTPStatus, appErr)
return
}
// Get tail parameter, default to 100 lines
tail := c.DefaultQuery("tail", "100")
if h.docker == nil {
// Docker client not available, return empty logs
c.JSON(http.StatusOK, LogsResponse{
Logs: []LogEntry{},
Total: 0,
})
return
}
reader, err := h.docker.GetContainerLogs(c.Request.Context(), instance.ContainerID, false, tail)
if err != nil {
h.logger.Error("failed to get container logs",
zap.String("instance_id", instanceID),
zap.String("container_id", instance.ContainerID),
zap.Error(err))
appErr := errors.InternalError("failed to get agent logs", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
defer reader.Close()
// Parse the logs - Docker log format has 8-byte header
logs := []LogEntry{}
scanner := bufio.NewScanner(reader)
for scanner.Scan() {
line := scanner.Text()
// Skip Docker header bytes if present
if len(line) > 8 {
line = line[8:]
}
logs = append(logs, LogEntry{
Timestamp: time.Now(), // Container logs include timestamp
Message: line,
Stream: "stdout",
})
}
c.JSON(http.StatusOK, LogsResponse{
Logs: logs,
Total: len(logs),
})
}
// ListAgents returns all active agent instances
// GET /api/v1/agents
func (h *Handler) ListAgents(c *gin.Context) {
instances := h.lifecycle.ListInstances()
agents := make([]AgentInstanceResponse, 0, len(instances))
for _, instance := range instances {
agents = append(agents, instanceToResponse(instance))
}
c.JSON(http.StatusOK, AgentsListResponse{
Agents: agents,
Total: len(agents),
})
}
// ListAgentTypes returns available agent types
// GET /api/v1/agents/types
func (h *Handler) ListAgentTypes(c *gin.Context) {
configs := h.registry.List()
types := make([]AgentTypeResponse, 0, len(configs))
for _, cfg := range configs {
types = append(types, AgentTypeResponse{
ID: cfg.ID,
Name: cfg.Name,
Description: cfg.Description,
Image: cfg.Image,
Capabilities: cfg.Capabilities,
Enabled: cfg.Enabled,
})
}
c.JSON(http.StatusOK, AgentTypesListResponse{
Types: types,
Total: len(types),
})
}
// GetAgentType returns a specific agent type
// GET /api/v1/agents/types/:typeId
func (h *Handler) GetAgentType(c *gin.Context) {
typeID := c.Param("typeId")
if typeID == "" {
appErr := errors.BadRequest("typeId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
cfg, err := h.registry.Get(typeID)
if err != nil {
appErr := errors.NotFound("agent type", typeID)
c.JSON(appErr.HTTPStatus, appErr)
return
}
resp := AgentTypeResponse{
ID: cfg.ID,
Name: cfg.Name,
Description: cfg.Description,
Image: cfg.Image,
Capabilities: cfg.Capabilities,
Enabled: cfg.Enabled,
}
c.JSON(http.StatusOK, resp)
}
// HealthCheck returns health status
// GET /health
func (h *Handler) HealthCheck(c *gin.Context) {
c.JSON(http.StatusOK, HealthResponse{
Status: "healthy",
Timestamp: time.Now(),
})
}
// SendPrompt sends a prompt to an agent
// POST /api/v1/agents/:instanceId/prompt
func (h *Handler) SendPrompt(c *gin.Context) {
instanceID := c.Param("instanceId")
if instanceID == "" {
appErr := errors.BadRequest("instanceId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
// Check if agent exists
_, found := h.lifecycle.GetInstance(instanceID)
if !found {
appErr := errors.NotFound("agent instance", instanceID)
c.JSON(appErr.HTTPStatus, appErr)
return
}
var req SendPromptRequest
if err := c.ShouldBindJSON(&req); err != nil {
appErr := errors.BadRequest("invalid request body: " + err.Error())
c.JSON(appErr.HTTPStatus, appErr)
return
}
if h.acp == nil {
appErr := errors.InternalError("ACP manager not available", nil)
c.JSON(appErr.HTTPStatus, appErr)
return
}
if err := h.acp.Prompt(c.Request.Context(), instanceID, req.Message); err != nil {
h.logger.Error("failed to send prompt to agent",
zap.String("instance_id", instanceID),
zap.Error(err))
appErr := errors.InternalError("failed to send prompt", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
sessionID, _ := h.acp.GetSessionID(instanceID)
c.JSON(http.StatusOK, SendPromptResponse{
Success: true,
SessionID: sessionID,
Message: "prompt sent successfully",
})
}
// CancelAgent cancels an agent's current operation
// POST /api/v1/agents/:instanceId/cancel
func (h *Handler) CancelAgent(c *gin.Context) {
instanceID := c.Param("instanceId")
if instanceID == "" {
appErr := errors.BadRequest("instanceId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
// Check if agent exists
_, found := h.lifecycle.GetInstance(instanceID)
if !found {
appErr := errors.NotFound("agent instance", instanceID)
c.JSON(appErr.HTTPStatus, appErr)
return
}
var req CancelAgentRequest
_ = c.ShouldBindJSON(&req) // Optional body
if h.acp == nil {
appErr := errors.InternalError("ACP manager not available", nil)
c.JSON(appErr.HTTPStatus, appErr)
return
}
if err := h.acp.Cancel(c.Request.Context(), instanceID, req.Reason); err != nil {
h.logger.Error("failed to cancel agent operation",
zap.String("instance_id", instanceID),
zap.Error(err))
appErr := errors.InternalError("failed to cancel operation", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusOK, CancelAgentResponse{
Success: true,
Message: "cancel request sent",
})
}
// GetSession returns the ACP session info for an agent
// GET /api/v1/agents/:instanceId/session
func (h *Handler) GetSession(c *gin.Context) {
instanceID := c.Param("instanceId")
if instanceID == "" {
appErr := errors.BadRequest("instanceId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
// Check if agent exists
instance, found := h.lifecycle.GetInstance(instanceID)
if !found {
appErr := errors.NotFound("agent instance", instanceID)
c.JSON(appErr.HTTPStatus, appErr)
return
}
sessionID := ""
if h.acp != nil {
sessionID, _ = h.acp.GetSessionID(instanceID)
}
c.JSON(http.StatusOK, GetSessionResponse{
InstanceID: instanceID,
SessionID: sessionID,
Status: string(instance.Status),
})
}
// instanceToResponse converts an AgentInstance to an AgentInstanceResponse
func instanceToResponse(instance *lifecycle.AgentInstance) AgentInstanceResponse {
return AgentInstanceResponse{
ID: instance.ID,
TaskID: instance.TaskID,
AgentType: instance.AgentType,
ContainerID: instance.ContainerID,
Status: string(instance.Status),
Progress: instance.Progress,
StartedAt: instance.StartedAt,
FinishedAt: instance.FinishedAt,
ExitCode: instance.ExitCode,
ErrorMessage: instance.ErrorMessage,
Metadata: instance.Metadata,
}
}
apps/backend/internal/agent/api/handlers_test.go
-573
View File
@@ -1,573 +0,0 @@
package api
import (
"bytes"
"context"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
"github.com/gin-gonic/gin"
"github.com/kandev/kandev/internal/agent/lifecycle"
"github.com/kandev/kandev/internal/agent/registry"
"github.com/kandev/kandev/internal/common/logger"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
func init() {
gin.SetMode(gin.TestMode)
}
func newTestLogger() *logger.Logger {
log, _ := logger.NewLogger(logger.LoggingConfig{
Level: "error",
Format: "json",
})
return log
}
// MockLifecycleManager implements a mock for lifecycle.Manager
type MockLifecycleManager struct {
LaunchFn func(ctx context.Context, req *lifecycle.LaunchRequest) (*lifecycle.AgentInstance, error)
StopAgentFn func(ctx context.Context, instanceID string, force bool) error
GetInstanceFn func(instanceID string) (*lifecycle.AgentInstance, bool)
ListInstancesFn func() []*lifecycle.AgentInstance
GetInstanceByTaskIDFn func(taskID string) (*lifecycle.AgentInstance, bool)
}
func (m *MockLifecycleManager) Launch(ctx context.Context, req *lifecycle.LaunchRequest) (*lifecycle.AgentInstance, error) {
if m.LaunchFn != nil {
return m.LaunchFn(ctx, req)
}
return &lifecycle.AgentInstance{
ID: "mock-instance-id",
TaskID: req.TaskID,
AgentType: req.AgentType,
ContainerID: "mock-container-id",
Status: v1.AgentStatusRunning,
StartedAt: time.Now(),
}, nil
}
func (m *MockLifecycleManager) StopAgent(ctx context.Context, instanceID string, force bool) error {
if m.StopAgentFn != nil {
return m.StopAgentFn(ctx, instanceID, force)
}
return nil
}
func (m *MockLifecycleManager) GetInstance(instanceID string) (*lifecycle.AgentInstance, bool) {
if m.GetInstanceFn != nil {
return m.GetInstanceFn(instanceID)
}
return nil, false
}
func (m *MockLifecycleManager) ListInstances() []*lifecycle.AgentInstance {
if m.ListInstancesFn != nil {
return m.ListInstancesFn()
}
return []*lifecycle.AgentInstance{}
}
func (m *MockLifecycleManager) GetInstanceByTaskID(taskID string) (*lifecycle.AgentInstance, bool) {
if m.GetInstanceByTaskIDFn != nil {
return m.GetInstanceByTaskIDFn(taskID)
}
return nil, false
}
// MockDockerClient for handler tests
type MockDockerClient struct {
GetContainerLogsFn func(ctx context.Context, containerID string, follow bool, tail string) (io.ReadCloser, error)
}
func (m *MockDockerClient) GetContainerLogs(ctx context.Context, containerID string, follow bool, tail string) (io.ReadCloser, error) {
if m.GetContainerLogsFn != nil {
return m.GetContainerLogsFn(ctx, containerID, follow, tail)
}
return io.NopCloser(strings.NewReader("test log line")), nil
}
func newTestRegistry() *registry.Registry {
log := newTestLogger()
reg := registry.NewRegistry(log)
reg.LoadDefaults()
return reg
}
// Helper to create a test router with handlers
func setupTestRouter(
mockLM *MockLifecycleManager,
reg *registry.Registry,
mockDocker *MockDockerClient,
) *gin.Engine {
log := newTestLogger()
router := gin.New()
apiV1 := router.Group("/api/v1")
// Create handler with lifecycle manager wrapper
handler := &Handler{
registry: reg,
docker: mockDocker,
acp: nil,
logger: log,
}
// Set up routes manually for testing
agents := apiV1.Group("/agents")
{
agents.GET("", func(c *gin.Context) {
instances := mockLM.ListInstances()
agentsList := make([]AgentInstanceResponse, 0, len(instances))
for _, instance := range instances {
agentsList = append(agentsList, instanceToResponse(instance))
}
c.JSON(http.StatusOK, AgentsListResponse{Agents: agentsList, Total: len(agentsList)})
})
agents.POST("/launch", func(c *gin.Context) {
var req LaunchAgentRequest
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
launchReq := &lifecycle.LaunchRequest{
TaskID: req.TaskID,
AgentType: req.AgentType,
WorkspacePath: req.WorkspacePath,
Env: req.Env,
Metadata: req.Metadata,
}
instance, err := mockLM.Launch(c.Request.Context(), launchReq)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusCreated, instanceToResponse(instance))
})
agents.GET("/types", handler.ListAgentTypes)
agents.GET("/types/:typeId", handler.GetAgentType)
agents.GET("/:instanceId/status", func(c *gin.Context) {
instanceID := c.Param("instanceId")
instance, found := mockLM.GetInstance(instanceID)
if !found {
c.JSON(http.StatusNotFound, gin.H{"error": "not found"})
return
}
c.JSON(http.StatusOK, instanceToResponse(instance))
})
agents.DELETE("/:instanceId", func(c *gin.Context) {
instanceID := c.Param("instanceId")
var req StopAgentRequest
_ = c.ShouldBindJSON(&req)
err := mockLM.StopAgent(c.Request.Context(), instanceID, req.Force)
if err != nil {
if strings.Contains(err.Error(), "not found") {
c.JSON(http.StatusNotFound, gin.H{"error": "not found"})
return
}
c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
return
}
c.JSON(http.StatusOK, gin.H{"message": "agent stopped successfully"})
})
}
router.GET("/health", handler.HealthCheck)
return router
}
func TestHandler_LaunchAgent(t *testing.T) {
mockLM := &MockLifecycleManager{}
reg := newTestRegistry()
mockDocker := &MockDockerClient{}
router := setupTestRouter(mockLM, reg, mockDocker)
reqBody := LaunchAgentRequest{
TaskID: "task-123",
AgentType: "augment-agent",
WorkspacePath: "/path/to/workspace",
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/api/v1/agents/launch", bytes.NewBuffer(body))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusCreated {
t.Errorf("expected status %d, got %d: %s", http.StatusCreated, w.Code, w.Body.String())
}
var resp AgentInstanceResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to parse response: %v", err)
}
if resp.TaskID != "task-123" {
t.Errorf("expected TaskID 'task-123', got %q", resp.TaskID)
}
}
func TestHandler_LaunchAgent_InvalidRequest(t *testing.T) {
mockLM := &MockLifecycleManager{}
reg := newTestRegistry()
mockDocker := &MockDockerClient{}
router := setupTestRouter(mockLM, reg, mockDocker)
// Missing required fields
reqBody := map[string]string{}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/api/v1/agents/launch", bytes.NewBuffer(body))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusBadRequest {
t.Errorf("expected status %d, got %d", http.StatusBadRequest, w.Code)
}
}
func TestHandler_StopAgent(t *testing.T) {
mockLM := &MockLifecycleManager{}
reg := newTestRegistry()
mockDocker := &MockDockerClient{}
router := setupTestRouter(mockLM, reg, mockDocker)
req := httptest.NewRequest("DELETE", "/api/v1/agents/instance-123", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status %d, got %d: %s", http.StatusOK, w.Code, w.Body.String())
}
}
func TestHandler_StopAgent_NotFound(t *testing.T) {
mockLM := &MockLifecycleManager{
StopAgentFn: func(ctx context.Context, instanceID string, force bool) error {
return &mockError{msg: "instance not found"}
},
}
reg := newTestRegistry()
mockDocker := &MockDockerClient{}
router := setupTestRouter(mockLM, reg, mockDocker)
req := httptest.NewRequest("DELETE", "/api/v1/agents/non-existent", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusNotFound {
t.Errorf("expected status %d, got %d", http.StatusNotFound, w.Code)
}
}
type mockError struct {
msg string
}
func (e *mockError) Error() string {
return e.msg
}
func TestHandler_GetAgentStatus(t *testing.T) {
now := time.Now()
mockLM := &MockLifecycleManager{
GetInstanceFn: func(instanceID string) (*lifecycle.AgentInstance, bool) {
if instanceID == "instance-123" {
return &lifecycle.AgentInstance{
ID: "instance-123",
TaskID: "task-456",
AgentType: "augment-agent",
ContainerID: "container-789",
Status: v1.AgentStatusRunning,
StartedAt: now,
Progress: 50,
}, true
}
return nil, false
},
}
reg := newTestRegistry()
mockDocker := &MockDockerClient{}
router := setupTestRouter(mockLM, reg, mockDocker)
req := httptest.NewRequest("GET", "/api/v1/agents/instance-123/status", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status %d, got %d: %s", http.StatusOK, w.Code, w.Body.String())
}
var resp AgentInstanceResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to parse response: %v", err)
}
if resp.ID != "instance-123" {
t.Errorf("expected ID 'instance-123', got %q", resp.ID)
}
if resp.Status != "RUNNING" {
t.Errorf("expected status 'RUNNING', got %q", resp.Status)
}
if resp.Progress != 50 {
t.Errorf("expected progress 50, got %d", resp.Progress)
}
}
func TestHandler_GetAgentStatus_NotFound(t *testing.T) {
mockLM := &MockLifecycleManager{
GetInstanceFn: func(instanceID string) (*lifecycle.AgentInstance, bool) {
return nil, false
},
}
reg := newTestRegistry()
mockDocker := &MockDockerClient{}
router := setupTestRouter(mockLM, reg, mockDocker)
req := httptest.NewRequest("GET", "/api/v1/agents/non-existent/status", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusNotFound {
t.Errorf("expected status %d, got %d", http.StatusNotFound, w.Code)
}
}
func TestHandler_ListAgents(t *testing.T) {
now := time.Now()
mockLM := &MockLifecycleManager{
ListInstancesFn: func() []*lifecycle.AgentInstance {
return []*lifecycle.AgentInstance{
{
ID: "instance-1",
TaskID: "task-1",
AgentType: "augment-agent",
ContainerID: "container-1",
Status: v1.AgentStatusRunning,
StartedAt: now,
},
{
ID: "instance-2",
TaskID: "task-2",
AgentType: "augment-agent",
ContainerID: "container-2",
Status: v1.AgentStatusCompleted,
StartedAt: now,
},
}
},
}
reg := newTestRegistry()
mockDocker := &MockDockerClient{}
router := setupTestRouter(mockLM, reg, mockDocker)
req := httptest.NewRequest("GET", "/api/v1/agents", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status %d, got %d: %s", http.StatusOK, w.Code, w.Body.String())
}
var resp AgentsListResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to parse response: %v", err)
}
if resp.Total != 2 {
t.Errorf("expected total 2, got %d", resp.Total)
}
if len(resp.Agents) != 2 {
t.Errorf("expected 2 agents, got %d", len(resp.Agents))
}
}
func TestHandler_ListAgents_Empty(t *testing.T) {
mockLM := &MockLifecycleManager{
ListInstancesFn: func() []*lifecycle.AgentInstance {
return []*lifecycle.AgentInstance{}
},
}
reg := newTestRegistry()
mockDocker := &MockDockerClient{}
router := setupTestRouter(mockLM, reg, mockDocker)
req := httptest.NewRequest("GET", "/api/v1/agents", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status %d, got %d", http.StatusOK, w.Code)
}
var resp AgentsListResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to parse response: %v", err)
}
if resp.Total != 0 {
t.Errorf("expected total 0, got %d", resp.Total)
}
}
func TestHandler_ListAgentTypes(t *testing.T) {
mockLM := &MockLifecycleManager{}
reg := newTestRegistry()
mockDocker := &MockDockerClient{}
router := setupTestRouter(mockLM, reg, mockDocker)
req := httptest.NewRequest("GET", "/api/v1/agents/types", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status %d, got %d: %s", http.StatusOK, w.Code, w.Body.String())
}
var resp AgentTypesListResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to parse response: %v", err)
}
// Should have at least the default agents
if resp.Total == 0 {
t.Error("expected at least one agent type")
}
}
func TestHandler_GetAgentType(t *testing.T) {
mockLM := &MockLifecycleManager{}
reg := newTestRegistry()
mockDocker := &MockDockerClient{}
router := setupTestRouter(mockLM, reg, mockDocker)
req := httptest.NewRequest("GET", "/api/v1/agents/types/augment-agent", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status %d, got %d: %s", http.StatusOK, w.Code, w.Body.String())
}
var resp AgentTypeResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to parse response: %v", err)
}
if resp.ID != "augment-agent" {
t.Errorf("expected ID 'augment-agent', got %q", resp.ID)
}
}
func TestHandler_GetAgentType_NotFound(t *testing.T) {
mockLM := &MockLifecycleManager{}
reg := newTestRegistry()
mockDocker := &MockDockerClient{}
router := setupTestRouter(mockLM, reg, mockDocker)
req := httptest.NewRequest("GET", "/api/v1/agents/types/non-existent", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusNotFound {
t.Errorf("expected status %d, got %d", http.StatusNotFound, w.Code)
}
}
func TestHandler_HealthCheck(t *testing.T) {
mockLM := &MockLifecycleManager{}
reg := newTestRegistry()
mockDocker := &MockDockerClient{}
router := setupTestRouter(mockLM, reg, mockDocker)
req := httptest.NewRequest("GET", "/health", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status %d, got %d: %s", http.StatusOK, w.Code, w.Body.String())
}
var resp HealthResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to parse response: %v", err)
}
if resp.Status != "healthy" {
t.Errorf("expected status 'healthy', got %q", resp.Status)
}
}
func TestInstanceToResponse(t *testing.T) {
now := time.Now()
exitCode := 0
finishedAt := now.Add(time.Hour)
instance := &lifecycle.AgentInstance{
ID: "test-id",
TaskID: "task-id",
AgentType: "test-agent",
ContainerID: "container-id",
Status: v1.AgentStatusCompleted,
StartedAt: now,
FinishedAt: &finishedAt,
ExitCode: &exitCode,
ErrorMessage: "",
Progress: 100,
Metadata: map[string]interface{}{"key": "value"},
}
resp := instanceToResponse(instance)
if resp.ID != instance.ID {
t.Errorf("expected ID %q, got %q", instance.ID, resp.ID)
}
if resp.TaskID != instance.TaskID {
t.Errorf("expected TaskID %q, got %q", instance.TaskID, resp.TaskID)
}
if resp.Status != string(instance.Status) {
t.Errorf("expected Status %q, got %q", instance.Status, resp.Status)
}
if resp.Progress != instance.Progress {
t.Errorf("expected Progress %d, got %d", instance.Progress, resp.Progress)
}
if resp.ExitCode == nil || *resp.ExitCode != exitCode {
t.Errorf("expected ExitCode %d, got %v", exitCode, resp.ExitCode)
}
if resp.Metadata["key"] != "value" {
t.Error("expected metadata to be preserved")
}
}
apps/backend/internal/agent/api/requests.go
-107
View File
@@ -1,107 +0,0 @@
// Package api provides HTTP handlers for the agent manager API.
package api
import "time"
// LaunchAgentRequest for launching an agent
type LaunchAgentRequest struct {
TaskID string `json:"task_id" binding:"required"`
AgentType string `json:"agent_type" binding:"required"`
WorkspacePath string `json:"workspace_path" binding:"required"`
Env map[string]string `json:"env,omitempty"`
Priority int `json:"priority"`
Metadata map[string]interface{} `json:"metadata,omitempty"`
}
// StopAgentRequest for stopping an agent
type StopAgentRequest struct {
Force bool `json:"force"`
Reason string `json:"reason,omitempty"`
}
// AgentInstanceResponse for agent status
type AgentInstanceResponse struct {
ID string `json:"id"`
TaskID string `json:"task_id"`
AgentType string `json:"agent_type"`
ContainerID string `json:"container_id"`
Status string `json:"status"`
Progress int `json:"progress"`
StartedAt time.Time `json:"started_at"`
FinishedAt *time.Time `json:"finished_at,omitempty"`
ExitCode *int `json:"exit_code,omitempty"`
ErrorMessage string `json:"error_message,omitempty"`
Metadata map[string]interface{} `json:"metadata,omitempty"`
}
// AgentTypeResponse for agent type listing
type AgentTypeResponse struct {
ID string `json:"id"`
Name string `json:"name"`
Description string `json:"description"`
Image string `json:"image"`
Capabilities []string `json:"capabilities"`
Enabled bool `json:"enabled"`
}
// AgentsListResponse for listing agents
type AgentsListResponse struct {
Agents []AgentInstanceResponse `json:"agents"`
Total int `json:"total"`
}
// AgentTypesListResponse for listing agent types
type AgentTypesListResponse struct {
Types []AgentTypeResponse `json:"types"`
Total int `json:"total"`
}
// LogEntry represents a single log entry
type LogEntry struct {
Timestamp time.Time `json:"timestamp"`
Message string `json:"message"`
Stream string `json:"stream"` // stdout or stderr
}
// LogsResponse for agent logs
type LogsResponse struct {
Logs []LogEntry `json:"logs"`
Total int `json:"total"`
}
// HealthResponse for health check
type HealthResponse struct {
Status string `json:"status"`
Timestamp time.Time `json:"timestamp"`
}
// SendPromptRequest for sending a prompt to an agent
type SendPromptRequest struct {
Message string `json:"message" binding:"required"`
}
// SendPromptResponse for prompt acknowledgment
type SendPromptResponse struct {
Success bool `json:"success"`
SessionID string `json:"session_id,omitempty"`
Message string `json:"message,omitempty"`
}
// CancelAgentRequest for cancelling agent operation
type CancelAgentRequest struct {
Reason string `json:"reason,omitempty"`
}
// CancelAgentResponse for cancel acknowledgment
type CancelAgentResponse struct {
Success bool `json:"success"`
Message string `json:"message,omitempty"`
}
// GetSessionResponse for getting session info
type GetSessionResponse struct {
InstanceID string `json:"instance_id"`
SessionID string `json:"session_id"`
Status string `json:"status"`
}
apps/backend/internal/agent/api/router.go
-47
View File
@@ -1,47 +0,0 @@
package api
import (
"github.com/gin-gonic/gin"
"github.com/kandev/kandev/internal/agent/lifecycle"
"github.com/kandev/kandev/internal/agent/registry"
"github.com/kandev/kandev/internal/common/logger"
)
// SetupRoutes configures the agent manager API routes
// router should be the /api/v1 group
func SetupRoutes(
router *gin.RouterGroup,
lm *lifecycle.Manager,
reg *registry.Registry,
docker DockerClient,
acp ACPManager,
log *logger.Logger,
) {
handler := NewHandler(lm, reg, docker, acp, log)
// Agent instance endpoints under /api/v1/agents
agents := router.Group("/agents")
{
// List all active agents
agents.GET("", handler.ListAgents)
// Launch a new agent
agents.POST("/launch", handler.LaunchAgent)
// Agent types endpoints
agents.GET("/types", handler.ListAgentTypes)
agents.GET("/types/:typeId", handler.GetAgentType)
// Single agent instance operations
agents.GET("/:instanceId/status", handler.GetAgentStatus)
agents.GET("/:instanceId/logs", handler.GetAgentLogs)
agents.DELETE("/:instanceId", handler.StopAgent)
// ACP communication endpoints
agents.POST("/:instanceId/prompt", handler.SendPrompt)
agents.POST("/:instanceId/cancel", handler.CancelAgent)
agents.GET("/:instanceId/session", handler.GetSession)
}
}
apps/backend/internal/agent/models/.gitkeep
View File
apps/backend/internal/agent/monitor/.gitkeep
View File
apps/backend/internal/common/acp/.gitkeep
View File
apps/backend/internal/common/auth/.gitkeep
View File
apps/backend/internal/events/publisher/.gitkeep
View File
apps/backend/internal/events/subscriber/.gitkeep
View File
apps/backend/internal/gateway/handlers/.gitkeep
View File
apps/backend/internal/gateway/middleware/.gitkeep
View File
apps/backend/internal/gateway/router/.gitkeep
View File
apps/backend/internal/gateway/streaming/.gitkeep
View File
apps/backend/internal/orchestrator/api/.gitkeep
View File
apps/backend/internal/orchestrator/api/handlers.go
-313
View File
@@ -1,313 +0,0 @@
package api
import (
"net/http"
"github.com/gin-gonic/gin"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/common/errors"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/orchestrator"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
// Handler contains HTTP handlers for the orchestrator API
type Handler struct {
service *orchestrator.Service
logger *logger.Logger
}
// NewHandler creates a new API handler
func NewHandler(service *orchestrator.Service, log *logger.Logger) *Handler {
return &Handler{
service: service,
logger: log.WithFields(zap.String("component", "orchestrator-api")),
}
}
// GetStatus returns the overall orchestrator status
// GET /api/v1/orchestrator/status
func (h *Handler) GetStatus(c *gin.Context) {
status := h.service.GetStatus()
c.JSON(http.StatusOK, status)
}
// GetQueue returns the current task queue
// GET /api/v1/orchestrator/queue
func (h *Handler) GetQueue(c *gin.Context) {
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,
})
}
c.JSON(http.StatusOK, QueueResponse{
Tasks: tasks,
Total: len(tasks),
})
}
// TriggerTask manually triggers orchestration for a specific task
// POST /api/v1/orchestrator/trigger
func (h *Handler) TriggerTask(c *gin.Context) {
var req TriggerTaskRequest
if err := c.ShouldBindJSON(&req); err != nil {
appErr := errors.ValidationError("request", err.Error())
c.JSON(appErr.HTTPStatus, appErr)
return
}
// Create a minimal task for enqueueing
task := &v1.Task{
ID: req.TaskID,
}
if err := h.service.EnqueueTask(c.Request.Context(), task); err != nil {
h.logger.Error("failed to trigger task", zap.String("task_id", req.TaskID), zap.Error(err))
appErr := errors.Wrap(err, "failed to trigger task")
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusAccepted, gin.H{
"message": "task triggered successfully",
"task_id": req.TaskID,
})
}
// StartTask starts agent execution for a task
// POST /api/v1/orchestrator/tasks/:taskId/start
func (h *Handler) StartTask(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
var req StartTaskRequest
if err := c.ShouldBindJSON(&req); err != nil {
// Allow empty body, use defaults
req = StartTaskRequest{}
}
execution, err := h.service.StartTask(c.Request.Context(), taskID, req.AgentType, req.Priority)
if err != nil {
h.logger.Error("failed to start task", zap.String("task_id", taskID), zap.Error(err))
appErr := errors.Wrap(err, "failed to start task")
c.JSON(appErr.HTTPStatus, appErr)
return
}
startedAt := execution.StartedAt
c.JSON(http.StatusAccepted, TaskStatusResponse{
TaskID: execution.TaskID,
State: string(execution.Status),
AgentInstanceID: execution.AgentInstanceID,
AgentType: execution.AgentType,
StartedAt: &startedAt,
Progress: execution.Progress,
})
}
// StopTask stops agent execution for a task
// POST /api/v1/orchestrator/tasks/:taskId/stop
func (h *Handler) StopTask(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
var req StopTaskRequest
if err := c.ShouldBindJSON(&req); err != nil {
// Allow empty body, use defaults
req = StopTaskRequest{}
}
if err := h.service.StopTask(c.Request.Context(), taskID, req.Reason, req.Force); err != nil {
h.logger.Error("failed to stop task", zap.String("task_id", taskID), zap.Error(err))
appErr := errors.Wrap(err, "failed to stop task")
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusOK, gin.H{
"message": "task stopped successfully",
"task_id": taskID,
})
}
// PromptTask sends a follow-up prompt to a running agent
// POST /api/v1/orchestrator/tasks/:taskId/prompt
func (h *Handler) PromptTask(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
var req PromptTaskRequest
if err := c.ShouldBindJSON(&req); err != nil {
appErr := errors.ValidationError("request", err.Error())
c.JSON(appErr.HTTPStatus, appErr)
return
}
if err := h.service.PromptTask(c.Request.Context(), taskID, req.Prompt); err != nil {
h.logger.Error("failed to prompt task", zap.String("task_id", taskID), zap.Error(err))
c.JSON(http.StatusInternalServerError, PromptTaskResponse{
Success: false,
Error: err.Error(),
})
return
}
c.JSON(http.StatusOK, PromptTaskResponse{Success: true})
}
// CompleteTask explicitly completes a running agent and marks the task as completed
// POST /api/v1/orchestrator/tasks/:taskId/complete
func (h *Handler) CompleteTask(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
if err := h.service.CompleteTask(c.Request.Context(), taskID); err != nil {
h.logger.Error("failed to complete task", zap.String("task_id", taskID), zap.Error(err))
c.JSON(http.StatusInternalServerError, gin.H{
"success": false,
"error": err.Error(),
})
return
}
c.JSON(http.StatusOK, gin.H{"success": true, "message": "task completed"})
}
// PauseTask pauses agent execution (if supported)
// POST /api/v1/orchestrator/tasks/:taskId/pause
func (h *Handler) PauseTask(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
// Pause is not currently implemented
c.JSON(http.StatusNotImplemented, gin.H{
"code": "NOT_IMPLEMENTED",
"message": "pause functionality is not yet implemented",
"task_id": taskID,
})
}
// ResumeTask resumes paused agent execution
// POST /api/v1/orchestrator/tasks/:taskId/resume
func (h *Handler) ResumeTask(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
// Resume is not currently implemented
c.JSON(http.StatusNotImplemented, gin.H{
"code": "NOT_IMPLEMENTED",
"message": "resume functionality is not yet implemented",
"task_id": taskID,
})
}
// GetTaskStatus returns detailed execution status for a task
// GET /api/v1/orchestrator/tasks/:taskId/status
func (h *Handler) GetTaskStatus(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
execution, found := h.service.GetTaskExecution(taskID)
if !found {
appErr := errors.NotFound("task execution", taskID)
c.JSON(appErr.HTTPStatus, appErr)
return
}
startedAt := execution.StartedAt
c.JSON(http.StatusOK, TaskStatusResponse{
TaskID: execution.TaskID,
State: string(execution.Status),
AgentInstanceID: execution.AgentInstanceID,
AgentType: execution.AgentType,
StartedAt: &startedAt,
Progress: execution.Progress,
})
}
// GetTaskLogs returns historical logs for a task's agent execution
// GET /api/v1/orchestrator/tasks/:taskId/logs
func (h *Handler) GetTaskLogs(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
// Check if task execution exists
_, found := h.service.GetTaskExecution(taskID)
if !found {
appErr := errors.NotFound("task execution", taskID)
c.JSON(appErr.HTTPStatus, appErr)
return
}
// Logs retrieval is not fully implemented yet
// Return empty logs for now
c.JSON(http.StatusOK, LogsResponse{
Logs: []LogEntry{},
Total: 0,
})
}
// GetTaskArtifacts lists artifacts generated by agent
// GET /api/v1/orchestrator/tasks/:taskId/artifacts
func (h *Handler) GetTaskArtifacts(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
// Check if task execution exists
_, found := h.service.GetTaskExecution(taskID)
if !found {
appErr := errors.NotFound("task execution", taskID)
c.JSON(appErr.HTTPStatus, appErr)
return
}
// Artifacts retrieval is not fully implemented yet
// Return empty artifacts for now
c.JSON(http.StatusOK, ArtifactsResponse{
Artifacts: []ArtifactResponse{},
})
}
apps/backend/internal/orchestrator/api/middleware.go
-164
View File
@@ -1,164 +0,0 @@
// Package api provides HTTP middleware for the Orchestrator API.
package api
import (
stderrors "errors"
"net/http"
"sync"
"time"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/common/errors"
"github.com/kandev/kandev/internal/common/logger"
)
// RequestLogger logs all incoming requests with detailed information.
func RequestLogger(log *logger.Logger) gin.HandlerFunc {
return func(c *gin.Context) {
start := time.Now()
// Generate request ID
requestID := uuid.New().String()
c.Set("request_id", requestID)
c.Header("X-Request-ID", requestID)
// Process request
c.Next()
// Log request details
duration := time.Since(start)
log.Info("Request completed",
zap.String("path", c.Request.URL.Path),
zap.String("method", c.Request.Method),
zap.Int("status", c.Writer.Status()),
zap.Duration("duration", duration),
zap.String("request_id", requestID),
)
}
}
// ErrorHandler handles errors and returns appropriate responses.
func ErrorHandler(log *logger.Logger) gin.HandlerFunc {
return func(c *gin.Context) {
c.Next()
// Check for errors
if len(c.Errors) > 0 {
err := c.Errors.Last().Err
// Check if it's an AppError
var appErr *errors.AppError
if stderrors.As(err, &appErr) {
log.Error("Request error",
zap.String("code", appErr.Code),
zap.String("message", appErr.Message),
zap.Int("status", appErr.HTTPStatus),
)
c.JSON(appErr.HTTPStatus, gin.H{
"error": gin.H{
"code": appErr.Code,
"message": appErr.Message,
},
})
return
}
// Default to internal server error
log.Error("Internal server error", zap.Error(err))
c.JSON(http.StatusInternalServerError, gin.H{
"error": gin.H{
"code": errors.ErrCodeInternalError,
"message": "An internal server error occurred",
},
})
}
}
}
// Recovery recovers from panics and logs them.
func Recovery(log *logger.Logger) gin.HandlerFunc {
return func(c *gin.Context) {
defer func() {
if r := recover(); r != nil {
log.Error("Panic recovered",
zap.Any("panic", r),
zap.String("path", c.Request.URL.Path),
zap.String("method", c.Request.Method),
)
c.AbortWithStatusJSON(http.StatusInternalServerError, gin.H{
"error": gin.H{
"code": errors.ErrCodeInternalError,
"message": "An internal server error occurred",
},
})
}
}()
c.Next()
}
}
// CORS adds CORS headers for cross-origin requests.
func CORS() gin.HandlerFunc {
return func(c *gin.Context) {
c.Header("Access-Control-Allow-Origin", "*")
c.Header("Access-Control-Allow-Methods", "GET, POST, PUT, PATCH, DELETE, OPTIONS")
c.Header("Access-Control-Allow-Headers", "Origin, Content-Type, Accept, Authorization, X-Request-ID")
c.Header("Access-Control-Expose-Headers", "X-Request-ID")
c.Header("Access-Control-Max-Age", "86400")
if c.Request.Method == "OPTIONS" {
c.AbortWithStatus(http.StatusNoContent)
return
}
c.Next()
}
}
// RateLimit provides basic rate limiting using a token bucket algorithm.
// This is a placeholder implementation - for production, consider using
// a distributed rate limiter like redis-based solutions.
func RateLimit(requestsPerSecond int) gin.HandlerFunc {
var (
mu sync.Mutex
tokens = float64(requestsPerSecond)
lastTime = time.Now()
)
return func(c *gin.Context) {
mu.Lock()
now := time.Now()
elapsed := now.Sub(lastTime).Seconds()
lastTime = now
// Refill tokens
tokens += elapsed * float64(requestsPerSecond)
if tokens > float64(requestsPerSecond) {
tokens = float64(requestsPerSecond)
}
// Check if we have tokens available
if tokens < 1 {
mu.Unlock()
c.AbortWithStatusJSON(http.StatusTooManyRequests, gin.H{
"error": gin.H{
"code": "RATE_LIMIT_EXCEEDED",
"message": "Too many requests, please try again later",
},
})
return
}
tokens--
mu.Unlock()
c.Next()
}
}
apps/backend/internal/orchestrator/api/requests.go
-94
View File
@@ -1,94 +0,0 @@
// Package api provides REST API handlers for the orchestrator service.
package api
import "time"
// TriggerTaskRequest for manually triggering a task
type TriggerTaskRequest struct {
TaskID string `json:"task_id" binding:"required"`
Force bool `json:"force"`
}
// StartTaskRequest for starting agent execution
type StartTaskRequest struct {
AgentType string `json:"agent_type,omitempty"`
Priority int `json:"priority"`
}
// StopTaskRequest for stopping agent execution
type StopTaskRequest struct {
Reason string `json:"reason,omitempty"`
Force bool `json:"force"`
}
// PromptTaskRequest for sending a follow-up prompt to a running agent
type PromptTaskRequest struct {
Prompt string `json:"prompt" binding:"required"`
}
// PromptTaskResponse for prompt endpoint
type PromptTaskResponse struct {
Success bool `json:"success"`
Error string `json:"error,omitempty"`
}
// CompleteTaskRequest for explicitly completing a task
type CompleteTaskRequest struct {
Message string `json:"message,omitempty"`
}
// TaskStatusResponse for task execution status
type TaskStatusResponse struct {
TaskID string `json:"task_id"`
State string `json:"state"`
AgentInstanceID string `json:"agent_instance_id,omitempty"`
AgentType string `json:"agent_type,omitempty"`
StartedAt *time.Time `json:"started_at,omitempty"`
Progress int `json:"progress"`
CurrentOperation string `json:"current_operation,omitempty"`
EstimatedCompletion *time.Time `json:"estimated_completion,omitempty"`
}
// QueuedTaskResponse for queue listing
type QueuedTaskResponse struct {
TaskID string `json:"task_id"`
Priority int `json:"priority"`
QueuedAt time.Time `json:"queued_at"`
EstimatedStart *time.Time `json:"estimated_start,omitempty"`
}
// QueueResponse for queue endpoint
type QueueResponse struct {
Tasks []QueuedTaskResponse `json:"tasks"`
Total int `json:"total"`
}
// LogEntry for log listing
type LogEntry struct {
Timestamp time.Time `json:"timestamp"`
Level string `json:"level"`
Message string `json:"message"`
Metadata map[string]interface{} `json:"metadata,omitempty"`
}
// LogsResponse for logs endpoint
type LogsResponse struct {
Logs []LogEntry `json:"logs"`
Total int `json:"total"`
}
// ArtifactResponse for artifact listing
type ArtifactResponse struct {
ID string `json:"id"`
Name string `json:"name"`
Type string `json:"type"`
Size int64 `json:"size"`
CreatedAt time.Time `json:"created_at"`
DownloadURL string `json:"download_url"`
}
// ArtifactsResponse for artifacts endpoint
type ArtifactsResponse struct {
Artifacts []ArtifactResponse `json:"artifacts"`
}
apps/backend/internal/orchestrator/api/router.go
-32
View File
@@ -1,32 +0,0 @@
package api
import (
"github.com/gin-gonic/gin"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/orchestrator"
)
// SetupRoutes configures the orchestrator API routes
func SetupRoutes(router *gin.RouterGroup, service *orchestrator.Service, log *logger.Logger) {
handler := NewHandler(service, log)
// Orchestrator control
router.GET("/status", handler.GetStatus)
router.GET("/queue", handler.GetQueue)
router.POST("/trigger", handler.TriggerTask)
// Task execution control
tasks := router.Group("/tasks/:taskId")
{
tasks.POST("/start", handler.StartTask)
tasks.POST("/stop", handler.StopTask)
tasks.POST("/prompt", handler.PromptTask)
tasks.POST("/complete", handler.CompleteTask)
tasks.POST("/pause", handler.PauseTask)
tasks.POST("/resume", handler.ResumeTask)
tasks.GET("/status", handler.GetTaskStatus)
tasks.GET("/logs", handler.GetTaskLogs)
tasks.GET("/artifacts", handler.GetTaskArtifacts)
}
}
apps/backend/internal/orchestrator/scheduler/scheduler_test.go
+4
View File
@@ -55,6 +55,10 @@ func (m *mockAgentManager) ListAgentTypes(ctx context.Context) ([]*v1.AgentType,
return []*v1.AgentType{}, nil
}
func (m *mockAgentManager) PromptAgent(ctx context.Context, agentInstanceID string, prompt string) error {
return nil
}
// testTaskRepository is an in-memory task repository for testing
type testTaskRepository struct {
tasks map[string]*v1.Task
apps/backend/internal/orchestrator/streaming/.gitkeep
View File
apps/backend/internal/orchestrator/streaming/client.go
-151
View File
@@ -1,151 +0,0 @@
package streaming
import (
"encoding/json"
"time"
"github.com/gorilla/websocket"
"go.uber.org/zap"
)
const (
writeWait = 10 * time.Second
pongWait = 60 * time.Second
pingPeriod = (pongWait * 9) / 10
maxMessageSize = 1024 * 1024 // 1MB
)
// SubscriptionMessage is sent by clients to subscribe/unsubscribe
type SubscriptionMessage struct {
Action string `json:"action"` // subscribe, unsubscribe
TaskIDs []string `json:"task_ids"`
}
// ReadPump reads messages from the WebSocket connection
func (c *Client) ReadPump() {
defer func() {
c.hub.Unregister(c)
c.conn.Close()
}()
c.conn.SetReadLimit(maxMessageSize)
c.conn.SetReadDeadline(time.Now().Add(pongWait))
c.conn.SetPongHandler(func(string) error {
c.conn.SetReadDeadline(time.Now().Add(pongWait))
return nil
})
for {
_, message, err := c.conn.ReadMessage()
if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
c.logger.Warn("WebSocket read error", zap.Error(err))
}
break
}
var subMsg SubscriptionMessage
if err := json.Unmarshal(message, &subMsg); err != nil {
c.logger.Warn("Invalid subscription message", zap.Error(err))
continue
}
switch subMsg.Action {
case "subscribe":
for _, taskID := range subMsg.TaskIDs {
c.Subscribe(taskID)
}
case "unsubscribe":
for _, taskID := range subMsg.TaskIDs {
c.Unsubscribe(taskID)
}
default:
c.logger.Warn("Unknown action", zap.String("action", subMsg.Action))
}
}
}
// WritePump writes messages to the WebSocket connection
func (c *Client) WritePump() {
ticker := time.NewTicker(pingPeriod)
defer func() {
ticker.Stop()
c.conn.Close()
}()
for {
select {
case message, ok := <-c.send:
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
if !ok {
// Hub closed the channel
c.conn.WriteMessage(websocket.CloseMessage, []byte{})
return
}
w, err := c.conn.NextWriter(websocket.TextMessage)
if err != nil {
return
}
w.Write(message)
// Add queued messages to the current websocket message
n := len(c.send)
for i := 0; i < n; i++ {
w.Write([]byte{'\n'})
w.Write(<-c.send)
}
if err := w.Close(); err != nil {
return
}
case <-ticker.C:
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil {
return
}
}
}
}
// Send sends a message to the client
func (c *Client) Send(msg []byte) bool {
select {
case c.send <- msg:
return true
default:
return false
}
}
// Close closes the client connection
func (c *Client) Close() {
c.hub.Unregister(c)
}
// Subscribe subscribes the client to a task
func (c *Client) Subscribe(taskID string) {
c.mu.Lock()
c.taskIDs[taskID] = true
c.mu.Unlock()
c.hub.SubscribeClient(c, taskID)
c.logger.Debug("Subscribed to task", zap.String("task_id", taskID))
}
// Unsubscribe unsubscribes the client from a task
func (c *Client) Unsubscribe(taskID string) {
c.mu.Lock()
delete(c.taskIDs, taskID)
c.mu.Unlock()
c.hub.UnsubscribeClient(c, taskID)
c.logger.Debug("Unsubscribed from task", zap.String("task_id", taskID))
}
// IsSubscribed returns true if the client is subscribed to a task
func (c *Client) IsSubscribed(taskID string) bool {
c.mu.RLock()
defer c.mu.RUnlock()
return c.taskIDs[taskID]
}
apps/backend/internal/orchestrator/streaming/handlers.go
-140
View File
@@ -1,140 +0,0 @@
// Package streaming provides WebSocket streaming handlers for ACP messages.
package streaming
import (
"net/http"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"github.com/gorilla/websocket"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/orchestrator"
)
var upgrader = websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
CheckOrigin: func(r *http.Request) bool {
// In production, validate origin
return true
},
}
// WSHandler handles WebSocket connections
type WSHandler struct {
hub *Hub
service *orchestrator.Service
logger *logger.Logger
}
// NewWSHandler creates a new WebSocket handler
func NewWSHandler(hub *Hub, service *orchestrator.Service, log *logger.Logger) *WSHandler {
return &WSHandler{
hub: hub,
service: service,
logger: log.WithFields(zap.String("component", "ws_handler")),
}
}
// StreamTask handles WebSocket connection for a specific task
// WS /api/v1/orchestrator/tasks/:taskId/stream
func (h *WSHandler) StreamTask(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
c.JSON(http.StatusBadRequest, gin.H{
"error": gin.H{
"code": "MISSING_TASK_ID",
"message": "Task ID is required",
},
})
return
}
// Validate JWT from query param or header (optional for now)
token := c.Query("token")
if token == "" {
token = c.GetHeader("Authorization")
}
// TODO: Implement proper JWT validation
_ = token
// Upgrade connection to WebSocket
conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
if err != nil {
h.logger.Error("Failed to upgrade connection",
zap.String("task_id", taskID),
zap.Error(err),
)
return
}
// Create a unique client ID
clientID := uuid.New().String()
h.logger.Info("WebSocket connection established for task",
zap.String("client_id", clientID),
zap.String("task_id", taskID),
)
// Create client and register with hub
client := NewClient(clientID, conn, h.hub, h.logger)
// Register client with hub first
h.hub.Register(client)
// Subscribe client to the specific task
client.Subscribe(taskID)
// Start read and write pumps in separate goroutines
go client.WritePump()
go client.ReadPump()
}
// StreamAll handles WebSocket connection for all tasks (with subscription)
// WS /api/v1/orchestrator/stream
func (h *WSHandler) StreamAll(c *gin.Context) {
// Validate JWT from query param or header (optional for now)
token := c.Query("token")
if token == "" {
token = c.GetHeader("Authorization")
}
// TODO: Implement proper JWT validation
_ = token
// Upgrade connection to WebSocket
conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
if err != nil {
h.logger.Error("Failed to upgrade connection", zap.Error(err))
return
}
// Create a unique client ID
clientID := uuid.New().String()
h.logger.Info("WebSocket connection established for all tasks",
zap.String("client_id", clientID),
)
// Create client and register with hub
client := NewClient(clientID, conn, h.hub, h.logger)
// Register client with hub
h.hub.Register(client)
// Start read and write pumps in separate goroutines
// The ReadPump handles subscription messages from the client
go client.WritePump()
go client.ReadPump()
}
// SetupWebSocketRoutes adds WebSocket routes to the router
func SetupWebSocketRoutes(router *gin.RouterGroup, handler *WSHandler) {
// Stream for a specific task
router.GET("/tasks/:taskId/stream", handler.StreamTask)
// Stream for all tasks (with dynamic subscription)
router.GET("/stream", handler.StreamAll)
}
apps/backend/internal/orchestrator/streaming/hub.go
-218
View File
@@ -1,218 +0,0 @@
// Package streaming handles WebSocket connections for real-time ACP message streaming.
package streaming
import (
"context"
"encoding/json"
"sync"
"github.com/gorilla/websocket"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/pkg/acp/protocol"
"go.uber.org/zap"
)
// Client represents a WebSocket client connection
type Client struct {
ID string
conn *websocket.Conn
taskIDs map[string]bool // Tasks this client is subscribed to
send chan []byte
hub *Hub
mu sync.RWMutex
logger *logger.Logger
}
// NewClient creates a new WebSocket client
func NewClient(id string, conn *websocket.Conn, hub *Hub, log *logger.Logger) *Client {
return &Client{
ID: id,
conn: conn,
taskIDs: make(map[string]bool),
send: make(chan []byte, 256),
hub: hub,
logger: log.WithFields(zap.String("client_id", id)),
}
}
// Hub manages all WebSocket clients
type Hub struct {
// Registered clients
clients map[*Client]bool
// Clients by task ID for efficient message routing
taskClients map[string]map[*Client]bool
// Channels
register chan *Client
unregister chan *Client
broadcast chan *BroadcastMessage
mu sync.RWMutex
logger *logger.Logger
}
// BroadcastMessage contains a message to broadcast
type BroadcastMessage struct {
TaskID string
Message *protocol.Message
}
// NewHub creates a new WebSocket hub
func NewHub(log *logger.Logger) *Hub {
return &Hub{
clients: make(map[*Client]bool),
taskClients: make(map[string]map[*Client]bool),
register: make(chan *Client),
unregister: make(chan *Client),
broadcast: make(chan *BroadcastMessage, 256),
logger: log.WithFields(zap.String("component", "websocket_hub")),
}
}
// Run starts the hub processing loop
func (h *Hub) Run(ctx context.Context) {
h.logger.Info("WebSocket hub started")
defer h.logger.Info("WebSocket hub stopped")
for {
select {
case <-ctx.Done():
// Close all client connections
h.mu.Lock()
for client := range h.clients {
close(client.send)
delete(h.clients, client)
}
h.taskClients = make(map[string]map[*Client]bool)
h.mu.Unlock()
return
case client := <-h.register:
h.mu.Lock()
h.clients[client] = true
h.mu.Unlock()
h.logger.Debug("Client registered", zap.String("client_id", client.ID))
case client := <-h.unregister:
h.mu.Lock()
if _, ok := h.clients[client]; ok {
delete(h.clients, client)
close(client.send)
// Remove from all task subscriptions
for taskID := range client.taskIDs {
if clients, ok := h.taskClients[taskID]; ok {
delete(clients, client)
if len(clients) == 0 {
delete(h.taskClients, taskID)
}
}
}
}
h.mu.Unlock()
h.logger.Debug("Client unregistered", zap.String("client_id", client.ID))
case msg := <-h.broadcast:
h.mu.RLock()
clients := h.taskClients[msg.TaskID]
h.mu.RUnlock()
if len(clients) == 0 {
continue
}
data, err := json.Marshal(msg.Message)
if err != nil {
h.logger.Error("Failed to marshal message", zap.Error(err))
continue
}
for client := range clients {
select {
case client.send <- data:
default:
// Client send buffer is full, close connection
h.mu.Lock()
close(client.send)
delete(h.clients, client)
for taskID := range client.taskIDs {
if taskClients, ok := h.taskClients[taskID]; ok {
delete(taskClients, client)
if len(taskClients) == 0 {
delete(h.taskClients, taskID)
}
}
}
h.mu.Unlock()
}
}
}
}
}
// Register adds a client to the hub
func (h *Hub) Register(client *Client) {
h.register <- client
}
// Unregister removes a client from the hub
func (h *Hub) Unregister(client *Client) {
h.unregister <- client
}
// Broadcast sends a message to all clients subscribed to a task
func (h *Hub) Broadcast(taskID string, msg *protocol.Message) {
h.broadcast <- &BroadcastMessage{
TaskID: taskID,
Message: msg,
}
}
// SubscribeClient subscribes a client to a task
func (h *Hub) SubscribeClient(client *Client, taskID string) {
h.mu.Lock()
defer h.mu.Unlock()
if _, ok := h.taskClients[taskID]; !ok {
h.taskClients[taskID] = make(map[*Client]bool)
}
h.taskClients[taskID][client] = true
h.logger.Debug("Client subscribed to task",
zap.String("client_id", client.ID),
zap.String("task_id", taskID))
}
// UnsubscribeClient unsubscribes a client from a task
func (h *Hub) UnsubscribeClient(client *Client, taskID string) {
h.mu.Lock()
defer h.mu.Unlock()
if clients, ok := h.taskClients[taskID]; ok {
delete(clients, client)
if len(clients) == 0 {
delete(h.taskClients, taskID)
}
}
h.logger.Debug("Client unsubscribed from task",
zap.String("client_id", client.ID),
zap.String("task_id", taskID))
}
// GetClientCount returns the number of connected clients
func (h *Hub) GetClientCount() int {
h.mu.RLock()
defer h.mu.RUnlock()
return len(h.clients)
}
// GetTaskSubscriberCount returns the number of clients subscribed to a task
func (h *Hub) GetTaskSubscriberCount(taskID string) int {
h.mu.RLock()
defer h.mu.RUnlock()
if clients, ok := h.taskClients[taskID]; ok {
return len(clients)
}
return 0
}
apps/backend/internal/task/api/handlers.go
-478
View File
@@ -1,478 +0,0 @@
package api
import (
"net/http"
"github.com/gin-gonic/gin"
"go.uber.org/zap"
"github.com/kandev/kandev/internal/common/errors"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/task/models"
"github.com/kandev/kandev/internal/task/service"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
// Handler contains HTTP handlers for the task API
type Handler struct {
service *service.Service
logger *logger.Logger
}
// NewHandler creates a new API handler
func NewHandler(svc *service.Service, log *logger.Logger) *Handler {
return &Handler{
service: svc,
logger: log,
}
}
// Task endpoints
// CreateTask creates a new task
// POST /api/v1/tasks
func (h *Handler) CreateTask(c *gin.Context) {
var req CreateTaskRequest
if err := c.ShouldBindJSON(&req); err != nil {
appErr := errors.BadRequest(err.Error())
c.JSON(appErr.HTTPStatus, appErr)
return
}
svcReq := &service.CreateTaskRequest{
BoardID: req.BoardID,
ColumnID: req.ColumnID,
Title: req.Title,
Description: req.Description,
Priority: req.Priority,
AgentType: req.AgentType,
RepositoryURL: req.RepositoryURL,
Branch: req.Branch,
Metadata: req.Metadata,
}
task, err := h.service.CreateTask(c.Request.Context(), svcReq)
if err != nil {
h.logger.Error("failed to create task", zap.Error(err))
appErr := errors.InternalError("failed to create task", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusCreated, taskToResponse(task))
}
// GetTask retrieves a task by ID
// GET /api/v1/tasks/:taskId
func (h *Handler) GetTask(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
task, err := h.service.GetTask(c.Request.Context(), taskID)
if err != nil {
appErr := errors.NotFound("task", taskID)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusOK, taskToResponse(task))
}
// UpdateTask updates an existing task
// PUT /api/v1/tasks/:taskId
func (h *Handler) UpdateTask(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
var req UpdateTaskRequest
if err := c.ShouldBindJSON(&req); err != nil {
appErr := errors.BadRequest(err.Error())
c.JSON(appErr.HTTPStatus, appErr)
return
}
svcReq := &service.UpdateTaskRequest{
Title: req.Title,
Description: req.Description,
Priority: req.Priority,
AgentType: req.AgentType,
Metadata: req.Metadata,
}
task, err := h.service.UpdateTask(c.Request.Context(), taskID, svcReq)
if err != nil {
h.logger.Error("failed to update task", zap.String("task_id", taskID), zap.Error(err))
appErr := errors.InternalError("failed to update task", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusOK, taskToResponse(task))
}
// DeleteTask deletes a task
// DELETE /api/v1/tasks/:taskId
func (h *Handler) DeleteTask(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
if err := h.service.DeleteTask(c.Request.Context(), taskID); err != nil {
h.logger.Error("failed to delete task", zap.String("task_id", taskID), zap.Error(err))
appErr := errors.InternalError("failed to delete task", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.Status(http.StatusNoContent)
}
// ListTasks returns all tasks for a board
// GET /api/v1/boards/:boardId/tasks
func (h *Handler) ListTasks(c *gin.Context) {
boardID := c.Param("boardId")
if boardID == "" {
appErr := errors.BadRequest("boardId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
tasks, err := h.service.ListTasks(c.Request.Context(), boardID)
if err != nil {
h.logger.Error("failed to list tasks", zap.String("board_id", boardID), zap.Error(err))
appErr := errors.InternalError("failed to list tasks", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
resp := TasksListResponse{
Tasks: make([]*TaskResponse, len(tasks)),
Total: len(tasks),
}
for i, t := range tasks {
resp.Tasks[i] = taskToResponse(t)
}
c.JSON(http.StatusOK, resp)
}
// UpdateTaskState updates the state of a task
// PUT /api/v1/tasks/:taskId/state
func (h *Handler) UpdateTaskState(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
var req UpdateTaskStateRequest
if err := c.ShouldBindJSON(&req); err != nil {
appErr := errors.BadRequest(err.Error())
c.JSON(appErr.HTTPStatus, appErr)
return
}
state := v1.TaskState(req.State)
task, err := h.service.UpdateTaskState(c.Request.Context(), taskID, state)
if err != nil {
h.logger.Error("failed to update task state", zap.String("task_id", taskID), zap.Error(err))
appErr := errors.InternalError("failed to update task state", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusOK, taskToResponse(task))
}
// MoveTask moves a task to a different column
// PUT /api/v1/tasks/:taskId/move
func (h *Handler) MoveTask(c *gin.Context) {
taskID := c.Param("taskId")
if taskID == "" {
appErr := errors.BadRequest("taskId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
var req MoveTaskRequest
if err := c.ShouldBindJSON(&req); err != nil {
appErr := errors.BadRequest(err.Error())
c.JSON(appErr.HTTPStatus, appErr)
return
}
task, err := h.service.MoveTask(c.Request.Context(), taskID, req.ColumnID, req.Position)
if err != nil {
h.logger.Error("failed to move task", zap.String("task_id", taskID), zap.Error(err))
appErr := errors.InternalError("failed to move task", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusOK, taskToResponse(task))
}
// Board endpoints
// CreateBoard creates a new board
// POST /api/v1/boards
func (h *Handler) CreateBoard(c *gin.Context) {
var req CreateBoardRequest
if err := c.ShouldBindJSON(&req); err != nil {
appErr := errors.BadRequest(err.Error())
c.JSON(appErr.HTTPStatus, appErr)
return
}
svcReq := &service.CreateBoardRequest{
Name: req.Name,
Description: req.Description,
OwnerID: "", // TODO: Get from authenticated user
}
board, err := h.service.CreateBoard(c.Request.Context(), svcReq)
if err != nil {
h.logger.Error("failed to create board", zap.Error(err))
appErr := errors.InternalError("failed to create board", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusCreated, boardToResponse(board))
}
// GetBoard retrieves a board by ID
// GET /api/v1/boards/:boardId
func (h *Handler) GetBoard(c *gin.Context) {
boardID := c.Param("boardId")
if boardID == "" {
appErr := errors.BadRequest("boardId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
board, err := h.service.GetBoard(c.Request.Context(), boardID)
if err != nil {
appErr := errors.NotFound("board", boardID)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusOK, boardToResponse(board))
}
// UpdateBoard updates an existing board
// PUT /api/v1/boards/:boardId
func (h *Handler) UpdateBoard(c *gin.Context) {
boardID := c.Param("boardId")
if boardID == "" {
appErr := errors.BadRequest("boardId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
var req UpdateBoardRequest
if err := c.ShouldBindJSON(&req); err != nil {
appErr := errors.BadRequest(err.Error())
c.JSON(appErr.HTTPStatus, appErr)
return
}
svcReq := &service.UpdateBoardRequest{
Name: req.Name,
Description: req.Description,
}
board, err := h.service.UpdateBoard(c.Request.Context(), boardID, svcReq)
if err != nil {
h.logger.Error("failed to update board", zap.String("board_id", boardID), zap.Error(err))
appErr := errors.InternalError("failed to update board", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusOK, boardToResponse(board))
}
// DeleteBoard deletes a board
// DELETE /api/v1/boards/:boardId
func (h *Handler) DeleteBoard(c *gin.Context) {
boardID := c.Param("boardId")
if boardID == "" {
appErr := errors.BadRequest("boardId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
if err := h.service.DeleteBoard(c.Request.Context(), boardID); err != nil {
h.logger.Error("failed to delete board", zap.String("board_id", boardID), zap.Error(err))
appErr := errors.InternalError("failed to delete board", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.Status(http.StatusNoContent)
}
// ListBoards returns all boards
// GET /api/v1/boards
func (h *Handler) ListBoards(c *gin.Context) {
boards, err := h.service.ListBoards(c.Request.Context())
if err != nil {
h.logger.Error("failed to list boards", zap.Error(err))
appErr := errors.InternalError("failed to list boards", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
resp := BoardsListResponse{
Boards: make([]*BoardResponse, len(boards)),
Total: len(boards),
}
for i, b := range boards {
resp.Boards[i] = boardToResponse(b)
}
c.JSON(http.StatusOK, resp)
}
// Column endpoints
// CreateColumn creates a new column
// POST /api/v1/boards/:boardId/columns
func (h *Handler) CreateColumn(c *gin.Context) {
boardID := c.Param("boardId")
if boardID == "" {
appErr := errors.BadRequest("boardId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
var req CreateColumnRequest
if err := c.ShouldBindJSON(&req); err != nil {
appErr := errors.BadRequest(err.Error())
c.JSON(appErr.HTTPStatus, appErr)
return
}
svcReq := &service.CreateColumnRequest{
BoardID: boardID,
Name: req.Name,
Position: req.Position,
State: v1.TaskState(req.State),
}
column, err := h.service.CreateColumn(c.Request.Context(), svcReq)
if err != nil {
h.logger.Error("failed to create column", zap.String("board_id", boardID), zap.Error(err))
appErr := errors.InternalError("failed to create column", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusCreated, columnToResponse(column))
}
// ListColumns returns all columns for a board
// GET /api/v1/boards/:boardId/columns
func (h *Handler) ListColumns(c *gin.Context) {
boardID := c.Param("boardId")
if boardID == "" {
appErr := errors.BadRequest("boardId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
columns, err := h.service.ListColumns(c.Request.Context(), boardID)
if err != nil {
h.logger.Error("failed to list columns", zap.String("board_id", boardID), zap.Error(err))
appErr := errors.InternalError("failed to list columns", err)
c.JSON(appErr.HTTPStatus, appErr)
return
}
resp := ColumnsListResponse{
Columns: make([]*ColumnResponse, len(columns)),
Total: len(columns),
}
for i, col := range columns {
resp.Columns[i] = columnToResponse(col)
}
c.JSON(http.StatusOK, resp)
}
// GetColumn retrieves a column by ID
// GET /api/v1/columns/:columnId
func (h *Handler) GetColumn(c *gin.Context) {
columnID := c.Param("columnId")
if columnID == "" {
appErr := errors.BadRequest("columnId is required")
c.JSON(appErr.HTTPStatus, appErr)
return
}
column, err := h.service.GetColumn(c.Request.Context(), columnID)
if err != nil {
appErr := errors.NotFound("column", columnID)
c.JSON(appErr.HTTPStatus, appErr)
return
}
c.JSON(http.StatusOK, columnToResponse(column))
}
// Helper functions to convert models to response types
func taskToResponse(t *models.Task) *TaskResponse {
return &TaskResponse{
ID: t.ID,
BoardID: t.BoardID,
ColumnID: t.ColumnID,
Title: t.Title,
Description: t.Description,
State: string(t.State),
Priority: t.Priority,
AgentType: t.AgentType,
RepositoryURL: t.RepositoryURL,
Branch: t.Branch,
Position: t.Position,
CreatedAt: t.CreatedAt,
UpdatedAt: t.UpdatedAt,
Metadata: t.Metadata,
}
}
func boardToResponse(b *models.Board) *BoardResponse {
return &BoardResponse{
ID: b.ID,
Name: b.Name,
Description: b.Description,
CreatedAt: b.CreatedAt,
UpdatedAt: b.UpdatedAt,
}
}
func columnToResponse(c *models.Column) *ColumnResponse {
return &ColumnResponse{
ID: c.ID,
BoardID: c.BoardID,
Name: c.Name,
Position: c.Position,
State: string(c.State),
CreatedAt: c.CreatedAt,
}
}
apps/backend/internal/task/api/handlers_test.go
-486
View File
@@ -1,486 +0,0 @@
package api
import (
"bytes"
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
"github.com/gin-gonic/gin"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/events/bus"
"github.com/kandev/kandev/internal/task/models"
"github.com/kandev/kandev/internal/task/repository"
"github.com/kandev/kandev/internal/task/service"
v1 "github.com/kandev/kandev/pkg/api/v1"
)
// MockEventBus implements bus.EventBus for testing
type MockEventBus struct {
mu sync.Mutex
}
func NewMockEventBus() *MockEventBus {
return &MockEventBus{}
}
func (m *MockEventBus) Publish(ctx context.Context, subject string, event *bus.Event) error {
return nil
}
func (m *MockEventBus) Subscribe(subject string, handler bus.EventHandler) (bus.Subscription, error) {
return nil, nil
}
func (m *MockEventBus) QueueSubscribe(subject, queue string, handler bus.EventHandler) (bus.Subscription, error) {
return nil, nil
}
func (m *MockEventBus) Request(ctx context.Context, subject string, event *bus.Event, timeout time.Duration) (*bus.Event, error) {
return nil, nil
}
func (m *MockEventBus) Close() {}
func (m *MockEventBus) IsConnected() bool {
return true
}
func setupTestHandler(t *testing.T) (*Handler, *repository.MemoryRepository, *gin.Engine) {
t.Helper()
gin.SetMode(gin.TestMode)
repo := repository.NewMemoryRepository()
eventBus := NewMockEventBus()
log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json", OutputPath: "stdout"})
svc := service.NewService(repo, eventBus, log)
handler := NewHandler(svc, log)
router := gin.New()
return handler, repo, router
}
// Task handler tests
func TestHandler_CreateTask(t *testing.T) {
handler, repo, router := setupTestHandler(t)
ctx := context.Background()
// Create board and column first
board := &models.Board{ID: "board-123", Name: "Test Board"}
_ = repo.CreateBoard(ctx, board)
column := &models.Column{ID: "col-123", BoardID: "board-123", Name: "To Do", State: v1.TaskStateTODO}
_ = repo.CreateColumn(ctx, column)
router.POST("/tasks", handler.CreateTask)
body := CreateTaskRequest{
BoardID: "board-123",
ColumnID: "col-123",
Title: "Test Task",
Description: "A test task",
Priority: 5,
}
jsonBody, _ := json.Marshal(body)
req := httptest.NewRequest(http.MethodPost, "/tasks", bytes.NewBuffer(jsonBody))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusCreated {
t.Errorf("expected status 201, got %d: %s", w.Code, w.Body.String())
}
var resp TaskResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to unmarshal response: %v", err)
}
if resp.Title != "Test Task" {
t.Errorf("expected title 'Test Task', got %s", resp.Title)
}
}
func TestHandler_GetTask(t *testing.T) {
handler, repo, router := setupTestHandler(t)
ctx := context.Background()
task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"}
_ = repo.CreateTask(ctx, task)
router.GET("/tasks/:taskId", handler.GetTask)
req := httptest.NewRequest(http.MethodGet, "/tasks/task-123", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status 200, got %d: %s", w.Code, w.Body.String())
}
var resp TaskResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to unmarshal response: %v", err)
}
if resp.Title != "Test Task" {
t.Errorf("expected title 'Test Task', got %s", resp.Title)
}
}
func TestHandler_GetTaskNotFound(t *testing.T) {
handler, _, router := setupTestHandler(t)
router.GET("/tasks/:taskId", handler.GetTask)
req := httptest.NewRequest(http.MethodGet, "/tasks/nonexistent", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusNotFound {
t.Errorf("expected status 404, got %d", w.Code)
}
}
func TestHandler_UpdateTask(t *testing.T) {
handler, repo, router := setupTestHandler(t)
ctx := context.Background()
task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Original"}
_ = repo.CreateTask(ctx, task)
router.PUT("/tasks/:taskId", handler.UpdateTask)
body := UpdateTaskRequest{Title: stringPtr("Updated Title")}
jsonBody, _ := json.Marshal(body)
req := httptest.NewRequest(http.MethodPut, "/tasks/task-123", bytes.NewBuffer(jsonBody))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status 200, got %d: %s", w.Code, w.Body.String())
}
var resp TaskResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to unmarshal response: %v", err)
}
if resp.Title != "Updated Title" {
t.Errorf("expected title 'Updated Title', got %s", resp.Title)
}
}
func TestHandler_DeleteTask(t *testing.T) {
handler, repo, router := setupTestHandler(t)
ctx := context.Background()
task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Test"}
_ = repo.CreateTask(ctx, task)
router.DELETE("/tasks/:taskId", handler.DeleteTask)
req := httptest.NewRequest(http.MethodDelete, "/tasks/task-123", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusNoContent {
t.Errorf("expected status 204, got %d", w.Code)
}
// Verify task is deleted
_, err := repo.GetTask(ctx, "task-123")
if err == nil {
t.Error("expected task to be deleted")
}
}
func TestHandler_ListTasks(t *testing.T) {
handler, repo, router := setupTestHandler(t)
ctx := context.Background()
_ = repo.CreateTask(ctx, &models.Task{ID: "task-1", BoardID: "board-123", ColumnID: "col-123", Title: "Task 1"})
_ = repo.CreateTask(ctx, &models.Task{ID: "task-2", BoardID: "board-123", ColumnID: "col-123", Title: "Task 2"})
router.GET("/boards/:boardId/tasks", handler.ListTasks)
req := httptest.NewRequest(http.MethodGet, "/boards/board-123/tasks", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status 200, got %d: %s", w.Code, w.Body.String())
}
var resp TasksListResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to unmarshal response: %v", err)
}
if len(resp.Tasks) != 2 {
t.Errorf("expected 2 tasks, got %d", len(resp.Tasks))
}
}
// Board handler tests
func TestHandler_CreateBoard(t *testing.T) {
handler, _, router := setupTestHandler(t)
router.POST("/boards", handler.CreateBoard)
body := CreateBoardRequest{
Name: "Test Board",
Description: "A test board",
}
jsonBody, _ := json.Marshal(body)
req := httptest.NewRequest(http.MethodPost, "/boards", bytes.NewBuffer(jsonBody))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusCreated {
t.Errorf("expected status 201, got %d: %s", w.Code, w.Body.String())
}
var resp BoardResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to unmarshal response: %v", err)
}
if resp.Name != "Test Board" {
t.Errorf("expected name 'Test Board', got %s", resp.Name)
}
}
func TestHandler_GetBoard(t *testing.T) {
handler, repo, router := setupTestHandler(t)
ctx := context.Background()
board := &models.Board{ID: "board-123", Name: "Test Board"}
_ = repo.CreateBoard(ctx, board)
router.GET("/boards/:boardId", handler.GetBoard)
req := httptest.NewRequest(http.MethodGet, "/boards/board-123", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status 200, got %d: %s", w.Code, w.Body.String())
}
var resp BoardResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to unmarshal response: %v", err)
}
if resp.Name != "Test Board" {
t.Errorf("expected name 'Test Board', got %s", resp.Name)
}
}
func TestHandler_GetBoardNotFound(t *testing.T) {
handler, _, router := setupTestHandler(t)
router.GET("/boards/:boardId", handler.GetBoard)
req := httptest.NewRequest(http.MethodGet, "/boards/nonexistent", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusNotFound {
t.Errorf("expected status 404, got %d", w.Code)
}
}
func TestHandler_UpdateBoard(t *testing.T) {
handler, repo, router := setupTestHandler(t)
ctx := context.Background()
board := &models.Board{ID: "board-123", Name: "Original"}
_ = repo.CreateBoard(ctx, board)
router.PUT("/boards/:boardId", handler.UpdateBoard)
body := UpdateBoardRequest{Name: stringPtr("Updated")}
jsonBody, _ := json.Marshal(body)
req := httptest.NewRequest(http.MethodPut, "/boards/board-123", bytes.NewBuffer(jsonBody))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status 200, got %d: %s", w.Code, w.Body.String())
}
var resp BoardResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to unmarshal response: %v", err)
}
if resp.Name != "Updated" {
t.Errorf("expected name 'Updated', got %s", resp.Name)
}
}
func TestHandler_DeleteBoard(t *testing.T) {
handler, repo, router := setupTestHandler(t)
ctx := context.Background()
board := &models.Board{ID: "board-123", Name: "Test Board"}
_ = repo.CreateBoard(ctx, board)
router.DELETE("/boards/:boardId", handler.DeleteBoard)
req := httptest.NewRequest(http.MethodDelete, "/boards/board-123", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusNoContent {
t.Errorf("expected status 204, got %d", w.Code)
}
// Verify board is deleted
_, err := repo.GetBoard(ctx, "board-123")
if err == nil {
t.Error("expected board to be deleted")
}
}
func TestHandler_ListBoards(t *testing.T) {
handler, repo, router := setupTestHandler(t)
ctx := context.Background()
_ = repo.CreateBoard(ctx, &models.Board{ID: "board-1", Name: "Board 1"})
_ = repo.CreateBoard(ctx, &models.Board{ID: "board-2", Name: "Board 2"})
router.GET("/boards", handler.ListBoards)
req := httptest.NewRequest(http.MethodGet, "/boards", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status 200, got %d: %s", w.Code, w.Body.String())
}
var resp BoardsListResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to unmarshal response: %v", err)
}
if len(resp.Boards) != 2 {
t.Errorf("expected 2 boards, got %d", len(resp.Boards))
}
}
// Column handler tests
func TestHandler_CreateColumn(t *testing.T) {
handler, repo, router := setupTestHandler(t)
ctx := context.Background()
board := &models.Board{ID: "board-123", Name: "Test Board"}
_ = repo.CreateBoard(ctx, board)
router.POST("/boards/:boardId/columns", handler.CreateColumn)
body := CreateColumnRequest{
Name: "To Do",
Position: 0,
State: "TODO",
}
jsonBody, _ := json.Marshal(body)
req := httptest.NewRequest(http.MethodPost, "/boards/board-123/columns", bytes.NewBuffer(jsonBody))
req.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusCreated {
t.Errorf("expected status 201, got %d: %s", w.Code, w.Body.String())
}
var resp ColumnResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to unmarshal response: %v", err)
}
if resp.Name != "To Do" {
t.Errorf("expected name 'To Do', got %s", resp.Name)
}
}
func TestHandler_GetColumn(t *testing.T) {
handler, repo, router := setupTestHandler(t)
ctx := context.Background()
column := &models.Column{ID: "col-123", BoardID: "board-123", Name: "To Do"}
_ = repo.CreateColumn(ctx, column)
router.GET("/columns/:columnId", handler.GetColumn)
req := httptest.NewRequest(http.MethodGet, "/columns/col-123", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status 200, got %d: %s", w.Code, w.Body.String())
}
var resp ColumnResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to unmarshal response: %v", err)
}
if resp.Name != "To Do" {
t.Errorf("expected name 'To Do', got %s", resp.Name)
}
}
func TestHandler_ListColumns(t *testing.T) {
handler, repo, router := setupTestHandler(t)
ctx := context.Background()
_ = repo.CreateColumn(ctx, &models.Column{ID: "col-1", BoardID: "board-123", Name: "To Do"})
_ = repo.CreateColumn(ctx, &models.Column{ID: "col-2", BoardID: "board-123", Name: "Done"})
router.GET("/boards/:boardId/columns", handler.ListColumns)
req := httptest.NewRequest(http.MethodGet, "/boards/board-123/columns", nil)
w := httptest.NewRecorder()
router.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("expected status 200, got %d: %s", w.Code, w.Body.String())
}
var resp ColumnsListResponse
if err := json.Unmarshal(w.Body.Bytes(), &resp); err != nil {
t.Fatalf("failed to unmarshal response: %v", err)
}
if len(resp.Columns) != 2 {
t.Errorf("expected 2 columns, got %d", len(resp.Columns))
}
}
// Helper function
func stringPtr(s string) *string {
return &s
}
apps/backend/internal/task/api/requests.go
-114
View File
@@ -1,114 +0,0 @@
// Package api provides HTTP handlers for the task service API.
package api
import "time"
// CreateTaskRequest for creating a task
type CreateTaskRequest struct {
BoardID string `json:"board_id" binding:"required"`
ColumnID string `json:"column_id" binding:"required"`
Title string `json:"title" binding:"required"`
Description string `json:"description"`
Priority int `json:"priority"`
AgentType string `json:"agent_type,omitempty"`
RepositoryURL string `json:"repository_url,omitempty"`
Branch string `json:"branch,omitempty"`
Metadata map[string]interface{} `json:"metadata,omitempty"`
}
// UpdateTaskRequest for updating a task
type UpdateTaskRequest struct {
Title *string `json:"title,omitempty"`
Description *string `json:"description,omitempty"`
Priority *int `json:"priority,omitempty"`
AgentType *string `json:"agent_type,omitempty"`
Metadata map[string]interface{} `json:"metadata,omitempty"`
}
// UpdateTaskStateRequest for changing task state
type UpdateTaskStateRequest struct {
State string `json:"state" binding:"required"`
}
// MoveTaskRequest for moving a task to a different column
type MoveTaskRequest struct {
ColumnID string `json:"column_id" binding:"required"`
Position int `json:"position"`
}
// CreateBoardRequest for creating a board
type CreateBoardRequest struct {
Name string `json:"name" binding:"required"`
Description string `json:"description"`
}
// UpdateBoardRequest for updating a board
type UpdateBoardRequest struct {
Name *string `json:"name,omitempty"`
Description *string `json:"description,omitempty"`
}
// CreateColumnRequest for creating a column
type CreateColumnRequest struct {
Name string `json:"name" binding:"required"`
Position int `json:"position"`
State string `json:"state"` // Maps to task state: TODO, IN_PROGRESS, DONE
}
// Response types
// TaskResponse represents a task in API responses
type TaskResponse struct {
ID string `json:"id"`
BoardID string `json:"board_id"`
ColumnID string `json:"column_id"`
Title string `json:"title"`
Description string `json:"description"`
State string `json:"state"`
Priority int `json:"priority"`
AgentType string `json:"agent_type,omitempty"`
RepositoryURL string `json:"repository_url,omitempty"`
Branch string `json:"branch,omitempty"`
Position int `json:"position"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
Metadata map[string]interface{} `json:"metadata,omitempty"`
}
// BoardResponse represents a board in API responses
type BoardResponse struct {
ID string `json:"id"`
Name string `json:"name"`
Description string `json:"description"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
// ColumnResponse represents a column in API responses
type ColumnResponse struct {
ID string `json:"id"`
BoardID string `json:"board_id"`
Name string `json:"name"`
Position int `json:"position"`
State string `json:"state"`
CreatedAt time.Time `json:"created_at"`
}
// TasksListResponse for listing tasks
type TasksListResponse struct {
Tasks []*TaskResponse `json:"tasks"`
Total int `json:"total"`
}
// BoardsListResponse for listing boards
type BoardsListResponse struct {
Boards []*BoardResponse `json:"boards"`
Total int `json:"total"`
}
// ColumnsListResponse for listing columns
type ColumnsListResponse struct {
Columns []*ColumnResponse `json:"columns"`
Total int `json:"total"`
}
apps/backend/internal/task/api/router.go
-46
View File
@@ -1,46 +0,0 @@
package api
import (
"github.com/gin-gonic/gin"
"github.com/kandev/kandev/internal/common/logger"
"github.com/kandev/kandev/internal/task/service"
)
// SetupRoutes configures the task API routes
func SetupRoutes(router *gin.RouterGroup, svc *service.Service, log *logger.Logger) {
handler := NewHandler(svc, log)
// Board routes
boards := router.Group("/boards")
{
boards.POST("", handler.CreateBoard)
boards.GET("", handler.ListBoards)
boards.GET("/:boardId", handler.GetBoard)
boards.PUT("/:boardId", handler.UpdateBoard)
boards.DELETE("/:boardId", handler.DeleteBoard)
// Board sub-resources
boards.GET("/:boardId/tasks", handler.ListTasks)
boards.POST("/:boardId/columns", handler.CreateColumn)
boards.GET("/:boardId/columns", handler.ListColumns)
}
// Task routes
tasks := router.Group("/tasks")
{
tasks.POST("", handler.CreateTask)
tasks.GET("/:taskId", handler.GetTask)
tasks.PUT("/:taskId", handler.UpdateTask)
tasks.DELETE("/:taskId", handler.DeleteTask)
tasks.PUT("/:taskId/state", handler.UpdateTaskState)
tasks.PUT("/:taskId/move", handler.MoveTask)
}
// Column routes
columns := router.Group("/columns")
{
columns.GET("/:columnId", handler.GetColumn)
}
}
apps/backend/kandev
BIN
View File
Binary file not shown.
apps/backend/pkg/acp/client/.gitkeep
View File
apps/backend/pkg/acp/server/.gitkeep
View File
deployments/docker/agents/.gitkeep
View File
deployments/docker/compose/.gitkeep
View File
migrations/postgres/.gitkeep
View File