feat(agentctl): protocol adapter abstraction with agent profile support (#15)

* feat(agentctl): implement protocol adapter abstraction

- Add AgentAdapter interface for protocol-agnostic agent communication
- Create ACPAdapter implementation for ACP protocol (JSON-RPC over stdin/stdout)
- Add Protocol field to agent registry (AgentTypeConfig)
- Update process.Manager to use adapters instead of hardcoded ACP
- Update API layer to use adapter interface
- Pass protocol from registry to agentctl via CLI flag and env var
- Support protocol configuration in both Docker and standalone modes

This refactoring allows agentctl to support multiple protocols (ACP, REST, MCP)
via a pluggable adapter pattern. The backend remains protocol-agnostic,
always communicating with agentctl's HTTP API.

* refactor(agentctl): consolidate types and clean up code

- Create shared types package (internal/agentctl/types) for permission types
- Create shared protocol package (pkg/agent) for Protocol type
- Remove duplicate permission types from acp/client.go and adapter.go
- Make adapter the single source of truth for session ID
- Remove unused Command field from adapter.Config
- Remove passthrough wrapper function handleAdapterPermissionRequest
- Clean up extra empty lines in process/manager.go and acp_adapter.go
- Re-export permission types from adapter package for convenience

* fix(agent-session): rename AgentType to AgentProfileID for consistency

- Rename AgentSession.AgentType to AgentProfileID in models
- Update agent_sessions table schema column from agent_type to agent_profile_id
- Update all SQL queries in sqlite.go to use agent_profile_id
- Fix ToAPI() method to return agent_profile_id key
- Remove misleading comments in executor.go

Also adds comprehensive tests for affected areas:
- adapter package unit tests (SessionUpdate, PlanEntry, AgentInfo, Config)
- profile resolver unit tests with mock repository
- AgentSession CRUD tests in sqlite_test.go
This commit is contained in:
João Salavisa
2026-01-14 18:16:26 +00:00
committed by GitHub
parent 3d01863d46
commit 85ecf7debb
22 changed files with 1759 additions and 272 deletions
+31 -12
View File
@@ -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, &notification); 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)
}
}()
@@ -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)
@@ -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"`
}
+102 -104
View File
@@ -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)
@@ -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())
}
}
@@ -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,
},
}
}
@@ -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
+6 -26
View File
@@ -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,
@@ -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)
@@ -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
}
@@ -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))
}
}
+36 -44
View File
@@ -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,
})
}
@@ -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),
@@ -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),
@@ -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,
}:
@@ -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)
@@ -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,
+2 -2
View File
@@ -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,
+14 -14
View File
@@ -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,
)
@@ -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")
}
}