From 0599800c40608fb81c2a7b50b664d0efe8a8cd0a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Fri, 16 Jan 2026 02:24:58 +0000 Subject: [PATCH] refactor(lifecycle): extract manager into focused components (#20) --- .../internal/agent/lifecycle/command.go | 81 ++++ .../internal/agent/lifecycle/container.go | 258 +++++++++++++ .../internal/agent/lifecycle/events.go | 353 ++++++++++++++++++ .../internal/agent/lifecycle/manager.go | 219 +++++------ .../agent/lifecycle/profile_resolver.go | 1 - .../agent/lifecycle/profile_resolver_test.go | 9 +- .../internal/agent/lifecycle/session.go | 169 +++++++++ .../internal/agent/lifecycle/streams.go | 185 +++++++++ .../internal/agent/registry/defaults.go | 22 +- .../internal/agent/registry/registry.go | 23 ++ 10 files changed, 1195 insertions(+), 125 deletions(-) create mode 100644 apps/backend/internal/agent/lifecycle/command.go create mode 100644 apps/backend/internal/agent/lifecycle/container.go create mode 100644 apps/backend/internal/agent/lifecycle/events.go create mode 100644 apps/backend/internal/agent/lifecycle/session.go create mode 100644 apps/backend/internal/agent/lifecycle/streams.go diff --git a/apps/backend/internal/agent/lifecycle/command.go b/apps/backend/internal/agent/lifecycle/command.go new file mode 100644 index 0000000000..e5545bb7f0 --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/command.go @@ -0,0 +1,81 @@ +// Package lifecycle manages agent instance lifecycles including tracking, +// state transitions, and cleanup. +package lifecycle + +import ( + "os" + "strings" + + "github.com/kandev/kandev/internal/agent/registry" +) + +// CommandBuilder builds agent commands from registry configuration +type CommandBuilder struct{} + +// NewCommandBuilder creates a new CommandBuilder +func NewCommandBuilder() *CommandBuilder { + return &CommandBuilder{} +} + +// CommandOptions contains options for building a command +type CommandOptions struct { + Model string // Model to use (appended via ModelFlag if set) + SessionID string // Session ID to resume (appended via SessionConfig.ResumeFlag if not ResumeViaACP) +} + +// BuildCommand builds a command slice from agent config and options +// Returns the command as a string slice ready for execution +func (cb *CommandBuilder) BuildCommand(agentConfig *registry.AgentTypeConfig, opts CommandOptions) []string { + // Start with base command from config + cmd := make([]string, len(agentConfig.Cmd)) + copy(cmd, agentConfig.Cmd) + + // Append model flag if agent supports it and model is specified + if agentConfig.ModelFlag != "" && opts.Model != "" { + cmd = append(cmd, agentConfig.ModelFlag, opts.Model) + } + + // Append session resume flag if: + // 1. Session ID is provided + // 2. Agent does NOT use ACP for session resumption (uses CLI flag instead) + // 3. Agent has a ResumeFlag configured + if opts.SessionID != "" && !agentConfig.SessionConfig.ResumeViaACP && agentConfig.SessionConfig.ResumeFlag != "" { + cmd = append(cmd, agentConfig.SessionConfig.ResumeFlag, opts.SessionID) + } + + return cmd +} + +// BuildCommandString builds a command as a single string (for standalone mode) +func (cb *CommandBuilder) BuildCommandString(agentConfig *registry.AgentTypeConfig, opts CommandOptions) string { + cmd := cb.BuildCommand(agentConfig, opts) + return strings.Join(cmd, " ") +} + +// ExpandSessionDir expands the session directory template from SessionConfig +// Replaces {home} with the user's home directory +// Returns empty string if no session directory is configured +func (cb *CommandBuilder) ExpandSessionDir(agentConfig *registry.AgentTypeConfig) string { + template := agentConfig.SessionConfig.SessionDirTemplate + if template == "" { + return "" + } + + homeDir, err := os.UserHomeDir() + if err != nil { + homeDir = "/tmp" + } + + result := strings.ReplaceAll(template, "{home}", homeDir) + + // Ensure the directory exists + _ = os.MkdirAll(result, 0755) + + return result +} + +// GetSessionDirTarget returns the container path for session directory mount +// Returns empty string if no session directory is configured +func (cb *CommandBuilder) GetSessionDirTarget(agentConfig *registry.AgentTypeConfig) string { + return agentConfig.SessionConfig.SessionDirTarget +} diff --git a/apps/backend/internal/agent/lifecycle/container.go b/apps/backend/internal/agent/lifecycle/container.go new file mode 100644 index 0000000000..5dab9f7618 --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/container.go @@ -0,0 +1,258 @@ +// Package lifecycle manages agent instance lifecycles including tracking, +// state transitions, and cleanup. +package lifecycle + +import ( + "context" + "fmt" + "strings" + "time" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/agent/agentctl" + "github.com/kandev/kandev/internal/agent/docker" + "github.com/kandev/kandev/internal/agent/registry" + "github.com/kandev/kandev/internal/common/logger" +) + +// ContainerConfig holds configuration for launching a Docker container +type ContainerConfig struct { + AgentConfig *registry.AgentTypeConfig + WorkspacePath string + TaskID string + TaskDescription string + Model string + SessionID string + Credentials map[string]string + ProfileInfo *AgentProfileInfo + InstanceID string + MainRepoGitDir string // Path to main repo's .git directory (for worktrees) +} + +// ContainerManager handles Docker container lifecycle operations +type ContainerManager struct { + dockerClient *docker.Client + commandBuilder *CommandBuilder + logger *logger.Logger + networkName string +} + +// NewContainerManager creates a new ContainerManager +func NewContainerManager(dockerClient *docker.Client, networkName string, log *logger.Logger) *ContainerManager { + return &ContainerManager{ + dockerClient: dockerClient, + commandBuilder: NewCommandBuilder(), + logger: log.WithFields(zap.String("component", "container-manager")), + networkName: networkName, + } +} + +// LaunchContainer creates and starts a Docker container for an agent +// Returns the container ID and agentctl client +func (cm *ContainerManager) LaunchContainer(ctx context.Context, config ContainerConfig) (string, *agentctl.Client, error) { + // Build container config + containerCfg, err := cm.buildContainerConfig(config) + if err != nil { + return "", nil, fmt.Errorf("failed to build container config: %w", err) + } + + // Create the container + containerID, err := cm.dockerClient.CreateContainer(ctx, containerCfg) + if err != nil { + return "", nil, fmt.Errorf("failed to create container: %w", err) + } + + // Start the container + if err := cm.dockerClient.StartContainer(ctx, containerID); err != nil { + _ = cm.dockerClient.RemoveContainer(ctx, containerID, true) + return "", nil, fmt.Errorf("failed to start container: %w", err) + } + + // Get container IP for agentctl communication + containerIP, err := cm.dockerClient.GetContainerIP(ctx, containerID) + if err != nil { + cm.logger.Warn("failed to get container IP, trying localhost", + zap.String("container_id", containerID), + zap.Error(err)) + containerIP = "127.0.0.1" + } + + // Create agentctl client pointing to the container + agentctlClient := agentctl.NewClient(containerIP, AgentCtlPort, cm.logger) + + cm.logger.Info("docker container launched", + zap.String("container_id", containerID), + zap.String("container_ip", containerIP), + zap.String("instance_id", config.InstanceID)) + + return containerID, agentctlClient, nil +} + +// StopContainer stops and removes a Docker container +func (cm *ContainerManager) StopContainer(ctx context.Context, containerID string, timeout time.Duration) error { + if containerID == "" { + return nil + } + + if err := cm.dockerClient.StopContainer(ctx, containerID, timeout); err != nil { + cm.logger.Warn("failed to stop container gracefully, forcing removal", + zap.String("container_id", containerID), + zap.Error(err)) + } + + if err := cm.dockerClient.RemoveContainer(ctx, containerID, true); err != nil { + return fmt.Errorf("failed to remove container: %w", err) + } + + cm.logger.Info("container stopped and removed", + zap.String("container_id", containerID)) + + return nil +} + +// buildContainerConfig builds the Docker container configuration +func (cm *ContainerManager) buildContainerConfig(config ContainerConfig) (docker.ContainerConfig, error) { + agentConfig := config.AgentConfig + + // Build image name with tag + imageName := agentConfig.Image + if agentConfig.Tag != "" { + imageName = fmt.Sprintf("%s:%s", agentConfig.Image, agentConfig.Tag) + } + + // Build command using CommandBuilder + cmdOpts := CommandOptions{ + Model: config.Model, + SessionID: config.SessionID, + } + cmd := cm.commandBuilder.BuildCommand(agentConfig, cmdOpts) + + // Expand mounts + mounts := cm.expandMounts(agentConfig.Mounts, config.WorkspacePath, agentConfig) + + // Add main repo .git directory mount for worktrees + if config.MainRepoGitDir != "" { + mounts = append(mounts, docker.MountConfig{ + Source: config.MainRepoGitDir, + Target: config.MainRepoGitDir, // Same path inside container + ReadOnly: false, + }) + cm.logger.Debug("added main repo .git directory mount for worktree", + zap.String("path", config.MainRepoGitDir)) + } + + // Build environment variables + env := cm.buildEnvVars(config) + + // Calculate resource limits + memoryBytes := agentConfig.ResourceLimits.MemoryMB * 1024 * 1024 + cpuQuota := int64(agentConfig.ResourceLimits.CPUCores * 100000) // Docker CPU quota + + containerName := fmt.Sprintf("kandev-agent-%s", config.InstanceID[:8]) + + return docker.ContainerConfig{ + Name: containerName, + Image: imageName, + Cmd: cmd, + Env: env, + WorkingDir: agentConfig.WorkingDir, + Mounts: mounts, + NetworkMode: cm.networkName, + Memory: memoryBytes, + CPUQuota: cpuQuota, + Labels: map[string]string{ + "kandev.managed": "true", + "kandev.instance_id": config.InstanceID, + "kandev.task_id": config.TaskID, + }, + AutoRemove: false, // We manage cleanup ourselves + }, nil +} + +// expandMounts expands mount templates with actual paths +func (cm *ContainerManager) expandMounts(templates []registry.MountTemplate, workspacePath string, agentConfig *registry.AgentTypeConfig) []docker.MountConfig { + mounts := make([]docker.MountConfig, 0, len(templates)+1) // +1 for potential session dir + + for _, mt := range templates { + // Skip workspace mounts if no workspace path is provided + if strings.Contains(mt.Source, "{workspace}") && workspacePath == "" { + cm.logger.Debug("skipping workspace mount - no workspace path provided", + zap.String("target", mt.Target)) + continue + } + + source := cm.expandMountSource(mt.Source, workspacePath) + mounts = append(mounts, docker.MountConfig{ + Source: source, + Target: mt.Target, + ReadOnly: mt.ReadOnly, + }) + } + + // Add session directory mount from SessionConfig + sessionDirSource := cm.commandBuilder.ExpandSessionDir(agentConfig) + sessionDirTarget := cm.commandBuilder.GetSessionDirTarget(agentConfig) + if sessionDirSource != "" && sessionDirTarget != "" { + mounts = append(mounts, docker.MountConfig{ + Source: sessionDirSource, + Target: sessionDirTarget, + ReadOnly: false, + }) + cm.logger.Debug("added session directory mount", + zap.String("source", sessionDirSource), + zap.String("target", sessionDirTarget)) + } + + return mounts +} + +// expandMountSource expands template variables in mount source paths +func (cm *ContainerManager) expandMountSource(source, workspacePath string) string { + result := source + result = strings.ReplaceAll(result, "{workspace}", workspacePath) + return result +} + +// buildEnvVars builds environment variables for the container +func (cm *ContainerManager) buildEnvVars(config ContainerConfig) []string { + agentConfig := config.AgentConfig + env := make([]string, 0) + + // Add default env from agent config + for k, v := range agentConfig.Env { + env = append(env, fmt.Sprintf("%s=%s", k, v)) + } + + // Add standard kandev env vars + env = append(env, + fmt.Sprintf("KANDEV_TASK_ID=%s", config.TaskID), + fmt.Sprintf("KANDEV_INSTANCE_ID=%s", config.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) + env = append(env, + "GIT_CONFIG_COUNT=1", + "GIT_CONFIG_KEY_0=safe.directory", + "GIT_CONFIG_VALUE_0=*", // Trust all directories in the container + ) + + // Inject credentials from the provided credentials map + for k, v := range config.Credentials { + env = append(env, fmt.Sprintf("%s=%s", k, v)) + } + + // Add profile-specific label if available + if config.ProfileInfo != nil && config.ProfileInfo.ProfileID != "" { + env = append(env, fmt.Sprintf("KANDEV_AGENT_PROFILE_ID=%s", config.ProfileInfo.ProfileID)) + } + + return env +} diff --git a/apps/backend/internal/agent/lifecycle/events.go b/apps/backend/internal/agent/lifecycle/events.go new file mode 100644 index 0000000000..2367d82d00 --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/events.go @@ -0,0 +1,353 @@ +// Package lifecycle manages agent instance lifecycles including tracking, +// state transitions, and cleanup. +package lifecycle + +import ( + "context" + "time" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/agent/agentctl" + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" +) + +// EventPublisher handles publishing agent lifecycle and session events to the event bus. +type EventPublisher struct { + eventBus bus.EventBus + logger *logger.Logger +} + +// NewEventPublisher creates a new EventPublisher with the given event bus and logger. +func NewEventPublisher(eventBus bus.EventBus, log *logger.Logger) *EventPublisher { + return &EventPublisher{ + eventBus: eventBus, + logger: log.WithFields(zap.String("component", "event-publisher")), + } +} + +// PublishAgentEvent publishes an agent lifecycle event (started, stopped, ready, completed, failed). +func (p *EventPublisher) PublishAgentEvent(ctx context.Context, eventType string, instance *AgentInstance) { + if p.eventBus == nil { + return + } + + data := map[string]interface{}{ + "instance_id": instance.ID, + "task_id": instance.TaskID, + "agent_profile_id": instance.AgentProfileID, + "container_id": instance.ContainerID, + "status": string(instance.Status), + "started_at": instance.StartedAt, + "progress": instance.Progress, + "error_message": instance.ErrorMessage, + } + + if instance.FinishedAt != nil { + data["finished_at"] = *instance.FinishedAt + } + if instance.ExitCode != nil { + data["exit_code"] = *instance.ExitCode + } + + event := bus.NewEvent(eventType, "agent-manager", data) + + if err := p.eventBus.Publish(ctx, eventType, event); err != nil { + p.logger.Error("failed to publish event", + zap.String("event_type", eventType), + zap.String("instance_id", instance.ID), + zap.Error(err)) + } else { + p.logger.Debug("published agent event", + zap.String("event_type", eventType), + zap.String("instance_id", instance.ID)) + } +} + +// PublishACPSessionCreated publishes an event when an ACP session is created. +func (p *EventPublisher) PublishACPSessionCreated(instance *AgentInstance, sessionID string) { + if p.eventBus == nil || sessionID == "" { + return + } + + data := map[string]interface{}{ + "task_id": instance.TaskID, + "agent_instance_id": instance.ID, + "acp_session_id": sessionID, + } + + event := bus.NewEvent(events.AgentACPSessionCreated, "agent-manager", data) + if err := p.eventBus.Publish(context.Background(), events.AgentACPSessionCreated, event); err != nil { + p.logger.Error("failed to publish ACP session event", + zap.String("event_type", events.AgentACPSessionCreated), + zap.String("instance_id", instance.ID), + zap.Error(err)) + } +} + +// PublishSessionUpdate publishes a session update to the event bus for WebSocket streaming. +func (p *EventPublisher) PublishSessionUpdate(instance *AgentInstance, update agentctl.SessionUpdate) { + if p.eventBus == nil { + return + } + + // 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().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 := p.eventBus.Publish(context.Background(), subject, event); err != nil { + p.logger.Error("failed to publish session update", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID), + zap.Error(err)) + } +} + +// PublishGitStatus publishes a git status update to the event bus. +func (p *EventPublisher) PublishGitStatus(instance *AgentInstance, update *agentctl.GitStatusUpdate) { + if p.eventBus == nil { + return + } + + data := map[string]interface{}{ + "task_id": instance.TaskID, + "agent_id": instance.ID, + "branch": update.Branch, + "remote_branch": update.RemoteBranch, + "modified": update.Modified, + "added": update.Added, + "deleted": update.Deleted, + "untracked": update.Untracked, + "renamed": update.Renamed, + "ahead": update.Ahead, + "behind": update.Behind, + "files": update.Files, + "timestamp": update.Timestamp.Format(time.RFC3339Nano), + } + + event := bus.NewEvent(events.GitStatusUpdated, "agent-manager", data) + subject := events.BuildGitStatusSubject(instance.TaskID) + + if err := p.eventBus.Publish(context.Background(), subject, event); err != nil { + p.logger.Error("failed to publish git status event", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID), + zap.Error(err)) + } +} + +// PublishFileChange publishes a file change notification to the event bus. +func (p *EventPublisher) PublishFileChange(instance *AgentInstance, notification *agentctl.FileChangeNotification) { + if p.eventBus == nil { + return + } + + data := map[string]interface{}{ + "task_id": instance.TaskID, + "agent_id": instance.ID, + "path": notification.Path, + "operation": notification.Operation, + "timestamp": notification.Timestamp.Format(time.RFC3339Nano), + } + + event := bus.NewEvent(events.FileChangeNotified, "agent-manager", data) + subject := events.BuildFileChangeSubject(instance.TaskID) + + if err := p.eventBus.Publish(context.Background(), subject, event); err != nil { + p.logger.Error("failed to publish file change event", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID), + zap.Error(err)) + } +} + +// PublishPromptComplete publishes a prompt_complete event when an agent finishes responding. +// This is used to save the agent's response as a comment on the task. +func (p *EventPublisher) PublishPromptComplete(instance *AgentInstance, agentMessage, reasoning, summary string) { + if p.eventBus == nil { + return + } + + // Only publish if there's actual content + if agentMessage == "" { + return + } + + data := map[string]interface{}{ + "type": "prompt_complete", + "timestamp": time.Now().UTC().Format(time.RFC3339Nano), + "agent_id": instance.ID, + "task_id": instance.TaskID, + "agent_message": agentMessage, + } + if reasoning != "" { + data["reasoning"] = reasoning + } + if summary != "" { + data["summary"] = summary + } + + event := bus.NewEvent(events.PromptComplete, "agent-manager", data) + // Publish on task-specific subject so orchestrator can subscribe + subject := events.PromptComplete + "." + instance.TaskID + + if err := p.eventBus.Publish(context.Background(), subject, event); err != nil { + p.logger.Error("failed to publish prompt_complete event", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID), + zap.Error(err)) + } +} + +// PublishToolCall publishes a tool call start event (to be saved as a comment). +func (p *EventPublisher) PublishToolCall(instance *AgentInstance, toolCallID, title, status string, args map[string]interface{}) { + if p.eventBus == nil { + return + } + + data := map[string]interface{}{ + "type": "tool_call", + "timestamp": time.Now().UTC().Format(time.RFC3339Nano), + "agent_id": instance.ID, + "task_id": instance.TaskID, + "tool_call_id": toolCallID, + "title": title, + "status": status, + } + if args != nil { + data["args"] = args + } + + event := bus.NewEvent(events.ToolCallStarted, "agent-manager", data) + subject := events.ToolCallStarted + "." + instance.TaskID + + if err := p.eventBus.Publish(context.Background(), subject, event); err != nil { + p.logger.Error("failed to publish tool_call event", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID), + zap.Error(err)) + } else { + p.logger.Debug("published tool_call event", + zap.String("task_id", instance.TaskID), + zap.String("tool_call_id", toolCallID), + zap.String("title", title)) + } +} + +// PublishToolCallComplete publishes a tool call completion event from a SessionUpdate. +func (p *EventPublisher) PublishToolCallComplete(instance *AgentInstance, update agentctl.SessionUpdate) { + if p.eventBus == nil { + return + } + + 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": update.ToolCallID, + "status": update.ToolStatus, + } + + event := bus.NewEvent(events.ToolCallComplete, "agent-manager", data) + subject := events.ToolCallComplete + "." + instance.TaskID + + if err := p.eventBus.Publish(context.Background(), subject, event); err != nil { + p.logger.Error("failed to publish tool_call_complete event", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID), + zap.Error(err)) + } +} + +// PublishPermissionRequest publishes a permission request event. +func (p *EventPublisher) PublishPermissionRequest(instance *AgentInstance, notification *agentctl.PermissionNotification) { + if p.eventBus == nil { + return + } + + // Convert options to a serializable format + options := make([]map[string]interface{}, len(notification.Options)) + for i, opt := range notification.Options { + options[i] = map[string]interface{}{ + "option_id": opt.OptionID, + "name": opt.Name, + "kind": opt.Kind, + } + } + + data := map[string]interface{}{ + "type": "permission_request", + "task_id": instance.TaskID, + "agent_instance_id": instance.ID, + "pending_id": notification.PendingID, + "session_id": notification.SessionID, + "tool_call_id": notification.ToolCallID, + "title": notification.Title, + "options": options, + "created_at": notification.CreatedAt, + "timestamp": time.Now().UTC().Format(time.RFC3339), + } + + event := bus.NewEvent(events.ACPMessage, "agent-manager", data) + subject := events.BuildACPSubject(instance.TaskID) + + if err := p.eventBus.Publish(context.Background(), subject, event); err != nil { + p.logger.Error("failed to publish permission_request event", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID), + zap.Error(err)) + } else { + p.logger.Info("published permission_request event", + zap.String("task_id", instance.TaskID), + zap.String("pending_id", notification.PendingID), + zap.String("title", notification.Title)) + } +} diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index d2403efa2b..dbe2e7f37a 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -120,6 +120,13 @@ type Manager struct { agentCfg config.AgentConfig standaloneCtl *agentctl.StandaloneCtl // Client for standalone agentctl control API + // Refactored components for separation of concerns + commandBuilder *CommandBuilder // Builds agent commands from registry config + sessionManager *SessionManager // Handles ACP session initialization + streamManager *StreamManager // Manages WebSocket streams + eventPublisher *EventPublisher // Publishes lifecycle events + containerManager *ContainerManager // Manages Docker containers + // Shell stream starter for auto-starting shell streams shellStreamStarter ShellStreamStarter @@ -143,19 +150,49 @@ func NewManager( agentCfg config.AgentConfig, log *logger.Logger, ) *Manager { - mgr := &Manager{ - docker: dockerClient, - registry: reg, - eventBus: eventBus, - agentCfg: agentCfg, - logger: log.WithFields(zap.String("component", "lifecycle-manager")), - instances: make(map[string]*AgentInstance), - byTask: make(map[string]string), - byContainer: make(map[string]string), - cleanupInterval: 30 * time.Second, - stopCh: make(chan struct{}), + componentLogger := log.WithFields(zap.String("component", "lifecycle-manager")) + + // Initialize command builder + commandBuilder := NewCommandBuilder() + + // Initialize session manager + sessionManager := NewSessionManager(log) + + // Initialize event publisher + eventPublisher := NewEventPublisher(eventBus, log) + + // Initialize container manager (only if docker client is available) + // Note: Network configuration is not used currently - containers use default Docker networking + var containerManager *ContainerManager + if dockerClient != nil { + containerManager = NewContainerManager(dockerClient, "", log) } + mgr := &Manager{ + docker: dockerClient, + registry: reg, + eventBus: eventBus, + agentCfg: agentCfg, + logger: componentLogger, + commandBuilder: commandBuilder, + sessionManager: sessionManager, + eventPublisher: eventPublisher, + containerManager: containerManager, + instances: make(map[string]*AgentInstance), + byTask: make(map[string]string), + byContainer: make(map[string]string), + cleanupInterval: 30 * time.Second, + stopCh: make(chan struct{}), + } + + // Initialize stream manager with callbacks that delegate to manager methods + mgr.streamManager = NewStreamManager(log, StreamCallbacks{ + OnSessionUpdate: mgr.handleSessionUpdate, + OnPermission: mgr.handlePermissionNotification, + OnGitStatus: mgr.handleGitStatusUpdate, + OnFileChange: mgr.handleFileChangeNotification, + }) + // Initialize standalone agentctl client if in standalone mode if agentCfg.Runtime == "standalone" { mgr.standaloneCtl = agentctl.NewStandaloneCtl( @@ -537,7 +574,7 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentInstanc m.publishEvent(ctx, events.AgentStarted, instance) // 10. Wait for agentctl to be ready and start the agent process - go m.initializeAgent(instance, req.TaskDescription) + go m.initializeAgent(instance, agentConfig, req.TaskDescription) m.logger.Info("agent launched successfully", zap.String("instance_id", instanceID), @@ -612,23 +649,17 @@ func (m *Manager) launchStandalone(ctx context.Context, instanceID string, req * // Build environment variables for the agent env := m.buildStandaloneEnv(instanceID, req, agentConfig) - // Build agent command from registry config - // If Cmd is specified, join it; otherwise use default auggie command - agentCommand := "" - if len(agentConfig.Cmd) > 0 { - agentCommand = strings.Join(agentConfig.Cmd, " ") + // Build agent command using CommandBuilder (configuration-driven, no hardcoded agent names) + model := "" + if profileInfo != nil { + model = profileInfo.Model } - // Append model flag if agent supports it and profile has a model configured - if agentConfig.ModelFlag != "" && profileInfo != nil && profileInfo.Model != "" { - agentCommand = agentCommand + " " + agentConfig.ModelFlag + " " + profileInfo.Model + cmdOpts := CommandOptions{ + Model: model, + SessionID: req.ACPSessionID, // CommandBuilder handles resume flag based on SessionConfig } - if req.ACPSessionID != "" && profileInfo != nil && profileInfo.AgentName == "auggie" { - m.logger.Info("adding auggie resume flag to command (standalone)", - zap.String("task_id", req.TaskID), - zap.String("session_id", req.ACPSessionID)) - agentCommand = agentCommand + " --resume " + req.ACPSessionID - } - // If empty, agentctl will use its default (auggie --acp) + agentCommand := m.commandBuilder.BuildCommandString(agentConfig, cmdOpts) + // If empty, agentctl will use its default // Create instance via control API createReq := &agentctl.CreateInstanceRequest{ @@ -754,7 +785,7 @@ func (m *Manager) getOrCreateWorktree(ctx context.Context, req *LaunchRequest) ( } // initializeAgent waits for agentctl to be ready and initializes the agent -func (m *Manager) initializeAgent(instance *AgentInstance, taskDescription string) { +func (m *Manager) initializeAgent(instance *AgentInstance, agentConfig *registry.AgentTypeConfig, taskDescription string) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) defer cancel() @@ -786,8 +817,8 @@ func (m *Manager) initializeAgent(instance *AgentInstance, taskDescription strin // Give the agent process a moment to initialize time.Sleep(500 * time.Millisecond) - // Initialize ACP session - if err := m.initializeACPSession(ctx, instance, taskDescription); err != nil { + // Initialize ACP session using configuration-driven SessionManager + if err := m.initializeACPSession(ctx, instance, agentConfig, taskDescription); err != nil { m.logger.Error("failed to initialize ACP session", zap.String("instance_id", instance.ID), zap.Error(err)) @@ -796,94 +827,43 @@ func (m *Manager) initializeAgent(instance *AgentInstance, taskDescription strin } } -// initializeACPSession sends the ACP initialization messages using the acp-go-sdk -func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInstance, taskDescription string) error { +// initializeACPSession sends the ACP initialization messages using the SessionManager +// The agentConfig is used for configuration-driven session resumption behavior +func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInstance, agentConfig *registry.AgentTypeConfig, taskDescription string) error { m.logger.Info("initializing ACP session", zap.String("instance_id", instance.ID), - zap.String("agentctl_url", instance.agentctl.BaseURL())) + zap.String("agentctl_url", instance.agentctl.BaseURL()), + zap.String("agent_type", agentConfig.ID)) - // 1. Send initialize request - m.logger.Info("sending ACP initialize request", - zap.String("instance_id", instance.ID)) - - agentInfo, err := instance.agentctl.Initialize(ctx, "kandev", "1.0.0") + // Use SessionManager for configuration-driven session initialization + // This replaces the hardcoded auggie-specific logic with registry-based configuration + result, err := m.sessionManager.InitializeSession( + ctx, + instance.agentctl, + agentConfig, + instance.ACPSessionID, // Existing session ID to resume (if any) + instance.WorkspacePath, + ) if err != nil { - m.logger.Error("ACP initialize failed", + m.logger.Error("session initialization failed", zap.String("instance_id", instance.ID), zap.Error(err)) - return fmt.Errorf("initialize failed: %w", err) + return err } - agentName := "unknown" - agentVersion := "unknown" - if agentInfo != nil { - agentName = agentInfo.Name - agentVersion = agentInfo.Version - } - m.logger.Info("ACP initialize response received", + + m.logger.Info("ACP session initialized", zap.String("instance_id", instance.ID), - zap.String("agent_name", agentName), - zap.String("agent_version", agentVersion)) + zap.String("agent_name", result.AgentName), + zap.String("agent_version", result.AgentVersion), + zap.String("session_id", result.SessionID)) - // 2. Create or resume ACP session - sessionID := "" - if instance.ACPSessionID != "" && agentName != "auggie" { - m.logger.Info("sending ACP session/load request", - zap.String("instance_id", instance.ID), - zap.String("session_id", instance.ACPSessionID)) + instance.ACPSessionID = result.SessionID + m.publishACPSessionCreated(instance, result.SessionID) - if err := instance.agentctl.LoadSession(ctx, instance.ACPSessionID); err != nil { - m.logger.Error("ACP session/load failed", - zap.String("instance_id", instance.ID), - zap.Error(err)) - return fmt.Errorf("session/load failed: %w", err) - } - sessionID = instance.ACPSessionID - m.logger.Info("ACP session loaded", - zap.String("instance_id", instance.ID), - zap.String("session_id", sessionID)) - } else { - if instance.ACPSessionID != "" && agentName == "auggie" { - m.logger.Info("skipping ACP session/load for auggie (resume handled via CLI)", - zap.String("instance_id", instance.ID), - zap.String("session_id", instance.ACPSessionID)) - } - m.logger.Info("sending ACP session/new request", - zap.String("instance_id", instance.ID), - zap.String("workspace_path", instance.WorkspacePath)) - - newSessionID, err := instance.agentctl.NewSession(ctx, instance.WorkspacePath) - if err != nil { - m.logger.Error("ACP session/new failed", - zap.String("instance_id", instance.ID), - zap.Error(err)) - return fmt.Errorf("session/new failed: %w", err) - } - sessionID = newSessionID - m.logger.Info("ACP session created", - zap.String("instance_id", instance.ID), - zap.String("session_id", sessionID)) - } - - instance.ACPSessionID = sessionID - m.logger.Info("ACP session ready", - zap.String("instance_id", instance.ID), - zap.String("session_id", sessionID), - zap.String("agent_name", agentName)) - m.publishACPSessionCreated(instance, sessionID) - - // 3. Set up updates stream to receive session notifications - // Use a ready channel to signal when the stream is connected + // Set up WebSocket streams using StreamManager + // Use a ready channel to signal when the updates stream is connected updatesReady := make(chan struct{}) - go m.handleUpdatesStream(instance, updatesReady) - - // 4. Set up permission stream to receive permission requests - go m.handlePermissionStream(instance) - - // 5. Set up git status stream to track workspace changes - go m.handleGitStatusStream(instance) - - // 6. Set up file changes stream to track filesystem changes - go m.handleFileChangesStream(instance) + m.streamManager.ConnectAll(instance, updatesReady) // Note: Shell stream is started on-demand when a client sends shell.subscribe // This ensures the buffered output (including initial prompt) is received @@ -902,7 +882,7 @@ func (m *Manager) initializeACPSession(ctx context.Context, instance *AgentInsta if taskDescription != "" { m.logger.Info("sending ACP prompt", zap.String("instance_id", instance.ID), - zap.String("session_id", sessionID), + zap.String("session_id", instance.ACPSessionID), zap.String("task_description", taskDescription)) // Clear buffers before starting prompt @@ -1354,6 +1334,11 @@ func (m *Manager) handleFileChangesStream(instance *AgentInstance) { zap.Int("max_retries", maxRetries)) } +// handleFileChangeNotification processes file change notifications from the workspace tracker +func (m *Manager) handleFileChangeNotification(instance *AgentInstance, notification *agentctl.FileChangeNotification) { + m.publishFileChange(instance, notification) +} + // publishFileChange publishes a file change notification to the event bus func (m *Manager) publishFileChange(instance *AgentInstance, notification *agentctl.FileChangeNotification) { if m.eventBus == nil { @@ -1621,18 +1606,16 @@ func (m *Manager) buildContainerConfig(instanceID string, req *LaunchRequest, ag containerName := fmt.Sprintf("kandev-agent-%s", instanceID[:8]) - // Build command, appending model flag if agent supports it and profile has a model - cmd := make([]string, len(agentConfig.Cmd)) - copy(cmd, agentConfig.Cmd) - if agentConfig.ModelFlag != "" && profileInfo != nil && profileInfo.Model != "" { - cmd = append(cmd, agentConfig.ModelFlag, profileInfo.Model) + // Build command using CommandBuilder (configuration-driven, no hardcoded agent names) + model := "" + if profileInfo != nil { + model = profileInfo.Model } - if req.ACPSessionID != "" && profileInfo != nil && profileInfo.AgentName == "auggie" { - m.logger.Info("adding auggie resume flag to command", - zap.String("task_id", req.TaskID), - zap.String("session_id", req.ACPSessionID)) - cmd = append(cmd, "--resume", req.ACPSessionID) + cmdOpts := CommandOptions{ + Model: model, + SessionID: req.ACPSessionID, // CommandBuilder handles resume flag based on SessionConfig } + cmd := m.commandBuilder.BuildCommand(agentConfig, cmdOpts) return docker.ContainerConfig{ Name: containerName, diff --git a/apps/backend/internal/agent/lifecycle/profile_resolver.go b/apps/backend/internal/agent/lifecycle/profile_resolver.go index 2968fb91a9..1579d53d7a 100644 --- a/apps/backend/internal/agent/lifecycle/profile_resolver.go +++ b/apps/backend/internal/agent/lifecycle/profile_resolver.go @@ -42,4 +42,3 @@ func (r *StoreProfileResolver) ResolveProfile(ctx context.Context, profileID str Plan: profile.Plan, }, nil } - diff --git a/apps/backend/internal/agent/lifecycle/profile_resolver_test.go b/apps/backend/internal/agent/lifecycle/profile_resolver_test.go index 4cd295f679..a059d9c194 100644 --- a/apps/backend/internal/agent/lifecycle/profile_resolver_test.go +++ b/apps/backend/internal/agent/lifecycle/profile_resolver_test.go @@ -11,10 +11,10 @@ import ( // 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) + 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) } @@ -214,4 +214,3 @@ func TestStoreProfileResolver_ResolveProfile_AgentNotFound(t *testing.T) { t.Errorf("expected error message to start with '%s', got '%s'", expectedMsg, err.Error()) } } - diff --git a/apps/backend/internal/agent/lifecycle/session.go b/apps/backend/internal/agent/lifecycle/session.go new file mode 100644 index 0000000000..ac641cdc4b --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/session.go @@ -0,0 +1,169 @@ +package lifecycle + +import ( + "context" + "fmt" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/agent/agentctl" + "github.com/kandev/kandev/internal/agent/registry" + "github.com/kandev/kandev/internal/common/logger" +) + +// SessionManager handles ACP session initialization and management +type SessionManager struct { + logger *logger.Logger +} + +// NewSessionManager creates a new SessionManager +func NewSessionManager(log *logger.Logger) *SessionManager { + return &SessionManager{ + logger: log, + } +} + +// InitializeResult contains the result of session initialization +type InitializeResult struct { + AgentName string + AgentVersion string + SessionID string +} + +// InitializeSession initializes an ACP session with the agent. +// It handles the initialize handshake and session creation/loading based on config. +// +// Session behavior: +// - If agentConfig.SessionConfig.ResumeViaACP is true AND existingSessionID is provided: use session/load +// - If agentConfig.SessionConfig.ResumeViaACP is false (CLI handles resume): always use session/new +// - Otherwise: use session/new +func (sm *SessionManager) InitializeSession( + ctx context.Context, + client *agentctl.Client, + agentConfig *registry.AgentTypeConfig, + existingSessionID string, + workspacePath string, +) (*InitializeResult, error) { + sm.logger.Info("initializing ACP session", + zap.String("agent_type", agentConfig.ID), + zap.String("workspace_path", workspacePath), + zap.Bool("resume_via_acp", agentConfig.SessionConfig.ResumeViaACP), + zap.String("existing_session_id", existingSessionID)) + + // Step 1: Send initialize request + sm.logger.Info("sending ACP initialize request", + zap.String("agent_type", agentConfig.ID)) + + agentInfo, err := client.Initialize(ctx, "kandev", "1.0.0") + if err != nil { + sm.logger.Error("ACP initialize failed", + zap.String("agent_type", agentConfig.ID), + zap.Error(err)) + return nil, fmt.Errorf("initialize failed: %w", err) + } + + result := &InitializeResult{ + AgentName: "unknown", + AgentVersion: "unknown", + } + if agentInfo != nil { + result.AgentName = agentInfo.Name + result.AgentVersion = agentInfo.Version + } + + sm.logger.Info("ACP initialize response received", + zap.String("agent_type", agentConfig.ID), + zap.String("agent_name", result.AgentName), + zap.String("agent_version", result.AgentVersion)) + + // Step 2: Create or resume ACP session based on configuration + sessionID, err := sm.createOrLoadSession(ctx, client, agentConfig, existingSessionID, workspacePath) + if err != nil { + return nil, err + } + + result.SessionID = sessionID + return result, nil +} + +// createOrLoadSession determines whether to create a new session or load an existing one +// based on the agent configuration and whether an existing session ID is provided. +func (sm *SessionManager) createOrLoadSession( + ctx context.Context, + client *agentctl.Client, + agentConfig *registry.AgentTypeConfig, + existingSessionID string, + workspacePath string, +) (string, error) { + // Use session/load only when: + // 1. ResumeViaACP is true (agent supports ACP-based resume) + // 2. An existing session ID is provided + shouldLoadSession := agentConfig.SessionConfig.ResumeViaACP && existingSessionID != "" + + if shouldLoadSession { + return sm.loadSession(ctx, client, agentConfig, existingSessionID) + } + + // Log why we're creating a new session when an existing session ID was provided + if existingSessionID != "" && !agentConfig.SessionConfig.ResumeViaACP { + sm.logger.Info("skipping ACP session/load (resume handled via CLI)", + zap.String("agent_type", agentConfig.ID), + zap.String("session_id", existingSessionID), + zap.String("resume_flag", agentConfig.SessionConfig.ResumeFlag)) + } + + return sm.createNewSession(ctx, client, agentConfig, workspacePath) +} + +// loadSession loads an existing session via ACP session/load +func (sm *SessionManager) loadSession( + ctx context.Context, + client *agentctl.Client, + agentConfig *registry.AgentTypeConfig, + sessionID string, +) (string, error) { + sm.logger.Info("sending ACP session/load request", + zap.String("agent_type", agentConfig.ID), + zap.String("session_id", sessionID)) + + if err := client.LoadSession(ctx, sessionID); err != nil { + sm.logger.Error("ACP session/load failed", + zap.String("agent_type", agentConfig.ID), + zap.String("session_id", sessionID), + zap.Error(err)) + return "", fmt.Errorf("session/load failed: %w", err) + } + + sm.logger.Info("ACP session loaded successfully", + zap.String("agent_type", agentConfig.ID), + zap.String("session_id", sessionID)) + + return sessionID, nil +} + +// createNewSession creates a new session via ACP session/new +func (sm *SessionManager) createNewSession( + ctx context.Context, + client *agentctl.Client, + agentConfig *registry.AgentTypeConfig, + workspacePath string, +) (string, error) { + sm.logger.Info("sending ACP session/new request", + zap.String("agent_type", agentConfig.ID), + zap.String("workspace_path", workspacePath)) + + sessionID, err := client.NewSession(ctx, workspacePath) + if err != nil { + sm.logger.Error("ACP session/new failed", + zap.String("agent_type", agentConfig.ID), + zap.String("workspace_path", workspacePath), + zap.Error(err)) + return "", fmt.Errorf("session/new failed: %w", err) + } + + sm.logger.Info("ACP session created successfully", + zap.String("agent_type", agentConfig.ID), + zap.String("session_id", sessionID)) + + return sessionID, nil +} diff --git a/apps/backend/internal/agent/lifecycle/streams.go b/apps/backend/internal/agent/lifecycle/streams.go new file mode 100644 index 0000000000..2a37b4b4b2 --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/streams.go @@ -0,0 +1,185 @@ +package lifecycle + +import ( + "context" + "time" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/agent/agentctl" + "github.com/kandev/kandev/internal/common/logger" +) + +// StreamCallbacks defines callbacks for stream events +type StreamCallbacks struct { + OnSessionUpdate func(instance *AgentInstance, update agentctl.SessionUpdate) + OnPermission func(instance *AgentInstance, notification *agentctl.PermissionNotification) + OnGitStatus func(instance *AgentInstance, update *agentctl.GitStatusUpdate) + OnFileChange func(instance *AgentInstance, notification *agentctl.FileChangeNotification) +} + +// StreamManager manages WebSocket streams to agent instances +type StreamManager struct { + logger *logger.Logger + callbacks StreamCallbacks +} + +// NewStreamManager creates a new StreamManager +func NewStreamManager(log *logger.Logger, callbacks StreamCallbacks) *StreamManager { + return &StreamManager{ + logger: log.WithFields(zap.String("component", "stream-manager")), + callbacks: callbacks, + } +} + +// ConnectAll connects to all streams for an instance. +// If ready is non-nil, it will be closed when the updates stream is connected. +func (sm *StreamManager) ConnectAll(instance *AgentInstance, ready chan<- struct{}) { + go sm.connectUpdatesStream(instance, ready) + go sm.connectPermissionStream(instance) + go sm.connectGitStatusStream(instance) + go sm.connectFileChangesStream(instance) +} + +// ReconnectAll reconnects to all streams (used after backend restart). +// This waits for agentctl to be ready before connecting to streams. +func (sm *StreamManager) ReconnectAll(instance *AgentInstance) { + sm.logger.Info("reconnecting to agent streams after recovery", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID)) + + // Wait a moment for any startup operations to settle + time.Sleep(500 * time.Millisecond) + + // Check if agentctl is responsive + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + if err := instance.agentctl.WaitForReady(ctx, 10*time.Second); err != nil { + sm.logger.Warn("agentctl not ready for stream reconnection", + zap.String("instance_id", instance.ID), + zap.Error(err)) + // Don't return - still try to connect to streams + } + + // Reconnect to WebSocket streams + sm.ConnectAll(instance, nil) + + sm.logger.Info("agent streams reconnected", + zap.String("instance_id", instance.ID), + zap.String("task_id", instance.TaskID)) +} + +// connectUpdatesStream handles the updates WebSocket stream with ready signaling +func (sm *StreamManager) connectUpdatesStream(instance *AgentInstance, ready chan<- struct{}) { + ctx := context.Background() + + err := instance.agentctl.StreamUpdates(ctx, func(update agentctl.SessionUpdate) { + if sm.callbacks.OnSessionUpdate != nil { + sm.callbacks.OnSessionUpdate(instance, update) + } + }) + + // Signal that the stream connection attempt is complete (success or failure) + // StreamUpdates returns immediately after establishing the WebSocket connection + // and starting the read goroutine, so this signals that we're ready to receive updates + if ready != nil { + close(ready) + } + + if err != nil { + sm.logger.Error("failed to connect to updates stream", + zap.String("instance_id", instance.ID), + zap.Error(err)) + } +} + +// connectPermissionStream handles the permission WebSocket stream +func (sm *StreamManager) connectPermissionStream(instance *AgentInstance) { + ctx := context.Background() + + err := instance.agentctl.StreamPermissions(ctx, func(notification *agentctl.PermissionNotification) { + if sm.callbacks.OnPermission != nil { + sm.callbacks.OnPermission(instance, notification) + } + }) + if err != nil { + sm.logger.Error("failed to connect to permission stream", + zap.String("instance_id", instance.ID), + zap.Error(err)) + } +} + +// connectGitStatusStream handles git status stream with retry logic +func (sm *StreamManager) connectGitStatusStream(instance *AgentInstance) { + ctx := context.Background() + + // Retry connection with exponential backoff + maxRetries := 5 + backoff := 1 * time.Second + + for attempt := 1; attempt <= maxRetries; attempt++ { + err := instance.agentctl.StreamGitStatus(ctx, func(update *agentctl.GitStatusUpdate) { + if sm.callbacks.OnGitStatus != nil { + sm.callbacks.OnGitStatus(instance, update) + } + }) + + if err == nil { + // Connection closed normally + return + } + + sm.logger.Debug("git status stream connection failed, retrying", + zap.String("instance_id", instance.ID), + zap.Int("attempt", attempt), + zap.Int("max_retries", maxRetries), + zap.Error(err)) + + if attempt < maxRetries { + time.Sleep(backoff) + backoff *= 2 // Exponential backoff + } + } + + sm.logger.Error("failed to connect to git status stream after retries", + zap.String("instance_id", instance.ID), + zap.Int("max_retries", maxRetries)) +} + +// connectFileChangesStream handles file changes stream with retry logic +func (sm *StreamManager) connectFileChangesStream(instance *AgentInstance) { + ctx := context.Background() + + // Retry connection with exponential backoff + maxRetries := 5 + backoff := 1 * time.Second + + for attempt := 1; attempt <= maxRetries; attempt++ { + err := instance.agentctl.StreamFileChanges(ctx, func(notification *agentctl.FileChangeNotification) { + if sm.callbacks.OnFileChange != nil { + sm.callbacks.OnFileChange(instance, notification) + } + }) + + if err == nil { + // Connection closed normally + return + } + + sm.logger.Debug("file changes stream connection failed, retrying", + zap.String("instance_id", instance.ID), + zap.Int("attempt", attempt), + zap.Int("max_retries", maxRetries), + zap.Error(err)) + + if attempt < maxRetries { + time.Sleep(backoff) + backoff *= 2 // Exponential backoff + } + } + + sm.logger.Error("failed to connect to file changes stream after retries", + zap.String("instance_id", instance.ID), + zap.Int("max_retries", maxRetries)) +} diff --git a/apps/backend/internal/agent/registry/defaults.go b/apps/backend/internal/agent/registry/defaults.go index b21276ad12..42df572305 100644 --- a/apps/backend/internal/agent/registry/defaults.go +++ b/apps/backend/internal/agent/registry/defaults.go @@ -19,7 +19,6 @@ func DefaultAgents() []*AgentTypeConfig { }, Mounts: []MountTemplate{ {Source: "{workspace}", Target: "/workspace", ReadOnly: false}, - {Source: "{augment_sessions}", Target: "/root/.augment/sessions", ReadOnly: false}, }, ResourceLimits: ResourceLimits{ MemoryMB: 4096, @@ -31,6 +30,12 @@ func DefaultAgents() []*AgentTypeConfig { ModelFlag: "--model", WorkspaceFlag: "--workspace-root", Protocol: agent.ProtocolACP, + SessionConfig: SessionConfig{ + ResumeViaACP: false, // Auggie handles resume via CLI flag + ResumeFlag: "--resume", + SessionDirTemplate: "{home}/.augment/sessions", + SessionDirTarget: "/root/.augment/sessions", + }, }, { ID: "claude-agent", @@ -54,6 +59,9 @@ func DefaultAgents() []*AgentTypeConfig { Enabled: true, ModelFlag: "--model", Protocol: agent.ProtocolACP, + SessionConfig: SessionConfig{ + ResumeViaACP: true, // Claude uses ACP session/load for resume + }, }, { ID: "codex-agent", @@ -77,6 +85,9 @@ func DefaultAgents() []*AgentTypeConfig { Enabled: true, ModelFlag: "--model", Protocol: agent.ProtocolCodex, + SessionConfig: SessionConfig{ + ResumeViaACP: true, // Codex uses ACP session/load for resume + }, }, { ID: "gemini-agent", @@ -100,6 +111,9 @@ func DefaultAgents() []*AgentTypeConfig { Enabled: true, ModelFlag: "--model", Protocol: agent.ProtocolACP, + SessionConfig: SessionConfig{ + ResumeViaACP: true, // Gemini uses ACP session/load for resume + }, }, { ID: "opencode-agent", @@ -123,6 +137,9 @@ func DefaultAgents() []*AgentTypeConfig { Enabled: true, ModelFlag: "--model", Protocol: agent.ProtocolACP, + SessionConfig: SessionConfig{ + ResumeViaACP: true, // OpenCode uses ACP session/load for resume + }, }, { ID: "copilot-agent", @@ -146,6 +163,9 @@ func DefaultAgents() []*AgentTypeConfig { Enabled: true, ModelFlag: "--model", Protocol: agent.ProtocolACP, + SessionConfig: SessionConfig{ + ResumeViaACP: true, // Copilot uses ACP session/load for resume + }, }, } } diff --git a/apps/backend/internal/agent/registry/registry.go b/apps/backend/internal/agent/registry/registry.go index 6cc5e5cea5..bdf28a5007 100644 --- a/apps/backend/internal/agent/registry/registry.go +++ b/apps/backend/internal/agent/registry/registry.go @@ -37,6 +37,29 @@ type AgentTypeConfig struct { // 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) + + // Session resumption configuration + SessionConfig SessionConfig `json:"session_config,omitempty"` // How to handle session resumption +} + +// SessionConfig defines how session resumption is handled for an agent type +type SessionConfig struct { + // ResumeViaACP indicates whether session resumption should use the ACP session/load method. + // If false, session resumption is handled via CLI flags (ResumeFlag). + ResumeViaACP bool `json:"resume_via_acp"` + + // ResumeFlag is the CLI flag used for session resumption when ResumeViaACP is false. + // Example: "--resume" for auggie, empty for agents that use ACP session/load. + ResumeFlag string `json:"resume_flag,omitempty"` + + // SessionDirTemplate is a template for the host directory where session data is stored. + // Supports variables: {home} (user home directory). + // Example: "{home}/.augment/sessions" for auggie. + SessionDirTemplate string `json:"session_dir_template,omitempty"` + + // SessionDirTarget is the container path where the session directory is mounted. + // Example: "/root/.augment/sessions" for auggie. + SessionDirTarget string `json:"session_dir_target,omitempty"` } // MountTemplate defines a mount with template support