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

chore: Remove dead code and consolidate agentctl modes (#17)

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