diff --git a/apps/backend/cmd/agentctl/main.go b/apps/backend/cmd/agentctl/main.go index b68b330069..a3f3044648 100644 --- a/apps/backend/cmd/agentctl/main.go +++ b/apps/backend/cmd/agentctl/main.go @@ -9,6 +9,7 @@ package main import ( "context" + "flag" "fmt" "net/http" "os" @@ -21,13 +22,42 @@ import ( "github.com/kandev/kandev/internal/agentctl/instance" "github.com/kandev/kandev/internal/agentctl/process" "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/pkg/agent" "go.uber.org/zap" ) +// Command-line flags +var ( + protocolFlag = flag.String("protocol", "", "Protocol for agent communication (acp, rest, mcp). Default: acp") + agentCommandFlag = flag.String("agent-command", "", "Command to run the agent") + workDirFlag = flag.String("workdir", "", "Working directory for the agent") + portFlag = flag.Int("port", 0, "HTTP server port") + modeFlag = flag.String("mode", "", "Operating mode (docker, standalone, auto)") +) + func main() { - // Load multi-instance configuration + flag.Parse() + + // Load multi-instance configuration (env vars as defaults, flags override) cfg := config.LoadMulti() + // Override with CLI flags if provided + if *protocolFlag != "" { + cfg.DefaultProtocol = agent.Protocol(*protocolFlag) + } + if *agentCommandFlag != "" { + cfg.DefaultAgentCommand = *agentCommandFlag + } + if *workDirFlag != "" { + cfg.DefaultWorkDir = *workDirFlag + } + if *portFlag != 0 { + cfg.ControlPort = *portFlag + } + if *modeFlag != "" { + cfg.Mode = *modeFlag + } + // Initialize logger log, err := logger.NewLogger(logger.LoggingConfig{ Level: cfg.LogLevel, @@ -104,6 +134,9 @@ func runDockerMode(cfg *config.MultiConfig, log *logger.Logger) { singleCfg := config.Load() // Override with multi-config values if they differ from defaults singleCfg.Port = cfg.ControlPort + if cfg.DefaultProtocol != "" { + singleCfg.Protocol = cfg.DefaultProtocol + } if cfg.DefaultAgentCommand != "" { singleCfg.AgentCommand = cfg.DefaultAgentCommand singleCfg.AgentArgs = parseCommand(cfg.DefaultAgentCommand) @@ -117,6 +150,7 @@ func runDockerMode(cfg *config.MultiConfig, log *logger.Logger) { log.Info("running in Docker mode (single instance)", zap.Int("port", singleCfg.Port), + zap.String("protocol", string(singleCfg.Protocol)), zap.String("agent_command", singleCfg.AgentCommand), zap.String("workdir", singleCfg.WorkDir), zap.Bool("auto_approve_permissions", singleCfg.AutoApprovePermissions)) diff --git a/apps/backend/internal/agent/agentctl/acp.go b/apps/backend/internal/agent/agentctl/acp.go index b2a337c9e1..eeef7d61c1 100644 --- a/apps/backend/internal/agent/agentctl/acp.go +++ b/apps/backend/internal/agent/agentctl/acp.go @@ -13,15 +13,21 @@ import ( "go.uber.org/zap" ) +// AgentInfo contains information about the connected agent. +type AgentInfo struct { + Name string `json:"name"` + Version string `json:"version"` +} + // InitializeResponse from agentctl type InitializeResponse struct { - Success bool `json:"success"` - Response *acp.InitializeResponse `json:"response,omitempty"` - Error string `json:"error,omitempty"` + Success bool `json:"success"` + AgentInfo *AgentInfo `json:"agent_info,omitempty"` + Error string `json:"error,omitempty"` } // Initialize sends the ACP initialize request -func (c *Client) Initialize(ctx context.Context, clientName, clientVersion string) (*acp.InitializeResponse, error) { +func (c *Client) Initialize(ctx context.Context, clientName, clientVersion string) (*AgentInfo, error) { reqBody := struct { ClientName string `json:"client_name"` ClientVersion string `json:"client_version"` @@ -54,7 +60,7 @@ func (c *Client) Initialize(ctx context.Context, clientName, clientVersion strin if !result.Success { return nil, fmt.Errorf("initialize failed: %s", result.Error) } - return result.Response, nil + return result.AgentInfo, nil } // NewSessionResponse from agentctl @@ -143,8 +149,21 @@ func (c *Client) Prompt(ctx context.Context, text string) (*PromptResponse, erro return &result, nil } -// SessionNotification represents a session update from the agent -type SessionNotification = acp.SessionNotification +// SessionUpdate represents a session update from the agent. +// This matches adapter.SessionUpdate that agentctl sends over WebSocket. +type SessionUpdate struct { + Type string `json:"type"` + SessionID string `json:"session_id,omitempty"` + Text string `json:"text,omitempty"` + ToolCallID string `json:"tool_call_id,omitempty"` + ToolName string `json:"tool_name,omitempty"` + ToolTitle string `json:"tool_title,omitempty"` + ToolStatus string `json:"tool_status,omitempty"` + ToolArgs map[string]interface{} `json:"tool_args,omitempty"` + ToolResult interface{} `json:"tool_result,omitempty"` + Error string `json:"error,omitempty"` + Data map[string]interface{} `json:"data,omitempty"` +} // PermissionOption represents a permission choice type PermissionOption struct { @@ -177,7 +196,7 @@ type PermissionRespondResponse struct { } // StreamUpdates opens a WebSocket connection for streaming session updates -func (c *Client) StreamUpdates(ctx context.Context, handler func(acp.SessionNotification)) error { +func (c *Client) StreamUpdates(ctx context.Context, handler func(SessionUpdate)) error { wsURL := "ws" + c.baseURL[4:] + "/api/v1/acp/stream" conn, _, err := websocket.DefaultDialer.DialContext(ctx, wsURL, nil) @@ -211,13 +230,13 @@ func (c *Client) StreamUpdates(ctx context.Context, handler func(acp.SessionNoti return } - var notification acp.SessionNotification - if err := json.Unmarshal(message, ¬ification); err != nil { - c.logger.Warn("failed to parse session notification", zap.Error(err)) + var update SessionUpdate + if err := json.Unmarshal(message, &update); err != nil { + c.logger.Warn("failed to parse session update", zap.Error(err)) continue } - handler(notification) + handler(update) } }() diff --git a/apps/backend/internal/agent/agentctl/launcher/launcher.go b/apps/backend/internal/agent/agentctl/launcher/launcher.go index 4ab241ec76..c8144907f3 100644 --- a/apps/backend/internal/agent/agentctl/launcher/launcher.go +++ b/apps/backend/internal/agent/agentctl/launcher/launcher.go @@ -123,8 +123,8 @@ func (l *Launcher) Start(ctx context.Context) error { // shutdown ourselves via Stop(). CommandContext sends SIGKILL on context // cancellation which prevents graceful shutdown. l.cmd = exec.Command(l.binaryPath, - "--mode=standalone", - fmt.Sprintf("--control-port=%d", l.port), + "-mode=standalone", + fmt.Sprintf("-port=%d", l.port), ) // Set environment variables (inherit from parent + add agentctl-specific ones) diff --git a/apps/backend/internal/agent/agentctl/standalone.go b/apps/backend/internal/agent/agentctl/standalone.go index ac1845ebcb..4894cee0fd 100644 --- a/apps/backend/internal/agent/agentctl/standalone.go +++ b/apps/backend/internal/agent/agentctl/standalone.go @@ -27,6 +27,7 @@ type CreateInstanceRequest struct { ID string `json:"id,omitempty"` WorkspacePath string `json:"workspace_path"` AgentCommand string `json:"agent_command,omitempty"` + Protocol string `json:"protocol,omitempty"` // Protocol adapter to use (acp, rest, mcp) Env map[string]string `json:"env,omitempty"` AutoStart bool `json:"auto_start,omitempty"` } diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index 4f534a386d..fa7bbabaaa 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -11,7 +11,6 @@ import ( "sync" "time" - acp "github.com/coder/acp-go-sdk" "github.com/google/uuid" "go.uber.org/zap" @@ -584,6 +583,7 @@ func (m *Manager) launchStandalone(ctx context.Context, instanceID string, req * ID: instanceID, WorkspacePath: req.WorkspacePath, AgentCommand: agentCommand, + Protocol: string(agentConfig.Protocol), // Pass protocol from registry Env: env, AutoStart: false, // We'll start via ACP initialize flow } @@ -768,17 +768,23 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta m.logger.Info("sending ACP initialize request", zap.String("instance_id", instance.ID)) - initResp, err := instance.agentctl.Initialize(ctx, "kandev", "1.0.0") + agentInfo, err := instance.agentctl.Initialize(ctx, "kandev", "1.0.0") if err != nil { m.logger.Error("ACP initialize failed", zap.String("instance_id", instance.ID), zap.Error(err)) return fmt.Errorf("initialize failed: %w", err) } + agentName := "unknown" + agentVersion := "unknown" + if agentInfo != nil { + agentName = agentInfo.Name + agentVersion = agentInfo.Version + } m.logger.Info("ACP initialize response received", zap.String("instance_id", instance.ID), - zap.String("agent_name", initResp.AgentInfo.Name), - zap.String("agent_version", initResp.AgentInfo.Version)) + zap.String("agent_name", agentName), + zap.String("agent_version", agentVersion)) // 2. Create new session m.logger.Info("sending ACP session/new request", @@ -888,8 +894,8 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta func (m *Manager) handleUpdatesStream(instance *AgentInstance) { ctx := context.Background() - err := instance.agentctl.StreamUpdates(ctx, func(notification agentctl.SessionNotification) { - m.handleSessionNotification(instance, notification) + err := instance.agentctl.StreamUpdates(ctx, func(update agentctl.SessionUpdate) { + m.handleSessionUpdate(instance, update) }) if err != nil { m.logger.Error("failed to connect to updates stream", @@ -1032,139 +1038,131 @@ func (m *Manager) publishPermissionRequest(instance *AgentInstance, notification } } -// handleSessionNotification processes incoming session notifications from the agent -func (m *Manager) handleSessionNotification(instance *AgentInstance, notification agentctl.SessionNotification) { - m.logger.Debug("received session notification", +// handleSessionUpdate processes incoming session updates from the agent +func (m *Manager) handleSessionUpdate(instance *AgentInstance, update agentctl.SessionUpdate) { + m.logger.Debug("received session update", zap.String("instance_id", instance.ID), - zap.String("session_id", string(notification.SessionId))) + zap.String("session_id", update.SessionID), + zap.String("type", update.Type)) - update := notification.Update - - // Handle different update types - if update.AgentMessageChunk != nil { + // Handle different update types based on the Type field + switch update.Type { + case "message_chunk": m.logger.Debug("agent message chunk", zap.String("instance_id", instance.ID)) m.updateInstanceProgress(instance.ID, 50) // Accumulate message content for saving as comment when a step completes - if update.AgentMessageChunk.Content.Text != nil { + if update.Text != "" { instance.messageMu.Lock() - instance.messageBuffer.WriteString(update.AgentMessageChunk.Content.Text.Text) + instance.messageBuffer.WriteString(update.Text) instance.messageMu.Unlock() } - } - if update.AgentThoughtChunk != nil { - m.logger.Debug("agent thought chunk", - zap.String("instance_id", instance.ID)) - } - - if update.ToolCall != nil { + case "tool_call": // Tool call starting marks a step boundary - flush the accumulated message as a comment // This way, each agent response before a tool use becomes a separate comment m.flushMessageBufferAsComment(instance) - toolCallID := string(update.ToolCall.ToolCallId) - toolTitle := update.ToolCall.Title - toolStatus := string(update.ToolCall.Status) - toolKind := string(update.ToolCall.Kind) - m.logger.Info("tool call started", zap.String("instance_id", instance.ID), - zap.String("tool_call_id", toolCallID), - zap.String("title", toolTitle), - zap.String("kind", toolKind)) + zap.String("tool_call_id", update.ToolCallID), + zap.String("title", update.ToolTitle), + zap.String("tool_name", update.ToolName)) m.updateInstanceProgress(instance.ID, 60) - // Extract args from the tool call for display - args := map[string]interface{}{} - - // Add tool kind - if toolKind != "" { - args["kind"] = toolKind - } - - // Add locations (file paths) - if len(update.ToolCall.Locations) > 0 { - locations := make([]map[string]interface{}, len(update.ToolCall.Locations)) - for i, loc := range update.ToolCall.Locations { - locMap := map[string]interface{}{"path": loc.Path} - if loc.Line != nil { - locMap["line"] = *loc.Line - } - locations[i] = locMap - } - args["locations"] = locations - - // Also set primary path for convenience - args["path"] = update.ToolCall.Locations[0].Path - } - - // Add raw input if available - if update.ToolCall.RawInput != nil { - args["raw_input"] = update.ToolCall.RawInput - } - // Publish tool call as a comment so it appears in the chat - m.publishToolCall(instance, toolCallID, toolTitle, toolStatus, args) - } - - if update.ToolCallUpdate != nil { - toolCallID := string(update.ToolCallUpdate.ToolCallId) + m.publishToolCall(instance, update.ToolCallID, update.ToolTitle, update.ToolStatus, update.ToolArgs) + case "tool_update": m.logger.Debug("tool call update", zap.String("instance_id", instance.ID), - zap.String("tool_call_id", toolCallID)) + zap.String("tool_call_id", update.ToolCallID)) // Check if tool call completed - if update.ToolCallUpdate.Status != nil { - status := string(*update.ToolCallUpdate.Status) - switch status { - case "complete": - m.logger.Info("tool call completed", - zap.String("instance_id", instance.ID)) - m.updateInstanceProgress(instance.ID, 80) - // Publish tool call completion - m.publishToolCallComplete(instance, toolCallID, update.ToolCallUpdate) - case "error": - m.logger.Error("tool call error", - zap.String("instance_id", instance.ID)) - m.publishToolCallComplete(instance, toolCallID, update.ToolCallUpdate) - } + switch update.ToolStatus { + case "complete", "completed": + m.logger.Info("tool call completed", + zap.String("instance_id", instance.ID)) + m.updateInstanceProgress(instance.ID, 80) + m.publishToolCallCompleteFromUpdate(instance, update) + case "error": + m.logger.Error("tool call error", + zap.String("instance_id", instance.ID)) + m.publishToolCallCompleteFromUpdate(instance, update) } - } - if update.Plan != nil { + case "plan": m.logger.Info("agent plan update", + zap.String("instance_id", instance.ID)) + + case "error": + m.logger.Error("agent error", zap.String("instance_id", instance.ID), - zap.Int("num_entries", len(update.Plan.Entries))) + zap.String("error", update.Error)) } - // Publish session notification to event bus for WebSocket streaming - m.publishSessionNotification(instance, notification) + // Publish session update to event bus for WebSocket streaming + m.publishSessionUpdate(instance, update) } -// publishSessionNotification publishes a session notification to the event bus -func (m *Manager) publishSessionNotification(instance *AgentInstance, notification agentctl.SessionNotification) { +// publishSessionUpdate publishes a session update to the event bus +func (m *Manager) publishSessionUpdate(instance *AgentInstance, update agentctl.SessionUpdate) { if m.eventBus == nil { return } - // Build ACP message data from the session notification + // Build the update data - our SessionUpdate type marshals cleanly + updateData := map[string]interface{}{ + "type": update.Type, + } + + if update.SessionID != "" { + updateData["session_id"] = update.SessionID + } + if update.Text != "" { + updateData["text"] = update.Text + } + if update.ToolCallID != "" { + updateData["tool_call_id"] = update.ToolCallID + } + if update.ToolName != "" { + updateData["tool_name"] = update.ToolName + } + if update.ToolTitle != "" { + updateData["tool_title"] = update.ToolTitle + } + if update.ToolStatus != "" { + updateData["tool_status"] = update.ToolStatus + } + if update.ToolArgs != nil { + updateData["tool_args"] = update.ToolArgs + } + if update.ToolResult != nil { + updateData["tool_result"] = update.ToolResult + } + if update.Error != "" { + updateData["error"] = update.Error + } + if update.Data != nil { + updateData["data"] = update.Data + } + + // Build ACP message data data := map[string]interface{}{ - "type": "session/update", - "timestamp": time.Now(), - "agent_id": instance.ID, - "task_id": instance.TaskID, - "session_id": string(notification.SessionId), - "data": notification.Update, + "type": "session/update", + "timestamp": time.Now().UTC().Format(time.RFC3339Nano), + "agent_id": instance.ID, + "task_id": instance.TaskID, + "session_id": update.SessionID, + "data": updateData, } event := bus.NewEvent(events.ACPMessage, "agent-manager", data) subject := events.BuildACPSubject(instance.TaskID) if err := m.eventBus.Publish(context.Background(), subject, event); err != nil { - m.logger.Error("failed to publish session notification", + m.logger.Error("failed to publish session update", zap.String("instance_id", instance.ID), zap.String("task_id", instance.TaskID), zap.Error(err)) @@ -1415,24 +1413,19 @@ func (m *Manager) publishToolCall(instance *AgentInstance, toolCallID, title, st } } -// publishToolCallComplete publishes a tool call completion event -func (m *Manager) publishToolCallComplete(instance *AgentInstance, toolCallID string, update *acp.SessionToolCallUpdate) { +// publishToolCallCompleteFromUpdate publishes a tool call completion event from a SessionUpdate +func (m *Manager) publishToolCallCompleteFromUpdate(instance *AgentInstance, update agentctl.SessionUpdate) { if m.eventBus == nil { return } - status := "" - if update.Status != nil { - status = string(*update.Status) - } - data := map[string]interface{}{ "type": "tool_call_complete", "timestamp": time.Now().UTC().Format(time.RFC3339Nano), "agent_id": instance.ID, "task_id": instance.TaskID, - "tool_call_id": toolCallID, - "status": status, + "tool_call_id": update.ToolCallID, + "status": update.ToolStatus, } event := bus.NewEvent(events.ToolCallComplete, "agent-manager", data) @@ -1548,6 +1541,11 @@ func (m *Manager) buildContainerConfig(instanceID string, req *LaunchRequest, ag fmt.Sprintf("KANDEV_INSTANCE_ID=%s", instanceID), ) + // Pass protocol to agentctl inside the container + if agentConfig.Protocol != "" { + env = append(env, fmt.Sprintf("AGENTCTL_PROTOCOL=%s", agentConfig.Protocol)) + } + // Configure Git to trust the workspace directory // This is needed because the container runs as root but files are owned by host user // Uses Git's environment-based config (GIT_CONFIG_COUNT, GIT_CONFIG_KEY_n, GIT_CONFIG_VALUE_n) diff --git a/apps/backend/internal/agent/lifecycle/profile_resolver_test.go b/apps/backend/internal/agent/lifecycle/profile_resolver_test.go new file mode 100644 index 0000000000..4cd295f679 --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/profile_resolver_test.go @@ -0,0 +1,217 @@ +package lifecycle + +import ( + "context" + "errors" + "testing" + + "github.com/kandev/kandev/internal/agent/settings/models" + "github.com/kandev/kandev/internal/agent/settings/store" +) + +// MockRepository implements store.Repository for testing +type MockRepository struct { + GetAgentFn func(ctx context.Context, id string) (*models.Agent, error) + GetAgentByNameFn func(ctx context.Context, name string) (*models.Agent, error) + GetAgentProfileFn func(ctx context.Context, id string) (*models.AgentProfile, error) + ListAgentsFn func(ctx context.Context) ([]*models.Agent, error) + ListAgentProfilesFn func(ctx context.Context, agentID string) ([]*models.AgentProfile, error) +} + +var _ store.Repository = (*MockRepository)(nil) + +func (m *MockRepository) CreateAgent(ctx context.Context, agent *models.Agent) error { + return nil +} + +func (m *MockRepository) GetAgent(ctx context.Context, id string) (*models.Agent, error) { + if m.GetAgentFn != nil { + return m.GetAgentFn(ctx, id) + } + return nil, errors.New("agent not found") +} + +func (m *MockRepository) GetAgentByName(ctx context.Context, name string) (*models.Agent, error) { + if m.GetAgentByNameFn != nil { + return m.GetAgentByNameFn(ctx, name) + } + return nil, errors.New("agent not found") +} + +func (m *MockRepository) UpdateAgent(ctx context.Context, agent *models.Agent) error { + return nil +} + +func (m *MockRepository) DeleteAgent(ctx context.Context, id string) error { + return nil +} + +func (m *MockRepository) ListAgents(ctx context.Context) ([]*models.Agent, error) { + if m.ListAgentsFn != nil { + return m.ListAgentsFn(ctx) + } + return []*models.Agent{}, nil +} + +func (m *MockRepository) CreateAgentProfile(ctx context.Context, profile *models.AgentProfile) error { + return nil +} + +func (m *MockRepository) UpdateAgentProfile(ctx context.Context, profile *models.AgentProfile) error { + return nil +} + +func (m *MockRepository) DeleteAgentProfile(ctx context.Context, id string) error { + return nil +} + +func (m *MockRepository) GetAgentProfile(ctx context.Context, id string) (*models.AgentProfile, error) { + if m.GetAgentProfileFn != nil { + return m.GetAgentProfileFn(ctx, id) + } + return nil, errors.New("profile not found") +} + +func (m *MockRepository) ListAgentProfiles(ctx context.Context, agentID string) ([]*models.AgentProfile, error) { + if m.ListAgentProfilesFn != nil { + return m.ListAgentProfilesFn(ctx, agentID) + } + return []*models.AgentProfile{}, nil +} + +func (m *MockRepository) Close() error { + return nil +} + +func TestNewStoreProfileResolver(t *testing.T) { + mockRepo := &MockRepository{} + + resolver := NewStoreProfileResolver(mockRepo) + + if resolver == nil { + t.Fatal("expected non-nil resolver") + } + if resolver.store != mockRepo { + t.Error("expected resolver to use the provided store") + } +} + +func TestStoreProfileResolver_ResolveProfile_Success(t *testing.T) { + mockRepo := &MockRepository{ + GetAgentProfileFn: func(ctx context.Context, id string) (*models.AgentProfile, error) { + return &models.AgentProfile{ + ID: "profile-123", + AgentID: "agent-456", + Name: "My Profile", + Model: "claude-3.5-sonnet", + AutoApprove: true, + DangerouslySkipPermissions: false, + Plan: "pro", + }, nil + }, + GetAgentFn: func(ctx context.Context, id string) (*models.Agent, error) { + return &models.Agent{ + ID: "agent-456", + Name: "claude", + }, nil + }, + } + + resolver := NewStoreProfileResolver(mockRepo) + ctx := context.Background() + + info, err := resolver.ResolveProfile(ctx, "profile-123") + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if info == nil { + t.Fatal("expected non-nil profile info") + } + if info.ProfileID != "profile-123" { + t.Errorf("expected ProfileID 'profile-123', got '%s'", info.ProfileID) + } + if info.ProfileName != "My Profile" { + t.Errorf("expected ProfileName 'My Profile', got '%s'", info.ProfileName) + } + if info.AgentID != "agent-456" { + t.Errorf("expected AgentID 'agent-456', got '%s'", info.AgentID) + } + if info.AgentName != "claude" { + t.Errorf("expected AgentName 'claude', got '%s'", info.AgentName) + } + if info.Model != "claude-3.5-sonnet" { + t.Errorf("expected Model 'claude-3.5-sonnet', got '%s'", info.Model) + } + if info.AutoApprove != true { + t.Error("expected AutoApprove to be true") + } + if info.DangerouslySkipPermissions != false { + t.Error("expected DangerouslySkipPermissions to be false") + } + if info.Plan != "pro" { + t.Errorf("expected Plan 'pro', got '%s'", info.Plan) + } +} + +func TestStoreProfileResolver_ResolveProfile_ProfileNotFound(t *testing.T) { + mockRepo := &MockRepository{ + GetAgentProfileFn: func(ctx context.Context, id string) (*models.AgentProfile, error) { + return nil, errors.New("profile not found") + }, + } + + resolver := NewStoreProfileResolver(mockRepo) + ctx := context.Background() + + info, err := resolver.ResolveProfile(ctx, "non-existent-profile") + + if err == nil { + t.Fatal("expected error for non-existent profile") + } + if info != nil { + t.Error("expected nil profile info on error") + } + if !errors.Is(err, errors.Unwrap(err)) && err.Error() == "" { + t.Error("expected error to contain message") + } + // Verify the error message contains "profile not found" + expectedMsg := "profile not found" + if err.Error()[:len(expectedMsg)] != expectedMsg { + t.Errorf("expected error message to start with '%s', got '%s'", expectedMsg, err.Error()) + } +} + +func TestStoreProfileResolver_ResolveProfile_AgentNotFound(t *testing.T) { + mockRepo := &MockRepository{ + GetAgentProfileFn: func(ctx context.Context, id string) (*models.AgentProfile, error) { + return &models.AgentProfile{ + ID: "profile-123", + AgentID: "non-existent-agent", + Name: "My Profile", + Model: "gpt-4", + }, nil + }, + GetAgentFn: func(ctx context.Context, id string) (*models.Agent, error) { + return nil, errors.New("agent not found") + }, + } + + resolver := NewStoreProfileResolver(mockRepo) + ctx := context.Background() + + info, err := resolver.ResolveProfile(ctx, "profile-123") + + if err == nil { + t.Fatal("expected error when agent not found") + } + if info != nil { + t.Error("expected nil profile info on error") + } + // Verify the error message contains "agent not found for profile" + expectedMsg := "agent not found for profile" + if err.Error()[:len(expectedMsg)] != expectedMsg { + t.Errorf("expected error message to start with '%s', got '%s'", expectedMsg, err.Error()) + } +} + diff --git a/apps/backend/internal/agent/registry/defaults.go b/apps/backend/internal/agent/registry/defaults.go index 2d31d9fcab..e74583bccc 100644 --- a/apps/backend/internal/agent/registry/defaults.go +++ b/apps/backend/internal/agent/registry/defaults.go @@ -1,5 +1,7 @@ package registry +import "github.com/kandev/kandev/pkg/agent" + // DefaultAgents returns the default agent configurations func DefaultAgents() []*AgentTypeConfig { return []*AgentTypeConfig{ @@ -27,6 +29,7 @@ func DefaultAgents() []*AgentTypeConfig { Capabilities: []string{"code_generation", "code_review", "refactoring", "testing", "shell_execution"}, Enabled: true, ModelFlag: "--model", + Protocol: agent.ProtocolACP, }, { ID: "claude-agent", @@ -49,6 +52,7 @@ func DefaultAgents() []*AgentTypeConfig { Capabilities: []string{"code_generation", "code_review", "refactoring", "testing", "shell_execution"}, Enabled: true, ModelFlag: "--model", + Protocol: agent.ProtocolACP, }, { ID: "codex-agent", @@ -71,6 +75,7 @@ func DefaultAgents() []*AgentTypeConfig { Capabilities: []string{"code_generation", "code_review", "refactoring", "testing", "shell_execution"}, Enabled: true, ModelFlag: "--model", + Protocol: agent.ProtocolACP, }, { ID: "gemini-agent", @@ -93,6 +98,7 @@ func DefaultAgents() []*AgentTypeConfig { Capabilities: []string{"code_generation", "code_review", "refactoring", "testing", "shell_execution"}, Enabled: true, ModelFlag: "--model", + Protocol: agent.ProtocolACP, }, { ID: "opencode-agent", @@ -115,6 +121,7 @@ func DefaultAgents() []*AgentTypeConfig { Capabilities: []string{"code_generation", "code_review", "refactoring", "testing", "shell_execution"}, Enabled: true, ModelFlag: "--model", + Protocol: agent.ProtocolACP, }, { ID: "copilot-agent", @@ -137,6 +144,7 @@ func DefaultAgents() []*AgentTypeConfig { Capabilities: []string{"code_generation", "code_review", "refactoring", "testing", "shell_execution"}, Enabled: true, ModelFlag: "--model", + Protocol: agent.ProtocolACP, }, } } diff --git a/apps/backend/internal/agent/registry/registry.go b/apps/backend/internal/agent/registry/registry.go index 208aa1fb78..38d8b4b2cb 100644 --- a/apps/backend/internal/agent/registry/registry.go +++ b/apps/backend/internal/agent/registry/registry.go @@ -11,6 +11,7 @@ import ( "go.uber.org/zap" "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/pkg/agent" v1 "github.com/kandev/kandev/pkg/api/v1" ) @@ -31,6 +32,10 @@ type AgentTypeConfig struct { Capabilities []string `json:"capabilities"` // What the agent can do Enabled bool `json:"enabled"` ModelFlag string `json:"model_flag,omitempty"` // CLI flag for model selection (e.g., "--model") + + // Protocol configuration + Protocol agent.Protocol `json:"protocol,omitempty"` // Communication protocol: "acp", "rest", "mcp" + ProtocolConfig map[string]string `json:"protocol_config,omitempty"` // Protocol-specific settings (e.g., base_url for REST) } // MountTemplate defines a mount with template support diff --git a/apps/backend/internal/agentctl/acp/client.go b/apps/backend/internal/agentctl/acp/client.go index cd3f9b4518..599d1c2651 100644 --- a/apps/backend/internal/agentctl/acp/client.go +++ b/apps/backend/internal/agentctl/acp/client.go @@ -10,6 +10,7 @@ import ( "sync" "github.com/coder/acp-go-sdk" + "github.com/kandev/kandev/internal/agentctl/types" "go.uber.org/zap" ) @@ -18,28 +19,7 @@ type UpdateHandler func(notification acp.SessionNotification) // PermissionRequestHandler is called when the agent requests permission // Returns the selected option ID, or empty string with cancelled=true to cancel -type PermissionRequestHandler func(ctx context.Context, req *PermissionRequest) (*PermissionResponse, error) - -// PermissionRequest represents a permission request from the agent -type PermissionRequest struct { - SessionID string `json:"session_id"` - ToolCallID string `json:"tool_call_id"` - Title string `json:"title"` - Options []PermissionOption `json:"options"` -} - -// PermissionOption represents a permission choice -type PermissionOption struct { - OptionID string `json:"option_id"` - Name string `json:"name"` - Kind string `json:"kind"` // allow_once, allow_always, reject_once, reject_always -} - -// PermissionResponse is the user's response to a permission request -type PermissionResponse struct { - OptionID string `json:"option_id,omitempty"` - Cancelled bool `json:"cancelled,omitempty"` -} +type PermissionRequestHandler func(ctx context.Context, req *types.PermissionRequest) (*types.PermissionResponse, error) // Client implements acp.Client interface and handles all agent requests type Client struct { @@ -148,10 +128,10 @@ func (c *Client) RequestPermission(ctx context.Context, p acp.RequestPermissionR // forwardPermissionRequest forwards the permission request to an external handler func (c *Client) forwardPermissionRequest(ctx context.Context, handler PermissionRequestHandler, p acp.RequestPermissionRequest) (acp.RequestPermissionResponse, error) { - // Convert ACP types to our types - options := make([]PermissionOption, len(p.Options)) + // Convert ACP types to shared types + options := make([]types.PermissionOption, len(p.Options)) for i, opt := range p.Options { - options[i] = PermissionOption{ + options[i] = types.PermissionOption{ OptionID: string(opt.OptionId), Name: opt.Name, Kind: string(opt.Kind), @@ -163,7 +143,7 @@ func (c *Client) forwardPermissionRequest(ctx context.Context, handler Permissio title = *p.ToolCall.Title } - req := &PermissionRequest{ + req := &types.PermissionRequest{ SessionID: string(p.SessionId), ToolCallID: string(p.ToolCall.ToolCallId), Title: title, diff --git a/apps/backend/internal/agentctl/adapter/acp_adapter.go b/apps/backend/internal/agentctl/adapter/acp_adapter.go new file mode 100644 index 0000000000..1eb657d41b --- /dev/null +++ b/apps/backend/internal/agentctl/adapter/acp_adapter.go @@ -0,0 +1,358 @@ +package adapter + +import ( + "context" + "fmt" + "io" + "log/slog" + "sync" + + "github.com/coder/acp-go-sdk" + acpclient "github.com/kandev/kandev/internal/agentctl/acp" + "github.com/kandev/kandev/internal/common/logger" + "go.uber.org/zap" +) + +// ACPAdapter implements AgentAdapter for agents using the ACP protocol. +// ACP (Agent Communication Protocol) uses JSON-RPC 2.0 over stdin/stdout. +// The subprocess is managed externally (by process.Manager) and stdin/stdout +// are passed to the adapter. +type ACPAdapter struct { + cfg *Config + logger *logger.Logger + + // Subprocess stdin/stdout (managed externally) + stdin io.Writer + stdout io.Reader + + // ACP SDK connection + acpClient *acpclient.Client + acpConn *acp.ClientSideConnection + sessionID string + + // Agent info (populated after Initialize) + agentInfo *AgentInfo + + // Update channel + updatesCh chan SessionUpdate + + // Permission handler + permissionHandler PermissionHandler + + // Synchronization + mu sync.RWMutex + closed bool +} + +// NewACPAdapter creates a new ACP protocol adapter. +// stdin and stdout are the subprocess's stdin/stdout pipes (managed by process.Manager). +func NewACPAdapter(stdin io.Writer, stdout io.Reader, cfg *Config, log *logger.Logger) *ACPAdapter { + return &ACPAdapter{ + cfg: cfg, + logger: log.WithFields(zap.String("adapter", "acp")), + stdin: stdin, + stdout: stdout, + updatesCh: make(chan SessionUpdate, 100), + } +} + +// Initialize establishes the ACP connection with the agent subprocess. +// The subprocess should already be running (started by process.Manager). +func (a *ACPAdapter) Initialize(ctx context.Context) error { + a.logger.Info("initializing ACP adapter", + zap.String("workdir", a.cfg.WorkDir)) + + // Create ACP client with update handler that converts to SessionUpdate + a.acpClient = acpclient.NewClient( + acpclient.WithLogger(a.logger.Zap()), + acpclient.WithWorkspaceRoot(a.cfg.WorkDir), + acpclient.WithUpdateHandler(a.handleACPUpdate), + acpclient.WithPermissionHandler(a.handlePermissionRequest), + ) + + // Create ACP SDK connection + a.acpConn = acp.NewClientSideConnection(a.acpClient, a.stdin, a.stdout) + a.acpConn.SetLogger(slog.Default().With("component", "acp-conn")) + + // Perform ACP handshake - this exchanges capabilities with the agent + resp, err := a.acpConn.Initialize(ctx, acp.InitializeRequest{ + ProtocolVersion: acp.ProtocolVersionNumber, + ClientInfo: &acp.Implementation{ + Name: "kandev-agentctl", + Version: "1.0.0", + }, + }) + if err != nil { + return fmt.Errorf("ACP initialize handshake failed: %w", err) + } + + // Store agent info + a.agentInfo = &AgentInfo{ + Name: "unknown", + Version: "unknown", + } + if resp.AgentInfo != nil { + a.agentInfo.Name = resp.AgentInfo.Name + a.agentInfo.Version = resp.AgentInfo.Version + } + a.logger.Info("ACP adapter initialized", + zap.String("agent_name", a.agentInfo.Name), + zap.String("agent_version", a.agentInfo.Version)) + + return nil +} + +// GetAgentInfo returns information about the connected agent. +func (a *ACPAdapter) GetAgentInfo() *AgentInfo { + return a.agentInfo +} + +// NewSession creates a new agent session. +func (a *ACPAdapter) NewSession(ctx context.Context) (string, error) { + a.mu.Lock() + defer a.mu.Unlock() + + if a.acpConn == nil { + return "", fmt.Errorf("adapter not initialized") + } + + resp, err := a.acpConn.NewSession(ctx, acp.NewSessionRequest{ + Cwd: a.cfg.WorkDir, + McpServers: []acp.McpServer{}, + }) + if err != nil { + return "", fmt.Errorf("failed to create session: %w", err) + } + + a.sessionID = string(resp.SessionId) + a.logger.Info("created new session", zap.String("session_id", a.sessionID)) + + return a.sessionID, nil +} + +// LoadSession resumes an existing session. +func (a *ACPAdapter) LoadSession(ctx context.Context, sessionID string) error { + a.mu.Lock() + defer a.mu.Unlock() + + if a.acpConn == nil { + return fmt.Errorf("adapter not initialized") + } + + _, err := a.acpConn.LoadSession(ctx, acp.LoadSessionRequest{ + SessionId: acp.SessionId(sessionID), + }) + if err != nil { + return fmt.Errorf("failed to load session: %w", err) + } + + a.sessionID = sessionID + a.logger.Info("loaded session", zap.String("session_id", a.sessionID)) + + return nil +} + +// Prompt sends a prompt to the agent. +func (a *ACPAdapter) Prompt(ctx context.Context, message string) error { + a.mu.RLock() + conn := a.acpConn + sessionID := a.sessionID + a.mu.RUnlock() + + if conn == nil { + return fmt.Errorf("adapter not initialized") + } + + a.logger.Info("sending prompt", zap.String("session_id", sessionID)) + + _, err := conn.Prompt(ctx, acp.PromptRequest{ + SessionId: acp.SessionId(sessionID), + Prompt: []acp.ContentBlock{acp.TextBlock(message)}, + }) + return err +} + +// Cancel cancels the current operation. +func (a *ACPAdapter) Cancel(ctx context.Context) error { + a.mu.RLock() + conn := a.acpConn + sessionID := a.sessionID + a.mu.RUnlock() + + if conn == nil { + return fmt.Errorf("adapter not initialized") + } + + a.logger.Info("cancelling session", zap.String("session_id", sessionID)) + + return conn.Cancel(ctx, acp.CancelNotification{ + SessionId: acp.SessionId(sessionID), + }) +} + +// Updates returns the channel for session updates. +func (a *ACPAdapter) Updates() <-chan SessionUpdate { + return a.updatesCh +} + +// GetSessionID returns the current session ID. +func (a *ACPAdapter) GetSessionID() string { + a.mu.RLock() + defer a.mu.RUnlock() + return a.sessionID +} + +// SetPermissionHandler sets the handler for permission requests. +func (a *ACPAdapter) SetPermissionHandler(handler PermissionHandler) { + a.mu.Lock() + defer a.mu.Unlock() + a.permissionHandler = handler +} + +// Close releases resources held by the adapter. +func (a *ACPAdapter) Close() error { + a.mu.Lock() + defer a.mu.Unlock() + + if a.closed { + return nil + } + a.closed = true + + a.logger.Info("closing ACP adapter") + + // Close update channel + close(a.updatesCh) + + // Note: We don't close stdin or manage the subprocess here. + // That's handled by process.Manager which owns the subprocess. + + return nil +} + +// GetACPConnection returns the underlying ACP connection for advanced usage. +func (a *ACPAdapter) GetACPConnection() *acp.ClientSideConnection { + a.mu.RLock() + defer a.mu.RUnlock() + return a.acpConn +} + +// handleACPUpdate converts ACP SessionNotification to protocol-agnostic SessionUpdate. +func (a *ACPAdapter) handleACPUpdate(n acp.SessionNotification) { + update := a.convertNotification(n) + if update != nil { + select { + case a.updatesCh <- *update: + default: + a.logger.Warn("updates channel full, dropping notification") + } + } +} + +// convertNotification converts an ACP SessionNotification to a SessionUpdate. +func (a *ACPAdapter) convertNotification(n acp.SessionNotification) *SessionUpdate { + u := n.Update + + switch { + case u.AgentMessageChunk != nil: + if u.AgentMessageChunk.Content.Text != nil { + return &SessionUpdate{ + Type: "message_chunk", + SessionID: string(n.SessionId), + Text: u.AgentMessageChunk.Content.Text.Text, + } + } + + case u.ToolCall != nil: + // Extract rich tool call information + args := map[string]interface{}{} + + // Add tool kind + if u.ToolCall.Kind != "" { + args["kind"] = string(u.ToolCall.Kind) + } + + // Add locations (file paths with line numbers) + if len(u.ToolCall.Locations) > 0 { + locations := make([]map[string]interface{}, len(u.ToolCall.Locations)) + for i, loc := range u.ToolCall.Locations { + locMap := map[string]interface{}{"path": loc.Path} + if loc.Line != nil { + locMap["line"] = *loc.Line + } + locations[i] = locMap + } + args["locations"] = locations + + // Also set primary path for convenience + args["path"] = u.ToolCall.Locations[0].Path + } + + // Add raw input if available + if u.ToolCall.RawInput != nil { + args["raw_input"] = u.ToolCall.RawInput + } + + return &SessionUpdate{ + Type: "tool_call", + SessionID: string(n.SessionId), + ToolCallID: string(u.ToolCall.ToolCallId), + ToolName: string(u.ToolCall.Kind), // Kind is effectively the tool name + ToolTitle: u.ToolCall.Title, + ToolStatus: string(u.ToolCall.Status), + ToolArgs: args, + } + + case u.ToolCallUpdate != nil: + status := "" + if u.ToolCallUpdate.Status != nil { + status = string(*u.ToolCallUpdate.Status) + } + return &SessionUpdate{ + Type: "tool_update", + SessionID: string(n.SessionId), + ToolCallID: string(u.ToolCallUpdate.ToolCallId), + ToolStatus: status, + } + + case u.Plan != nil: + entries := make([]PlanEntry, len(u.Plan.Entries)) + for i, e := range u.Plan.Entries { + entries[i] = PlanEntry{ + Description: e.Content, + Status: string(e.Status), + Priority: string(e.Priority), + } + } + return &SessionUpdate{ + Type: "plan", + SessionID: string(n.SessionId), + PlanEntries: entries, + } + } + + return nil +} + +// handlePermissionRequest handles permission requests from the agent. +// Since both acpclient and adapter now use the shared types package, +// no conversion is needed - we just forward to the handler. +func (a *ACPAdapter) handlePermissionRequest(ctx context.Context, req *PermissionRequest) (*PermissionResponse, error) { + a.mu.RLock() + handler := a.permissionHandler + a.mu.RUnlock() + + if handler == nil { + // Auto-approve if no handler + if len(req.Options) > 0 { + return &PermissionResponse{OptionID: req.Options[0].OptionID}, nil + } + return &PermissionResponse{Cancelled: true}, nil + } + + // Forward directly to handler - types are already compatible + return handler(ctx, req) +} + +// Verify interface implementation +var _ AgentAdapter = (*ACPAdapter)(nil) diff --git a/apps/backend/internal/agentctl/adapter/adapter.go b/apps/backend/internal/agentctl/adapter/adapter.go new file mode 100644 index 0000000000..7cfc242358 --- /dev/null +++ b/apps/backend/internal/agentctl/adapter/adapter.go @@ -0,0 +1,121 @@ +// Package adapter provides protocol adapters for different agent communication protocols. +// This abstraction allows agentctl to work with agents using ACP, REST, MCP, or custom protocols. +package adapter + +import ( + "context" + + "github.com/kandev/kandev/internal/agentctl/types" +) + +// Re-export permission types from the shared types package for convenience. +// This allows users of the adapter package to access these types without +// importing the types package directly. +type ( + PermissionRequest = types.PermissionRequest + PermissionResponse = types.PermissionResponse + PermissionOption = types.PermissionOption + PermissionHandler = types.PermissionHandler +) + +// SessionUpdate is a protocol-agnostic update from the agent. +// All protocol adapters normalize their updates to this format. +type SessionUpdate struct { + // Type identifies the update type: "message_chunk", "tool_call", "tool_update", "plan", "complete", "error" + Type string `json:"type"` + + // SessionID is the current session identifier + SessionID string `json:"session_id,omitempty"` + + // Message fields (for "message_chunk" type) + Text string `json:"text,omitempty"` + + // Tool call fields (for "tool_call" and "tool_update" types) + ToolCallID string `json:"tool_call_id,omitempty"` + ToolName string `json:"tool_name,omitempty"` + ToolTitle string `json:"tool_title,omitempty"` + ToolStatus string `json:"tool_status,omitempty"` // "started", "running", "completed", "error" + ToolArgs map[string]interface{} `json:"tool_args,omitempty"` + ToolResult interface{} `json:"tool_result,omitempty"` + + // Plan fields (for "plan" type) + PlanEntries []PlanEntry `json:"plan_entries,omitempty"` + + // Error fields (for "error" type) + Error string `json:"error,omitempty"` + + // Raw data for protocol-specific extensions + Data map[string]interface{} `json:"data,omitempty"` +} + +// PlanEntry represents an entry in the agent's execution plan +type PlanEntry struct { + Description string `json:"description,omitempty"` // Content/description of the task + Status string `json:"status,omitempty"` // "pending", "in_progress", "completed", "failed" + Priority string `json:"priority,omitempty"` // Relative importance +} + +// AgentInfo contains information about the connected agent. +type AgentInfo struct { + Name string `json:"name"` + Version string `json:"version"` +} + +// AgentAdapter defines the interface for protocol adapters. +// Each adapter translates a specific protocol (ACP, REST, MCP, etc.) into the +// normalized SessionUpdate format that agentctl exposes via its HTTP API. +type AgentAdapter interface { + // Initialize establishes the connection with the agent and exchanges capabilities. + // For subprocess-based agents (ACP), this sends the initialize request. + // For HTTP-based agents (REST), this might do a health check. + Initialize(ctx context.Context) error + + // GetAgentInfo returns information about the connected agent. + // Returns nil if Initialize has not been called yet. + GetAgentInfo() *AgentInfo + + // NewSession creates a new agent session and returns the session ID. + NewSession(ctx context.Context) (string, error) + + // LoadSession resumes an existing session by ID. + LoadSession(ctx context.Context, sessionID string) error + + // Prompt sends a prompt to the agent. + // The agent's responses are streamed via the Updates channel. + Prompt(ctx context.Context, message string) error + + // Cancel cancels the current operation. + Cancel(ctx context.Context) error + + // Updates returns a channel that receives session updates. + // The channel is closed when the adapter is closed. + Updates() <-chan SessionUpdate + + // GetSessionID returns the current session ID. + GetSessionID() string + + // SetPermissionHandler sets the handler for permission requests. + SetPermissionHandler(handler PermissionHandler) + + // Close releases resources held by the adapter. + Close() error +} + +// Config holds configuration for creating adapters +type Config struct { + // WorkDir is the working directory for the agent + WorkDir string + + // AutoApprove automatically approves permission requests + AutoApprove bool + + // For HTTP-based adapters (REST) + BaseURL string // Base URL of the agent's HTTP API + AuthHeader string // Optional auth header name + AuthValue string // Optional auth header value + Headers map[string]string // Additional headers + + // Protocol-specific configuration + Extra map[string]string +} + diff --git a/apps/backend/internal/agentctl/adapter/adapter_test.go b/apps/backend/internal/agentctl/adapter/adapter_test.go new file mode 100644 index 0000000000..07a2d9a04e --- /dev/null +++ b/apps/backend/internal/agentctl/adapter/adapter_test.go @@ -0,0 +1,331 @@ +package adapter + +import ( + "encoding/json" + "testing" +) + +func TestSessionUpdate_Fields(t *testing.T) { + update := SessionUpdate{ + Type: "tool_call", + SessionID: "session-123", + Text: "test message", + ToolCallID: "tool-call-456", + ToolName: "file_read", + ToolTitle: "Reading file", + ToolStatus: "running", + ToolArgs: map[string]interface{}{"path": "/test/file.txt"}, + ToolResult: map[string]interface{}{"content": "file contents"}, + PlanEntries: []PlanEntry{ + {Description: "Step 1", Status: "completed", Priority: "high"}, + }, + Error: "", + Data: map[string]interface{}{"custom": "data"}, + } + + if update.Type != "tool_call" { + t.Errorf("expected Type 'tool_call', got %q", update.Type) + } + if update.SessionID != "session-123" { + t.Errorf("expected SessionID 'session-123', got %q", update.SessionID) + } + if update.Text != "test message" { + t.Errorf("expected Text 'test message', got %q", update.Text) + } + if update.ToolCallID != "tool-call-456" { + t.Errorf("expected ToolCallID 'tool-call-456', got %q", update.ToolCallID) + } + if update.ToolName != "file_read" { + t.Errorf("expected ToolName 'file_read', got %q", update.ToolName) + } + if update.ToolTitle != "Reading file" { + t.Errorf("expected ToolTitle 'Reading file', got %q", update.ToolTitle) + } + if update.ToolStatus != "running" { + t.Errorf("expected ToolStatus 'running', got %q", update.ToolStatus) + } + if update.ToolArgs["path"] != "/test/file.txt" { + t.Errorf("expected ToolArgs[path] '/test/file.txt', got %v", update.ToolArgs["path"]) + } + if len(update.PlanEntries) != 1 { + t.Errorf("expected 1 PlanEntry, got %d", len(update.PlanEntries)) + } + if update.Data["custom"] != "data" { + t.Errorf("expected Data[custom] 'data', got %v", update.Data["custom"]) + } +} + +func TestSessionUpdate_JSONSerialization(t *testing.T) { + update := SessionUpdate{ + Type: "message_chunk", + SessionID: "session-abc", + Text: "Hello, world!", + ToolCallID: "", + } + + data, err := json.Marshal(update) + if err != nil { + t.Fatalf("failed to marshal SessionUpdate: %v", err) + } + + var decoded SessionUpdate + if err := json.Unmarshal(data, &decoded); err != nil { + t.Fatalf("failed to unmarshal SessionUpdate: %v", err) + } + + if decoded.Type != update.Type { + t.Errorf("expected Type %q, got %q", update.Type, decoded.Type) + } + if decoded.SessionID != update.SessionID { + t.Errorf("expected SessionID %q, got %q", update.SessionID, decoded.SessionID) + } + if decoded.Text != update.Text { + t.Errorf("expected Text %q, got %q", update.Text, decoded.Text) + } +} + +func TestSessionUpdate_JSONOmitEmpty(t *testing.T) { + // Test that omitempty fields are not included when empty + update := SessionUpdate{ + Type: "error", + } + + data, err := json.Marshal(update) + if err != nil { + t.Fatalf("failed to marshal SessionUpdate: %v", err) + } + + var parsed map[string]interface{} + if err := json.Unmarshal(data, &parsed); err != nil { + t.Fatalf("failed to unmarshal to map: %v", err) + } + + // Type should be present + if _, ok := parsed["type"]; !ok { + t.Error("expected 'type' field to be present") + } + + // These should be omitted due to omitempty + if _, ok := parsed["session_id"]; ok { + t.Error("expected 'session_id' to be omitted when empty") + } + if _, ok := parsed["text"]; ok { + t.Error("expected 'text' to be omitted when empty") + } + if _, ok := parsed["tool_call_id"]; ok { + t.Error("expected 'tool_call_id' to be omitted when empty") + } +} + +func TestPlanEntry_Fields(t *testing.T) { + entry := PlanEntry{ + Description: "Analyze the codebase", + Status: "in_progress", + Priority: "high", + } + + if entry.Description != "Analyze the codebase" { + t.Errorf("expected Description 'Analyze the codebase', got %q", entry.Description) + } + if entry.Status != "in_progress" { + t.Errorf("expected Status 'in_progress', got %q", entry.Status) + } + if entry.Priority != "high" { + t.Errorf("expected Priority 'high', got %q", entry.Priority) + } +} + +func TestPlanEntry_JSONSerialization(t *testing.T) { + entries := []PlanEntry{ + {Description: "Task 1", Status: "pending", Priority: "low"}, + {Description: "Task 2", Status: "completed", Priority: "high"}, + {Description: "Task 3", Status: "failed", Priority: "medium"}, + } + + data, err := json.Marshal(entries) + if err != nil { + t.Fatalf("failed to marshal PlanEntry slice: %v", err) + } + + var decoded []PlanEntry + if err := json.Unmarshal(data, &decoded); err != nil { + t.Fatalf("failed to unmarshal PlanEntry slice: %v", err) + } + + if len(decoded) != 3 { + t.Fatalf("expected 3 entries, got %d", len(decoded)) + } + + for i, entry := range entries { + if decoded[i].Description != entry.Description { + t.Errorf("entry %d: expected Description %q, got %q", i, entry.Description, decoded[i].Description) + } + if decoded[i].Status != entry.Status { + t.Errorf("entry %d: expected Status %q, got %q", i, entry.Status, decoded[i].Status) + } + if decoded[i].Priority != entry.Priority { + t.Errorf("entry %d: expected Priority %q, got %q", i, entry.Priority, decoded[i].Priority) + } + } +} + +func TestAgentInfo_Fields(t *testing.T) { + info := AgentInfo{ + Name: "test-agent", + Version: "1.0.0", + } + + if info.Name != "test-agent" { + t.Errorf("expected Name 'test-agent', got %q", info.Name) + } + if info.Version != "1.0.0" { + t.Errorf("expected Version '1.0.0', got %q", info.Version) + } +} + +func TestAgentInfo_JSONSerialization(t *testing.T) { + info := AgentInfo{ + Name: "claude-agent", + Version: "2.5.0", + } + + data, err := json.Marshal(info) + if err != nil { + t.Fatalf("failed to marshal AgentInfo: %v", err) + } + + var decoded AgentInfo + if err := json.Unmarshal(data, &decoded); err != nil { + t.Fatalf("failed to unmarshal AgentInfo: %v", err) + } + + if decoded.Name != info.Name { + t.Errorf("expected Name %q, got %q", info.Name, decoded.Name) + } + if decoded.Version != info.Version { + t.Errorf("expected Version %q, got %q", info.Version, decoded.Version) + } +} + +func TestConfig_Fields(t *testing.T) { + cfg := Config{ + WorkDir: "/workspace", + AutoApprove: true, + BaseURL: "http://localhost:8080", + AuthHeader: "Authorization", + AuthValue: "Bearer token123", + Headers: map[string]string{"X-Custom": "value"}, + Extra: map[string]string{"debug": "true"}, + } + + if cfg.WorkDir != "/workspace" { + t.Errorf("expected WorkDir '/workspace', got %q", cfg.WorkDir) + } + if !cfg.AutoApprove { + t.Error("expected AutoApprove true, got false") + } + if cfg.BaseURL != "http://localhost:8080" { + t.Errorf("expected BaseURL 'http://localhost:8080', got %q", cfg.BaseURL) + } + if cfg.AuthHeader != "Authorization" { + t.Errorf("expected AuthHeader 'Authorization', got %q", cfg.AuthHeader) + } + if cfg.AuthValue != "Bearer token123" { + t.Errorf("expected AuthValue 'Bearer token123', got %q", cfg.AuthValue) + } + if cfg.Headers["X-Custom"] != "value" { + t.Errorf("expected Headers[X-Custom] 'value', got %v", cfg.Headers["X-Custom"]) + } + if cfg.Extra["debug"] != "true" { + t.Errorf("expected Extra[debug] 'true', got %v", cfg.Extra["debug"]) + } +} + +func TestSessionUpdate_UpdateTypes(t *testing.T) { + // Test various update type constants + testCases := []struct { + name string + updateType string + }{ + {"message chunk", "message_chunk"}, + {"tool call", "tool_call"}, + {"tool update", "tool_update"}, + {"plan", "plan"}, + {"complete", "complete"}, + {"error", "error"}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + update := SessionUpdate{Type: tc.updateType} + if update.Type != tc.updateType { + t.Errorf("expected Type %q, got %q", tc.updateType, update.Type) + } + }) + } +} + +func TestPlanEntry_StatusValues(t *testing.T) { + // Test various status values + testCases := []struct { + status string + }{ + {"pending"}, + {"in_progress"}, + {"completed"}, + {"failed"}, + } + + for _, tc := range testCases { + t.Run(tc.status, func(t *testing.T) { + entry := PlanEntry{Status: tc.status} + if entry.Status != tc.status { + t.Errorf("expected Status %q, got %q", tc.status, entry.Status) + } + }) + } +} + +func TestSessionUpdate_WithComplexToolArgs(t *testing.T) { + // Test with complex nested ToolArgs + update := SessionUpdate{ + Type: "tool_call", + ToolCallID: "call-123", + ToolName: "file_edit", + ToolArgs: map[string]interface{}{ + "path": "/src/main.go", + "locations": []map[string]interface{}{ + {"path": "/src/main.go", "line": 42}, + {"path": "/src/util.go", "line": 10}, + }, + "raw_input": map[string]interface{}{ + "old_text": "foo", + "new_text": "bar", + }, + }, + } + + // Marshal and unmarshal to verify complex structures work + data, err := json.Marshal(update) + if err != nil { + t.Fatalf("failed to marshal: %v", err) + } + + var decoded SessionUpdate + if err := json.Unmarshal(data, &decoded); err != nil { + t.Fatalf("failed to unmarshal: %v", err) + } + + if decoded.ToolArgs["path"] != "/src/main.go" { + t.Errorf("expected path '/src/main.go', got %v", decoded.ToolArgs["path"]) + } + + locations, ok := decoded.ToolArgs["locations"].([]interface{}) + if !ok { + t.Fatal("expected locations to be a slice") + } + if len(locations) != 2 { + t.Errorf("expected 2 locations, got %d", len(locations)) + } +} + diff --git a/apps/backend/internal/agentctl/api/acp.go b/apps/backend/internal/agentctl/api/acp.go index fa1294505d..dac1e5024e 100644 --- a/apps/backend/internal/agentctl/api/acp.go +++ b/apps/backend/internal/agentctl/api/acp.go @@ -6,7 +6,6 @@ import ( "net/http" "time" - "github.com/coder/acp-go-sdk" "github.com/gin-gonic/gin" "github.com/gorilla/websocket" "go.uber.org/zap" @@ -18,11 +17,17 @@ type InitializeRequest struct { ClientVersion string `json:"client_version"` } +// AgentInfoResponse contains information about the connected agent. +type AgentInfoResponse struct { + Name string `json:"name"` + Version string `json:"version"` +} + // InitializeResponse is the response to an initialize call type InitializeResponse struct { - Success bool `json:"success"` - Response *acp.InitializeResponse `json:"response,omitempty"` - Error string `json:"error,omitempty"` + Success bool `json:"success"` + AgentInfo *AgentInfoResponse `json:"agent_info,omitempty"` + Error string `json:"error,omitempty"` } func (s *Server) handleACPInitialize(c *gin.Context) { @@ -38,8 +43,8 @@ func (s *Server) handleACPInitialize(c *gin.Context) { ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) defer cancel() - conn := s.procMgr.GetConnection() - if conn == nil { + adapter := s.procMgr.GetAdapter() + if adapter == nil { c.JSON(http.StatusServiceUnavailable, InitializeResponse{ Success: false, Error: "agent not running", @@ -47,13 +52,7 @@ func (s *Server) handleACPInitialize(c *gin.Context) { return } - resp, err := conn.Initialize(ctx, acp.InitializeRequest{ - ProtocolVersion: acp.ProtocolVersionNumber, - ClientInfo: &acp.Implementation{ - Name: req.ClientName, - Version: req.ClientVersion, - }, - }) + err := adapter.Initialize(ctx) if err != nil { s.logger.Error("initialize failed", zap.Error(err)) c.JSON(http.StatusInternalServerError, InitializeResponse{ @@ -63,9 +62,18 @@ func (s *Server) handleACPInitialize(c *gin.Context) { return } + // Get agent info after successful initialization + var agentInfoResp *AgentInfoResponse + if info := adapter.GetAgentInfo(); info != nil { + agentInfoResp = &AgentInfoResponse{ + Name: info.Name, + Version: info.Version, + } + } + c.JSON(http.StatusOK, InitializeResponse{ - Success: true, - Response: &resp, + Success: true, + AgentInfo: agentInfoResp, }) } @@ -94,8 +102,8 @@ func (s *Server) handleACPNewSession(c *gin.Context) { ctx, cancel := context.WithTimeout(c.Request.Context(), 30*time.Second) defer cancel() - conn := s.procMgr.GetConnection() - if conn == nil { + adapter := s.procMgr.GetAdapter() + if adapter == nil { c.JSON(http.StatusServiceUnavailable, NewSessionResponse{ Success: false, Error: "agent not running", @@ -103,15 +111,7 @@ func (s *Server) handleACPNewSession(c *gin.Context) { return } - cwd := req.Cwd - if cwd == "" { - cwd = "/workspace" - } - - resp, err := conn.NewSession(ctx, acp.NewSessionRequest{ - Cwd: cwd, - McpServers: []acp.McpServer{}, // Required: must be an empty array, not nil - }) + sessionID, err := adapter.NewSession(ctx) if err != nil { s.logger.Error("new session failed", zap.Error(err)) c.JSON(http.StatusInternalServerError, NewSessionResponse{ @@ -121,12 +121,10 @@ func (s *Server) handleACPNewSession(c *gin.Context) { return } - // Store session ID - s.procMgr.SetSessionID(resp.SessionId) - + // Session ID is stored in the adapter - no need to duplicate in Manager c.JSON(http.StatusOK, NewSessionResponse{ Success: true, - SessionID: string(resp.SessionId), + SessionID: sessionID, }) } @@ -137,9 +135,9 @@ type PromptRequest struct { // PromptResponse is the response to a prompt call type PromptResponse struct { - Success bool `json:"success"` - StopReason acp.StopReason `json:"stop_reason,omitempty"` - Error string `json:"error,omitempty"` + Success bool `json:"success"` + StopReason string `json:"stop_reason,omitempty"` + Error string `json:"error,omitempty"` } func (s *Server) handleACPPrompt(c *gin.Context) { @@ -152,8 +150,8 @@ func (s *Server) handleACPPrompt(c *gin.Context) { return } - conn := s.procMgr.GetConnection() - if conn == nil { + adapter := s.procMgr.GetAdapter() + if adapter == nil { c.JSON(http.StatusServiceUnavailable, PromptResponse{ Success: false, Error: "agent not running", @@ -174,11 +172,7 @@ func (s *Server) handleACPPrompt(c *gin.Context) { ctx, cancel := context.WithTimeout(c.Request.Context(), 10*time.Minute) defer cancel() - // Build prompt content - simple text block - resp, err := conn.Prompt(ctx, acp.PromptRequest{ - SessionId: sessionID, - Prompt: []acp.ContentBlock{acp.TextBlock(req.Text)}, - }) + err := adapter.Prompt(ctx, req.Text) if err != nil { s.logger.Error("prompt failed", zap.Error(err)) c.JSON(http.StatusInternalServerError, PromptResponse{ @@ -188,12 +182,10 @@ func (s *Server) handleACPPrompt(c *gin.Context) { return } - s.logger.Info("prompt completed", - zap.String("stop_reason", string(resp.StopReason))) + s.logger.Info("prompt completed") c.JSON(http.StatusOK, PromptResponse{ - Success: true, - StopReason: resp.StopReason, + Success: true, }) } diff --git a/apps/backend/internal/agentctl/config/config.go b/apps/backend/internal/agentctl/config/config.go index 33f4e8a14b..f1dc01ed4c 100644 --- a/apps/backend/internal/agentctl/config/config.go +++ b/apps/backend/internal/agentctl/config/config.go @@ -5,6 +5,8 @@ import ( "os" "strconv" "strings" + + "github.com/kandev/kandev/pkg/agent" ) // Config holds the agentctl configuration @@ -12,6 +14,9 @@ type Config struct { // HTTP server port Port int + // Protocol used to communicate with the agent + Protocol agent.Protocol + // Agent command to execute (e.g., "auggie --acp") AgentCommand string @@ -45,6 +50,7 @@ func Load() *Config { cfg := &Config{ Port: getEnvInt("AGENTCTL_PORT", 9999), + Protocol: agent.Protocol(getEnv("AGENTCTL_PROTOCOL", string(agent.ProtocolACP))), AgentCommand: getEnv("AGENTCTL_AGENT_COMMAND", defaultCmd), WorkDir: workDir, AutoStart: getEnvBool("AGENTCTL_AUTO_START", false), diff --git a/apps/backend/internal/agentctl/config/multi_config.go b/apps/backend/internal/agentctl/config/multi_config.go index 30bfdc8c25..8e6e730c77 100644 --- a/apps/backend/internal/agentctl/config/multi_config.go +++ b/apps/backend/internal/agentctl/config/multi_config.go @@ -2,6 +2,8 @@ // This file contains multi-instance configuration for running multiple agent instances. package config +import "github.com/kandev/kandev/pkg/agent" + // MultiConfig holds configuration for multi-instance mode. // It extends the base configuration with settings for managing multiple concurrent agent instances. type MultiConfig struct { @@ -17,6 +19,9 @@ type MultiConfig struct { // MaxInstances is the maximum number of concurrent instances MaxInstances int + // DefaultProtocol is the default protocol for agents + DefaultProtocol agent.Protocol + // DefaultAgentCommand is the default command for agents DefaultAgentCommand string @@ -44,6 +49,7 @@ func LoadMulti() *MultiConfig { InstancePortBase: getEnvInt("AGENTCTL_INSTANCE_PORT_BASE", 10001), InstancePortMax: getEnvInt("AGENTCTL_INSTANCE_PORT_MAX", 10100), MaxInstances: getEnvInt("AGENTCTL_MAX_INSTANCES", 10), + DefaultProtocol: agent.Protocol(getEnv("AGENTCTL_PROTOCOL", string(agent.ProtocolACP))), DefaultAgentCommand: getEnv("AGENTCTL_AGENT_COMMAND", "auggie --acp"), DefaultWorkDir: getEnv("AGENTCTL_WORKDIR", "/workspace"), AutoApprovePermissions: getEnvBool("AGENTCTL_AUTO_APPROVE_PERMISSIONS", false), diff --git a/apps/backend/internal/agentctl/process/manager.go b/apps/backend/internal/agentctl/process/manager.go index 8ac43661ac..51ee114f24 100644 --- a/apps/backend/internal/agentctl/process/manager.go +++ b/apps/backend/internal/agentctl/process/manager.go @@ -6,22 +6,18 @@ import ( "context" "fmt" "io" - "log/slog" "os/exec" "sync" "sync/atomic" "time" - "github.com/coder/acp-go-sdk" - acpclient "github.com/kandev/kandev/internal/agentctl/acp" + "github.com/kandev/kandev/internal/agentctl/adapter" "github.com/kandev/kandev/internal/agentctl/config" "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/pkg/agent" "go.uber.org/zap" ) -// Compile-time check that we export our client -var _ acp.Client = (*acpclient.Client)(nil) - // Status represents the agent process status type Status string @@ -42,19 +38,19 @@ type errorWrapper struct { // PendingPermission represents a permission request waiting for user response type PendingPermission struct { ID string - Request *acpclient.PermissionRequest - ResponseCh chan *acpclient.PermissionResponse + Request *adapter.PermissionRequest + ResponseCh chan *adapter.PermissionResponse CreatedAt time.Time } // PermissionNotification is sent when the agent requests permission type PermissionNotification struct { - PendingID string `json:"pending_id"` - SessionID string `json:"session_id"` - ToolCallID string `json:"tool_call_id"` - Title string `json:"title"` - Options []acpclient.PermissionOption `json:"options"` - CreatedAt time.Time `json:"created_at"` + PendingID string `json:"pending_id"` + SessionID string `json:"session_id"` + ToolCallID string `json:"tool_call_id"` + Title string `json:"title"` + Options []adapter.PermissionOption `json:"options"` + CreatedAt time.Time `json:"created_at"` } // Manager manages the agent subprocess @@ -74,13 +70,11 @@ type Manager struct { // Workspace tracker for git status and file changes workspaceTracker *WorkspaceTracker - // ACP SDK connection - acpClient *acpclient.Client - acpConn *acp.ClientSideConnection - sessionID acp.SessionId + // Protocol adapter for agent communication + adapter adapter.AgentAdapter - // Session update notifications - updatesCh chan acp.SessionNotification + // Session update notifications (protocol-agnostic) + updatesCh chan adapter.SessionUpdate // Permission request notifications (sent to backend) permissionCh chan *PermissionNotification @@ -103,7 +97,7 @@ func NewManager(cfg *config.Config, log *logger.Logger) *Manager { cfg: cfg, logger: log.WithFields(zap.String("component", "process-manager")), workspaceTracker: NewWorkspaceTracker(cfg.WorkDir, log), - updatesCh: make(chan acp.SessionNotification, 100), + updatesCh: make(chan adapter.SessionUpdate, 100), permissionCh: make(chan *PermissionNotification, 10), pendingPermissions: make(map[string]*PendingPermission), } @@ -147,6 +141,7 @@ func (m *Manager) Start(ctx context.Context) error { } m.logger.Info("starting agent process", + zap.String("protocol", string(m.cfg.Protocol)), zap.Strings("args", m.cfg.AgentArgs), zap.String("workdir", m.cfg.WorkDir)) @@ -195,29 +190,21 @@ func (m *Manager) Start(ctx context.Context) error { m.stopCh = make(chan struct{}) m.doneCh = make(chan struct{}) - // Create ACP client that will handle agent requests - m.acpClient = acpclient.NewClient( - acpclient.WithLogger(m.logger.Zap()), - acpclient.WithWorkspaceRoot(m.cfg.WorkDir), - acpclient.WithUpdateHandler(func(n acp.SessionNotification) { - select { - case m.updatesCh <- n: - default: - m.logger.Warn("updates channel full, dropping notification") - } - }), - acpclient.WithPermissionHandler(m.handlePermissionRequest), - ) - - // Create ACP SDK connection - it handles JSON-RPC communication - m.acpConn = acp.NewClientSideConnection(m.acpClient, m.stdin, m.stdout) - m.acpConn.SetLogger(slog.Default().With("component", "acp-conn")) + // Create protocol adapter based on configuration + if err := m.createAdapter(); err != nil { + m.status.Store(StatusError) + return fmt.Errorf("failed to create adapter: %w", err) + } // Start stderr reader and exit waiter m.wg.Add(2) go m.readStderr() go m.waitForExit() + // Forward adapter updates to our channel + m.wg.Add(1) + go m.forwardUpdates() + // Start workspace tracker with background context (not tied to HTTP request) m.workspaceTracker.Start(context.Background()) @@ -227,30 +214,76 @@ func (m *Manager) Start(ctx context.Context) error { return nil } +// createAdapter creates the appropriate protocol adapter based on configuration +func (m *Manager) createAdapter() error { + protocol := m.cfg.Protocol + if protocol == "" { + protocol = agent.ProtocolACP // Default to ACP + } + + adapterCfg := &adapter.Config{ + WorkDir: m.cfg.WorkDir, + AutoApprove: m.cfg.AutoApprovePermissions, + } + + switch protocol { + case agent.ProtocolACP: + m.adapter = adapter.NewACPAdapter(m.stdin, m.stdout, adapterCfg, m.logger) + default: + return fmt.Errorf("unsupported protocol: %s", protocol) + } + + // Set the permission handler + m.adapter.SetPermissionHandler(m.handlePermissionRequest) + + return nil +} + +// forwardUpdates forwards updates from the adapter to the manager's channel +func (m *Manager) forwardUpdates() { + defer m.wg.Done() + + updatesCh := m.adapter.Updates() + for { + select { + case update, ok := <-updatesCh: + if !ok { + return + } + select { + case m.updatesCh <- update: + default: + m.logger.Warn("updates channel full, dropping notification") + } + case <-m.stopCh: + return + } + } +} + // GetUpdates returns the channel for session update notifications -func (m *Manager) GetUpdates() <-chan acp.SessionNotification { +func (m *Manager) GetUpdates() <-chan adapter.SessionUpdate { return m.updatesCh } -// GetConnection returns the ACP connection (for advanced usage) -func (m *Manager) GetConnection() *acp.ClientSideConnection { +// GetAdapter returns the protocol adapter +func (m *Manager) GetAdapter() adapter.AgentAdapter { m.mu.RLock() defer m.mu.RUnlock() - return m.acpConn + return m.adapter } -// GetSessionID returns the current session ID -func (m *Manager) GetSessionID() acp.SessionId { +// GetSessionID returns the current session ID from the adapter. +// The adapter is the single source of truth for session ID. +func (m *Manager) GetSessionID() string { m.mu.RLock() - defer m.mu.RUnlock() - return m.sessionID -} + a := m.adapter + m.mu.RUnlock() -// SetSessionID sets the current session ID -func (m *Manager) SetSessionID(id acp.SessionId) { - m.mu.Lock() - defer m.mu.Unlock() - m.sessionID = id + if a != nil { + return a.GetSessionID() + } + return "" } // Stop stops the agent process @@ -271,6 +304,11 @@ func (m *Manager) Stop(ctx context.Context) error { m.workspaceTracker.Stop() } + // Close the adapter + if m.adapter != nil { + m.adapter.Close() + } + // Close stop channel to signal readers if m.stopCh != nil { close(m.stopCh) @@ -303,8 +341,6 @@ func (m *Manager) Stop(ctx context.Context) error { return nil } - - // readStderr reads and logs stderr from the agent func (m *Manager) readStderr() { defer m.wg.Done() @@ -341,8 +377,6 @@ func (m *Manager) waitForExit() { m.status.Store(StatusStopped) } - - // GetProcessInfo returns information about the process func (m *Manager) GetProcessInfo() map[string]interface{} { info := map[string]interface{}{ @@ -363,7 +397,7 @@ func (m *Manager) GetProcessInfo() map[string]interface{} { // handlePermissionRequest handles permission requests from the agent // It stores the pending request and waits for a response from the backend -func (m *Manager) handlePermissionRequest(ctx context.Context, req *acpclient.PermissionRequest) (*acpclient.PermissionResponse, error) { +func (m *Manager) handlePermissionRequest(ctx context.Context, req *adapter.PermissionRequest) (*adapter.PermissionResponse, error) { // Generate a unique ID for this permission request pendingID := fmt.Sprintf("%s-%s-%d", req.SessionID, req.ToolCallID, time.Now().UnixNano()) @@ -383,7 +417,7 @@ func (m *Manager) handlePermissionRequest(ctx context.Context, req *acpclient.Pe pending := &PendingPermission{ ID: pendingID, Request: req, - ResponseCh: make(chan *acpclient.PermissionResponse, 1), + ResponseCh: make(chan *adapter.PermissionResponse, 1), CreatedAt: time.Now(), } @@ -414,24 +448,24 @@ func (m *Manager) handlePermissionRequest(ctx context.Context, req *acpclient.Pe case <-ctx.Done(): m.logger.Warn("permission request context cancelled", zap.String("pending_id", pendingID)) - return &acpclient.PermissionResponse{Cancelled: true}, nil + return &adapter.PermissionResponse{Cancelled: true}, nil case <-time.After(5 * time.Minute): m.logger.Warn("permission request timed out", zap.String("pending_id", pendingID)) - return &acpclient.PermissionResponse{Cancelled: true}, nil + return &adapter.PermissionResponse{Cancelled: true}, nil } } // autoApprovePermission automatically approves a permission request // by selecting the first "allow" option, or the first option if no allow option exists -func (m *Manager) autoApprovePermission(req *acpclient.PermissionRequest) (*acpclient.PermissionResponse, error) { +func (m *Manager) autoApprovePermission(req *adapter.PermissionRequest) (*adapter.PermissionResponse, error) { if len(req.Options) == 0 { m.logger.Warn("no options available for auto-approve, cancelling") - return &acpclient.PermissionResponse{Cancelled: true}, nil + return &adapter.PermissionResponse{Cancelled: true}, nil } // Find the first "allow" option - var selectedOption *acpclient.PermissionOption + var selectedOption *adapter.PermissionOption for i := range req.Options { opt := &req.Options[i] if opt.Kind == "allow_once" || opt.Kind == "allow_always" { @@ -450,7 +484,7 @@ func (m *Manager) autoApprovePermission(req *acpclient.PermissionRequest) (*acpc zap.String("option_name", selectedOption.Name), zap.String("kind", selectedOption.Kind)) - return &acpclient.PermissionResponse{ + return &adapter.PermissionResponse{ OptionID: selectedOption.OptionID, }, nil } @@ -501,7 +535,7 @@ func (m *Manager) RespondToPermission(pendingID string, optionID string, cancell // Send response (non-blocking since channel is buffered) select { - case pending.ResponseCh <- &acpclient.PermissionResponse{ + case pending.ResponseCh <- &adapter.PermissionResponse{ OptionID: optionID, Cancelled: cancelled, }: diff --git a/apps/backend/internal/agentctl/types/types.go b/apps/backend/internal/agentctl/types/types.go new file mode 100644 index 0000000000..8d63e7b3ff --- /dev/null +++ b/apps/backend/internal/agentctl/types/types.go @@ -0,0 +1,30 @@ +// Package types provides shared types for agentctl packages. +// This breaks import cycles between adapter and acp packages. +package types + +import "context" + +// PermissionRequest represents a permission request from the agent +type PermissionRequest struct { + SessionID string `json:"session_id"` + ToolCallID string `json:"tool_call_id"` + Title string `json:"title"` + Options []PermissionOption `json:"options"` +} + +// PermissionOption represents a permission choice +type PermissionOption struct { + OptionID string `json:"option_id"` + Name string `json:"name"` + Kind string `json:"kind"` // allow_once, allow_always, reject_once, reject_always +} + +// PermissionResponse is the user's response to a permission request +type PermissionResponse struct { + OptionID string `json:"option_id,omitempty"` + Cancelled bool `json:"cancelled,omitempty"` +} + +// PermissionHandler is called when the agent requests permission for an action +type PermissionHandler func(ctx context.Context, req *PermissionRequest) (*PermissionResponse, error) + diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index f67a76d01b..fd12342ddc 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -107,7 +107,7 @@ func FromAgentSession(s *models.AgentSession) *TaskExecution { return &TaskExecution{ TaskID: s.TaskID, AgentInstanceID: s.AgentInstanceID, - AgentProfileID: s.AgentType, // AgentSession still uses AgentType field in DB + AgentProfileID: s.AgentProfileID, StartedAt: s.StartedAt, Status: agentSessionStatusToV1(s.Status), Progress: s.Progress, @@ -328,7 +328,7 @@ func (e *Executor) Execute(ctx context.Context, task *v1.Task) (*TaskExecution, TaskID: task.ID, AgentInstanceID: resp.AgentInstanceID, ContainerID: resp.ContainerID, - AgentType: *task.AgentProfileID, // AgentSession still uses AgentType field in DB + AgentProfileID: *task.AgentProfileID, Status: v1StatusToAgentSessionStatus(resp.Status), Progress: 0, StartedAt: now, diff --git a/apps/backend/internal/task/models/models.go b/apps/backend/internal/task/models/models.go index 6708820db0..0a6c54ebda 100644 --- a/apps/backend/internal/task/models/models.go +++ b/apps/backend/internal/task/models/models.go @@ -152,7 +152,7 @@ type AgentSession struct { TaskID string `json:"task_id"` AgentInstanceID string `json:"agent_instance_id"` // Docker container/agent instance ContainerID string `json:"container_id"` // Docker container ID for cleanup - AgentType string `json:"agent_type"` // e.g., "augment-agent" + AgentProfileID string `json:"agent_profile_id"` // ID of the agent profile used ACPSessionID string `json:"acp_session_id"` // ACP protocol session for resumption ExecutorID string `json:"executor_id"` EnvironmentID string `json:"environment_id"` @@ -173,7 +173,7 @@ func (s *AgentSession) ToAPI() map[string]interface{} { "task_id": s.TaskID, "agent_instance_id": s.AgentInstanceID, "container_id": s.ContainerID, - "agent_type": s.AgentType, + "agent_profile_id": s.AgentProfileID, "acp_session_id": s.ACPSessionID, "executor_id": s.ExecutorID, "environment_id": s.EnvironmentID, diff --git a/apps/backend/internal/task/repository/sqlite.go b/apps/backend/internal/task/repository/sqlite.go index 36b11fac59..d8e0856bb5 100644 --- a/apps/backend/internal/task/repository/sqlite.go +++ b/apps/backend/internal/task/repository/sqlite.go @@ -242,7 +242,7 @@ func (r *SQLiteRepository) initSchema() error { task_id TEXT NOT NULL, agent_instance_id TEXT NOT NULL DEFAULT '', container_id TEXT NOT NULL DEFAULT '', - agent_type TEXT NOT NULL, + agent_profile_id TEXT NOT NULL, acp_session_id TEXT DEFAULT '', executor_id TEXT DEFAULT '', environment_id TEXT DEFAULT '', @@ -1570,10 +1570,10 @@ func (r *SQLiteRepository) CreateAgentSession(ctx context.Context, session *mode _, err = r.db.ExecContext(ctx, ` INSERT INTO agent_sessions ( - id, task_id, agent_instance_id, container_id, agent_type, acp_session_id, executor_id, environment_id, + id, task_id, agent_instance_id, container_id, agent_profile_id, acp_session_id, executor_id, environment_id, status, progress, error_message, metadata, started_at, completed_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - `, session.ID, session.TaskID, session.AgentInstanceID, session.ContainerID, session.AgentType, session.ACPSessionID, + `, session.ID, session.TaskID, session.AgentInstanceID, session.ContainerID, session.AgentProfileID, session.ACPSessionID, session.ExecutorID, session.EnvironmentID, string(session.Status), session.Progress, session.ErrorMessage, string(metadataJSON), session.StartedAt, session.CompletedAt, session.UpdatedAt) @@ -1588,11 +1588,11 @@ func (r *SQLiteRepository) GetAgentSession(ctx context.Context, id string) (*mod var completedAt sql.NullTime err := r.db.QueryRowContext(ctx, ` - SELECT id, task_id, agent_instance_id, container_id, agent_type, acp_session_id, executor_id, environment_id, + SELECT id, task_id, agent_instance_id, container_id, agent_profile_id, acp_session_id, executor_id, environment_id, status, progress, error_message, metadata, started_at, completed_at, updated_at FROM agent_sessions WHERE id = ? `, id).Scan( - &session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentType, + &session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentProfileID, &session.ACPSessionID, &session.ExecutorID, &session.EnvironmentID, &status, &session.Progress, &session.ErrorMessage, &metadataJSON, &session.StartedAt, &completedAt, &session.UpdatedAt, ) @@ -1625,11 +1625,11 @@ func (r *SQLiteRepository) GetAgentSessionByTaskID(ctx context.Context, taskID s var completedAt sql.NullTime err := r.db.QueryRowContext(ctx, ` - SELECT id, task_id, agent_instance_id, container_id, agent_type, acp_session_id, executor_id, environment_id, + SELECT id, task_id, agent_instance_id, container_id, agent_profile_id, acp_session_id, executor_id, environment_id, status, progress, error_message, metadata, started_at, completed_at, updated_at FROM agent_sessions WHERE task_id = ? ORDER BY started_at DESC LIMIT 1 `, taskID).Scan( - &session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentType, + &session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentProfileID, &session.ACPSessionID, &session.ExecutorID, &session.EnvironmentID, &status, &session.Progress, &session.ErrorMessage, &metadataJSON, &session.StartedAt, &completedAt, &session.UpdatedAt, ) @@ -1662,13 +1662,13 @@ func (r *SQLiteRepository) GetActiveAgentSessionByTaskID(ctx context.Context, ta var completedAt sql.NullTime err := r.db.QueryRowContext(ctx, ` - SELECT id, task_id, agent_instance_id, container_id, agent_type, acp_session_id, executor_id, environment_id, + SELECT id, task_id, agent_instance_id, container_id, agent_profile_id, acp_session_id, executor_id, environment_id, status, progress, error_message, metadata, started_at, completed_at, updated_at FROM agent_sessions WHERE task_id = ? AND status IN ('pending', 'running', 'waiting') ORDER BY started_at DESC LIMIT 1 `, taskID).Scan( - &session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentType, + &session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentProfileID, &session.ACPSessionID, &session.ExecutorID, &session.EnvironmentID, &status, &session.Progress, &session.ErrorMessage, &metadataJSON, &session.StartedAt, &completedAt, &session.UpdatedAt, ) @@ -1704,10 +1704,10 @@ func (r *SQLiteRepository) UpdateAgentSession(ctx context.Context, session *mode result, err := r.db.ExecContext(ctx, ` UPDATE agent_sessions SET - agent_instance_id = ?, container_id = ?, agent_type = ?, acp_session_id = ?, executor_id = ?, environment_id = ?, + agent_instance_id = ?, container_id = ?, agent_profile_id = ?, acp_session_id = ?, executor_id = ?, environment_id = ?, status = ?, progress = ?, error_message = ?, metadata = ?, completed_at = ?, updated_at = ? WHERE id = ? - `, session.AgentInstanceID, session.ContainerID, session.AgentType, session.ACPSessionID, session.ExecutorID, session.EnvironmentID, + `, session.AgentInstanceID, session.ContainerID, session.AgentProfileID, session.ACPSessionID, session.ExecutorID, session.EnvironmentID, string(session.Status), session.Progress, session.ErrorMessage, string(metadataJSON), session.CompletedAt, session.UpdatedAt, session.ID) if err != nil { @@ -1747,7 +1747,7 @@ func (r *SQLiteRepository) UpdateAgentSessionStatus(ctx context.Context, id stri // ListAgentSessions returns all agent sessions for a task func (r *SQLiteRepository) ListAgentSessions(ctx context.Context, taskID string) ([]*models.AgentSession, error) { rows, err := r.db.QueryContext(ctx, ` - SELECT id, task_id, agent_instance_id, container_id, agent_type, acp_session_id, executor_id, environment_id, + SELECT id, task_id, agent_instance_id, container_id, agent_profile_id, acp_session_id, executor_id, environment_id, status, progress, error_message, metadata, started_at, completed_at, updated_at FROM agent_sessions WHERE task_id = ? ORDER BY started_at DESC `, taskID) @@ -1762,7 +1762,7 @@ func (r *SQLiteRepository) ListAgentSessions(ctx context.Context, taskID string) // ListActiveAgentSessions returns all active agent sessions across all tasks func (r *SQLiteRepository) ListActiveAgentSessions(ctx context.Context) ([]*models.AgentSession, error) { rows, err := r.db.QueryContext(ctx, ` - SELECT id, task_id, agent_instance_id, container_id, agent_type, acp_session_id, executor_id, environment_id, + SELECT id, task_id, agent_instance_id, container_id, agent_profile_id, acp_session_id, executor_id, environment_id, status, progress, error_message, metadata, started_at, completed_at, updated_at FROM agent_sessions WHERE status IN ('pending', 'running', 'waiting') ORDER BY started_at DESC `) @@ -1784,7 +1784,7 @@ func (r *SQLiteRepository) scanAgentSessions(rows *sql.Rows) ([]*models.AgentSes var completedAt sql.NullTime err := rows.Scan( - &session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentType, + &session.ID, &session.TaskID, &session.AgentInstanceID, &session.ContainerID, &session.AgentProfileID, &session.ACPSessionID, &session.ExecutorID, &session.EnvironmentID, &status, &session.Progress, &session.ErrorMessage, &metadataJSON, &session.StartedAt, &completedAt, &session.UpdatedAt, ) diff --git a/apps/backend/internal/task/repository/sqlite_test.go b/apps/backend/internal/task/repository/sqlite_test.go index 40e2c101ec..d62febbadd 100644 --- a/apps/backend/internal/task/repository/sqlite_test.go +++ b/apps/backend/internal/task/repository/sqlite_test.go @@ -632,3 +632,320 @@ func TestSQLiteRepository_CommentWithRequestsInput(t *testing.T) { t.Errorf("expected ACPSessionID 'session-abc', got %s", retrieved.ACPSessionID) } } + +// AgentSession CRUD tests + +func TestSQLiteRepository_AgentSessionCRUD(t *testing.T) { + repo, cleanup := createTestSQLiteRepo(t) + defer cleanup() + ctx := context.Background() + + // Create board, column, and task first (required for foreign key constraints) + board := &models.Board{ID: "board-123", Name: "Test Board"} + _ = repo.CreateBoard(ctx, board) + column := &models.Column{ID: "col-123", BoardID: "board-123", Name: "Test Column"} + _ = repo.CreateColumn(ctx, column) + task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"} + _ = repo.CreateTask(ctx, task) + + // Create agent session + session := &models.AgentSession{ + TaskID: "task-123", + AgentInstanceID: "instance-abc", + ContainerID: "container-xyz", + AgentProfileID: "profile-123", + ACPSessionID: "acp-session-123", + ExecutorID: "executor-1", + EnvironmentID: "env-1", + Status: models.AgentSessionStatusPending, + Progress: 0, + Metadata: map[string]interface{}{"key": "value"}, + } + if err := repo.CreateAgentSession(ctx, session); err != nil { + t.Fatalf("failed to create agent session: %v", err) + } + if session.ID == "" { + t.Error("expected session ID to be set") + } + if session.StartedAt.IsZero() { + t.Error("expected StartedAt to be set") + } + if session.UpdatedAt.IsZero() { + t.Error("expected UpdatedAt to be set") + } + + // Get agent session + retrieved, err := repo.GetAgentSession(ctx, session.ID) + if err != nil { + t.Fatalf("failed to get agent session: %v", err) + } + if retrieved.TaskID != "task-123" { + t.Errorf("expected TaskID 'task-123', got %s", retrieved.TaskID) + } + if retrieved.AgentProfileID != "profile-123" { + t.Errorf("expected AgentProfileID 'profile-123', got %s", retrieved.AgentProfileID) + } + if retrieved.Status != models.AgentSessionStatusPending { + t.Errorf("expected status 'pending', got %s", retrieved.Status) + } + if retrieved.Metadata["key"] != "value" { + t.Errorf("expected metadata key 'value', got %v", retrieved.Metadata["key"]) + } + + // Update agent session + session.Status = models.AgentSessionStatusRunning + session.Progress = 50 + if err := repo.UpdateAgentSession(ctx, session); err != nil { + t.Fatalf("failed to update agent session: %v", err) + } + retrieved, _ = repo.GetAgentSession(ctx, session.ID) + if retrieved.Status != models.AgentSessionStatusRunning { + t.Errorf("expected status 'running', got %s", retrieved.Status) + } + if retrieved.Progress != 50 { + t.Errorf("expected progress 50, got %d", retrieved.Progress) + } + + // Delete agent session + if err := repo.DeleteAgentSession(ctx, session.ID); err != nil { + t.Fatalf("failed to delete agent session: %v", err) + } + _, err = repo.GetAgentSession(ctx, session.ID) + if err == nil { + t.Error("expected agent session to be deleted") + } +} + +func TestSQLiteRepository_AgentSessionNotFound(t *testing.T) { + repo, cleanup := createTestSQLiteRepo(t) + defer cleanup() + ctx := context.Background() + + _, err := repo.GetAgentSession(ctx, "nonexistent") + if err == nil { + t.Error("expected error for nonexistent agent session") + } + + err = repo.UpdateAgentSession(ctx, &models.AgentSession{ID: "nonexistent", TaskID: "task-123"}) + if err == nil { + t.Error("expected error for updating nonexistent agent session") + } + + err = repo.DeleteAgentSession(ctx, "nonexistent") + if err == nil { + t.Error("expected error for deleting nonexistent agent session") + } +} + +func TestSQLiteRepository_AgentSessionByTaskID(t *testing.T) { + repo, cleanup := createTestSQLiteRepo(t) + defer cleanup() + ctx := context.Background() + + // Create board, column, and task + board := &models.Board{ID: "board-123", Name: "Test Board"} + _ = repo.CreateBoard(ctx, board) + column := &models.Column{ID: "col-123", BoardID: "board-123", Name: "Test Column"} + _ = repo.CreateColumn(ctx, column) + task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"} + _ = repo.CreateTask(ctx, task) + + // Create multiple sessions for the same task (simulating session history) + session1 := &models.AgentSession{ + ID: "session-1", + TaskID: "task-123", + AgentProfileID: "profile-1", + Status: models.AgentSessionStatusCompleted, + } + _ = repo.CreateAgentSession(ctx, session1) + + session2 := &models.AgentSession{ + ID: "session-2", + TaskID: "task-123", + AgentProfileID: "profile-2", + Status: models.AgentSessionStatusRunning, + } + _ = repo.CreateAgentSession(ctx, session2) + + // GetAgentSessionByTaskID should return the most recent session + retrieved, err := repo.GetAgentSessionByTaskID(ctx, "task-123") + if err != nil { + t.Fatalf("failed to get agent session by task ID: %v", err) + } + if retrieved.ID != "session-2" { + t.Errorf("expected session-2 (most recent), got %s", retrieved.ID) + } + + // GetActiveAgentSessionByTaskID should return the active session + active, err := repo.GetActiveAgentSessionByTaskID(ctx, "task-123") + if err != nil { + t.Fatalf("failed to get active agent session by task ID: %v", err) + } + if active.ID != "session-2" { + t.Errorf("expected session-2 (active), got %s", active.ID) + } + if active.Status != models.AgentSessionStatusRunning { + t.Errorf("expected status 'running', got %s", active.Status) + } + + // Test when no active session exists + session2.Status = models.AgentSessionStatusCompleted + _ = repo.UpdateAgentSession(ctx, session2) + + _, err = repo.GetActiveAgentSessionByTaskID(ctx, "task-123") + if err == nil { + t.Error("expected error when no active session exists") + } + + // Test for nonexistent task + _, err = repo.GetAgentSessionByTaskID(ctx, "nonexistent-task") + if err == nil { + t.Error("expected error for nonexistent task") + } +} + +func TestSQLiteRepository_ListAgentSessions(t *testing.T) { + repo, cleanup := createTestSQLiteRepo(t) + defer cleanup() + ctx := context.Background() + + // Create board, column, and tasks + board := &models.Board{ID: "board-123", Name: "Test Board"} + _ = repo.CreateBoard(ctx, board) + column := &models.Column{ID: "col-123", BoardID: "board-123", Name: "Test Column"} + _ = repo.CreateColumn(ctx, column) + task1 := &models.Task{ID: "task-1", BoardID: "board-123", ColumnID: "col-123", Title: "Task 1"} + _ = repo.CreateTask(ctx, task1) + task2 := &models.Task{ID: "task-2", BoardID: "board-123", ColumnID: "col-123", Title: "Task 2"} + _ = repo.CreateTask(ctx, task2) + + // Create sessions for different tasks + _ = repo.CreateAgentSession(ctx, &models.AgentSession{ID: "session-1", TaskID: "task-1", AgentProfileID: "profile-1", Status: models.AgentSessionStatusCompleted}) + _ = repo.CreateAgentSession(ctx, &models.AgentSession{ID: "session-2", TaskID: "task-1", AgentProfileID: "profile-1", Status: models.AgentSessionStatusRunning}) + _ = repo.CreateAgentSession(ctx, &models.AgentSession{ID: "session-3", TaskID: "task-2", AgentProfileID: "profile-2", Status: models.AgentSessionStatusPending}) + + // List sessions for task-1 + sessions, err := repo.ListAgentSessions(ctx, "task-1") + if err != nil { + t.Fatalf("failed to list agent sessions: %v", err) + } + if len(sessions) != 2 { + t.Errorf("expected 2 sessions for task-1, got %d", len(sessions)) + } + + // List all active sessions + activeSessions, err := repo.ListActiveAgentSessions(ctx) + if err != nil { + t.Fatalf("failed to list active agent sessions: %v", err) + } + if len(activeSessions) != 2 { + t.Errorf("expected 2 active sessions, got %d", len(activeSessions)) + } + + // Verify only active statuses are returned + for _, s := range activeSessions { + if s.Status != models.AgentSessionStatusPending && s.Status != models.AgentSessionStatusRunning && s.Status != models.AgentSessionStatusWaiting { + t.Errorf("expected active status, got %s", s.Status) + } + } +} + +func TestSQLiteRepository_UpdateAgentSessionStatus(t *testing.T) { + repo, cleanup := createTestSQLiteRepo(t) + defer cleanup() + ctx := context.Background() + + // Create board, column, and task + board := &models.Board{ID: "board-123", Name: "Test Board"} + _ = repo.CreateBoard(ctx, board) + column := &models.Column{ID: "col-123", BoardID: "board-123", Name: "Test Column"} + _ = repo.CreateColumn(ctx, column) + task := &models.Task{ID: "task-123", BoardID: "board-123", ColumnID: "col-123", Title: "Test Task"} + _ = repo.CreateTask(ctx, task) + + // Create an agent session + session := &models.AgentSession{ + ID: "session-123", + TaskID: "task-123", + AgentProfileID: "profile-1", + Status: models.AgentSessionStatusPending, + } + _ = repo.CreateAgentSession(ctx, session) + + // Update to running status + err := repo.UpdateAgentSessionStatus(ctx, "session-123", models.AgentSessionStatusRunning, "") + if err != nil { + t.Fatalf("failed to update agent session status: %v", err) + } + retrieved, _ := repo.GetAgentSession(ctx, "session-123") + if retrieved.Status != models.AgentSessionStatusRunning { + t.Errorf("expected status 'running', got %s", retrieved.Status) + } + if retrieved.CompletedAt != nil { + t.Error("expected CompletedAt to be nil for running status") + } + + // Update to completed status (should set CompletedAt) + err = repo.UpdateAgentSessionStatus(ctx, "session-123", models.AgentSessionStatusCompleted, "") + if err != nil { + t.Fatalf("failed to update agent session status to completed: %v", err) + } + retrieved, _ = repo.GetAgentSession(ctx, "session-123") + if retrieved.Status != models.AgentSessionStatusCompleted { + t.Errorf("expected status 'completed', got %s", retrieved.Status) + } + if retrieved.CompletedAt == nil { + t.Error("expected CompletedAt to be set for completed status") + } + + // Test failed status with error message + session2 := &models.AgentSession{ + ID: "session-456", + TaskID: "task-123", + AgentProfileID: "profile-1", + Status: models.AgentSessionStatusRunning, + } + _ = repo.CreateAgentSession(ctx, session2) + + err = repo.UpdateAgentSessionStatus(ctx, "session-456", models.AgentSessionStatusFailed, "connection timeout") + if err != nil { + t.Fatalf("failed to update agent session status to failed: %v", err) + } + retrieved, _ = repo.GetAgentSession(ctx, "session-456") + if retrieved.Status != models.AgentSessionStatusFailed { + t.Errorf("expected status 'failed', got %s", retrieved.Status) + } + if retrieved.ErrorMessage != "connection timeout" { + t.Errorf("expected error message 'connection timeout', got %s", retrieved.ErrorMessage) + } + if retrieved.CompletedAt == nil { + t.Error("expected CompletedAt to be set for failed status") + } + + // Test stopped status + session3 := &models.AgentSession{ + ID: "session-789", + TaskID: "task-123", + AgentProfileID: "profile-1", + Status: models.AgentSessionStatusRunning, + } + _ = repo.CreateAgentSession(ctx, session3) + + err = repo.UpdateAgentSessionStatus(ctx, "session-789", models.AgentSessionStatusStopped, "") + if err != nil { + t.Fatalf("failed to update agent session status to stopped: %v", err) + } + retrieved, _ = repo.GetAgentSession(ctx, "session-789") + if retrieved.Status != models.AgentSessionStatusStopped { + t.Errorf("expected status 'stopped', got %s", retrieved.Status) + } + if retrieved.CompletedAt == nil { + t.Error("expected CompletedAt to be set for stopped status") + } + + // Test nonexistent session + err = repo.UpdateAgentSessionStatus(ctx, "nonexistent", models.AgentSessionStatusRunning, "") + if err == nil { + t.Error("expected error for updating nonexistent session status") + } +} diff --git a/apps/backend/pkg/agent/protocol.go b/apps/backend/pkg/agent/protocol.go new file mode 100644 index 0000000000..7c9bea955d --- /dev/null +++ b/apps/backend/pkg/agent/protocol.go @@ -0,0 +1,30 @@ +// Package agent provides shared types for agent communication. +package agent + +// Protocol defines the communication protocol used by an agent. +type Protocol string + +const ( + // ProtocolACP is the Agent Communication Protocol (JSON-RPC over stdin/stdout). + ProtocolACP Protocol = "acp" + // ProtocolREST is for agents with REST APIs. + ProtocolREST Protocol = "rest" + // ProtocolMCP is the Model Context Protocol. + ProtocolMCP Protocol = "mcp" +) + +// String returns the string representation of the protocol. +func (p Protocol) String() string { + return string(p) +} + +// IsValid returns true if the protocol is a known valid protocol. +func (p Protocol) IsValid() bool { + switch p { + case ProtocolACP, ProtocolREST, ProtocolMCP: + return true + default: + return false + } +} +