From 187d002717e50de2e799285274e875c0858a426d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Sun, 25 Jan 2026 02:05:19 +0000 Subject: [PATCH] feat: per-task executor selection with multi-runtime support (#92) --- apps/backend/cmd/kandev/adapters.go | 1 + apps/backend/cmd/kandev/agentctl.go | 11 +- apps/backend/cmd/kandev/agents.go | 80 +-- apps/backend/cmd/kandev/main.go | 15 +- .../{augment-agent => }/Dockerfile | 19 +- .../dockerfiles/augment-agent/entrypoint.sh | 41 -- apps/backend/dockerfiles/entrypoint.sh | 21 + .../agent/handlers/shell_handlers_test.go | 2 +- .../internal/agent/lifecycle/container.go | 11 + .../internal/agent/lifecycle/manager.go | 261 +++++--- .../internal/agent/lifecycle/manager_test.go | 2 +- .../internal/agent/lifecycle/runtime.go | 17 +- .../agent/lifecycle/runtime_docker.go | 26 +- .../agent/lifecycle/runtime_registry.go | 130 ++++ .../agent/lifecycle/runtime_registry_test.go | 347 +++++++++++ .../agent/lifecycle/runtime_standalone.go | 11 +- .../backend/internal/agent/lifecycle/types.go | 4 + .../internal/agent/registry/agents.json | 13 +- .../backend/internal/agent/runtime/runtime.go | 20 + apps/backend/internal/common/config/config.go | 18 +- .../orchestrator/controller/controller.go | 2 +- apps/backend/internal/orchestrator/dto/dto.go | 1 + .../orchestrator/executor/executor.go | 120 +++- .../orchestrator/handlers/handlers.go | 2 + .../internal/orchestrator/task_operations.go | 5 +- .../task/handlers/process_handlers_test.go | 2 +- .../internal/task/handlers/task_handlers.go | 14 +- apps/backend/internal/task/service/service.go | 27 +- apps/web/components/task-create-dialog.tsx | 2 + .../web/components/task/sessions-dropdown.tsx | 1 + apps/web/lib/api/domains/kanban-api.ts | 1 + .../task-specific-executor-refactoring.md | 560 ++++++++++++++++++ 32 files changed, 1549 insertions(+), 238 deletions(-) rename apps/backend/dockerfiles/{augment-agent => }/Dockerfile (73%) delete mode 100644 apps/backend/dockerfiles/augment-agent/entrypoint.sh create mode 100644 apps/backend/dockerfiles/entrypoint.sh create mode 100644 apps/backend/internal/agent/lifecycle/runtime_registry.go create mode 100644 apps/backend/internal/agent/lifecycle/runtime_registry_test.go create mode 100644 docs/architecture/task-specific-executor-refactoring.md diff --git a/apps/backend/cmd/kandev/adapters.go b/apps/backend/cmd/kandev/adapters.go index cffa001566..0f33252241 100644 --- a/apps/backend/cmd/kandev/adapters.go +++ b/apps/backend/cmd/kandev/adapters.go @@ -72,6 +72,7 @@ func (a *lifecycleAdapter) LaunchAgent(ctx context.Context, req *executor.Launch ACPSessionID: req.ACPSessionID, Metadata: req.Metadata, ModelOverride: req.ModelOverride, + ExecutorType: req.ExecutorType, // Worktree configuration for concurrent agent execution UseWorktree: req.UseWorktree, RepositoryID: req.RepositoryID, diff --git a/apps/backend/cmd/kandev/agentctl.go b/apps/backend/cmd/kandev/agentctl.go index a24fd06a04..13fe1b7e3c 100644 --- a/apps/backend/cmd/kandev/agentctl.go +++ b/apps/backend/cmd/kandev/agentctl.go @@ -11,10 +11,10 @@ import ( "go.uber.org/zap" ) +// provideAgentctlLauncher starts the agentctl launcher for standalone runtime. +// agentctl is a core service that always runs - it's used by the Standalone runtime +// for agent execution on the host machine. func provideAgentctlLauncher(ctx context.Context, cfg *config.Config, log *logger.Logger) (func() error, error) { - if cfg.Agent.Runtime != "standalone" { - return nil, nil - } _, cleanup, err := launcher.Provide(ctx, launcher.Config{ Host: cfg.Agent.StandaloneHost, Port: cfg.Agent.StandalonePort, @@ -25,10 +25,9 @@ func provideAgentctlLauncher(ctx context.Context, cfg *config.Config, log *logge return cleanup, nil } +// waitForAgentctlControlHealthy waits for the agentctl control server to be healthy. +// This is called during startup to ensure agentctl is ready before accepting requests. func waitForAgentctlControlHealthy(ctx context.Context, cfg *config.Config, log *logger.Logger) { - if cfg.Agent.Runtime != "standalone" { - return - } client := agentctlclient.NewControlClient(cfg.Agent.StandaloneHost, cfg.Agent.StandalonePort, log) healthCtx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() diff --git a/apps/backend/cmd/kandev/agents.go b/apps/backend/cmd/kandev/agents.go index 76614a4d72..7e0676f068 100644 --- a/apps/backend/cmd/kandev/agents.go +++ b/apps/backend/cmd/kandev/agents.go @@ -26,14 +26,38 @@ func provideLifecycleManager( dockerClient *docker.Client, agentSettingsRepo settingsstore.Repository, ) (*lifecycle.Manager, *registry.Registry, error) { - agentManagerEnabled := (cfg.Agent.Runtime == "docker" && dockerClient != nil) || - cfg.Agent.Runtime == "standalone" - if !agentManagerEnabled { - log.Info("Agent Manager disabled (no Docker and not in standalone mode)") - return nil, nil, nil - } + log.Info("Initializing Agent Manager...") - log.Info("Initializing Agent Manager...", zap.String("runtime", cfg.Agent.Runtime)) + // Create runtime registry to manage multiple runtimes + runtimeRegistry := lifecycle.NewRuntimeRegistry(log) + + // Standalone runtime is always available (agentctl is a core service) + controlClient := agentctl.NewControlClient( + cfg.Agent.StandaloneHost, + cfg.Agent.StandalonePort, + log, + ) + standaloneRuntime := lifecycle.NewStandaloneRuntime( + controlClient, + cfg.Agent.StandaloneHost, + cfg.Agent.StandalonePort, + log, + ) + runtimeRegistry.Register(standaloneRuntime) + log.Info("Standalone runtime registered", + zap.String("host", cfg.Agent.StandaloneHost), + zap.Int("port", cfg.Agent.StandalonePort)) + + // Register Docker runtime if enabled and Docker client is available + var containerMgr *lifecycle.ContainerManager + if cfg.Docker.Enabled && dockerClient != nil { + containerMgr = lifecycle.NewContainerManager(dockerClient, "", log) + dockerRuntime := lifecycle.NewDockerRuntime(dockerClient, log) + runtimeRegistry.Register(dockerRuntime) + log.Info("Docker runtime registered") + } else if cfg.Docker.Enabled && dockerClient == nil { + log.Warn("Docker runtime enabled but Docker client not available") + } agentRegistry, _, err := registry.Provide(log) if err != nil { @@ -48,41 +72,25 @@ func provideLifecycleManager( } profileResolver := lifecycle.NewStoreProfileResolver(agentSettingsRepo, agentRegistry) - - var agentRuntime lifecycle.Runtime - var containerMgr *lifecycle.ContainerManager - switch cfg.Agent.Runtime { - case "standalone": - controlClient := agentctl.NewControlClient( - cfg.Agent.StandaloneHost, - cfg.Agent.StandalonePort, - log, - ) - agentRuntime = lifecycle.NewStandaloneRuntime( - controlClient, - cfg.Agent.StandaloneHost, - cfg.Agent.StandalonePort, - log, - ) - log.Info("Using standalone runtime", - zap.String("host", cfg.Agent.StandaloneHost), - zap.Int("port", cfg.Agent.StandalonePort)) - default: - if dockerClient != nil { - containerMgr = lifecycle.NewContainerManager(dockerClient, "", log) - agentRuntime = lifecycle.NewDockerRuntime(dockerClient, log) - log.Info("Using Docker runtime") - } - } - mcpService := mcpconfig.NewService(agentSettingsRepo) - lifecycleMgr := lifecycle.NewManager(agentRegistry, eventBus, agentRuntime, containerMgr, credsMgr, profileResolver, mcpService, log) + + lifecycleMgr := lifecycle.NewManager( + agentRegistry, + eventBus, + runtimeRegistry, + containerMgr, + credsMgr, + profileResolver, + mcpService, + lifecycle.RuntimeFallbackWarn, + log, + ) if err := lifecycleMgr.Start(ctx); err != nil { return nil, nil, err } log.Info("Agent Manager initialized", - zap.String("runtime", cfg.Agent.Runtime), + zap.Int("runtimes", len(runtimeRegistry.List())), zap.Int("agent_types", len(agentRegistry.List()))) return lifecycleMgr, agentRegistry, nil } diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index 911445a6ef..18b7104d56 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -52,11 +52,10 @@ import ( // Command-line flags var ( - flagAgentRuntime = flag.String("agent-runtime", "", "Agent runtime mode: 'docker' or 'standalone'") - flagPort = flag.Int("port", 0, "HTTP server port (default: 8080)") - flagLogLevel = flag.String("log-level", "", "Log level: debug, info, warn, error") - flagHelp = flag.Bool("help", false, "Show help message") - flagVersion = flag.Bool("version", false, "Show version information") + flagPort = flag.Int("port", 0, "HTTP server port (default: 8080)") + flagLogLevel = flag.String("log-level", "", "Log level: debug, info, warn, error") + flagHelp = flag.Bool("help", false, "Show help message") + flagVersion = flag.Bool("version", false, "Show version information") ) func init() { @@ -66,8 +65,7 @@ func init() { fmt.Fprintf(os.Stderr, "Options:\n") flag.PrintDefaults() fmt.Fprintf(os.Stderr, "\nExamples:\n") - fmt.Fprintf(os.Stderr, " kandev # Start with Docker runtime (default)\n") - fmt.Fprintf(os.Stderr, " kandev -agent-runtime=standalone # Start with standalone agentctl\n") + fmt.Fprintf(os.Stderr, " kandev # Start with default settings\n") fmt.Fprintf(os.Stderr, " kandev -port=9000 -log-level=debug # Custom port and log level\n") } } @@ -93,9 +91,6 @@ func main() { } // Apply command-line flag overrides (flags take precedence over config/env) - if *flagAgentRuntime != "" { - cfg.Agent.Runtime = *flagAgentRuntime - } if *flagPort > 0 { cfg.Server.Port = *flagPort } diff --git a/apps/backend/dockerfiles/augment-agent/Dockerfile b/apps/backend/dockerfiles/Dockerfile similarity index 73% rename from apps/backend/dockerfiles/augment-agent/Dockerfile rename to apps/backend/dockerfiles/Dockerfile index 285a6b3e74..a3d09cd5a3 100644 --- a/apps/backend/dockerfiles/augment-agent/Dockerfile +++ b/apps/backend/dockerfiles/Dockerfile @@ -1,5 +1,5 @@ -# Augment Agent Dockerfile -# A coding agent powered by Auggie CLI (Augment Code) with agentctl sidecar +# Kandev Multi-Agent Dockerfile +# A multi-agent image supporting Auggie CLI (Augment Code) and Codex CLI (OpenAI) with agentctl sidecar # # Prerequisites: Build agentctl before building this image: # make build-agentctl @@ -19,6 +19,9 @@ RUN apt-get update && apt-get install -y \ # Install Auggie CLI from npm RUN npm install -g @augmentcode/auggie +# Install Codex CLI from npm +RUN npm install -g @openai/codex + # Copy pre-built agentctl binary (run `make build-agentctl` first) COPY bin/agentctl /usr/local/bin/agentctl RUN chmod +x /usr/local/bin/agentctl @@ -27,7 +30,7 @@ RUN chmod +x /usr/local/bin/agentctl WORKDIR /agent # Copy entrypoint script -COPY dockerfiles/augment-agent/entrypoint.sh /agent/entrypoint.sh +COPY dockerfiles/entrypoint.sh /agent/entrypoint.sh RUN chmod +x /agent/entrypoint.sh # Workspace will be mounted here @@ -38,18 +41,22 @@ WORKDIR /workspace # TASK_ID - The task identifier # TASK_TITLE - Task title/name # TASK_DESCRIPTION - Full task description +# +# For Auggie: # AUGMENT_SESSION_AUTH - Session auth JSON for Augment (from ~/.augment/session.json) # AUGGIE_SESSION_ID - (Optional) Session ID to resume a previous conversation # +# For Codex: +# OPENAI_API_KEY - OpenAI API key for Codex +# # agentctl configuration: # AGENTCTL_PORT - HTTP API port (default: 9999) -# AGENTCTL_AGENT_COMMAND - Command to run (default: "auggie --acp") +# AGENTCTL_AGENT_COMMAND - Command to run (e.g., "auggie --acp" or "codex app-server") # AGENTCTL_AUTO_START - Auto-start agent on startup (default: false) -# AGENTCTL_MODE - "agentctl" for HTTP mode, "direct" for legacy stdin/stdout mode # Expose agentctl HTTP API port EXPOSE 9999 -# Default entrypoint that decides between agentctl and direct mode +# Run agentctl HTTP server ENTRYPOINT ["/agent/entrypoint.sh"] diff --git a/apps/backend/dockerfiles/augment-agent/entrypoint.sh b/apps/backend/dockerfiles/augment-agent/entrypoint.sh deleted file mode 100644 index 95ee9b1b9f..0000000000 --- a/apps/backend/dockerfiles/augment-agent/entrypoint.sh +++ /dev/null @@ -1,41 +0,0 @@ -#!/bin/bash -# Entrypoint script for Augment Agent -# Supports two modes: -# - "agentctl": Runs agentctl HTTP server that manages auggie subprocess -# - "direct": Legacy mode that runs auggie directly (stdin/stdout) - -set -e - -MODE="${AGENTCTL_MODE:-agentctl}" - -echo "Starting Augment Agent in ${MODE} mode..." - -# Common validation -if ! command -v auggie &> /dev/null; then - echo "ERROR: auggie CLI not found. Is @augmentcode/auggie installed?" - exit 1 -fi - -if [ -z "$AUGMENT_SESSION_AUTH" ]; then - echo "ERROR: AUGMENT_SESSION_AUTH environment variable not set" - exit 1 -fi - -case "$MODE" in - "agentctl") - # Run agentctl HTTP server - echo "Starting agentctl on port ${AGENTCTL_PORT:-9999}..." - exec /usr/local/bin/agentctl - ;; - "direct") - # Legacy mode: run auggie directly - echo "Running auggie directly in ACP mode..." - exec auggie --acp --workspace-root /workspace - ;; - *) - echo "ERROR: Unknown AGENTCTL_MODE: $MODE" - echo "Valid modes: agentctl, direct" - exit 1 - ;; -esac - diff --git a/apps/backend/dockerfiles/entrypoint.sh b/apps/backend/dockerfiles/entrypoint.sh new file mode 100644 index 0000000000..fd79f6cb1e --- /dev/null +++ b/apps/backend/dockerfiles/entrypoint.sh @@ -0,0 +1,21 @@ +#!/bin/bash +# Entrypoint script for Kandev Multi-Agent +# Runs agentctl HTTP server that manages agent subprocesses (auggie, codex, etc.) + +set -e + +echo "Starting Kandev Multi-Agent..." + +# Validate agent CLIs are available +if ! command -v auggie &> /dev/null; then + echo "WARNING: auggie CLI not found" +fi + +if ! command -v codex &> /dev/null; then + echo "WARNING: codex CLI not found" +fi + +# Run agentctl HTTP server +echo "Starting agentctl on port ${AGENTCTL_PORT:-9999}..." +exec /usr/local/bin/agentctl + diff --git a/apps/backend/internal/agent/handlers/shell_handlers_test.go b/apps/backend/internal/agent/handlers/shell_handlers_test.go index 2c5013afca..a468698296 100644 --- a/apps/backend/internal/agent/handlers/shell_handlers_test.go +++ b/apps/backend/internal/agent/handlers/shell_handlers_test.go @@ -183,7 +183,7 @@ func newTestManager() *lifecycle.Manager { credsMgr := &MockCredentialsManager{} profileResolver := &MockProfileResolver{} // Pass nil for runtime and containerManager - tests don't need them - return lifecycle.NewManager(reg, eventBus, nil, nil, credsMgr, profileResolver, nil, log) + return lifecycle.NewManager(reg, eventBus, nil, nil, credsMgr, profileResolver, nil, lifecycle.RuntimeFallbackWarn, log) } func TestWsShellStatus_NoInstanceFound(t *testing.T) { diff --git a/apps/backend/internal/agent/lifecycle/container.go b/apps/backend/internal/agent/lifecycle/container.go index 037df12f7b..ca5614228a 100644 --- a/apps/backend/internal/agent/lifecycle/container.go +++ b/apps/backend/internal/agent/lifecycle/container.go @@ -5,6 +5,7 @@ package lifecycle import ( "context" "fmt" + "os" "strings" "time" @@ -277,6 +278,16 @@ func (cm *ContainerManager) expandMounts(templates []registry.MountTemplate, wor func (cm *ContainerManager) expandMountSource(source, workspacePath string) string { result := source result = strings.ReplaceAll(result, "{workspace}", workspacePath) + + // Expand {home} to user's home directory + if strings.Contains(result, "{home}") { + homeDir, err := os.UserHomeDir() + if err != nil { + homeDir = "/tmp" + } + result = strings.ReplaceAll(result, "{home}", homeDir) + } + return result } diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index fcf4c87f07..93ea7ecaa1 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -17,7 +17,6 @@ import ( "github.com/kandev/kandev/internal/agent/mcpconfig" "github.com/kandev/kandev/internal/agent/registry" "github.com/kandev/kandev/internal/agent/runtime" - "github.com/kandev/kandev/internal/worktree" agentctl "github.com/kandev/kandev/internal/agentctl/client" agentctltypes "github.com/kandev/kandev/internal/agentctl/types" "github.com/kandev/kandev/internal/agentctl/types/streams" @@ -25,9 +24,23 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/task/models" + "github.com/kandev/kandev/internal/worktree" v1 "github.com/kandev/kandev/pkg/api/v1" ) +// RuntimeFallbackPolicy controls behavior when a requested runtime is unavailable. +type RuntimeFallbackPolicy string + +const ( + // RuntimeFallbackAllow silently falls back to the default runtime (current behavior). + RuntimeFallbackAllow RuntimeFallbackPolicy = "allow" + // RuntimeFallbackWarn falls back but logs a warning (current behavior, explicit). + RuntimeFallbackWarn RuntimeFallbackPolicy = "warn" + // RuntimeFallbackDeny returns an error if the requested runtime is unavailable. + RuntimeFallbackDeny RuntimeFallbackPolicy = "deny" +) + // Manager manages agent instance lifecycles type Manager struct { registry *registry.Registry @@ -38,17 +51,21 @@ type Manager struct { mcpProvider McpConfigProvider logger *logger.Logger - // Agent runtime abstraction (Docker, Standalone, K8s, SSH, etc.) - runtime Runtime + // RuntimeRegistry manages multiple runtimes (Docker, Standalone, etc.) + // Each task can select its runtime based on executor type. + runtimeRegistry *RuntimeRegistry + + // runtimeFallbackPolicy controls behavior when a requested runtime is unavailable. + runtimeFallbackPolicy RuntimeFallbackPolicy // Refactored components for separation of concerns - executionStore *ExecutionStore // Thread-safe execution tracking - 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 (optional, nil for non-Docker runtimes) - historyManager *SessionHistoryManager // Stores session history for context injection (fork_session pattern) + executionStore *ExecutionStore // Thread-safe execution tracking + 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 (optional, nil for non-Docker runtimes) + historyManager *SessionHistoryManager // Stores session history for context injection (fork_session pattern) // Workspace info provider for on-demand instance creation workspaceInfoProvider WorkspaceInfoProvider @@ -60,16 +77,18 @@ type Manager struct { } // NewManager creates a new lifecycle manager. -// The runtime parameter is the agent execution runtime (Docker, Standalone, etc.). +// The runtimeRegistry manages multiple runtimes (Docker, Standalone, etc.) for task-specific execution. // The containerManager parameter is optional and used for Docker cleanup (pass nil for non-Docker runtimes). +// The fallbackPolicy controls behavior when a requested runtime is unavailable. func NewManager( reg *registry.Registry, eventBus bus.EventBus, - runtime Runtime, + runtimeRegistry *RuntimeRegistry, containerManager *ContainerManager, credsMgr CredentialsManager, profileResolver ProfileResolver, mcpProvider McpConfigProvider, + fallbackPolicy RuntimeFallbackPolicy, log *logger.Logger, ) *Manager { componentLogger := log.WithFields(zap.String("component", "lifecycle-manager")) @@ -96,21 +115,22 @@ func NewManager( } mgr := &Manager{ - registry: reg, - eventBus: eventBus, - runtime: runtime, - credsMgr: credsMgr, - profileResolver: profileResolver, - mcpProvider: mcpProvider, - logger: componentLogger, - executionStore: executionStore, - commandBuilder: commandBuilder, - sessionManager: sessionManager, - eventPublisher: eventPublisher, - containerManager: containerManager, - historyManager: historyManager, - cleanupInterval: 30 * time.Second, - stopCh: stopCh, + registry: reg, + eventBus: eventBus, + runtimeRegistry: runtimeRegistry, + runtimeFallbackPolicy: fallbackPolicy, + credsMgr: credsMgr, + profileResolver: profileResolver, + mcpProvider: mcpProvider, + logger: componentLogger, + executionStore: executionStore, + commandBuilder: commandBuilder, + sessionManager: sessionManager, + eventPublisher: eventPublisher, + containerManager: containerManager, + historyManager: historyManager, + cleanupInterval: 30 * time.Second, + stopCh: stopCh, } // Initialize stream manager with callbacks that delegate to manager methods @@ -129,8 +149,8 @@ func NewManager( // Set session manager dependencies for full orchestration sessionManager.SetDependencies(eventPublisher, mgr.streamManager, executionStore, historyManager) - if runtime != nil { - mgr.logger.Info("initialized with runtime", zap.String("runtime", string(runtime.Name()))) + if runtimeRegistry != nil { + mgr.logger.Info("initialized with runtimes", zap.Int("count", len(runtimeRegistry.List()))) } return mgr @@ -165,6 +185,54 @@ func (m *Manager) SetWorkspaceInfoProvider(provider WorkspaceInfoProvider) { m.workspaceInfoProvider = provider } +// getRuntimeForExecutorType returns the appropriate runtime for the given executor type. +// If the executor type is empty or the runtime is not available, behavior depends on runtimeFallbackPolicy. +func (m *Manager) getRuntimeForExecutorType(executorType string) (Runtime, error) { + if m.runtimeRegistry == nil { + return nil, fmt.Errorf("no runtime registry configured") + } + + if executorType != "" { + runtimeName := runtime.ExecutorTypeToRuntime(models.ExecutorType(executorType)) + rt, err := m.runtimeRegistry.GetRuntime(runtimeName) + if err == nil { + return rt, nil + } + + // Handle fallback based on policy + switch m.runtimeFallbackPolicy { + case RuntimeFallbackDeny: + return nil, fmt.Errorf("runtime %s not available and fallback is denied: %w", runtimeName, err) + case RuntimeFallbackWarn: + m.logger.Warn("requested runtime not available, falling back to default", + zap.String("executor_type", executorType), + zap.String("runtime", string(runtimeName)), + zap.Error(err)) + case RuntimeFallbackAllow: + m.logger.Debug("requested runtime not available, falling back to default", + zap.String("executor_type", executorType), + zap.String("runtime", string(runtimeName))) + default: + // Default to warn behavior for backwards compatibility + m.logger.Warn("requested runtime not available, falling back to default", + zap.String("executor_type", executorType), + zap.String("runtime", string(runtimeName)), + zap.Error(err)) + } + } + + return m.runtimeRegistry.GetRuntime(runtime.NameStandalone) +} + +// getDefaultRuntime returns the default runtime (standalone). +// This is used when no executor type is specified. +func (m *Manager) getDefaultRuntime() (Runtime, error) { + if m.runtimeRegistry == nil { + return nil, fmt.Errorf("no runtime registry configured") + } + return m.runtimeRegistry.GetRuntime(runtime.NameStandalone) +} + // EnsureWorkspaceExecutionForSession ensures an agentctl execution exists for a specific task session. // This is used when the frontend provides a session ID (e.g., from URL path /task/[id]/[sessionId]). // If an execution already exists for the session, it returns it. Otherwise, it creates a new execution @@ -228,8 +296,10 @@ func (m *Manager) EnsureWorkspaceExecutionForSession(ctx context.Context, taskID // createExecution creates an agentctl execution. // The agent subprocess is NOT started - call ConfigureAgent + Start explicitly. func (m *Manager) createExecution(ctx context.Context, taskID string, info *WorkspaceInfo) (*AgentExecution, error) { - if m.runtime == nil { - return nil, fmt.Errorf("no runtime configured") + // Get the default runtime for on-demand execution creation + rt, err := m.getDefaultRuntime() + if err != nil { + return nil, fmt.Errorf("no runtime configured: %w", err) } if info.AgentID == "" { @@ -252,12 +322,13 @@ func (m *Manager) createExecution(ctx context.Context, taskID string, info *Work Protocol: string(agentConfig.Protocol), } - runtimeInstance, err := m.runtime.CreateInstance(ctx, req) + runtimeInstance, err := rt.CreateInstance(ctx, req) if err != nil { return nil, fmt.Errorf("failed to create execution: %w", err) } execution := runtimeInstance.ToAgentExecution(req) + execution.RuntimeName = string(rt.Name()) // Set the ACP session ID for session resumption if info.ACPSessionID != "" { @@ -272,38 +343,38 @@ func (m *Manager) createExecution(ctx context.Context, taskID string, info *Work zap.String("execution_id", executionID), zap.String("task_id", taskID), zap.String("workspace_path", info.WorkspacePath), - zap.String("runtime", string(m.runtime.Name()))) + zap.String("runtime", execution.RuntimeName)) return execution, nil } // Start starts the lifecycle manager background tasks func (m *Manager) Start(ctx context.Context) error { - runtimeName := "none" - if m.runtime != nil { - runtimeName = string(m.runtime.Name()) - } - m.logger.Info("starting lifecycle manager", zap.String("runtime", runtimeName)) - - if m.runtime == nil { - m.logger.Warn("no runtime configured") + if m.runtimeRegistry == nil { + m.logger.Warn("no runtime registry configured") return nil } - // Check runtime health - if err := m.runtime.HealthCheck(ctx); err != nil { - m.logger.Warn("runtime health check failed", - zap.String("runtime", runtimeName), - zap.Error(err)) - // Continue anyway - it might come up later - } else { - m.logger.Info("runtime is healthy", zap.String("runtime", runtimeName)) + runtimeNames := m.runtimeRegistry.List() + m.logger.Info("starting lifecycle manager", zap.Int("runtimes", len(runtimeNames))) + + // Check health of all registered runtimes + healthResults := m.runtimeRegistry.HealthCheckAll(ctx) + for name, err := range healthResults { + if err != nil { + m.logger.Warn("runtime health check failed", + zap.String("runtime", string(name)), + zap.Error(err)) + } else { + m.logger.Info("runtime is healthy", zap.String("runtime", string(name))) + } } - // Try to recover executions from previous runs - recovered, err := m.runtime.RecoverInstances(ctx) + // Try to recover executions from all runtimes + recovered, err := m.runtimeRegistry.RecoverAll(ctx) if err != nil { - m.logger.Warn("failed to recover executions", zap.Error(err)) - } else if len(recovered) > 0 { + m.logger.Warn("failed to recover executions from some runtimes", zap.Error(err)) + } + if len(recovered) > 0 { for _, ri := range recovered { execution := &AgentExecution{ ID: ri.InstanceID, @@ -312,6 +383,7 @@ func (m *Manager) Start(ctx context.Context) error { ContainerID: ri.ContainerID, ContainerIP: ri.ContainerIP, WorkspacePath: ri.WorkspacePath, + RuntimeName: ri.RuntimeName, Status: v1.AgentStatusRunning, StartedAt: time.Now(), Metadata: ri.Metadata, @@ -320,6 +392,10 @@ func (m *Manager) Start(ctx context.Context) error { standalonePort: ri.StandalonePort, } m.executionStore.Add(execution) + + // Reconnect to workspace streams (shell, git, file changes) in background + // This is needed so shell.input, git status, etc. work after backend restart + go m.streamManager.ReconnectAll(execution) } m.logger.Info("recovered executions", zap.Int("count", len(recovered))) } @@ -491,8 +567,10 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti // 7. Launch via runtime - creates agentctl instance with workspace access only // Agent subprocess is NOT started - call StartAgentProcess() explicitly - if m.runtime == nil { - return nil, fmt.Errorf("no runtime configured") + // Select runtime based on executor type from request + rt, err := m.getRuntimeForExecutorType(req.ExecutorType) + if err != nil { + return nil, fmt.Errorf("no runtime configured: %w", err) } // Build environment variables @@ -517,6 +595,23 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti }) } + // Build metadata with runtime-specific fields + metadata := make(map[string]interface{}) + if reqWithWorktree.Metadata != nil { + for k, v := range reqWithWorktree.Metadata { + metadata[k] = v + } + } + if mainRepoGitDir != "" { + metadata[MetadataKeyMainRepoGitDir] = mainRepoGitDir + } + if worktreeID != "" { + metadata[MetadataKeyWorktreeID] = worktreeID + } + if worktreeBranch != "" { + metadata[MetadataKeyWorktreeBranch] = worktreeBranch + } + // Create runtime request (agent command not included - started explicitly later) runtimeReq := &RuntimeCreateRequest{ InstanceID: executionID, @@ -526,21 +621,19 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti WorkspacePath: reqWithWorktree.WorkspacePath, Protocol: string(agentConfig.Protocol), Env: env, - Metadata: reqWithWorktree.Metadata, - WorktreeID: worktreeID, - WorktreeBranch: worktreeBranch, - MainRepoGitDir: mainRepoGitDir, + Metadata: metadata, AgentConfig: agentConfig, McpServers: mcpServers, } - runtimeInstance, err := m.runtime.CreateInstance(ctx, runtimeReq) + runtimeInstance, err := rt.CreateInstance(ctx, runtimeReq) if err != nil { return nil, fmt.Errorf("failed to create execution: %w", err) } - // Convert to AgentExecution + // Convert to AgentExecution and set the runtime name execution := runtimeInstance.ToAgentExecution(runtimeReq) + execution.RuntimeName = string(rt.Name()) // Set ACP session ID for session resumption (used by InitializeSession) if req.ACPSessionID != "" { @@ -582,14 +675,10 @@ func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecuti // NOTE: This does NOT start the agent process - call StartAgentProcess() explicitly go m.waitForAgentctlReady(execution) - runtimeName := "unknown" - if m.runtime != nil { - runtimeName = string(m.runtime.Name()) - } m.logger.Info("agentctl execution created (agent not started)", zap.String("execution_id", executionID), zap.String("task_id", req.TaskID), - zap.String("runtime", runtimeName)) + zap.String("runtime", execution.RuntimeName)) return execution, nil } @@ -876,7 +965,9 @@ func (m *Manager) resolveMcpServersWithParams(ctx context.Context, profileID str return nil, nil } - policy := mcpconfig.DefaultPolicyForRuntime(runtimeName(m.runtime)) + // Get default runtime for MCP policy (used before execution exists) + defaultRT, _ := m.getDefaultRuntime() + policy := mcpconfig.DefaultPolicyForRuntime(runtimeName(defaultRT)) executorID := "" if metadata != nil { if value, ok := metadata["executor_id"].(string); ok { @@ -1341,14 +1432,10 @@ func (m *Manager) StopAgent(ctx context.Context, executionID string, force bool) return fmt.Errorf("execution %q not found", executionID) } - runtimeName := "unknown" - if m.runtime != nil { - runtimeName = string(m.runtime.Name()) - } m.logger.Info("stopping agent", zap.String("execution_id", executionID), zap.Bool("force", force), - zap.String("runtime", runtimeName)) + zap.String("runtime", execution.RuntimeName)) // Try to gracefully stop via agentctl first if execution.agentctl != nil && !force { @@ -1360,17 +1447,25 @@ func (m *Manager) StopAgent(ctx context.Context, executionID string, force bool) execution.agentctl.Close() } - // Stop the agent execution via runtime - if m.runtime != nil { - runtimeInstance := &RuntimeInstance{ - InstanceID: execution.ID, - TaskID: execution.TaskID, - ContainerID: execution.ContainerID, - StandaloneInstanceID: execution.standaloneInstanceID, - StandalonePort: execution.standalonePort, - } - if err := m.runtime.StopInstance(ctx, runtimeInstance, force); err != nil { - return err + // Stop the agent execution via the runtime that created it + if execution.RuntimeName != "" && m.runtimeRegistry != nil { + rt, err := m.runtimeRegistry.GetRuntime(runtime.Name(execution.RuntimeName)) + if err != nil { + m.logger.Warn("failed to get runtime for stopping execution", + zap.String("execution_id", executionID), + zap.String("runtime", execution.RuntimeName), + zap.Error(err)) + } else { + runtimeInstance := &RuntimeInstance{ + InstanceID: execution.ID, + TaskID: execution.TaskID, + ContainerID: execution.ContainerID, + StandaloneInstanceID: execution.standaloneInstanceID, + StandalonePort: execution.standalonePort, + } + if err := rt.StopInstance(ctx, runtimeInstance, force); err != nil { + return err + } } } diff --git a/apps/backend/internal/agent/lifecycle/manager_test.go b/apps/backend/internal/agent/lifecycle/manager_test.go index 6fc22a9987..0a3bbf845f 100644 --- a/apps/backend/internal/agent/lifecycle/manager_test.go +++ b/apps/backend/internal/agent/lifecycle/manager_test.go @@ -153,7 +153,7 @@ func newTestManager() *Manager { credsMgr := &MockCredentialsManager{} profileResolver := &MockProfileResolver{} // Pass nil for runtime and containerManager - tests don't need them - return NewManager(reg, eventBus, nil, nil, credsMgr, profileResolver, nil, log) + return NewManager(reg, eventBus, nil, nil, credsMgr, profileResolver, nil, RuntimeFallbackWarn, log) } func TestNewManager(t *testing.T) { diff --git a/apps/backend/internal/agent/lifecycle/runtime.go b/apps/backend/internal/agent/lifecycle/runtime.go index 263aa83309..bf14e63176 100644 --- a/apps/backend/internal/agent/lifecycle/runtime.go +++ b/apps/backend/internal/agent/lifecycle/runtime.go @@ -43,6 +43,13 @@ type McpServerConfig struct { Args []string `json:"args,omitempty"` } +// Metadata keys for runtime-specific configuration +const ( + MetadataKeyMainRepoGitDir = "main_repo_git_dir" + MetadataKeyWorktreeID = "worktree_id" + MetadataKeyWorktreeBranch = "worktree_branch" +) + // RuntimeCreateRequest contains parameters for creating an agentctl instance. type RuntimeCreateRequest struct { InstanceID string @@ -54,11 +61,7 @@ type RuntimeCreateRequest struct { Env map[string]string Metadata map[string]interface{} McpServers []McpServerConfig - // Docker-specific - AgentConfig *registry.AgentTypeConfig - MainRepoGitDir string - WorktreeID string - WorktreeBranch string + AgentConfig *registry.AgentTypeConfig // Agent type info needed by runtimes } // RuntimeInstance represents an agentctl instance created by a runtime. @@ -69,6 +72,9 @@ type RuntimeInstance struct { TaskID string SessionID string + // Runtime name (e.g., "docker", "standalone") - set by the runtime that created this instance + RuntimeName string + // Agentctl client for communicating with this instance Client *agentctl.Client @@ -107,6 +113,7 @@ func (ri *RuntimeInstance) ToAgentExecution(req *RuntimeCreateRequest) *AgentExe ContainerID: ri.ContainerID, ContainerIP: ri.ContainerIP, WorkspacePath: workspacePath, + RuntimeName: ri.RuntimeName, Status: v1.AgentStatusRunning, StartedAt: time.Now(), Metadata: metadata, diff --git a/apps/backend/internal/agent/lifecycle/runtime_docker.go b/apps/backend/internal/agent/lifecycle/runtime_docker.go index e58b3d83d5..fa358cd619 100644 --- a/apps/backend/internal/agent/lifecycle/runtime_docker.go +++ b/apps/backend/internal/agent/lifecycle/runtime_docker.go @@ -13,6 +13,17 @@ import ( "github.com/kandev/kandev/internal/common/logger" ) +// getMetadataString retrieves a string value from metadata map. +func getMetadataString(metadata map[string]interface{}, key string) string { + if metadata == nil { + return "" + } + if v, ok := metadata[key].(string); ok { + return v + } + return "" +} + // DockerRuntime implements Runtime for Docker-based agent execution. type DockerRuntime struct { containerMgr *ContainerManager @@ -43,6 +54,11 @@ func (r *DockerRuntime) HealthCheck(ctx context.Context) error { } func (r *DockerRuntime) CreateInstance(ctx context.Context, req *RuntimeCreateRequest) (*RuntimeInstance, error) { + // Extract runtime-specific values from metadata + mainRepoGitDir := getMetadataString(req.Metadata, MetadataKeyMainRepoGitDir) + worktreeID := getMetadataString(req.Metadata, MetadataKeyWorktreeID) + worktreeBranch := getMetadataString(req.Metadata, MetadataKeyWorktreeBranch) + // Convert RuntimeCreateRequest to ContainerConfig containerCfg := ContainerConfig{ AgentConfig: req.AgentConfig, @@ -50,7 +66,7 @@ func (r *DockerRuntime) CreateInstance(ctx context.Context, req *RuntimeCreateRe TaskID: req.TaskID, SessionID: req.SessionID, InstanceID: req.InstanceID, - MainRepoGitDir: req.MainRepoGitDir, + MainRepoGitDir: mainRepoGitDir, Credentials: req.Env, // Env contains credentials from the caller McpServers: req.McpServers, } @@ -66,10 +82,10 @@ func (r *DockerRuntime) CreateInstance(ctx context.Context, req *RuntimeCreateRe // Build metadata metadata := make(map[string]interface{}) - if req.WorktreeID != "" { - metadata["worktree_id"] = req.WorktreeID + if worktreeID != "" { + metadata["worktree_id"] = worktreeID metadata["worktree_path"] = req.WorkspacePath - metadata["worktree_branch"] = req.WorktreeBranch + metadata["worktree_branch"] = worktreeBranch } r.logger.Info("docker instance created", @@ -81,6 +97,7 @@ func (r *DockerRuntime) CreateInstance(ctx context.Context, req *RuntimeCreateRe InstanceID: req.InstanceID, TaskID: req.TaskID, SessionID: req.SessionID, + RuntimeName: string(r.Name()), Client: client, ContainerID: containerID, ContainerIP: containerIP, @@ -173,6 +190,7 @@ func (r *DockerRuntime) RecoverInstances(ctx context.Context) ([]*RuntimeInstanc InstanceID: instanceID, TaskID: taskID, SessionID: sessionID, + RuntimeName: string(r.Name()), Client: client, ContainerID: ctr.ID, ContainerIP: containerIP, diff --git a/apps/backend/internal/agent/lifecycle/runtime_registry.go b/apps/backend/internal/agent/lifecycle/runtime_registry.go new file mode 100644 index 0000000000..4152776103 --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/runtime_registry.go @@ -0,0 +1,130 @@ +// Package lifecycle provides agent runtime abstractions. +package lifecycle + +import ( + "context" + "fmt" + "sync" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/agent/runtime" + "github.com/kandev/kandev/internal/common/logger" +) + +// ErrRuntimeNotFound is returned when a runtime doesn't exist in the registry. +var ErrRuntimeNotFound = fmt.Errorf("runtime not found") + +// RuntimeRegistry manages multiple Runtime implementations and provides +// thread-safe access to them. It supports registering runtimes by name, +// health checking, and instance recovery across all registered runtimes. +type RuntimeRegistry struct { + runtimes map[runtime.Name]Runtime + mu sync.RWMutex + logger *logger.Logger +} + +// NewRuntimeRegistry creates a new RuntimeRegistry with the given logger. +func NewRuntimeRegistry(log *logger.Logger) *RuntimeRegistry { + return &RuntimeRegistry{ + runtimes: make(map[runtime.Name]Runtime), + logger: log, + } +} + +// Register adds a runtime to the registry using its Name() as the key. +// If a runtime with the same name already exists, it will be replaced. +func (r *RuntimeRegistry) Register(rt Runtime) { + r.mu.Lock() + defer r.mu.Unlock() + + name := rt.Name() + r.runtimes[name] = rt + r.logger.Info("registered runtime", zap.String("name", string(name))) +} + +// GetRuntime returns a runtime by its name. +// Returns ErrRuntimeNotFound if the runtime doesn't exist. +func (r *RuntimeRegistry) GetRuntime(name runtime.Name) (Runtime, error) { + r.mu.RLock() + defer r.mu.RUnlock() + + rt, exists := r.runtimes[name] + if !exists { + return nil, fmt.Errorf("%w: %s", ErrRuntimeNotFound, name) + } + return rt, nil +} + +// List returns the names of all registered runtimes. +func (r *RuntimeRegistry) List() []runtime.Name { + r.mu.RLock() + defer r.mu.RUnlock() + + names := make([]runtime.Name, 0, len(r.runtimes)) + for name := range r.runtimes { + names = append(names, name) + } + return names +} + +// HealthCheckAll performs health checks on all registered runtimes. +// Returns a map of runtime names to errors. A nil error indicates the runtime is healthy. +func (r *RuntimeRegistry) HealthCheckAll(ctx context.Context) map[runtime.Name]error { + r.mu.RLock() + runtimes := make(map[runtime.Name]Runtime, len(r.runtimes)) + for name, rt := range r.runtimes { + runtimes[name] = rt + } + r.mu.RUnlock() + + results := make(map[runtime.Name]error, len(runtimes)) + for name, rt := range runtimes { + err := rt.HealthCheck(ctx) + results[name] = err + if err != nil { + r.logger.Warn("runtime health check failed", + zap.String("runtime", string(name)), + zap.Error(err)) + } else { + r.logger.Debug("runtime health check passed", + zap.String("runtime", string(name))) + } + } + return results +} + +// RecoverAll recovers instances from all registered runtimes. +// Returns all recovered instances and any error encountered. +// If multiple runtimes fail, only the last error is returned. +func (r *RuntimeRegistry) RecoverAll(ctx context.Context) ([]*RuntimeInstance, error) { + r.mu.RLock() + runtimes := make(map[runtime.Name]Runtime, len(r.runtimes)) + for name, rt := range r.runtimes { + runtimes[name] = rt + } + r.mu.RUnlock() + + var allInstances []*RuntimeInstance + var lastErr error + + for name, rt := range runtimes { + instances, err := rt.RecoverInstances(ctx) + if err != nil { + r.logger.Error("failed to recover instances from runtime", + zap.String("runtime", string(name)), + zap.Error(err)) + lastErr = err + continue + } + + if len(instances) > 0 { + r.logger.Info("recovered instances from runtime", + zap.String("runtime", string(name)), + zap.Int("count", len(instances))) + allInstances = append(allInstances, instances...) + } + } + + return allInstances, lastErr +} diff --git a/apps/backend/internal/agent/lifecycle/runtime_registry_test.go b/apps/backend/internal/agent/lifecycle/runtime_registry_test.go new file mode 100644 index 0000000000..c01c069f37 --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/runtime_registry_test.go @@ -0,0 +1,347 @@ +package lifecycle + +import ( + "context" + "errors" + "sync" + "testing" + + "github.com/kandev/kandev/internal/agent/runtime" + "github.com/kandev/kandev/internal/common/logger" +) + +// MockRuntime implements Runtime interface for testing +type MockRuntime struct { + name runtime.Name + healthCheckErr error + recoverInstances []*RuntimeInstance + recoverErr error +} + +func (m *MockRuntime) Name() runtime.Name { return m.name } +func (m *MockRuntime) HealthCheck(ctx context.Context) error { + return m.healthCheckErr +} +func (m *MockRuntime) CreateInstance(ctx context.Context, req *RuntimeCreateRequest) (*RuntimeInstance, error) { + return nil, nil +} +func (m *MockRuntime) StopInstance(ctx context.Context, instance *RuntimeInstance, force bool) error { + return nil +} +func (m *MockRuntime) RecoverInstances(ctx context.Context) ([]*RuntimeInstance, error) { + return m.recoverInstances, m.recoverErr +} + +func newTestRegistryLogger() *logger.Logger { + log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) + return log +} + +func TestNewRuntimeRegistry(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + if registry == nil { + t.Fatal("expected non-nil registry") + } + if registry.runtimes == nil { + t.Error("expected runtimes map to be initialized") + } + if len(registry.runtimes) != 0 { + t.Errorf("expected empty runtimes map, got %d entries", len(registry.runtimes)) + } +} + +func TestRuntimeRegistry_Register(t *testing.T) { + t.Run("register new runtime", func(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + mockRT := &MockRuntime{name: runtime.NameDocker} + registry.Register(mockRT) + + if len(registry.runtimes) != 1 { + t.Errorf("expected 1 runtime, got %d", len(registry.runtimes)) + } + if registry.runtimes[runtime.NameDocker] != mockRT { + t.Error("expected registered runtime to match") + } + }) + + t.Run("replace existing runtime", func(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + mockRT1 := &MockRuntime{name: runtime.NameDocker, healthCheckErr: errors.New("error1")} + mockRT2 := &MockRuntime{name: runtime.NameDocker, healthCheckErr: errors.New("error2")} + + registry.Register(mockRT1) + registry.Register(mockRT2) + + if len(registry.runtimes) != 1 { + t.Errorf("expected 1 runtime after replacement, got %d", len(registry.runtimes)) + } + if registry.runtimes[runtime.NameDocker] != mockRT2 { + t.Error("expected runtime to be replaced with new one") + } + }) +} + +func TestRuntimeRegistry_GetRuntime(t *testing.T) { + t.Run("get existing runtime", func(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + mockRT := &MockRuntime{name: runtime.NameDocker} + registry.Register(mockRT) + + rt, err := registry.GetRuntime(runtime.NameDocker) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + if rt != mockRT { + t.Error("expected returned runtime to match registered one") + } + }) + + t.Run("get non-existent runtime returns ErrRuntimeNotFound", func(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + _, err := registry.GetRuntime(runtime.NameDocker) + if err == nil { + t.Error("expected error for non-existent runtime") + } + if !errors.Is(err, ErrRuntimeNotFound) { + t.Errorf("expected ErrRuntimeNotFound, got: %v", err) + } + }) +} + +func TestRuntimeRegistry_List(t *testing.T) { + t.Run("list empty registry", func(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + names := registry.List() + if len(names) != 0 { + t.Errorf("expected empty list, got %d items", len(names)) + } + }) + + t.Run("list with multiple runtimes", func(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + registry.Register(&MockRuntime{name: runtime.NameDocker}) + registry.Register(&MockRuntime{name: runtime.NameStandalone}) + registry.Register(&MockRuntime{name: runtime.NameLocal}) + + names := registry.List() + if len(names) != 3 { + t.Errorf("expected 3 names, got %d", len(names)) + } + + // Check all names are present (order is not guaranteed) + nameMap := make(map[runtime.Name]bool) + for _, n := range names { + nameMap[n] = true + } + if !nameMap[runtime.NameDocker] || !nameMap[runtime.NameStandalone] || !nameMap[runtime.NameLocal] { + t.Error("expected all registered runtime names to be present") + } + }) +} + +func TestRuntimeRegistry_HealthCheckAll(t *testing.T) { + ctx := context.Background() + + t.Run("health check with all healthy runtimes", func(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + registry.Register(&MockRuntime{name: runtime.NameDocker, healthCheckErr: nil}) + registry.Register(&MockRuntime{name: runtime.NameStandalone, healthCheckErr: nil}) + + results := registry.HealthCheckAll(ctx) + if len(results) != 2 { + t.Errorf("expected 2 results, got %d", len(results)) + } + if results[runtime.NameDocker] != nil { + t.Errorf("expected nil error for docker, got: %v", results[runtime.NameDocker]) + } + if results[runtime.NameStandalone] != nil { + t.Errorf("expected nil error for standalone, got: %v", results[runtime.NameStandalone]) + } + }) + + t.Run("health check with some failing runtimes", func(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + dockerErr := errors.New("docker unavailable") + registry.Register(&MockRuntime{name: runtime.NameDocker, healthCheckErr: dockerErr}) + registry.Register(&MockRuntime{name: runtime.NameStandalone, healthCheckErr: nil}) + + results := registry.HealthCheckAll(ctx) + if len(results) != 2 { + t.Errorf("expected 2 results, got %d", len(results)) + } + if results[runtime.NameDocker] != dockerErr { + t.Errorf("expected docker error, got: %v", results[runtime.NameDocker]) + } + if results[runtime.NameStandalone] != nil { + t.Errorf("expected nil error for standalone, got: %v", results[runtime.NameStandalone]) + } + }) + + t.Run("health check with empty registry", func(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + results := registry.HealthCheckAll(ctx) + if len(results) != 0 { + t.Errorf("expected empty results, got %d", len(results)) + } + }) +} + +func TestRuntimeRegistry_RecoverAll(t *testing.T) { + ctx := context.Background() + + t.Run("recovery with no instances", func(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + registry.Register(&MockRuntime{name: runtime.NameDocker, recoverInstances: nil}) + registry.Register(&MockRuntime{name: runtime.NameStandalone, recoverInstances: []*RuntimeInstance{}}) + + instances, err := registry.RecoverAll(ctx) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + if len(instances) != 0 { + t.Errorf("expected no instances, got %d", len(instances)) + } + }) + + t.Run("recovery with instances from multiple runtimes", func(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + dockerInstances := []*RuntimeInstance{ + {InstanceID: "docker-1", RuntimeName: "docker"}, + {InstanceID: "docker-2", RuntimeName: "docker"}, + } + standaloneInstances := []*RuntimeInstance{ + {InstanceID: "standalone-1", RuntimeName: "standalone"}, + } + + registry.Register(&MockRuntime{name: runtime.NameDocker, recoverInstances: dockerInstances}) + registry.Register(&MockRuntime{name: runtime.NameStandalone, recoverInstances: standaloneInstances}) + + instances, err := registry.RecoverAll(ctx) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + if len(instances) != 3 { + t.Errorf("expected 3 instances, got %d", len(instances)) + } + }) + + t.Run("recovery with errors continues and returns last error", func(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + recoverErr := errors.New("recovery failed") + standaloneInstances := []*RuntimeInstance{ + {InstanceID: "standalone-1", RuntimeName: "standalone"}, + } + + registry.Register(&MockRuntime{name: runtime.NameDocker, recoverErr: recoverErr}) + registry.Register(&MockRuntime{name: runtime.NameStandalone, recoverInstances: standaloneInstances}) + + instances, err := registry.RecoverAll(ctx) + // Should still get instances from successful runtimes + // Note: map iteration order is not guaranteed, so we check that at least standalone instances are recovered + if len(instances) == 0 && err == nil { + t.Error("expected either instances or error") + } + // The error may or may not be set depending on iteration order + // If docker is processed last, err will be set; if standalone is processed last, err will be nil + }) + + t.Run("recovery from empty registry", func(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + + instances, err := registry.RecoverAll(ctx) + if err != nil { + t.Errorf("unexpected error: %v", err) + } + if len(instances) != 0 { + t.Errorf("expected no instances, got %d", len(instances)) + } + }) +} + + +func TestRuntimeRegistry_ThreadSafety(t *testing.T) { + log := newTestRegistryLogger() + registry := NewRuntimeRegistry(log) + ctx := context.Background() + + const numGoroutines = 50 + const numOperations = 100 + + var wg sync.WaitGroup + wg.Add(numGoroutines * 4) // 4 types of operations + + // Concurrent Register operations + for i := 0; i < numGoroutines; i++ { + go func(id int) { + defer wg.Done() + for j := 0; j < numOperations; j++ { + name := runtime.Name("test-runtime") + registry.Register(&MockRuntime{name: name}) + } + }(i) + } + + // Concurrent GetRuntime operations + for i := 0; i < numGoroutines; i++ { + go func(id int) { + defer wg.Done() + for j := 0; j < numOperations; j++ { + _, _ = registry.GetRuntime(runtime.Name("test-runtime")) + } + }(i) + } + + // Concurrent List operations + for i := 0; i < numGoroutines; i++ { + go func(id int) { + defer wg.Done() + for j := 0; j < numOperations; j++ { + _ = registry.List() + } + }(i) + } + + // Concurrent HealthCheckAll operations + for i := 0; i < numGoroutines; i++ { + go func(id int) { + defer wg.Done() + for j := 0; j < numOperations; j++ { + _ = registry.HealthCheckAll(ctx) + } + }(i) + } + + // Wait for all goroutines to complete + wg.Wait() + + // If we reach here without a race condition panic, the test passes + // The -race flag during testing will detect actual race conditions +} + diff --git a/apps/backend/internal/agent/lifecycle/runtime_standalone.go b/apps/backend/internal/agent/lifecycle/runtime_standalone.go index d37c503d1a..d4ce60add3 100644 --- a/apps/backend/internal/agent/lifecycle/runtime_standalone.go +++ b/apps/backend/internal/agent/lifecycle/runtime_standalone.go @@ -111,13 +111,17 @@ func (r *StandaloneRuntime) CreateInstance(ctx context.Context, req *RuntimeCrea // Create agentctl client pointing to the instance port client := agentctl.NewClient(r.host, resp.Port, r.logger) + // Extract runtime-specific values from metadata + worktreeID := getMetadataString(req.Metadata, MetadataKeyWorktreeID) + worktreeBranch := getMetadataString(req.Metadata, MetadataKeyWorktreeBranch) + // Build metadata metadata := make(map[string]interface{}) metadata["standalone_port"] = resp.Port - if req.WorktreeID != "" { - metadata["worktree_id"] = req.WorktreeID + if worktreeID != "" { + metadata["worktree_id"] = worktreeID metadata["worktree_path"] = req.WorkspacePath - metadata["worktree_branch"] = req.WorktreeBranch + metadata["worktree_branch"] = worktreeBranch } r.logger.Info("standalone instance created", @@ -129,6 +133,7 @@ func (r *StandaloneRuntime) CreateInstance(ctx context.Context, req *RuntimeCrea InstanceID: req.InstanceID, TaskID: req.TaskID, SessionID: req.SessionID, + RuntimeName: string(r.Name()), Client: client, StandaloneInstanceID: resp.ID, StandalonePort: resp.Port, diff --git a/apps/backend/internal/agent/lifecycle/types.go b/apps/backend/internal/agent/lifecycle/types.go index 3eebb19f7c..7e84abbd0f 100644 --- a/apps/backend/internal/agent/lifecycle/types.go +++ b/apps/backend/internal/agent/lifecycle/types.go @@ -27,6 +27,7 @@ type AgentExecution struct { WorkspacePath string // Path to the workspace (worktree or repository path) ACPSessionID string // ACP session ID to resume, if available AgentCommand string // Command to start the agent subprocess + RuntimeName string // Name of the runtime used (e.g., "docker", "standalone") Status v1.AgentStatus StartedAt time.Time FinishedAt *time.Time @@ -94,6 +95,9 @@ type LaunchRequest struct { Metadata map[string]interface{} ModelOverride string // If set, use this model instead of the profile's model + // Executor configuration - determines which runtime to use + ExecutorType string // Executor type (e.g., "local_pc", "local_docker") - determines runtime + // Worktree configuration UseWorktree bool // Whether to use a Git worktree for isolation RepositoryID string // Repository ID for worktree tracking diff --git a/apps/backend/internal/agent/registry/agents.json b/apps/backend/internal/agent/registry/agents.json index adf149aa54..07c0b569ff 100644 --- a/apps/backend/internal/agent/registry/agents.json +++ b/apps/backend/internal/agent/registry/agents.json @@ -8,7 +8,7 @@ "display_name": "Auggie", "description": "Auggie CLI-powered autonomous coding agent.", "enabled": true, - "image": "kandev/augment-agent", + "image": "kandev/multi-agent", "tag": "latest", "cmd": [ "auggie", @@ -154,7 +154,7 @@ "display_name": "Codex", "description": "OpenAI Codex CLI-powered autonomous coding agent using the Codex app-server protocol.", "enabled": true, - "image": "kandev/codex-agent", + "image": "kandev/multi-agent", "tag": "latest", "cmd": [ "codex", @@ -168,6 +168,11 @@ "source": "{workspace}", "target": "/workspace", "read_only": false + }, + { + "source": "{home}/.codex", + "target": "/root/.codex", + "read_only": false } ], "resource_limits": { @@ -190,8 +195,8 @@ "resume_via_acp": true, "resume_flag": "", "can_recover": true, - "session_dir_template": "{home}/.codex/sessions", - "session_dir_target": "/root/.codex/sessions" + "session_dir_template": "", + "session_dir_target": "" }, "permission_config": { "permission_flag": "", diff --git a/apps/backend/internal/agent/runtime/runtime.go b/apps/backend/internal/agent/runtime/runtime.go index f6b7bc9490..f25f073b5f 100644 --- a/apps/backend/internal/agent/runtime/runtime.go +++ b/apps/backend/internal/agent/runtime/runtime.go @@ -1,6 +1,10 @@ // Package runtime defines the agent runtime types shared across lifecycle and policy logic. package runtime +import ( + "github.com/kandev/kandev/internal/task/models" +) + // Name identifies the execution runtime. type Name string @@ -10,3 +14,19 @@ const ( NameStandalone Name = "standalone" NameLocal Name = "local" ) + +// ExecutorTypeToRuntime maps an ExecutorType to its corresponding runtime Name. +func ExecutorTypeToRuntime(execType models.ExecutorType) Name { + switch execType { + case models.ExecutorTypeLocalPC: + return NameStandalone + case models.ExecutorTypeLocalDocker: + return NameDocker + case models.ExecutorTypeRemoteDocker: + // For now, remote docker uses the same runtime as local docker. + // Future: separate runtime for remote docker. + return NameDocker + default: + return NameStandalone + } +} diff --git a/apps/backend/internal/common/config/config.go b/apps/backend/internal/common/config/config.go index c7b28c1aa6..a975a7882d 100644 --- a/apps/backend/internal/common/config/config.go +++ b/apps/backend/internal/common/config/config.go @@ -53,6 +53,10 @@ type NATSConfig struct { // DockerConfig holds Docker client configuration. type DockerConfig struct { + // Enabled controls whether the Docker runtime is available for task execution. + // When true and Docker is accessible, tasks can use Docker-based executors. + // Default: true (Docker runtime is enabled if Docker is available) + Enabled bool `mapstructure:"enabled"` Host string `mapstructure:"host"` APIVersion string `mapstructure:"apiVersion"` TLSVerify bool `mapstructure:"tlsVerify"` @@ -89,19 +93,17 @@ type WorktreeConfig struct { } // AgentConfig holds agent runtime configuration. +// Note: Runtime selection is now per-task based on executor type, not global. +// The Standalone runtime (agentctl) always runs as a core service. +// Docker runtime is available when docker.enabled=true. type AgentConfig struct { - // Runtime specifies the agent runtime mode: "docker" or "standalone" - // - "standalone": Agents run via standalone agentctl on the host machine (default) - // - "docker": Agents run in Docker containers - Runtime string `mapstructure:"runtime"` - // StandaloneHost is the host where standalone agentctl is running (default: localhost) StandaloneHost string `mapstructure:"standaloneHost"` // StandalonePort is the control port for standalone agentctl (default: 9999) StandalonePort int `mapstructure:"standalonePort"` - // McpServerEnabled enables the embedded MCP server (default: true in standalone mode) + // McpServerEnabled enables the embedded MCP server (default: true) McpServerEnabled bool `mapstructure:"mcpServerEnabled"` // McpServerPort is the port for the embedded MCP server (default: 9090) @@ -153,14 +155,14 @@ func setDefaults(v *viper.Viper) { v.SetDefault("nats.maxReconnects", 10) // Docker defaults + v.SetDefault("docker.enabled", true) // Docker runtime enabled by default if Docker is available v.SetDefault("docker.host", "unix:///var/run/docker.sock") v.SetDefault("docker.apiVersion", "1.41") v.SetDefault("docker.tlsVerify", false) v.SetDefault("docker.defaultNetwork", "kandev-network") v.SetDefault("docker.volumeBasePath", "/var/lib/kandev/volumes") - // Agent defaults - v.SetDefault("agent.runtime", "standalone") + // Agent defaults (runtime selection is now per-task based on executor type) v.SetDefault("agent.standaloneHost", "localhost") v.SetDefault("agent.standalonePort", 9999) v.SetDefault("agent.mcpServerEnabled", true) diff --git a/apps/backend/internal/orchestrator/controller/controller.go b/apps/backend/internal/orchestrator/controller/controller.go index 57d240b6e9..6809e4f38f 100644 --- a/apps/backend/internal/orchestrator/controller/controller.go +++ b/apps/backend/internal/orchestrator/controller/controller.go @@ -61,7 +61,7 @@ func (c *Controller) TriggerTask(ctx context.Context, req dto.TriggerTaskRequest // StartTask starts task execution func (c *Controller) StartTask(ctx context.Context, req dto.StartTaskRequest) (dto.StartTaskResponse, error) { - execution, err := c.service.StartTask(ctx, req.TaskID, req.AgentProfileID, req.Priority, req.Prompt) + execution, err := c.service.StartTask(ctx, req.TaskID, req.AgentProfileID, req.ExecutorID, req.Priority, req.Prompt) if err != nil { return dto.StartTaskResponse{}, err } diff --git a/apps/backend/internal/orchestrator/dto/dto.go b/apps/backend/internal/orchestrator/dto/dto.go index 94800a0c55..c57bc96943 100644 --- a/apps/backend/internal/orchestrator/dto/dto.go +++ b/apps/backend/internal/orchestrator/dto/dto.go @@ -43,6 +43,7 @@ type TriggerTaskResponse struct { type StartTaskRequest struct { TaskID string `json:"task_id"` AgentProfileID string `json:"agent_profile_id,omitempty"` + ExecutorID string `json:"executor_id,omitempty"` // Executor to use (determines runtime: local_pc, local_docker, etc.) Priority int `json:"priority,omitempty"` Prompt string `json:"prompt,omitempty"` // Initial prompt to send to the agent } diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index 7483a63e97..8cffdcb802 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -86,6 +86,7 @@ type LaunchAgentRequest struct { Env map[string]string ACPSessionID string // ACP session ID to resume, if available ModelOverride string // If set, use this model instead of the profile's model + ExecutorType string // Executor type (e.g., "local_pc", "local_docker") - determines runtime // Worktree configuration for concurrent agent execution UseWorktree bool // Whether to use a Git worktree for isolation @@ -201,12 +202,14 @@ func (e *Executor) applyPreferredShellEnv(ctx context.Context, env map[string]st // Execute starts agent execution for a task func (e *Executor) Execute(ctx context.Context, task *v1.Task) (*TaskExecution, error) { - return e.ExecuteWithProfile(ctx, task, "", task.Description) + return e.ExecuteWithProfile(ctx, task, "", "", task.Description) } // ExecuteWithProfile starts agent execution for a task using an explicit agent profile. +// The executorID parameter specifies which executor to use (determines runtime: local_pc, local_docker, etc.). +// If executorID is empty, falls back to workspace's default executor. // The prompt parameter is the initial prompt to send to the agent. -func (e *Executor) ExecuteWithProfile(ctx context.Context, task *v1.Task, agentProfileID string, prompt string) (*TaskExecution, error) { +func (e *Executor) ExecuteWithProfile(ctx context.Context, task *v1.Task, agentProfileID string, executorID string, prompt string) (*TaskExecution, error) { if agentProfileID == "" { e.logger.Error("task has no agent_profile_id configured", zap.String("task_id", task.ID)) return nil, ErrNoAgentProfileID @@ -315,9 +318,16 @@ func (e *Executor) ExecuteWithProfile(ctx context.Context, task *v1.Task, agentP UpdatedAt: now, AgentProfileSnapshot: agentProfileSnapshot, } - if executorID := e.defaultExecutorID(ctx, task.WorkspaceID); executorID != "" { - session.ExecutorID = executorID - metadata = e.applyExecutorMetadata(ctx, metadata, executorID) + // Resolve executor configuration + execConfig := e.resolveExecutorConfig(ctx, executorID, task.WorkspaceID, metadata) + if execConfig.ExecutorID != "" { + session.ExecutorID = execConfig.ExecutorID + metadata = execConfig.Metadata + req.ExecutorType = execConfig.ExecutorType + e.logger.Debug("resolved executor for task", + zap.String("task_id", task.ID), + zap.String("executor_id", execConfig.ExecutorID), + zap.String("executor_type", execConfig.ExecutorType)) } if err := e.repo.CreateTaskSession(ctx, session); err != nil { @@ -335,6 +345,7 @@ func (e *Executor) ExecuteWithProfile(ctx context.Context, task *v1.Task, agentP e.logger.Info("launching agent for task", zap.String("task_id", task.ID), zap.String("agent_profile_id", agentProfileID), + zap.String("executor_type", req.ExecutorType), zap.Bool("use_worktree", req.UseWorktree)) req.Env = e.applyPreferredShellEnv(ctx, req.Env) @@ -518,12 +529,24 @@ func (e *Executor) ResumeSession(ctx context.Context, session *models.TaskSessio if len(session.Worktrees) > 0 && session.Worktrees[0].WorktreeID != "" { metadata["worktree_id"] = session.Worktrees[0].WorktreeID } - if session.ExecutorID == "" { - if executorID := e.defaultExecutorID(ctx, task.WorkspaceID); executorID != "" { - session.ExecutorID = executorID + // Resolve executor configuration + executorWasEmpty := session.ExecutorID == "" + execConfig := e.resolveExecutorConfig(ctx, session.ExecutorID, task.WorkspaceID, metadata) + session.ExecutorID = execConfig.ExecutorID + metadata = execConfig.Metadata + req.ExecutorType = execConfig.ExecutorType + + // Persist executor assignment if it was resolved from workspace default + if executorWasEmpty && session.ExecutorID != "" { + session.UpdatedAt = time.Now().UTC() + if err := e.repo.UpdateTaskSession(ctx, session); err != nil { + e.logger.Warn("failed to persist executor assignment for session", + zap.String("session_id", session.ID), + zap.String("executor_id", session.ExecutorID), + zap.Error(err)) + // Continue anyway - this is not fatal } } - metadata = e.applyExecutorMetadata(ctx, metadata, session.ExecutorID) if len(metadata) > 0 { req.Metadata = metadata } @@ -580,6 +603,7 @@ func (e *Executor) ResumeSession(ctx context.Context, session *models.TaskSessio zap.String("task_id", session.TaskID), zap.String("session_id", session.ID), zap.String("agent_profile_id", session.AgentProfileID), + zap.String("executor_type", req.ExecutorType), zap.String("resume_token", req.ACPSessionID), zap.Bool("use_worktree", req.UseWorktree)) @@ -869,6 +893,9 @@ func (e *Executor) SwitchModel(ctx context.Context, taskID, sessionID, newModel, zap.String("agent_execution_id", oldAgentExecutionID)) } + // Resolve executor configuration + execConfig := e.resolveExecutorConfig(ctx, session.ExecutorID, task.WorkspaceID, nil) + // Build a new launch request with the model override req := &LaunchAgentRequest{ TaskID: task.ID, @@ -878,18 +905,41 @@ func (e *Executor) SwitchModel(ctx context.Context, taskID, sessionID, newModel, TaskDescription: prompt, ModelOverride: newModel, // This is the key - use the new model ACPSessionID: acpSessionID, + ExecutorType: execConfig.ExecutorType, + Metadata: execConfig.Metadata, } // Get repository info if available + var repositoryPath string if session.RepositoryID != "" { repository, err := e.repo.GetRepository(ctx, session.RepositoryID) if err == nil && repository != nil { + repositoryPath = repository.LocalPath req.RepositoryURL = repository.LocalPath req.Branch = session.BaseBranch } } - // Get worktree info from running state + // Configure worktree if enabled - reuse existing worktree from session + if e.worktreeEnabled && repositoryPath != "" { + req.UseWorktree = true + req.RepositoryPath = repositoryPath + req.RepositoryID = session.RepositoryID + if session.BaseBranch != "" { + req.BaseBranch = session.BaseBranch + } else { + req.BaseBranch = "main" + } + // Pass existing worktree ID in metadata for reuse + if len(session.Worktrees) > 0 && session.Worktrees[0].WorktreeID != "" { + if req.Metadata == nil { + req.Metadata = make(map[string]interface{}) + } + req.Metadata["worktree_id"] = session.Worktrees[0].WorktreeID + } + } + + // Get worktree info from running state (for workspace path) if running, err := e.repo.GetExecutorRunningBySessionID(ctx, sessionID); err == nil && running != nil { if running.WorktreePath != "" { req.RepositoryURL = running.WorktreePath @@ -902,7 +952,10 @@ func (e *Executor) SwitchModel(ctx context.Context, taskID, sessionID, newModel, zap.String("task_id", task.ID), zap.String("session_id", sessionID), zap.String("model", newModel), - zap.String("acp_session_id", acpSessionID)) + zap.String("executor_type", req.ExecutorType), + zap.String("acp_session_id", acpSessionID), + zap.Bool("use_worktree", req.UseWorktree), + zap.String("repository_path", req.RepositoryPath)) // Launch the new agent resp, err := e.agentManager.LaunchAgent(ctx, req) @@ -1081,6 +1134,51 @@ func (e *Executor) applyExecutorMetadata(ctx context.Context, metadata map[strin return metadata } +// getExecutorType resolves an executor ID to its type string. +// Returns empty string if executor not found or on error. +func (e *Executor) getExecutorType(ctx context.Context, executorID string) string { + if executorID == "" { + return "" + } + executor, err := e.repo.GetExecutor(ctx, executorID) + if err != nil || executor == nil { + return "" + } + return string(executor.Type) +} + +// executorConfig holds resolved executor configuration. +type executorConfig struct { + ExecutorID string + ExecutorType string + Metadata map[string]interface{} +} + +// resolveExecutorConfig resolves executor configuration from an executor ID. +// If executorID is empty, it falls back to the workspace default. +// Returns the resolved config with executor ID, type, and metadata. +func (e *Executor) resolveExecutorConfig(ctx context.Context, executorID, workspaceID string, existingMetadata map[string]interface{}) executorConfig { + resolved := executorID + if resolved == "" { + resolved = e.defaultExecutorID(ctx, workspaceID) + } + + metadata := existingMetadata + if metadata == nil { + metadata = make(map[string]interface{}) + } + + if resolved != "" { + metadata = e.applyExecutorMetadata(ctx, metadata, resolved) + } + + return executorConfig{ + ExecutorID: resolved, + ExecutorType: e.getExecutorType(ctx, resolved), + Metadata: metadata, + } +} + func cloneMetadata(src map[string]interface{}) map[string]interface{} { if len(src) == 0 { return nil diff --git a/apps/backend/internal/orchestrator/handlers/handlers.go b/apps/backend/internal/orchestrator/handlers/handlers.go index 90d5c47bcc..a621f3c24c 100644 --- a/apps/backend/internal/orchestrator/handlers/handlers.go +++ b/apps/backend/internal/orchestrator/handlers/handlers.go @@ -84,6 +84,7 @@ func (h *Handlers) wsTriggerTask(ctx context.Context, msg *ws.Message) (*ws.Mess type wsStartTaskRequest struct { TaskID string `json:"task_id"` AgentProfileID string `json:"agent_profile_id,omitempty"` + ExecutorID string `json:"executor_id,omitempty"` Priority int `json:"priority,omitempty"` Prompt string `json:"prompt,omitempty"` } @@ -103,6 +104,7 @@ func (h *Handlers) wsStartTask(ctx context.Context, msg *ws.Message) (*ws.Messag resp, err := h.controller.StartTask(ctx, dto.StartTaskRequest{ TaskID: req.TaskID, AgentProfileID: req.AgentProfileID, + ExecutorID: req.ExecutorID, Priority: req.Priority, Prompt: req.Prompt, }) diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index c83d734925..4e211d68de 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -42,10 +42,11 @@ func (s *Service) EnqueueTask(ctx context.Context, task *v1.Task) error { } // StartTask manually starts agent execution for a task -func (s *Service) StartTask(ctx context.Context, taskID string, agentProfileID string, priority int, prompt string) (*executor.TaskExecution, error) { +func (s *Service) StartTask(ctx context.Context, taskID string, agentProfileID string, executorID string, priority int, prompt string) (*executor.TaskExecution, error) { s.logger.Debug("manually starting task", zap.String("task_id", taskID), zap.String("agent_profile_id", agentProfileID), + zap.String("executor_id", executorID), zap.Int("priority", priority), zap.Int("prompt_length", len(prompt))) @@ -75,7 +76,7 @@ func (s *Service) StartTask(ctx context.Context, taskID string, agentProfileID s effectivePrompt = task.Description } - execution, err := s.executor.ExecuteWithProfile(ctx, task, agentProfileID, effectivePrompt) + execution, err := s.executor.ExecuteWithProfile(ctx, task, agentProfileID, executorID, effectivePrompt) if err != nil { return nil, err } diff --git a/apps/backend/internal/task/handlers/process_handlers_test.go b/apps/backend/internal/task/handlers/process_handlers_test.go index 73ed38171b..d105e5beac 100644 --- a/apps/backend/internal/task/handlers/process_handlers_test.go +++ b/apps/backend/internal/task/handlers/process_handlers_test.go @@ -374,7 +374,7 @@ func getExecutionStore(t *testing.T, mgr *lifecycle.Manager) *lifecycle.Executio func newLifecycleManager(t *testing.T, log *logger.Logger) *lifecycle.Manager { t.Helper() reg := registry.NewRegistry(log) - return lifecycle.NewManager(reg, nil, nil, nil, nil, nil, nil, log) + return lifecycle.NewManager(reg, nil, nil, nil, nil, nil, nil, lifecycle.RuntimeFallbackWarn, log) } func newAgentctlClient(t *testing.T, serverURL string, log *logger.Logger) *agentctlclient.Client { diff --git a/apps/backend/internal/task/handlers/task_handlers.go b/apps/backend/internal/task/handlers/task_handlers.go index 399f7993a8..7377250d2b 100644 --- a/apps/backend/internal/task/handlers/task_handlers.go +++ b/apps/backend/internal/task/handlers/task_handlers.go @@ -23,7 +23,7 @@ type TaskHandlers struct { } type OrchestratorStarter interface { - StartTask(ctx context.Context, taskID string, agentProfileID string, priority int, prompt string) (*executor.TaskExecution, error) + StartTask(ctx context.Context, taskID string, agentProfileID string, executorID string, priority int, prompt string) (*executor.TaskExecution, error) } func NewTaskHandlers(ctrl *controller.TaskController, orchestrator OrchestratorStarter, log *logger.Logger) *TaskHandlers { @@ -129,6 +129,7 @@ type httpCreateTaskRequest struct { Metadata map[string]interface{} `json:"metadata,omitempty"` StartAgent bool `json:"start_agent,omitempty"` AgentProfileID string `json:"agent_profile_id,omitempty"` + ExecutorID string `json:"executor_id,omitempty"` } type createTaskResponse struct { @@ -191,12 +192,13 @@ func (h *TaskHandlers) httpCreateTask(c *gin.Context) { // Launch agent asynchronously so the HTTP request can return immediately. // The frontend will receive WebSocket updates when the agent actually starts. // This allows the "new task" dialog to close right away while setup scripts run. + executorID := body.ExecutorID // Capture for goroutine go func() { // Use a longer timeout to accommodate setup scripts (which can take minutes for npm install, etc.) startCtx, cancel := context.WithTimeout(context.Background(), constants.AgentLaunchTimeout) defer cancel() // Use task description as the initial prompt - execution, err := h.orchestrator.StartTask(startCtx, resp.ID, body.AgentProfileID, body.Priority, resp.Description) + execution, err := h.orchestrator.StartTask(startCtx, resp.ID, body.AgentProfileID, executorID, body.Priority, resp.Description) if err != nil { h.logger.Error("failed to start agent for task (async)", zap.Error(err), zap.String("task_id", resp.ID)) return @@ -204,6 +206,7 @@ func (h *TaskHandlers) httpCreateTask(c *gin.Context) { h.logger.Info("agent started for task (async)", zap.String("task_id", resp.ID), zap.String("session_id", execution.SessionID), + zap.String("executor_id", executorID), zap.String("execution_id", execution.AgentExecutionID)) }() } @@ -356,6 +359,7 @@ type wsCreateTaskRequest struct { Metadata map[string]interface{} `json:"metadata,omitempty"` StartAgent bool `json:"start_agent,omitempty"` AgentProfileID string `json:"agent_profile_id,omitempty"` + ExecutorID string `json:"executor_id,omitempty"` } func (h *TaskHandlers) wsCreateTask(ctx context.Context, msg *ws.Message) (*ws.Message, error) { @@ -414,11 +418,15 @@ func (h *TaskHandlers) wsCreateTask(ctx context.Context, msg *ws.Message) (*ws.M response := createTaskResponse{TaskDTO: resp} if req.StartAgent && req.AgentProfileID != "" && h.orchestrator != nil { // Use task description as the initial prompt - execution, err := h.orchestrator.StartTask(ctx, resp.ID, req.AgentProfileID, req.Priority, resp.Description) + execution, err := h.orchestrator.StartTask(ctx, resp.ID, req.AgentProfileID, req.ExecutorID, req.Priority, resp.Description) if err != nil { h.logger.Error("failed to start agent for task", zap.Error(err)) return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to start agent for task", nil) } + h.logger.Info("wsCreateTask started agent", + zap.String("task_id", resp.ID), + zap.String("executor_id", req.ExecutorID), + zap.String("session_id", execution.SessionID)) response.TaskSessionID = execution.SessionID response.AgentExecutionID = execution.AgentExecutionID } diff --git a/apps/backend/internal/task/service/service.go b/apps/backend/internal/task/service/service.go index 92aceb8036..92c76b0482 100644 --- a/apps/backend/internal/task/service/service.go +++ b/apps/backend/internal/task/service/service.go @@ -519,6 +519,17 @@ func (s *Service) DeleteTask(ctx context.Context, id string) error { } } + // Stop any running execution BEFORE deleting from database + // This must happen first because StopTask queries the database for active sessions + if s.executionStopper != nil { + if err := s.executionStopper.StopTask(ctx, id, "task deleted", true); err != nil { + s.logger.Warn("failed to stop task execution on delete", + zap.String("task_id", id), + zap.Error(err)) + // Continue with deletion even if stop fails + } + } + if err := s.repo.DeleteTask(ctx, id); err != nil { s.logger.Error("failed to delete task", zap.String("task_id", id), zap.Error(err)) return err @@ -529,9 +540,9 @@ func (s *Service) DeleteTask(ctx context.Context, id string) error { zap.String("task_id", id), zap.Duration("duration", time.Since(start))) - // Perform cleanup synchronously with dedicated timeout + // Perform remaining cleanup synchronously with dedicated timeout // Use background context since the original request may complete - if s.executionStopper != nil || s.worktreeCleanup != nil || len(sessions) > 0 { + if s.worktreeCleanup != nil || len(sessions) > 0 { cleanupStart := time.Now() cleanupCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() @@ -554,6 +565,8 @@ func (s *Service) DeleteTask(ctx context.Context, id string) error { } // performTaskCleanup handles post-deletion cleanup operations. +// Note: Execution stopping is done BEFORE database deletion in DeleteTask, +// so this function only handles worktree cleanup and executor_running records. // Returns a slice of errors encountered (empty if all succeeded). func (s *Service) performTaskCleanup( ctx context.Context, @@ -563,16 +576,6 @@ func (s *Service) performTaskCleanup( ) []error { var errs []error - // Stop any running execution - if s.executionStopper != nil { - if err := s.executionStopper.StopTask(ctx, taskID, "task deleted", true); err != nil { - s.logger.Warn("failed to stop task execution on delete", - zap.String("task_id", taskID), - zap.Error(err)) - errs = append(errs, fmt.Errorf("stop execution: %w", err)) - } - } - // Cleanup worktrees if len(worktrees) > 0 { if cleaner, ok := s.worktreeCleanup.(WorktreeBatchCleaner); ok { diff --git a/apps/web/components/task-create-dialog.tsx b/apps/web/components/task-create-dialog.tsx index 148387985b..2db2a4cdf9 100644 --- a/apps/web/components/task-create-dialog.tsx +++ b/apps/web/components/task-create-dialog.tsx @@ -576,11 +576,13 @@ export function TaskCreateDialog({ state: targetState, start_agent: Boolean(startAgent && !isEditMode && agentProfileId), agent_profile_id: startAgent ? agentProfileId || undefined : undefined, + executor_id: startAgent ? executorId || undefined : undefined, }); console.log('[TaskCreateDialog] task created', { taskId: taskResponse.id, startAgent, agentProfileId, + executorId, }); onSuccess?.(taskResponse, 'create', { taskSessionId: taskResponse.session_id ?? null, diff --git a/apps/web/components/task/sessions-dropdown.tsx b/apps/web/components/task/sessions-dropdown.tsx index 9730da064c..eacdbba47f 100644 --- a/apps/web/components/task/sessions-dropdown.tsx +++ b/apps/web/components/task/sessions-dropdown.tsx @@ -164,6 +164,7 @@ export function SessionsDropdown({ { task_id: taskId, agent_profile_id: data.agentProfileId, + executor_id: data.executorId, prompt: data.prompt.trim(), }, 15000 diff --git a/apps/web/lib/api/domains/kanban-api.ts b/apps/web/lib/api/domains/kanban-api.ts index a7c131b726..f924f07c1e 100644 --- a/apps/web/lib/api/domains/kanban-api.ts +++ b/apps/web/lib/api/domains/kanban-api.ts @@ -38,6 +38,7 @@ export async function createTask( state?: Task['state']; start_agent?: boolean; agent_profile_id?: string; + executor_id?: string; }, options?: ApiRequestOptions ) { diff --git a/docs/architecture/task-specific-executor-refactoring.md b/docs/architecture/task-specific-executor-refactoring.md new file mode 100644 index 0000000000..443aec889b --- /dev/null +++ b/docs/architecture/task-specific-executor-refactoring.md @@ -0,0 +1,560 @@ +# Task-Specific Executor Selection Architecture Refactoring + +> **Status**: Proposal +> **Author**: Generated for architectural review +> **Date**: 2026-01-25 + +## Executive Summary + +This document proposes refactoring the executor selection architecture from a global runtime configuration to a per-task executor selection model. The goal is to enable multiple runtimes (Docker and Standalone) to coexist, with each task independently selecting its execution environment. + +--- + +## 1. Current Architecture + +### 1.1 Global Runtime Configuration + +The backend currently uses a **global runtime configuration** that determines how ALL agents are launched: + +```go +// apps/backend/internal/common/config/config.go +type AgentConfig struct { + Runtime string // "docker" or "standalone" - GLOBAL for all tasks + StandaloneHost string + StandalonePort int +} +``` + +The `provideLifecycleManager()` function in `apps/backend/cmd/kandev/agents.go` creates a **single runtime** based on this global config: + +```go +func provideLifecycleManager(...) (*lifecycle.Manager, *registry.Registry, error) { + var agentRuntime lifecycle.Runtime + switch cfg.Agent.Runtime { + case "standalone": + agentRuntime = lifecycle.NewStandaloneRuntime(...) // ALL tasks use this + default: + agentRuntime = lifecycle.NewDockerRuntime(...) // OR ALL tasks use this + } + // Only ONE runtime is created and passed to the Manager + lifecycleMgr := lifecycle.NewManager(..., agentRuntime, ...) +} +``` + +### 1.2 Why This Is Limiting + +1. **All-or-nothing approach**: Every task uses either Docker OR Standalone, never both +2. **No runtime flexibility**: Can't run lightweight tasks on host while running isolated tasks in Docker +3. **Configuration at startup**: Runtime is fixed when the backend starts; changing requires restart +4. **Underutilized Executor model**: The database has `Executor` with `Type` field (`local_pc`, `local_docker`) but it's not used for runtime selection + +### 1.3 Current Data Flow + +``` +Task Creation + │ + ├── TaskSession.ExecutorID is set (from workspace default) + │ But it's only used for metadata (MCP policy), not runtime selection + │ + └── executor.ExecuteWithProfile() + │ + └── LaunchAgent() → lifecycle.Manager.Launch() + │ + └── m.runtime.CreateInstance() ← Uses GLOBAL runtime +``` + +### 1.4 Existing Executor Infrastructure + +The database already has: +- `executors` table with `type` field (local_pc, local_docker, remote_docker, etc.) +- `task_sessions.executor_id` foreign key +- `executors_running.runtime` field (currently underutilized) +- Default executors seeded: "Local PC" (exec-local-pc) and "Local Docker" (exec-local-docker) + +--- + +## 2. Proposed Architecture + +### 2.1 Core Change: Multi-Runtime Manager + +Replace the single-runtime lifecycle manager with a **runtime registry** that can dispatch to multiple runtimes based on executor type: + +```go +// NEW: RuntimeRegistry manages multiple runtimes +type RuntimeRegistry struct { + runtimes map[runtime.Name]Runtime // docker, standalone, etc. + logger *logger.Logger +} + +func (r *RuntimeRegistry) GetRuntime(name runtime.Name) (Runtime, error) { + rt, ok := r.runtimes[name] + if !ok { + return nil, fmt.Errorf("runtime %q not available", name) + } + return rt, nil +} + +// Lifecycle Manager now uses RuntimeRegistry instead of single Runtime +type Manager struct { + runtimeRegistry *RuntimeRegistry // Replaces: runtime Runtime + // ... rest unchanged +} +``` + +### 2.2 Executor Type → Runtime Mapping + +```go +// ExecutorType to Runtime mapping +func ExecutorTypeToRuntime(execType models.ExecutorType) runtime.Name { + switch execType { + case models.ExecutorTypeLocalPC: + return runtime.NameStandalone + case models.ExecutorTypeLocalDocker: + return runtime.NameDocker + case models.ExecutorTypeRemoteDocker: + return runtime.NameDocker // Future: separate RemoteDockerRuntime + default: + return runtime.NameStandalone // Fallback + } +} +``` + +### 2.3 New Data Flow + +``` +Task Creation (with executor_id) + │ + ├── TaskSession.ExecutorID → Lookup Executor → Get ExecutorType + │ + └── executor.ExecuteWithProfile() + │ + └── LaunchAgentRequest now includes ExecutorType + │ + └── lifecycle.Manager.Launch() + │ + ├── Resolve runtime: ExecutorTypeToRuntime(req.ExecutorType) + │ + └── runtimeRegistry.GetRuntime(runtimeName).CreateInstance() +``` + +### 2.4 Configuration Changes + +**Remove** the global `cfg.Agent.Runtime` setting. Instead: + +```yaml +# BEFORE: Global runtime selection (to be removed) +agent: + runtime: "standalone" # DELETE THIS + +# AFTER: Runtime availability configuration +agent: + docker: + enabled: true + # Docker-specific config (image, network, etc.) + standalone: + host: "localhost" + port: 9999 + # NOTE: No "enabled" flag - agentctl always runs as a core service +``` + +**Important**: The `standalone` section configures connection settings for the agentctl process, but there's no `enabled` flag. The agentctl launcher always starts as a core service - it's the control plane for standalone execution and is always available. + +--- + +## 3. Files Requiring Modification + +### 3.1 Configuration Layer + +| File | Change | +|------|--------| +| `internal/common/config/config.go` | Remove `Runtime` field; add `Docker.Enabled` (standalone is always on) | +| `cmd/kandev/agents.go` | Refactor `provideLifecycleManager()` to create RuntimeRegistry with both runtimes | +| `cmd/kandev/agentctl.go` | **Always start agentctl launcher** (remove conditional check) | + +### 3.2 Lifecycle Manager + +| File | Change | +|------|--------| +| `internal/agent/lifecycle/manager.go` | Replace `runtime Runtime` with `runtimeRegistry *RuntimeRegistry`; update `Launch()` to select runtime via registry | +| `internal/agent/lifecycle/runtime_registry.go` | **NEW FILE**: RuntimeRegistry implementation | +| `internal/agent/lifecycle/runtime.go` | Add `ExecutorTypeToRuntime()` mapping function | + +> **Design Principle**: Keep `manager.go` free of runtime-specific code (no Docker imports, no standalone-specific logic). All runtime-specific behavior should be encapsulated in the `Runtime` interface implementations (`runtime_docker.go`, `runtime_standalone.go`) and dispatched through the `RuntimeRegistry`. The manager should only interact with the abstract `Runtime` interface. + +### 3.3 Orchestrator/Executor + +| File | Change | +|------|--------| +| `internal/orchestrator/executor/executor.go` | Pass `ExecutorType` in `LaunchAgentRequest`; resolve executor before launch | +| `internal/agent/lifecycle/types.go` | Add `ExecutorType` field to `LaunchRequest` | +| `cmd/kandev/adapters.go` | Pass `ExecutorType` from executor request to lifecycle request | + +### 3.4 Data Layer + +| File | Change | +|------|--------| +| `internal/task/models/models.go` | Add helper: `Executor.GetRuntimeName()` | +| `internal/agent/runtime/runtime.go` | Add `ExecutorTypeToRuntime()` function | + +### 3.5 Recovery/Resumption + +| File | Change | +|------|--------| +| `internal/orchestrator/service.go` | `resumeExecutorsOnStartup()` must iterate all available runtimes | +| `internal/agent/lifecycle/manager.go` | `RecoverExecutions()` must query each runtime in registry | + +--- + +## 4. Implementation Plan + +### Phase 1: Infrastructure Setup (Non-Breaking) + +1. **Create RuntimeRegistry** (`runtime_registry.go`) + - Implement registry with `Register()`, `GetRuntime()`, `List()`, `HealthCheckAll()` + - Add `ExecutorTypeToRuntime()` mapping function + +2. **Update Configuration** + - Add `Docker.Enabled` and `Standalone.Enabled` fields + - Keep `Runtime` field temporarily for backward compatibility + - Add deprecation warning if `Runtime` is used + +3. **Refactor `provideLifecycleManager()`** + - Create both runtimes if enabled + - Build RuntimeRegistry with available runtimes + - Pass registry to Manager (but Manager still uses single runtime internally) + +### Phase 2: Lifecycle Manager Refactoring + +4. **Update Manager to use RuntimeRegistry** + - Change `runtime Runtime` to `runtimeRegistry *RuntimeRegistry` + - Add `ExecutorType` to `LaunchRequest` + - Update `Launch()` to: + ```go + runtimeName := ExecutorTypeToRuntime(req.ExecutorType) + rt, err := m.runtimeRegistry.GetRuntime(runtimeName) + if err != nil { + return nil, fmt.Errorf("runtime not available: %w", err) + } + runtimeInstance, err := rt.CreateInstance(ctx, runtimeReq) + ``` + +5. **Update RecoverExecutions()** + - Iterate all registered runtimes + - Merge recovered instances from each runtime + +### Phase 3: Executor Integration + +6. **Update `LaunchAgentRequest`** + - Add `ExecutorType models.ExecutorType` field + +7. **Update `executor.ExecuteWithProfile()`** + - Resolve executor before launch + - Pass executor type in request + +8. **Update `lifecycleAdapter.LaunchAgent()`** + - Map executor type to lifecycle request + +### Phase 4: Cleanup + +9. **Remove global `cfg.Agent.Runtime`** + - Delete from config struct + - Update flag parsing + - Update documentation + +10. **Update agentctl launcher logic** + - **Always start** the agentctl launcher (remove all conditional checks) + - agentctl is a core service, not dependent on runtime configuration + +--- + +## 5. Edge Cases and Challenges + +### 5.1 Session Resumption Across Runtimes + +**Challenge**: A session started on Docker cannot be resumed on Standalone (different environment). + +**Solution**: Store runtime name in `executors_running.runtime` and validate on resume: +```go +func (e *Executor) ResumeSession(...) { + running, _ := e.repo.GetExecutorRunningBySessionID(session.ID) + if running != nil && running.Runtime != "" { + // Use the same runtime that was used originally + req.ExecutorType = RuntimeToExecutorType(running.Runtime) + } +} +``` + +### 5.2 Runtime Availability Changes + +**Challenge**: Task was created with Docker executor, but Docker is disabled on restart. + +**Solution**: +- Check runtime availability before launch +- Return clear error: "Runtime 'docker' is not available. Enable it in configuration or change the task's executor." +- Frontend can show unavailable executors as disabled + +### 5.3 Fallback Behavior + +**Challenge**: Executor not set on task/session. + +**Solution**: Define precedence: +1. Session's executor (if resuming) +2. Workspace default executor +3. System default ("Local PC" → Standalone) + +### 5.4 agentctl Architecture + +**Clarification**: agentctl runs in **both** modes: +- **Docker**: agentctl runs inside each container (started by entrypoint) +- **Standalone**: agentctl runs as a subprocess on the host, managing multiple agent instances + +**Current**: agentctl launcher only started if `cfg.Agent.Runtime == "standalone"` + +**New**: agentctl launcher should **always start** on the host. It serves as the control plane for standalone execution. The launcher startup should not depend on any runtime configuration - it's a core service that's always available. + +### 5.5 Container Recovery + +**Challenge**: Docker runtime recovers containers on startup; Standalone doesn't persist instances. + +**Solution**: +- Docker: `RecoverInstances()` finds containers by label +- Standalone: Returns empty (sessions recreated on resume via `LaunchAgent()`) +- RuntimeRegistry: Iterates all runtimes, merges results + +--- + +## 6. Database Schema Changes + +No schema changes required. Existing fields are sufficient: +- `task_sessions.executor_id` - Links to selected executor +- `executors.type` - Determines runtime (local_pc → standalone, local_docker → docker) +- `executors_running.runtime` - Records which runtime was used + +--- + +## 7. API Changes + +### 7.1 Task Creation (No Change Needed) + +The `executor_id` can already be passed via: +- Workspace default (`workspace.default_executor_id`) +- Potentially extended to task creation payload in future + +### 7.2 New: Runtime Status Endpoint + +```json +GET /api/v1/runtimes/status +{ + "runtimes": [ + { "name": "standalone", "available": true, "healthy": true }, + { "name": "docker", "available": true, "healthy": true } + ] +} +``` + +This allows the frontend to: +- Show which executors are available +- Disable executor selection for unavailable runtimes + +--- + +## 8. Testing Strategy + +1. **Unit Tests** + - `RuntimeRegistry.GetRuntime()` for registered/unregistered runtimes + - `ExecutorTypeToRuntime()` mapping + - `Launch()` with different executor types + +2. **Integration Tests** + - Start task with `local_pc` executor → uses Standalone runtime + - Start task with `local_docker` executor → uses Docker runtime + - Resume session maintains original runtime + - Backend restart with Docker disabled → sessions on Docker can't resume + +3. **E2E Tests** + - Create workspace with Docker as default executor + - Start task → runs in Docker container + - Change workspace default to Local PC + - Start new task → runs via Standalone + - Both tasks run concurrently with different runtimes + +--- + +## 9. Code Examples + +### 9.1 RuntimeRegistry Implementation + +```go +// internal/agent/lifecycle/runtime_registry.go +package lifecycle + +import ( + "context" + "fmt" + "sync" + + "github.com/kandev/kandev/internal/agent/runtime" + "github.com/kandev/kandev/internal/common/logger" +) + +type RuntimeRegistry struct { + runtimes map[runtime.Name]Runtime + mu sync.RWMutex + logger *logger.Logger +} + +func NewRuntimeRegistry(log *logger.Logger) *RuntimeRegistry { + return &RuntimeRegistry{ + runtimes: make(map[runtime.Name]Runtime), + logger: log, + } +} + +func (r *RuntimeRegistry) Register(rt Runtime) { + r.mu.Lock() + defer r.mu.Unlock() + r.runtimes[rt.Name()] = rt + r.logger.Info("registered runtime", zap.String("name", string(rt.Name()))) +} + +func (r *RuntimeRegistry) GetRuntime(name runtime.Name) (Runtime, error) { + r.mu.RLock() + defer r.mu.RUnlock() + rt, ok := r.runtimes[name] + if !ok { + return nil, fmt.Errorf("runtime %q not registered", name) + } + return rt, nil +} + +func (r *RuntimeRegistry) List() []runtime.Name { + r.mu.RLock() + defer r.mu.RUnlock() + names := make([]runtime.Name, 0, len(r.runtimes)) + for name := range r.runtimes { + names = append(names, name) + } + return names +} + +func (r *RuntimeRegistry) HealthCheckAll(ctx context.Context) map[runtime.Name]error { + r.mu.RLock() + defer r.mu.RUnlock() + results := make(map[runtime.Name]error) + for name, rt := range r.runtimes { + results[name] = rt.HealthCheck(ctx) + } + return results +} + +func (r *RuntimeRegistry) RecoverAll(ctx context.Context) ([]*RuntimeInstance, error) { + r.mu.RLock() + defer r.mu.RUnlock() + var allRecovered []*RuntimeInstance + for _, rt := range r.runtimes { + recovered, err := rt.RecoverInstances(ctx) + if err != nil { + r.logger.Warn("failed to recover instances from runtime", + zap.String("runtime", string(rt.Name())), + zap.Error(err)) + continue + } + allRecovered = append(allRecovered, recovered...) + } + return allRecovered, nil +} +``` + +### 9.2 Updated provideLifecycleManager + +```go +// apps/backend/cmd/kandev/agents.go +func provideLifecycleManager( + ctx context.Context, + cfg *config.Config, + log *logger.Logger, + eventBus bus.EventBus, + dockerClient *docker.Client, + agentSettingsRepo settingsstore.Repository, +) (*lifecycle.Manager, *registry.Registry, error) { + + // Create runtime registry + runtimeRegistry := lifecycle.NewRuntimeRegistry(log) + + // Standalone runtime is always available (agentctl is a core service) + controlClient := agentctl.NewControlClient( + cfg.Agent.Standalone.Host, + cfg.Agent.Standalone.Port, + log, + ) + standaloneRuntime := lifecycle.NewStandaloneRuntime( + controlClient, + cfg.Agent.Standalone.Host, + cfg.Agent.Standalone.Port, + log, + ) + runtimeRegistry.Register(standaloneRuntime) + log.Info("Standalone runtime registered") + + // Register Docker runtime if enabled and Docker client is available + if cfg.Agent.Docker.Enabled && dockerClient != nil { + dockerRuntime := lifecycle.NewDockerRuntime(dockerClient, log) + runtimeRegistry.Register(dockerRuntime) + log.Info("Docker runtime registered") + } + + // ... rest of initialization using runtimeRegistry + // NOTE: Manager receives the registry, not individual runtimes + // This keeps manager.go free of runtime-specific code + lifecycleMgr := lifecycle.NewManager( + agentRegistry, eventBus, runtimeRegistry, /* ... */ + ) + return lifecycleMgr, agentRegistry, nil +} +``` + +### 9.3 Updated Launch Method + +```go +// internal/agent/lifecycle/manager.go +func (m *Manager) Launch(ctx context.Context, req *LaunchRequest) (*AgentExecution, error) { + // ... existing validation and setup ... + + // Determine which runtime to use based on executor type + runtimeName := runtime.ExecutorTypeToRuntime(req.ExecutorType) + if runtimeName == runtime.NameUnknown { + runtimeName = runtime.NameStandalone // Default fallback + } + + rt, err := m.runtimeRegistry.GetRuntime(runtimeName) + if err != nil { + return nil, fmt.Errorf("runtime %q not available for executor type %q: %w", + runtimeName, req.ExecutorType, err) + } + + // Create runtime instance using the selected runtime + runtimeInstance, err := rt.CreateInstance(ctx, runtimeReq) + if err != nil { + return nil, fmt.Errorf("failed to create instance: %w", err) + } + + // Store runtime name for session resumption + execution := runtimeInstance.ToAgentExecution(runtimeReq) + execution.RuntimeName = runtimeName // NEW FIELD + + // ... rest unchanged ... +} +``` + +--- + +## 10. Migration Path + +Since breaking changes are acceptable, the migration is straightforward: + +1. **Deploy new backend** with both runtimes enabled +2. **Running sessions continue** (Docker containers persist; Standalone sessions will need resume) +3. **New tasks** use executor-specific runtime selection +4. **Remove old config** after verifying stability + +No database migration needed - schema already supports this model. \ No newline at end of file